mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-13 07:22:32 +00:00
perf(otlp): build descriptor rows without the per-resource BTreeMap
Projecting a resource allocated a BTreeMap and then collected it into the row key, and every attribute was matched against the allowlist by linear scan. Collect the tags into a Vec and sort once, and match the allowlist instead of scanning it. Measured on the conversion path: descriptor work drops 16-18%, from 10.6% to 8.9% of conversion CPU on the worst shape (1000 resources with 4 data points each), where the cost tracks resource count rather than data-point count. Also trims the comments and tests added with the descriptor to what carries information. Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
@@ -1003,7 +1003,6 @@ mod tests {
|
||||
"service.instance"
|
||||
]
|
||||
);
|
||||
// host identity is host.id only; host.name stays descriptive
|
||||
assert_eq!(declarations[1].id_columns, vec!["host.id"]);
|
||||
assert_eq!(declarations[1].descriptive_columns, vec!["host.name"]);
|
||||
assert_eq!(declarations[3].id_columns, vec!["job"]);
|
||||
@@ -1011,20 +1010,18 @@ mod tests {
|
||||
declarations[3].descriptive_columns,
|
||||
vec!["service.name", "service.namespace"]
|
||||
);
|
||||
// no descriptive_rest: other entities' identifying attributes must
|
||||
// not leak into service.instance
|
||||
// no descriptive_rest: other entities' ids must not be filed under
|
||||
// service.instance
|
||||
assert_eq!(declarations[4].id_columns, vec!["job", "instance"]);
|
||||
assert!(declarations[4].descriptive_columns.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn otel_resource_info_missing_id_columns_drop_entities() {
|
||||
let info = prom_table_info(
|
||||
// an id column the table lacks drops that entity, not the table
|
||||
let partial = prom_table_info(
|
||||
"otel_resource_info",
|
||||
&["job", "service.name", "host.id"],
|
||||
OTEL_STAMPS,
|
||||
);
|
||||
let types: Vec<String> = sorted_declarations(&info)
|
||||
let types: Vec<String> = sorted_declarations(&partial)
|
||||
.into_iter()
|
||||
.map(|d| d.entity_type)
|
||||
.collect();
|
||||
|
||||
@@ -84,6 +84,10 @@ fn trace_permission_targets(
|
||||
PermissionTableTargets::resolved(targets)
|
||||
}
|
||||
|
||||
/// Whether the table under the descriptor's name is the one ingestion
|
||||
/// 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
|
||||
@@ -178,11 +182,9 @@ impl OpenTelemetryProtocolHandler for Instance {
|
||||
.context(error::ExecuteGrpcQuerySnafu)?
|
||||
};
|
||||
|
||||
// The descriptor is derived enrichment written after the main data is
|
||||
// committed: neither a permission denial on its table nor a write
|
||||
// failure (e.g. a conflicting pre-existing table, or auto-create
|
||||
// disabled) must fail the request and trigger client retries of
|
||||
// already-accepted data; both degrade to a partial-success warning.
|
||||
// Derived enrichment, written after the metric data is committed:
|
||||
// failing here would make the client retry data the server already
|
||||
// accepted, so every failure degrades to a warning instead.
|
||||
let mut warning = None;
|
||||
if let Some(resource_info) = resource_info {
|
||||
let existing = self
|
||||
|
||||
@@ -32,9 +32,8 @@ virtual_dst_candidates:
|
||||
- { column: span_attributes.server.address, connection_type: virtual_node }
|
||||
|
||||
# Implicit declarations for greptime_trace_v1 tables; the pod UID bridges
|
||||
# trace-side pods onto the kube-state-metrics entities below. Host and
|
||||
# container identities are the stable ids only — host.name and container.name
|
||||
# vary by SDK and resource detector, so they stay descriptive.
|
||||
# trace-side pods onto the kube-state-metrics entities below. host.name and
|
||||
# container.name vary by SDK and resource detector, so identity is the id.
|
||||
otlp_trace_entities:
|
||||
- entity: service
|
||||
id: [service_name]
|
||||
@@ -128,12 +127,10 @@ prometheus_info_metrics:
|
||||
|
||||
# Implicit declarations for the OTLP resource descriptor synthesized at
|
||||
# metrics ingestion, gated on source=opentelemetry and metric.type=info.
|
||||
# Columns are the raw OTel attribute keys; job/instance are the derived
|
||||
# Prometheus-compatible service identity, so services line up with
|
||||
# target_info-sourced entities. Every descriptive column is explicit: the
|
||||
# descriptor's attribute set is a known allowlist, and identifying attributes
|
||||
# of other entities must not leak into service.instance the way target_info's
|
||||
# descriptive_rest would.
|
||||
# job/instance are the derived Prometheus-compatible identity, so services
|
||||
# line up with target_info. Unlike target_info the attribute set is known, so
|
||||
# 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:
|
||||
- entity: service
|
||||
|
||||
@@ -90,14 +90,13 @@ 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).
|
||||
/// 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.
|
||||
/// The synthesized resource descriptor, written separately from the
|
||||
/// metric data. See [`resource_info`].
|
||||
pub resource_info: Option<RowInsertRequests>,
|
||||
}
|
||||
|
||||
@@ -153,9 +152,8 @@ pub fn to_grpc_insert_requests(
|
||||
|
||||
let (requests, rows) = table_writer.into_row_insert_requests();
|
||||
|
||||
// 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.
|
||||
// The metric and the descriptor would fight over the same table across
|
||||
// two engines. The user's metric wins.
|
||||
let resource_info = if requests
|
||||
.inserts
|
||||
.iter()
|
||||
@@ -290,8 +288,7 @@ fn from_metric_type(data: &metric::Data) -> MetricType {
|
||||
}
|
||||
}
|
||||
|
||||
/// Renders a scalar attribute value as a string label value. Non-scalar
|
||||
/// values (bool, arrays, maps, bytes) are not representable as tags.
|
||||
/// Non-scalar values (bool, arrays, maps, bytes) are not representable as tags.
|
||||
fn scalar_value_string(value: Option<&AnyValue>) -> Option<String> {
|
||||
match value.and_then(|v| v.value.as_ref())? {
|
||||
any_value::Value::StringValue(s) => Some(s.clone()),
|
||||
@@ -301,11 +298,9 @@ fn scalar_value_string(value: Option<&AnyValue>) -> Option<String> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Prometheus-style service identity of a resource. `job` is
|
||||
/// `<service.namespace>/<service.name>` when the namespace is present (per
|
||||
/// the OTel Prometheus compatibility spec), bare `service.name` otherwise,
|
||||
/// `None` when `service.name` is absent (no `unknown_service` fabrication).
|
||||
/// `instance` is `service.instance.id` verbatim.
|
||||
/// Prometheus-style `(job, instance)` identity. Per the OTel Prometheus
|
||||
/// compatibility spec `job` folds in `service.namespace` when present; a
|
||||
/// resource without `service.name` gets no job rather than a fabricated one.
|
||||
pub(crate) fn service_identity(attrs: &[KeyValue]) -> (Option<String>, Option<String>) {
|
||||
let mut name = None;
|
||||
let mut namespace = None;
|
||||
@@ -1082,41 +1077,14 @@ mod tests {
|
||||
process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
|
||||
assert_eq!(attr_value(&attrs, "job").as_deref(), Some("api"));
|
||||
|
||||
// no service.name: no job is fabricated, namespace stays a plain attr
|
||||
let mut attrs = vec![keyvalue("service.namespace", "shop")];
|
||||
process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
|
||||
assert_eq!(attr_value(&attrs, "job"), None);
|
||||
assert_eq!(
|
||||
attr_value(&attrs, "service.namespace").as_deref(),
|
||||
Some("shop")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_resource_attrs_remap_and_promotion() {
|
||||
// no service.name: no job is fabricated
|
||||
let mut attrs = vec![
|
||||
keyvalue("service.name", "api"),
|
||||
keyvalue("service.namespace", "shop"),
|
||||
keyvalue("service.instance.id", "inst-1"),
|
||||
keyvalue("host.id", "h-1"),
|
||||
];
|
||||
process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
|
||||
assert_eq!(attr_value(&attrs, "instance").as_deref(), Some("inst-1"));
|
||||
// raw identity attrs are in the default promote list and survive
|
||||
assert_eq!(attr_value(&attrs, "service.name").as_deref(), Some("api"));
|
||||
// non-promoted attrs are dropped
|
||||
assert_eq!(attr_value(&attrs, "host.id"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_legacy_mode_keeps_resource_attrs_untouched() {
|
||||
let ctx = OtlpMetricCtx {
|
||||
is_legacy: true,
|
||||
..Default::default()
|
||||
};
|
||||
let mut attrs = vec![keyvalue("service.name", "api"), keyvalue("host.id", "h-1")];
|
||||
process_resource_attrs(&mut attrs, &ctx);
|
||||
assert_eq!(attr_value(&attrs, "job"), None);
|
||||
assert_eq!(attr_value(&attrs, "host.id").as_deref(), Some("h-1"));
|
||||
assert_eq!(attr_value(&attrs, "instance").as_deref(), Some("inst-1"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -12,22 +12,13 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! OTel-native resource descriptor synthesized from OTLP metrics requests.
|
||||
//! Resource descriptor synthesized from OTLP metrics requests, so the entity
|
||||
//! graph reads one table instead of scanning every logical metric table for
|
||||
//! attributes the promote filter may have dropped.
|
||||
//!
|
||||
//! 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`.
|
||||
//! Columns are a fixed allowlist under the raw OTel attribute names: the
|
||||
//! conventions whitelist matches fixed names, so they must not follow the
|
||||
//! per-request label translation strategy or the promote/ignore headers.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
@@ -42,66 +33,73 @@ 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",
|
||||
];
|
||||
/// Attributes projected under their raw OTel keys. `service.instance.id` is
|
||||
/// absent on purpose: it lands in `instance`.
|
||||
///
|
||||
/// Matched instead of scanned: this runs for every attribute of every
|
||||
/// resource, and the compiler turns it into a length-and-prefix dispatch.
|
||||
fn is_projected_attr(key: &str) -> bool {
|
||||
matches!(
|
||||
key,
|
||||
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.
|
||||
/// Upper bound of [`is_projected_attr`] plus the derived `job`/`instance`,
|
||||
/// 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.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct ResourceInfoData {
|
||||
rows: BTreeMap<Vec<(String, String)>, i64>,
|
||||
}
|
||||
|
||||
impl ResourceInfoData {
|
||||
/// Projects one resource's raw (unfiltered) attributes; resources whose
|
||||
/// projection is empty contribute nothing.
|
||||
/// Takes the raw attributes, before the promote filter runs on them.
|
||||
pub fn observe(&mut self, raw_attrs: &[KeyValue], max_ts_nanos: i64) {
|
||||
let mut tags = BTreeMap::new();
|
||||
let mut tags = Vec::with_capacity(MAX_PROJECTED_TAGS);
|
||||
let (job, instance) = service_identity(raw_attrs);
|
||||
if let Some(job) = job {
|
||||
tags.insert(JOB_KEY.to_string(), job);
|
||||
tags.push((JOB_KEY.to_string(), job));
|
||||
}
|
||||
if let Some(instance) = instance {
|
||||
tags.insert(INSTANCE_KEY.to_string(), instance);
|
||||
tags.push((INSTANCE_KEY.to_string(), instance));
|
||||
}
|
||||
for kv in raw_attrs {
|
||||
if RESOURCE_INFO_ATTRS.contains(&kv.key.as_str())
|
||||
if is_projected_attr(&kv.key)
|
||||
&& let Some(value) = scalar_value_string(kv.value.as_ref())
|
||||
{
|
||||
tags.insert(kv.key.clone(), value);
|
||||
tags.push((kv.key.clone(), value));
|
||||
}
|
||||
}
|
||||
if tags.is_empty() {
|
||||
return;
|
||||
}
|
||||
// Sorted so equal attribute sets share a key, and so the emitted
|
||||
// columns keep a stable order.
|
||||
tags.sort_unstable();
|
||||
|
||||
let entry = self
|
||||
.rows
|
||||
.entry(tags.into_iter().collect())
|
||||
.or_insert(i64::MIN);
|
||||
*entry = (*entry).max(max_ts_nanos);
|
||||
self.rows
|
||||
.entry(tags)
|
||||
.and_modify(|ts| *ts = (*ts).max(max_ts_nanos))
|
||||
.or_insert(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.
|
||||
/// Every projected attribute becomes a tag, so auto-create puts it in the
|
||||
/// primary key where the conventions expect it.
|
||||
pub fn into_row_insert_requests(self) -> Result<Option<RowInsertRequests>> {
|
||||
if self.rows.is_empty() {
|
||||
return Ok(None);
|
||||
@@ -110,7 +108,7 @@ impl ResourceInfoData {
|
||||
let mut writer = MultiTableData::default();
|
||||
let table = writer.get_or_default_table_data(
|
||||
OTEL_RESOURCE_INFO_TABLE_NAME,
|
||||
RESOURCE_INFO_ATTRS.len() + 4,
|
||||
MAX_PROJECTED_TAGS + 2,
|
||||
self.rows.len(),
|
||||
);
|
||||
for (tags, ts_nanos) in self.rows {
|
||||
@@ -132,8 +130,8 @@ impl ResourceInfoData {
|
||||
}
|
||||
}
|
||||
|
||||
/// The newest data-point timestamp under a resource, the descriptor row's
|
||||
/// observation time. `None` when the resource carries no data points at all.
|
||||
/// 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)));
|
||||
@@ -191,7 +189,6 @@ mod tests {
|
||||
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();
|
||||
@@ -199,7 +196,6 @@ mod tests {
|
||||
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")
|
||||
@@ -207,14 +203,11 @@ mod tests {
|
||||
|
||||
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());
|
||||
// nothing allowlisted: no row at all, rather than an empty one
|
||||
let mut empty = ResourceInfoData::default();
|
||||
empty.observe(&[kv("os.type", "linux")], 100);
|
||||
assert!(empty.into_row_insert_requests().unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Reference in New Issue
Block a user