Merge branch 'worktree-real-metrics-overhaul'

Cargo.lock

@@ -2291,6 +2291,7 @@ dependencies = [
  "criterion",
  "fastrand",
  "futures",
+ "http 1.3.1",
  "jsonwebtoken",
  "lazy_static",
  "ml",
@@ -2313,6 +2314,8 @@ dependencies = [
  "tokio-test",
  "toml",
  "tonic 0.14.2",
+ "tower-layer",
+ "tower-service",
  "tracing",
  "tracing-appender",
  "tracing-opentelemetry",
@@ -11543,6 +11546,8 @@ dependencies = [
  "common",
  "futures-util",
  "jsonwebtoken",
+ "once_cell",
+ "prometheus",
  "prost 0.14.1",
  "serde",
  "serde_json",

crates/common/Cargo.toml

@@ -47,8 +47,11 @@ opentelemetry.workspace = true
 opentelemetry-otlp.workspace = true
 opentelemetry_sdk.workspace = true
 
-# gRPC (for correlation ID propagation)
+# gRPC (for correlation ID propagation and metrics layer)
 tonic.workspace = true
+http.workspace = true
+tower-layer.workspace = true
+tower-service.workspace = true
 
 # Configuration
 toml.workspace = true

crates/common/src/metrics/grpc_metrics.rs

@@ -0,0 +1,245 @@
+//! gRPC metrics Tower layer for automatic request instrumentation.
+//!
+//! Provides `grpc_server_started_total`, `grpc_server_handled_total`,
+//! and `grpc_server_handling_seconds` metrics for all gRPC handlers.
+//!
+//! # Usage
+//!
+//! ```rust,ignore
+//! use common::metrics::grpc_metrics::GrpcMetricsLayer;
+//! use tonic::transport::Server;
+//!
+//! Server::builder()
+//!     .layer(GrpcMetricsLayer::new("trading_service"))
+//!     .add_service(my_service)
+//!     .serve(addr)
+//!     .await?;
+//! ```
+
+use once_cell::sync::Lazy;
+use prometheus::{register_counter_vec, register_histogram_vec, CounterVec, HistogramVec};
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tower_layer::Layer;
+use tower_service::Service;
+
+/// Counter for gRPC requests started.
+static GRPC_STARTED: Lazy<CounterVec> = Lazy::new(|| {
+    register_counter_vec!(
+        "grpc_server_started_total",
+        "Total gRPC RPCs started on the server",
+        &["service", "grpc_method"]
+    )
+    .unwrap_or_else(|e| {
+        // Metric registration failure at startup is unrecoverable.
+        // The process cannot operate without metrics instrumentation.
+        eprintln!("FATAL: failed to register grpc_server_started_total: {e}");
+        std::process::abort()
+    })
+});
+
+/// Counter for gRPC requests completed with status code.
+static GRPC_HANDLED: Lazy<CounterVec> = Lazy::new(|| {
+    register_counter_vec!(
+        "grpc_server_handled_total",
+        "Total gRPC RPCs completed on the server with status code",
+        &["service", "grpc_method", "grpc_code"]
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register grpc_server_handled_total: {e}");
+        std::process::abort()
+    })
+});
+
+/// Histogram for gRPC request handling duration.
+static GRPC_HANDLING_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
+    register_histogram_vec!(
+        "grpc_server_handling_seconds",
+        "Histogram of gRPC request handling duration in seconds",
+        &["service", "grpc_method"],
+        vec![0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0]
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register grpc_server_handling_seconds: {e}");
+        std::process::abort()
+    })
+});
+
+/// Initialize gRPC metrics (force lazy registration).
+///
+/// Call this early in service startup to ensure metrics are registered
+/// before any requests arrive. It is safe to call multiple times.
+pub fn init_grpc_metrics() {
+    let _ = &*GRPC_STARTED;
+    let _ = &*GRPC_HANDLED;
+    let _ = &*GRPC_HANDLING_SECONDS;
+}
+
+/// Extract the method name from a gRPC URI path.
+///
+/// gRPC paths follow the format `/package.Service/Method`.
+/// This extracts the last segment (the method name).
+fn extract_method(path: &str) -> &str {
+    match path.rsplit('/').next() {
+        Some(m) if !m.is_empty() => m,
+        _ => "unknown",
+    }
+}
+
+/// Tower Layer that instruments gRPC handlers with Prometheus metrics.
+///
+/// Records three metric families:
+/// - `grpc_server_started_total` — incremented when a request begins
+/// - `grpc_server_handled_total` — incremented when a request completes (with status code)
+/// - `grpc_server_handling_seconds` — histogram of request duration
+#[derive(Debug, Clone)]
+pub struct GrpcMetricsLayer {
+    service_name: &'static str,
+}
+
+impl GrpcMetricsLayer {
+    /// Create a new gRPC metrics layer for the given service.
+    ///
+    /// `service_name` is used as the `service` label on all emitted metrics.
+    pub fn new(service_name: &'static str) -> Self {
+        init_grpc_metrics();
+        Self { service_name }
+    }
+}
+
+impl<S> Layer<S> for GrpcMetricsLayer {
+    type Service = GrpcMetricsService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        GrpcMetricsService {
+            inner,
+            service_name: self.service_name,
+        }
+    }
+}
+
+/// Tower Service that records gRPC metrics around the inner service.
+#[derive(Debug, Clone)]
+pub struct GrpcMetricsService<S> {
+    inner: S,
+    service_name: &'static str,
+}
+
+impl<S, ReqBody, ResBody> Service<http::Request<ReqBody>> for GrpcMetricsService<S>
+where
+    S: Service<http::Request<ReqBody>, Response = http::Response<ResBody>>
+        + Clone
+        + Send
+        + 'static,
+    S::Future: Send + 'static,
+    S::Error: Send + 'static,
+    ReqBody: Send + 'static,
+{
+    type Response = S::Response;
+    type Error = S::Error;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: http::Request<ReqBody>) -> Self::Future {
+        let service_name = self.service_name;
+        let method = extract_method(req.uri().path()).to_string();
+
+        GRPC_STARTED
+            .with_label_values(&[service_name, &method])
+            .inc();
+
+        let start = std::time::Instant::now();
+        // Clone inner before calling to satisfy tower's Service contract:
+        // poll_ready was called on `self`, so we must call on `self`
+        // (not the clone). We swap so the clone becomes `self` for
+        // the next call, and we consume the ready instance.
+        let mut inner = self.inner.clone();
+        std::mem::swap(&mut self.inner, &mut inner);
+
+        Box::pin(async move {
+            let result = inner.call(req).await;
+            let elapsed = start.elapsed().as_secs_f64();
+
+            let code = match &result {
+                Ok(resp) => resp
+                    .headers()
+                    .get("grpc-status")
+                    .and_then(|v| v.to_str().ok())
+                    .map_or_else(|| "0".to_string(), |s| s.to_string()),
+                Err(_) => "13".to_string(), // INTERNAL
+            };
+
+            GRPC_HANDLED
+                .with_label_values(&[service_name, &method, &code])
+                .inc();
+
+            GRPC_HANDLING_SECONDS
+                .with_label_values(&[service_name, &method])
+                .observe(elapsed);
+
+            result
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_extract_method_normal_path() {
+        assert_eq!(extract_method("/package.Service/MethodName"), "MethodName");
+    }
+
+    #[test]
+    fn test_extract_method_simple_path() {
+        assert_eq!(extract_method("/Method"), "Method");
+    }
+
+    #[test]
+    fn test_extract_method_empty_path() {
+        assert_eq!(extract_method(""), "unknown");
+    }
+
+    #[test]
+    fn test_extract_method_trailing_slash() {
+        // rsplit('/').next() on "foo/" yields "" (empty), so fallback
+        assert_eq!(extract_method("/Service/"), "unknown");
+    }
+
+    #[test]
+    fn test_extract_method_root_slash() {
+        assert_eq!(extract_method("/"), "unknown");
+    }
+
+    #[test]
+    fn test_metrics_layer_creates_service() {
+        let layer = GrpcMetricsLayer::new("test_svc");
+        assert_eq!(layer.service_name, "test_svc");
+    }
+
+    #[test]
+    fn test_init_grpc_metrics_is_idempotent() {
+        // Calling init multiple times should not panic
+        init_grpc_metrics();
+        init_grpc_metrics();
+        init_grpc_metrics();
+    }
+
+    #[test]
+    fn test_metrics_registered() {
+        init_grpc_metrics();
+        // Verify metrics are accessible via the global statics
+        GRPC_STARTED
+            .with_label_values(&["test_svc", "TestMethod"])
+            .inc();
+        let val = GRPC_STARTED
+            .with_label_values(&["test_svc", "TestMethod"])
+            .get();
+        assert!(val >= 1.0, "Counter should have been incremented");
+    }
+}

crates/common/src/metrics/mod.rs

@@ -3,6 +3,7 @@
 // Provides Prometheus-based metrics collection with a global registry
 // and helper functions for all services.
 
+pub mod grpc_metrics;
 pub mod registry;
 
 // Re-export commonly used functions
@@ -14,5 +15,7 @@ pub use registry::{
     set_gauge, set_gauge_vec, REGISTRY,
 };
 
+pub use grpc_metrics::{init_grpc_metrics, GrpcMetricsLayer};
+
 #[cfg(test)]
 mod tests;

crates/ml/src/training.rs

@@ -12,6 +12,7 @@
 
 // Sub-modules for specialized training components
 pub mod orchestrator;
+pub mod push_metrics;
 pub mod unified_data_loader;
 pub mod unified_trainer; // NEW: Unified training trait for all models // NEW: Model-agnostic training orchestrator
 

crates/ml/src/training/push_metrics.rs

@@ -0,0 +1,211 @@
+//! Prometheus Pushgateway integration for training job metrics
+//!
+//! Training jobs are short-lived K8s Jobs. They push epoch-level metrics
+//! to the Pushgateway so Prometheus can scrape them persistently.
+
+use std::fmt::Write;
+
+/// Pushgateway client for training metrics
+#[derive(Debug)]
+pub struct TrainingMetricsPusher {
+    pushgateway_url: String,
+    job_id: String,
+    model_name: String,
+    client: reqwest::Client,
+}
+
+impl TrainingMetricsPusher {
+    /// Create a new pusher. Falls back to in-cluster Pushgateway if `PUSHGATEWAY_URL` not set.
+    pub fn new(job_id: &str, model_name: &str) -> Self {
+        let pushgateway_url = std::env::var("PUSHGATEWAY_URL")
+            .unwrap_or_else(|_| "http://pushgateway.foxhunt.svc.cluster.local:9091".to_string());
+        Self {
+            pushgateway_url,
+            job_id: job_id.to_string(),
+            model_name: model_name.to_string(),
+            client: reqwest::Client::new(),
+        }
+    }
+
+    /// Push current training state to Pushgateway.
+    /// Non-fatal: logs errors but never panics.
+    pub async fn push(&self, state: &TrainingState) {
+        let mut body = String::new();
+
+        // Epoch progress
+        let _ = writeln!(body, "# HELP foxhunt_training_current_epoch Current training epoch");
+        let _ = writeln!(body, "# TYPE foxhunt_training_current_epoch gauge");
+        let _ = writeln!(body, "foxhunt_training_current_epoch {}", state.epoch);
+
+        // Loss
+        let _ = writeln!(
+            body,
+            "# HELP foxhunt_training_epoch_loss Training loss value"
+        );
+        let _ = writeln!(body, "# TYPE foxhunt_training_epoch_loss gauge");
+        let _ = writeln!(
+            body,
+            "foxhunt_training_epoch_loss{{split=\"train\"}} {}",
+            state.train_loss
+        );
+        if let Some(val_loss) = state.val_loss {
+            let _ = writeln!(
+                body,
+                "foxhunt_training_epoch_loss{{split=\"val\"}} {}",
+                val_loss
+            );
+        }
+
+        // Accuracy
+        if let Some(accuracy) = state.accuracy {
+            let _ = writeln!(
+                body,
+                "# HELP foxhunt_training_eval_accuracy Model evaluation accuracy"
+            );
+            let _ = writeln!(body, "# TYPE foxhunt_training_eval_accuracy gauge");
+            let _ = writeln!(body, "foxhunt_training_eval_accuracy {}", accuracy);
+        }
+
+        // Batches processed
+        let _ = writeln!(
+            body,
+            "# HELP foxhunt_training_batches_processed Total batches processed"
+        );
+        let _ = writeln!(body, "# TYPE foxhunt_training_batches_processed counter");
+        let _ = writeln!(
+            body,
+            "foxhunt_training_batches_processed {}",
+            state.batches_processed
+        );
+
+        // Training speed
+        if state.batches_per_second > 0.0 {
+            let _ = writeln!(
+                body,
+                "# HELP foxhunt_training_batches_per_second Training throughput"
+            );
+            let _ = writeln!(body, "# TYPE foxhunt_training_batches_per_second gauge");
+            let _ = writeln!(
+                body,
+                "foxhunt_training_batches_per_second {}",
+                state.batches_per_second
+            );
+        }
+
+        // Learning rate
+        let _ = writeln!(
+            body,
+            "# HELP foxhunt_training_learning_rate Current learning rate"
+        );
+        let _ = writeln!(body, "# TYPE foxhunt_training_learning_rate gauge");
+        let _ = writeln!(
+            body,
+            "foxhunt_training_learning_rate {}",
+            state.learning_rate
+        );
+
+        // NaN / gradient events
+        if state.nan_count > 0 {
+            let _ = writeln!(
+                body,
+                "# HELP foxhunt_training_nan_detected_total NaN gradient events"
+            );
+            let _ = writeln!(body, "# TYPE foxhunt_training_nan_detected_total counter");
+            let _ = writeln!(
+                body,
+                "foxhunt_training_nan_detected_total {}",
+                state.nan_count
+            );
+        }
+        if state.gradient_explosion_count > 0 {
+            let _ = writeln!(
+                body,
+                "# HELP foxhunt_training_gradient_explosion_total Gradient clipping events"
+            );
+            let _ = writeln!(
+                body,
+                "# TYPE foxhunt_training_gradient_explosion_total counter"
+            );
+            let _ = writeln!(
+                body,
+                "foxhunt_training_gradient_explosion_total {}",
+                state.gradient_explosion_count
+            );
+        }
+
+        // Checkpoint saves
+        let _ = writeln!(
+            body,
+            "# HELP foxhunt_training_checkpoint_saves_total Checkpoints saved"
+        );
+        let _ = writeln!(
+            body,
+            "# TYPE foxhunt_training_checkpoint_saves_total counter"
+        );
+        let _ = writeln!(
+            body,
+            "foxhunt_training_checkpoint_saves_total {}",
+            state.checkpoint_saves
+        );
+
+        let url = format!(
+            "{}/metrics/job/{}/model/{}",
+            self.pushgateway_url, self.job_id, self.model_name
+        );
+
+        match self.client.put(&url).body(body).send().await {
+            Ok(resp) if resp.status().is_success() => {}
+            Ok(resp) => {
+                eprintln!(
+                    "Pushgateway returned {}: {}",
+                    resp.status(),
+                    resp.text().await.unwrap_or_default()
+                );
+            }
+            Err(e) => {
+                eprintln!("Failed to push metrics to Pushgateway: {e}");
+            }
+        }
+    }
+
+    /// Delete metrics for this job from Pushgateway (call on training completion).
+    pub async fn cleanup(&self) {
+        let url = format!(
+            "{}/metrics/job/{}/model/{}",
+            self.pushgateway_url, self.job_id, self.model_name
+        );
+        let _ = self.client.delete(&url).send().await;
+    }
+}
+
+/// Current training state to push
+#[derive(Debug)]
+pub struct TrainingState {
+    pub epoch: u64,
+    pub train_loss: f64,
+    pub val_loss: Option<f64>,
+    pub accuracy: Option<f64>,
+    pub batches_processed: u64,
+    pub batches_per_second: f64,
+    pub learning_rate: f64,
+    pub nan_count: u64,
+    pub gradient_explosion_count: u64,
+    pub checkpoint_saves: u64,
+}
+
+impl Default for TrainingState {
+    fn default() -> Self {
+        Self {
+            epoch: 0,
+            train_loss: 0.0,
+            val_loss: None,
+            accuracy: None,
+            batches_processed: 0,
+            batches_per_second: 0.0,
+            learning_rate: 0.001,
+            nan_count: 0,
+            gradient_explosion_count: 0,
+            checkpoint_saves: 0,
+        }
+    }
+}

crates/web-gateway/Cargo.toml

@@ -43,6 +43,10 @@ tracing-subscriber.workspace = true
 # Shared types
 common.workspace = true
 
+# Metrics
+prometheus.workspace = true
+once_cell.workspace = true
+
 # Time and IDs
 chrono.workspace = true
 uuid.workspace = true

crates/web-gateway/src/lib.rs

@@ -10,6 +10,7 @@ pub mod auth;
 pub mod config;
 pub mod error;
 pub mod grpc;
+pub mod metrics;
 pub mod rate_limit;
 pub mod routes;
 pub mod state;

crates/web-gateway/src/main.rs

@@ -75,6 +75,52 @@ async fn main() -> Result<()> {
         .layer(TraceLayer::new_for_http())
         .layer(middleware::from_fn(request_id_middleware));
 
+    // Initialize Prometheus metrics
+    web_gateway::metrics::init_metrics();
+    let service_start = std::time::Instant::now();
+
+    // Spawn uptime updater
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
+        loop {
+            interval.tick().await;
+            web_gateway::metrics::update_uptime(service_start);
+        }
+    });
+
+    // Start Prometheus metrics HTTP endpoint on a separate port
+    let metrics_port: u16 = std::env::var("METRICS_PORT")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(9098);
+    tokio::spawn(async move {
+        use axum::{routing::get, Router};
+        use prometheus::{Encoder, TextEncoder};
+
+        async fn metrics_handler() -> String {
+            let encoder = TextEncoder::new();
+            let metric_families = prometheus::gather();
+            let mut buffer = vec![];
+            let _ = encoder.encode(&metric_families, &mut buffer);
+            String::from_utf8(buffer).unwrap_or_else(|_| String::new())
+        }
+
+        let app = Router::new().route("/metrics", get(metrics_handler));
+        let addr = format!("0.0.0.0:{}", metrics_port);
+        tracing::info!("Prometheus metrics endpoint listening on http://{}", addr);
+
+        let metrics_listener = match tokio::net::TcpListener::bind(&addr).await {
+            Ok(l) => l,
+            Err(e) => {
+                tracing::error!("Failed to bind metrics endpoint {}: {}", addr, e);
+                return;
+            }
+        };
+        if let Err(e) = axum::serve(metrics_listener, app).await {
+            tracing::error!("Metrics server failed: {}", e);
+        }
+    });
+
     let listener = tokio::net::TcpListener::bind(&listen_addr).await?;
     info!("Web gateway listening on {}", listen_addr);
 

crates/web-gateway/src/metrics.rs

@@ -0,0 +1,84 @@
+//! Prometheus metrics for Web Gateway
+
+use once_cell::sync::Lazy;
+use prometheus::{
+    register_counter_vec, register_gauge, register_histogram_vec, CounterVec, Gauge, HistogramVec,
+};
+
+/// Service uptime in seconds
+pub static SERVICE_UPTIME: Lazy<Gauge> = Lazy::new(|| {
+    register_gauge!(
+        "foxhunt_web_gateway_uptime_seconds",
+        "Service uptime in seconds"
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register uptime gauge: {e}");
+        std::process::abort()
+    })
+});
+
+/// HTTP requests total
+pub static HTTP_REQUESTS: Lazy<CounterVec> = Lazy::new(|| {
+    register_counter_vec!(
+        "foxhunt_web_gateway_http_requests_total",
+        "Total HTTP requests",
+        &["method", "path", "status"]
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register http_requests counter: {e}");
+        std::process::abort()
+    })
+});
+
+/// HTTP request duration
+pub static HTTP_REQUEST_DURATION: Lazy<HistogramVec> = Lazy::new(|| {
+    register_histogram_vec!(
+        "foxhunt_web_gateway_http_request_duration_seconds",
+        "HTTP request duration in seconds",
+        &["method", "path"],
+        vec![0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0]
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register http_request_duration histogram: {e}");
+        std::process::abort()
+    })
+});
+
+/// Active WebSocket connections
+pub static WS_CONNECTIONS: Lazy<Gauge> = Lazy::new(|| {
+    register_gauge!(
+        "foxhunt_web_gateway_ws_connections_active",
+        "Active WebSocket connections"
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register ws_connections gauge: {e}");
+        std::process::abort()
+    })
+});
+
+/// WebSocket messages total
+pub static WS_MESSAGES: Lazy<CounterVec> = Lazy::new(|| {
+    register_counter_vec!(
+        "foxhunt_web_gateway_ws_messages_total",
+        "Total WebSocket messages",
+        &["direction", "topic"]
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register ws_messages counter: {e}");
+        std::process::abort()
+    })
+});
+
+/// Initialize all metrics (force lazy registration)
+pub fn init_metrics() {
+    let _ = &*SERVICE_UPTIME;
+    let _ = &*HTTP_REQUESTS;
+    let _ = &*HTTP_REQUEST_DURATION;
+    let _ = &*WS_CONNECTIONS;
+    let _ = &*WS_MESSAGES;
+}
+
+/// Update service uptime metric
+pub fn update_uptime(start_time: std::time::Instant) {
+    SERVICE_UPTIME.set(start_time.elapsed().as_secs_f64());
+}

