servers/otlp/metrics/
resource_info.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Resource descriptor synthesized from OTLP metrics requests, so the entity
16//! graph reads one table instead of scanning every logical metric table for
17//! attributes the promote filter may have dropped.
18//!
19//! Columns are a fixed allowlist under the raw OTel attribute names: the
20//! conventions whitelist matches fixed names, so they must not follow the
21//! per-request label translation strategy or the promote/ignore headers.
22
23use std::collections::BTreeMap;
24
25use api::v1::RowInsertRequests;
26use common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS;
27use common_grpc::precision::Precision;
28use common_query::prelude::{greptime_timestamp, greptime_value};
29use otel_arrow_rust::proto::opentelemetry::common::v1::KeyValue;
30
31use crate::error::Result;
32use crate::otlp::metrics::{
33    INSTANCE_KEY, JOB_KEY, ServiceIdentity, scalar_value_string, service_identity,
34};
35use crate::otlp::trace::{
36    KEY_CONTAINER_ID, KEY_CONTAINER_NAME, KEY_HOST_ID, KEY_HOST_NAME, KEY_K8S_CONTAINER_NAME,
37    KEY_K8S_NAMESPACE_NAME, KEY_K8S_NODE_NAME, KEY_K8S_POD_NAME, KEY_K8S_POD_UID, KEY_SERVICE_NAME,
38    KEY_SERVICE_NAMESPACE,
39};
40use crate::row_writer::{self, MultiTableData};
41
42/// Prefixed like the other engine-managed tables, so a user metric is
43/// unlikely to claim the name.
44pub const OTEL_RESOURCE_INFO_TABLE_NAME: &str = "greptime_otel_resource_info";
45
46/// Attributes projected under their raw OTel keys. `service.instance.id` is
47/// absent on purpose: it lands in `instance`.
48///
49/// Matched instead of scanned: this runs for every attribute of every
50/// resource, and the compiler turns it into a length-and-prefix dispatch.
51fn is_projected_attr(key: &str) -> bool {
52    matches!(
53        key,
54        KEY_SERVICE_NAME
55            | KEY_SERVICE_NAMESPACE
56            | KEY_HOST_ID
57            | KEY_HOST_NAME
58            | KEY_CONTAINER_ID
59            | KEY_CONTAINER_NAME
60            | KEY_K8S_POD_UID
61            | KEY_K8S_POD_NAME
62            | KEY_K8S_CONTAINER_NAME
63            | KEY_K8S_NAMESPACE_NAME
64            | KEY_K8S_NODE_NAME
65    )
66}
67
68/// Upper bound of [`is_projected_attr`] plus the derived `job`/`instance`,
69/// used to size the per-row buffers.
70const MAX_PROJECTED_TAGS: usize = 13;
71
72/// Projected attributes (sorted `(name, value)` pairs) -> graph window ->
73/// the newest data-point time seen in that window, which is what the row for
74/// that window is stamped with.
75///
76/// Windows are an inner map so the attributes are stored, and moved, once per
77/// resource. They are keyed separately because one request may carry data for
78/// several of them: a single row per resource would describe only the newest
79/// window, leaving the earlier ones with metric rows but no entities.
80#[derive(Debug, Default)]
81pub struct ResourceInfoData {
82    rows: BTreeMap<Vec<(String, String)>, BTreeMap<i64, i64>>,
83}
84
85impl ResourceInfoData {
86    pub(crate) fn is_empty(&self) -> bool {
87        self.rows.is_empty()
88    }
89
90    pub(super) fn observe_times(
91        &mut self,
92        raw_attrs: &[KeyValue],
93        times: impl IntoIterator<Item = i64>,
94    ) {
95        let mut tags = Vec::with_capacity(MAX_PROJECTED_TAGS);
96        let ServiceIdentity { job, instance } = service_identity(raw_attrs);
97        if let Some(job) = job {
98            tags.push((JOB_KEY.to_string(), job));
99        }
100        if let Some(instance) = instance {
101            tags.push((INSTANCE_KEY.to_string(), instance));
102        }
103        for kv in raw_attrs {
104            if is_projected_attr(&kv.key)
105                && let Some(value) = scalar_value_string(kv.value.as_ref())
106            {
107                tags.push((kv.key.clone(), value));
108            }
109        }
110        if tags.is_empty() {
111            return;
112        }
113        // Sorted so equal attribute sets share a key, and so the emitted
114        // columns keep a stable order.
115        tags.sort_unstable();
116
117        let mut observed: BTreeMap<i64, i64> = BTreeMap::new();
118        for ts in times {
119            let window = ts - ts.rem_euclid(SEMANTIC_GRAPH_WINDOW_NANOS);
120            observed
121                .entry(window)
122                .and_modify(|newest| *newest = (*newest).max(ts))
123                .or_insert(ts);
124        }
125        if observed.is_empty() {
126            return;
127        }
128
129        let windows = self.rows.entry(tags).or_default();
130        for (window, newest) in observed {
131            windows
132                .entry(window)
133                .and_modify(|seen| *seen = (*seen).max(newest))
134                .or_insert(newest);
135        }
136    }
137
138    /// Every projected attribute becomes a tag, so auto-create puts it in the
139    /// primary key where the conventions expect it.
140    pub fn into_row_insert_requests(self) -> Result<Option<RowInsertRequests>> {
141        if self.rows.is_empty() {
142            return Ok(None);
143        }
144
145        let mut writer = MultiTableData::default();
146        let table = writer.get_or_default_table_data(
147            OTEL_RESOURCE_INFO_TABLE_NAME,
148            MAX_PROJECTED_TAGS + 2,
149            self.rows.values().map(BTreeMap::len).sum(),
150        );
151        for (tags, windows) in &self.rows {
152            for ts_nanos in windows.values().copied() {
153                let mut row = table.alloc_one_row();
154                row_writer::write_tags(table, tags.iter().cloned(), &mut row)?;
155                row_writer::write_f64(table, greptime_value(), 1.0, &mut row)?;
156                row_writer::write_ts_to_millis(
157                    table,
158                    greptime_timestamp(),
159                    Some(ts_nanos),
160                    Precision::Nanosecond,
161                    &mut row,
162                )?;
163                table.add_row(row);
164            }
165        }
166
167        let (requests, _) = writer.into_row_insert_requests();
168        Ok(Some(requests))
169    }
170}
171
172#[cfg(test)]
173mod tests {
174    use api::v1::SemanticType;
175    use api::v1::value::ValueData;
176    use common_query::prelude::set_default_prefix;
177    use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, any_value};
178    use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
179        AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Metric,
180        ResourceMetrics, ScopeMetrics, metric,
181    };
182
183    use super::*;
184
185    mod delta;
186
187    fn kv(key: &str, value: &str) -> KeyValue {
188        KeyValue {
189            key: key.into(),
190            value: Some(AnyValue {
191                value: Some(any_value::Value::StringValue(value.into())),
192            }),
193        }
194    }
195
196    #[test]
197    fn observe_projects_allowlist_and_dedups_per_request() {
198        let mut data = ResourceInfoData::default();
199        let attrs = vec![
200            kv("service.name", "api"),
201            kv("service.namespace", "shop"),
202            kv("service.instance.id", "inst-1"),
203            kv("host.id", "h-1"),
204            kv("k8s.node.name", "node-a"),
205            kv("os.type", "linux"),
206        ];
207        data.observe_times(&attrs, [100, 50]);
208        assert_eq!(data.rows.len(), 1);
209        let (tags, windows) = data.rows.iter().next().unwrap();
210        assert_eq!(windows.values().copied().collect::<Vec<_>>(), vec![100]);
211        assert!(tags.contains(&("job".to_string(), "shop/api".to_string())));
212        assert!(tags.contains(&("instance".to_string(), "inst-1".to_string())));
213        assert!(tags.contains(&("service.name".to_string(), "api".to_string())));
214        assert!(tags.contains(&("k8s.node.name".to_string(), "node-a".to_string())));
215        assert!(
216            tags.iter()
217                .all(|(k, _)| k != "os.type" && k != "service.instance.id")
218        );
219
220        let mut reordered = attrs.clone();
221        reordered.reverse();
222        data.observe_times(&reordered, [200, 75]);
223        assert_eq!(data.rows.len(), 1);
224        assert_eq!(
225            data.rows.values().next().unwrap().values().next(),
226            Some(&200)
227        );
228
229        data.observe_times(&[kv("host.id", "h-2")], [10]);
230        assert_eq!(data.rows.len(), 2);
231
232        let mut empty = ResourceInfoData::default();
233        empty.observe_times(&[kv("os.type", "linux")], [100]);
234        assert!(empty.into_row_insert_requests().unwrap().is_none());
235    }
236
237    #[test]
238    fn observe_without_accepted_points_does_not_emit_rows() {
239        let mut data = ResourceInfoData::default();
240        data.observe_times(&[kv("service.name", "api")], []);
241        assert!(data.is_empty());
242        assert!(data.into_row_insert_requests().unwrap().is_none());
243    }
244
245    /// Earlier windows would keep their metric rows but lose their entities.
246    #[test]
247    fn observe_keeps_one_row_per_graph_window() {
248        let window = SEMANTIC_GRAPH_WINDOW_NANOS;
249        let mut data = ResourceInfoData::default();
250        data.observe_times(
251            &[kv("service.name", "api")],
252            [window + 1, window + 2, 3 * window + 7],
253        );
254
255        let windows = data.rows.values().next().unwrap();
256        assert_eq!(
257            windows.iter().collect::<Vec<_>>(),
258            vec![(&window, &(window + 2)), (&(3 * window), &(3 * window + 7))]
259        );
260    }
261
262    /// Describing a resource whose only data the encoder drops invents an
263    /// entity with no measurements.
264    #[test]
265    fn observe_ignores_data_the_encoder_drops() {
266        let exponential = |temporality: AggregationTemporality| ResourceMetrics {
267            scope_metrics: vec![ScopeMetrics {
268                metrics: vec![Metric {
269                    name: "latency".into(),
270                    data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram {
271                        data_points: vec![ExponentialHistogramDataPoint {
272                            time_unix_nano: 100,
273                            ..Default::default()
274                        }],
275                        aggregation_temporality: temporality as i32,
276                    })),
277                    ..Default::default()
278                }],
279                ..Default::default()
280            }],
281            ..Default::default()
282        };
283        for temporality in [
284            AggregationTemporality::Delta,
285            AggregationTemporality::Unspecified,
286        ] {
287            let mut resource = exponential(temporality);
288            resource.resource = Some(
289                otel_arrow_rust::proto::opentelemetry::resource::v1::Resource {
290                    attributes: vec![kv("service.name", "api")],
291                    ..Default::default()
292                },
293            );
294            let conversion = crate::otlp::metrics::to_grpc_insert_requests(
295                otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest { resource_metrics: vec![resource] },
296                &mut session::protocol_ctx::OtlpMetricCtx { resource_info: true, ..Default::default() },
297            ).unwrap();
298            assert_eq!(
299                conversion.resource_info.is_none(),
300                temporality == AggregationTemporality::Unspecified
301            );
302        }
303    }
304
305    #[test]
306    fn rows_carry_raw_key_tags_value_and_millis_timestamp() {
307        set_default_prefix(None).unwrap();
308        let mut data = ResourceInfoData::default();
309        data.observe_times(
310            &[kv("service.name", "api"), kv("host.id", "h-1")],
311            [1_700_000_000_123_456_789],
312        );
313        let requests = data.into_row_insert_requests().unwrap().unwrap();
314        assert_eq!(requests.inserts.len(), 1);
315        let insert = &requests.inserts[0];
316        assert_eq!(insert.table_name, OTEL_RESOURCE_INFO_TABLE_NAME);
317
318        let rows = insert.rows.as_ref().unwrap();
319        let names = rows
320            .schema
321            .iter()
322            .map(|c| c.column_name.as_str())
323            .collect::<Vec<_>>();
324        assert_eq!(
325            names,
326            vec![
327                "host.id",
328                "job",
329                "service.name",
330                greptime_value(),
331                greptime_timestamp()
332            ]
333        );
334        for column in &rows.schema[..3] {
335            assert_eq!(column.semantic_type, SemanticType::Tag as i32);
336        }
337
338        assert_eq!(rows.rows.len(), 1);
339        let values = &rows.rows[0].values;
340        assert_eq!(values[3].value_data, Some(ValueData::F64Value(1.0)));
341        assert_eq!(
342            values[4].value_data,
343            Some(ValueData::TimestampMillisecondValue(1_700_000_000_123))
344        );
345    }
346}
servers/otlp/metrics/resource_info.rs

servers/otlp/metrics/
resource_info.rs
