perf-sentinel-core 0.9.23

//! Alumet scraper task, HTTP client, and error types. See
//! `docs/design/05-GREENOPS-AND-CARBON.md` for the methodology and
//! `docs/LIMITATIONS.md#alumet-precision-bounds` for the precision
//! bounds.

use std::sync::Arc;
use std::time::Duration;

use crate::http_client::{self, FetchError, HttpClient};
use crate::ingest::auth_header::{AuthHeader, ScraperAuthOutcome, parse_scraper_auth_header};
use crate::report::metrics::{AlumetScrapeReason, MetricsState};
use crate::score::ops_snapshot_diff::OpsSnapshotDiff;
use crate::score::prom_parser::parse_metric_samples;

use super::apply::apply_scrape;
use super::config::AlumetConfig;
use super::state::{AlumetState, DbEnergyState, monotonic_ms};

/// Number of consecutive scrape failures before [`run_scraper_loop`]
/// emits the one-shot "likely misconfigured endpoint" warning. Same
/// rationale as Scaphandre's threshold.
const UNSUPPORTED_PLATFORM_FAILURE_THRESHOLD: u32 = 3;

/// Consecutive HTTP-200 ticks with zero matching samples before the
/// warn-once fires. Catches an operator-supplied `metric_name` that does
/// not exist on the wire, the most likely Alumet misconfiguration since
/// the exporter's `prefix`/`suffix` shape the name.
const ZERO_SAMPLE_WARN_THRESHOLD: u32 = 3;

/// Scrape the Alumet endpoint once via the hyper-util client.
pub(super) async fn fetch_metrics_once(
    client: &HttpClient,
    uri: &hyper::Uri,
    auth: Option<&AuthHeader>,
) -> Result<String, ScraperError> {
    let bytes = http_client::fetch_get(
        client,
        uri,
        "perf-sentinel/alumet-scraper",
        Duration::from_secs(3),
        auth,
    )
    .await
    .map_err(ScraperError::Fetch)?;
    String::from_utf8(bytes.to_vec()).map_err(ScraperError::Utf8)
}

/// Errors the scraper task might emit. Never returned to the caller,
/// logged via `tracing` with the warn-once pattern and the task keeps
/// running. URI parsing failures are handled separately at startup and
/// never reach this enum.
#[derive(Debug, thiserror::Error)]
pub(super) enum ScraperError {
    #[error("Alumet fetch failed")]
    Fetch(#[source] FetchError),
    #[error("Alumet response was not valid UTF-8")]
    Utf8(#[source] std::string::FromUtf8Error),
}

/// Map a [`ScraperError`] to the `reason` label used by
/// `perf_sentinel_alumet_scrape_failed_total`.
pub(super) fn scraper_error_reason(err: &ScraperError) -> AlumetScrapeReason {
    match err {
        ScraperError::Fetch(fe) => fetch_error_reason(fe),
        ScraperError::Utf8(_) => AlumetScrapeReason::InvalidUtf8,
    }
}

fn fetch_error_reason(err: &FetchError) -> AlumetScrapeReason {
    match err {
        FetchError::Transport(_) => AlumetScrapeReason::Unreachable,
        FetchError::Timeout => AlumetScrapeReason::Timeout,
        FetchError::HttpStatus(_) => AlumetScrapeReason::HttpError,
        FetchError::BodyRead(_) => AlumetScrapeReason::BodyReadError,
        FetchError::RequestBuild(_) => AlumetScrapeReason::RequestError,
    }
}

/// Spawn the periodic Alumet scraper task.
///
/// Returns a `JoinHandle` the daemon captures and aborts on Ctrl-C.
/// The task reads per-service op counts from `MetricsState`, scrapes the
/// Alumet endpoint, converts each interval-energy reading into a
/// kWh-per-op coefficient, and publishes via [`AlumetState::publish`].
#[must_use]
pub fn spawn_scraper(
    cfg: AlumetConfig,
    state: Arc<AlumetState>,
    db_state: Option<Arc<DbEnergyState>>,
    broker_state: Option<Arc<DbEnergyState>>,
    metrics: Arc<MetricsState>,
) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        run_scraper_loop(cfg, state, db_state, broker_state, metrics).await;
    })
}

