runtimes/core: Add requests counter metric

Author: fredr

Cargo.lock

@@ -10,6 +10,7 @@ rttrace = []
 [dependencies]
 pingora = { version = "0.4", features = ["lb", "openssl"] }
 anyhow = "1.0.76"
+async-trait = "0.1"
 base64 = "0.21.5"
 gjson = "0.8.1"
 prost = "0.12.3"
@@ -123,6 +124,12 @@ email_address = "0.2.9"
 cookie = "0.18.1"
 malachite = "0.6.1"
 byteorder = "1.5.0"
+metrics = "0.24.2"
+metrics-util = { version = "0.20.0", features = ["registry", "storage"] }
+dashmap = "6.1.0"
+google-cloud-monitoring-v3 = "1.0.0"
+google-cloud-api = "1.0.0"
+google-cloud-wkt = "1.0.0"
 
 [build-dependencies]
 prost-build = "0.12.3"

runtimes/core/Cargo.toml

@@ -2,6 +2,7 @@ use crate::api::auth::{AuthHandler, AuthPayload, AuthRequest, AuthResponse};
 use crate::api::schema::encoding::Schema;
 use crate::api::{APIResult, HandlerResponse, HandlerResponseInner, PValues};
 use crate::log::LogFromRust;
+use crate::metrics::Counter;
 use crate::model::{AuthRequestData, RequestData};
 use crate::trace::Tracer;
 use crate::{api, model, EndpointName};
@@ -14,6 +15,7 @@ pub struct LocalAuthHandler {
     pub schema: Schema,
     pub handler: RwLock<Option<Arc<dyn api::TypedHandler>>>,
     pub tracer: Tracer,
+    pub requests_total: Counter,
 }
 
 impl LocalAuthHandler {
@@ -152,7 +154,9 @@ impl AuthHandler for LocalAuthHandler {
                             user_id: auth_uid.clone(),
                         })),
                     };
+
                     self.tracer.request_span_end(&model_resp, false);
+                    self.requests_total.increment_with([("code", "ok")]);
                     Ok(AuthResponse::Authenticated {
                         auth_uid,
                         auth_data,
@@ -165,6 +169,8 @@ impl AuthHandler for LocalAuthHandler {
                         data: model::ResponseData::Auth(Err(e.clone())),
                     };
                     self.tracer.request_span_end(&model_resp, false);
+                    self.requests_total
+                        .increment_with([("code", e.code.to_string())]);
                     Err(e)
                 }
             }

runtimes/core/src/api/auth/local.rs

@@ -23,6 +23,7 @@ use crate::api::{jsonschema, schema, ErrCode, Error};
 use crate::encore::parser::meta::v1::rpc;
 use crate::encore::parser::meta::v1::{self as meta, selector};
 use crate::log::LogFromRust;
+use crate::metrics::Counter;
 use crate::model::StreamDirection;
 use crate::names::EndpointName;
 use crate::trace;
@@ -406,6 +407,7 @@ pub(super) struct EndpointHandler {
     pub endpoint: Arc<Endpoint>,
     pub handler: Arc<dyn BoxedHandler>,
     pub shared: Arc<SharedEndpointData>,
+    pub requests_total: Counter,
 }
 
 #[derive(Debug)]
@@ -427,6 +429,7 @@ impl Clone for EndpointHandler {
             endpoint: self.endpoint.clone(),
             handler: self.handler.clone(),
             shared: self.shared.clone(),
+            requests_total: self.requests_total.clone(),
         }
     }
 }
@@ -601,7 +604,6 @@ impl EndpointHandler {
             let duration = tokio::time::Instant::now().duration_since(request.start);
 
             // If we had a request failure, log that separately.
-
             if let ResponseData::Typed(Err(err)) = &resp {
                 logger.error(Some(&request), "request failed", Some(err), {
                     let mut fields = crate::log::Fields::new();
@@ -629,6 +631,12 @@ impl EndpointHandler {
                 });
             }
 
+            let code = match &resp {
+                ResponseData::Typed(Ok(_)) => "ok".to_string(),
+                ResponseData::Typed(Err(err)) => err.code.to_string(),
+                ResponseData::Raw(resp) => ErrCode::from(resp.status()).to_string(),
+            };
+
             logger.info(Some(&request), "request completed", {
                 let mut fields = crate::log::Fields::new();
                 let dur_ms = (duration.as_secs() as f64 * 1000f64)
@@ -644,13 +652,7 @@ impl EndpointHandler {
                     )),
                 );
 
-                let code = match &resp {
-                    ResponseData::Typed(Ok(_)) => "ok".to_string(),
-                    ResponseData::Typed(Err(err)) => err.code.to_string(),
-                    ResponseData::Raw(resp) => ErrCode::from(resp.status()).to_string(),
-                };
-
-                fields.insert("code".into(), serde_json::Value::String(code));
+                fields.insert("code".into(), serde_json::Value::String(code.clone()));
                 Some(fields)
             });
 
