diff --git a/.claude/skills/connector-runtime/SKILL.md b/.claude/skills/connector-runtime/SKILL.md index ce194c08e6..cee028180d 100644 --- a/.claude/skills/connector-runtime/SKILL.md +++ b/.claude/skills/connector-runtime/SKILL.md @@ -211,7 +211,7 @@ Fatal errors propagate to `main` and exit. Per-connector / per-message errors ar All families labeled by `connector_key` + `connector_type` (histogram adds `stage`): -- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered,errors}_total`. +- **Counters**: `iggy_connector_messages_{produced,sent,consumed,processed,filtered}_total` and `iggy_connector_errors_total`. These are the *rendered* names; each is registered without the `_total`, which the OpenMetrics encoder appends. - `messages_filtered_total` - intentional drops via transform `Ok(None)`. - `errors_total` - unexpected drops (decode/encode/build failure, missing field, ...) + batch-level failures. - **Histograms**: `iggy_connector_stage_duration_seconds{stage}` (snake_case stage labels - `prepare`, `ffi`, `decode`, `iggy_send`, `state_save`, `total`). Buckets `STAGE_BUCKETS_SECONDS`. Always populated regardless of any flag. Scraped at `/metrics` when `[http.metrics] enabled = true`. @@ -221,7 +221,7 @@ All families labeled by `connector_key` + `connector_type` (histogram adds `stag When adding a metric: -- Add family to `Metrics` struct + `init`, register with name + help text. +- Add family to `Metrics` struct + `init`, register with name + help text. Never end a `Counter` family's registered name in `_total`: the encoder appends it and the series renders `_total_total`. Gauges get no suffix, so a gauge name may end in `_total` literally. - New label sets define `EncodeLabelSet` struct + label enum (hand-impl `EncodeLabelValue` for snake_case values - the derive emits PascalCase). - Histograms: pass `fn() -> Histogram` to `Family::new_with_constructor`. - Add unit tests under `mod tests` with `given_*_when_*_should_*` BDD names. diff --git a/core/connectors/runtime/src/metrics.rs b/core/connectors/runtime/src/metrics.rs index ca6d79515c..71b5f93090 100644 --- a/core/connectors/runtime/src/metrics.rs +++ b/core/connectors/runtime/src/metrics.rs @@ -208,33 +208,35 @@ impl Metrics { "Sinks in Running status", sinks_running.clone(), ); + // Counter families are registered without the `_total` suffix: the + // OpenMetrics encoder appends it, so a literal one renders doubled. registry.register( - "iggy_connector_messages_produced_total", + "iggy_connector_messages_produced", "Messages received from source plugin poll", messages_produced.clone(), ); registry.register( - "iggy_connector_messages_sent_total", + "iggy_connector_messages_sent", "Messages sent to Iggy (source)", messages_sent.clone(), ); registry.register( - "iggy_connector_messages_consumed_total", + "iggy_connector_messages_consumed", "Messages consumed from Iggy (sink)", messages_consumed.clone(), ); registry.register( - "iggy_connector_messages_processed_total", + "iggy_connector_messages_processed", "Messages processed and sent to sink plugin", messages_processed.clone(), ); registry.register( - "iggy_connector_messages_filtered_total", + "iggy_connector_messages_filtered", "Messages intentionally dropped by transforms returning Ok(None)", messages_filtered.clone(), ); registry.register( - "iggy_connector_errors_total", + "iggy_connector_errors", "Errors encountered", errors.clone(), ); @@ -520,6 +522,83 @@ fn stage_histogram() -> Histogram { mod tests { use super::*; + /// Every series the registry renders once each family holds a sample. + /// Counters carry exactly one `_total`, gauges and histogram buckets carry + /// the suffixes the OpenMetrics encoder gives them. + const RENDERED_SERIES_NAMES: [&str; 13] = [ + "iggy_connectors_sources_total", + "iggy_connectors_sources_running", + "iggy_connectors_sinks_total", + "iggy_connectors_sinks_running", + "iggy_connector_messages_produced_total", + "iggy_connector_messages_sent_total", + "iggy_connector_messages_consumed_total", + "iggy_connector_messages_processed_total", + "iggy_connector_messages_filtered_total", + "iggy_connector_errors_total", + "iggy_connector_stage_duration_seconds_sum", + "iggy_connector_stage_duration_seconds_count", + "iggy_connector_stage_duration_seconds_bucket", + ]; + + /// Exercise every family so each one renders at least one sample: a + /// `Family` with no label set encodes its `# TYPE` header and nothing else, + /// which would hide the rendered series names from any assertion. + fn populated_metrics() -> Metrics { + let metrics = Metrics::init(); + metrics.set_sources_total(1); + metrics.set_sinks_total(1); + metrics.increment_sources_running(); + metrics.increment_sinks_running(); + metrics.increment_messages_produced("k", 1); + metrics.increment_messages_sent("k", 1); + metrics.increment_messages_consumed("k", 1); + metrics.increment_messages_processed("k", 1); + metrics.increment_messages_filtered("k", ConnectorType::Sink, 1); + metrics.increment_errors("k", ConnectorType::Sink); + metrics.observe_stage_duration( + "k", + ConnectorType::Sink, + Stage::Ffi, + Duration::from_micros(120), + ); + metrics + } + + /// Sample lines are `name{labels} value` or `name value`; comments and the + /// `# EOF` trailer are skipped. + fn rendered_series_names(output: &str) -> Vec { + let mut names: Vec = output + .lines() + .filter(|line| !line.is_empty() && !line.starts_with('#')) + .map(|line| line.split(['{', ' ']).next().unwrap_or_default().to_owned()) + .collect(); + names.sort_unstable(); + names.dedup(); + names + } + + #[test] + fn given_every_family_when_encoded_should_render_one_total_suffix() { + let output = populated_metrics().get_formatted_output(); + + assert!( + !output.contains("_total_total"), + "counter registered with a literal `_total` renders doubled:\n{output}" + ); + + let mut expected: Vec = RENDERED_SERIES_NAMES + .iter() + .map(|name| (*name).to_owned()) + .collect(); + expected.sort_unstable(); + assert_eq!( + rendered_series_names(&output), + expected, + "rendered series names drifted; dashboards key off these:\n{output}" + ); + } + #[test] fn test_metrics_init() { let metrics = Metrics::init(); diff --git a/core/integration/tests/connectors/runtime/benchmark.rs b/core/integration/tests/connectors/runtime/benchmark.rs index 77225a3653..063d83f408 100644 --- a/core/integration/tests/connectors/runtime/benchmark.rs +++ b/core/integration/tests/connectors/runtime/benchmark.rs @@ -435,6 +435,13 @@ async fn given_runtime_processing_batches_when_metrics_scraped_should_expose_sta "expected stage histogram in /metrics, got:\n{metrics_body}" ); + // Both connectors have processed batches by now, so every counter family + // renders a sample and a doubled suffix would be visible on the wire. + assert!( + !metrics_body.contains("_total_total"), + "counter names must carry exactly one `_total`, got:\n{metrics_body}" + ); + let sink_total_count = parse_metric_value( &metrics_body, "iggy_connector_stage_duration_seconds_count",