// Same call as the Scaphandre and cloud scrapers: splitting the loop
// lifecycle across helpers fragments it without clarity gain.
#[allow(clippy::too_many_lines)]
async fn run_scraper_loop(
    cfg: AlumetConfig,
    state: Arc<AlumetState>,
    db_state: Option<Arc<DbEnergyState>>,
    broker_state: Option<Arc<DbEnergyState>>,
    metrics: Arc<MetricsState>,
) {
    use std::str::FromStr;

    let uri = match hyper::Uri::from_str(&cfg.endpoint) {
        Ok(u) => u,
        Err(e) => {
            // Defense in depth: validate_alumet already rejects `@` in
            // the authority, but log the redacted string in case a
            // future caller skips validation.
            tracing::error!(
                endpoint = %http_client::redact_endpoint_str(&cfg.endpoint),
                error = %e,
                "Alumet scraper aborting on invalid endpoint URI"
            );
            return;
        }
    };
    let redacted = http_client::redact_endpoint(&uri);

    let parsed_auth: Option<AuthHeader> = match parse_scraper_auth_header(
        cfg.auth_header.as_deref(),
        &cfg.endpoint,
        &redacted,
        "alumet",
    ) {
        ScraperAuthOutcome::Invalid => return,
        ScraperAuthOutcome::None => None,
        ScraperAuthOutcome::Some(h) => Some(h),
    };

    let client = http_client::build_client();

    let mut ticker = tokio::time::interval(cfg.scrape_interval);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    ticker.tick().await;

    let mut snapshot_diff = OpsSnapshotDiff::default();
    let mut failure_streak_warned = false;
    let mut consecutive_failures: u32 = 0;
    let mut unsupported_platform_warned = false;
    let mut latches = ScrapeLatches::default();
    // Track the last successful scrape so the `last_scrape_age_seconds`
    // gauge advances on every failure tick. Seeded to scraper-start time
    // so an Alumet endpoint broken from boot still climbs the gauge.
    let mut last_success_ms: u64 = monotonic_ms();

    // `energy_interval_secs` is echoed at startup on purpose: it is the
    // one config value that silently rescales every reading when it
    // drifts from the Alumet-side `poll_interval`, and it cannot be
    // cross-checked against the wire.
    tracing::info!(
        endpoint = %redacted,
        scrape_interval_secs = cfg.scrape_interval.as_secs(),
        metric = %cfg.metric_name,
        label = %cfg.label_key,
        energy_interval_secs = cfg.energy_interval_secs,
        service_count = cfg.service_mappings.len(),
        "Alumet scraper started"
    );
    // A database-only configuration is legitimate, no warning then.
    if cfg.service_mappings.is_empty() && cfg.database.is_none() {
        tracing::warn!(
            endpoint = %redacted,
            "[green.alumet] service_mappings is empty: the scraper will \
             poll the endpoint but can never attribute energy to a \
             service. Add mappings to publish measured coefficients."
        );
    }

    loop {
        ticker.tick().await;

        let current_ops = metrics.snapshot_service_io_ops();
        let deltas = snapshot_diff.delta_and_advance(current_ops);

        match fetch_metrics_once(&client, &uri, parsed_auth.as_ref()).await {
            Ok(body) => {
                failure_streak_warned = false;
                consecutive_failures = 0;
                let samples = parse_metric_samples(&body, &cfg.metric_name, &cfg.label_key);
                // Mirror Scaphandre: timestamp after the fetch resolves
                // so `last_update_ms` reflects when the data landed.
                let now = monotonic_ms();
                let matched = apply_scrape(
                    &state,
                    db_state.as_deref(),
                    broker_state.as_deref(),
                    &samples,
                    &deltas,
                    &cfg,
                    now,
                );
                last_success_ms = now;
                metrics.alumet_last_scrape_age_seconds.set(0.0);
                metrics.alumet_scrape_success.inc();
                post_scrape_bookkeeping(
                    &samples,
                    matched,
                    deltas.len(),
                    &cfg,
                    &redacted,
                    db_state.as_deref(),
                    broker_state.as_deref(),
                    now,
                    &mut latches,
                );
            }
            Err(e) => {
                consecutive_failures = consecutive_failures.saturating_add(1);
                // Reset the latches on HTTP failure: the failure-side
                // warning covers flapping endpoints.
                latches.reset_all();
                handle_alumet_failure(
                    &e,
                    &metrics,
                    &redacted,
                    last_success_ms,
                    monotonic_ms(),
                    consecutive_failures,
                    &mut failure_streak_warned,
                    &mut unsupported_platform_warned,
                );
            }
        }
    }
}

/// Success-branch liveness and diagnostics, extracted from
/// [`run_scraper_loop`] for the line-count limit. A successful scrape
/// proves the chain is alive: banked database energy survives idle
/// spells and label renames.
/// The four independent warn-once latches of one scraper loop, reset on
/// HTTP error so a flapping endpoint does not falsely trip them.
///
/// One streak per cause on purpose: a `metric_name` matching nothing and
/// a `service_mappings` table matching nothing are distinct
/// misconfigurations with distinct fixes, and sharing one counter would
/// let one cause fire another's message, or latch it away for good.
#[derive(Default)]
pub(super) struct ScrapeLatches {
    no_samples: WarnOnceStreak,
    no_match: WarnOnceStreak,
    db_missing: WarnOnceStreak,
    broker_missing: WarnOnceStreak,
}

