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;
30use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
31    AggregationTemporality, ResourceMetrics, metric,
32};
33use session::protocol_ctx::OtlpMetricCtx;
34
35use crate::error::Result;
36use crate::otlp::metrics::{
37    INSTANCE_KEY, JOB_KEY, ServiceIdentity, exponential_histogram_gate,
38    exponential_histogram_value, histogram_data_point_rejection, scalar_value_string,
39    service_identity,
40};
41use crate::otlp::trace::{
42    KEY_CONTAINER_ID, KEY_CONTAINER_NAME, KEY_HOST_ID, KEY_HOST_NAME, KEY_K8S_CONTAINER_NAME,
43    KEY_K8S_NAMESPACE_NAME, KEY_K8S_NODE_NAME, KEY_K8S_POD_NAME, KEY_K8S_POD_UID, KEY_SERVICE_NAME,
44    KEY_SERVICE_NAMESPACE,
45};
46use crate::row_writer::{self, MultiTableData};
47
48/// Prefixed like the other engine-managed tables, so a user metric is
49/// unlikely to claim the name.
50pub const OTEL_RESOURCE_INFO_TABLE_NAME: &str = "greptime_otel_resource_info";
51
52/// Attributes projected under their raw OTel keys. `service.instance.id` is
53/// absent on purpose: it lands in `instance`.
54///
55/// Matched instead of scanned: this runs for every attribute of every
56/// resource, and the compiler turns it into a length-and-prefix dispatch.
57fn is_projected_attr(key: &str) -> bool {
58    matches!(
59        key,
60        KEY_SERVICE_NAME
61            | KEY_SERVICE_NAMESPACE
62            | KEY_HOST_ID
63            | KEY_HOST_NAME
64            | KEY_CONTAINER_ID
65            | KEY_CONTAINER_NAME
66            | KEY_K8S_POD_UID
67            | KEY_K8S_POD_NAME
68            | KEY_K8S_CONTAINER_NAME
69            | KEY_K8S_NAMESPACE_NAME
70            | KEY_K8S_NODE_NAME
71    )
72}
73
74/// Upper bound of [`is_projected_attr`] plus the derived `job`/`instance`,
75/// used to size the per-row buffers.
76const MAX_PROJECTED_TAGS: usize = 13;
77
78/// Projected attributes (sorted `(name, value)` pairs) -> graph window ->
79/// the newest data-point time seen in that window, which is what the row for
80/// that window is stamped with.
81///
82/// Windows are an inner map so the attributes are stored, and moved, once per
83/// resource. They are keyed separately because one request may carry data for
84/// several of them: a single row per resource would describe only the newest
85/// window, leaving the earlier ones with metric rows but no entities.
86#[derive(Debug, Default)]
87pub struct ResourceInfoData {
88    rows: BTreeMap<Vec<(String, String)>, BTreeMap<i64, i64>>,
89}
90
91impl ResourceInfoData {
92    /// Takes the raw attributes, before the promote filter runs on them.
93    pub fn observe(
94        &mut self,
95        raw_attrs: &[KeyValue],
96        resource: &ResourceMetrics,
97        metric_ctx: &OtlpMetricCtx,
98    ) {
99        let mut tags = Vec::with_capacity(MAX_PROJECTED_TAGS);
100        let ServiceIdentity { job, instance } = service_identity(raw_attrs);
101        if let Some(job) = job {
102            tags.push((JOB_KEY.to_string(), job));
103        }
104        if let Some(instance) = instance {
105            tags.push((INSTANCE_KEY.to_string(), instance));
106        }
107        for kv in raw_attrs {
108            if is_projected_attr(&kv.key)
109                && let Some(value) = scalar_value_string(kv.value.as_ref())
110            {
111                tags.push((kv.key.clone(), value));
112            }
113        }
114        if tags.is_empty() {
115            return;
116        }
117        // Sorted so equal attribute sets share a key, and so the emitted
118        // columns keep a stable order.
119        tags.sort_unstable();
120
121        let mut observed: BTreeMap<i64, i64> = BTreeMap::new();
122        for_each_encoded_time(resource, metric_ctx, |ts| {
123            let window = ts - ts.rem_euclid(SEMANTIC_GRAPH_WINDOW_NANOS);
124            observed
125                .entry(window)
126                .and_modify(|newest| *newest = (*newest).max(ts))
127                .or_insert(ts);
128        });
129        if observed.is_empty() {
130            return;
131        }
132
133        let windows = self.rows.entry(tags).or_default();
134        for (window, newest) in observed {
135            windows
136                .entry(window)
137                .and_modify(|seen| *seen = (*seen).max(newest))
138                .or_insert(newest);
139        }
140    }
141
142    /// Every projected attribute becomes a tag, so auto-create puts it in the
143    /// primary key where the conventions expect it.
144    pub fn into_row_insert_requests(self) -> Result<Option<RowInsertRequests>> {
145        if self.rows.is_empty() {
146            return Ok(None);
147        }
148
149        let mut writer = MultiTableData::default();
150        let table = writer.get_or_default_table_data(
151            OTEL_RESOURCE_INFO_TABLE_NAME,
152            MAX_PROJECTED_TAGS + 2,
153            self.rows.values().map(BTreeMap::len).sum(),
154        );
155        for (tags, windows) in &self.rows {
156            for ts_nanos in windows.values().copied() {
157                let mut row = table.alloc_one_row();
158                row_writer::write_tags(table, tags.iter().cloned(), &mut row)?;
159                row_writer::write_f64(table, greptime_value(), 1.0, &mut row)?;
160                row_writer::write_ts_to_millis(
161                    table,
162                    greptime_timestamp(),
163                    Some(ts_nanos),
164                    Precision::Nanosecond,
165                    &mut row,
166                )?;
167                table.add_row(row);
168            }
169        }
170
171        let (requests, _) = writer.into_row_insert_requests();
172        Ok(Some(requests))
173    }
174}
175
176/// Visits the times of the data points the encoder writes rows for, so a
177/// resource is described exactly where it is measured rather than wherever
178/// its request happens to reach.
179fn for_each_encoded_time(
180    resource: &ResourceMetrics,
181    metric_ctx: &OtlpMetricCtx,
182    mut visit: impl FnMut(i64),
183) {
184    fn visit_all(points: impl Iterator<Item = u64>, visit: &mut impl FnMut(i64)) {
185        for ts in points {
186            visit(ts as i64);
187        }
188    }
189    for scope in &resource.scope_metrics {
190        for m in &scope.metrics {
191            match &m.data {
192                Some(metric::Data::Gauge(g)) => {
193                    visit_all(g.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
194                }
195                Some(metric::Data::Sum(s)) => {
196                    visit_all(s.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
197                }
198                Some(metric::Data::Histogram(h)) => {
199                    let is_delta = matches!(
200                        AggregationTemporality::try_from(h.aggregation_temporality),
201                        Ok(AggregationTemporality::Delta)
202                    );
203                    visit_all(
204                        h.data_points
205                            .iter()
206                            .filter(|point| {
207                                histogram_data_point_rejection(point, is_delta).is_none()
208                            })
209                            .map(|point| point.time_unix_nano),
210                        &mut visit,
211                    )
212                }
213                Some(metric::Data::Summary(s)) => {
214                    visit_all(s.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
215                }
216                Some(metric::Data::ExponentialHistogram(h))
217                    if exponential_histogram_gate(h, metric_ctx).is_ok() =>
218                {
219                    for point in &h.data_points {
220                        if let Ok((_, ts)) = exponential_histogram_value(point) {
221                            visit(ts);
222                        }
223                    }
224                }
225                Some(metric::Data::ExponentialHistogram(_)) => {}
226                None => {}
227            }
228        }
229    }
230}
231
232#[cfg(test)]
233mod tests {
234    use api::v1::SemanticType;
235    use api::v1::value::ValueData;
236    use common_query::prelude::set_default_prefix;
237    use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, any_value};
238    use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
239        AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Metric,
240        NumberDataPoint, ScopeMetrics,
241    };
242
243    use super::*;
244
245    mod delta;
246
247    fn kv(key: &str, value: &str) -> KeyValue {
248        KeyValue {
249            key: key.into(),
250            value: Some(AnyValue {
251                value: Some(any_value::Value::StringValue(value.into())),
252            }),
253        }
254    }
255
256    fn gauge_at(times: &[i64]) -> ResourceMetrics {
257        ResourceMetrics {
258            scope_metrics: vec![ScopeMetrics {
259                metrics: vec![Metric {
260                    data: Some(metric::Data::Gauge(Gauge {
261                        data_points: times
262                            .iter()
263                            .map(|ts| NumberDataPoint {
264                                time_unix_nano: *ts as u64,
265                                ..Default::default()
266                            })
267                            .collect(),
268                    })),
269                    ..Default::default()
270                }],
271                ..Default::default()
272            }],
273            ..Default::default()
274        }
275    }
276
277    #[test]
278    fn observe_projects_allowlist_and_dedups_per_request() {
279        let mut data = ResourceInfoData::default();
280        let attrs = vec![
281            kv("service.name", "api"),
282            kv("service.namespace", "shop"),
283            kv("service.instance.id", "inst-1"),
284            kv("host.id", "h-1"),
285            kv("k8s.node.name", "node-a"),
286            kv("os.type", "linux"),
287        ];
288        data.observe(&attrs, &gauge_at(&[100, 50]), &OtlpMetricCtx::default());
289        assert_eq!(data.rows.len(), 1);
290        let (tags, windows) = data.rows.iter().next().unwrap();
291        assert_eq!(windows.values().copied().collect::<Vec<_>>(), vec![100]);
292        assert!(tags.contains(&("job".to_string(), "shop/api".to_string())));
293        assert!(tags.contains(&("instance".to_string(), "inst-1".to_string())));
294        assert!(tags.contains(&("service.name".to_string(), "api".to_string())));
295        assert!(tags.contains(&("k8s.node.name".to_string(), "node-a".to_string())));
296        assert!(
297            tags.iter()
298                .all(|(k, _)| k != "os.type" && k != "service.instance.id")
299        );
300
301        data.observe(
302            &[kv("host.id", "h-2")],
303            &gauge_at(&[10]),
304            &OtlpMetricCtx::default(),
305        );
306        assert_eq!(data.rows.len(), 2);
307
308        let mut empty = ResourceInfoData::default();
309        empty.observe(
310            &[kv("os.type", "linux")],
311            &gauge_at(&[100]),
312            &OtlpMetricCtx::default(),
313        );
314        assert!(empty.into_row_insert_requests().unwrap().is_none());
315    }
316
317    /// Earlier windows would keep their metric rows but lose their entities.
318    #[test]
319    fn observe_keeps_one_row_per_graph_window() {
320        let window = SEMANTIC_GRAPH_WINDOW_NANOS;
321        let mut data = ResourceInfoData::default();
322        data.observe(
323            &[kv("service.name", "api")],
324            &gauge_at(&[window + 1, window + 2, 3 * window + 7]),
325            &OtlpMetricCtx::default(),
326        );
327
328        let windows = data.rows.values().next().unwrap();
329        assert_eq!(
330            windows.iter().collect::<Vec<_>>(),
331            vec![(&window, &(window + 2)), (&(3 * window), &(3 * window + 7))]
332        );
333    }
334
335    /// Describing a resource whose only data the encoder drops invents an
336    /// entity with no measurements.
337    #[test]
338    fn observe_ignores_data_the_encoder_drops() {
339        let exponential = |temporality: AggregationTemporality| ResourceMetrics {
340            scope_metrics: vec![ScopeMetrics {
341                metrics: vec![Metric {
342                    data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram {
343                        data_points: vec![ExponentialHistogramDataPoint {
344                            time_unix_nano: 100,
345                            ..Default::default()
346                        }],
347                        aggregation_temporality: temporality as i32,
348                    })),
349                    ..Default::default()
350                }],
351                ..Default::default()
352            }],
353            ..Default::default()
354        };
355        let enabled = OtlpMetricCtx {
356            experimental_enable_exponential_histogram: true,
357            ..Default::default()
358        };
359
360        for (resource, ctx) in [
361            (
362                exponential(AggregationTemporality::Cumulative),
363                OtlpMetricCtx::default(),
364            ),
365            (exponential(AggregationTemporality::Delta), enabled),
366        ] {
367            let mut data = ResourceInfoData::default();
368            data.observe(&[kv("service.name", "api")], &resource, &ctx);
369            assert!(data.into_row_insert_requests().unwrap().is_none());
370        }
371    }
372
373    #[test]
374    fn rows_carry_raw_key_tags_value_and_millis_timestamp() {
375        set_default_prefix(None).unwrap();
376        let mut data = ResourceInfoData::default();
377        data.observe(
378            &[kv("service.name", "api"), kv("host.id", "h-1")],
379            &gauge_at(&[1_700_000_000_123_456_789]),
380            &OtlpMetricCtx::default(),
381        );
382        let requests = data.into_row_insert_requests().unwrap().unwrap();
383        assert_eq!(requests.inserts.len(), 1);
384        let insert = &requests.inserts[0];
385        assert_eq!(insert.table_name, OTEL_RESOURCE_INFO_TABLE_NAME);
386
387        let rows = insert.rows.as_ref().unwrap();
388        let names = rows
389            .schema
390            .iter()
391            .map(|c| c.column_name.as_str())
392            .collect::<Vec<_>>();
393        assert_eq!(
394            names,
395            vec![
396                "host.id",
397                "job",
398                "service.name",
399                greptime_value(),
400                greptime_timestamp()
401            ]
402        );
403        for column in &rows.schema[..3] {
404            assert_eq!(column.semantic_type, SemanticType::Tag as i32);
405        }
406
407        assert_eq!(rows.rows.len(), 1);
408        let values = &rows.rows[0].values;
409        assert_eq!(values[3].value_data, Some(ValueData::F64Value(1.0)));
410        assert_eq!(
411            values[4].value_data,
412            Some(ValueData::TimestampMillisecondValue(1_700_000_000_123))
413        );
414    }
415}
servers/otlp/metrics/resource_info.rs

servers/otlp/metrics/
resource_info.rs
