From bb6e1591d5db933ea98b39b086af83523a01c170 Mon Sep 17 00:00:00 2001 From: LFC <990479+MichaelScofield@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:40:45 +0800 Subject: [PATCH] refactor: expose OTLP write and batching helpers for downstream routers [Backport release/v1.3] (#9462) refactor: expose OTLP write and batching helpers for downstream routers (#9401) (cherry picked from commit 48a2ca63409dbdb7d46fef038af91baf61fbc008) Signed-off-by: shuiyisong Co-authored-by: shuiyisong <113876041+shuiyisong@users.noreply.github.com> --- src/frontend/AGENTS.md | 4 +- src/frontend/src/instance/logical_batcher.rs | 34 +++--- src/frontend/src/instance/otlp.rs | 119 +++++++++++-------- src/frontend/src/lib.rs | 2 + src/frontend/src/server.rs | 8 +- src/servers/src/http.rs | 2 +- 6 files changed, 99 insertions(+), 70 deletions(-) diff --git a/src/frontend/AGENTS.md b/src/frontend/AGENTS.md index 49a4b8fcd0f..697ceb1ce2d 100644 --- a/src/frontend/AGENTS.md +++ b/src/frontend/AGENTS.md @@ -53,7 +53,9 @@ remote datanodes via `operator`/`client`. - **Logical-table batching** (`instance/logical_batcher.rs`): `Services` initializes one shared batcher for opted-in HTTP Prom and nonlegacy OTLP metric-engine - writes. OTLP checks operator eligibility and falls back for incompatible tables. + writes. Downstream routers can initialize it when enabling replacement endpoints. + `Instance::handle_otlp_metric_row_inserts` shares OTLP eligibility and dispatch + between converters and falls back for incompatible tables. The schema adapter holds a weak instance reference to avoid an ownership cycle. - **Table batching** (`instance/builder.rs`): protocol entry points opt in through diff --git a/src/frontend/src/instance/logical_batcher.rs b/src/frontend/src/instance/logical_batcher.rs index e68dcf644d4..255766ac14d 100644 --- a/src/frontend/src/instance/logical_batcher.rs +++ b/src/frontend/src/instance/logical_batcher.rs @@ -26,23 +26,25 @@ use crate::frontend::FrontendOptions; use crate::instance::Instance; impl Instance { - pub(crate) fn init_logical_batcher(self: &Arc, options: &FrontendOptions) { + /// Initializes the shared logical batcher for enabled ingestion endpoints. + /// Disabled endpoints leave it uninitialized so downstream routers can enable it. + pub fn init_logical_batcher(self: &Arc, options: &FrontendOptions) { + let options_batcher = options.logical_batcher_options(); + let enabled = options_batcher + .protocols + .iter() + .any(|protocol| match protocol { + BatchingProtocol::Prom => options.prom_store.enable, + BatchingProtocol::Otlp => options.otlp.enable, + _ => false, + }); + if !options.prom_store.with_metric_engine + || !enabled + || !options_batcher.pending_rows_batching_enabled() + { + return; + } self.logical_batcher.get_or_init(|| { - let options_batcher = options.logical_batcher_options(); - let enabled = options_batcher - .protocols - .iter() - .any(|protocol| match protocol { - BatchingProtocol::Prom => options.prom_store.enable, - BatchingProtocol::Otlp => options.otlp.enable, - _ => false, - }); - if !options.prom_store.with_metric_engine - || !enabled - || !options_batcher.pending_rows_batching_enabled() - { - return None; - } LogicalTablePendingRowsBatcher::try_new( self.partition_manager().clone(), self.node_manager().clone(), diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 5132b9f8a8d..5d96138319a 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/src/frontend/src/instance/otlp.rs @@ -18,6 +18,7 @@ pub mod trace_types; use std::sync::Arc; +use api::v1::RowInsertRequests; use async_trait::async_trait; use auth::{ OTLP_WRITE, PermissionChecker, PermissionCheckerRef, PermissionReq, PermissionTableTarget, @@ -43,6 +44,7 @@ use servers::query_handler::{ MetricsIngestOutcome, OpenTelemetryProtocolHandler, PipelineHandlerRef, TraceIngestOutcome, }; use session::context::QueryContextRef; +use session::protocol_ctx::OtlpMetricCtx; use snafu::ResultExt; use table::requests::{ OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM, SEMANTIC_PER_TABLE_INDEX_KEY, @@ -166,57 +168,9 @@ impl OpenTelemetryProtocolHandler for Instance { Arc::new(c) }; - let physical_table = ctx - .extension(PHYSICAL_TABLE_PARAM) - .unwrap_or(GREPTIME_PHYSICAL_TABLE) - .to_string(); - // The bulk path converts each request's time index unit to the - // destination table's unit during batch alignment, so no pre-gate - // alignment is needed here. - let batcher = self.logical_batcher().filter(|_| { - ctx.logical_batching_enabled() && !metric_ctx.is_legacy && metric_ctx.with_metric_engine - }); - let batcher = if batcher.is_some() - && self - .inserter - .can_batch_metric_rows(&requests, &ctx, &physical_table) - .await - .map_err(BoxedError::new) - .context(error::ExecuteGrpcQuerySnafu)? - { - batcher - } else { - None - }; - - // OTLP tables have one sample field in both the legacy and physical paths. - let output = if let Some(batcher) = batcher { - let (rows, cost) = batcher - .submit_with(requests, ctx.clone(), |mut requests| { - let ctx = ctx.clone(); - async move { - Inserter::meter_row_inserts(&mut requests, &ctx) - .await - .map_err(BoxedError::new) - .context(error::ExecuteGrpcQuerySnafu) - } - }) - .await?; - Output::new( - OutputData::AffectedRows(rows as usize), - OutputMeta::new_with_cost(cost as _), - ) - } else if metric_ctx.is_legacy || !metric_ctx.with_metric_engine { - self.handle_row_inserts(requests, ctx.clone(), false, true) - .await - .map_err(BoxedError::new) - .context(error::ExecuteGrpcQuerySnafu)? - } else { - self.handle_metric_row_inserts(requests, ctx.clone(), physical_table) - .await - .map_err(BoxedError::new) - .context(error::ExecuteGrpcQuerySnafu)? - }; + let output = self + .handle_otlp_metric_row_inserts(requests, ctx.clone(), &metric_ctx) + .await?; outcome.write_cost = output.meta.cost; // Derived enrichment follows the accepted metric submission, which may @@ -362,6 +316,69 @@ impl OpenTelemetryProtocolHandler for Instance { } } +impl Instance { + /// Inserts converted OTLP metrics through the eligible batching or ordinary write path. + /// The caller must check table permissions and admit the complete request first. + pub async fn handle_otlp_metric_row_inserts( + &self, + requests: RowInsertRequests, + ctx: QueryContextRef, + metric_ctx: &OtlpMetricCtx, + ) -> ServerResult { + let physical_table = ctx + .extension(PHYSICAL_TABLE_PARAM) + .unwrap_or(GREPTIME_PHYSICAL_TABLE) + .to_string(); + // The bulk path converts each request's time index unit to the + // destination table's unit during batch alignment, so no pre-gate + // alignment is needed here. + let batcher = self.logical_batcher().filter(|_| { + ctx.logical_batching_enabled() && !metric_ctx.is_legacy && metric_ctx.with_metric_engine + }); + let batcher = if batcher.is_some() + && self + .inserter + .can_batch_metric_rows(&requests, &ctx, &physical_table) + .await + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu)? + { + batcher + } else { + None + }; + + // OTLP tables have one sample field in both the legacy and physical paths. + if let Some(batcher) = batcher { + let (rows, cost) = batcher + .submit_with(requests, ctx.clone(), |mut requests| { + let ctx = ctx.clone(); + async move { + Inserter::meter_row_inserts(&mut requests, &ctx) + .await + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu) + } + }) + .await?; + Ok(Output::new( + OutputData::AffectedRows(rows as usize), + OutputMeta::new_with_cost(cost as _), + )) + } else if metric_ctx.is_legacy || !metric_ctx.with_metric_engine { + self.handle_row_inserts(requests, ctx.clone(), false, true) + .await + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu) + } else { + self.handle_metric_row_inserts(requests, ctx.clone(), physical_table) + .await + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu) + } + } +} + #[cfg(test)] mod tests { use std::sync::Arc; diff --git a/src/frontend/src/lib.rs b/src/frontend/src/lib.rs index 1468fe7c0b2..7c68e445242 100644 --- a/src/frontend/src/lib.rs +++ b/src/frontend/src/lib.rs @@ -23,3 +23,5 @@ pub(crate) mod metrics; pub mod server; pub mod service_config; mod stream_wrapper; + +pub use metrics::{OTLP_METRICS_ROWS, OTLP_RESOURCE_INFO_WRITE_ERRORS}; diff --git a/src/frontend/src/server.rs b/src/frontend/src/server.rs index a5568b24be5..cebb27d13d0 100644 --- a/src/frontend/src/server.rs +++ b/src/frontend/src/server.rs @@ -451,7 +451,8 @@ fn effective_prom_store_options(opts: &FrontendOptions) -> PromStoreOptions { prom_store } -fn effective_http_options(opts: &FrontendOptions) -> HttpOptions { +/// Resolves HTTP options, allowing synchronous batches enough time to flush. +pub fn effective_http_options(opts: &FrontendOptions) -> HttpOptions { effective_http_options_with_sync(opts, pending_rows_batch_sync_enabled()) } @@ -550,6 +551,11 @@ mod tests { .await .unwrap(), ); + let mut disabled_options = options.clone(); + disabled_options.otlp.enable = false; + instance.init_logical_batcher(&disabled_options); + assert!(instance.logical_batcher().is_none()); + let services = Services::new(options.clone(), instance.clone(), Plugins::default()); let batcher = instance.logical_batcher().unwrap().clone(); instance.init_logical_batcher(&options); diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index b9fcd0a9420..fd60ee2e040 100644 --- a/src/servers/src/http.rs +++ b/src/servers/src/http.rs @@ -225,7 +225,7 @@ pub fn is_api_listener_path(path: &str) -> bool { struct LogicalBatchingProtocols(Vec); /// Sets a local-only write selector after authentication creates the context. -async fn set_http_write_batching( +pub async fn set_http_write_batching( State(protocol): State, mut req: Request, next: Next,