impl ScrapeLatches {
    /// Reset every cause. Called on HTTP failure, where the
    /// failure-side warning already covers a flapping endpoint.
    pub(super) fn reset_all(&mut self) {
        self.no_samples.reset();
        self.no_match.reset();
        self.db_missing.reset();
        self.broker_missing.reset();
    }
}

#[allow(clippy::too_many_arguments)] // per-scrape context, each one distinct
pub(super) fn post_scrape_bookkeeping(
    samples: &[crate::score::prom_parser::PromSample],
    matched: usize,
    deltas_len: usize,
    cfg: &AlumetConfig,
    redacted: &str,
    db_state: Option<&DbEnergyState>,
    broker_state: Option<&DbEnergyState>,
    now_ms: u64,
    latches: &mut ScrapeLatches,
) {
    if let Some(db) = db_state {
        db.mark_alive(now_ms);
    }
    if let Some(broker) = broker_state {
        broker.mark_alive(now_ms);
    }
    track_zero_sample_streak(
        samples.len(),
        matched,
        deltas_len,
        cfg.service_mappings.len(),
        redacted,
        &cfg.metric_name,
        &cfg.label_key,
        &mut latches.no_samples,
        &mut latches.no_match,
    );
    track_workload_label_streak(
        samples,
        cfg.database.as_ref().map(|d| d.label_value.as_str()),
        "[green.alumet.database]",
        cfg,
        redacted,
        &mut latches.db_missing,
    );
    track_workload_label_streak(
        samples,
        cfg.broker.as_ref().map(|b| b.label_value.as_str()),
        "[green.alumet.broker]",
        cfg,
        redacted,
        &mut latches.broker_missing,
    );
}

/// Warn-once when a declared workload label never appears among
/// non-empty samples. An empty exposition belongs to the `no_samples`
/// cause and says nothing about the declared label.
///
/// `section` is the TOML section name, so one latch serves both the
/// database and the broker declaration.
pub(super) fn track_workload_label_streak(
    samples: &[crate::score::prom_parser::PromSample],
    declared: Option<&str>,
    section: &str,
    cfg: &AlumetConfig,
    redacted: &str,
    streak: &mut WarnOnceStreak,
) {
    let Some(label_value) = declared else {
        return;
    };
    if samples.is_empty() {
        return;
    }
    if samples.iter().any(|s| s.label_value == label_value) {
        streak.reset();
    } else if streak.tick() {
        tracing::warn!(
            endpoint = %redacted,
            label = %cfg.label_key,
            label_value = %label_value,
            section = %section,
            "Alumet samples flowed but the declared label_value was absent \
             under label_key across the last {ZERO_SAMPLE_WARN_THRESHOLD} such \
             ticks, so no energy is accumulating for that section. Either the \
             value is mistyped (it must match the wire verbatim) or the \
             workload is absent from the exposition."
        );
    }
}

/// Failure-branch bookkeeping: advance the staleness gauge, bump the
/// reason counter, log once at warn then debug, and emit the one-shot
/// "likely misconfigured" warning after three consecutive failures.
/// Extracted so [`run_scraper_loop`] stays under the line-count limit.
#[allow(clippy::too_many_arguments)]
fn handle_alumet_failure(
    err: &ScraperError,
    metrics: &MetricsState,
    redacted: &str,
    last_success_ms: u64,
    now_ms: u64,
    consecutive_failures: u32,
    failure_streak_warned: &mut bool,
    unsupported_platform_warned: &mut bool,
) {
    // Wall-clock age since the last successful scrape (or scraper
    // start time, when no success has happened yet) so Grafana alerts
    // on a hung scraper fire reliably from boot.
    let age_secs = now_ms.saturating_sub(last_success_ms) as f64 / 1000.0;
    metrics.alumet_last_scrape_age_seconds.set(age_secs);
    let reason = scraper_error_reason(err);
    metrics.alumet_scrape_failed.inc();
    metrics
        .alumet_scrape_failed_total
        .with_label_values(&[reason.as_str()])
        .inc();
    if *failure_streak_warned {
        tracing::debug!(error = %err, "Alumet scrape failed again");
    } else {
        tracing::warn!(
            error = %err,
            endpoint = %redacted,
            "Alumet scrape failed; subsequent failures will log at debug level"
        );
        *failure_streak_warned = true;
    }
    if !*unsupported_platform_warned
        && consecutive_failures >= UNSUPPORTED_PLATFORM_FAILURE_THRESHOLD
    {
        tracing::warn!(
            endpoint = %redacted,
            consecutive_failures = consecutive_failures,
            "Alumet endpoint has been unreachable for {UNSUPPORTED_PLATFORM_FAILURE_THRESHOLD} consecutive scrapes. \
             Check that the Alumet agent is running with the prometheus-exporter plugin enabled \
             and serving metrics at the configured endpoint. \
             The daemon is falling back through the precedence chain for affected services. \
             See docs/LIMITATIONS.md#alumet-precision-bounds."
        );
        *unsupported_platform_warned = true;
    }
}

