Metrics

rust-camel records route and exchange metrics through the MetricsCollector trait. PrometheusMetrics is the production implementation. You can supply your own.

Custom metrics collector

Implement MetricsCollector to receive route-level metrics. The runtime calls your methods as exchanges flow, errors fire, and circuit breakers change state. The trait covers exchange duration, error and exchange counts, queue depth, and circuit-breaker transitions.

/// A simple console-based metrics collector that prints stats on shutdown.
#[derive(Debug)]
struct ConsoleMetrics {
    exchanges_processed: AtomicU64,
    errors: AtomicU64,
    total_duration_ns: AtomicU64,
}

impl ConsoleMetrics {
    fn new() -> Self {
        Self {
            exchanges_processed: AtomicU64::new(0),
            errors: AtomicU64::new(0),
            total_duration_ns: AtomicU64::new(0),
        }
    }

    fn print_stats(&self) {
        let processed = self.exchanges_processed.load(Ordering::Relaxed);
        let errors = self.errors.load(Ordering::Relaxed);
        let total_ns = self.total_duration_ns.load(Ordering::Relaxed);
        let avg_ms = (total_ns.checked_div(processed).unwrap_or(0)) as f64 / 1_000_000.0;

        println!(
            "[METRICS] exchanges={} errors={} avg_duration_ms={:.2}",
            processed, errors, avg_ms
        );
    }
}

impl MetricsCollector for ConsoleMetrics {
    fn record_exchange_duration(&self, _route_id: &str, duration: Duration) {
        self.exchanges_processed.fetch_add(1, Ordering::Relaxed);
        self.total_duration_ns
            .fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
    }

    fn increment_errors(&self, _route_id: &str, _error_type: &str) {
        self.errors.fetch_add(1, Ordering::Relaxed);
    }

    fn increment_exchanges(&self, _route_id: &str) {
        self.exchanges_processed.fetch_add(1, Ordering::Relaxed);
    }

    fn set_queue_depth(&self, _route_id: &str, _depth: usize) {
        // Not tracked in this simple example
    }

    fn record_circuit_breaker_change(&self, _route_id: &str, from: &str, to: &str) {
        println!("[METRICS] Circuit breaker: {} -> {}", from, to);
    }
}

Register the collector with CamelContext::builder().metrics(...):

    // Create context with custom metrics collector
    let mut ctx = CamelContext::builder().metrics(metrics).build().await?;

    ctx.register_component(TimerComponent::new());
    ctx.register_component(LogComponent::new());

    let route = RouteBuilder::from("timer:metrics?period=500&repeatCount=10")
        .route_id("metrics-demo")
        .process(|mut exchange| {
            let start = Instant::now();
            async move {
                // Simulate some work
                tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

                exchange.input.body = Body::Json(serde_json::json!({
                    "message": "processed",
                    "elapsed_us": start.elapsed().as_micros(),
                }));
                Ok(exchange)
            }
        })
        .to("log:output?showBody=true&showCorrelationId=true")
        .build()?;

    ctx.add_route_definition(route).await?;

Note: Service registration is Rust API only. YAML routes compile to the same RouteDefinition. The service wiring stays in application code.

routes:
  - id: metrics-demo
    from: timer:metrics?period=500&repeatCount=10
    steps:
      # The Rust route measures elapsed time and sets the body in an
      # inline process() closure. Move that logic into a bean step.
      - set_body:
          value:
            message: processed
      - to: log:output?showBody=true&showCorrelationId=true

Reference: MetricsCollector contract

Prometheus integration

PrometheusService is a Lifecycle implementation. It owns a PrometheusMetrics collector and an HTTP server, exposes metrics at /metrics, and merges the health routes from camel-health.

    let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 9090);
    let prometheus = PrometheusService::new(addr);

    let mut ctx = CamelContext::builder()
        .build()
        .await
        .unwrap() // allow-unwrap
        .with_lifecycle(prometheus)
        .with_tracing()
        .await;

Note: Service registration is Rust API only. YAML routes compile to the same RouteDefinition. The service wiring stays in application code.

The server exposes these endpoints:

Exposure posture

Diagnostic endpoints follow the Prometheus scrape convention: unauthenticated by default, with TLS and auth as opt-in hooks. The default bind address prefers loopback. Binding to 0.0.0.0 emits a startup warning.

See ADR-0052 for the full posture.

Cardinality contract

Metric label values must come from a closed or bounded set. Never pass raw Exchange body, header, property, or correlation-key data as a label value. Each distinct label combination creates a new Prometheus series. The registry has no cardinality cap or eviction.

See ADR-0032 for the exchange-data trust boundary.

Reference: camel-prometheus crate