infra/k8s/monitoring/dashboards/foxhunt-traces.json

@@ -1,13 +1,29 @@
 {
   "__inputs": [
-    { "name": "DS_PROMETHEUS", "label": "Prometheus", "description": "", "type": "datasource", "pluginId": "prometheus" },
-    { "name": "DS_TEMPO", "label": "Tempo", "description": "", "type": "datasource", "pluginId": "tempo" }
+    {
+      "name": "DS_PROMETHEUS",
+      "label": "Prometheus",
+      "description": "",
+      "type": "datasource",
+      "pluginId": "prometheus"
+    },
+    {
+      "name": "DS_TEMPO",
+      "label": "Tempo",
+      "description": "",
+      "type": "datasource",
+      "pluginId": "tempo"
+    }
   ],
   "id": null,
   "uid": "foxhunt-traces",
   "title": "Foxhunt - Traces",
   "description": "Distributed tracing dashboard: span rates, error rates, latency percentiles, service map, trace search",
-  "tags": ["foxhunt", "traces", "tempo"],
+  "tags": [
+    "foxhunt",
+    "traces",
+    "tempo"
+  ],
   "timezone": "browser",
   "schemaVersion": 39,
   "version": 1,
@@ -25,12 +41,19 @@
       {
         "name": "service",
         "type": "query",
-        "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+        "datasource": {
+          "type": "prometheus",
+          "uid": "${DS_PROMETHEUS}"
+        },
         "query": "label_values(traces_service_graph_request_total, client)",
         "refresh": 2,
         "multi": true,
         "includeAll": true,
-        "current": { "selected": true, "text": "All", "value": "$__all" }
+        "current": {
+          "selected": true,
+          "text": "All",
+          "value": "$__all"
+        }
       }
     ]
   },