/// One warn-once streak: counts consecutive bad ticks for a single
/// cause and fires at most once per streak. Shared with the Kepler
/// scraper, which imports it the same way `kepler::state` borrows the
/// Scaphandre monotonic clock.
#[derive(Default)]
pub(crate) struct WarnOnceStreak {
    ticks: u32,
    warned: bool,
}

impl WarnOnceStreak {
    /// Record one bad tick. Returns `true` exactly when the warn should
    /// fire now (threshold reached, not yet warned this streak).
    pub(crate) fn tick(&mut self) -> bool {
        self.ticks = self.ticks.saturating_add(1);
        if !self.warned && self.ticks >= ZERO_SAMPLE_WARN_THRESHOLD {
            self.warned = true;
            return true;
        }
        false
    }

    pub(crate) fn reset(&mut self) {
        self.ticks = 0;
        self.warned = false;
    }

    #[cfg(test)]
    pub(crate) fn has_warned(&self) -> bool {
        self.warned
    }
}

/// Success-branch latches: warn once per streak of
/// [`ZERO_SAMPLE_WARN_THRESHOLD`] consecutive HTTP-200 ticks that
/// produced nothing usable. Extracted for the line-count limit and to
/// unit-test the warn-once edges.
///
/// Two independent causes, two independent streaks:
///
/// - `no_samples` fires when `metric_name`/`label_key` match nothing on
///   the wire. A tick with no samples says nothing about the mappings,
///   so it neither advances nor resets `no_match`.
/// - `no_match` fires when samples flow but zero `service_mappings`
///   label values are present. That is either mistyped mapping values
///   or every mapped workload currently absent from the exposition, the
///   message names both since the wire cannot tell them apart. Gated on
///   a non-empty mappings table (an empty table trivially matches
///   nothing and gets its own startup warning instead).
///
/// A partially wrong table (some mappings match, others never do) trips
/// neither latch, the per-tick `services_matched` debug field and the
/// report-level `per_service_energy_model` are the signals for that.
#[allow(clippy::too_many_arguments)]
pub(super) fn track_zero_sample_streak(
    samples_len: usize,
    services_matched: usize,
    services_with_ops: usize,
    mapping_count: usize,
    redacted: &str,
    metric_name: &str,
    label_key: &str,
    no_samples: &mut WarnOnceStreak,
    no_match: &mut WarnOnceStreak,
) {
    if samples_len == 0 {
        if no_samples.tick() {
            tracing::warn!(
                endpoint = %redacted,
                metric = metric_name,
                label = label_key,
                "Alumet endpoint replied HTTP 200 but no samples matched \
                 the configured metric across the last {ZERO_SAMPLE_WARN_THRESHOLD} ticks. \
                 Most common cause: metric_name does not match the wire. \
                 Alumet's prometheus-exporter prepends `prefix` and \
                 appends `suffix` (default '_alumet') to every metric \
                 name, and an energy-attribution series is named after \
                 the operator's formula. Run \
                 `curl <endpoint> | grep -i energy` and copy the name \
                 verbatim. Other cause: label_key absent from the series.",
            );
        }
    } else {
        no_samples.reset();
        if services_matched == 0 && mapping_count > 0 {
            if no_match.tick() {
                tracing::warn!(
                    endpoint = %redacted,
                    metric = metric_name,
                    label = label_key,
                    samples = samples_len,
                    "Alumet endpoint replied HTTP 200 and metric_name matched \
                     samples, but none of the configured service_mappings \
                     label values were present under label_key across the \
                     last {ZERO_SAMPLE_WARN_THRESHOLD} such ticks, so no measured \
                     coefficient is being published. Either the mapping \
                     values are mistyped (they must match the label value \
                     verbatim), or every mapped workload is currently absent \
                     from the exposition (scaled to zero, not yet scheduled). \
                     Inspect the live values for the configured label key \
                     with curl against the endpoint.",
                );
            }
        } else {
            no_match.reset();
        }
    }
    tracing::debug!(
        samples = samples_len,
        services_matched = services_matched,
        services_with_ops = services_with_ops,
        "Alumet scrape succeeded"
    );
}