refactor: expose OTLP write and batching helpers for downstream routers (#9401)

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

src/frontend/AGENTS.md

@@ -57,7 +57,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

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<Self>, 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<Self>, 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(),

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<Output> {
+        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;

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};

src/frontend/src/server.rs

@@ -455,7 +455,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())
 }
 
@@ -555,6 +556,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);

src/servers/src/http.rs

@@ -225,7 +225,7 @@ pub fn is_api_listener_path(path: &str) -> bool {
 struct LogicalBatchingProtocols(Vec<BatchingProtocol>);
 
 /// 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<BatchingProtocol>,
     mut req: Request,
     next: Next,