Instrument kafka_send with Counter, Latency Histogram, and In-Flight Gauge

Outcome

Add three Prometheus metrics directly inside the kafka_send static async method in rust-capture/src/sinks/kafka_event_sink.rs. Because send_batch fans out by spawning one kafka_send task per event, instrumenting kafka_send automatically covers both the single-event path (KafkaSink::send) and all per-event paths inside batch sends.

Metrics to be added:

Metric	Type	Labels	When emitted
`kafka_producer_in_flight_messages`	Gauge	`topic`	Incremented just before `producer.send()` is called; decremented on delivery resolution (success or failure)
`kafka_producer_produce_duration_seconds`	Histogram	`topic`	Observed after the delivery future resolves (success path only); measures time from just before `producer.send()` to delivery ack
`kafka_producer_messages_total`	Counter	`topic`, `status`	Incremented after the delivery future resolves; `status="success"` or `status="failure"`

Acceptance Criteria

Success counter – Given a single event is sent via KafkaSink::send and the broker acknowledges it, when GET /metrics is scraped, then kafka_producer_messages_total{topic="<topic>",status="success"} has been incremented by 1.
Failure counter + error propagation – Given a single event is sent via KafkaSink::send and the broker returns an error, when GET /metrics is scraped, then kafka_producer_messages_total{topic="<topic>",status="failure"} has been incremented by 1 and the Err is still returned from kafka_send (existing error-propagation behavior is unchanged).
In-flight gauge lifecycle – Given a message is in-flight (i.e., producer.send() has been called but has not yet resolved), when GET /metrics is scraped, then kafka_producer_in_flight_messages{topic="<topic>"} is greater than zero; after the delivery future resolves the gauge returns to its pre-call value.
Latency histogram observation – Given a successful delivery, when GET /metrics is scraped, then kafka_producer_produce_duration_seconds{topic="<topic>"} contains at least one observation with a positive duration.
Batch fan-out coverage – Each event processed by send_batch emits all three metrics individually (one counter increment, one histogram observation on the success path, and one gauge increment/decrement pair), not just a single aggregate per batch call.
No behavior change – The existing send_batch error-swallowing behavior (debug-logged join results) is unchanged. Metric emission must not alter the Ok(()) return from send_batch.
Serialization errors unaffected – If serde_json::to_string(&event) fails (early ? return before producer.send() is called), no gauge increment or counter bump occurs. This is correct: no in-flight message was ever enqueued.

Dependencies

task_1 (blocking predecessor): task_1 is expected to extend setup_metrics_recorder() in metrics_recorder.rs to register custom histogram bucket boundaries for kafka_producer_produce_duration_seconds via an additional set_buckets_for_metric call. This task must not be merged before task_1 is complete; otherwise the histogram will be emitted with the Prometheus exporter's default buckets rather than the tight sub-second buckets specified in FR-2.

Codebase Context

Key file: `rust-capture/src/sinks/kafka_event_sink.rs`

Current kafka_send body (lines 55–107):

async fn kafka_send(
    producer: FutureProducer,
    topic: String,
    event: EventTypes,
) -> Result<()> {
    tracing::debug!("Sending event to kafka: {:?}", event);
    let payload = serde_json::to_string(&event)?;       // early-return if serialization fails
    let key = event.key();
    let timeout = Timeout::Never;
    let partition = get_partition(&event.project_id(), 18);
    let record = FutureRecord::to(topic.as_str())
        .partition(partition)
        .key(&key)
        .payload(&payload);

    let result = producer.send(record, timeout).await;  // delivery future

    match result {
        Ok((partition, offset)) => {
            tracing::debug!("Message delivered to partition {} at offset {}", partition, offset);
        }
        Err((err, _)) => {
            return Err(anyhow!("failed to produce event {}", err));
        }
    }
    Ok(())
}

KafkaSink::send (lines 112–119) calls Self::kafka_send(self.producer.clone(), self.topic.clone(), event).await.

send_batch (lines 120–142) clones producer and topic per event, spawns Self::kafka_send(producer, topic, event) into a JoinSet, joins all tasks, and returns Ok(()) unconditionally (join results are debug-logged and discarded).

Existing `metrics_recorder.rs`

setup_metrics_recorder() already installs the global Prometheus recorder with PrometheusBuilder::new()...install_recorder().unwrap(). It currently registers one bucket set for http_requests_duration_seconds. task_1 will add the equivalent for kafka_producer_produce_duration_seconds.

track_metrics middleware demonstrates the correct macro call patterns already in use:

