feat(otlp): synthesize otel_resource_info at OTLP metrics ingestion

Ordinary OTLP metrics scatter filtered resource attributes as tags over
every logical metric table, so metrics-only services contribute nothing
to the semantic entity graph. Each request now also projects its
distinct resources into one info-metric-shaped mito table,
otel_resource_info: a fixed allowlist of identity-relevant attributes
under their raw OTel keys (independent of the label translation
strategy and the promote/ignore headers) plus derived job/instance
compatibility columns, value 1.0, and the newest data-point timestamp.

The descriptor is written after the main insert is committed; a failure
there (conflicting pre-existing table, auto-create disabled) degrades
to an OTLP partial_success warning with rejected_data_points = 0
instead of failing the request and triggering client retries of
already-accepted data. A request writing a metric named
otel_resource_info suppresses synthesis. Legacy mode is unchanged.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

src/frontend/src/instance/otlp.rs

@@ -38,7 +38,7 @@ use servers::interceptor::{OpenTelemetryProtocolInterceptor, OpenTelemetryProtoc
 use servers::otlp;
 use servers::otlp::trace::span::TraceSpanGroup;
 use servers::query_handler::{
-    OpenTelemetryProtocolHandler, PipelineHandlerRef, TraceIngestOutcome,
+    OpenTelemetryProtocolHandler, OtlpMetricsOutcome, PipelineHandlerRef, TraceIngestOutcome,
 };
 use session::context::QueryContextRef;
 use snafu::ResultExt;
@@ -50,7 +50,7 @@ use table::requests::{
 
 use self::trace_ingest::trace_conventions;
 use crate::instance::Instance;
-use crate::metrics::{OTLP_LOGS_ROWS, OTLP_METRICS_ROWS};
+use crate::metrics::{OTLP_LOGS_ROWS, OTLP_METRICS_ROWS, OTLP_RESOURCE_INFO_WRITE_ERRORS};
 
 fn trace_permission_targets(
     table_name: &str,
@@ -90,7 +90,7 @@ impl OpenTelemetryProtocolHandler for Instance {
         &self,
         request: ExportMetricsServiceRequest,
         ctx: QueryContextRef,
-    ) -> ServerResult<Output> {
+    ) -> ServerResult<OtlpMetricsOutcome> {
         self.plugins
             .get::<PermissionCheckerRef>()
             .as_ref()
@@ -120,10 +120,18 @@ impl OpenTelemetryProtocolHandler for Instance {
             .unwrap_or_default();
         metric_ctx.is_legacy = is_legacy;
 
-        let (requests, rows, semantic_index) =
-            otlp::metrics::to_grpc_insert_requests(request, &mut metric_ctx)?;
+        let otlp::metrics::MetricsConversion {
+            requests,
+            rows,
+            semantic_index,
+            resource_info,
+        } = otlp::metrics::to_grpc_insert_requests(request, &mut metric_ctx)?;
         self.check_row_insert_permission(&requests, &ctx, PermissionReq::Action(OTLP_WRITE))
             .context(AuthSnafu)?;
+        if let Some(resource_info) = &resource_info {
+            self.check_row_insert_permission(resource_info, &ctx, PermissionReq::Action(OTLP_WRITE))
+                .context(AuthSnafu)?;
+        }
         self.cache_otlp_legacy(&input_names, &ctx, is_legacy)?;
         OTLP_METRICS_ROWS.inc_by(rows as u64);
 
@@ -144,21 +152,41 @@ impl OpenTelemetryProtocolHandler for Instance {
         };
 
         // If the user uses the legacy path, it is by default without metric engine.
-        if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
-            self.handle_row_inserts(requests, ctx, false, false)
+        let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
+            self.handle_row_inserts(requests, ctx.clone(), false, false)
                 .await
                 .map_err(BoxedError::new)
-                .context(error::ExecuteGrpcQuerySnafu)
+                .context(error::ExecuteGrpcQuerySnafu)?
         } else {
             let physical_table = ctx
                 .extension(PHYSICAL_TABLE_PARAM)
                 .unwrap_or(GREPTIME_PHYSICAL_TABLE)
                 .to_string();
-            self.handle_metric_row_inserts(requests, ctx, physical_table.clone())
+            self.handle_metric_row_inserts(requests, ctx.clone(), physical_table.clone())
                 .await
                 .map_err(BoxedError::new)
-                .context(error::ExecuteGrpcQuerySnafu)
+                .context(error::ExecuteGrpcQuerySnafu)?
+        };
+
+        // The descriptor is derived enrichment written after the main data is
+        // committed: a failure here (e.g. a conflicting pre-existing table, or
+        // auto-create disabled) must not fail the request and trigger client
+        // retries of already-accepted data; it degrades to a partial-success
+        // warning.
+        let mut warning = None;
+        if let Some(resource_info) = resource_info
+            && let Err(e) = self.handle_row_inserts(resource_info, ctx, false, false).await
+        {
+            OTLP_RESOURCE_INFO_WRITE_ERRORS.inc();
+            common_telemetry::warn!(e; "Failed to write the OTLP resource descriptor table");
+            warning = Some(format!(
+                "metric data was accepted, but writing the resource \
+                 descriptor table `{}` failed: {e}",
+                otlp::metrics::OTEL_RESOURCE_INFO_TABLE_NAME
+            ));
         }
+
+        Ok(OtlpMetricsOutcome { output, warning })
     }
 
     #[tracing::instrument(skip_all)]

src/frontend/src/metrics.rs

@@ -45,6 +45,14 @@ lazy_static! {
     )
     .unwrap();
 
+    /// Failed writes of the synthesized OTLP resource descriptor table; these
+    /// surface as OTLP partial-success warnings, not request failures.
+    pub static ref OTLP_RESOURCE_INFO_WRITE_ERRORS: IntCounter = register_int_counter!(
+        "greptime_frontend_otlp_resource_info_write_errors",
+        "frontend otlp resource descriptor write errors"
+    )
+    .unwrap();
+
     /// The number of OpenTelemetry traces send by frontend node.
     pub static ref OTLP_TRACES_ROWS: IntCounter = register_int_counter!(
         "greptime_frontend_otlp_traces_rows",

src/servers/src/http/otlp.rs

@@ -27,7 +27,9 @@ use mime_guess::mime;
 use opentelemetry_proto::tonic::collector::logs::v1::{
     ExportLogsServiceRequest, ExportLogsServiceResponse,
 };
-use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceResponse;
+use opentelemetry_proto::tonic::collector::metrics::v1::{
+    ExportMetricsPartialSuccess, ExportMetricsServiceResponse,
+};
 use opentelemetry_proto::tonic::collector::trace::v1::{
     ExportTracePartialSuccess, ExportTraceServiceRequest, ExportTraceServiceResponse,
 };
@@ -123,9 +125,14 @@ pub async fn metrics(
         .await
         .map(|o| OtlpResponse {
             resp_body: ExportMetricsServiceResponse {
-                partial_success: None,
+                // rejected_data_points = 0: all metric data was accepted, the
+                // message only carries a derived-write warning.
+                partial_success: o.warning.map(|error_message| ExportMetricsPartialSuccess {
+                    rejected_data_points: 0,
+                    error_message,
+                }),
             },
-            write_cost: o.meta.cost,
+            write_cost: o.output.meta.cost,
         })
 }
 

src/servers/src/otel_arrow.rs

@@ -15,7 +15,7 @@
 use auth::UserProviderRef;
 use common_error::ext::ErrorExt;
 use common_error::status_code::status_to_tonic_code;
-use common_telemetry::error;
+use common_telemetry::{error, warn};
 use futures::SinkExt;
 use otel_arrow_rust::Consumer;
 use otel_arrow_rust::proto::opentelemetry::arrow::v1::arrow_metrics_service_server::ArrowMetricsService;
@@ -99,15 +99,24 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
                     }
                 };
                 // use metric engine by default
-                if let Err(e) = handler.metrics(request, query_ctx.clone()).await {
-                    let _ = sender
-                        .send(Err(Status::new(
-                            status_to_tonic_code(e.status_code()),
-                            e.to_string(),
-                        )))
-                        .await;
-                    error!(e; "Failed to ingest metrics from otel-arrow");
-                    return;
+                match handler.metrics(request, query_ctx.clone()).await {
+                    Ok(outcome) => {
+                        // BatchStatus has no partial-success channel; a
+                        // derived-write warning is only logged here.
+                        if let Some(warning) = outcome.warning {
+                            warn!("otel-arrow metrics ingestion warning: {warning}");
+                        }
+                    }
+                    Err(e) => {
+                        let _ = sender
+                            .send(Err(Status::new(
+                                status_to_tonic_code(e.status_code()),
+                                e.to_string(),
+                            )))
+                            .await;
+                        error!(e; "Failed to ingest metrics from otel-arrow");
+                        return;
+                    }
                 }
                 let _ = sender.send(Ok(batch_status)).await;
             }

src/servers/src/otlp/metrics.rs

@@ -16,6 +16,7 @@ use ahash::HashSet;
 use api::v1::{RowInsertRequests, Value};
 use common_grpc::precision::Precision;
 use common_query::prelude::{GREPTIME_COUNT, greptime_timestamp, greptime_value};
+use common_telemetry::warn;
 use lazy_static::lazy_static;
 use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
 use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, KeyValue, any_value};
@@ -35,8 +36,11 @@ use crate::semantic::{
     METRIC_TYPE_UPDOWN_COUNTER,
 };
 
+mod resource_info;
 mod translator;
 
+pub use resource_info::OTEL_RESOURCE_INFO_TABLE_NAME;
+use resource_info::ResourceInfoData;
 pub use translator::legacy_normalize_otlp_name;
 pub(crate) use translator::ucum_to_openmetrics_unit;
 use translator::{translate_label_name, translate_metric_name};
@@ -83,22 +87,41 @@ const OTEL_SCOPE_NAME: &str = "name";
 const OTEL_SCOPE_VERSION: &str = "version";
 const OTEL_SCOPE_SCHEMA_URL: &str = "schema_url";
 
+/// Result of converting one OTLP metrics request.
+pub struct MetricsConversion {
+    pub requests: RowInsertRequests,
+    /// Row count of `requests` (the resource descriptor is not counted).
+    pub rows: usize,
+    /// Per-table semantic index for the auto-create path to stamp as table
+    /// options; covers the descriptor table too.
+    pub semantic_index: SemanticIndex,
+    /// The synthesized resource descriptor insert. `None` in legacy mode,
+    /// when no resource projected any allowlisted attribute, or when the
+    /// request itself writes a metric named like the descriptor table.
+    pub resource_info: Option<RowInsertRequests>,
+}
+
 /// Convert OpenTelemetry metrics to GreptimeDB insert requests
 ///
 /// See
 /// <https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/metrics/v1/metrics.proto>
 /// for data structure of OTLP metrics.
-///
-/// Returns `InsertRequests`, total number of rows to ingest, and the per-table
-/// semantic index for the auto-create path to stamp as table options.
 pub fn to_grpc_insert_requests(
     request: ExportMetricsServiceRequest,
     metric_ctx: &mut OtlpMetricCtx,
-) -> Result<(RowInsertRequests, usize, SemanticIndex)> {
+) -> Result<MetricsConversion> {
     let mut table_writer = MultiTableData::default();
     let mut semantic_index = SemanticIndex::default();
+    let mut resource_info = ResourceInfoData::default();
 
     for resource in &request.resource_metrics {
+        if !metric_ctx.is_legacy
+            && let Some(r) = resource.resource.as_ref()
+            && let Some(max_ts) = resource_info::max_data_point_time_nanos(resource)
+        {
+            resource_info.observe(&r.attributes, max_ts);
+        }
+
         let resource_attrs = resource.resource.as_ref().map(|r| {
             let mut attrs = r.attributes.clone();
             process_resource_attrs(&mut attrs, metric_ctx);
@@ -129,7 +152,42 @@ pub fn to_grpc_insert_requests(
     }
 
     let (requests, rows) = table_writer.into_row_insert_requests();
-    Ok((requests, rows, semantic_index))
+
+    // A metric emitting a table named like the descriptor would fight over
+    // the table (metric engine vs plain mito); the metric wins, synthesis is
+    // suppressed for the whole request.
+    let resource_info = if requests
+        .inserts
+        .iter()
+        .any(|r| r.table_name == OTEL_RESOURCE_INFO_TABLE_NAME)
+    {
+        warn!(
+            "Skipping OTLP resource descriptor synthesis: the request writes \
+             a metric named `{OTEL_RESOURCE_INFO_TABLE_NAME}`"
+        );
+        None
+    } else {
+        resource_info.into_row_insert_requests()?
+    };
+    if resource_info.is_some() {
+        semantic_index.record_scalar(
+            OTEL_RESOURCE_INFO_TABLE_NAME,
+            SEMANTIC_METRIC_TYPE,
+            crate::semantic::METRIC_TYPE_INFO,
+        );
+        semantic_index.record_scalar(
+            OTEL_RESOURCE_INFO_TABLE_NAME,
+            SEMANTIC_METRIC_METADATA_QUALITY,
+            METADATA_QUALITY_DECLARED,
+        );
+    }
+
+    Ok(MetricsConversion {
+        requests,
+        rows,
+        semantic_index,
+        resource_info,
+    })
 }
 
 /// The tables a metric emits and their per-table `metric.type`. Histogram fans
@@ -894,6 +952,115 @@ mod tests {
             .and_then(|kv| scalar_value_string(kv.value.as_ref()))
     }
 
+    fn gauge_request(
+        resource_attrs: Vec<KeyValue>,
+        metric_name: &str,
+    ) -> ExportMetricsServiceRequest {
+        use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource;
+        ExportMetricsServiceRequest {
+            resource_metrics: vec![ResourceMetrics {
+                resource: Some(Resource {
+                    attributes: resource_attrs,
+                    ..Default::default()
+                }),
+                scope_metrics: vec![ScopeMetrics {
+                    metrics: vec![Metric {
+                        name: metric_name.to_string(),
+                        data: Some(metric::Data::Gauge(Gauge {
+                            data_points: vec![NumberDataPoint {
+                                time_unix_nano: 1_000_000,
+                                value: Some(Value::AsInt(1)),
+                                ..Default::default()
+                            }],
+                        })),
+                        ..Default::default()
+                    }],
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        }
+    }
+
+    fn column_names(request: &RowInsertRequests, table: &str) -> Vec<String> {
+        request
+            .inserts
+            .iter()
+            .find(|r| r.table_name == table)
+            .unwrap_or_else(|| panic!("missing table {table}"))
+            .rows
+            .as_ref()
+            .unwrap()
+            .schema
+            .iter()
+            .map(|c| c.column_name.clone())
+            .collect()
+    }
+
+    #[test]
+    fn test_conversion_synthesizes_resource_descriptor() {
+        set_default_prefix(None).unwrap();
+        let request = gauge_request(
+            vec![keyvalue("service.name", "api"), keyvalue("host.id", "h-1")],
+            "my_gauge",
+        );
+        let conversion =
+            to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();
+
+        // descriptor keeps raw OTel keys while the metric table's labels went
+        // through the (default underscore-escaping) translation strategy
+        let resource_info = conversion.resource_info.expect("descriptor synthesized");
+        let descriptor_cols = column_names(&resource_info, OTEL_RESOURCE_INFO_TABLE_NAME);
+        assert!(descriptor_cols.contains(&"host.id".to_string()));
+        assert!(descriptor_cols.contains(&"service.name".to_string()));
+        assert!(descriptor_cols.contains(&"job".to_string()));
+        let metric_cols = column_names(&conversion.requests, "my_gauge");
+        assert!(metric_cols.contains(&"service_name".to_string()));
+        assert!(!metric_cols.contains(&"service.name".to_string()));
+        // host.id is not in the promote list: only the descriptor keeps it
+        assert!(!metric_cols.contains(&"host_id".to_string()));
+
+        let decoded = decode(&conversion.semantic_index);
+        let t = &decoded[OTEL_RESOURCE_INFO_TABLE_NAME];
+        assert_eq!(t.get(SEMANTIC_METRIC_TYPE).map(String::as_str), Some("info"));
+        assert_eq!(
+            t.get(SEMANTIC_METRIC_METADATA_QUALITY).map(String::as_str),
+            Some("declared")
+        );
+    }
+
+    #[test]
+    fn test_conversion_skips_descriptor_for_legacy_mode() {
+        set_default_prefix(None).unwrap();
+        let request = gauge_request(vec![keyvalue("service.name", "api")], "my_gauge");
+        let mut ctx = OtlpMetricCtx {
+            is_legacy: true,
+            ..Default::default()
+        };
+        let conversion = to_grpc_insert_requests(request, &mut ctx).unwrap();
+        assert!(conversion.resource_info.is_none());
+    }
+
+    #[test]
+    fn test_conversion_skips_descriptor_on_metric_name_collision() {
+        set_default_prefix(None).unwrap();
+        let request = gauge_request(
+            vec![keyvalue("service.name", "api")],
+            OTEL_RESOURCE_INFO_TABLE_NAME,
+        );
+        let conversion =
+            to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();
+        assert!(conversion.resource_info.is_none());
+        // the metric itself still goes through the main path
+        assert!(
+            conversion
+                .requests
+                .inserts
+                .iter()
+                .any(|r| r.table_name == OTEL_RESOURCE_INFO_TABLE_NAME)
+        );
+    }
+
     #[test]
     fn test_job_composition_follows_service_namespace() {
         let mut attrs = vec![

src/servers/src/otlp/metrics/resource_info.rs

@@ -0,0 +1,258 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! OTel-native resource descriptor synthesized from OTLP metrics requests.
+//!
+//! Ordinary OTLP metrics scatter (filtered) resource attributes as tags over
+//! every emitted metric table, which is useless for entity extraction: the
+//! entity graph would have to scan every logical table and would still miss
+//! attributes dropped by the promote filter. Instead, each request projects
+//! its distinct resources into one info-metric-shaped table,
+//! [`OTEL_RESOURCE_INFO_TABLE_NAME`], which the entity-graph conventions
+//! whitelist by name.
+//!
+//! Columns are a fixed allowlist keyed by the raw OTel attribute names —
+//! deliberately independent of both the per-request label translation
+//! strategy (the conventions match fixed column names) and the resource-attr
+//! promote/ignore headers (this is entity metadata, not label promotion).
+//! `job`/`instance` are derived compatibility columns aligning the service
+//! identity with Prometheus-sourced `target_info`.
+
+use std::collections::BTreeMap;
+
+use api::v1::RowInsertRequests;
+use common_grpc::precision::Precision;
+use common_query::prelude::{greptime_timestamp, greptime_value};
+use otel_arrow_rust::proto::opentelemetry::common::v1::KeyValue;
+use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ResourceMetrics, metric};
+
+use super::{INSTANCE_KEY, JOB_KEY, scalar_value_string, service_identity};
+use crate::error::Result;
+use crate::otlp::trace::{KEY_SERVICE_NAME, KEY_SERVICE_NAMESPACE};
+use crate::row_writer::{self, MultiTableData};
+
+/// Table name of the synthesized resource descriptor; reserved in the sense
+/// that an incoming metric with the same name suppresses synthesis for its
+/// request.
+pub const OTEL_RESOURCE_INFO_TABLE_NAME: &str = "otel_resource_info";
+
+/// Resource attributes projected verbatim under their raw OTel keys.
+/// `service.instance.id` is not listed: it lands unchanged in `instance`.
+const RESOURCE_INFO_ATTRS: [&str; 9] = [
+    KEY_SERVICE_NAME,
+    KEY_SERVICE_NAMESPACE,
+    "host.id",
+    "host.name",
+    "container.id",
+    "container.name",
+    "k8s.pod.uid",
+    "k8s.pod.name",
+    "k8s.namespace.name",
+];
+
+/// Request-local resource snapshots: one row per distinct projected attribute
+/// set, stamped with the newest data-point timestamp that observed it.
+/// Cross-request dedup is left to the storage engine's last-row merge.
+#[derive(Debug, Default)]
+pub struct ResourceInfoData {
+    rows: BTreeMap<Vec<(String, String)>, i64>,
+}
+
+impl ResourceInfoData {
+    /// Projects one resource's raw (unfiltered) attributes; resources whose
+    /// projection is empty contribute nothing.
+    pub fn observe(&mut self, raw_attrs: &[KeyValue], max_ts_nanos: i64) {
+        let mut tags = BTreeMap::new();
+        let (job, instance) = service_identity(raw_attrs);
+        if let Some(job) = job {
+            tags.insert(JOB_KEY.to_string(), job);
+        }
+        if let Some(instance) = instance {
+            tags.insert(INSTANCE_KEY.to_string(), instance);
+        }
+        for kv in raw_attrs {
+            if RESOURCE_INFO_ATTRS.contains(&kv.key.as_str())
+                && let Some(value) = scalar_value_string(kv.value.as_ref())
+            {
+                tags.insert(kv.key.clone(), value);
+            }
+        }
+        if tags.is_empty() {
+            return;
+        }
+
+        let entry = self.rows.entry(tags.into_iter().collect()).or_insert(i64::MIN);
+        *entry = (*entry).max(max_ts_nanos);
+    }
+
+    /// Builds the descriptor insert: all projected attributes as tags plus
+    /// `greptime_value = 1.0`, timestamps in milliseconds. `None` when no
+    /// resource was observed.
+    pub fn into_row_insert_requests(self) -> Result<Option<RowInsertRequests>> {
+        if self.rows.is_empty() {
+            return Ok(None);
+        }
+
+        let mut writer = MultiTableData::default();
+        let table = writer.get_or_default_table_data(
+            OTEL_RESOURCE_INFO_TABLE_NAME,
+            RESOURCE_INFO_ATTRS.len() + 4,
+            self.rows.len(),
+        );
+        for (tags, ts_nanos) in self.rows {
+            let mut row = table.alloc_one_row();
+            row_writer::write_tags(table, tags.into_iter(), &mut row)?;
+            row_writer::write_f64(table, greptime_value(), 1.0, &mut row)?;
+            row_writer::write_ts_to_millis(
+                table,
+                greptime_timestamp(),
+                Some(ts_nanos),
+                Precision::Nanosecond,
+                &mut row,
+            )?;
+            table.add_row(row);
+        }
+
+        let (requests, _) = writer.into_row_insert_requests();
+        Ok(Some(requests))
+    }
+}
+
+/// The newest data-point timestamp under a resource, the descriptor row's
+/// observation time. `None` when the resource carries no data points at all.
+pub(crate) fn max_data_point_time_nanos(resource: &ResourceMetrics) -> Option<i64> {
+    let mut max_ts: Option<u64> = None;
+    let mut fold = |ts: u64| max_ts = Some(max_ts.map_or(ts, |cur| cur.max(ts)));
+    for scope in &resource.scope_metrics {
+        for m in &scope.metrics {
+            match &m.data {
+                Some(metric::Data::Gauge(g)) => {
+                    g.data_points.iter().for_each(|p| fold(p.time_unix_nano))
+                }
+                Some(metric::Data::Sum(s)) => {
+                    s.data_points.iter().for_each(|p| fold(p.time_unix_nano))
+                }
+                Some(metric::Data::Histogram(h)) => {
+                    h.data_points.iter().for_each(|p| fold(p.time_unix_nano))
+                }
+                Some(metric::Data::ExponentialHistogram(h)) => {
+                    h.data_points.iter().for_each(|p| fold(p.time_unix_nano))
+                }
+                Some(metric::Data::Summary(s)) => {
+                    s.data_points.iter().for_each(|p| fold(p.time_unix_nano))
+                }
+                None => {}
+            }
+        }
+    }
+    max_ts.map(|ts| ts as i64)
+}
+
+#[cfg(test)]
+mod tests {
+    use api::v1::SemanticType;
+    use api::v1::value::ValueData;
+    use common_query::prelude::set_default_prefix;
+    use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, any_value};
+
+    use super::*;
+
+    fn kv(key: &str, value: &str) -> KeyValue {
+        KeyValue {
+            key: key.into(),
+            value: Some(AnyValue {
+                value: Some(any_value::Value::StringValue(value.into())),
+            }),
+        }
+    }
+
+    #[test]
+    fn observe_projects_allowlist_and_dedups_per_request() {
+        let mut data = ResourceInfoData::default();
+        let attrs = vec![
+            kv("service.name", "api"),
+            kv("service.namespace", "shop"),
+            kv("service.instance.id", "inst-1"),
+            kv("host.id", "h-1"),
+            kv("os.type", "linux"),
+        ];
+        data.observe(&attrs, 100);
+        // the same resource seen again keeps one row with the newest timestamp
+        data.observe(&attrs, 50);
+        assert_eq!(data.rows.len(), 1);
+        let (tags, ts) = data.rows.iter().next().unwrap();
+        assert_eq!(*ts, 100);
+        assert!(tags.contains(&("job".to_string(), "shop/api".to_string())));
+        assert!(tags.contains(&("instance".to_string(), "inst-1".to_string())));
+        assert!(tags.contains(&("service.name".to_string(), "api".to_string())));
+        // not allowlisted / folded into instance
+        assert!(
+            tags.iter()
+                .all(|(k, _)| k != "os.type" && k != "service.instance.id")
+        );
+
+        data.observe(&[kv("host.id", "h-2")], 10);
+        assert_eq!(data.rows.len(), 2);
+    }
+
+    #[test]
+    fn observe_skips_resources_with_empty_projection() {
+        let mut data = ResourceInfoData::default();
+        data.observe(&[kv("os.type", "linux")], 100);
+        assert!(data.rows.is_empty());
+        assert!(data.into_row_insert_requests().unwrap().is_none());
+    }
+
+    #[test]
+    fn rows_carry_raw_key_tags_value_and_millis_timestamp() {
+        set_default_prefix(None).unwrap();
+        let mut data = ResourceInfoData::default();
+        data.observe(
+            &[kv("service.name", "api"), kv("host.id", "h-1")],
+            1_700_000_000_123_456_789,
+        );
+        let requests = data.into_row_insert_requests().unwrap().unwrap();
+        assert_eq!(requests.inserts.len(), 1);
+        let insert = &requests.inserts[0];
+        assert_eq!(insert.table_name, OTEL_RESOURCE_INFO_TABLE_NAME);
+
+        let rows = insert.rows.as_ref().unwrap();
+        let names = rows
+            .schema
+            .iter()
+            .map(|c| c.column_name.as_str())
+            .collect::<Vec<_>>();
+        assert_eq!(
+            names,
+            vec![
+                "host.id",
+                "job",
+                "service.name",
+                greptime_value(),
+                greptime_timestamp()
+            ]
+        );
+        for column in &rows.schema[..3] {
+            assert_eq!(column.semantic_type, SemanticType::Tag as i32);
+        }
+
+        assert_eq!(rows.rows.len(), 1);
+        let values = &rows.rows[0].values;
+        assert_eq!(values[3].value_data, Some(ValueData::F64Value(1.0)));
+        assert_eq!(
+            values[4].value_data,
+            Some(ValueData::TimestampMillisecondValue(1_700_000_000_123))
+        );
+    }
+}

src/servers/src/query_handler.rs

@@ -129,6 +129,15 @@ pub trait PromStoreProtocolHandler {
     async fn ingest_metrics(&self, metrics: Metrics) -> Result<()>;
 }
 
+/// Outcome of an OTLP metrics ingestion: the main write output plus an
+/// optional warning for the OTLP `partial_success` response (with
+/// `rejected_data_points = 0`) when a derived write — the resource
+/// descriptor — failed after the main data was committed.
+pub struct OtlpMetricsOutcome {
+    pub output: Output,
+    pub warning: Option<String>,
+}
+
 #[async_trait]
 pub trait OpenTelemetryProtocolHandler: PipelineHandler {
     /// Handling opentelemetry metrics request
@@ -136,7 +145,7 @@ pub trait OpenTelemetryProtocolHandler: PipelineHandler {
         &self,
         request: ExportMetricsServiceRequest,
         ctx: QueryContextRef,
-    ) -> Result<Output>;
+    ) -> Result<OtlpMetricsOutcome>;
 
     /// Handling opentelemetry traces request
     async fn traces(