feat: expose standalone metric ingestion hooks

Expose embedded object storage and converted OTLP metric ingestion hooks with phase instrumentation.

Files: `src/datanode/src/datanode.rs`, `src/frontend/src/instance.rs`, `src/frontend/src/instance/otlp.rs`, `src/frontend/AGENTS.md`, `src/servers/src/http/otlp.rs`, `src/servers/src/metrics.rs`.
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

src/datanode/src/datanode.rs

@@ -82,6 +82,7 @@ pub struct Datanode {
     heartbeat_task: Option<HeartbeatTask>,
     region_event_receiver: Option<RegionServerEventReceiver>,
     region_server: RegionServer,
+    object_store_manager: ObjectStoreManagerRef,
     greptimedb_telemetry_task: Arc<GreptimeDBTelemetryTask>,
     leases_notifier: Option<Arc<Notify>>,
     plugins: Plugins,
@@ -151,6 +152,10 @@ impl Datanode {
         self.region_server.clone()
     }
 
+    pub fn object_store_manager(&self) -> &ObjectStoreManagerRef {
+        &self.object_store_manager
+    }
+
     pub fn plugins(&self) -> Plugins {
         self.plugins.clone()
     }
@@ -293,8 +298,10 @@ impl DatanodeBuilder {
             Some(node_id),
             gc_enabled,
         ));
+        let object_store_manager = Self::build_object_store_manager(&self.opts.storage).await?;
         let region_server = self
             .new_region_server(
+                object_store_manager.clone(),
                 schema_metadata_manager,
                 region_event_listener,
                 file_ref_manager,
@@ -359,6 +366,7 @@ impl DatanodeBuilder {
             services: ServerHandlers::default(),
             heartbeat_task,
             region_server,
+            object_store_manager,
             greptimedb_telemetry_task,
             region_event_receiver,
             leases_notifier,
@@ -434,6 +442,7 @@ impl DatanodeBuilder {
 
     async fn new_region_server(
         &mut self,
+        object_store_manager: ObjectStoreManagerRef,
         schema_metadata_manager: SchemaMetadataManagerRef,
         event_listener: RegionServerEventListenerRef,
         file_ref_manager: FileReferenceManagerRef,
@@ -470,7 +479,6 @@ impl DatanodeBuilder {
         );
         region_server.install_remote_dyn_filter_receiver_injector(&self.plugins);
 
-        let object_store_manager = Self::build_object_store_manager(&opts.storage).await?;
         let engines = self
             .build_store_engines(
                 object_store_manager,

src/frontend/AGENTS.md

@@ -47,6 +47,7 @@ remote datanodes via `operator`/`client`.
 ## Public surface
 
 - `Instance` (`instance.rs`) — the business-logic container.
+- `OtlpMetricIngestor` (`instance/otlp.rs`, re-exported by `instance`) — optional converted-metric storage hook, after OTLP permission checks and translation.
 - `Frontend` (`frontend.rs`) — lifecycle wrapper around `Instance` + servers + heartbeat.
 - Created from `cmd`: `src/cmd/src/frontend.rs` (distributed) and
   `src/cmd/src/standalone.rs` (standalone, with embedded datanode).

src/frontend/src/instance.rs

@@ -22,6 +22,7 @@ mod log_handler;
 mod logs;
 mod opentsdb;
 mod otlp;
+pub use otlp::{OtlpMetricIngestor, OtlpMetricIngestorRef};
 pub mod prom_store;
 mod promql;
 mod region_query;

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,
@@ -52,6 +53,20 @@ use self::trace_ingest::trace_conventions;
 use crate::instance::Instance;
 use crate::metrics::{OTLP_LOGS_ROWS, OTLP_METRICS_ROWS, OTLP_RESOURCE_INFO_WRITE_ERRORS};
 
+/// Optional storage path for converted OTLP metrics, after protocol and table permissions.
+#[async_trait]
+pub trait OtlpMetricIngestor: Send + Sync {
+    async fn check_write(&self) -> ServerResult<()>;
+
+    async fn ingest(
+        &self,
+        requests: RowInsertRequests,
+        ctx: QueryContextRef,
+    ) -> ServerResult<Output>;
+}
+
+pub type OtlpMetricIngestorRef = Arc<dyn OtlpMetricIngestor>;
+
 fn trace_permission_targets(
     table_name: &str,
     groups: &[TraceSpanGroup],
@@ -101,6 +116,10 @@ impl OpenTelemetryProtocolHandler for Instance {
             .plugins
             .get::<OpenTelemetryProtocolInterceptorRef<servers::error::Error>>();
         interceptor_ref.pre_execute(ctx.clone())?;
+        let metric_ingestor = self.plugins.get::<OtlpMetricIngestorRef>();
+        if let Some(ingestor) = &metric_ingestor {
+            ingestor.check_write().await?;
+        }
         let ctx = Arc::new(ctx.fork());
 
         let input_names = request
@@ -121,6 +140,9 @@ impl OpenTelemetryProtocolHandler for Instance {
         metric_ctx.is_legacy = is_legacy;
         metric_ctx.resource_info = self.otlp_resource_info;
 
+        let conversion_timer = servers::metrics::METRIC_OTLP_METRICS_STAGE_ELAPSED
+            .with_label_values(&["convert"])
+            .start_timer();
         let otlp::metrics::MetricsConversion {
             requests,
             rows,
@@ -128,6 +150,7 @@ impl OpenTelemetryProtocolHandler for Instance {
             resource_info,
             mut outcome,
         } = otlp::metrics::to_grpc_insert_requests(request, &mut metric_ctx)?;
+        drop(conversion_timer);
         if outcome.rejected_data_points > 0 {
             warn!(
                 "Rejected {} OTLP metrics data points: {}",
@@ -161,7 +184,9 @@ impl OpenTelemetryProtocolHandler for Instance {
         };
 
         // OTLP tables have one sample field in both the legacy and physical paths.
-        let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
+        let output = if let Some(ingestor) = metric_ingestor {
+            ingestor.ingest(requests, ctx.clone()).await
+        } 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)

src/servers/src/http/otlp.rs

@@ -100,11 +100,15 @@ pub async fn metrics(
     let _timer = crate::metrics::METRIC_HTTP_OPENTELEMETRY_METRICS_ELAPSED
         .with_label_values(&[db.as_str()])
         .start_timer();
+    let decode_timer = crate::metrics::METRIC_OTLP_METRICS_STAGE_ELAPSED
+        .with_label_values(&["http_decode"])
+        .start_timer();
     let request = ExportMetricsServiceRequest::decode(bytes).with_context(|_| {
         error::DecodeOtlpRequestSnafu {
             content_type: content_type_to_string(content_type.as_ref()),
         }
     })?;
+    drop(decode_timer);
 
     let OtlpState {
         with_metric_engine,

src/servers/src/metrics.rs

@@ -51,6 +51,13 @@ pub(crate) const METRIC_FAILURE_VALUE: &str = "failure";
 
 lazy_static! {
 
+    pub static ref METRIC_OTLP_METRICS_STAGE_ELAPSED: HistogramVec = register_histogram_vec!(
+        "greptime_servers_otlp_metrics_stage_seconds",
+        "OTLP metrics decoding and conversion duration in seconds",
+        &["stage"],
+        prometheus::exponential_buckets(0.00001, 4.0, 12).unwrap()
+    ).unwrap();
+
     pub static ref HTTP_REQUEST_COUNTER: IntCounterVec = register_int_counter_vec!(
         "greptime_servers_http_request_counter",
         "servers http request counter",