diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 459a05321c..8692a2ad35 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/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 { + ) -> ServerResult { self.plugins .get::() .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)] diff --git a/src/frontend/src/metrics.rs b/src/frontend/src/metrics.rs index aba33637cf..f649db0b46 100644 --- a/src/frontend/src/metrics.rs +++ b/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", diff --git a/src/servers/src/http/otlp.rs b/src/servers/src/http/otlp.rs index 6540f6d4c4..c82acbfc9a 100644 --- a/src/servers/src/http/otlp.rs +++ b/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, }) } diff --git a/src/servers/src/otel_arrow.rs b/src/servers/src/otel_arrow.rs index f905b39f5b..10b52102df 100644 --- a/src/servers/src/otel_arrow.rs +++ b/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 { + // 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; } diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 0c8794a08a..e17fd2321b 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/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, +} + /// Convert OpenTelemetry metrics to GreptimeDB insert requests /// /// See /// /// 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 { 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, + 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 { + 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![ diff --git a/src/servers/src/otlp/metrics/resource_info.rs b/src/servers/src/otlp/metrics/resource_info.rs new file mode 100644 index 0000000000..47710815dc --- /dev/null +++ b/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, 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> { + 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 { + let mut max_ts: Option = 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::>(); + 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)) + ); + } +} diff --git a/src/servers/src/query_handler.rs b/src/servers/src/query_handler.rs index a6a086e0f2..86258a3ea8 100644 --- a/src/servers/src/query_handler.rs +++ b/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, +} + #[async_trait] pub trait OpenTelemetryProtocolHandler: PipelineHandler { /// Handling opentelemetry metrics request @@ -136,7 +145,7 @@ pub trait OpenTelemetryProtocolHandler: PipelineHandler { &self, request: ExportMetricsServiceRequest, ctx: QueryContextRef, - ) -> Result; + ) -> Result; /// Handling opentelemetry traces request async fn traces(