metrics::increment_counter!("http_requests_total", &labels);
metrics::histogram!("http_requests_duration_seconds", latency, &labels);

Crate versions (`rust-capture/Cargo.toml`)

metrics = "0.21.0"
metrics-exporter-prometheus = "0.12.1"

No new dependencies are required.

Implementation Approach

File changed: `rust-capture/src/sinks/kafka_event_sink.rs` only

1. Add `use std::time::Instant;`

Add at the top of the file alongside existing use declarations. (Instant is already used in metrics_recorder.rs but is not yet imported in the sink file.)

2. Instrument `kafka_send`

Insert the following steps inside the body of kafka_send. All additions are after the existing serialization + record-construction block (i.e., after let record = FutureRecord::to(...)), so serialization errors still early-return cleanly without touching any gauge or counter.

Step A – Increment in-flight gauge and capture start time (just before producer.send()):

metrics::gauge!("kafka_producer_in_flight_messages", 1.0_f64, "topic" => topic.clone());
let send_start = Instant::now();

Step B – Await the delivery future (existing producer.send(record, timeout).await — unchanged).

Step C – Emit metrics in both match arms:

Ok arm (after the existing tracing::debug!):

let elapsed = send_start.elapsed().as_secs_f64();
metrics::gauge!("kafka_producer_in_flight_messages", -1.0_f64, "topic" => topic.clone());
metrics::histogram!("kafka_producer_produce_duration_seconds", elapsed, "topic" => topic.clone());
metrics::increment_counter!("kafka_producer_messages_total", "topic" => topic.clone(), "status" => "success");

Err arm (before the return Err(...)):

metrics::gauge!("kafka_producer_in_flight_messages", -1.0_f64, "topic" => topic.clone());
metrics::increment_counter!("kafka_producer_messages_total", "topic" => topic.clone(), "status" => "failure");
return Err(anyhow!("failed to produce event {}", err));

Note on gauge macro: The metrics 0.21 crate uses metrics::gauge!(name, value, labels) with a signed f64 for both increment (+1.0) and decrement (−1.0). There are no separate increment_gauge! / decrement_gauge! macros in this version.

Note on histogram failure path: The latency histogram is intentionally recorded on the success path only. A failed delivery does not represent a valid round-trip latency (it may have timed out or been rejected immediately). The failure counter is the appropriate signal for failed sends.

3. No changes to `send_batch`

The fan-out structure, JoinSet joining behavior, and Ok(()) return are left entirely intact.

4. No changes to `metrics_recorder.rs` in this task

Bucket registration for kafka_producer_produce_duration_seconds is task_1's responsibility.

Label value source

The topic label is sourced directly from the topic: String parameter already present in every kafka_send call (cloned from self.topic at the call site). No hard-coded strings are introduced.

Risks and Mitigations

Risk	Likelihood	Mitigation
Histogram uses default buckets if merged before task_1	Medium	Enforce merge order; PR description must explicitly reference the task_1 dependency
`topic` string `.clone()` overhead per metric call in hot path	Low	`topic` is a short runtime string; three extra shallow clones per message are negligible at expected event throughput
In-flight gauge remains elevated if a spawned task panics	Low	`JoinSet::join_next` returns a `JoinError` on panic; the gauge increment in the panicking task is lost. Acceptable for initial implementation; panics are already fatal in the current setup
`metrics::gauge!` signed-delta semantics misunderstood	Low (mitigated)	Implementation approach section explicitly documents the `+1.0` / `-1.0` signed-float pattern for `metrics` 0.21

Open Questions

Histogram on failure path: Should kafka_producer_produce_duration_seconds also be recorded for failed sends (to measure how long failed attempts took)? The current plan omits it to keep failure latency separate from success latency. This is a product decision — no implementation change is needed unless the answer is yes.
task_1 bucket scope confirmation: Does task_1 add the set_buckets_for_metric call specifically for kafka_producer_produce_duration_seconds? If not, a small additive change to setup_metrics_recorder() will be needed in this task or a follow-up.

Out of Scope

kafka_producer_batch_size histogram — that is task_3/task_4 per the epic PRD (FR-4).
Any changes to the consumer or worker binaries.
Changes to the HTTP metrics middleware or the metrics server port/endpoint.
Integration test harness changes (test coverage of the metrics endpoint is a separate concern).

Was this article helpful?

Outcome

Acceptance Criteria

Dependencies

Codebase Context

Key file: rust-capture/src/sinks/kafka_event_sink.rs

Existing metrics_recorder.rs

Crate versions (rust-capture/Cargo.toml)