servers/otlp/metrics/
exemplars.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
15use std::collections::{BTreeMap, HashSet};
16use std::sync::Arc;
17
18use api::v1::{ColumnSchema, SemanticType, Value, value::ValueData};
19use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar};
20use session::protocol_ctx::OtlpMetricCtx;
21use snafu::{OptionExt, ensure};
22
23use crate::error::{InvalidOtlpMetricInputSnafu, Result};
24use crate::exemplar::{self as normalized, Exemplar as Observation};
25use crate::otlp::metrics::{scalar_value_string, translate_label_name};
26
27/// Observations whose inherited labels are bound after source-row normalization.
28pub(crate) struct RowExemplars {
29    pub(crate) metric: Arc<str>,
30    /// Starts as an emitted table's row offset; preparation maps it into `row_locations`.
31    pub(crate) row: usize,
32    observations: Vec<Observation>,
33}
34
35impl RowExemplars {
36    /// Drops observations whose inherited tags cannot be represented by the exemplar schema.
37    pub(crate) fn finish(mut self, schema: &[ColumnSchema], row: &[Value]) -> Vec<Observation> {
38        let labels = normalized::labels(
39            schema.iter().zip(row).filter_map(|(column, value)| {
40                if column.semantic_type != SemanticType::Tag as i32 {
41                    return None;
42                }
43                match &value.value_data {
44                    Some(ValueData::StringValue(value)) => {
45                        Some((column.column_name.clone(), value.clone()))
46                    }
47                    _ => None,
48                }
49            }),
50            false,
51            |name| !name.is_empty(),
52        );
53        let Ok(labels) = labels else {
54            crate::metrics::EXEMPLARS_DROPPED_TOTAL
55                .with_label_values(&["otlp"])
56                .inc_by(self.observations.len() as u64);
57            return Vec::new();
58        };
59        let labels = Arc::new(labels);
60        for observation in &mut self.observations {
61            observation.series_labels = labels.clone();
62        }
63        self.observations
64    }
65}
66
67pub(super) fn value(exemplar: &Exemplar) -> Result<f64> {
68    match exemplar
69        .value
70        .as_ref()
71        .context(InvalidOtlpMetricInputSnafu {
72            reason: "exemplar is missing its value",
73        })? {
74        exemplar::Value::AsDouble(value) => Ok(*value),
75        exemplar::Value::AsInt(value) => Ok(*value as f64),
76    }
77}
78
79/// Collects exemplars for metric-engine ingestion, retaining each emitted row's identity
80/// until destination defaults have been filled.
81pub(crate) fn append<'a>(
82    output: &mut Vec<RowExemplars>,
83    exemplars: impl IntoIterator<Item = &'a Exemplar>,
84    metric: &str,
85    row: usize,
86    ctx: &OtlpMetricCtx,
87) {
88    let mut exemplars = exemplars.into_iter().peekable();
89    if exemplars.peek().is_none() {
90        return;
91    }
92    if !ctx.with_metric_engine {
93        crate::metrics::EXEMPLARS_DROPPED_TOTAL
94            .with_label_values(&["otlp"])
95            .inc_by(exemplars.count() as u64);
96        return;
97    }
98    let validate_name = |name: &[u8]| !name.is_empty();
99    let series_labels = Arc::new(BTreeMap::new());
100    let metric: Arc<str> = metric.into();
101    let mut pending = RowExemplars {
102        metric: metric.clone(),
103        row,
104        observations: Vec::new(),
105    };
106    for exemplar in exemplars {
107        let observation: Result<_> = (|| {
108            let mut names = HashSet::new();
109            let mut attributes = Vec::new();
110            for attribute in &exemplar.filtered_attributes {
111                ensure!(
112                    names.insert(&attribute.key),
113                    InvalidOtlpMetricInputSnafu {
114                        reason: format!("duplicate filtered exemplar attribute: {}", attribute.key)
115                    }
116                );
117                if let Some(value) = scalar_value_string(attribute.value.as_ref()) {
118                    let name =
119                        translate_label_name(&attribute.key, ctx.metric_translation_strategy);
120                    attributes.push((name, value));
121                }
122            }
123            let mut labels = normalized::labels(attributes, true, validate_name)?;
124            for (name, bytes, length) in [
125                ("trace_id", &exemplar.trace_id, 16),
126                ("span_id", &exemplar.span_id, 8),
127            ] {
128                if !bytes.is_empty() {
129                    ensure!(
130                        bytes.len() == length,
131                        InvalidOtlpMetricInputSnafu {
132                            reason: format!("exemplar {name} must contain {length} bytes")
133                        }
134                    );
135                    labels.insert(name.to_string(), hex::encode(bytes));
136                }
137            }
138            let labels = normalized::labels(labels, true, validate_name)?;
139            let timestamp = i64::try_from(exemplar.time_unix_nano).map_err(|_| {
140                InvalidOtlpMetricInputSnafu {
141                    reason: "exemplar timestamp exceeds signed nanosecond range",
142                }
143                .build()
144            })?;
145            Ok(Observation {
146                metric: metric.clone(),
147                series_labels: series_labels.clone(),
148                timestamp,
149                value: value(exemplar)?,
150                labels,
151            })
152        })();
153        match observation {
154            Ok(observation) => pending.observations.push(observation),
155            Err(_) => crate::metrics::EXEMPLARS_DROPPED_TOTAL
156                .with_label_values(&["otlp"])
157                .inc(),
158        }
159    }
160    if !pending.observations.is_empty() {
161        output.push(pending);
162    }
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168    use crate::otlp::metrics::string_key_value;
169    use crate::row_writer;
170    use crate::row_writer::TableData;
171
172    #[test]
173    fn test_exemplars_require_metric_engine() {
174        let exemplar = Exemplar {
175            time_unix_nano: 1,
176            value: Some(exemplar::Value::AsDouble(0.5)),
177            ..Default::default()
178        };
179        let dropped = crate::metrics::EXEMPLARS_DROPPED_TOTAL.with_label_values(&["otlp"]);
180        for enabled in [false, true] {
181            let before = dropped.get();
182            let mut output = Vec::new();
183            append(
184                &mut output,
185                [&exemplar],
186                "latency",
187                0,
188                &OtlpMetricCtx {
189                    with_metric_engine: enabled,
190                    ..Default::default()
191                },
192            );
193            assert_eq!(output.len(), usize::from(enabled));
194            assert_eq!(dropped.get() - before, u64::from(!enabled));
195        }
196    }
197
198    #[test]
199    fn test_exemplar_labels_follow_otlp_translation_strategy() {
200        use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
201            Gauge, NumberDataPoint, number_data_point,
202        };
203        use session::protocol_ctx::OtlpMetricTranslationStrategy;
204
205        use crate::otlp::metrics::encode_gauge;
206        use crate::row_writer::MultiTableData;
207
208        for strategy in [
209            OtlpMetricTranslationStrategy::NoTranslation,
210            OtlpMetricTranslationStrategy::NoUtf8EscapingWithSuffixes,
211            OtlpMetricTranslationStrategy::UnderscoreEscapingWithSuffixes,
212            OtlpMetricTranslationStrategy::UnderscoreEscapingWithoutSuffixes,
213        ] {
214            for name in ["request.method", "请求"] {
215                let gauge = Gauge {
216                    data_points: vec![NumberDataPoint {
217                        attributes: vec![string_key_value(name, "GET".into())],
218                        time_unix_nano: 1,
219                        value: Some(number_data_point::Value::AsDouble(1.0)),
220                        exemplars: vec![Exemplar {
221                            filtered_attributes: vec![string_key_value(name, "POST".into())],
222                            time_unix_nano: 1,
223                            value: Some(exemplar::Value::AsDouble(0.5)),
224                            ..Default::default()
225                        }],
226                        ..Default::default()
227                    }],
228                };
229                let mut tables = MultiTableData::default();
230                let mut output = Vec::new();
231                encode_gauge(
232                    &mut tables,
233                    "requests",
234                    &gauge,
235                    None,
236                    None,
237                    &OtlpMetricCtx {
238                        with_metric_engine: true,
239                        metric_translation_strategy: strategy,
240                        ..Default::default()
241                    },
242                    &mut output,
243                )
244                .unwrap();
245                let (requests, rows) = tables.into_row_insert_requests();
246                assert_eq!(rows, 1);
247                let rows = requests.inserts[0].rows.as_ref().unwrap();
248                let output = output
249                    .into_iter()
250                    .flat_map(|pending| {
251                        let row = pending.row;
252                        pending.finish(&rows.schema, &rows.rows[row].values)
253                    })
254                    .collect::<Vec<_>>();
255                let name = translate_label_name(name, strategy);
256                assert_eq!(output.len(), 1);
257                assert_eq!(output[0].series_labels[&name], "GET");
258                assert_eq!(output[0].labels[&name], "POST");
259                assert!(
260                    requests.inserts[0]
261                        .rows
262                        .as_ref()
263                        .unwrap()
264                        .schema
265                        .iter()
266                        .any(|column| column.column_name == name)
267                );
268            }
269        }
270    }
271
272    #[test]
273    fn own_time_labels_and_dedicated_id_precedence() {
274        let mut table = TableData::new(2, 1);
275        let mut row = table.alloc_one_row();
276        row_writer::write_tag(&mut table, "job", "api", &mut row).unwrap();
277        let exemplar = Exemplar {
278            filtered_attributes: vec![
279                string_key_value("trace_id", "fallback".into()),
280                string_key_value("span_id", "fallback_span".into()),
281                string_key_value("request.id", "r1".into()),
282            ],
283            time_unix_nano: 1720000000123456789,
284            trace_id: vec![0xab; 16],
285            span_id: vec![0xcd; 8],
286            value: Some(exemplar::Value::AsDouble(0.32)),
287        };
288        let ctx = OtlpMetricCtx {
289            with_metric_engine: true,
290            ..Default::default()
291        };
292        let mut output = Vec::new();
293        append(&mut output, [&exemplar], "latency", table.num_rows(), &ctx);
294        let first = output.pop().unwrap().finish(table.columns(), &row);
295        assert_eq!(first[0].timestamp, 1720000000123456789);
296        assert_eq!(first[0].series_labels["job"], "api");
297        assert_eq!(first[0].labels["trace_id"], "ab".repeat(16));
298        assert_eq!(first[0].labels["span_id"], "cd".repeat(8));
299        let mut without_id = exemplar.clone();
300        without_id.trace_id.clear();
301        append(
302            &mut output,
303            [&without_id],
304            "latency",
305            table.num_rows(),
306            &ctx,
307        );
308        let second = output.pop().unwrap().finish(table.columns(), &row);
309        assert_eq!(second[0].labels["trace_id"], "fallback");
310    }
311}
servers/otlp/metrics/exemplars.rs

servers/otlp/metrics/
exemplars.rs
