From 39bba171c9fca786fc193f49c6a91fd7b4353c5e Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Mon, 17 Aug 2026 23:43:27 +0800 Subject: [PATCH] fix(otlp): guard descriptor writes with semantic ownership markers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- src/frontend/src/instance/entity_graph.rs | 33 ++++++-- src/frontend/src/instance/otlp.rs | 81 ++++++++++++++----- .../statement/semantic_graph/conventions.yaml | 13 +-- src/servers/src/otel_arrow.rs | 8 +- tests-integration/tests/http.rs | 76 +++++++++++++++++ .../common/system/semantic_graph.result | 5 +- .../common/system/semantic_graph.sql | 5 +- 7 files changed, 179 insertions(+), 42 deletions(-) diff --git a/src/frontend/src/instance/entity_graph.rs b/src/frontend/src/instance/entity_graph.rs index 8136aee37b0..0c4c21c1266 100644 --- a/src/frontend/src/instance/entity_graph.rs +++ b/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>, expected_source: &str, + expected_metric_type: Option<&str>, declarations: &mut Vec, ) { 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] diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 067e47073ee..ce00270b055 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/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 + )); + } } } diff --git a/src/operator/src/statement/semantic_graph/conventions.yaml b/src/operator/src/statement/semantic_graph/conventions.yaml index 0546e763940..f4b2f92eee6 100644 --- a/src/operator/src/statement/semantic_graph/conventions.yaml +++ b/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 diff --git a/src/servers/src/otel_arrow.rs b/src/servers/src/otel_arrow.rs index 10b52102dff..9c6fc112ee8 100644 --- a/src/servers/src/otel_arrow.rs +++ b/src/servers/src/otel_arrow.rs @@ -73,7 +73,7 @@ impl ArrowMetricsService for OtelArrowServiceHandler { - // 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) => { diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index c3e3271eaba..d903424f0d7 100644 --- a/tests-integration/tests/http.rs +++ b/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; } diff --git a/tests/cases/standalone/common/system/semantic_graph.result b/tests/cases/standalone/common/system/semantic_graph.result index 64ffca7c688..18f61ba9923 100644 --- a/tests/cases/standalone/common/system/semantic_graph.result +++ b/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 diff --git a/tests/cases/standalone/common/system/semantic_graph.sql b/tests/cases/standalone/common/system/semantic_graph.sql index 32eb7c34837..837396fbed7 100644 --- a/tests/cases/standalone/common/system/semantic_graph.sql +++ b/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