Add error logs subsystem (#117)

* Add error logs subsystem, hidden behind errorlogs flag

* Update monitor config to include cloudflare api token

* Switch branch for sdk for testing

* Add support for wrangler file bindings (#118)

* Add support for wrangler file bindings

* Add local wrangler files to gitignore

* Add support for routes

* PR comment updates

* Update test to match config change

* Bump version to 0.3.1 (#119)

* Fix Wrangler path issue (#121)

* Fix Wrangler path issue

* Add tests

* Bump version to v0.3.2 (#122)

* Limit files that get uploaded to Cloudflare (#123)

* Limit files that get uploaded to CF and add tests

* Remove outdated test

* Fix another outdated tests

* Add error logs subsystem, hidden behind errorlogs flag

* Update monitor config to include cloudflare api token

* Switch branch for sdk for testing

* Handle cancel canary

* Add rest of code

* Update cargo toml

Author: EricGhildyal

Cargo.lock

@@ -91,6 +91,7 @@ tempfile = "3.8"
 
 [features]
 proxy = ["dep:pingora"]
+errorlogs = []
 
 # The profile that 'dist' will build with
 [profile.dist]

Cargo.toml

@@ -8,7 +8,7 @@ use crate::{fs::Session, metrics::ResponseStatusCode, utils::circuit_breaker::Ht
 use bon::Builder;
 use chrono::DateTime;
 use miette::{IntoDiagnostic, Result, bail};
-use multitool_sdk::models::Rollout;
+use multitool_sdk::models::{CreateErrorRequest, Rollout};
 use multitool_sdk::{
     apis::{Api, ApiClient, configuration::Configuration},
     models::{
@@ -348,6 +348,30 @@ impl BackendClient {
         Ok(())
     }
 
+    /// Upload errors to the backend.
+    pub(crate) async fn upload_errors(
+        &self,
+        meta: &RolloutMetadata,
+        url_path: String,
+        status_code: i32,
+        logs: Vec<String>,
+    ) -> Result<()> {
+        trace!("Uploading error logs to backend");
+        let workspace_id = *meta.workspace_id();
+        let application_id = *meta.application_id();
+        let rollout_id = *meta.rollout_id();
+
+        let request = CreateErrorRequest::new(logs, status_code, url_path);
+
+        self.client
+            .errors_api()
+            .log_error(workspace_id, application_id, rollout_id, request)
+            .await
+            .into_diagnostic()?;
+
+        Ok(())
+    }
+
     /// Return information about the workspace given its name.
     pub(crate) async fn create_workspace<T: AsRef<str>>(
         &self,
@@ -583,6 +607,7 @@ pub enum MonitorConfig {
         dimensions: Vec<CloudWatchDimensions>,
     },
     CloudflareWorkersObservability {
+        api_token: String,
         account_id: String,
         worker_name: String,
     },

src/adapters/backend/mod.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 use serde::Deserialize;
 
 #[derive(Deserialize, Debug)]
@@ -17,3 +19,101 @@ pub struct Aggregate {
     #[serde(default)]
     pub count: u32,
 }
+
+#[derive(Deserialize, Debug)]
+pub struct ErrorLogsResponse {
+    #[serde(default)]
+    pub invocations: HashMap<String, Vec<InvocationData>>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct InvocationData {
+    #[serde(rename = "$workers")]
+    pub workers: WorkersData,
+    #[serde(rename = "$metadata")]
+    pub metadata: MetadataData,
+    pub source: SourceData,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct WorkersData {
+    pub event: EventData,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct EventData {
+    pub request: RequestData,
+    #[serde(default)]
+    pub response: Option<ResponseData>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct RequestData {
+    pub url: String,
+    pub method: String,
+    pub path: String,
+}
+
+#[derive(Deserialize, Debug, Clone)]
+pub struct ResponseData {
+    pub status: i32,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct MetadataData {
+    #[serde(rename = "type")]
+    pub event_type: String,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct SourceData {
+    pub message: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct CloudflareErrorLog {
+    pub method: String,
+    pub path: String,
+    pub status_code: i32,
+    pub logs: Vec<String>,
+}
+
+impl Into<Vec<CloudflareErrorLog>> for ErrorLogsResponse {
+    fn into(self) -> Vec<CloudflareErrorLog> {
+        let mut error_logs = Vec::new();
+
+        for (_request_id, invocations) in self.invocations {
+            // The cf-worker-event entry contains the request/response details
+            let event_entry = invocations
+                .iter()
+                .find(|inv| inv.metadata.event_type == "cf-worker-event");
+
+            if let Some(event) = event_entry {
+                let request = &event.workers.event.request;
+                let method = request.method.clone();
+                let path = request.path.clone();
+
+                // Extract log line from each invocation event and combine
+                if let Some(response) = &event.workers.event.response {
+                    let mut logs: Vec<String> = invocations
+                        .iter()
+                        .map(|inv| inv.source.message.clone())
+                        .collect();
+
+                    logs.reverse(); // Reverse to maintain chronological order
+
+                    let error_log = CloudflareErrorLog {
+                        method,
+                        path,
+                        status_code: response.status,
+                        logs,
+                    };
+
+                    error_logs.push(error_log);
+                }
+            }
+        }
+
+        error_logs
+    }
+}

src/adapters/cloudflare/metrics.rs

@@ -12,7 +12,7 @@ use uploads::{UploadVersionRequest, UploadVersionResponse};
 use url::Url;
 
 use deployments::{CreateDeploymentRequest, DeploymentResponse};
-use metrics::MetricsResponse;
+use metrics::{CloudflareErrorLog, ErrorLogsResponse, MetricsResponse};
 use responses::CloudflareResponse;
 use routes::{CloudflareRoute, CreateCloudflareRouteRequest, UpdateCloudflareRouteRequest};
 
@@ -33,6 +33,8 @@ pub struct CloudflareClient {
     account_id: String,
     /// The name of the Cloudflare worker
     worker_name: String,
+    /// The API token used for authentication
+    api_token: String,
 }
 
 impl CloudflareClient {
@@ -51,6 +53,7 @@ impl CloudflareClient {
 
         Self {
             client,
+            api_token: token.to_string(),
             account_id,
             worker_name,
         }
@@ -569,6 +572,87 @@ impl CloudflareClient {
         Ok(())
     }
 
+    /// Collects the logs produced by Cloudflare worker invocations that result in errors
+    /// Only collects 5xx errrors for now.
+    /// Returns grouped error logs maintaining the original invocation structure.
+    pub async fn collect_errors(
+        &self,
+        cf_worker_name: String,
+        from_time: DateTime<chrono::Utc>,
+        to_time: DateTime<chrono::Utc>,
+    ) -> Result<Vec<CloudflareErrorLog>> {
+        let account_id = &self.account_id;
+        let path = format!("accounts/{account_id}/workers/observability/telemetry/query");
+        let url = Self::url_with_path(&path);
+
+        // Convert DateTime to Unix Millisecond timestamps
+        let from_timestamp = from_time.timestamp_millis() as u64;
+        let to_timestamp = to_time.timestamp_millis() as u64;
+
+        let query_body = serde_json::json!({
+            "view": "invocations",
+            "queryId": "workers-logs-invocations",
+            "parameters": {
+                "datasets": [],
+                "filters": [
+                    {
+                        "key": "$metadata.service",
+                        "operation": "eq",
+                        "type": "string",
+                        "value": cf_worker_name
+                    },
+                    {
+                        "key": "$workers.event.response.status",
+                        "operation": "gte",
+                        "type": "number",
+                        "value": 500
+                    },
+                    {
+                        "key": "$workers.event.response.status",
+                        "operation": "lte",
+                        "type": "number",
+                        "value": 599
+                    }
+                ],
+                "calculations": [],
+                "groupBys": [],
+                "havings": []
+            },
+            "timeframe": {
+                "to": to_timestamp,
+                "from": from_timestamp
+            }
+        });
+
+        let response = self
+            .client
+            .post(url)
+            .json(&query_body)
+            .send()
+            .await
+            .into_diagnostic()?;
+
+        // If there's an error, just return empty results
+        if !response.status().is_success() {
+            error!(
+                "Failed to query Cloudflare error logs for worker: {}, error: {:?}",
+                cf_worker_name,
+                response.json::<serde_json::Value>().await
+            );
+            return Ok(Vec::new());
+        }
+
+        let error_logs_response = response
+            .json::<CloudflareResponse<ErrorLogsResponse>>()
+            .await
+            .into_diagnostic()?;
+
+        // Convert the enourmous CF response into our error log groups
+        let error_log_groups = error_logs_response.result.into();
+
+        Ok(error_log_groups)
+    }
+
     fn base_url() -> &'static Url {
         URL.get_or_init(init_url)
     }

src/adapters/cloudflare/mod.rs

@@ -47,6 +47,7 @@ impl Monitor for CloudflareMonitor {
 
     fn get_config(&self) -> MonitorConfig {
         MonitorConfig::CloudflareWorkersObservability {
+            api_token: self.client.api_token().clone(),
             account_id: self.client.account_id().clone(),
             worker_name: self.client.worker_name().clone(),
         }

src/adapters/monitors/cloudflare.rs

@@ -10,7 +10,7 @@ pub struct RunSubcommand {
     workspace: Option<String>,
     #[arg(short, long, env = "MULTI_APPLICATION")]
     application: Option<String>,
-    #[arg(long, short = 'o', default_value = Some(MULTITOOL_ORIGIN))]
+    #[arg(long, short = 'o', default_value = Some(MULTITOOL_ORIGIN), env = "ORIGIN")]
     origin: Option<String>,
 
     ///Cloudflare config

src/config/run/mod.rs

@@ -2,12 +2,13 @@ use async_trait::async_trait;
 use bon::bon;
 use miette::{Report, Result};
 use tokio_graceful_shutdown::{IntoSubsystem, SubsystemBuilder, SubsystemHandle};
-use tracing::{debug, trace};
+use tracing::{debug, error, trace};
 
 use crate::adapters::{BackendClient, BoxedIngress, BoxedMonitor, BoxedPlatform, RolloutMetadata};
 use crate::subsystems::PLATFORM_SUBSYSTEM_NAME;
 use crate::{IngressSubsystem, PlatformSubsystem};
 
+use crate::subsystems::{ERROR_LOGS_SUBSYSTEM_NAME, ErrorLogsController};
 use monitor::{MONITOR_CONTROLLER_SUBSYSTEM_NAME, MonitorController};
 
 use super::{INGRESS_SUBSYSTEM_NAME, RELAY_SUBSYSTEM_NAME, RelaySubsystem};
@@ -62,22 +63,33 @@ impl IntoSubsystem<Report> for ControllerSubsystem {
         let platform_subsystem = PlatformSubsystem::new(self.platform);
         let platform_handle = platform_subsystem.handle();
 
+        // Extract monitor config before moving the monitor so we can use it
+        // in the ErrorLogsController.
+        let monitor_config = self.monitor.get_config();
+
         let mut monitor_controller = MonitorController::builder().monitor(self.monitor).build();
         let observation_stream = monitor_controller.stream()?;
 
         let baseline_sender = monitor_controller.get_baseline_sender();
         let canary_sender = monitor_controller.get_canary_sender();
 
         let relay_subsystem = RelaySubsystem::builder()
-            .backend(self.backend)
+            .backend(self.backend.clone())
             .observations(observation_stream)
             .platform(platform_handle)
             .ingress(ingress_handle)
-            .meta(self.meta)
+            .meta(self.meta.clone())
             .baseline_sender(baseline_sender)
             .canary_sender(canary_sender)
             .build();
 
+        let error_logs_controller = ErrorLogsController::builder()
+            .metadata(self.meta)
+            .monitor(monitor_config)
+            .backend(self.backend)
+            .build()
+            .map_err(|e| miette::miette!("Failed to create ErrorLogsController: {}", e))?;
+
         // • Start the ingress subsystem.
         subsys.start(SubsystemBuilder::new(
             INGRESS_SUBSYSTEM_NAME,
@@ -102,6 +114,11 @@ impl IntoSubsystem<Report> for ControllerSubsystem {
             relay_subsystem.into_subsystem(),
         ));
 
+        subsys.start(SubsystemBuilder::new(
+            ERROR_LOGS_SUBSYSTEM_NAME,
+            error_logs_controller.into_subsystem(),
+        ));
+
         subsys.wait_for_children().await;
         Ok(())
     }