feat: add support for host logging, deprecating old topics_logger in favor of host-managed topics logger. updates examples. (#21)

Author: tylerburdsall

Cargo.lock

@@ -34,4 +34,5 @@ serde_json              = { version = "1" }
 sha2                    = { version = "0" }
 tiktoken-rs             = { version = "0" }
 thiserror               = { version = "2" }
+time                    = { version = "0.3" }
 wit-bindgen             = { version = "0.41" }

Cargo.toml

@@ -16,6 +16,7 @@ pub mod aws;
 pub mod cache;
 pub mod encoding;
 pub mod http;
+pub mod logging;
 pub mod redis;
 mod spawn;
 pub mod topics;

momento-functions-host/src/lib.rs

@@ -0,0 +1,185 @@
+//! Host interfaces for working with host logging, allowing you to send
+//! logs to different destinations
+use momento_functions_wit::host::momento::host::logging;
+use thiserror::Error;
+
+/// Where do you want your logs to go?
+pub enum LogDestination {
+    /// Momento topic within the same cache as your function
+    Topic {
+        /// Name of the topic
+        topic: String,
+    },
+    /// AWS CloudWatch Log Group for your function's logs
+    CloudWatch {
+        /// ARN of the IAM role for Momento to assume
+        iam_role_arn: String,
+        /// ARN of the CloudWatch Log Group for Momento to publish your
+        /// function logs to
+        log_group_name: String,
+    },
+}
+
+impl LogDestination {
+    /// Creates a Topic destination
+    pub fn topic(name: impl Into<String>) -> Self {
+        Self::Topic { topic: name.into() }
+    }
+    /// Creates a CloudWatch destination.
+    /// Reach out to us at `[email protected]` for details on how to properly
+    /// set up your log configuration.
+    pub fn cloudwatch(iam_role_arn: impl Into<String>, log_group_name: impl Into<String>) -> Self {
+        Self::CloudWatch {
+            iam_role_arn: iam_role_arn.into(),
+            log_group_name: log_group_name.into(),
+        }
+    }
+}
+
+/// A single configuration for a destination
+pub struct LogConfiguration {
+    /// At what level would you like your function's logs to be filtered into this destination?
+    log_level: log::LevelFilter,
+    /// At what level would you like Momento's system logs to be filtered into this destination?
+    system_log_level: log::LevelFilter,
+    /// The specific destination
+    destination: LogDestination,
+}
+
+impl LogConfiguration {
+    /// Constructs a single logging input with a desired destination. System logs will be at default level (INFO).
+    pub fn new(destination: LogDestination) -> Self {
+        Self {
+            log_level: log::LevelFilter::Info,
+            system_log_level: log::LevelFilter::Info,
+            destination,
+        }
+    }
+
+    /// Constructs a single logging input with a desired destination as well as a specified logs filter.
+    pub fn with_log_level(mut self, log_level: log::LevelFilter) -> Self {
+        self.log_level = log_level;
+        self
+    }
+
+    /// Constructs a single logging input with a desired destination as well as a specified system logs filter.
+    pub fn with_system_log_level(mut self, system_log_level: log::LevelFilter) -> Self {
+        self.system_log_level = system_log_level;
+        self
+    }
+}
+
+impl From<LogDestination> for LogConfiguration {
+    fn from(value: LogDestination) -> Self {
+        match value {
+            LogDestination::Topic { topic } => Self::new(LogDestination::topic(topic)),
+            LogDestination::CloudWatch {
+                iam_role_arn,
+                log_group_name,
+            } => Self::new(LogDestination::cloudwatch(iam_role_arn, log_group_name)),
+        }
+    }
+}
+
+/// Create a single `LogConfiguration` given a `LogDestination`.
+pub fn log_configuration(destination: LogDestination) -> LogConfiguration {
+    LogConfiguration::new(destination)
+}
+
+impl From<LogDestination> for logging::Destination {
+    fn from(value: LogDestination) -> Self {
+        match value {
+            LogDestination::Topic { topic } => {
+                momento_functions_wit::host::momento::host::logging::Destination::Topic(
+                    logging::TopicDestination { topic_name: topic },
+                )
+            }
+            LogDestination::CloudWatch {
+                iam_role_arn,
+                log_group_name,
+            } => momento_functions_wit::host::momento::host::logging::Destination::Cloudwatch(
+                logging::CloudwatchDestination {
+                    iam_role_arn,
+                    log_group_name,
+                },
+            ),
+        }
+    }
+}
+
+impl From<LogConfiguration> for logging::ConfigureLoggingInput {
+    fn from(value: LogConfiguration) -> Self {
+        Self {
+            log_level: match value.log_level {
+                log::LevelFilter::Off => logging::LogLevel::Off,
+                log::LevelFilter::Error => logging::LogLevel::Error,
+                log::LevelFilter::Warn => logging::LogLevel::Warn,
+                log::LevelFilter::Info => logging::LogLevel::Info,
+                log::LevelFilter::Debug => logging::LogLevel::Debug,
+                // Momento does not publish Trace logs
+                log::LevelFilter::Trace => logging::LogLevel::Debug,
+            },
+            system_logs_level: match value.system_log_level {
+                log::LevelFilter::Off => logging::LogLevel::Off,
+                log::LevelFilter::Error => logging::LogLevel::Error,
+                log::LevelFilter::Warn => logging::LogLevel::Warn,
+                log::LevelFilter::Info => logging::LogLevel::Info,
+                log::LevelFilter::Debug => logging::LogLevel::Debug,
+                // Momento does not publish Trace logs
+                log::LevelFilter::Trace => logging::LogLevel::Debug,
+            },
+            destination: value.destination.into(),
+        }
+    }
+}
+
+/// Captures any errors Momento has detected during log configuration
+#[derive(Debug, Error)]
+pub enum LogConfigurationError {
+    #[error("Invalid auth provided for configuration! {message}")]
+    /// Invalid auth was provided
+    Auth {
+        /// The error message bubbled back up
+        message: String,
+    },
+    #[error("Something went wrong while trying to configure logs! {message}")]
+    /// Something went wrong
+    Unknown {
+        /// The error message bubbled back up
+        message: String,
+    },
+}
+
+impl From<logging::LogConfigurationError> for LogConfigurationError {
+    fn from(value: logging::LogConfigurationError) -> Self {
+        match value {
+            logging::LogConfigurationError::Auth(e) => Self::Auth { message: e },
+        }
+    }
+}
+
+/// Configures logging via Momento host functions
+pub fn configure_host_logging(
+    configurations: impl IntoIterator<Item = LogConfiguration>,
+) -> Result<(), LogConfigurationError> {
+    let configurations = configurations
+        .into_iter()
+        .map(|configuration| configuration.into())
+        .collect::<Vec<logging::ConfigureLoggingInput>>();
+    Ok(logging::configure_logging(&configurations)?)
+}
+
+/// Logs a given string
+pub fn log(input: &str, level: log::Level) {
+    logging::log(
+        input,
+        match level {
+            log::Level::Error => logging::LogLevel::Error,
+            log::Level::Warn => logging::LogLevel::Warn,
+            log::Level::Info => logging::LogLevel::Info,
+            log::Level::Debug => logging::LogLevel::Debug,
+            // Momento does not publish Trace logs
+            log::Level::Trace => logging::LogLevel::Debug,
+        },
+    )
+}

momento-functions-host/src/logging.rs

@@ -14,3 +14,4 @@ categories.workspace = true
 log                     = { workspace = true }
 momento-functions-host  = { workspace = true }
 thiserror               = { workspace = true }
+time                    = { workspace = true, features = ["formatting"] }

momento-functions-log/Cargo.toml

@@ -0,0 +1,59 @@
+use std::fmt::Write;
+
+use log::{Log, set_logger_racy, set_max_level};
+use momento_functions_host::logging::{LogConfiguration, LogConfigurationError};
+use time::format_description::well_known::Rfc3339;
+
+pub struct HostLog {}
+
+impl HostLog {
+    pub fn init(
+        configurations: impl IntoIterator<Item = LogConfiguration>,
+    ) -> Result<(), LogConfigurationError> {
+        static mut LOGGER: Option<HostLog> = None;
+        // We're setting this to DEBUG so all logs are captured and sent to the host serving
+        // the function. The host will determine whether to log the mesage.
+        set_max_level(log::LevelFilter::Debug);
+        momento_functions_host::logging::configure_host_logging(configurations)?;
+        #[allow(static_mut_refs)]
+        #[allow(clippy::expect_used)]
+        // SAFETY: concurrency requirement is satisfied by the single threaded nature
+        // of the Function environment.
+        unsafe {
+            LOGGER.replace(HostLog {});
+            set_logger_racy(LOGGER.as_mut().expect("logger just set")).map_err(|e| {
+                LogConfigurationError::Unknown {
+                    message: format!("Failed to configure logger! {e:?}"),
+                }
+            })
+        }
+    }
+}
+
+impl Log for HostLog {
+    fn enabled(&self, _metadata: &log::Metadata) -> bool {
+        // Host logging will filter out logs based on level
+        true
+    }
+
+    fn log(&self, record: &log::Record) {
+        let mut buffer = String::with_capacity(128);
+        let utc_now = time::OffsetDateTime::now_utc();
+        let timestamp = utc_now.format(&Rfc3339).unwrap_or("<unknown>".to_string());
+        let record_level = record.level();
+        let level = record_level.as_str();
+        let module = record.module_path().unwrap_or("<unknown>");
+        let file = record.file().unwrap_or("<unknown>");
+        let line = record.line().unwrap_or(0);
+        let log_message = record.args();
+
+        let _ = write!(
+            &mut buffer,
+            "{level} {timestamp} {module} {file}:{line} {log_message}"
+        );
+
+        momento_functions_host::logging::log(buffer.as_str(), record_level);
+    }
+
+    fn flush(&self) {}
+}

momento-functions-log/src/host_logging.rs

@@ -11,35 +11,14 @@
 //! * [`momento-functions`](https://crates.io/crates/momento-functions): Code generators for Functions.
 //! * [`momento-functions-host`](https://crates.io/crates/momento-functions-host): Interfaces and tools for calling host interfaces.
 
-use log::SetLoggerError;
-use thiserror::Error;
+use momento_functions_host::logging::{LogConfiguration, LogConfigurationError};
 
-mod topic_logger;
+use crate::host_logging::HostLog;
+mod host_logging;
 
-/// Which logging mode to use?
-pub enum LogMode {
-    Topic {
-        /// The topic to send logs to.
-        ///
-        /// You can get the logs with the `momento` CLI, or on the Momento topics dashboard at gomomento.com.
-        /// The CLI command would be `momento topic subscribe $topic`
-        /// Log messages will stream to your terminal.
-        topic: String,
-    },
-}
-
-#[derive(Debug, Error)]
-pub enum LogConfigError {
-    #[error("Failed to initialize topics logger: {cause}")]
-    TopicsInit { cause: SetLoggerError },
-}
-
-/// Initializes the logging system with the specified log level and mode.
-///
-/// You **must** only call this function once.
-pub fn configure_logging(level: log::LevelFilter, mode: LogMode) -> Result<(), LogConfigError> {
-    match mode {
-        LogMode::Topic { topic } => topic_logger::TopicLog::init(level, topic)
-            .map_err(|e| LogConfigError::TopicsInit { cause: e }),
-    }
+/// Entrypoint for configuring logs to be delivered to a destination(s)
+pub fn configure_logs(
+    configurations: impl IntoIterator<Item = LogConfiguration>,
+) -> Result<(), LogConfigurationError> {
+    HostLog::init(configurations)
 }