[RFC] Lambda Logs API Integration

[nmoutschen]

The purpose of this issue is to define and agree on an library implementation for the Lambda Logs API.

Before going into proposed user experiences, there are a few characteristics of the Logs API that are important to consider:

1. Only extensions can subscribe to logs. You need to first register the extension, then subscribe for the logs using the same extension identifier.
2. Upon subscribing, Lambda will deliver logs to a local endpoint (HTTP or TCP). Meaning we only send a single API call on initialization.

Challenges with current implementation

Right now, all libraries (lambda_runtime and lambda_extension) abstract the registration process under the hood. That means developers don't have control between the registration process and the first invocation. However, to work correctly with the Lambda Logs API, a developer needs to first register the extension, then send a call to subscribe for logs.

That means that to implement this feature, we either need to give more flexibility, or create a new abstraction just for the logs API. With the former, this would make it harder to move all handler traits to tower::Service (see #374) unless we implement a factory pattern. With the latter, we would remove the possibility of having an extension serving multiple purposes.

Proposition 1: High-level library

The user experience would resemble the ones from other crates in this project. It would also take care of running a TCP server, and send log entries to a function or trait implementation provided.

Pros:

• High-level library that hides the Logs API complexity under the hood
• Works out of the box with migrating to tower::Service

Cons:

• Less flexible for those that want to react to events or do something else while the extension is running

use lambda_logs::{log_fn, Error, LogEvent};

async fn log_processor(log_events: &[LogEvent]) -> Result<(), Error> {
    // Process a batch of log entries here
    todo!();
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let func = log_fn(log_processor);
    lambda_logs::run(func).await
}

Proposition 2: Low-level library with init function

This would provide a low-level library only. The developers would be responsible for starting an HTTP or TCP server, but would still benefit from a function that will subscribe for logs, and structs for log entries. Developers would still use lambda_extension to actually register the extension, and then subscribe to the logs.

Pros:

• Give more flexibility to the developers

Cons:

• Does not support migrating to tower::Service (extra init function)
• Developers need to start their own TCP/HTTP server

use lambda_extension::{Extension, Error, LambdaEvent};
use lambda_logs::{
    // Function to make the API call
    subscribe_logs,
    // Schema for log events
    LogEvent.
};
use std::{
    future::{ready, Future},
    pin::Pin,
};

#[derive(Default)]
struct LogExtension;

impl Extension for LogExtension {
    type Fut = Pin<Box<dyn Future<Output = Result<(), Error>>>>;

    fn init(&mut self, extension_id: &str) -> Result<(), Error> {
        // Start an HTTP server
       let endpoint: String = my_function_to_make_a_server();
        subscribe_logs(extension_id, &endpoint);
    }

    fn call(&mut self, event: LambdaEvent) -> Self::Fut {
        // Process events from the extension
    }
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    run(LogExtension::default()).await
}

Proposition 3: Low-level library with factory pattern

TODO

Pros:

• Give more flexibility to the developers
• Supports migrating to tower::Service

Cons:

• Developers need to start their own TCP/HTTP server

Proposition 4: your idea here

If you have an idea that gives flexibility while providing a high-level interface, that'd be awesome, but I'm unsure what it would look like.

[calavera]

is there a way to have a builder pattern and still provide the HTTP server? I'm thinking about something like this, but I have not tested if this could be possible:

async fn event_processor(event: LambdaEvent) -> Result<(), Error> {
  Ok(())
}

async fn log_processor(log_events: &[LogEvent]) -> Result<(), Error> {
    // Process a batch of log entries here
    todo!();
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let extension = lambda_extension::builder()
                                .with_event_processor(event_processor)
                                .with_logs_processor(log_processor)
                                .build()?;
    lambda_extension::run(extension).await
}

[tannerntannern]

@calavera This is very similar to what I had in mind. I'd be in favor of this.

Personally, I don't see a need to support bringing your own HTTP/TCP server, at least initially. It seems like boilerplate the majority of users won't want to bother with, similar to the extension registration which is hidden from the user entirely. Also the docs recommend using HTTP instead of TCP, so I think we would be justified in having HTTP implemented by default.

[nmoutschen]

@calavera I like the idea of using a builder pattern. I'd just make a small change so we can support a trait pattern (for future implementation of tower::Service):

use lambda_extension::{service_fn, LambdaEvent, LogEvent};

async fn event_processor(event: LambdaEvent) -> Result<(), Error> {
  Ok(())
}

async fn log_processor(log_events: &[LogEvent]) -> Result<(), Error> {
    // Process a batch of log entries here
    todo!();
    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let extension = lambda_extension::builder()
                                .with_event_processor(service_fn(event_processor))
                                .with_logs_processor(service_fn(log_processor))
                                .build()?;
    lambda_extension::run(extension).await
}

This way we could also do this in the long term:

use lambda_extension::LogEvent;
use tower::Service;

#[derive(Default)]
struct MyLogProcessor;

impl Service<&[LogEvent]> for MyLogProcessor {
// implementation skipped
}

#[tokio::main]
async fn main() -> Result<(), Error> {
    let extension = lambda_extension::builder()
                                // We could also skip the event processor - this could use a default one instead
                                .with_logs_processor(MyLogProcessor::default())
                                .build()?;
    lambda_extension::run(extension).await
}

[calavera]

I'd just make a small change so we can support a trait pattern

that makes sense to me 👍

[nmoutschen]

Small change to my previous comment: I've replaced extension_fn and log_fn with service_fn.

[calavera]

yup, if you look at the examples in #407, I'm already using service_fn 😄