@@ -685,6 +687,7 @@ impl EndpointHandler {
                     }),
                 };
                 self.shared.tracer.request_span_end(&model_resp, sensitive);
+                self.requests_total.increment_with([("code", code)]);
             }
 
             if let Ok(val) = HeaderValue::from_str(request.span.0.serialize_encore().as_str()) {

runtimes/core/src/api/endpoint.rs

@@ -18,6 +18,7 @@ use crate::api::{
 };
 use crate::encore::parser::meta::v1 as meta;
 use crate::encore::runtime::v1 as runtime;
+use crate::metrics::requests_total_counter;
 use crate::trace::Tracer;
 use crate::{api, model, pubsub, secrets, EncoreName, EndpointName, Hosted};
 
@@ -283,6 +284,7 @@ fn build_auth_handler(
     // let is_local = hosted_services.contains(&explicit.service_name);
     let is_local = true;
     let name = EndpointName::new(explicit.service_name.clone(), auth.name.clone());
+    let requests_total = requests_total_counter(&explicit.service_name, &auth.name);
 
     let auth_data = registry.schema(auth_data_schema_idx);
     let auth_handler = if is_local {
@@ -294,6 +296,7 @@ fn build_auth_handler(
                 schema,
                 handler: Default::default(),
                 tracer,
+                requests_total,
             },
         )?
     } else {

runtimes/core/src/api/manager.rs

@@ -12,6 +12,7 @@ use crate::api::static_assets::StaticAssetsHandler;
 use crate::api::{self, ToResponse};
 use crate::api::{paths, reqauth, schema, BoxedHandler, EndpointMap};
 use crate::encore::parser::meta::v1 as meta;
+use crate::metrics::requests_total_counter;
 use crate::names::EndpointName;
 use crate::trace;
 
@@ -92,11 +93,14 @@ impl Server {
                             // For static asset routes, configure the static asset handler directly.
                             // There's no need to defer it for dynamic runtime registration.
                             let static_handler = StaticAssetsHandler::new(assets);
+                            let requests_total =
+                                requests_total_counter(&ep.name.service(), &ep.name.endpoint());
 
                             let handler = EndpointHandler {
                                 endpoint: ep.clone(),
                                 handler: Arc::new(static_handler),
                                 shared: shared.clone(),
+                                requests_total,
                             };
                             server_handler.set(handler);
                         }
@@ -153,11 +157,14 @@ impl Server {
             None => Ok(()), // anyhow::bail!("no handler found for endpoint: {}", endpoint_name),
             Some(h) => {
                 let endpoint = self.endpoints.get(&endpoint_name).unwrap().to_owned();
+                let requests_total =
+                    requests_total_counter(&endpoint.name.service(), &endpoint.name.endpoint());
 
                 let handler = EndpointHandler {
                     endpoint,
                     handler,
                     shared: self.shared.clone(),
+                    requests_total,
                 };
 
                 h.add(handler);

runtimes/core/src/api/server.rs

@@ -24,6 +24,8 @@ pub mod error;
 pub mod infracfg;
 pub mod log;
 pub mod meta;
+pub mod metadata;
+pub mod metrics;
 pub mod model;
 mod names;
 pub mod objects;
@@ -209,6 +211,7 @@ pub struct Runtime {
     app_meta: meta::AppMeta,
     compute: ComputeConfig,
     runtime: tokio::runtime::Runtime,
+    metrics: metrics::Manager,
 }
 
 impl Runtime {
@@ -238,6 +241,7 @@ impl Runtime {
 
         let mut deployment = cfg.deployment.take().unwrap_or_default();
         let service_discovery = deployment.service_discovery.take().unwrap_or_default();
+        let observability = deployment.observability.take().unwrap_or_default();
 
         let http_client = reqwest::Client::builder()
             .build()
@@ -250,11 +254,18 @@ impl Runtime {
         );
         let platform_validator = Arc::new(platform_validator);
 
+        // Initialize metrics manager from runtime config
+        let metrics_manager = metrics::Manager::from_runtime_config(
+            &observability,
+            &environment,
+            &http_client,
+            tokio_rt.handle().clone(),
+        );
+
         // Set up observability.
         let disable_tracing =
             testing || std::env::var("ENCORE_NOTRACE").is_ok_and(|v| !v.is_empty());
         let tracer = if !disable_tracing {
-            let observability = deployment.observability.take().unwrap_or_default();
             let trace_endpoint = observability
                 .tracing
                 .into_iter()
@@ -411,6 +422,7 @@ impl Runtime {
             app_meta,
             compute,
             runtime: tokio_rt,
+            metrics: metrics_manager,
         })
     }
 
@@ -444,6 +456,11 @@ impl Runtime {
         &self.api
     }
 
+    #[inline]
+    pub fn metrics(&self) -> &metrics::Manager {
+        &self.metrics
+    }
+
     #[inline]
     pub fn endpoints(&self) -> &api::EndpointMap {
         self.api.endpoints()