From e2ee0407d7494bdf14678c83b075a932f8ee6efd Mon Sep 17 00:00:00 2001 From: Chet Nichols III Date: Tue, 14 Jul 2026 23:32:45 -0700 Subject: [PATCH] feat(scout): give the on-host discovery agent a /metrics endpoint scout -- the on-host control plane -- was invisible to Prometheus: no metrics and no /metrics endpoint, only logs. This stands one up through the shared metrics-endpoint crate and counts the control loop's work. - With `--metrics-listen-addr` set, scout installs the framework meter, registers `carbide_log_events_total` (so the existing error/warn sites gain an error-rate signal for free), and serves `/metrics`. Without the flag scout stays a pure client as before -- the endpoint is opt-in, and the counters record nothing until it is configured. - `carbide_scout_actions_total{action, outcome}` counts each control-loop action by kind and result; `carbide_scout_stream_connections_total{outcome}` and `carbide_scout_stream_reconnects_total` track the ScoutStream lifecycle. - A failed log-error action now propagates instead of being swallowed, so its outcome records as an error rather than a silent success. - The existing `error!`/`warn!` lines stay as logs -- their rate rides on `carbide_log_events_total`, so there are no per-site metric siblings. scout's /metrics is binary-local (not scraped by `test_integration`), so there is no catalogue change. This supports https://github.com/NVIDIA/infra-controller/issues/3173 Signed-off-by: Chet Nichols III --- Cargo.lock | 3 + crates/scout/Cargo.toml | 4 + crates/scout/src/cfg/command_line.rs | 9 ++ crates/scout/src/main.rs | 51 ++++++- crates/scout/src/metrics.rs | 205 +++++++++++++++++++++++++++ crates/scout/src/stream.rs | 27 +++- 6 files changed, 290 insertions(+), 9 deletions(-) create mode 100644 crates/scout/src/metrics.rs diff --git a/Cargo.lock b/Cargo.lock index f61320fa9c..143fdb352c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2896,9 +2896,11 @@ version = "0.0.1" dependencies = [ "axum", "carbide-host-support", + "carbide-instrument", "carbide-libmlx", "carbide-machine-validation", "carbide-rpc", + "carbide-test-support", "carbide-tls", "carbide-utils", "carbide-uuid", @@ -2911,6 +2913,7 @@ dependencies = [ "hex", "http", "lazy_static", + "metrics-endpoint", "once_cell", "pwhash", "regex", diff --git a/crates/scout/Cargo.toml b/crates/scout/Cargo.toml index cc28995899..8a1f7e5040 100644 --- a/crates/scout/Cargo.toml +++ b/crates/scout/Cargo.toml @@ -34,12 +34,14 @@ path = "src/main.rs" # [local-dependencies] carbide-rpc = { path = "../rpc" } carbide-host-support = { path = "../host-support" } +carbide-instrument = { path = "../instrument" } carbide-tls = { path = "../tls" } carbide-uuid = { path = "../uuid" } carbide-version = { path = "../version" } carbide-utils = { path = "../utils" } carbide-machine-validation = { path = "../machine-validation" } carbide-libmlx = { path = "../libmlx" } +metrics-endpoint = { path = "../metrics-endpoint" } chrono = { workspace = true } clap = { workspace = true } @@ -73,6 +75,8 @@ x509-parser = { workspace = true } [dev-dependencies] axum = { workspace = true } +carbide-instrument = { path = "../instrument", features = ["test-support"] } +carbide-test-support = { path = "../test-support" } [build-dependencies] carbide-version = { path = "../version" } diff --git a/crates/scout/src/cfg/command_line.rs b/crates/scout/src/cfg/command_line.rs index 9ef8c60f94..853e667814 100644 --- a/crates/scout/src/cfg/command_line.rs +++ b/crates/scout/src/cfg/command_line.rs @@ -14,6 +14,8 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +use std::net::SocketAddr; + use carbide_uuid::machine_validation::MachineValidationId; use clap::{Parser, Subcommand, ValueEnum}; use forge_tls::default as tls_default; @@ -108,6 +110,13 @@ pub(crate) struct Options { )] pub tpm_path: String, + #[clap( + long, + help = "HTTP listen address for the metrics/health endpoint (e.g. 127.0.0.1:9091). \ + When omitted the endpoint is not served and no metrics are collected." + )] + pub metrics_listen_addr: Option, + #[clap(subcommand)] pub subcmd: Option, } diff --git a/crates/scout/src/main.rs b/crates/scout/src/main.rs index 10c334e89e..7e5c9fba08 100644 --- a/crates/scout/src/main.rs +++ b/crates/scout/src/main.rs @@ -22,6 +22,7 @@ use std::time::Duration; use carbide_host_support::dpa_cmds::{DpaCommand, OpCode}; use carbide_host_support::registration; +use carbide_instrument::{Outcome, emit}; use carbide_uuid::machine::MachineId; use cfg::{AutoDetect, Command, MlxAction, Mode, Options}; use chrono::{DateTime, Days, TimeDelta, Utc}; @@ -54,6 +55,7 @@ mod deprovision; mod discovery; mod firmware_upgrade; mod machine_validation; +mod metrics; mod mlx_device; mod platform; mod register; @@ -167,6 +169,35 @@ async fn initial_setup(config: &Options) -> Result<(uuid::Uuid, MachineId), eyre } async fn run_as_service(config: &Options) -> Result<(), eyre::Report> { + // Stand up the metrics/scrape endpoint when it is configured. The meter + // provider must stay alive for the lifetime of the service: dropping it + // shuts down the Prometheus exporter. The endpoint is opt-in -- without + // --metrics-listen-addr scout installs no meter and the counters below are + // no-ops. + let _metrics_guard = match config.metrics_listen_addr { + Some(address) => { + let metrics_setup = + metrics_endpoint::new_metrics_setup("nico-scout", "forge-system", true)?; + carbide_instrument::log_events::register(&metrics_setup.meter); + let metrics_config = metrics_endpoint::MetricsEndpointConfig { + address, + registry: metrics_setup.registry, + health_controller: Some(metrics_setup.health_controller), + additional_prefix: None, + }; + // The endpoint's /health and /ready report process liveness (the + // default HealthController state), not scout readiness. + tokio::spawn(async move { + tracing::info!("Spawning metrics endpoint on {}", metrics_config.address); + if let Err(e) = metrics_endpoint::run_metrics_endpoint(&metrics_config).await { + tracing::error!("Metrics endpoint error: {e}"); + } + }); + Some(metrics_setup.meter_provider) + } + None => None, + }; + // Implement the logic to run as a service here let (machine_interface_id, machine_id) = initial_setup(config).await?; @@ -210,7 +241,14 @@ async fn run_as_service(config: &Options) -> Result<(), eyre::Report> { }; if let Some(action) = controller_response.action { let action_str = action.as_str_name().to_owned(); - match handle_action(action, &machine_id, machine_interface_id, config).await { + // Capture the action label before handle_action consumes `action`. + let scout_action = metrics::ScoutAction::from(&action); + let result = handle_action(action, &machine_id, machine_interface_id, config).await; + emit(metrics::ScoutActionHandled { + action: scout_action, + outcome: Outcome::from(&result), + }); + match result { Ok(_) => tracing::info!("Successfully served {}", action_str), Err(e) => tracing::info!("Failed to serve {}: Err {}", action_str, e), }; @@ -348,10 +386,13 @@ async fn handle_action( unimplemented!("Rebuild not written yet"); } fac::Action::Noop(_) => {} - fac::Action::LogError(_) => match logerror_to_carbide(config, machine_interface_id).await { - Ok(()) => (), - Err(e) => tracing::info!("Forge Scout logerror_to_carbide error: {}", e), - }, + fac::Action::LogError(_) => logerror_to_carbide(config, machine_interface_id) + .await + // Propagate the failure so `carbide_scout_actions_total` records this + // as `outcome = error`, not a silent success. + .map_err(|e| { + CarbideClientError::GenericError(format!("logerror_to_carbide failed: {e}")) + })?, fac::Action::Retry(_) => { panic!( "Retrieved Retry action, which should be handled internally by query_api_with_retries" diff --git a/crates/scout/src/metrics.rs b/crates/scout/src/metrics.rs new file mode 100644 index 0000000000..316d31e618 --- /dev/null +++ b/crates/scout/src/metrics.rs @@ -0,0 +1,205 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +//! Control-loop and stream counters for scout. Every event here is metric-only +//! (`log = off`): the human-readable side is already carried by the existing +//! `tracing` lines at each site, and the process-wide log rate rides on +//! `carbide_log_events_total{level, component}` once `log_events::register` +//! runs. These counters add the per-action and per-stream-outcome rates that a +//! log line alone can't total. + +use carbide_instrument::{Event, LabelValue, Outcome}; +use rpc::forge_agent_control_response as fac; + +/// Which control-loop action scout handled, as a bounded metric label: one +/// variant per [`fac::Action`] arm the service loop can dispatch. +#[derive(Debug, Clone, Copy, PartialEq, Eq, LabelValue)] +pub enum ScoutAction { + Noop, + Reset, + Discovery, + Rebuild, + Retry, + Measure, + LogError, + MachineValidation, + MlxAction, + FirmwareUpgrade, +} + +impl From<&fac::Action> for ScoutAction { + fn from(action: &fac::Action) -> Self { + match action { + fac::Action::Noop(_) => Self::Noop, + fac::Action::Reset(_) => Self::Reset, + fac::Action::Discovery(_) => Self::Discovery, + fac::Action::Rebuild(_) => Self::Rebuild, + fac::Action::Retry(_) => Self::Retry, + fac::Action::Measure(_) => Self::Measure, + fac::Action::LogError(_) => Self::LogError, + fac::Action::MachineValidation(_) => Self::MachineValidation, + fac::Action::MlxAction(_) => Self::MlxAction, + fac::Action::FirmwareUpgrade(_) => Self::FirmwareUpgrade, + } + } +} + +/// Scout finished handling one control-loop action, whatever the result. +#[derive(Event)] +#[event( + name = "carbide_scout_actions_total", + component = "nico-scout", + log = off, + metric = counter, + describe = "Number of scout control-loop actions handled, by action and outcome." +)] +pub struct ScoutActionHandled { + #[label] + pub action: ScoutAction, + #[label] + pub outcome: Outcome, +} + +/// A scout stream connection attempt resolved -- `ok` once the bidirectional +/// stream is established, `error` when the client could not be built. +#[derive(Event)] +#[event( + name = "carbide_scout_stream_connections_total", + component = "nico-scout", + log = off, + metric = counter, + describe = "Number of scout stream connection attempts, by outcome." +)] +pub struct ScoutStreamConnection { + #[label] + pub outcome: Outcome, +} + +/// The scout stream closed or errored and the loop looped back to re-establish +/// it after the reconnect delay. +#[derive(Event)] +#[event( + name = "carbide_scout_stream_reconnects_total", + component = "nico-scout", + log = off, + metric = counter, + describe = "Number of scout stream reconnect cycles after a stream closed or errored." +)] +pub struct ScoutStreamReconnect {} + +#[cfg(test)] +mod tests { + use carbide_instrument::emit; + use carbide_instrument::testing::MetricsCapture; + use carbide_test_support::{Check, check_values}; + + use super::*; + + #[test] + fn scout_action_maps_every_dispatchable_action() { + check_values( + [ + Check { + scenario: "noop", + input: fac::Action::Noop(fac::Noop {}), + expect: ScoutAction::Noop, + }, + Check { + scenario: "reset", + input: fac::Action::Reset(fac::Reset {}), + expect: ScoutAction::Reset, + }, + Check { + scenario: "discovery", + input: fac::Action::Discovery(fac::Discovery {}), + expect: ScoutAction::Discovery, + }, + Check { + scenario: "rebuild", + input: fac::Action::Rebuild(fac::Rebuild {}), + expect: ScoutAction::Rebuild, + }, + Check { + scenario: "retry", + input: fac::Action::Retry(fac::Retry {}), + expect: ScoutAction::Retry, + }, + Check { + scenario: "measure", + input: fac::Action::Measure(fac::Measure {}), + expect: ScoutAction::Measure, + }, + Check { + scenario: "log error", + input: fac::Action::LogError(fac::LogError {}), + expect: ScoutAction::LogError, + }, + Check { + scenario: "machine validation", + input: fac::Action::MachineValidation(fac::MachineValidation::default()), + expect: ScoutAction::MachineValidation, + }, + Check { + scenario: "mlx action", + input: fac::Action::MlxAction(fac::MlxAction::default()), + expect: ScoutAction::MlxAction, + }, + Check { + scenario: "firmware upgrade", + input: fac::Action::FirmwareUpgrade(fac::FirmwareUpgrade::default()), + expect: ScoutAction::FirmwareUpgrade, + }, + ], + |action| ScoutAction::from(&action), + ); + } + + #[test] + fn scout_counters_move_per_label() { + let metrics = MetricsCapture::start(); + + // Labels chosen so no other test in this binary shares them. + emit(ScoutActionHandled { + action: ScoutAction::FirmwareUpgrade, + outcome: Outcome::Error, + }); + emit(ScoutStreamConnection { + outcome: Outcome::Ok, + }); + emit(ScoutStreamReconnect {}); + emit(ScoutStreamReconnect {}); + + assert_eq!( + metrics.counter_delta( + "carbide_scout_actions_total", + &[("action", "firmware_upgrade"), ("outcome", "error")], + ), + 1.0 + ); + assert_eq!( + metrics.counter_delta( + "carbide_scout_stream_connections_total", + &[("outcome", "ok")], + ), + 1.0 + ); + assert_eq!( + metrics.counter_delta("carbide_scout_stream_reconnects_total", &[]), + 2.0 + ); + } +} diff --git a/crates/scout/src/stream.rs b/crates/scout/src/stream.rs index a2d7a4307b..46bdfd0c46 100644 --- a/crates/scout/src/stream.rs +++ b/crates/scout/src/stream.rs @@ -17,6 +17,7 @@ use std::time::Duration; +use carbide_instrument::{Outcome, emit}; use carbide_uuid::machine::MachineId; use libmlx::profile::error::MlxProfileError; use rpc::forge::ScoutStreamApiBoundMessage; @@ -24,6 +25,7 @@ use rpc::protos::forge::{scout_stream_api_bound_message, scout_stream_scout_boun use tokio::sync::mpsc; use crate::cfg::Options; +use crate::metrics::{ScoutStreamConnection, ScoutStreamReconnect}; use crate::{client, mlx_device}; // ScoutStreamError represents errors that can @@ -71,6 +73,7 @@ pub fn start_scout_stream(machine_id: MachineId, options: &Options) -> tokio::ta ); } } + emit(ScoutStreamReconnect {}); tracing::warn!( "scout stream reconnecting (api:{}, machine_id:{machine_id}): 10s delay", options.api @@ -86,9 +89,12 @@ async fn run_scout_stream_loop( machine_id: MachineId, options: &Options, ) -> Result<(), ScoutStreamError> { - let mut client = client::create_forge_client(options) - .await - .map_err(|e| ScoutStreamError::ClientError(e.to_string()))?; + let mut client = client::create_forge_client(options).await.map_err(|e| { + emit(ScoutStreamConnection { + outcome: Outcome::Error, + }); + ScoutStreamError::ClientError(e.to_string()) + })?; // Create channels for bidirectional streaming. let (tx, rx) = mpsc::channel::(100); @@ -110,7 +116,20 @@ async fn run_scout_stream_loop( })?; // Now create the response handler. - let mut response_stream = client.scout_stream(request_stream).await?.into_inner(); + let mut response_stream = client + .scout_stream(request_stream) + .await + .map_err(|e| { + emit(ScoutStreamConnection { + outcome: Outcome::Error, + }); + ScoutStreamError::from(e) + })? + .into_inner(); + + emit(ScoutStreamConnection { + outcome: Outcome::Ok, + }); tracing::info!( "scout stream connection established (api:{}, machine_id:{machine_id})",