@@ -39,7 +62,12 @@
       "id": 1,
       "type": "row",
       "title": "Overview Stats",
-      "gridPos": { "h": 1, "w": 24, "x": 0, "y": 0 },
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 0
+      },
       "collapsed": false,
       "panels": []
     },
@@ -47,23 +75,42 @@
       "id": 2,
       "type": "stat",
       "title": "Spans/sec",
-      "gridPos": { "h": 4, "w": 6, "x": 0, "y": 1 },
-      "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+      "gridPos": {
+        "h": 4,
+        "w": 6,
+        "x": 0,
+        "y": 1
+      },
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
       "fieldConfig": {
         "defaults": {
           "thresholds": {
             "mode": "absolute",
             "steps": [
-              { "color": "green", "value": null }
+              {
+                "color": "green",
+                "value": null
+              }
             ]
           },
           "mappings": [],
-          "color": { "mode": "thresholds" }
+          "color": {
+            "mode": "thresholds"
+          }
         },
         "overrides": []
       },
       "options": {
-        "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false },
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
         "orientation": "auto",
         "textMode": "auto",
         "colorMode": "background",
@@ -73,8 +120,11 @@
       "targets": [
         {
           "refId": "A",
-          "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
-          "expr": "sum(rate(traces_spanmetrics_calls_total[5m]))",
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "sum(rate(tempo_distributor_spans_received_total[5m]))",
           "instant": true
         }
       ]
@@ -83,26 +133,51 @@
       "id": 3,
       "type": "stat",
       "title": "Error Rate",
-      "gridPos": { "h": 4, "w": 6, "x": 6, "y": 1 },
-      "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+      "gridPos": {
+        "h": 4,
+        "w": 6,
+        "x": 6,
+        "y": 1
+      },
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
       "fieldConfig": {
         "defaults": {
           "unit": "percent",
           "thresholds": {
             "mode": "absolute",
             "steps": [
-              { "color": "green", "value": null },
-              { "color": "yellow", "value": 1 },
-              { "color": "red", "value": 5 }
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "yellow",
+                "value": 1
+              },
+              {
+                "color": "red",
+                "value": 5
+              }
             ]
           },
           "mappings": [],
-          "color": { "mode": "thresholds" }
+          "color": {
+            "mode": "thresholds"
+          }
         },
         "overrides": []
       },
       "options": {
-        "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false },
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
         "orientation": "auto",
         "textMode": "auto",
         "colorMode": "background",
@@ -112,8 +187,11 @@
       "targets": [
         {
           "refId": "A",
-          "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
-          "expr": "sum(rate(traces_spanmetrics_calls_total{status_code=\"STATUS_CODE_ERROR\"}[5m])) / sum(rate(traces_spanmetrics_calls_total[5m])) * 100",
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "sum(rate(tempo_distributor_spans_received_total{status=\"error\"}[5m])) / sum(rate(tempo_distributor_spans_received_total[5m])) * 100",
           "instant": true
         }
       ]
@@ -122,24 +200,43 @@
       "id": 4,
       "type": "stat",
       "title": "Avg Duration",
-      "gridPos": { "h": 4, "w": 6, "x": 12, "y": 1 },
-      "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+      "gridPos": {
+        "h": 4,
+        "w": 6,
+        "x": 12,
+        "y": 1
+      },
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
       "fieldConfig": {
         "defaults": {
           "unit": "s",
           "thresholds": {
             "mode": "absolute",
             "steps": [
-              { "color": "green", "value": null }
+              {
+                "color": "green",
+                "value": null
+              }
             ]
           },
           "mappings": [],
-          "color": { "mode": "thresholds" }
+          "color": {
+            "mode": "thresholds"
+          }
         },
         "overrides": []
       },
       "options": {
-        "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false },
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
         "orientation": "auto",
         "textMode": "auto",
         "colorMode": "background",
@@ -149,8 +246,11 @@
       "targets": [
         {
           "refId": "A",
-          "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
-          "expr": "sum(rate(traces_spanmetrics_latency_sum[5m])) / sum(rate(traces_spanmetrics_calls_total[5m]))",
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "sum(rate(tempo_spanmetrics_latency_sum[5m])) / sum(rate(tempo_spanmetrics_latency_count[5m]))",
           "instant": true
         }
       ]
@@ -159,23 +259,42 @@
       "id": 5,
       "type": "stat",
       "title": "Active Services",
-      "gridPos": { "h": 4, "w": 6, "x": 18, "y": 1 },
-      "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+      "gridPos": {
+        "h": 4,
+        "w": 6,
+        "x": 18,
+        "y": 1
+      },
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
       "fieldConfig": {
         "defaults": {
           "thresholds": {
             "mode": "absolute",
             "steps": [
-              { "color": "green", "value": null }
+              {
+                "color": "green",
+                "value": null
+              }
             ]
           },
           "mappings": [],
-          "color": { "mode": "thresholds" }
+          "color": {
+            "mode": "thresholds"
+          }
         },
         "overrides": []
       },
       "options": {
-        "reduceOptions": { "calcs": ["lastNotNull"], "fields": "", "values": false },
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
         "orientation": "auto",
         "textMode": "auto",
         "colorMode": "background",
@@ -185,8 +304,11 @@
       "targets": [
         {
           "refId": "A",
-          "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
-          "expr": "count(count by (service_name)(traces_spanmetrics_calls_total))",
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "count(count by (service_name)(tempo_distributor_spans_received_total))",
           "instant": true
         }
       ]
@@ -195,7 +317,12 @@
       "id": 20,
       "type": "row",
       "title": "Service Map + Trace Search",
-      "gridPos": { "h": 1, "w": 24, "x": 0, "y": 5 },
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 5
+      },
       "collapsed": false,
       "panels": []
     },
