feat: instrument request counter metrics

Author: ymgyt

.dev/otelcol-config.yaml

@@ -3,7 +3,6 @@ receivers:
     protocols:
       grpc:
         endpoint: localhost:4317
-
 processors:
   resource/loki:
     attributes:
@@ -13,7 +12,6 @@ processors:
       - action: insert
         key: loki.format
         value: logfmt # logfmt | json   raw
-
 exporters:
   loki:
     endpoint: ${env:LOKI_ENDPOINT}
@@ -22,14 +20,14 @@ exporters:
       job: false
       instance: false
       level: false
-      
   debug:
     verbosity: detailed
-
 service:
   pipelines:
-    logs:
+    # logs:
+    #   receivers: [otlp]
+    #   processors: [resource/loki]
+    #   exporters: [debug]
+    metrics:
       receivers: [otlp]
-      processors: [resource/loki]
-      exporters: [debug,loki]
-    
+      exporters: [debug]

crates/synd_api/src/main.rs

@@ -6,9 +6,10 @@ use synd_api::{
     config,
     dependency::Dependency,
     repository::kvsd::ConnectKvsdFailed,
-    serve::listen_and_serve,
+    serve::{layer::request_metrics::METRICS_TARGET, listen_and_serve},
     shutdown::Shutdown,
 };
+use tracing_subscriber::filter::filter_fn;
 
 fn init_tracing() -> Option<OpenTelemetryGuard> {
     use synd_o11y::{
@@ -35,6 +36,8 @@ fn init_tracing() -> Option<OpenTelemetryGuard> {
                 let resource =
                     synd_o11y::opentelemetry::resource(config::NAME, config::VERSION, "local");
 
+                tracing::info!(endpoint, ?resource, "Export opentelemetry signals");
+
                 let trace_layer = otel_trace::layer(resource.clone());
                 let log_layer = otel_log::layer(endpoint, resource.clone());
                 let metrics_layer = otel_metrics::layer(resource);
@@ -55,6 +58,7 @@ fn init_tracing() -> Option<OpenTelemetryGuard> {
                 .with_file(show_src)
                 .with_line_number(show_src)
                 .with_target(show_target)
+                .with_filter(filter_fn(|metadata| metadata.target() != METRICS_TARGET))
                 .and_then(opentelemetry_layers)
                 .with_filter(
                     EnvFilter::try_from_default_env()

crates/synd_api/src/serve/layer/mod.rs

@@ -1,2 +1,3 @@
 pub mod authenticate;
+pub mod request_metrics;
 pub mod trace;

crates/synd_api/src/serve/layer/request_metrics.rs

@@ -0,0 +1,63 @@
+use std::{
+    convert::Infallible,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use axum::{extract::Request, response::Response};
+use futures_util::Future;
+use tower::{Layer, Service};
+use tracing::Level;
+
+pub const METRICS_TARGET: &str = "metrics";
+
+#[derive(Clone)]
+pub struct RequestMetricsLayer {}
+
+impl RequestMetricsLayer {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl<S> Layer<S> for RequestMetricsLayer {
+    type Service = RequestMetricsService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        RequestMetricsService { inner }
+    }
+}
+
+#[derive(Clone)]
+pub struct RequestMetricsService<S> {
+    inner: S,
+}
+
+impl<S> Service<Request> for RequestMetricsService<S>
+where
+    S: Service<Request, Response = Response, Error = Infallible> + Clone + Send + 'static,
+    S::Future: Send,
+{
+    type Response = Response;
+    type Error = Infallible;
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: Request) -> Self::Future {
+        let path = req.uri().path().to_owned();
+
+        let mut this = self.clone();
+        Box::pin(async move {
+            let response = this.inner.call(req).await.unwrap();
+            let status = response.status().as_u16();
+
+            tracing::event!(target: METRICS_TARGET, Level::INFO, monotonic_counter.request = 1, path, status);
+
+            Ok(response)
+        })
+    }
+}

crates/synd_api/src/serve/mod.rs

@@ -17,7 +17,7 @@ use crate::{
     config,
     dependency::Dependency,
     gql,
-    serve::layer::{authenticate, trace},
+    serve::layer::{authenticate, request_metrics::RequestMetricsLayer, trace},
     shutdown::Shutdown,
 };
 
@@ -68,7 +68,8 @@ pub async fn serve(
                 .layer(RequestBodyLimitLayer::new(2048))
                 .layer(CorsLayer::new()),
         )
-        .route("/healthcheck", get(probe::healthcheck));
+        .route("/healthcheck", get(probe::healthcheck))
+        .layer(RequestMetricsLayer::new());
 
     axum_server::from_tcp_rustls(listener.into_std()?, tls_config)
         .handle(shutdown.into_handle())

crates/synd_o11y/src/opentelemetry/resource.rs

@@ -1,7 +1,8 @@
 use opentelemetry::KeyValue;
 use opentelemetry_sdk::Resource;
-use opentelemetry_semantic_conventions::resource::{
-    DEPLOYMENT_ENVIRONMENT, SERVICE_NAME, SERVICE_NAMESPACE, SERVICE_VERSION,
+use opentelemetry_semantic_conventions::{
+    resource::{DEPLOYMENT_ENVIRONMENT, SERVICE_NAME, SERVICE_NAMESPACE, SERVICE_VERSION},
+    SCHEMA_URL,
 };
 use std::borrow::Cow;
 
@@ -10,7 +11,7 @@ pub fn resource(
     service_version: impl Into<Cow<'static, str>>,
     deployment_environment: impl Into<Cow<'static, str>>,
 ) -> Resource {
-    Resource::new(
+    Resource::from_schema_url(
         [
             (SERVICE_NAME, service_name.into()),
             (SERVICE_VERSION, service_version.into()),
@@ -19,5 +20,6 @@ pub fn resource(
         ]
         .into_iter()
         .map(|(key, value)| KeyValue::new(key, value)),
+        SCHEMA_URL,
     )
 }

crates/synd_o11y/src/tracing_subscriber/otel_metrics/mod.rs

@@ -17,7 +17,7 @@ fn init_meter_provider(resource: Resource) -> MeterProvider {
         .metrics(runtime::Tokio)
         .with_resource(resource)
         .with_exporter(opentelemetry_otlp::new_exporter().tonic())
-        .with_period(Duration::from_secs(60 * 3))
+        .with_period(Duration::from_secs(60))
         .build()
         .unwrap();
 

justfile

@@ -92,7 +92,7 @@ api *flags:
 
 # Run term
 term *flags:
-  cd crates/synd_term; cargo run -- --log /tmp/syndterm.log {{flags}}
+  cd crates/synd_term; cargo run -- --log /tmp/syndterm.log --endpoint https://localhost:5959/graphql {{flags}}
 
 # Run opentelemetry-collector-contrib
 @otelcol: