rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-16 17:02:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(otlp): gate the resource descriptor, and fix what review found

Browse Source 
Synthesizing greptime_otel_resource_info creates and writes a table the
user never sent, so it is now off unless
otlp.experimental_enable_resource_info says otherwise. With it off the
request costs exactly what it did before the descriptor existed: nothing
is projected, no table is created, no write and no permission check
happen. Tests run with it on. StandaloneOptions carried no otlp field,
so the whole [otlp] section was silently dropped in standalone mode; map
it through, or the new option (and trace_ingest_chunk_size before it)
would do nothing there.

Renamed from otel_resource_info: the greptime_ prefix marks the table as
engine-managed and makes a collision with a user metric unlikely, which
is what the pre-existing-table ownership check and its per-request
catalog lookup were defending against. Both are gone.

A request may carry data for several graph windows, but the descriptor
folded every data-point time into one row at the newest of them, leaving
the earlier windows with metric rows and no entities. Key the rows by
window as well, and take the times from the data points the encoder
actually writes: it drops exponential histograms, and a resource
carrying nothing else was being described as an entity with no
measurements.

Projecting a resource cloned its attributes once per data point. Nest
the windows under the attributes instead, so they are moved once per
resource, and walk the data-point times through a visitor rather than
collecting a Vec per metric.

Also documents what the two maps key and hold, and lifts the projected
attribute names to constants beside KEY_SERVICE_NAME.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
Dennis Zhuang
2026-08-22 19:35:03 +08:00
parent 2b5046382e
commit 28245cdad5
25 changed files with 323 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/catalog/src/consts.rs
+5
View File
@@ -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";
src/frontend/src/instance.rs
+1
View File
@@ -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<AtomicBool>,
// cache for otlp metrics
src/frontend/src/instance/builder.rs
+1
View File
@@ -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)),
})
}
src/frontend/src/instance/entity_graph.rs
+21 -8
View File
@@ -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()
);
}
src/frontend/src/instance/otlp.rs
+13 -43
View File
@@ -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 {
src/frontend/src/service_config/otlp.rs
+7
View File
@@ -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);
src/operator/src/statement/semantic_graph.rs
+7 -5
View File
@@ -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";
src/operator/src/statement/semantic_graph/conventions.rs
+4 -3
View File
@@ -75,10 +75,11 @@ pub struct Conventions {
pub trace_co_declared_edges: Vec<EdgeRule>,
pub virtual_dst_candidates: Vec<VirtualDstCandidate>,
pub otlp_trace_entities: Vec<ImplicitEntity>,
/// Table name -> the entities that table declares, for Prometheus-sourced
/// descriptor metrics (`source = prometheus`).
pub prometheus_info_metrics: BTreeMap<String, Vec<ImplicitEntity>>,
/// 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<String, Vec<ImplicitEntity>>,
}
src/operator/src/statement/semantic_graph/conventions.yaml
+1 -1
View File
@@ -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:
src/servers/src/http/otlp.rs
+2 -1
View File
@@ -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,
}));
src/servers/src/otlp/metrics.rs
+14 -6
View File
@@ -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<String> {
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!(
src/servers/src/otlp/metrics/resource_info.rs
+149 -53
View File
@@ -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<Vec<(String, String)>, i64>,
rows: BTreeMap<Vec<(String, String)>, BTreeMap<i64, i64>>,
}
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<i64, i64> = 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<i64> {
let mut max_ts: Option<u64> = None;
let mut fold = |ts: u64| max_ts = Some(max_ts.map_or(ts, |cur| cur.max(ts)));
/// 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<Item = u64>| {
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<_>>(), 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<_>>(),
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);
src/servers/src/otlp/trace.rs
+7
View File
@@ -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
src/session/src/protocol_ctx.rs
+3
View File
@@ -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,
}
src/standalone/src/options.rs
+4 -1
View File
@@ -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,