@@ -203,12 +330,23 @@
       "id": 6,
       "type": "nodeGraph",
       "title": "Service Map",
-      "gridPos": { "h": 10, "w": 24, "x": 0, "y": 6 },
-      "datasource": { "type": "tempo", "uid": "${DS_TEMPO}" },
+      "gridPos": {
+        "h": 10,
+        "w": 24,
+        "x": 0,
+        "y": 6
+      },
+      "datasource": {
+        "type": "tempo",
+        "uid": "${DS_TEMPO}"
+      },
       "targets": [
         {
           "refId": "A",
-          "datasource": { "type": "tempo", "uid": "${DS_TEMPO}" },
+          "datasource": {
+            "type": "tempo",
+            "uid": "${DS_TEMPO}"
+          },
           "queryType": "serviceMap"
         }
       ]
@@ -217,7 +355,12 @@
       "id": 21,
       "type": "row",
       "title": "Latency and Errors by Service",
-      "gridPos": { "h": 1, "w": 24, "x": 0, "y": 16 },
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 16
+      },
       "collapsed": false,
       "panels": []
     },
@@ -225,15 +368,26 @@
       "id": 7,
       "type": "timeseries",
       "title": "Latency by Service p95",
-      "gridPos": { "h": 7, "w": 12, "x": 0, "y": 17 },
-      "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+      "gridPos": {
+        "h": 7,
+        "w": 12,
+        "x": 0,
+        "y": 17
+      },
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
       "fieldConfig": {
         "defaults": {
           "unit": "s",
           "thresholds": {
             "mode": "absolute",
             "steps": [
-              { "color": "green", "value": null }
+              {
+                "color": "green",
+                "value": null
+              }
             ]
           },
           "custom": {
@@ -248,22 +402,39 @@
             "axisBorderShow": false,
             "axisCenteredZero": false,
             "axisColorMode": "text",
-            "scaleDistribution": { "type": "linear" }
+            "scaleDistribution": {
+              "type": "linear"
+            }
+          },
+          "color": {
+            "mode": "palette-classic"
           },
-          "color": { "mode": "palette-classic" },
           "mappings": []
         },
         "overrides": []
       },
       "options": {
-        "tooltip": { "mode": "multi", "sort": "desc" },
-        "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] }
+        "tooltip": {
+          "mode": "multi",
+          "sort": "desc"
+        },
+        "legend": {
+          "displayMode": "table",
+          "placement": "bottom",
+          "calcs": [
+            "mean",
+            "max"
+          ]
+        }
       },
       "targets": [
         {
           "refId": "A",
-          "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
-          "expr": "histogram_quantile(0.95, sum by (le, service_name)(rate(traces_spanmetrics_latency_bucket[5m])))",
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "histogram_quantile(0.95, sum by (le, service_name)(rate(tempo_spanmetrics_latency_bucket[5m])))",
           "legendFormat": "{{service_name}}",
           "instant": false,
           "range": true
@@ -274,15 +445,26 @@
       "id": 8,
       "type": "timeseries",
       "title": "Error Rate by Service",
-      "gridPos": { "h": 7, "w": 12, "x": 12, "y": 17 },
-      "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
+      "gridPos": {
+        "h": 7,
+        "w": 12,
+        "x": 12,
+        "y": 17
+      },
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
       "fieldConfig": {
         "defaults": {
           "unit": "reqps",
           "thresholds": {
             "mode": "absolute",
             "steps": [
-              { "color": "green", "value": null }
+              {
+                "color": "green",
+                "value": null
+              }
             ]
           },
           "custom": {
@@ -297,22 +479,39 @@
             "axisBorderShow": false,
             "axisCenteredZero": false,
             "axisColorMode": "text",
-            "scaleDistribution": { "type": "linear" }
+            "scaleDistribution": {
+              "type": "linear"
+            }
+          },
+          "color": {
+            "mode": "palette-classic"
           },
-          "color": { "mode": "palette-classic" },
           "mappings": []
         },
         "overrides": []
       },
       "options": {
-        "tooltip": { "mode": "multi", "sort": "desc" },
-        "legend": { "displayMode": "table", "placement": "bottom", "calcs": ["mean", "max"] }
+        "tooltip": {
+          "mode": "multi",
+          "sort": "desc"
+        },
+        "legend": {
+          "displayMode": "table",
+          "placement": "bottom",
+          "calcs": [
+            "mean",
+            "max"
+          ]
+        }
       },
       "targets": [
         {
           "refId": "A",
-          "datasource": { "type": "prometheus", "uid": "${DS_PROMETHEUS}" },
-          "expr": "sum by (service_name)(rate(traces_spanmetrics_calls_total{status_code=\"STATUS_CODE_ERROR\"}[5m]))",
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "expr": "sum by (service_name)(rate(tempo_distributor_spans_received_total{status=\"error\"}[5m]))",
           "legendFormat": "{{service_name}}",
           "instant": false,
           "range": true
@@ -323,7 +522,12 @@
       "id": 22,
       "type": "row",
       "title": "Trace Search",
-      "gridPos": { "h": 1, "w": 24, "x": 0, "y": 24 },
+      "gridPos": {
+        "h": 1,
+        "w": 24,
+        "x": 0,
+        "y": 24
+      },
       "collapsed": false,
       "panels": []
     },
@@ -331,14 +535,25 @@
       "id": 10,
       "type": "table",
       "title": "Recent Traces",
-      "gridPos": { "h": 8, "w": 24, "x": 0, "y": 25 },
-      "datasource": { "type": "tempo", "uid": "${DS_TEMPO}" },
+      "gridPos": {
+        "h": 8,
+        "w": 24,
+        "x": 0,
+        "y": 25
+      },
+      "datasource": {
+        "type": "tempo",
+        "uid": "${DS_TEMPO}"
+      },
       "targets": [
         {
           "refId": "A",
-          "datasource": { "type": "tempo", "uid": "${DS_TEMPO}" },
-          "queryType": "nativeSearch",
-          "serviceName": "$service",
+          "datasource": {
+            "type": "tempo",
+            "uid": "${DS_TEMPO}"
+          },
+          "queryType": "traceql",
+          "query": "{status = error} || {duration > 1s}",
           "limit": 20
         }
       ]

infra/k8s/monitoring/prometheus-values.yaml

@@ -164,3 +164,9 @@ prometheus:
           - source_labels: [__meta_kubernetes_pod_ip]
             target_label: __address__
             replacement: $1:9080
+
+      # --- Pushgateway (training job metrics) ---
+      - job_name: pushgateway
+        honor_labels: true
+        static_configs:
+          - targets: ['pushgateway.foxhunt.svc.cluster.local:9091']

infra/k8s/monitoring/pushgateway.yaml

@@ -0,0 +1,61 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: pushgateway
+  namespace: foxhunt
+  labels:
+    app.kubernetes.io/name: pushgateway
+    app.kubernetes.io/part-of: monitoring
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: pushgateway
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/name: pushgateway
+    spec:
+      nodeSelector:
+        k8s.scaleway.com/pool-name: services
+      containers:
+        - name: pushgateway
+          image: prom/pushgateway:v1.11.0
+          args:
+            - "--persistence.interval=5m"
+          ports:
+            - containerPort: 9091
+              name: http
+          resources:
+            requests:
+              cpu: 50m
+              memory: 32Mi
+            limits:
+              cpu: 200m
+              memory: 128Mi
+          livenessProbe:
+            httpGet:
+              path: /-/healthy
+              port: 9091
+            initialDelaySeconds: 5
+          readinessProbe:
+            httpGet:
+              path: /-/ready
+              port: 9091
+            initialDelaySeconds: 5
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: pushgateway
+  namespace: foxhunt
+  labels:
+    app.kubernetes.io/name: pushgateway
+spec:
+  type: ClusterIP
+  ports:
+    - port: 9091
+      targetPort: 9091
+      name: http
+  selector:
+    app.kubernetes.io/name: pushgateway

infra/k8s/services/web-gateway.yaml

@@ -18,6 +18,10 @@ spec:
       app.kubernetes.io/name: web-gateway
   template:
     metadata:
+      annotations:
+        gitlab.com/prometheus_scrape: "true"
+        gitlab.com/prometheus_port: "9098"
+        gitlab.com/prometheus_path: "/metrics"
       labels:
         app.kubernetes.io/name: web-gateway
     spec:
@@ -86,6 +90,8 @@ spec:
           ports:
             - containerPort: 3000
               name: http
+            - containerPort: 9098
+              name: metrics
           env:
             - name: JWT_SECRET
               valueFrom:
@@ -155,3 +161,6 @@ spec:
     - port: 3000
       targetPort: 3000
       name: http
+    - port: 9098
+      targetPort: 9098
+      name: metrics

services/api_gateway/src/main.rs

@@ -11,6 +11,8 @@ use std::sync::Arc;
 use std::time::Duration;
 use tracing::{error, info, warn};
 
+use common::metrics::GrpcMetricsLayer;
+
 // Import all needed types from the library
 use api_gateway::auth::jwt::JwtConfig;
 use api_gateway::auth::{
@@ -496,12 +498,14 @@ async fn main() -> Result<()> {
     // Build server with HTTP/2 optimizations
     let mut server_builder = if let Some(ref tls) = tls_config {
         tonic::transport::Server::builder()
+            .layer(GrpcMetricsLayer::new("api-gateway"))
             .tls_config(tls.to_server_tls_config())?
             .max_concurrent_streams(Some(10_000))
             .http2_keepalive_interval(Some(Duration::from_secs(30)))
             .http2_keepalive_timeout(Some(Duration::from_secs(10)))
     } else {
         tonic::transport::Server::builder()
+            .layer(GrpcMetricsLayer::new("api-gateway"))
             .max_concurrent_streams(Some(10_000))
             .http2_keepalive_interval(Some(Duration::from_secs(30)))
             .http2_keepalive_timeout(Some(Duration::from_secs(10)))

services/backtesting_service/src/main.rs

@@ -13,6 +13,8 @@ use std::net::SocketAddr;
 use tonic::transport::Server;
 use tracing::{error, info, warn};
 
+use common::metrics::GrpcMetricsLayer;
+
 mod health;
 
 use backtesting_service::foxhunt;
@@ -173,7 +175,8 @@ async fn main() -> Result<()> {
     }
 
     // Start the server with HTTP/2 optimizations
-    let mut server_builder = Server::builder();
+    let mut server_builder = Server::builder()
+        .layer(GrpcMetricsLayer::new("backtesting-service"));
 
     if enable_http2_opts {
         server_builder = server_builder

services/broker_gateway_service/src/main.rs

@@ -12,6 +12,8 @@ use tokio::signal;
 use tonic::transport::Server;
 use tracing::{error, info, warn};
 
+use common::metrics::GrpcMetricsLayer;
+
 // Service configuration
 const DEFAULT_GRPC_PORT: u16 = 50056;
 const DEFAULT_HEALTH_PORT: u16 = 8086;
@@ -142,6 +144,7 @@ async fn main() -> Result<()> {
 
     // Build gRPC server
     let server = Server::builder()
+        .layer(GrpcMetricsLayer::new("broker-gateway"))
         .add_service(health_service)
         .add_service(BrokerGatewayServiceServer::new(service))
         .serve_with_shutdown(addr, shutdown_signal());

services/data_acquisition_service/src/lib.rs

@@ -12,6 +12,7 @@
 
 pub mod downloader;
 pub mod error;
+pub mod metrics;
 pub mod service;
 pub mod uploader;
 pub mod validator;

services/data_acquisition_service/src/main.rs

@@ -11,6 +11,8 @@ use std::net::SocketAddr;
 use tonic::transport::Server;
 use tracing::info;
 
+use common::metrics::GrpcMetricsLayer;
+
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about = None)]
 struct Args {
@@ -43,6 +45,52 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     info!("gRPC port: {}", args.port);
     info!("Health port: {}", args.health_port);
 
+    // Initialize Prometheus metrics
+    data_acquisition_service::metrics::init_metrics();
+    let service_start = std::time::Instant::now();
+
+    // Spawn uptime updater
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(1));
+        loop {
+            interval.tick().await;
+            data_acquisition_service::metrics::update_uptime(service_start);
+        }
+    });
+
+    // Start Prometheus metrics HTTP endpoint on port 9097
+    let metrics_port: u16 = std::env::var("METRICS_PORT")
+        .ok()
+        .and_then(|s| s.parse().ok())
+        .unwrap_or(9097);
+    tokio::spawn(async move {
+        use axum::{routing::get, Router};
+        use prometheus::{Encoder, TextEncoder};
+
+        async fn metrics_handler() -> String {
+            let encoder = TextEncoder::new();
+            let metric_families = prometheus::gather();
+            let mut buffer = vec![];
+            let _ = encoder.encode(&metric_families, &mut buffer);
+            String::from_utf8(buffer).unwrap_or_else(|_| String::new())
+        }
+
+        let app = Router::new().route("/metrics", get(metrics_handler));
+        let addr = format!("0.0.0.0:{metrics_port}");
+        tracing::info!("Prometheus metrics endpoint listening on http://{}", addr);
+
+        let listener = match tokio::net::TcpListener::bind(&addr).await {
+            Ok(l) => l,
+            Err(e) => {
+                tracing::error!("Failed to bind metrics endpoint {}: {}", addr, e);
+                return;
+            }
+        };
+        if let Err(e) = axum::serve(listener, app).await {
+            tracing::error!("Metrics server failed: {}", e);
+        }
+    });
+
     // Create service instance
     let service = DataAcquisitionServiceImpl::default();
 
@@ -53,6 +101,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // Start gRPC server with graceful shutdown
     Server::builder()
+        .layer(GrpcMetricsLayer::new("data-acquisition-service"))
         .add_service(DataAcquisitionServiceServer::new(service))
         .serve_with_shutdown(addr, shutdown_signal())
         .await?;

services/data_acquisition_service/src/metrics.rs

@@ -0,0 +1,84 @@
+//! Prometheus metrics for Data Acquisition Service
+
+use once_cell::sync::Lazy;
+use prometheus::{
+    register_counter, register_counter_vec, register_gauge, register_histogram_vec, Counter,
+    CounterVec, Gauge, HistogramVec,
+};
+
+/// Service uptime in seconds
+pub static SERVICE_UPTIME: Lazy<Gauge> = Lazy::new(|| {
+    register_gauge!(
+        "foxhunt_data_acquisition_uptime_seconds",
+        "Service uptime in seconds"
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register uptime gauge: {e}");
+        std::process::abort()
+    })
+});
+
+/// Active data feeds
+pub static FEEDS_ACTIVE: Lazy<Gauge> = Lazy::new(|| {
+    register_gauge!(
+        "foxhunt_data_acquisition_feeds_active",
+        "Number of active data feeds"
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register feeds_active gauge: {e}");
+        std::process::abort()
+    })
+});
+
+/// Total records received
+pub static RECORDS_RECEIVED: Lazy<CounterVec> = Lazy::new(|| {
+    register_counter_vec!(
+        "foxhunt_data_acquisition_records_received_total",
+        "Total records received from data feeds",
+        &["symbol", "feed_type"]
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register records_received counter: {e}");
+        std::process::abort()
+    })
+});
+
+/// Total errors
+pub static ERRORS: Lazy<Counter> = Lazy::new(|| {
+    register_counter!(
+        "foxhunt_data_acquisition_errors_total",
+        "Total data acquisition errors"
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register errors counter: {e}");
+        std::process::abort()
+    })
+});
+
+/// Feed processing latency
+pub static FEED_LATENCY: Lazy<HistogramVec> = Lazy::new(|| {
+    register_histogram_vec!(
+        "foxhunt_data_acquisition_feed_latency_seconds",
+        "Feed processing latency in seconds",
+        &["feed_type"],
+        vec![0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0]
+    )
+    .unwrap_or_else(|e| {
+        eprintln!("FATAL: failed to register feed_latency histogram: {e}");
+        std::process::abort()
+    })
+});
+
+/// Initialize all metrics (force lazy registration)
+pub fn init_metrics() {
+    let _ = &*SERVICE_UPTIME;
+    let _ = &*FEEDS_ACTIVE;
+    let _ = &*RECORDS_RECEIVED;
+    let _ = &*ERRORS;
+    let _ = &*FEED_LATENCY;
+}
+
+/// Update service uptime metric
+pub fn update_uptime(start_time: std::time::Instant) {
+    SERVICE_UPTIME.set(start_time.elapsed().as_secs_f64());
+}

