This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 3cd0860a9 fix(connectors): stop rendering counter metrics with a
doubled _total (#3921)
3cd0860a9 is described below
commit 3cd0860a97492cc566d3c2e3b1752c18168b7580
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Wed Aug 19 13:09:30 2026 +0200
fix(connectors): stop rendering counter metrics with a doubled _total
(#3921)
The runtime registered its six counter families under names that
already ended in _total, and the OpenMetrics encoder appends that
suffix itself, so every scrape exposed names like
iggy_connector_errors_total_total. The runtime README documented the
un-doubled names all along, so no scraper ever saw what the docs
promised.
Drop the literal suffix at registration and let the encoder add it.
Gauges stay as they are: the encoder appends nothing to them, so
their names already rendered correctly, including the ones that end
in _total.
The existing tests could not catch this. They asserted with
contains() on a name that is a substring of the doubled one, which
passes on broken and on fixed code alike. The new test compares the
full set of rendered series names, so a silent rename fails too, and
the live-runtime scrape test now rejects a doubled suffix on the
wire.
This changes operator-facing series names, so anything scraping the
doubled names has to be updated.
---
.claude/skills/connector-runtime/SKILL.md | 4 +-
core/connectors/runtime/src/metrics.rs | 91 ++++++++++++++++++++--
.../tests/connectors/runtime/benchmark.rs | 7 ++
3 files changed, 94 insertions(+), 8 deletions(-)
diff --git a/.claude/skills/connector-runtime/SKILL.md
b/.claude/skills/connector-runtime/SKILL.md
index ce194c08e..cee028180 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 ca6d79515..71b5f9309 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<String> {
+ let mut names: Vec<String> = 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<String> = 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 77225a365..063d83f40 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",
