feat: Base metrics: add generic ingress handler metrics

Author: keivenchang

components/router/src/main.rs

@@ -66,14 +66,16 @@ async fn app(runtime: Runtime) -> Result<()> {
 
     let selector = Box::new(CustomWorkerSelector::default());
 
-    let router = KvRouter::new(component.clone(), args.block_size, Some(selector), true).await?;
-    let router = Ingress::for_engine(Arc::new(router))?;
-
-    component
+    let endpoint = component
         .service_builder()
         .create()
         .await?
-        .endpoint("generate")
+        .endpoint("generate");
+
+    let router = KvRouter::new(component.clone(), args.block_size, Some(selector), true).await?;
+    let router = Ingress::for_engine_with_metrics(Arc::new(router), &endpoint)?;
+
+    endpoint
         .endpoint_builder()
         .handler(router)
         .start()

deploy/metrics/prometheus.yml

@@ -38,7 +38,7 @@ scrape_configs:
   - job_name: 'llm-demo'
     scrape_interval: 10s
     static_configs:
-      - targets: ['host.docker.internal:8000']  # on the "monitoring" network
+      - targets: ['host.docker.internal:8080']  # on the "monitoring" network
 
   # This is another demo aggregator that needs to be launched manually. See components/metrics/README.md
   # Note that you may need to disable the firewall on your host. On Ubuntu: sudo ufw allow 9091/tcp

lib/llm/src/kv_router/publisher.rs

@@ -499,11 +499,11 @@ impl WorkerMetricsPublisher {
 
     pub async fn create_endpoint(&self, component: Component) -> Result<()> {
         let mut metrics_rx = self.rx.clone();
-        let handler = Arc::new(KvLoadEndpoingHander::new(metrics_rx.clone()));
-        let handler = Ingress::for_engine(handler)?;
+        let endpoint = component.endpoint(KV_METRICS_ENDPOINT);
+        let handler = Arc::new(KvLoadEndpoingHander::new(metrics_rx.clone(), endpoint.clone()));
+        let handler = Ingress::for_engine_with_metrics(handler, &endpoint)?;
 
-        component
-            .endpoint(KV_METRICS_ENDPOINT)
+        endpoint
             .endpoint_builder()
             .stats_handler(move |_| {
                 let metrics = metrics_rx.borrow_and_update().clone();
@@ -517,11 +517,16 @@ impl WorkerMetricsPublisher {
 
 struct KvLoadEndpoingHander {
     metrics_rx: tokio::sync::watch::Receiver<Arc<ForwardPassMetrics>>,
+    #[allow(dead_code)]
+    endpoint: dynamo_runtime::component::Endpoint,
 }
 
 impl KvLoadEndpoingHander {
-    pub fn new(metrics_rx: tokio::sync::watch::Receiver<Arc<ForwardPassMetrics>>) -> Self {
-        Self { metrics_rx }
+    pub fn new(
+        metrics_rx: tokio::sync::watch::Receiver<Arc<ForwardPassMetrics>>,
+        endpoint: dynamo_runtime::component::Endpoint,
+    ) -> Self {
+        Self { metrics_rx, endpoint }
     }
 }