fix(otlp): guard descriptor writes with semantic ownership markers

A pre-existing schema-compatible table named otel_resource_info would
silently receive descriptor rows while its missing semantic stamps kept
it out of the entity graph. The descriptor write now requires the
auto-created table's ownership markers (mito engine + signal_type +
source + metric.type=info + metadata_quality=declared) and otherwise
degrades to the partial-success warning; the entity-graph gate for the
otel whitelist likewise requires metric.type=info, so a user table
stamped with only signal/source no longer picks up implicit
declarations.

Also fold the descriptor write cost into the response and surface the
degrade warning through the otel-arrow BatchStatus status_message.
Integration tests pin the full marker set on auto-create and that an
existing owned descriptor keeps accepting writes without degrading —
a missing marker would otherwise silently stop every descriptor write
after the first request.

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

src/frontend/src/instance/entity_graph.rs

@@ -60,8 +60,9 @@ use table::TableRef;
 use table::metadata::TableInfo;
 use table::predicate::{TimeRangeExtraction, extract_time_range_strict};
 use table::requests::{
-    EntityRole, SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SIGNAL_TYPE_METRIC, SOURCE_OPENTELEMETRY,
-    SOURCE_PROMETHEUS, is_trace_v1_table, parse_entity_columns, parse_entity_option_key,
+    EntityRole, SEMANTIC_METRIC_TYPE, SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SIGNAL_TYPE_METRIC,
+    SOURCE_OPENTELEMETRY, SOURCE_PROMETHEUS, is_trace_v1_table, parse_entity_columns,
+    parse_entity_option_key,
 };
 
 use crate::error;
@@ -209,12 +210,14 @@ impl EntityGraphProviderImpl {
             table_info,
             &conventions.prometheus_info_metrics,
             SOURCE_PROMETHEUS,
+            None,
             &mut declarations,
         );
         Self::extend_with_info_metric_conventions(
             table_info,
             &conventions.otel_info_metrics,
             SOURCE_OPENTELEMETRY,
+            Some(servers::semantic::METRIC_TYPE_INFO),
             &mut declarations,
         );
         declarations
