Emit SUT metrics in real time (#161)

* disk metrics should be counters

Entire-Checkpoint: 2b1195615a61

* make SUT metrics streamable

Entire-Checkpoint: 4d2e6e04b1b1

* streaming SUT metrics

* simplify

Entire-Checkpoint: 80f190a205d4

Author: Jeadie

crates/system-adapter-protocol/src/lib.rs

@@ -222,22 +222,22 @@ pub struct MetricsRequest {
 /// Resource utilization snapshot from the system under test
 #[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
 pub struct ResourceMetrics {
-    /// CPU utilization as a percentage (0.0–100.0)
+    /// Cumulative CPU seconds used
     #[serde(skip_serializing_if = "Option::is_none")]
     pub cpu_usage_percent: Option<f64>,
     /// Resident memory usage in bytes
     #[serde(skip_serializing_if = "Option::is_none")]
     pub memory_usage_bytes: Option<u64>,
-    /// Disk bytes read since last scrape
+    /// Cumulative disk bytes read
     #[serde(skip_serializing_if = "Option::is_none")]
     pub disk_read_bytes: Option<u64>,
-    /// Disk bytes written since last scrape
+    /// Cumulative disk bytes written
     #[serde(skip_serializing_if = "Option::is_none")]
     pub disk_write_bytes: Option<u64>,
-    /// Disk read IOPS since last scrape
+    /// Cumulative disk read operations
     #[serde(skip_serializing_if = "Option::is_none")]
     pub disk_read_iops: Option<u64>,
-    /// Disk write IOPS since last scrape
+    /// Cumulative disk write operations
     #[serde(skip_serializing_if = "Option::is_none")]
     pub disk_write_iops: Option<u64>,
 }

crates/test-framework/src/telemetry/mod.rs

@@ -23,6 +23,7 @@ use anyhow::Result;
 
 use opentelemetry::metrics::{Meter, MeterProvider};
 
+use opentelemetry_sdk::metrics::PeriodicReader;
 use opentelemetry_sdk::metrics::exporter::PushMetricExporter;
 use opentelemetry_sdk::metrics::reader::MetricReader;
 use opentelemetry_sdk::{
@@ -229,3 +230,99 @@ impl Telemetry {
         Ok(())
     }
 }
+
+/// A dedicated periodic metrics pipeline for SUT resource metrics.
+///
+/// Always exports to the Arrow backend (when `SPICEAI_BENCHMARK_METRICS_KEY` is set).
+/// Additionally exports to an OTLP endpoint when configured.
+/// Instruments created on the returned [`Meter`] are exported every 5 seconds.
+pub struct SutMetricsPipeline {
+    provider: SdkMeterProvider,
+    meter: Meter,
+}
+
+impl SutMetricsPipeline {
+    /// Create the pipeline.
+    ///
+    /// - `api_key_name`: env var name for the Arrow backend API key (e.g. `"SPICEAI_BENCHMARK_METRICS_KEY"`).
+    /// - `otlp_endpoint`: optional OTLP gRPC endpoint to also export to.
+    /// - `resource`: OTel resource attributes for exported metrics.
+    pub async fn new(
+        api_key_name: &str,
+        otlp_endpoint: Option<&str>,
+        resource: Resource,
+    ) -> Result<Self> {
+        let mut builder = SdkMeterProvider::builder().with_resource(resource);
+
+        match std::env::var(api_key_name) {
+            Ok(key) => {
+                let token = Arc::new(SecretString::new(key.into()));
+                // Arrow periodic reader (always, when API key is present)
+                let arrow_exporter = otel_arrow::OtelArrowExporter::new(
+                    TelemetryExporterBuilder::new()
+                        .with_credentials(flight_client::Credentials::Bearer {
+                            token,
+                            prefix: false,
+                        })
+                        .with_service_name("benchmarks_telemetry".into())
+                        .with_endpoint(Arc::clone(&ENDPOINT))
+                        .build()
+                        .await?,
+                );
+                let reader = PeriodicReader::builder(arrow_exporter)
+                    .with_interval(Duration::from_secs(5))
+                    .build();
+                builder = builder.with_reader(reader);
+                println!(
+                    "SUT metrics: Arrow periodic exporter enabled (endpoint: {})",
+                    *ENDPOINT
+                )
+            }
+            Err(e) => {
+                eprintln!("Failed to create Arrow exporter for SUT metrics: {e}");
+            }
+        };
+
+        // OTLP periodic reader (when --otlp-endpoint is configured)
+        if let Some(endpoint) = otlp_endpoint {
+            match MetricExporter::builder()
+                .with_tonic()
+                .with_timeout(Duration::from_secs(10))
+                .with_endpoint(endpoint)
+                .build()
+            {
+                Ok(otlp_exporter) => {
+                    let reader = PeriodicReader::builder(otlp_exporter)
+                        .with_interval(Duration::from_secs(5))
+                        .build();
+                    builder = builder.with_reader(reader);
+                    println!("SUT metrics: OTLP periodic exporter enabled (endpoint: {endpoint})");
+                }
+                Err(e) => {
+                    eprintln!("Failed to create OTLP exporter for SUT metrics: {e}");
+                }
+            }
+        }
+
+        let provider = builder.build();
+        let meter = provider.meter("spicebench-sut");
+
+        Ok(Self { provider, meter })
+    }
+
+    /// Get the meter for creating SUT instruments.
+    #[must_use]
+    pub fn meter(&self) -> Meter {
+        self.meter.clone()
+    }
+
+    /// Flush and shut down the pipeline.
+    pub fn shutdown(&self) {
+        if let Err(e) = self.provider.force_flush() {
+            eprintln!("Failed to flush SUT metrics pipeline: {e}");
+        }
+        if let Err(e) = self.provider.shutdown() {
+            eprintln!("Failed to shutdown SUT metrics pipeline: {e}");
+        }
+    }
+}