diff --git a/src/datanode/src/metrics.rs b/src/datanode/src/metrics.rs index e9f947ef9da..af9a82ed551 100644 --- a/src/datanode/src/metrics.rs +++ b/src/datanode/src/metrics.rs @@ -85,6 +85,15 @@ lazy_static! { ) .unwrap(); + /// Ingest-task join failures, labeled by request type and reason (`panic` or `cancelled`). + /// Counts tasks, not write entries: a failed task can leave entry outcomes unknown. + pub static ref INGEST_TASK_FAILURES_TOTAL: IntCounterVec = register_int_counter_vec!( + "greptime_datanode_ingest_task_failures_total", + "Ingest-task join failures, not failed write entries", + &[REGION_REQUEST_TYPE, "reason"] + ) + .unwrap(); + /// Failed region open attempts, including individual and batch requests. /// Counts every request in an engine group when no individual results are available. /// Failed retries count separately; explicitly ignored missing regions do not count. diff --git a/src/datanode/src/region_server.rs b/src/datanode/src/region_server.rs index 135d069fd77..3acf188ad19 100644 --- a/src/datanode/src/region_server.rs +++ b/src/datanode/src/region_server.rs @@ -49,7 +49,7 @@ use common_query::OutputData; use common_query::request::QueryRequest; use common_recordbatch::adapter::RecordBatchMetrics; use common_recordbatch::{OrderOption, RecordBatch, RecordBatchStream, SendableRecordBatchStream}; -use common_runtime::Runtime; +use common_runtime::{JoinHandle, Runtime}; use common_telemetry::tracing::{self, info_span}; use common_telemetry::tracing_context::{FutureExt, TracingContext}; use common_telemetry::{debug, error, info, warn}; @@ -250,11 +250,10 @@ impl RegionServer { if RegionServerInner::is_ingest_request(&request) { let inner = self.inner.clone(); let request_type = request.request_type(); - return common_runtime::spawn_ingest(async move { + let task = common_runtime::spawn_ingest(async move { inner.handle_request(region_id, request).await - }) - .await - .context(RuntimeJoinSnafu { request_type })?; + }); + return join_ingest_task(task, request_type).await?; } self.inner.handle_request(region_id, request).await @@ -679,14 +678,13 @@ impl RegionServer { "RegionServer::handle_metric_batch_puts", batch_size = batch_size )); - let results = common_runtime::spawn_ingest(async move { + let task = common_runtime::spawn_ingest(async move { metric_engine .put_regions_batch(put_requests) .trace(span) .await - }) - .await - .context(RuntimeJoinSnafu { request_type })?; + }); + let results = join_ingest_task(task, request_type).await?; let mut response = BatchRegionResponse { affected_rows: 0, extensions: HashMap::new(), @@ -746,7 +744,7 @@ impl RegionServer { .collect(); let started = Instant::now(); let tracing_context = TracingContext::from_current_span(); - common_runtime::spawn_ingest(async move { + let task = common_runtime::spawn_ingest(async move { let mut pending = engine .handle_batch_put_requests(puts) .into_iter() @@ -800,10 +798,8 @@ impl RegionServer { } response.failures.sort_unstable_by_key(|(index, _)| *index); response - }) - .await - .context(RuntimeJoinSnafu { request_type }) - .map(Either::Left) + }); + join_ingest_task(task, request_type).await.map(Either::Left) } /// Returns the mito engine when there are at least two requests, every one a put into a ready @@ -2292,6 +2288,22 @@ fn record_region_open_failures(engine: &str, status_code: StatusCode, count: usi .inc_by(count as u64); } +/// Awaits an ingest task without treating a join failure as an entry-level write result. +async fn join_ingest_task(task: JoinHandle, request_type: &'static str) -> Result { + task.await + .inspect_err(|error| { + let reason = if error.is_panic() { + "panic" + } else { + "cancelled" + }; + crate::metrics::INGEST_TASK_FAILURES_TOTAL + .with_label_values(&[request_type, reason]) + .inc(); + }) + .context(RuntimeJoinSnafu { request_type }) +} + /// Records the result of `count` ingest requests answered by one response: the rows they /// changed, or their failures. fn record_ingest_result(request_type: &str, count: usize, result: &Result) {