Skip to content

Commit 1bca29c

Browse files
feat(query-engine): log precompute ingest throughput (#716)
* feat(query-engine): log precompute ingest throughput samples_ingested was an AtomicU64 incremented on every HTTP ingest request but never read. Spawn a self-contained 30s ticker inside HttpIngestSource::run that swaps it to 0 each tick and logs the resulting samples/sec at debug level, so ingest throughput shows up in query_engine.log for any run with debug logging enabled. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PR8ENqodwuB1o8k4YAcjcJ * fix(query-engine): stop ingest throughput ticker leaking on source exit roborev (job 196) on fef024e: the diagnostics ticker was spawned before TcpListener::bind, so a bind failure left it running forever, holding the HttpIngestState alive and logging 0 samples/sec indefinitely. Bind first, then spawn the ticker only on success, and abort it once axum::serve(...) resolves (success or error) so it never outlives the ingest server. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PR8ENqodwuB1o8k4YAcjcJ * fix(query-engine): abort ingest ticker on cancellation, not just on exit roborev (job 199) on e937288: ticker.abort() only ran on axum::serve returning, so if run()'s own future were ever dropped/cancelled mid-await by a future caller, the abort call would never execute — a bare JoinHandle detaches rather than cancels its task on drop. Wrap the ticker's JoinHandle in a small AbortOnDrop guard so it's aborted on every exit path, including cancellation, and add a regression test that proves drop actually cancels the wrapped task. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PR8ENqodwuB1o8k4YAcjcJ * fix(query-engine): make abort-on-drop test actually falsifiable roborev (job 200) on d90a0a7: the regression test checked a flag set after a 60s sleep, but only waited 50ms before asserting — so it passed even with AbortOnDrop::drop() doing nothing, since 50ms never approaches 60s either way. Confirmed by temporarily neutering the abort call: the old test still passed. Rewrote it to check ongoing progress instead: the wrapped task increments a counter on a 5ms cadence, the test waits for it to start, drops the guard, then asserts the counter stops advancing. Verified this version fails without the real abort call and passes with it. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PR8ENqodwuB1o8k4YAcjcJ * fix(query-engine): measure real elapsed time for ingest throughput rate The deferred Low finding from roborev job 196: tokio::time::interval defaults to MissedTickBehavior::Burst, and the rate always divided by the fixed INGEST_DIAG_INTERVAL constant rather than actual elapsed time. Under a runtime stall this would replay missed ticks back to back, each still claiming the fixed 30s window and misreporting the rate. Track Instant::now() across ticks and divide by measured elapsed time instead, and set MissedTickBehavior::Delay so a stall produces one accurately-labeled longer interval rather than a burst of catch-up ticks. Verified via a shortened-interval smoke test against test_e2e_precompute: elapsed time and rate match the raw sample count under normal operation. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PR8ENqodwuB1o8k4YAcjcJ --------- Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
1 parent 3f4f83c commit 1bca29c

1 file changed

Lines changed: 106 additions & 6 deletions

File tree

‎asap-query-engine/src/precompute_engine/ingest_handler.rs‎

Lines changed: 106 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,11 @@ use crate::precompute_engine::ingest_source::{route_decoded_samples, IngestConte
44
use axum::{body::Bytes, extract::State, http::StatusCode, routing::post, Router};
55
use std::sync::atomic::{AtomicU64, Ordering};
66
use std::sync::Arc;
7-
use std::time::Instant;
7+
use std::time::{Duration, Instant};
88
use tokio::net::TcpListener;
9-
use tracing::{info, warn};
9+
use tracing::{debug, info, warn};
10+
11+
const INGEST_DIAG_INTERVAL: Duration = Duration::from_secs(30);
1012

1113
pub struct HttpIngestConfig {
1214
pub port: u16,
@@ -33,15 +35,23 @@ impl IngestSource for HttpIngestSource {
3335
samples_ingested: AtomicU64::new(0),
3436
});
3537

38+
let addr = format!("0.0.0.0:{}", self.config.port);
39+
// Bind before spawning the ticker so a bind failure returns early
40+
// without leaving an orphaned diagnostics task behind.
41+
let listener = TcpListener::bind(&addr).await?;
42+
info!("HTTP ingest server listening on {}", addr);
43+
44+
// Held for its Drop impl: aborts the ticker on every exit path,
45+
// including the run() future itself being dropped/cancelled by a
46+
// future caller mid-await (a plain JoinHandle would not — dropping
47+
// it just detaches the task, leaving it running).
48+
let _ticker_guard = AbortOnDrop(tokio::spawn(log_ingest_throughput(state.clone())));
49+
3650
let app = Router::new()
3751
.route("/api/v1/write", post(handle_prometheus_ingest))
3852
.route("/api/v1/import", post(handle_victoriametrics_ingest))
3953
.with_state(state);
4054

41-
let addr = format!("0.0.0.0:{}", self.config.port);
42-
info!("HTTP ingest server listening on {}", addr);
43-
44-
let listener = TcpListener::bind(&addr).await?;
4555
axum::serve(listener, app).await?;
4656
Ok(())
4757
}
@@ -53,6 +63,44 @@ struct HttpIngestState {
5363
samples_ingested: AtomicU64,
5464
}
5565

66+
/// Aborts the wrapped task when dropped, so it can't outlive its owner.
67+
struct AbortOnDrop(tokio::task::JoinHandle<()>);
68+
69+
impl Drop for AbortOnDrop {
70+
fn drop(&mut self) {
71+
self.0.abort();
72+
}
73+
}
74+
75+
/// Logs ingest throughput every `INGEST_DIAG_INTERVAL`, resetting the counter
76+
/// each tick so the log reports a per-interval rate rather than a lifetime total.
77+
///
78+
/// Measures actual elapsed time (rather than assuming exactly
79+
/// `INGEST_DIAG_INTERVAL` passed) and uses `MissedTickBehavior::Delay` so a
80+
/// runtime stall produces one accurately-labeled longer interval instead of a
81+
/// burst of back-to-back catch-up ticks each misreporting against the fixed
82+
/// interval.
83+
async fn log_ingest_throughput(state: Arc<HttpIngestState>) {
84+
let mut interval = tokio::time::interval(INGEST_DIAG_INTERVAL);
85+
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
86+
interval.tick().await; // first tick fires immediately; skip it
87+
let mut last = Instant::now();
88+
loop {
89+
interval.tick().await;
90+
let now = Instant::now();
91+
let elapsed = now.duration_since(last);
92+
last = now;
93+
let samples = state.samples_ingested.swap(0, Ordering::Relaxed);
94+
let secs = elapsed.as_secs_f64();
95+
debug!(
96+
"[INGEST_DIAG] samples_ingested: {} in {:.1}s ({:.1} samples/sec)",
97+
samples,
98+
secs,
99+
samples as f64 / secs,
100+
);
101+
}
102+
}
103+
56104
async fn handle_prometheus_ingest(
57105
State(state): State<Arc<HttpIngestState>>,
58106
body: Bytes,
@@ -100,3 +148,55 @@ async fn handle_victoriametrics_ingest(
100148
}
101149
}
102150
}
151+
152+
#[cfg(test)]
153+
mod tests {
154+
use super::AbortOnDrop;
155+
use std::sync::atomic::{AtomicU64, Ordering};
156+
use std::sync::Arc;
157+
use std::time::Duration;
158+
159+
/// Regression test for roborev job 199: a plain `JoinHandle` does not
160+
/// cancel its task on drop, so the diagnostics ticker would keep running
161+
/// (and keep the ingest state alive) if `HttpIngestSource::run` were ever
162+
/// dropped/cancelled mid-flight by a future caller. `AbortOnDrop` must
163+
/// actually abort the task, not just detach it.
164+
///
165+
/// Checks ongoing progress rather than a single far-future flag (roborev
166+
/// job 200): the wrapped task increments a counter on a fast, observable
167+
/// cadence, so a merely-detached task (bug) keeps advancing the counter
168+
/// after drop, while a genuinely aborted task (fix) does not. An earlier
169+
/// version of this test only checked a flag set after a 60s sleep — that
170+
/// passed even with `AbortOnDrop::drop` doing nothing, since the check
171+
/// window (50ms) never reached the 60s mark either way.
172+
#[tokio::test]
173+
async fn abort_on_drop_cancels_the_wrapped_task() {
174+
let counter = Arc::new(AtomicU64::new(0));
175+
let task_counter = counter.clone();
176+
let guard = AbortOnDrop(tokio::spawn(async move {
177+
loop {
178+
task_counter.fetch_add(1, Ordering::SeqCst);
179+
tokio::time::sleep(Duration::from_millis(5)).await;
180+
}
181+
}));
182+
183+
// Synchronize with task startup before measuring.
184+
while counter.load(Ordering::SeqCst) == 0 {
185+
tokio::task::yield_now().await;
186+
}
187+
188+
drop(guard);
189+
let count_at_drop = counter.load(Ordering::SeqCst);
190+
191+
// A merely-detached task has ample time here to advance the counter
192+
// several more times (100ms / 5ms per tick); an aborted task cannot
193+
// make any further progress at all.
194+
tokio::time::sleep(Duration::from_millis(100)).await;
195+
196+
assert_eq!(
197+
counter.load(Ordering::SeqCst),
198+
count_at_drop,
199+
"task kept running after the guard was dropped — it was merely detached, not aborted"
200+
);
201+
}
202+
}

0 commit comments

Comments
 (0)