services/ml_training_service/src/main.rs

@@ -15,6 +15,8 @@ use clap::{Args, Parser, Subcommand};
 use tonic::transport::Server;
 use tonic_reflection::server::Builder as ReflectionBuilder;
 use tracing::{debug, error, info, warn};
+
+use common::metrics::GrpcMetricsLayer;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
 // Import from library instead of duplicating module declarations
@@ -454,7 +456,8 @@ async fn serve(args: ServeArgs) -> Result<()> {
         .and_then(|v| v.parse().ok())
         .unwrap_or(true);
 
-    let mut server_builder = Server::builder();
+    let mut server_builder = Server::builder()
+        .layer(GrpcMetricsLayer::new("ml-training-service"));
 
     if enable_http2_opts {
         info!("HTTP/2 optimizations enabled:");

services/trading_agent_service/src/main.rs

@@ -12,6 +12,8 @@ use tokio::sync::Mutex;
 use tonic::transport::{Certificate, Identity, Server, ServerTlsConfig};
 use tracing::{error, info};
 
+use common::metrics::GrpcMetricsLayer;
+
 use common::DatabasePool;
 use config::manager::ConfigManager;
 use config::DatabaseConfig;
@@ -93,11 +95,13 @@ async fn main() -> Result<()> {
     info!("Starting gRPC server on {}", addr);
 
     // Build server with optional TLS
-    let mut server_builder = Server::builder();
+    let mut server_builder = Server::builder()
+        .layer(GrpcMetricsLayer::new("trading-agent-service"));
 
     if let Some(tls) = tls_config {
         info!("🔒 TLS enabled for Trading Agent Service");
         server_builder = Server::builder()
+            .layer(GrpcMetricsLayer::new("trading-agent-service"))
             .tls_config(tls)
             .context("Failed to apply TLS configuration")?;
     } else {

services/trading_service/src/main.rs

@@ -43,6 +43,8 @@ use trading_service::feedback_loop::{FeedbackLoop, FeedbackLoopConfig};
 use trading_service::questdb_metrics::QuestDBMetricsProvider;
 use trading_service::state::TradingServiceState;
 
+use common::metrics::GrpcMetricsLayer;
+
 /// Default configuration values
 const DEFAULT_GRPC_PORT: u16 = 50052; // Trading Service port (API Gateway uses 50051)
 const DEFAULT_HEALTH_PORT: u16 = 8081; // Health check port (separate from API Gateway 8080)
@@ -942,9 +944,11 @@ async fn main() -> Result<()> {
     // Apply authentication interceptor to all gRPC services
     let mut server_builder = match tls_config {
         Some(tls) => Server::builder()
+            .layer(GrpcMetricsLayer::new("trading-service"))
             .tls_config(tls)
             .context("Failed to configure TLS")?,
-        None => Server::builder(),
+        None => Server::builder()
+            .layer(GrpcMetricsLayer::new("trading-service")),
     };
 
     // Apply HTTP/2 optimizations if enabled