Plat 330/deliver metrics to otel (#36)

Author: AlexanderKlump

Cargo.lock

@@ -49,10 +49,6 @@ opentelemetry_sdk = { version = "0.31", default-features = false, features = [
     "logs",
     "metrics",
 ] }
-opentelemetry-stdout = { version = "0.31", default-features = false, features = [
-    "logs",
-    "metrics",
-] }
 prometheus = "0.14"
 s3s = "0.13.0-alpha.3"
 s3s-aws = "0.13.0-alpha.3"

Cargo.toml

@@ -5,7 +5,7 @@ WORKDIR /app
 COPY Cargo.toml Cargo.lock ./
 COPY src/ src/
 
-RUN cargo build --release --bin s3_cache
+RUN cargo build --bin s3_cache
 
 FROM ubuntu:24.04
 
@@ -14,7 +14,7 @@ RUN apt-get update && apt-get install -y \
     curl \
     && rm -rf /var/lib/apt/lists/*
 
-COPY --from=builder /app/target/release/s3_cache /usr/local/bin/s3_cache
+COPY --from=builder /app/target/debug/s3_cache /usr/local/bin/s3_cache
 
 RUN mkdir -p /metrics
 

dev/Dockerfile.s3-cache

@@ -33,17 +33,37 @@ services:
       CACHE_MAX_SIZE_BYTES: "1073741824"
       CACHE_MAX_OBJECT_SIZE_BYTES: "10485760"
       CACHE_TTL_SECONDS: "86400"
-      PROMETHEUS_TEXTFILE_DIR: /metrics
+      OTEL_GRPC_ENDPOINT_URL: "http://otel-collector:4317"
+      OTEL_EXPORT_METRICS: "true"
       RUST_LOG: info
     depends_on:
       minio:
         condition: service_healthy
+      otel-collector:
+        condition: service_started
     healthcheck:
       test: [ "CMD", "curl", "-f", "http://localhost:8080/health" ]
       interval: 5s
       timeout: 3s
       retries: 10
 
+  otel-collector:
+    image: otel/opentelemetry-collector-contrib:0.147.0
+    volumes:
+      - ./otel-collector-config.yml:/etc/otelcol-contrib/config.yaml:ro
+    ports:
+      - "127.0.0.1:4317:4317"
+      - "127.0.0.1:8889:8889"
+
+  prometheus:
+    image: prom/prometheus:v3.10.0
+    volumes:
+      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
+    ports:
+      - "127.0.0.1:9090:9090"
+    depends_on:
+      - otel-collector
+
   test-runner:
     image: amazon/aws-cli
     entrypoint: [ "/bin/bash", "/scripts/test.sh" ]

dev/docker-compose.yml

@@ -80,6 +80,14 @@ pub struct Config {
     #[arg(long, env = "OTEL_GRPC_ENDPOINT_URL")]
     pub otel_grpc_endpoint_url: Option<String>,
 
+    /// Export metrics via OTLP gRPC (requires otel_grpc_endpoint_url)
+    #[arg(long, env = "OTEL_EXPORT_METRICS", default_value_t = false, action = clap::ArgAction::Set)]
+    pub otel_export_metrics: bool,
+
+    /// Export logs via OTLP gRPC (requires otel_grpc_endpoint_url)
+    #[arg(long, env = "OTEL_EXPORT_LOGS", default_value_t = false, action = clap::ArgAction::Set)]
+    pub otel_export_logs: bool,
+
     /// Prometheus textfile collector directory
     #[arg(long, env = "PROMETHEUS_TEXTFILE_DIR")]
     pub prometheus_textfile_dir: Option<String>,
@@ -114,6 +122,14 @@ impl Config {
         if self.worker_threads == 0 {
             panic!("Invalid configuration: worker_threads must be greater than 0");
         }
+
+        if (self.otel_export_metrics || self.otel_export_logs)
+            && self.otel_grpc_endpoint_url.is_none()
+        {
+            panic!(
+                "Invalid configuration: otel_export_metrics and otel_export_logs require otel_grpc_endpoint_url to be set"
+            );
+        }
     }
 }
 
@@ -123,7 +139,8 @@ impl Display for Config {
             f,
             "Config{{ listen_addr: {}, upstream_endpoint: {}, upstream_region: {}, \
              cache_max_entries: {}, cache_max_size_bytes: {}, cache_ttl_seconds: {}, \
-             max_cacheable_object_size: {}, otel_grpc_endpoint_url: {:?}, cache_shards: {}, \
+             max_cacheable_object_size: {}, otel_grpc_endpoint_url: {:?}, \
+             otel_export_metrics: {}, otel_export_logs: {}, cache_shards: {}, \
              cache_dry_run: {}, worker_threads: {}, prometheus_textfile_dir: {:?} }}",
             self.listen_addr,
             self.upstream_endpoint,
@@ -133,6 +150,8 @@ impl Display for Config {
             self.cache_ttl_seconds,
             self.cache_max_object_size_bytes,
             self.otel_grpc_endpoint_url,
+            self.otel_export_metrics,
+            self.otel_export_logs,
             self.cache_shards,
             self.cache_dry_run,
             self.worker_threads,
@@ -165,6 +184,8 @@ mod tests {
             cache_ttl_seconds: 86_400,
             worker_threads: 4,
             otel_grpc_endpoint_url: None,
+            otel_export_metrics: false,
+            otel_export_logs: false,
             prometheus_textfile_dir: None,
         }
     }
@@ -210,4 +231,33 @@ mod tests {
         config.worker_threads = 0;
         config.validate();
     }
+
+    #[test]
+    #[should_panic(
+        expected = "otel_export_metrics and otel_export_logs require otel_grpc_endpoint_url"
+    )]
+    fn config_otel_export_metrics_without_endpoint() {
+        let mut config = minimal_config();
+        config.otel_export_metrics = true;
+        config.validate();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "otel_export_metrics and otel_export_logs require otel_grpc_endpoint_url"
+    )]
+    fn config_otel_export_logs_without_endpoint() {
+        let mut config = minimal_config();
+        config.otel_export_logs = true;
+        config.validate();
+    }
+
+    #[test]
+    fn config_otel_export_with_endpoint() {
+        let mut config = minimal_config();
+        config.otel_grpc_endpoint_url = Some("http://localhost:4317".to_string());
+        config.otel_export_metrics = true;
+        config.otel_export_logs = true;
+        config.validate();
+    }
 }

src/config.rs

@@ -84,12 +84,16 @@ const EVICTION_AGE_BUCKETS: &[f64] = &[
 pub(crate) fn initialize_telemetry(
     config: &Config,
 ) -> crate::Result<(
-    opentelemetry_sdk::metrics::SdkMeterProvider,
+    Option<opentelemetry_sdk::metrics::SdkMeterProvider>,
     Option<opentelemetry_sdk::logs::SdkLoggerProvider>,
 )> {
     let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
 
-    let logs_provider = init_logs(config.otel_grpc_endpoint_url.as_deref())?;
+    let otel_logs_endpoint = config
+        .otel_grpc_endpoint_url
+        .as_deref()
+        .filter(|_| config.otel_export_logs);
+    let logs_provider = init_logs(otel_logs_endpoint)?;
 
     match logs_provider.as_ref() {
         None => {
@@ -109,7 +113,11 @@ pub(crate) fn initialize_telemetry(
         }
     }
 
-    let metrics_provider = init_metrics(config.otel_grpc_endpoint_url.as_deref())?;
+    let otel_metrics_endpoint = config
+        .otel_grpc_endpoint_url
+        .as_deref()
+        .filter(|_| config.otel_export_metrics);
+    let metrics_provider = init_metrics(otel_metrics_endpoint)?;
 
     Ok((metrics_provider, logs_provider))
 }
@@ -147,35 +155,37 @@ pub(crate) fn shutdown_logs(logs_provider: Option<opentelemetry_sdk::logs::SdkLo
 
 fn init_metrics(
     otel_grpc_endpoint_url: Option<&str>,
-) -> crate::Result<opentelemetry_sdk::metrics::SdkMeterProvider> {
-    let builder =
-        opentelemetry_sdk::metrics::SdkMeterProvider::builder().with_resource(RESOURCE.clone());
+) -> crate::Result<Option<opentelemetry_sdk::metrics::SdkMeterProvider>> {
+    let Some(otel_grpc_endpoint_url) = otel_grpc_endpoint_url else {
+        info!("OTLP metrics export disabled");
+        return Ok(None);
+    };
 
-    let provider = match otel_grpc_endpoint_url {
-        None => {
-            info!("opentelemetry_stdout initialized");
-            builder.with_periodic_exporter(opentelemetry_stdout::MetricExporter::default())
-        }
-        Some(otel_grpc_endpoint_url) => {
-            info!("opentelemetry_otlp initialized");
-            let otlp_exporter = opentelemetry_otlp::MetricExporter::builder()
-                .with_tonic()
-                .with_compression(Compression::Gzip)
-                .with_endpoint(otel_grpc_endpoint_url)
-                .with_timeout(Duration::from_secs(5))
-                .build()?;
-
-            builder.with_periodic_exporter(otlp_exporter)
-        }
-    }
-    .build();
+    info!("OTLP metrics export enabled (endpoint: {otel_grpc_endpoint_url})");
+    let otlp_exporter = opentelemetry_otlp::MetricExporter::builder()
+        .with_tonic()
+        .with_compression(Compression::Gzip)
+        .with_endpoint(otel_grpc_endpoint_url)
+        .with_timeout(Duration::from_secs(5))
+        .build()?;
+
+    let provider = opentelemetry_sdk::metrics::SdkMeterProvider::builder()
+        .with_resource(RESOURCE.clone())
+        .with_periodic_exporter(otlp_exporter)
+        .build();
 
     opentelemetry::global::set_meter_provider(provider.clone());
 
-    Ok(provider)
+    Ok(Some(provider))
 }
 
-pub(crate) fn shutdown_metrics(metric_provider: opentelemetry_sdk::metrics::SdkMeterProvider) {
+pub(crate) fn shutdown_metrics(
+    metric_provider: Option<opentelemetry_sdk::metrics::SdkMeterProvider>,
+) {
+    let Some(metric_provider) = metric_provider else {
+        return;
+    };
+
     if let Err(error) = metric_provider.shutdown() {
         error!("Error during metric shutdown: {error:?}");
     }

src/telemetry.rs

@@ -90,6 +90,8 @@ pub async fn start_proxy(minio_api_port: u16) -> TestProxy {
         cache_ttl_seconds: 86_400,
         worker_threads: 4,
         otel_grpc_endpoint_url: None, // no network side effects in tests
+        otel_export_logs: false,
+        otel_export_metrics: false,
         prometheus_textfile_dir: None, // no filesystem side effects in tests
     };
 

tests/common/docker.rs

@@ -28,6 +28,8 @@ async fn start_test_server_with_upstream(
         cache_ttl_seconds: 60,
         worker_threads: 2,
         otel_grpc_endpoint_url: None,
+        otel_export_metrics: false,
+        otel_export_logs: false,
         prometheus_textfile_dir: None,
     };