@@ -231,13 +234,15 @@ impl EntityGraphProviderImpl {
     /// Implicit declarations of the well-known entity-descriptor metrics
     /// (the `prometheus_info_metrics` / `otel_info_metrics` whitelists of
     /// `conventions.yaml`), gated on the ingest-stamped `signal_type=metric`
-    /// option plus the whitelist's expected `source`, and keyed by table
-    /// name. The metric engine's physical table aggregates every logical
-    /// table's columns and must not contribute a duplicate source.
+    /// option plus the whitelist's expected `source`; OTel descriptors also
+    /// require `metric.type=info`. The metric engine's physical table
+    /// aggregates every logical table's columns and must not contribute a
+    /// duplicate source.
     fn extend_with_info_metric_conventions(
         table_info: &TableInfo,
         whitelist: &BTreeMap<String, Vec<ImplicitEntity>>,
         expected_source: &str,
+        expected_metric_type: Option<&str>,
         declarations: &mut Vec<EntityDeclaration>,
     ) {
         let Some(implicit_entities) = whitelist.get(&table_info.name) else {
@@ -246,11 +251,14 @@ impl EntityGraphProviderImpl {
         let options = &table_info.meta.options.extra_options;
         if options.get(SEMANTIC_SIGNAL_TYPE).map(String::as_str) != Some(SIGNAL_TYPE_METRIC)
             || options.get(SEMANTIC_SOURCE).map(String::as_str) != Some(expected_source)
+            || expected_metric_type.is_some_and(|expected| {
+                options.get(SEMANTIC_METRIC_TYPE).map(String::as_str) != Some(expected)
+            })
             || table_info.is_physical_table()
         {
             debug!(
-                "Table `{}` matches the info-metric whitelist but is not a `{expected_source}` \
-                 logical metric table; skipping its implicit declarations",
+                "Table `{}` matches the info-metric whitelist but is not an eligible \
+                 `{expected_source}` info-metric source; skipping its implicit declarations",
                 table_info.name
             );
             return;
@@ -958,6 +966,7 @@ mod tests {
     const OTEL_STAMPS: &[(&str, &str)] = &[
         (SEMANTIC_SIGNAL_TYPE, SIGNAL_TYPE_METRIC),
         (SEMANTIC_SOURCE, SOURCE_OPENTELEMETRY),
+        (SEMANTIC_METRIC_TYPE, servers::semantic::METRIC_TYPE_INFO),
     ];
 
     #[test]
@@ -1042,6 +1051,16 @@ mod tests {
                 .otel_info_metrics
                 .contains_key(servers::otlp::metrics::OTEL_RESOURCE_INFO_TABLE_NAME)
         );
+
+        let wrong_type = [
+            (SEMANTIC_SIGNAL_TYPE, SIGNAL_TYPE_METRIC),
+            (SEMANTIC_SOURCE, SOURCE_OPENTELEMETRY),
+            (SEMANTIC_METRIC_TYPE, servers::semantic::METRIC_TYPE_GAUGE),
+        ];
+        assert!(
+            sorted_declarations(&prom_table_info("otel_resource_info", labels, &wrong_type))
+                .is_empty()
+        );
     }
 
     #[test]

src/frontend/src/instance/otlp.rs

@@ -24,7 +24,7 @@ use auth::{
     PermissionTableTargets,
 };
 use client::Output;
-use common_catalog::consts::{trace_operations_table_name, trace_services_table_name};
+use common_catalog::consts::{MITO_ENGINE, trace_operations_table_name, trace_services_table_name};
 use common_error::ext::BoxedError;
 use common_query::prelude::GREPTIME_PHYSICAL_TABLE;
 use common_telemetry::tracing;
@@ -43,7 +43,8 @@ use servers::query_handler::{
 use session::context::QueryContextRef;
 use snafu::ResultExt;
 use table::requests::{
-    OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM, SEMANTIC_PER_TABLE_INDEX_KEY,
+    METADATA_QUALITY_DECLARED, OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM,
+    SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_METRIC_TYPE, SEMANTIC_PER_TABLE_INDEX_KEY,
     SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SIGNAL_TYPE_LOG, SIGNAL_TYPE_METRIC,
     SOURCE_OPENTELEMETRY,
 };
@@ -83,6 +84,19 @@ fn trace_permission_targets(
     PermissionTableTargets::resolved(targets)
 }
 
+fn is_owned_resource_info_table(table: &table::metadata::TableInfo) -> bool {
+    let options = &table.meta.options.extra_options;
+    table.meta.engine == MITO_ENGINE
+        && options.get(SEMANTIC_SIGNAL_TYPE).map(String::as_str) == Some(SIGNAL_TYPE_METRIC)
+        && options.get(SEMANTIC_SOURCE).map(String::as_str) == Some(SOURCE_OPENTELEMETRY)
+        && options.get(SEMANTIC_METRIC_TYPE).map(String::as_str)
+            == Some(servers::semantic::METRIC_TYPE_INFO)
+        && options
+            .get(SEMANTIC_METRIC_METADATA_QUALITY)
+            .map(String::as_str)
+            == Some(METADATA_QUALITY_DECLARED)
+}
+
 #[async_trait]
 impl OpenTelemetryProtocolHandler for Instance {
     #[tracing::instrument(skip_all)]
@@ -148,7 +162,7 @@ impl OpenTelemetryProtocolHandler for Instance {
         };
 
         // If the user uses the legacy path, it is by default without metric engine.
-        let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
+        let mut 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)
@@ -171,26 +185,49 @@ impl OpenTelemetryProtocolHandler for Instance {
         // already-accepted data; both degrade to a partial-success warning.
         let mut warning = None;
         if let Some(resource_info) = resource_info {
-            let written = match self.check_row_insert_permission(
-                &resource_info,
-                &ctx,
-                PermissionReq::Action(OTLP_WRITE),
-            ) {
-                Ok(_) => self
-                    .handle_row_inserts(resource_info, ctx, false, false)
-                    .await
-                    .map(|_| ())
-                    .map_err(BoxedError::new),
-                Err(e) => Err(BoxedError::new(e)),
+            let existing = self
+                .catalog_manager()
+                .table(
+                    ctx.current_catalog(),
+                    &ctx.current_schema(),
+                    otlp::metrics::OTEL_RESOURCE_INFO_TABLE_NAME,
+                    None,
+                )
+                .await;
+            let written = match existing {
+                Ok(Some(table)) if !is_owned_resource_info_table(&table.table_info()) => {
+                    Err(format!(
+                        "table `{}` already exists without the descriptor's semantic ownership markers",
+                        otlp::metrics::OTEL_RESOURCE_INFO_TABLE_NAME
+                    ))
+                }
+                Ok(_) => match self.check_row_insert_permission(
+                    &resource_info,
+                    &ctx,
+                    PermissionReq::Action(OTLP_WRITE),
+                ) {
+                    Ok(_) => self
+                        .handle_row_inserts(resource_info, ctx, false, false)
+                        .await
+                        .map_err(BoxedError::new)
+                        .map_err(|e| e.to_string()),
+                    Err(e) => Err(e.to_string()),
+                },
+                Err(e) => Err(e.to_string()),
             };
-            if let Err(e) = written {
-                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
-                ));
+            match written {
+                Ok(descriptor_output) => output.meta.cost += descriptor_output.meta.cost,
+                Err(e) => {
+                    OTLP_RESOURCE_INFO_WRITE_ERRORS.inc();
+                    common_telemetry::warn!(
+                        "Failed to write the OTLP resource descriptor table: {e}"
+                    );
+                    warning = Some(format!(
+                        "metric data was accepted, but writing the resource \
+                         descriptor table `{}` failed: {e}",
+                        otlp::metrics::OTEL_RESOURCE_INFO_TABLE_NAME
+                    ));
+                }
             }
         }
 

src/operator/src/statement/semantic_graph/conventions.yaml

@@ -127,12 +127,13 @@ prometheus_info_metrics:
       descriptive_rest: true
 
 # Implicit declarations for the OTLP resource descriptor synthesized at
-# metrics ingestion, gated on source=opentelemetry. Columns are the raw OTel
-# attribute keys; job/instance are the derived Prometheus-compatible service
-# identity, so services line up with target_info-sourced entities. Every
-# descriptive column is explicit: the descriptor's attribute set is a known
-# allowlist, and identifying attributes of other entities must not leak into
-# service.instance the way target_info's descriptive_rest would.
+# metrics ingestion, gated on source=opentelemetry and metric.type=info.
+# Columns are the raw OTel attribute keys; job/instance are the derived
+# Prometheus-compatible service identity, so services line up with
+# target_info-sourced entities. Every descriptive column is explicit: the
+# descriptor's attribute set is a known allowlist, and identifying attributes
+# of other entities must not leak into service.instance the way target_info's
+# descriptive_rest would.
 otel_info_metrics:
   otel_resource_info:
     - entity: service

src/servers/src/otel_arrow.rs

@@ -73,7 +73,7 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
                         return;
                     }
                 };
-                let batch_status = BatchStatus {
+                let mut batch_status = BatchStatus {
                     batch_id: batch.batch_id,
                     status_code: 0,
                     status_message: Default::default(),
@@ -101,10 +101,12 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
                 // use metric engine by default
                 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}");
+                            // OTAP has no rejected-count field. Keep the batch
+                            // successful and carry the derived-write warning in
+                            // its diagnostic message.
+                            batch_status.status_message = warning;
                         }
                     }
                     Err(e) => {

tests-integration/tests/http.rs

@@ -6571,12 +6571,16 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) {
         "[[\"claude-code\",\"claude-code\",1.0,1753780559836]]",
     )
     .await;
+    // All four ownership markers must land on auto-create: the descriptor
+    // write path refuses tables that miss any of them, so a missing stamp
+    // here would silently stop every follow-up descriptor write.
     validate_data(
         "otlp_metrics_resource_info_options",
         &client,
         "select count(*) from information_schema.tables where table_name = 'otel_resource_info' \
          and engine = 'mito' \
          and create_options like '%greptime.semantic.metric.type=info%' \
+         and create_options like '%greptime.semantic.metric.metadata_quality=declared%' \
          and create_options like '%greptime.semantic.signal_type=metric%' \
          and create_options like '%greptime.semantic.source=opentelemetry%';",
         "[[1]]",
@@ -6968,6 +6972,78 @@ pub async fn test_otlp_metrics_resource_info_conflicts(store_type: StorageType)
     )
     .await;
 
+    let res = execute_sql(&client, "drop table otel_resource_info;").await;
+    assert_eq!(res.status(), StatusCode::OK);
+    let res = execute_sql(
+        &client,
+        "create table otel_resource_info (\
+         greptime_timestamp timestamp time index, greptime_value double, \
+         job string, \"service.name\" string, primary key(job, \"service.name\"));",
+    )
+    .await;
+    assert_eq!(res.status(), StatusCode::OK);
+
+    // A schema-compatible user table is still not descriptor-owned. Writing
+    // into it would appear successful while its missing semantic stamps keep
+    // it out of the entity graph, so reject only the derived write and warn.
+    let req: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap();
+    let res = send_req(
+        &client,
+        vec![content_type()],
+        "/v1/otlp/v1/metrics",
+        req.encode_to_vec(),
+        false,
+    )
+    .await;
+    assert_eq!(StatusCode::OK, res.status());
+    let body = ExportMetricsServiceResponse::decode(res.bytes().await).unwrap();
+    let partial_success = body.partial_success.as_ref().unwrap();
+    assert_eq!(partial_success.rejected_data_points, 0);
+    assert!(
+        partial_success.error_message.contains("semantic ownership"),
+        "unexpected warning: {}",
+        partial_success.error_message
+    );
+    validate_data(
+        "otlp_metrics_compatible_foreign_table_untouched",
+        &client,
+        "select count(*) from otel_resource_info;",
+        "[[0]]",
+    )
+    .await;
+
+    // Once the foreign table is gone, the descriptor is auto-created and a
+    // follow-up write into the now-existing owned table passes the ownership
+    // check without degrading — guards against a marker drift that would
+    // silently stop every descriptor write after the first request.
+    let res = execute_sql(&client, "drop table otel_resource_info;").await;
+    assert_eq!(res.status(), StatusCode::OK);
+    for round in 0..2 {
+        let req: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap();
+        let res = send_req(
+            &client,
+            vec![content_type()],
+            "/v1/otlp/v1/metrics",
+            req.encode_to_vec(),
+            false,
+        )
+        .await;
+        assert_eq!(StatusCode::OK, res.status());
+        let body = ExportMetricsServiceResponse::decode(res.bytes().await).unwrap();
+        assert!(
+            body.partial_success.is_none(),
+            "round {round} unexpectedly degraded: {:?}",
+            body.partial_success
+        );
+    }
+    validate_data(
+        "otlp_metrics_owned_descriptor_keeps_accepting",
+        &client,
+        "select count(*) from otel_resource_info;",
+        "[[1]]",
+    )
+    .await;
+
     guard.remove_all().await;
 }
 

tests/cases/standalone/common/system/semantic_graph.result

@@ -628,7 +628,7 @@ drop table http_requests_total;
 Affected Rows: 0
 
 -- OTel conventions: the ingestion-synthesized otel_resource_info descriptor
--- (stamped signal_type=metric + source=opentelemetry) gets implicit
+-- (stamped signal_type=metric + source=opentelemetry + metric.type=info) gets implicit
 -- declarations under the raw OTel column names. host/container identities are
 -- the stable ids only, so rows with an empty host.id / container.id link no
 -- infrastructure, and a row without an instance value declares no
@@ -649,7 +649,8 @@ create table otel_resource_info (
     "host.id", "host.name", "container.id", "container.name")
 ) with (
   'greptime.semantic.signal_type' = 'metric',
-  'greptime.semantic.source' = 'opentelemetry'
+  'greptime.semantic.source' = 'opentelemetry',
+  'greptime.semantic.metric.type' = 'info'
 );
 
 Affected Rows: 0

tests/cases/standalone/common/system/semantic_graph.sql

@@ -389,7 +389,7 @@ drop table target_info;
 drop table http_requests_total;
 
 -- OTel conventions: the ingestion-synthesized otel_resource_info descriptor
--- (stamped signal_type=metric + source=opentelemetry) gets implicit
+-- (stamped signal_type=metric + source=opentelemetry + metric.type=info) gets implicit
 -- declarations under the raw OTel column names. host/container identities are
 -- the stable ids only, so rows with an empty host.id / container.id link no
 -- infrastructure, and a row without an instance value declares no
@@ -410,7 +410,8 @@ create table otel_resource_info (
     "host.id", "host.name", "container.id", "container.name")
 ) with (
   'greptime.semantic.signal_type' = 'metric',
-  'greptime.semantic.source' = 'opentelemetry'
+  'greptime.semantic.source' = 'opentelemetry',
+  'greptime.semantic.metric.type' = 'info'
 );
 
 insert into otel_resource_info values