diff --git a/config/config.md b/config/config.md index 123eaf90a20..14387741297 100644 --- a/config/config.md +++ b/config/config.md @@ -75,6 +75,7 @@ | `otlp` | -- | -- | OpenTelemetry protocol options. | | `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. | | `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. | +| `otlp.experimental_enable_resource_info` | Bool | `false` | Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric
resource attributes, so metrics-only services reach the semantic graph. | | `prom_store` | -- | -- | Prometheus remote storage options | | `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. | | `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. | @@ -317,6 +318,7 @@ | `otlp` | -- | -- | OpenTelemetry protocol options. | | `otlp.enable` | Bool | `true` | Whether to enable OpenTelemetry protocol in HTTP API. | | `otlp.trace_ingest_chunk_size` | Integer | `512` | Maximum spans per trace ingest chunk. Set to 0 to disable splitting. | +| `otlp.experimental_enable_resource_info` | Bool | `false` | Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric
resource attributes, so metrics-only services reach the semantic graph. | | `prom_store` | -- | -- | Prometheus remote storage options | | `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. | | `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index 260984a4ab3..42da7f81fe5 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -239,6 +239,9 @@ enable = true enable = true ## Maximum spans per trace ingest chunk. Set to 0 to disable splitting. trace_ingest_chunk_size = 512 +## Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric +## resource attributes, so metrics-only services reach the semantic graph. +experimental_enable_resource_info = false ## Prometheus remote storage options [prom_store] diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 068025e2a28..32ed0b12442 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -206,6 +206,9 @@ enable = true enable = true ## Maximum spans per trace ingest chunk. Set to 0 to disable splitting. trace_ingest_chunk_size = 512 +## Whether to synthesize the `greptime_otel_resource_info` table from OTLP metric +## resource attributes, so metrics-only services reach the semantic graph. +experimental_enable_resource_info = false ## Prometheus remote storage options [prom_store] diff --git a/src/common/catalog/src/consts.rs b/src/common/catalog/src/consts.rs index b976cbc256e..63aa103b83d 100644 --- a/src/common/catalog/src/consts.rs +++ b/src/common/catalog/src/consts.rs @@ -220,6 +220,11 @@ pub const SEMANTIC_RELATIONSHIPS_TABLE_NAME: &str = "semantic_relationships"; /// `semantic_relationships`; see [`is_ddl_reserved_table`] for its lifecycle. pub const SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME: &str = "semantic_relationships_declared"; +/// Width of the window the graph derivation bins observations into. Sources +/// synthesizing observations must land a row in every window they describe, +/// so this is shared rather than restated per crate. +pub const SEMANTIC_GRAPH_WINDOW_NANOS: i64 = 60 * 1_000_000_000; + // Column names of the graph tables: `catalog` exposes these schemas and the // read-time plans in `operator` must project exactly them. pub const OBSERVED_AT_COLUMN: &str = "observed_at"; diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index 78b701a3442..412ff495f65 100644 --- a/src/frontend/src/instance.rs +++ b/src/frontend/src/instance.rs @@ -134,6 +134,7 @@ pub struct Instance { slow_query_options: SlowQueryOptions, influxdb_default_merge_mode: InfluxdbMergeMode, trace_ingest_chunk_size: usize, + otlp_resource_info: bool, suspend: Arc, // cache for otlp metrics diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs index c49a8394beb..0f87dc424f0 100644 --- a/src/frontend/src/instance/builder.rs +++ b/src/frontend/src/instance/builder.rs @@ -367,6 +367,7 @@ impl FrontendBuilder { slow_query_options: self.options.slow_query.clone(), influxdb_default_merge_mode: self.options.influxdb.default_merge_mode, trace_ingest_chunk_size: self.options.otlp.trace_ingest_chunk_size, + otlp_resource_info: self.options.otlp.experimental_enable_resource_info, suspend: Arc::new(AtomicBool::new(false)), }) } diff --git a/src/frontend/src/instance/entity_graph.rs b/src/frontend/src/instance/entity_graph.rs index cd15f0e017e..3d2f5fe6e3c 100644 --- a/src/frontend/src/instance/entity_graph.rs +++ b/src/frontend/src/instance/entity_graph.rs @@ -998,9 +998,9 @@ mod tests { ]; #[test] - fn otel_resource_info_gets_implicit_declarations() { + fn greptime_otel_resource_info_gets_implicit_declarations() { let info = prom_table_info( - "otel_resource_info", + "greptime_otel_resource_info", &[ "job", "instance", @@ -1045,7 +1045,7 @@ mod tests { // an id column the table lacks drops that entity, not the table let partial = prom_table_info( - "otel_resource_info", + "greptime_otel_resource_info", &["job", "service.name", "host.id"], OTEL_STAMPS, ); @@ -1061,13 +1061,22 @@ mod tests { let labels: &[&str] = &["job", "instance", "host.id"]; // a prometheus-stamped table under the otel-whitelisted name assert!( - sorted_declarations(&prom_table_info("otel_resource_info", labels, PROM_STAMPS)) - .is_empty() + sorted_declarations(&prom_table_info( + "greptime_otel_resource_info", + labels, + PROM_STAMPS + )) + .is_empty() ); let mut stamps = OTEL_STAMPS.to_vec(); stamps.push((PHYSICAL_TABLE_METADATA_KEY, "true")); assert!( - sorted_declarations(&prom_table_info("otel_resource_info", labels, &stamps)).is_empty() + sorted_declarations(&prom_table_info( + "greptime_otel_resource_info", + labels, + &stamps + )) + .is_empty() ); // drift guard: the ingestion-side table name must stay whitelisted assert!( @@ -1083,8 +1092,12 @@ mod tests { (SEMANTIC_METRIC_TYPE, servers::semantic::METRIC_TYPE_GAUGE), ]; assert!( - sorted_declarations(&prom_table_info("otel_resource_info", labels, &wrong_type)) - .is_empty() + sorted_declarations(&prom_table_info( + "greptime_otel_resource_info", + labels, + &wrong_type + )) + .is_empty() ); } diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 8310bd6854b..2a369944c3f 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::{MITO_ENGINE, trace_operations_table_name, trace_services_table_name}; +use common_catalog::consts::{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,8 +43,7 @@ use servers::query_handler::{ use session::context::QueryContextRef; use snafu::ResultExt; use table::requests::{ - METADATA_QUALITY_DECLARED, OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM, - SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_METRIC_TYPE, SEMANTIC_PER_TABLE_INDEX_KEY, + OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM, SEMANTIC_PER_TABLE_INDEX_KEY, SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SIGNAL_TYPE_LOG, SIGNAL_TYPE_METRIC, SOURCE_OPENTELEMETRY, }; @@ -88,18 +87,6 @@ fn trace_permission_targets( /// created, so a user's same-named table never receives descriptor rows. /// Not atomic: the insert path resolves the name again, so a concurrent DROP /// plus CREATE can still slip a foreign table into that window. -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 { @@ -137,6 +124,7 @@ impl OpenTelemetryProtocolHandler for Instance { .cloned() .unwrap_or_default(); metric_ctx.is_legacy = is_legacy; + metric_ctx.resource_info = self.otlp_resource_info; let otlp::metrics::MetricsConversion { requests, @@ -187,34 +175,16 @@ impl OpenTelemetryProtocolHandler for Instance { // accepted, so every failure degrades to a warning instead. let mut warning = None; if let Some(resource_info) = resource_info { - 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()), - }, + 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_err(BoxedError::new) + .map_err(|e| e.to_string()), Err(e) => Err(e.to_string()), }; match written { diff --git a/src/frontend/src/service_config/otlp.rs b/src/frontend/src/service_config/otlp.rs index 3b01f422894..c7ec7e71486 100644 --- a/src/frontend/src/service_config/otlp.rs +++ b/src/frontend/src/service_config/otlp.rs @@ -22,6 +22,11 @@ pub struct OtlpOptions { pub enable: bool, /// Maximum spans per trace ingest chunk. Set to 0 to disable splitting. pub trace_ingest_chunk_size: usize, + /// Whether to synthesize the `greptime_otel_resource_info` descriptor + /// table from the resource attributes of OTLP metrics, so metrics-only + /// services reach the semantic graph. Off by default: it creates and + /// writes a table the user did not send. + pub experimental_enable_resource_info: bool, } impl Default for OtlpOptions { @@ -29,6 +34,7 @@ impl Default for OtlpOptions { Self { enable: true, trace_ingest_chunk_size: DEFAULT_TRACE_INGEST_CHUNK_SIZE, + experimental_enable_resource_info: false, } } } @@ -42,6 +48,7 @@ mod tests { let default = OtlpOptions::default(); assert!(default.enable); assert_eq!(default.trace_ingest_chunk_size, 512); + assert!(!default.experimental_enable_resource_info); let options: OtlpOptions = toml::from_str("enable = false").unwrap(); assert!(!options.enable); diff --git a/src/operator/src/statement/semantic_graph.rs b/src/operator/src/statement/semantic_graph.rs index bc8f28b3963..c4d3db749be 100644 --- a/src/operator/src/statement/semantic_graph.rs +++ b/src/operator/src/statement/semantic_graph.rs @@ -41,8 +41,9 @@ use common_catalog::consts::{ ENTITY_DESCRIPTIVE_COLUMN, ENTITY_ID_ATTRS_COLUMN, ENTITY_ID_COLUMN, ENTITY_SCOPE_COLUMN, ENTITY_TYPE_COLUMN, ERROR_COUNT_COLUMN, FRESH_UNTIL_COLUMN, GENERATION_ID_COLUMN, OBSERVED_AT_COLUMN, PROVENANCE_COLUMN, REL_TYPE_COLUMN, REQUEST_COUNT_COLUMN, - SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME, SOURCE_TABLES_COLUMN, SRC_ID_COLUMN, - SRC_TYPE_COLUMN, VALID_FROM_COLUMN, VALID_UNTIL_COLUMN, WINDOW_END_COLUMN, WINDOW_START_COLUMN, + SEMANTIC_GRAPH_WINDOW_NANOS, SEMANTIC_RELATIONSHIPS_DECLARED_TABLE_NAME, SOURCE_TABLES_COLUMN, + SRC_ID_COLUMN, SRC_TYPE_COLUMN, VALID_FROM_COLUMN, VALID_UNTIL_COLUMN, WINDOW_END_COLUMN, + WINDOW_START_COLUMN, }; use common_function::function::FunctionContext; use common_function::function_registry::FUNCTION_REGISTRY; @@ -122,9 +123,10 @@ pub fn declared_relationships_schema_matches(table_info: &table::metadata::Table }) } -/// Bin width for the temporal window of derived rows: 60s buckets, matching the -/// service-graph convention. -const BIN_NANOS: i64 = 60 * 1_000_000_000; +/// Bin width for the temporal window of derived rows, matching the +/// service-graph convention. Shared so ingestion-synthesized observations +/// land in the same buckets. +const BIN_NANOS: i64 = SEMANTIC_GRAPH_WINDOW_NANOS; /// Default retention for the declared-edge table; expiry slides the topology window. const DEFAULT_DECLARED_RELATIONSHIPS_TTL: &str = "90d"; diff --git a/src/operator/src/statement/semantic_graph/conventions.rs b/src/operator/src/statement/semantic_graph/conventions.rs index 365beb3b7cc..9ec6f52e158 100644 --- a/src/operator/src/statement/semantic_graph/conventions.rs +++ b/src/operator/src/statement/semantic_graph/conventions.rs @@ -75,10 +75,11 @@ pub struct Conventions { pub trace_co_declared_edges: Vec, pub virtual_dst_candidates: Vec, pub otlp_trace_entities: Vec, + /// Table name -> the entities that table declares, for Prometheus-sourced + /// descriptor metrics (`source = prometheus`). pub prometheus_info_metrics: BTreeMap>, - /// Implicit declarations of OTLP-sourced descriptor tables (the - /// ingestion-synthesized `otel_resource_info`), gated on - /// `source = opentelemetry` instead of `prometheus`. + /// Table name -> the entities that table declares, for OTLP-sourced + /// descriptor tables (`source = opentelemetry`). pub otel_info_metrics: BTreeMap>, } diff --git a/src/operator/src/statement/semantic_graph/conventions.yaml b/src/operator/src/statement/semantic_graph/conventions.yaml index 539903833c8..c64c9d21c21 100644 --- a/src/operator/src/statement/semantic_graph/conventions.yaml +++ b/src/operator/src/statement/semantic_graph/conventions.yaml @@ -136,7 +136,7 @@ prometheus_info_metrics: # descriptive columns are listed instead of swept up by descriptive_rest, # which would misfile other entities' ids under service.instance. otel_info_metrics: - otel_resource_info: + greptime_otel_resource_info: - entity: service id: [job] descriptive: diff --git a/src/servers/src/http/otlp.rs b/src/servers/src/http/otlp.rs index c82acbfc9a5..a5dfc17039e 100644 --- a/src/servers/src/http/otlp.rs +++ b/src/servers/src/http/otlp.rs @@ -113,8 +113,9 @@ pub async fn metrics( resource_attrs: http_opts.resource_attrs, promote_scope_attrs: http_opts.promote_scope_attrs, with_metric_engine, - // set is_legacy later + // set by the frontend from its config is_legacy: false, + resource_info: false, metric_type: MetricType::Init, metric_translation_strategy: http_opts.metric_translation_strategy, })); diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 47048ba677d..68bbe692d01 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -114,11 +114,11 @@ pub fn to_grpc_insert_requests( let mut resource_info = ResourceInfoData::default(); for resource in &request.resource_metrics { - if !metric_ctx.is_legacy + if metric_ctx.resource_info + && !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); + resource_info.observe(&r.attributes, resource); } let resource_attrs = resource.resource.as_ref().map(|r| { @@ -940,6 +940,14 @@ mod tests { } } + /// The descriptor is opt-in; the tests covering it turn it on. + fn descriptor_ctx() -> OtlpMetricCtx { + OtlpMetricCtx { + resource_info: true, + ..Default::default() + } + } + fn attr_value(attrs: &[KeyValue], key: &str) -> Option { attrs .iter() @@ -999,7 +1007,7 @@ mod tests { vec![keyvalue("service.name", "api"), keyvalue("host.id", "h-1")], "my_gauge", ); - let conversion = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap(); + let conversion = to_grpc_insert_requests(request, &mut descriptor_ctx()).unwrap(); // descriptor keeps raw OTel keys while the metric table's labels went // through the (default underscore-escaping) translation strategy @@ -1035,7 +1043,7 @@ mod tests { ); let mut ctx = OtlpMetricCtx { is_legacy: true, - ..Default::default() + ..descriptor_ctx() }; let conversion = to_grpc_insert_requests(request, &mut ctx).unwrap(); assert!(conversion.resource_info.is_none()); @@ -1054,7 +1062,7 @@ mod tests { vec![keyvalue("service.name", "api")], OTEL_RESOURCE_INFO_TABLE_NAME, ); - let conversion = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap(); + let conversion = to_grpc_insert_requests(request, &mut descriptor_ctx()).unwrap(); assert!(conversion.resource_info.is_none()); // the metric itself still goes through the main path assert!( diff --git a/src/servers/src/otlp/metrics/resource_info.rs b/src/servers/src/otlp/metrics/resource_info.rs index 950ced3e6ed..2d3511373d1 100644 --- a/src/servers/src/otlp/metrics/resource_info.rs +++ b/src/servers/src/otlp/metrics/resource_info.rs @@ -23,6 +23,7 @@ use std::collections::BTreeMap; use api::v1::RowInsertRequests; +use common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS; use common_grpc::precision::Precision; use common_query::prelude::{greptime_timestamp, greptime_value}; use otel_arrow_rust::proto::opentelemetry::common::v1::KeyValue; @@ -30,10 +31,15 @@ use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ResourceMetrics, metric use crate::error::Result; use crate::otlp::metrics::{INSTANCE_KEY, JOB_KEY, scalar_value_string, service_identity}; -use crate::otlp::trace::{KEY_SERVICE_NAME, KEY_SERVICE_NAMESPACE}; +use crate::otlp::trace::{ + KEY_CONTAINER_ID, KEY_CONTAINER_NAME, KEY_HOST_ID, KEY_HOST_NAME, KEY_K8S_NAMESPACE_NAME, + KEY_K8S_POD_NAME, KEY_K8S_POD_UID, KEY_SERVICE_NAME, KEY_SERVICE_NAMESPACE, +}; use crate::row_writer::{self, MultiTableData}; -pub const OTEL_RESOURCE_INFO_TABLE_NAME: &str = "otel_resource_info"; +/// Prefixed like the other engine-managed tables, so a user metric is +/// unlikely to claim the name. +pub const OTEL_RESOURCE_INFO_TABLE_NAME: &str = "greptime_otel_resource_info"; /// Attributes projected under their raw OTel keys. `service.instance.id` is /// absent on purpose: it lands in `instance`. @@ -45,13 +51,13 @@ fn is_projected_attr(key: &str) -> bool { key, KEY_SERVICE_NAME | KEY_SERVICE_NAMESPACE - | "host.id" - | "host.name" - | "container.id" - | "container.name" - | "k8s.pod.uid" - | "k8s.pod.name" - | "k8s.namespace.name" + | KEY_HOST_ID + | KEY_HOST_NAME + | KEY_CONTAINER_ID + | KEY_CONTAINER_NAME + | KEY_K8S_POD_UID + | KEY_K8S_POD_NAME + | KEY_K8S_NAMESPACE_NAME ) } @@ -59,17 +65,23 @@ fn is_projected_attr(key: &str) -> bool { /// used to size the per-row buffers. const MAX_PROJECTED_TAGS: usize = 11; -/// One row per distinct projected attribute set in a request, stamped with -/// the newest data-point time that observed it. Dedup is request-local; the -/// storage engine merges repeats across requests. +/// Projected attributes (sorted `(name, value)` pairs) -> graph window -> +/// the newest data-point time seen in that window, which is what the row for +/// that window is stamped with. +/// +/// Windows are an inner map so the attributes are stored, and moved, once per +/// resource. They are keyed separately because one request may carry data for +/// several of them: a single row per resource would describe only the newest +/// window, leaving the earlier ones with metric rows but no entities. #[derive(Debug, Default)] pub struct ResourceInfoData { - rows: BTreeMap, i64>, + rows: BTreeMap, BTreeMap>, } impl ResourceInfoData { - /// Takes the raw attributes, before the promote filter runs on them. - pub fn observe(&mut self, raw_attrs: &[KeyValue], max_ts_nanos: i64) { + /// Takes the raw attributes, before the promote filter runs on them, and + /// the times of the data points this resource actually writes. + pub fn observe(&mut self, raw_attrs: &[KeyValue], resource: &ResourceMetrics) { let mut tags = Vec::with_capacity(MAX_PROJECTED_TAGS); let (job, instance) = service_identity(raw_attrs); if let Some(job) = job { @@ -92,10 +104,25 @@ impl ResourceInfoData { // columns keep a stable order. tags.sort_unstable(); - self.rows - .entry(tags) - .and_modify(|ts| *ts = (*ts).max(max_ts_nanos)) - .or_insert(max_ts_nanos); + let mut observed: BTreeMap = BTreeMap::new(); + for_each_encoded_time(resource, |ts| { + let window = ts - ts.rem_euclid(SEMANTIC_GRAPH_WINDOW_NANOS); + observed + .entry(window) + .and_modify(|newest| *newest = (*newest).max(ts)) + .or_insert(ts); + }); + if observed.is_empty() { + return; + } + + let windows = self.rows.entry(tags).or_default(); + for (window, newest) in observed { + windows + .entry(window) + .and_modify(|seen| *seen = (*seen).max(newest)) + .or_insert(newest); + } } /// Every projected attribute becomes a tag, so auto-create puts it in the @@ -109,20 +136,22 @@ impl ResourceInfoData { let table = writer.get_or_default_table_data( OTEL_RESOURCE_INFO_TABLE_NAME, MAX_PROJECTED_TAGS + 2, - self.rows.len(), + self.rows.values().map(BTreeMap::len).sum(), ); - 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); + for (tags, windows) in &self.rows { + for ts_nanos in windows.values().copied() { + let mut row = table.alloc_one_row(); + row_writer::write_tags(table, tags.iter().cloned(), &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(); @@ -130,34 +159,35 @@ impl ResourceInfoData { } } -/// The descriptor row's observation time, so it shares a window with the -/// metric rows it describes. `None` when the resource has no data points. -pub(crate) fn max_data_point_time_nanos(resource: &ResourceMetrics) -> Option { - let mut max_ts: Option = None; - let mut fold = |ts: u64| max_ts = Some(max_ts.map_or(ts, |cur| cur.max(ts))); +/// Visits the times of the data points the encoder writes rows for, so a +/// resource is described exactly where it is measured. Exponential histograms +/// are excluded: [`super::encode_metrics`] drops them, and describing a +/// resource whose only data was dropped would invent an entity out of nothing. +fn for_each_encoded_time(resource: &ResourceMetrics, mut visit: impl FnMut(i64)) { + let mut visit_all = |points: &mut dyn Iterator| { + for ts in points { + visit(ts as i64); + } + }; 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)) + visit_all(&mut g.data_points.iter().map(|p| p.time_unix_nano)) } Some(metric::Data::Sum(s)) => { - s.data_points.iter().for_each(|p| fold(p.time_unix_nano)) + visit_all(&mut s.data_points.iter().map(|p| 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)) + visit_all(&mut h.data_points.iter().map(|p| p.time_unix_nano)) } Some(metric::Data::Summary(s)) => { - s.data_points.iter().for_each(|p| fold(p.time_unix_nano)) + visit_all(&mut s.data_points.iter().map(|p| p.time_unix_nano)) } - None => {} + Some(metric::Data::ExponentialHistogram(_)) | None => {} } } } - max_ts.map(|ts| ts as i64) } #[cfg(test)] @@ -166,6 +196,10 @@ mod tests { use api::v1::value::ValueData; use common_query::prelude::set_default_prefix; use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, any_value}; + use otel_arrow_rust::proto::opentelemetry::metrics::v1::{ + ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Metric, NumberDataPoint, + ScopeMetrics, + }; use super::*; @@ -178,6 +212,27 @@ mod tests { } } + fn gauge_at(times: &[i64]) -> ResourceMetrics { + ResourceMetrics { + scope_metrics: vec![ScopeMetrics { + metrics: vec![Metric { + data: Some(metric::Data::Gauge(Gauge { + data_points: times + .iter() + .map(|ts| NumberDataPoint { + time_unix_nano: *ts as u64, + ..Default::default() + }) + .collect(), + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + } + } + #[test] fn observe_projects_allowlist_and_dedups_per_request() { let mut data = ResourceInfoData::default(); @@ -188,11 +243,10 @@ mod tests { kv("host.id", "h-1"), kv("os.type", "linux"), ]; - data.observe(&attrs, 100); - data.observe(&attrs, 50); + data.observe(&attrs, &gauge_at(&[100, 50])); assert_eq!(data.rows.len(), 1); - let (tags, ts) = data.rows.iter().next().unwrap(); - assert_eq!(*ts, 100); + let (tags, windows) = data.rows.iter().next().unwrap(); + assert_eq!(windows.values().copied().collect::>(), vec![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()))); @@ -201,22 +255,64 @@ mod tests { .all(|(k, _)| k != "os.type" && k != "service.instance.id") ); - data.observe(&[kv("host.id", "h-2")], 10); + data.observe(&[kv("host.id", "h-2")], &gauge_at(&[10])); assert_eq!(data.rows.len(), 2); // nothing allowlisted: no row at all, rather than an empty one let mut empty = ResourceInfoData::default(); - empty.observe(&[kv("os.type", "linux")], 100); + empty.observe(&[kv("os.type", "linux")], &gauge_at(&[100])); assert!(empty.into_row_insert_requests().unwrap().is_none()); } + /// Earlier windows would keep their metric rows but lose their entities. + #[test] + fn observe_keeps_one_row_per_graph_window() { + let window = SEMANTIC_GRAPH_WINDOW_NANOS; + let mut data = ResourceInfoData::default(); + data.observe( + &[kv("service.name", "api")], + &gauge_at(&[window + 1, window + 2, 3 * window + 7]), + ); + + let windows = data.rows.values().next().unwrap(); + assert_eq!( + windows.iter().collect::>(), + vec![(&window, &(window + 2)), (&(3 * window), &(3 * window + 7))] + ); + } + + /// Describing a resource whose only data the encoder drops invents an + /// entity with no measurements. + #[test] + fn observe_ignores_data_the_encoder_drops() { + let mut data = ResourceInfoData::default(); + let dropped = ResourceMetrics { + scope_metrics: vec![ScopeMetrics { + metrics: vec![Metric { + data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram { + data_points: vec![ExponentialHistogramDataPoint { + time_unix_nano: 100, + ..Default::default() + }], + ..Default::default() + })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }; + data.observe(&[kv("service.name", "api")], &dropped); + 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, + &gauge_at(&[1_700_000_000_123_456_789]), ); let requests = data.into_row_insert_requests().unwrap().unwrap(); assert_eq!(requests.inserts.len(), 1); diff --git a/src/servers/src/otlp/trace.rs b/src/servers/src/otlp/trace.rs index 0915c5874de..9ee16507bda 100644 --- a/src/servers/src/otlp/trace.rs +++ b/src/servers/src/otlp/trace.rs @@ -48,6 +48,13 @@ pub const TRACE_STATE_COLUMN: &str = "trace_state"; pub const KEY_SERVICE_NAME: &str = "service.name"; pub const KEY_SERVICE_NAMESPACE: &str = "service.namespace"; pub const KEY_SERVICE_INSTANCE_ID: &str = "service.instance.id"; +pub const KEY_HOST_ID: &str = "host.id"; +pub const KEY_HOST_NAME: &str = "host.name"; +pub const KEY_CONTAINER_ID: &str = "container.id"; +pub const KEY_CONTAINER_NAME: &str = "container.name"; +pub const KEY_K8S_POD_UID: &str = "k8s.pod.uid"; +pub const KEY_K8S_POD_NAME: &str = "k8s.pod.name"; +pub const KEY_K8S_NAMESPACE_NAME: &str = "k8s.namespace.name"; pub const KEY_SPAN_KIND: &str = "span.kind"; // jaeger const keys, not sure if they are general diff --git a/src/session/src/protocol_ctx.rs b/src/session/src/protocol_ctx.rs index 6e84761c2d9..5165def3e49 100644 --- a/src/session/src/protocol_ctx.rs +++ b/src/session/src/protocol_ctx.rs @@ -54,6 +54,9 @@ pub struct OtlpMetricCtx { pub promote_scope_attrs: bool, pub with_metric_engine: bool, pub is_legacy: bool, + /// Set from the server's `otlp.experimental_enable_resource_info`; off + /// means the resource descriptor is not synthesized at all. + pub resource_info: bool, pub metric_type: MetricType, pub metric_translation_strategy: OtlpMetricTranslationStrategy, } diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index bedf9eebc70..4f31ac2ebdf 100644 --- a/src/standalone/src/options.rs +++ b/src/standalone/src/options.rs @@ -24,7 +24,7 @@ use file_engine::config::EngineConfig as FileEngineConfig; use flow::FlowConfig; use frontend::frontend::FrontendOptions; use frontend::service_config::{ - InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, PostgresOptions, + InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, OtlpOptions, PostgresOptions, PromStoreOptions, }; use mito2::config::MitoConfig; @@ -56,6 +56,7 @@ pub struct StandaloneOptions { pub opentsdb: OpentsdbOptions, pub influxdb: InfluxdbOptions, pub jaeger: JaegerOptions, + pub otlp: OtlpOptions, pub prom_store: PromStoreOptions, pub wal: DatanodeWalConfig, pub storage: StorageConfig, @@ -94,6 +95,7 @@ impl Default for StandaloneOptions { opentsdb: OpentsdbOptions::default(), influxdb: InfluxdbOptions::default(), jaeger: JaegerOptions::default(), + otlp: OtlpOptions::default(), prom_store: PromStoreOptions::default(), wal: DatanodeWalConfig::default(), storage: StorageConfig::default(), @@ -153,6 +155,7 @@ impl StandaloneOptions { opentsdb: cloned_opts.opentsdb, influxdb: cloned_opts.influxdb, jaeger: cloned_opts.jaeger, + otlp: cloned_opts.otlp, prom_store: cloned_opts.prom_store, meta_client: None, logging: cloned_opts.logging, diff --git a/tests-integration/src/instance.rs b/tests-integration/src/instance.rs index bdcf393c5da..ada4247e558 100644 --- a/tests-integration/src/instance.rs +++ b/tests-integration/src/instance.rs @@ -729,7 +729,7 @@ mod tests { if let auth::PermissionTableTargets::Resolved(targets) = targets && targets .iter() - .any(|target| target.table == "otel_resource_info") + .any(|target| target.table == "greptime_otel_resource_info") { return Ok(auth::PermissionResp::Reject); } @@ -782,7 +782,7 @@ mod tests { .unwrap(); let warning = outcome.warning.expect("denial must surface as a warning"); assert!( - warning.contains("otel_resource_info"), + warning.contains("greptime_otel_resource_info"), "unexpected warning: {warning}" ); @@ -800,7 +800,7 @@ mod tests { "metric data was not committed:\n{pretty_print}" ); assert!( - !pretty_print.contains("otel_resource_info"), + !pretty_print.contains("greptime_otel_resource_info"), "denied descriptor table was created:\n{pretty_print}" ); } diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 1085c21983e..16fc2e65634 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -369,6 +369,11 @@ impl GreptimeDbStandaloneBuilder { grpc: GrpcOptions::default().with_server_addr("127.0.0.1:4001"), slow_query: self.slow_query_options.clone(), auto_create_table: self.auto_create_table, + // Tests cover the descriptor, so they run with it enabled. + otlp: frontend::service_config::OtlpOptions { + experimental_enable_resource_info: true, + ..Default::default() + }, ..StandaloneOptions::default() }; diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 08617db9623..720ae6b8124 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -6493,7 +6493,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) { .await; assert_eq!(StatusCode::OK, res.status()); - let expected = "[[\"claude_code_cost_usage_USD_total\"],[\"claude_code_token_usage_tokens_total\"],[\"demo\"],[\"greptime_physical_table\"],[\"numbers\"],[\"otel_resource_info\"]]"; + let expected = "[[\"claude_code_cost_usage_USD_total\"],[\"claude_code_token_usage_tokens_total\"],[\"demo\"],[\"greptime_otel_resource_info\"],[\"greptime_physical_table\"],[\"numbers\"]]"; validate_data("otlp_metrics_all_tables", &client, "show tables;", expected).await; // Metric-engine logical table carries the semantic identity. Match substrings @@ -6567,7 +6567,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) { validate_data( "otlp_metrics_resource_info", &client, - "select job, \"service.name\", greptime_value, greptime_timestamp from otel_resource_info;", + "select job, \"service.name\", greptime_value, greptime_timestamp from greptime_otel_resource_info;", "[[\"claude-code\",\"claude-code\",1.0,1753780559836]]", ) .await; @@ -6577,7 +6577,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) { validate_data( "otlp_metrics_resource_info_options", &client, - "select count(*) from information_schema.tables where table_name = 'otel_resource_info' \ + "select count(*) from information_schema.tables where table_name = 'greptime_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%' \ @@ -6599,7 +6599,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) { .await; assert_eq!(res.status(), StatusCode::OK); let res = client - .get("/v1/sql?sql=drop table otel_resource_info;") + .get("/v1/sql?sql=drop table greptime_otel_resource_info;") .send() .await; assert_eq!(res.status(), StatusCode::OK); @@ -6734,7 +6734,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) { "otlp_metrics_resource_info_recreated", &client, "select count(*) from information_schema.tables \ - where table_name = 'otel_resource_info' and engine = 'mito';", + where table_name = 'greptime_otel_resource_info' and engine = 'mito';", "[[1]]", ) .await; @@ -6751,7 +6751,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) { .await; assert_eq!(res.status(), StatusCode::OK); let res = client - .get("/v1/sql?sql=drop table otel_resource_info;") + .get("/v1/sql?sql=drop table greptime_otel_resource_info;") .send() .await; assert_eq!(res.status(), StatusCode::OK); @@ -6777,7 +6777,7 @@ pub async fn test_otlp_metrics_new(store_type: StorageType) { validate_data( "otlp_metrics_namespace_job_descriptor", &client, - "select job, \"service.name\", \"service.namespace\", \"host.id\" from otel_resource_info;", + "select job, \"service.name\", \"service.namespace\", \"host.id\" from greptime_otel_resource_info;", "[[\"shop/api\",\"api\",\"shop\",\"h-1\"]]", ) .await; @@ -6856,7 +6856,7 @@ pub async fn test_otlp_metric_translation_strategies(store_type: StorageType) { validate_data( "otlp_metric_translation_strategy_resource_info", &client, - "select job, \"service.name\" from otel_resource_info;", + "select job, \"service.name\" from greptime_otel_resource_info;", "[[\"strategy-service\",\"strategy-service\"]]", ) .await; @@ -6864,7 +6864,7 @@ pub async fn test_otlp_metric_translation_strategies(store_type: StorageType) { "otlp_metric_translation_strategy_resource_info_allowlist", &client, "select count(*) from information_schema.columns \ - where table_name = 'otel_resource_info' and column_name = 'resource.attr';", + where table_name = 'greptime_otel_resource_info' and column_name = 'resource.attr';", "[[0]]", ) .await; @@ -6908,7 +6908,7 @@ pub async fn test_otlp_metrics_resource_info_conflicts(store_type: StorageType) // A metric named like the descriptor wins the table name: it goes through // the normal metric-engine path and no descriptor is synthesized. let content = r#" -{"resourceMetrics":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"api"}},{"key":"host.id","value":{"stringValue":"h-1"}}]},"scopeMetrics":[{"scope":{"name":"s"},"metrics":[{"name":"otel_resource_info","gauge":{"dataPoints":[{"timeUnixNano":"1753780559836000000","asDouble":1.0}]}}]}]}]} +{"resourceMetrics":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"api"}},{"key":"host.id","value":{"stringValue":"h-1"}}]},"scopeMetrics":[{"scope":{"name":"s"},"metrics":[{"name":"greptime_otel_resource_info","gauge":{"dataPoints":[{"timeUnixNano":"1753780559836000000","asDouble":1.0}]}}]}]}]} "#; let req: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap(); let res = send_req( @@ -6926,11 +6926,11 @@ pub async fn test_otlp_metrics_resource_info_conflicts(store_type: StorageType) "otlp_metrics_name_collision_engine", &client, "select count(*) from information_schema.tables \ - where table_name = 'otel_resource_info' and engine = 'metric';", + where table_name = 'greptime_otel_resource_info' and engine = 'metric';", "[[1]]", ) .await; - let res = execute_sql(&client, "drop table otel_resource_info;").await; + let res = execute_sql(&client, "drop table greptime_otel_resource_info;").await; assert_eq!(res.status(), StatusCode::OK); // A pre-existing table with an incompatible schema fails the descriptor @@ -6938,7 +6938,7 @@ pub async fn test_otlp_metrics_resource_info_conflicts(store_type: StorageType) // metric data itself is accepted. let res = execute_sql( &client, - "create table otel_resource_info (ts timestamp time index, job double, greptime_value double);", + "create table greptime_otel_resource_info (ts timestamp time index, job double, greptime_value double);", ) .await; assert_eq!(res.status(), StatusCode::OK); @@ -6960,7 +6960,9 @@ pub async fn test_otlp_metrics_resource_info_conflicts(store_type: StorageType) let partial_success = body.partial_success.as_ref().unwrap(); assert_eq!(partial_success.rejected_data_points, 0); assert!( - partial_success.error_message.contains("otel_resource_info"), + partial_success + .error_message + .contains("greptime_otel_resource_info"), "unexpected warning: {}", partial_success.error_message ); @@ -6972,51 +6974,9 @@ 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; + // The descriptor is auto-created again once the conflicting table is + // gone, and a follow-up write into it does not degrade. + let res = execute_sql(&client, "drop table greptime_otel_resource_info;").await; assert_eq!(res.status(), StatusCode::OK); for round in 0..2 { let req: ExportMetricsServiceRequest = serde_json::from_str(content).unwrap(); @@ -7039,7 +6999,7 @@ pub async fn test_otlp_metrics_resource_info_conflicts(store_type: StorageType) validate_data( "otlp_metrics_owned_descriptor_keeps_accepting", &client, - "select count(*) from otel_resource_info;", + "select count(*) from greptime_otel_resource_info;", "[[1]]", ) .await; diff --git a/tests/cases/standalone/common/system/semantic_graph.result b/tests/cases/standalone/common/system/semantic_graph.result index e7611068195..f8894319842 100644 --- a/tests/cases/standalone/common/system/semantic_graph.result +++ b/tests/cases/standalone/common/system/semantic_graph.result @@ -627,14 +627,14 @@ drop table http_requests_total; Affected Rows: 0 --- OTel conventions: the ingestion-synthesized otel_resource_info descriptor +-- OTel conventions: the ingestion-synthesized greptime_otel_resource_info descriptor -- (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 -- service.instance. The same table name stamped source=prometheus is not -- whitelisted. -create table otel_resource_info ( +create table greptime_otel_resource_info ( greptime_timestamp timestamp(3) time index, job string, instance string, @@ -655,7 +655,7 @@ create table otel_resource_info ( Affected Rows: 0 -insert into otel_resource_info values +insert into greptime_otel_resource_info values (now(), 'shop/api', 'inst-1', 'api', 'shop', 'h-1', 'node-a', 'c-1', 'api-ctr', 1), (now(), 'shop/api', 'inst-2', 'api', 'shop', '', 'laptop', '', '', 1), (now(), 'worker', '', 'worker', '', 'h-1', 'node-a', '', '', 1); @@ -667,16 +667,16 @@ select entity_type, entity_id, source_tables from greptime_private.semantic_entities order by entity_type, entity_id, source_tables; -+------------------+-----------------+-------------------------------+ -| entity_type | entity_id | source_tables | -+------------------+-----------------+-------------------------------+ -| container | c-1 | ["public.otel_resource_info"] | -| host | h-1 | ["public.otel_resource_info"] | -| service | shop/api | ["public.otel_resource_info"] | -| service | worker | ["public.otel_resource_info"] | -| service.instance | shop/api,inst-1 | ["public.otel_resource_info"] | -| service.instance | shop/api,inst-2 | ["public.otel_resource_info"] | -+------------------+-----------------+-------------------------------+ ++------------------+-----------------+----------------------------------------+ +| entity_type | entity_id | source_tables | ++------------------+-----------------+----------------------------------------+ +| container | c-1 | ["public.greptime_otel_resource_info"] | +| host | h-1 | ["public.greptime_otel_resource_info"] | +| service | shop/api | ["public.greptime_otel_resource_info"] | +| service | worker | ["public.greptime_otel_resource_info"] | +| service.instance | shop/api,inst-1 | ["public.greptime_otel_resource_info"] | +| service.instance | shop/api,inst-2 | ["public.greptime_otel_resource_info"] | ++------------------+-----------------+----------------------------------------+ -- SQLNESS PROTOCOL MYSQL select src_type, src_id, dst_type, dst_id, rel_type, provenance @@ -693,13 +693,13 @@ order by rel_type, src_id, dst_id; | service.instance | shop/api,inst-1 | host | h-1 | runs_on | attribute | +------------------+-----------------+-----------+----------+----------+------------+ -drop table otel_resource_info; +drop table greptime_otel_resource_info; Affected Rows: 0 -- The descriptor whitelist is gated on source=opentelemetry: the same table -- shape stamped as a prometheus source contributes nothing. -create table otel_resource_info ( +create table greptime_otel_resource_info ( greptime_timestamp timestamp(3) time index, job string, instance string, @@ -713,7 +713,7 @@ create table otel_resource_info ( Affected Rows: 0 -insert into otel_resource_info values +insert into greptime_otel_resource_info values (now(), 'shop/api', 'inst-1', 'h-1', 1); Affected Rows: 1 @@ -725,7 +725,7 @@ order by entity_type, entity_id; affected_rows: 0 -drop table otel_resource_info; +drop table greptime_otel_resource_info; Affected Rows: 0 @@ -751,7 +751,7 @@ create table graph_xsignal_traces ( Affected Rows: 0 -create table otel_resource_info ( +create table greptime_otel_resource_info ( greptime_timestamp timestamp(3) time index, job string, instance string, @@ -773,7 +773,7 @@ insert into graph_xsignal_traces values Affected Rows: 2 -insert into otel_resource_info values +insert into greptime_otel_resource_info values (now(), 'shop/api', 'inst-1', 'api', 'shop', 1), (now(), 'worker', 'inst-2', 'worker', '', 1); @@ -785,24 +785,24 @@ from greptime_private.semantic_entities where entity_type in ('service', 'service.instance') order by entity_type, entity_id, source_tables; -+------------------+-----------------+---------------------------------+ -| entity_type | entity_id | source_tables | -+------------------+-----------------+---------------------------------+ -| service | shop/api | ["public.otel_resource_info"] | -| service | shop/api | ["public.graph_xsignal_traces"] | -| service | worker | ["public.otel_resource_info"] | -| service | worker | ["public.graph_xsignal_traces"] | -| service.instance | shop/api,inst-1 | ["public.otel_resource_info"] | -| service.instance | shop/api,inst-1 | ["public.graph_xsignal_traces"] | -| service.instance | worker,inst-2 | ["public.otel_resource_info"] | -| service.instance | worker,inst-2 | ["public.graph_xsignal_traces"] | -+------------------+-----------------+---------------------------------+ ++------------------+-----------------+----------------------------------------+ +| entity_type | entity_id | source_tables | ++------------------+-----------------+----------------------------------------+ +| service | shop/api | ["public.graph_xsignal_traces"] | +| service | shop/api | ["public.greptime_otel_resource_info"] | +| service | worker | ["public.graph_xsignal_traces"] | +| service | worker | ["public.greptime_otel_resource_info"] | +| service.instance | shop/api,inst-1 | ["public.graph_xsignal_traces"] | +| service.instance | shop/api,inst-1 | ["public.greptime_otel_resource_info"] | +| service.instance | worker,inst-2 | ["public.graph_xsignal_traces"] | +| service.instance | worker,inst-2 | ["public.greptime_otel_resource_info"] | ++------------------+-----------------+----------------------------------------+ drop table graph_xsignal_traces; Affected Rows: 0 -drop table otel_resource_info; +drop table greptime_otel_resource_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 5a92177f02b..60d95630d91 100644 --- a/tests/cases/standalone/common/system/semantic_graph.sql +++ b/tests/cases/standalone/common/system/semantic_graph.sql @@ -388,14 +388,14 @@ drop table target_info; drop table http_requests_total; --- OTel conventions: the ingestion-synthesized otel_resource_info descriptor +-- OTel conventions: the ingestion-synthesized greptime_otel_resource_info descriptor -- (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 -- service.instance. The same table name stamped source=prometheus is not -- whitelisted. -create table otel_resource_info ( +create table greptime_otel_resource_info ( greptime_timestamp timestamp(3) time index, job string, instance string, @@ -414,7 +414,7 @@ create table otel_resource_info ( 'greptime.semantic.metric.type' = 'info' ); -insert into otel_resource_info values +insert into greptime_otel_resource_info values (now(), 'shop/api', 'inst-1', 'api', 'shop', 'h-1', 'node-a', 'c-1', 'api-ctr', 1), (now(), 'shop/api', 'inst-2', 'api', 'shop', '', 'laptop', '', '', 1), (now(), 'worker', '', 'worker', '', 'h-1', 'node-a', '', '', 1); @@ -429,11 +429,11 @@ select src_type, src_id, dst_type, dst_id, rel_type, provenance from greptime_private.semantic_relationships order by rel_type, src_id, dst_id; -drop table otel_resource_info; +drop table greptime_otel_resource_info; -- The descriptor whitelist is gated on source=opentelemetry: the same table -- shape stamped as a prometheus source contributes nothing. -create table otel_resource_info ( +create table greptime_otel_resource_info ( greptime_timestamp timestamp(3) time index, job string, instance string, @@ -445,7 +445,7 @@ create table otel_resource_info ( 'greptime.semantic.source' = 'prometheus' ); -insert into otel_resource_info values +insert into greptime_otel_resource_info values (now(), 'shop/api', 'inst-1', 'h-1', 1); -- SQLNESS PROTOCOL MYSQL @@ -453,7 +453,7 @@ select entity_type, entity_id from greptime_private.semantic_entities order by entity_type, entity_id; -drop table otel_resource_info; +drop table greptime_otel_resource_info; -- Cross-signal identity: the same service reaches the graph as trace spans and -- as an ingestion-synthesized descriptor. The two sources name their identity @@ -475,7 +475,7 @@ create table graph_xsignal_traces ( "resource_attributes.service.instance.id") ) with ('table_data_model' = 'greptime_trace_v1', 'append_mode' = 'true'); -create table otel_resource_info ( +create table greptime_otel_resource_info ( greptime_timestamp timestamp(3) time index, job string, instance string, @@ -493,7 +493,7 @@ insert into graph_xsignal_traces values (now(), 't1', 's1', NULL, 'SPAN_KIND_SERVER', 'STATUS_CODE_UNSET', 'api', 100, 'shop', 'inst-1'), (now(), 't2', 's2', NULL, 'SPAN_KIND_SERVER', 'STATUS_CODE_UNSET', 'worker', 100, '', 'inst-2'); -insert into otel_resource_info values +insert into greptime_otel_resource_info values (now(), 'shop/api', 'inst-1', 'api', 'shop', 1), (now(), 'worker', 'inst-2', 'worker', '', 1); @@ -505,4 +505,4 @@ order by entity_type, entity_id, source_tables; drop table graph_xsignal_traces; -drop table otel_resource_info; +drop table greptime_otel_resource_info; diff --git a/tests/conf/frontend-test.toml.template b/tests/conf/frontend-test.toml.template index 632947a3cde..5a879e04882 100644 --- a/tests/conf/frontend-test.toml.template +++ b/tests/conf/frontend-test.toml.template @@ -7,3 +7,6 @@ parallelism = 4 [event_recorder] event_types = [] + +[otlp] +experimental_enable_resource_info = true diff --git a/tests/conf/standalone-test.toml.template b/tests/conf/standalone-test.toml.template index ebe25f82487..8d7e307d511 100644 --- a/tests/conf/standalone-test.toml.template +++ b/tests/conf/standalone-test.toml.template @@ -53,3 +53,6 @@ retry_delay = "500ms" [event_recorder] event_types = [] + +[otlp] +experimental_enable_resource_info = true