feat(): add more worker metrics

j-white · j-white · commit 4a99d0e661fc · 2026-01-28T08:35:04.000-05:00
diff --git a/gql/workers_query.graphql b/gql/workers_query.graphql
@@ -15,6 +15,8 @@ query GetWorkersAnalyticsQuery($accountTag: string!, $datetimeStart: Time!, $dat
           requests
           errors
           duration
+          wallTime
+          subrequests
         }
 
         quantiles {
diff --git a/src/gql.rs b/src/gql.rs
@@ -4,7 +4,6 @@ use graphql_client::{GraphQLQuery, Response};
 use opentelemetry_proto::tonic::metrics::v1::Metric;
 use prometheus::{CounterVec, GaugeVec, Opts, Registry};
 use crate::metrics::prometheus_registry_to_opentelemetry_metrics;
-use web_time::SystemTime;
 use worker::console_log;
 
 // The paths are relative to the directory where your `Cargo.toml` is located.
@@ -72,9 +71,11 @@ type float64 = f64;
 #[allow(non_camel_case_types)]
 type uint16 = u16;
 
-pub async fn do_get_workers_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_workers_analytics_query::Variables) -> Result<Vec<Metric>, Box<dyn Error>> {
+pub async fn do_get_workers_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_workers_analytics_query::Variables, debug_logging: bool, fallback_timestamp_nanos: u64) -> Result<Vec<Metric>, Box<dyn Error>> {
     let request_body = GetWorkersAnalyticsQuery::build_query(variables);
-    console_log!("[Workers] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    if debug_logging {
+        console_log!("[Workers] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    }
     let client = reqwest::Client::new();
     let res = client.post(cloudflare_api_url)
         .bearer_auth(cloudflare_api_key)
@@ -86,7 +87,9 @@ pub async fn do_get_workers_analytics_query(cloudflare_api_url: &String, cloudfl
     }
 
     let response_text = res.text().await?;
-    console_log!("[Workers] GraphQL response: {}", response_text);
+    if debug_logging {
+        console_log!("[Workers] GraphQL response: {}", response_text);
+    }
     let response_body: Response<get_workers_analytics_query::ResponseData> = serde_json::from_str(&response_text)?;
     if response_body.errors.is_some() {
         console_log!("[Workers] GraphQL query failed: {:?}", response_body.errors);
@@ -111,6 +114,14 @@ pub async fn do_get_workers_analytics_query(cloudflare_api_url: &String, cloudfl
     let worker_duration = GaugeVec::new(worker_duration_opts, &["script_name", "quantile"]).unwrap();
     registry.register(Box::new(worker_duration.clone())).unwrap();
 
+    let worker_wall_time_opts = Opts::new("cloudflare_worker_wall_time", "Sum of wall time - microseconds");
+    let worker_wall_time = CounterVec::new(worker_wall_time_opts, &["script_name"]).unwrap();
+    registry.register(Box::new(worker_wall_time.clone())).unwrap();
+
+    let worker_subrequests_opts = Opts::new("cloudflare_worker_subrequests", "Sum of subrequests");
+    let worker_subrequests = CounterVec::new(worker_subrequests_opts, &["script_name"]).unwrap();
+    registry.register(Box::new(worker_subrequests.clone())).unwrap();
+
     let mut last_datetime: Option<Time> = None;
     for account in response_data.viewer.unwrap().accounts.iter() {
         for worker in account.workers_invocations_adaptive.iter() {
@@ -122,6 +133,8 @@ pub async fn do_get_workers_analytics_query(cloudflare_api_url: &String, cloudfl
 
             worker_requests.with_label_values(&[script_name.as_str()]).inc_by(sum.requests as f64);
             worker_errors.with_label_values(&[script_name.as_str()]).inc_by(sum.errors as f64);
+            worker_wall_time.with_label_values(&[script_name.as_str()]).inc_by(sum.wall_time as f64);
+            worker_subrequests.with_label_values(&[script_name.as_str()]).inc_by(sum.subrequests as f64);
             worker_cpu_time.with_label_values(&[script_name.as_str(), "P50"]).set(quantiles.cpu_time_p50 as f64);
             worker_cpu_time.with_label_values(&[script_name.as_str(), "P75"]).set(quantiles.cpu_time_p75 as f64);
             worker_cpu_time.with_label_values(&[script_name.as_str(), "P99"]).set(quantiles.cpu_time_p99 as f64);
@@ -136,16 +149,16 @@ pub async fn do_get_workers_analytics_query(cloudflare_api_url: &String, cloudfl
     let timestamp_nanos: u64 = last_datetime.map(|datetime| {
         let datetime: NaiveDateTime = NaiveDateTime::parse_from_str(&datetime, "%+").unwrap();
         datetime.and_utc().timestamp_nanos_opt().unwrap_or(0) as u64
-    }).unwrap_or_else(|| {
-        systemtime_to_nanos(SystemTime::now())
-    });
+    }).unwrap_or(fallback_timestamp_nanos);
 
     Ok(prometheus_registry_to_opentelemetry_metrics(registry, timestamp_nanos))
 }
 
-pub async fn do_get_d1_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_d1_analytics_query::Variables) -> Result<Vec<Metric>, Box<dyn Error>> {
+pub async fn do_get_d1_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_d1_analytics_query::Variables, debug_logging: bool, fallback_timestamp_nanos: u64) -> Result<Vec<Metric>, Box<dyn Error>> {
     let request_body = GetD1AnalyticsQuery::build_query(variables);
-    console_log!("[D1] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    if debug_logging {
+        console_log!("[D1] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    }
     let client = reqwest::Client::new();
     let res = client.post(cloudflare_api_url)
         .bearer_auth(cloudflare_api_key)
@@ -157,7 +170,9 @@ pub async fn do_get_d1_analytics_query(cloudflare_api_url: &String, cloudflare_a
     }
 
     let response_text = res.text().await?;
-    console_log!("[D1] GraphQL response: {}", response_text);
+    if debug_logging {
+        console_log!("[D1] GraphQL response: {}", response_text);
+    }
     let response_body: Response<get_d1_analytics_query::ResponseData> = serde_json::from_str(&response_text)?;
     if response_body.errors.is_some() {
         console_log!("[D1] GraphQL query failed: {:?}", response_body.errors);
@@ -214,16 +229,16 @@ pub async fn do_get_d1_analytics_query(cloudflare_api_url: &String, cloudflare_a
     let timestamp_nanos: u64 = last_datetime.map(|datetime| {
         let datetime: NaiveDateTime = NaiveDateTime::parse_from_str(&datetime, "%+").unwrap();
         datetime.and_utc().timestamp_nanos_opt().unwrap_or(0) as u64
-    }).unwrap_or_else(|| {
-        systemtime_to_nanos(SystemTime::now())
-    });
+    }).unwrap_or(fallback_timestamp_nanos);
 
     Ok(prometheus_registry_to_opentelemetry_metrics(registry, timestamp_nanos))
 }
 
-pub async fn do_get_durableobjects_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_durable_objects_analytics_query::Variables) -> Result<Vec<Metric>, Box<dyn Error>> {
+pub async fn do_get_durableobjects_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_durable_objects_analytics_query::Variables, debug_logging: bool, fallback_timestamp_nanos: u64) -> Result<Vec<Metric>, Box<dyn Error>> {
     let request_body = GetDurableObjectsAnalyticsQuery::build_query(variables);
-    console_log!("[DurableObjects] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    if debug_logging {
+        console_log!("[DurableObjects] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    }
     let client = reqwest::Client::new();
     let res = client.post(cloudflare_api_url)
         .bearer_auth(cloudflare_api_key)
@@ -235,7 +250,9 @@ pub async fn do_get_durableobjects_analytics_query(cloudflare_api_url: &String,
     }
 
     let response_text = res.text().await?;
-    console_log!("[DurableObjects] GraphQL response: {}", response_text);
+    if debug_logging {
+        console_log!("[DurableObjects] GraphQL response: {}", response_text);
+    }
     let response_body: Response<get_durable_objects_analytics_query::ResponseData> = serde_json::from_str(&response_text)?;
     if response_body.errors.is_some() {
         console_log!("[DurableObjects] GraphQL query failed: {:?}", response_body.errors);
@@ -291,16 +308,16 @@ pub async fn do_get_durableobjects_analytics_query(cloudflare_api_url: &String,
     let timestamp_nanos: u64 = last_datetime.map(|datetime| {
         let datetime: NaiveDateTime = NaiveDateTime::parse_from_str(&datetime, "%+").unwrap();
         datetime.and_utc().timestamp_nanos_opt().unwrap_or(0) as u64
-    }).unwrap_or_else(|| {
-        systemtime_to_nanos(SystemTime::now())
-    });
+    }).unwrap_or(fallback_timestamp_nanos);
 
     Ok(prometheus_registry_to_opentelemetry_metrics(registry, timestamp_nanos))
 }
 
-pub async fn do_get_queue_backlog_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_queue_backlog_analytics_query::Variables) -> Result<Vec<Metric>, Box<dyn Error>> {
+pub async fn do_get_queue_backlog_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_queue_backlog_analytics_query::Variables, debug_logging: bool, fallback_timestamp_nanos: u64) -> Result<Vec<Metric>, Box<dyn Error>> {
     let request_body = GetQueueBacklogAnalyticsQuery::build_query(variables);
-    console_log!("[QueueBacklog] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    if debug_logging {
+        console_log!("[QueueBacklog] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    }
     let client = reqwest::Client::new();
     let res = client.post(cloudflare_api_url)
         .bearer_auth(cloudflare_api_key)
@@ -312,7 +329,9 @@ pub async fn do_get_queue_backlog_analytics_query(cloudflare_api_url: &String, c
     }
 
     let response_text = res.text().await?;
-    console_log!("[QueueBacklog] GraphQL response: {}", response_text);
+    if debug_logging {
+        console_log!("[QueueBacklog] GraphQL response: {}", response_text);
+    }
     let response_body: Response<get_queue_backlog_analytics_query::ResponseData> = serde_json::from_str(&response_text)?;
     if response_body.errors.is_some() {
         console_log!("[QueueBacklog] GraphQL query failed: {:?}", response_body.errors);
@@ -350,16 +369,16 @@ pub async fn do_get_queue_backlog_analytics_query(cloudflare_api_url: &String, c
     let timestamp_nanos: u64 = last_datetime.map(|datetime| {
         let datetime: NaiveDateTime = NaiveDateTime::parse_from_str(&datetime, "%+").unwrap();
         datetime.and_utc().timestamp_nanos_opt().unwrap_or(0) as u64
-    }).unwrap_or_else(|| {
-        systemtime_to_nanos(SystemTime::now())
-    });
+    }).unwrap_or(fallback_timestamp_nanos);
 
     Ok(prometheus_registry_to_opentelemetry_metrics(registry, timestamp_nanos))
 }
 
-pub async fn do_get_queue_operations_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_queue_operations_analytics_query::Variables) -> Result<Vec<Metric>, Box<dyn Error>> {
+pub async fn do_get_queue_operations_analytics_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_queue_operations_analytics_query::Variables, debug_logging: bool, fallback_timestamp_nanos: u64) -> Result<Vec<Metric>, Box<dyn Error>> {
     let request_body = GetQueueOperationsAnalyticsQuery::build_query(variables);
-    console_log!("[QueueOperations] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    if debug_logging {
+        console_log!("[QueueOperations] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    }
     let client = reqwest::Client::new();
     let res = client.post(cloudflare_api_url)
         .bearer_auth(cloudflare_api_key)
@@ -371,7 +390,9 @@ pub async fn do_get_queue_operations_analytics_query(cloudflare_api_url: &String
     }
 
     let response_text = res.text().await?;
-    console_log!("[QueueOperations] GraphQL response: {}", response_text);
+    if debug_logging {
+        console_log!("[QueueOperations] GraphQL response: {}", response_text);
+    }
     let response_body: Response<get_queue_operations_analytics_query::ResponseData> = serde_json::from_str(&response_text)?;
     if response_body.errors.is_some() {
         console_log!("[QueueOperations] GraphQL query failed: {:?}", response_body.errors);
@@ -424,16 +445,16 @@ pub async fn do_get_queue_operations_analytics_query(cloudflare_api_url: &String
     let timestamp_nanos: u64 = last_datetime.map(|datetime| {
         let datetime: NaiveDateTime = NaiveDateTime::parse_from_str(&datetime, "%+").unwrap();
         datetime.and_utc().timestamp_nanos_opt().unwrap_or(0) as u64
-    }).unwrap_or_else(|| {
-        systemtime_to_nanos(SystemTime::now())
-    });
+    }).unwrap_or(fallback_timestamp_nanos);
 
     Ok(prometheus_registry_to_opentelemetry_metrics(registry, timestamp_nanos))
 }
 
-pub async fn do_get_zone_http_requests_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_zone_http_requests_query::Variables) -> Result<Vec<Metric>, Box<dyn Error>> {
+pub async fn do_get_zone_http_requests_query(cloudflare_api_url: &String, cloudflare_api_key: &String, variables: get_zone_http_requests_query::Variables, debug_logging: bool, fallback_timestamp_nanos: u64) -> Result<Vec<Metric>, Box<dyn Error>> {
     let request_body = GetZoneHttpRequestsQuery::build_query(variables);
-    console_log!("[ZoneHttpRequests] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    if debug_logging {
+        console_log!("[ZoneHttpRequests] GraphQL request: {}", serde_json::to_string_pretty(&request_body).unwrap_or_default());
+    }
     let client = reqwest::Client::new();
     let res = client.post(cloudflare_api_url)
         .bearer_auth(cloudflare_api_key)
@@ -445,7 +466,9 @@ pub async fn do_get_zone_http_requests_query(cloudflare_api_url: &String, cloudf
     }
 
     let response_text = res.text().await?;
-    console_log!("[ZoneHttpRequests] GraphQL response: {}", response_text);
+    if debug_logging {
+        console_log!("[ZoneHttpRequests] GraphQL response: {}", response_text);
+    }
     let response_body: Response<get_zone_http_requests_query::ResponseData> = serde_json::from_str(&response_text)?;
     if response_body.errors.is_some() {
         console_log!("[ZoneHttpRequests] GraphQL query failed: {:?}", response_body.errors);
@@ -487,14 +510,8 @@ pub async fn do_get_zone_http_requests_query(cloudflare_api_url: &String, cloudf
     let timestamp_nanos: u64 = last_datetime.map(|datetime| {
         let datetime: NaiveDateTime = NaiveDateTime::parse_from_str(&datetime, "%+").unwrap();
         datetime.and_utc().timestamp_nanos_opt().unwrap_or(0) as u64
-    }).unwrap_or_else(|| {
-        systemtime_to_nanos(SystemTime::now())
-    });
+    }).unwrap_or(fallback_timestamp_nanos);
 
     Ok(prometheus_registry_to_opentelemetry_metrics(registry, timestamp_nanos))
 }
 
-fn systemtime_to_nanos(time: web_time::SystemTime) -> u64 {
-    let duration = time.duration_since(web_time::SystemTime::UNIX_EPOCH).unwrap();
-    duration.as_nanos() as u64
-}
diff --git a/src/lib.rs b/src/lib.rs
@@ -12,6 +12,16 @@ use crate::gql::{get_workers_analytics_query, do_get_workers_analytics_query, do
 mod gql;
 mod metrics;
 
+const DEFAULT_SCRAPE_DELAY_SECONDS: i64 = 300;
+
+fn get_scrape_delay_seconds(env: &Env) -> i64 {
+    env.var("SCRAPE_DELAY")
+        .ok()
+        .and_then(|val| val.to_string().parse::<i64>().ok())
+        .filter(|val| *val >= 0)
+        .unwrap_or(DEFAULT_SCRAPE_DELAY_SECONDS)
+}
+
 #[worker::send]
 pub async fn do_fetch(
     url: String,
@@ -60,9 +70,15 @@ async fn do_trigger(env: Env) -> Result<()> {
     let cloudflare_api_url = env.var("CLOUDFLARE_API_URL")?.to_string();
     let cloudflare_api_key = env.var("CLOUDFLARE_API_KEY")?.to_string();
     let cloudflare_account_id = env.var("CLOUDFLARE_ACCOUNT_ID")?.to_string();
+    let debug_logging: bool = match env.var("DEBUG_LOGGING") {
+        Ok(val) => matches!(val.to_string().to_lowercase().as_str(), "true" | "1" | "yes"),
+        Err(_) => false,
+    };
 
-    let end = chrono::Utc::now().round_subsecs(0);
+    let scrape_delay_seconds = get_scrape_delay_seconds(&env);
+    let end = (chrono::Utc::now() - chrono::Duration::seconds(scrape_delay_seconds)).round_subsecs(0);
     let start = (end - chrono::Duration::minutes(1)).round_subsecs(0);
+    let fallback_timestamp_nanos = end.timestamp_nanos_opt().unwrap_or(0) as u64;
 
     console_log!("Fetching!");
     let mut all_metrics = Vec::new();
@@ -72,7 +88,7 @@ async fn do_trigger(env: Env) -> Result<()> {
         datetime_start: start.to_rfc3339(),
         datetime_end: end.to_rfc3339(),
         limit: 9999,
-    }).await;
+    }, debug_logging, fallback_timestamp_nanos).await;
     match result {
         Ok(metrics) => {
             for metric in metrics {
@@ -90,7 +106,7 @@ async fn do_trigger(env: Env) -> Result<()> {
         datetime_start: start.to_rfc3339(),
         datetime_end: end.to_rfc3339(),
         limit: 9999,
-    }).await;
+    }, debug_logging, fallback_timestamp_nanos).await;
     match result {
         Ok(metrics) => {
             for metric in metrics {
@@ -108,7 +124,7 @@ async fn do_trigger(env: Env) -> Result<()> {
         datetime_start: start.to_rfc3339(),
         datetime_end: end.to_rfc3339(),
         limit: 9999,
-    }).await;
+    }, debug_logging, fallback_timestamp_nanos).await;
     match result {
         Ok(metrics) => {
             for metric in metrics {
@@ -126,7 +142,7 @@ async fn do_trigger(env: Env) -> Result<()> {
         datetime_start: start.to_rfc3339(),
         datetime_end: end.to_rfc3339(),
         limit: 9999,
-    }).await;
+    }, debug_logging, fallback_timestamp_nanos).await;
     match result {
         Ok(metrics) => {
             for metric in metrics {
@@ -144,7 +160,7 @@ async fn do_trigger(env: Env) -> Result<()> {
         datetime_start: start.to_rfc3339(),
         datetime_end: end.to_rfc3339(),
         limit: 9999,
-    }).await;
+    }, debug_logging, fallback_timestamp_nanos).await;
     match result {
         Ok(metrics) => {
             for metric in metrics {
@@ -171,7 +187,7 @@ async fn do_trigger(env: Env) -> Result<()> {
                 datetime_start: start.to_rfc3339(),
                 datetime_end: end.to_rfc3339(),
                 limit: 9999,
-            }).await;
+            }, debug_logging, fallback_timestamp_nanos).await;
             match result {
                 Ok(metrics) => {
                     for metric in metrics {
@@ -188,10 +204,10 @@ async fn do_trigger(env: Env) -> Result<()> {
 
     console_log!("Done fetching!");
 
-    do_push_metrics(env, all_metrics).await
+    do_push_metrics(env, all_metrics, debug_logging).await
 }
 
-async fn do_push_metrics(env: Env, metrics: Vec<Metric>) -> Result<()> {
+async fn do_push_metrics(env: Env, metrics: Vec<Metric>, debug_logging: bool) -> Result<()> {
     let metrics_url = env.var("METRICS_URL")?.to_string();
     let otlp_headers = match env.var("OTLP_HEADERS") {
         Ok(val) => val.to_string(),
@@ -225,8 +241,10 @@ async fn do_push_metrics(env: Env, metrics: Vec<Metric>) -> Result<()> {
     };
 
     // Log the OTLP payload as JSON for debugging
-    let metrics_json_for_logging = serde_json::to_string_pretty(&export_request).unwrap();
-    console_log!("OTLP metrics payload:\n{}", metrics_json_for_logging);
+    if debug_logging {
+        let metrics_json_for_logging = serde_json::to_string_pretty(&export_request).unwrap();
+        console_log!("OTLP metrics payload:\n{}", metrics_json_for_logging);
+    }
 
     let js_value: JsValue;
     let content_type: String;

Original file line number	Diff line number	Diff line change
`@@ -15,6 +15,8 @@ query GetWorkersAnalyticsQuery($accountTag: string!, $datetimeStart: Time!, $dat`
`15`	`15`	`requests`
`16`	`16`	`errors`
`17`	`17`	`duration`
	`18`	`+ wallTime`
	`19`	`+ subrequests`
`18`	`20`	`}`
`19`	`21`
`20`	`22`	`quantiles {`