servers/otlp/metrics/
prepare.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::{HashMap, HashSet};
16use std::hash::{BuildHasher, Hash, Hasher};
17use std::ops::Range;
18
19use ahash::{HashMap as AHashMap, HashMapExt};
20use api::v1::{ColumnSchema, RowInsertRequest, RowInsertRequests, Rows, SemanticType};
21use common_base::regex_pattern::NAME_PATTERN_REG;
22use common_query::native_histogram::native_histogram_value_type;
23use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
24use otel_arrow_rust::proto::opentelemetry::common::v1::{KeyValue, any_value};
25use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Metric, NumberDataPoint, metric};
26use session::protocol_ctx::OtlpMetricCtx;
27use snafu::{OptionExt, ensure};
28use table::requests::{
29    METADATA_QUALITY_DECLARED, SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_METRIC_TYPE,
30};
31
32use crate::error::{self, Result};
33use crate::metrics::OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS;
34use crate::otlp::metrics::exemplars::RowExemplars;
35use crate::otlp::metrics::{
36    MetricMetadata, MetricsConversion, OTEL_RESOURCE_INFO_TABLE_NAME, ResourceInfoData,
37    add_accepted_data_points, encode_metrics, from_metric_type, has_no_recorded_value,
38    process_resource_attrs, process_scope_attrs, reject_data_points, validate_sample_kind,
39};
40use crate::query_handler::MetricsIngestOutcome;
41use crate::row_writer::{MultiTableData, TableData};
42use crate::semantic::SemanticIndex;
43
44/// Rows share exact input schemas while retaining their source-point boundaries.
45pub struct PreparedMetrics {
46    /// Destination/schema batches. Normalization must preserve batch and row order/counts.
47    pub requests: Vec<RowInsertRequest>,
48    points: Vec<PreparedPoint>,
49    row_locations: Vec<(usize, usize)>,
50    metrics: Vec<MetricMetadata>,
51    resources: Vec<Vec<KeyValue>>,
52    outcome: MetricsIngestOutcome,
53}
54
55struct PreparedPoint {
56    metric: usize,
57    resource: usize,
58    timestamp: i64,
59    rows: Range<usize>,
60    exemplars: Vec<RowExemplars>,
61}
62
63#[derive(Default)]
64struct MetricWriter {
65    tables: MultiTableData,
66    // TableData schemas only grow. A new column invalidates the cached batch.
67    batches: AHashMap<String, (usize, usize)>,
68}
69
70pub fn prepare_metrics(
71    request: ExportMetricsServiceRequest,
72    metric_ctx: &mut OtlpMetricCtx,
73) -> Result<PreparedMetrics> {
74    let mut prepared = PreparedMetrics {
75        requests: Vec::new(),
76        points: Vec::new(),
77        row_locations: Vec::new(),
78        metrics: Vec::new(),
79        resources: Vec::new(),
80        outcome: MetricsIngestOutcome::default(),
81    };
82    let mut schema_batches = AHashMap::<u64, Vec<usize>>::new();
83    for resource in request.resource_metrics {
84        let raw_attrs = resource.resource.map(|r| r.attributes).unwrap_or_default();
85        let mut attrs = raw_attrs.clone();
86        process_resource_attrs(&mut attrs, metric_ctx);
87        let resource_index = prepared.resources.len();
88        prepared.resources.push(if metric_ctx.resource_info {
89            raw_attrs
90        } else {
91            Vec::new()
92        });
93        for scope in resource.scope_metrics {
94            let scope_attrs = process_scope_attrs(&scope, metric_ctx);
95            for mut metric in scope.metrics {
96                let Some(data) = metric.data.as_ref() else {
97                    continue;
98                };
99                metric_ctx.set_metric_type(from_metric_type(data));
100                let metric_index = prepared.metrics.len();
101                prepared
102                    .metrics
103                    .push(MetricMetadata::new(&metric, metric_ctx)?);
104                let mut writer = MetricWriter::default();
105                let mut layout: (Vec<String>, bool) = (Vec::new(), false);
106                let mut writers = AHashMap::new();
107                // Move each payload once; never clone the accumulated request.
108                macro_rules! prepare_points {
109                    ($data:ident, $variant:ident, $omits_value:expr) => {
110                        for point in std::mem::take(&mut $data.data_points) {
111                            // Omissions must not become explicit nulls in a retained schema.
112                            let omits_value = ($omits_value)(&point);
113                            if omits_value != layout.1
114                                || !scalar_attribute_names(&point.attributes)
115                                    .eq(layout.0.iter().map(String::as_str))
116                            {
117                                let next_layout = (
118                                    scalar_attribute_names(&point.attributes)
119                                        .map(str::to_owned)
120                                        .collect(),
121                                    omits_value,
122                                );
123                                let next_writer = writers.remove(&next_layout).unwrap_or_default();
124                                let previous = std::mem::replace(&mut writer, next_writer);
125                                let previous_layout = std::mem::replace(&mut layout, next_layout);
126                                if !previous.batches.is_empty() {
127                                    writers.insert(previous_layout, previous);
128                                }
129                            }
130                            let timestamp = point.time_unix_nano;
131                            let start = point.start_time_unix_nano;
132                            $data.data_points.push(point);
133                            metric.data = Some(metric::Data::$variant($data));
134                            if !prepared.push_point(
135                                &metric,
136                                metric_index,
137                                timestamp,
138                                start,
139                                &attrs,
140                                scope_attrs.as_ref(),
141                                resource_index,
142                                metric_ctx,
143                                &mut writer,
144                                &mut schema_batches,
145                            )? {
146                                writer = MetricWriter::default();
147                            }
148                            let Some(metric::Data::$variant(mut remaining)) = metric.data.take()
149                            else {
150                                unreachable!()
151                            };
152                            remaining.data_points.clear();
153                            $data = remaining;
154                        }
155                    };
156                }
157                let Some(data) = metric.data.take() else {
158                    continue;
159                };
160                match data {
161                    metric::Data::Gauge(mut data) => {
162                        prepare_points!(data, Gauge, |point: &NumberDataPoint| point
163                            .value
164                            .is_none())
165                    }
166                    metric::Data::Sum(mut data) => {
167                        prepare_points!(data, Sum, |point: &NumberDataPoint| point.value.is_none()
168                            && !has_no_recorded_value(point.flags))
169                    }
170                    metric::Data::Histogram(mut data) => {
171                        prepare_points!(data, Histogram, |_| false)
172                    }
173                    metric::Data::Summary(mut data) => prepare_points!(data, Summary, |_| false),
174                    metric::Data::ExponentialHistogram(mut data) => {
175                        prepare_points!(data, ExponentialHistogram, |_| false)
176                    }
177                }
178            }
179        }
180    }
181    Ok(prepared)
182}
183
184fn scalar_attribute_names(attrs: &[KeyValue]) -> impl Iterator<Item = &str> {
185    attrs.iter().filter_map(|attr| {
186        matches!(
187            attr.value.as_ref()?.value.as_ref()?,
188            any_value::Value::StringValue(_)
189                | any_value::Value::BoolValue(_)
190                | any_value::Value::IntValue(_)
191                | any_value::Value::DoubleValue(_)
192        )
193        .then_some(attr.key.as_str())
194    })
195}
196
197impl PreparedMetrics {
198    #[allow(clippy::too_many_arguments)]
199    fn push_point(
200        &mut self,
201        metric: &Metric,
202        metric_index: usize,
203        timestamp: u64,
204        start: u64,
205        resource_attrs: &Vec<KeyValue>,
206        scope_attrs: Option<&Vec<KeyValue>>,
207        resource: usize,
208        ctx: &OtlpMetricCtx,
209        writer: &mut MetricWriter,
210        schema_batches: &mut AHashMap<u64, Vec<usize>>,
211    ) -> Result<bool> {
212        let metadata = &self.metrics[metric_index];
213        let exponential = metadata.native_schema.is_some();
214        // OTLP gauges ignore StartTimeUnixNano.
215        let invalid_interval =
216            !matches!(metric.data, Some(metric::Data::Gauge(_))) && start > timestamp;
217        let mut outcome = MetricsIngestOutcome::default();
218        let mut exemplars = Vec::new();
219        let encoded = if !exponential && (timestamp > i64::MAX as u64 || invalid_interval) {
220            Err(error::InvalidOtlpMetricInputSnafu {
221                reason: "invalid OTLP data-point timestamp or interval",
222            }
223            .build())
224        } else {
225            encode_metrics(
226                &mut writer.tables,
227                metric,
228                metadata,
229                Some(resource_attrs),
230                scope_attrs,
231                ctx,
232                &mut outcome,
233                &mut exemplars,
234            )
235        };
236        if let Err(err) = encoded {
237            match err {
238                error::Error::InvalidOtlpMetricInput { .. }
239                | error::Error::IncompatibleSchema { .. }
240                | error::Error::TimestampOverflow { .. }
241                | error::Error::TimePrecision { .. }
242                | error::Error::InvalidParameter { .. } => {
243                    self.reject(exponential, err.to_string())?;
244                    return Ok(false);
245                }
246                _ => return Err(err),
247            }
248        }
249        if outcome.rejected_data_points != 0 {
250            // Payload encoders already counted histogram-specific rejections.
251            reject_data_points(&mut self.outcome, 1, || {
252                outcome.error_message.unwrap_or_default()
253            })?;
254            return Ok(false);
255        }
256        if let Some((name, _)) = writer.tables.iter().find(|(name, table)| {
257            table.num_rows() != 0
258                && !writer.batches.contains_key(*name)
259                && !NAME_PATTERN_REG.is_match(name)
260        }) {
261            self.reject(exponential, format!("invalid output table name `{name}`"))?;
262            return Ok(false);
263        }
264        let first_row = self.row_locations.len();
265        for (name, schema, rows) in writer.tables.drain_rows() {
266            for pending in exemplars
267                .iter_mut()
268                .filter(|pending| pending.metric.as_ref() == name)
269            {
270                pending.row = self.row_locations.len().checked_add(pending.row).context(
271                    error::InternalSnafu {
272                        err_msg: "exemplar source row offset overflow",
273                    },
274                )?;
275            }
276            let batch = match writer.batches.get(name) {
277                Some(&(columns, batch)) if columns == schema.len() => batch,
278                _ => {
279                    let mut hash = schema_batches.hasher().build_hasher();
280                    name.hash(&mut hash);
281                    for column in schema {
282                        (&column.column_name, column.datatype, column.semantic_type)
283                            .hash(&mut hash);
284                    }
285                    let candidates = schema_batches.entry(hash.finish()).or_default();
286                    let batch = candidates.iter().copied().find(|&index| {
287                        self.requests[index].table_name == name
288                            && self.requests[index]
289                                .rows
290                                .as_ref()
291                                .is_some_and(|rows| rows.schema == schema)
292                    });
293                    let batch = if let Some(batch) = batch {
294                        batch
295                    } else {
296                        validate_sample_kind(name, schema)?;
297                        let batch = self.requests.len();
298                        self.requests.push(RowInsertRequest {
299                            table_name: name.to_owned(),
300                            rows: Some(Rows {
301                                schema: schema.to_vec(),
302                                rows: Vec::new(),
303                            }),
304                        });
305                        candidates.push(batch);
306                        batch
307                    };
308                    writer
309                        .batches
310                        .insert(name.to_owned(), (schema.len(), batch));
311                    batch
312                }
313            };
314            let output = self.requests[batch]
315                .rows
316                .as_mut()
317                .context(error::InternalSnafu {
318                    err_msg: "missing prepared batch",
319                })?;
320            for row in rows {
321                self.row_locations.push((batch, output.rows.len()));
322                output.rows.push(row);
323            }
324        }
325        self.points.push(PreparedPoint {
326            metric: metric_index,
327            resource,
328            timestamp: timestamp as i64,
329            rows: first_row..self.row_locations.len(),
330            exemplars,
331        });
332        Ok(true)
333    }
334
335    fn reject(&mut self, exponential: bool, reason: String) -> Result<()> {
336        reject_data_points(&mut self.outcome, 1, || reason)?;
337        if exponential {
338            OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS
339                .with_label_values(&["invalid_data_point"])
340                .inc();
341        }
342        Ok(())
343    }
344
345    /// Number of source points that survived payload conversion.
346    pub fn num_points(&self) -> usize {
347        self.points.len()
348    }
349
350    fn validate_batches(&self) -> Result<()> {
351        let mut counts = vec![0; self.requests.len()];
352        for &(batch, row) in &self.row_locations {
353            let count = counts.get_mut(batch).context(error::InternalSnafu {
354                err_msg: "missing prepared batch",
355            })?;
356            ensure!(
357                row == *count,
358                error::InternalSnafu {
359                    err_msg: "invalid prepared row order",
360                }
361            );
362            *count += 1;
363        }
364        ensure!(
365            self.requests.iter().zip(counts).all(|(request, count)| {
366                count != 0
367                    && request
368                        .rows
369                        .as_ref()
370                        .is_some_and(|rows| rows.rows.len() == count)
371            }),
372            error::InternalSnafu {
373                err_msg: "OTLP prepared batch row counts changed"
374            }
375        );
376        Ok(())
377    }
378
379    /// Rejects a source point if any row in its fan-out failed destination validation.
380    pub fn point_rejections(&self, rows: Vec<Vec<Option<String>>>) -> Result<Vec<Option<String>>> {
381        self.validate_batches()?;
382        ensure!(
383            rows.len() == self.requests.len()
384                && rows.iter().zip(&self.requests).all(|(errors, request)| {
385                    request
386                        .rows
387                        .as_ref()
388                        .is_some_and(|rows| errors.len() == rows.rows.len())
389                }),
390            error::InternalSnafu {
391                err_msg: "OTLP batch rejections must match prepared rows"
392            }
393        );
394        Ok(self
395            .points
396            .iter()
397            .map(|point| {
398                self.row_locations[point.rows.clone()]
399                    .iter()
400                    .find_map(|&(batch, row)| rows[batch][row].clone())
401            })
402            .collect())
403    }
404
405    /// Complete authoritative validation before calling this method. Fresh
406    /// logical and physical conflicts are then collected from one survivor set.
407    pub fn finish(
408        mut self,
409        mut rejections: Vec<Option<String>>,
410        physical_tables: &HashMap<String, String>,
411    ) -> Result<MetricsConversion> {
412        ensure!(
413            rejections.len() == self.points.len(),
414            error::InternalSnafu {
415                err_msg: "OTLP prepared points and rejections must have equal lengths",
416            }
417        );
418        self.validate_batches()?;
419        let conflicts = conflicting_points(
420            &self.requests,
421            &self.points,
422            &self.row_locations,
423            &rejections,
424            physical_tables,
425        );
426        for index in conflicts {
427            rejections[index] =
428                Some("incompatible metric sample kinds or column definitions".to_string());
429        }
430        // Bind before survivor merging moves batch schemas and row values.
431        let mut exemplars = Vec::new();
432        for (point, rejection) in self.points.iter_mut().zip(&rejections) {
433            if rejection.is_some() {
434                continue;
435            }
436            for pending in std::mem::take(&mut point.exemplars) {
437                let &(batch, row) =
438                    self.row_locations
439                        .get(pending.row)
440                        .context(error::InternalSnafu {
441                            err_msg: "missing exemplar source row",
442                        })?;
443                let rows = self.requests[batch]
444                    .rows
445                    .as_ref()
446                    .context(error::InternalSnafu {
447                        err_msg: "missing prepared batch",
448                    })?;
449                exemplars.extend(pending.finish(&rows.schema, &rows.rows[row].values));
450            }
451        }
452        let reuse_batches = rejections.iter().all(Option::is_none) && {
453            let mut tables = HashSet::new();
454            self.requests.len() <= 1 || self.requests.iter().all(|r| tables.insert(&r.table_name))
455        };
456        let mut outputs: Vec<(String, TableData)> = Vec::new();
457        let mut output_indices = AHashMap::new();
458        let mut mappings: Vec<Option<(usize, Vec<usize>, bool)>> = if reuse_batches {
459            Vec::new()
460        } else {
461            vec![None; self.requests.len()]
462        };
463        let mut semantic_index = SemanticIndex::default();
464        let mut resource_info = ResourceInfoData::default();
465        let mut resource_times = vec![Vec::new(); self.resources.len()];
466        for (point, rejection) in std::mem::take(&mut self.points).into_iter().zip(rejections) {
467            if let Some(reason) = rejection {
468                let metadata = &self.metrics[point.metric];
469                self.reject(
470                    metadata.native_schema.is_some(),
471                    format!("metric `{}`: {reason}", metadata.original_name),
472                )?;
473                continue;
474            }
475            for &(batch, row) in &self.row_locations[point.rows] {
476                let request = &mut self.requests[batch];
477                semantic_index.merge_table_from(
478                    &mut self.metrics[point.metric].semantics,
479                    &request.table_name,
480                );
481                if reuse_batches {
482                    continue;
483                }
484                let rows = request.rows.as_mut().context(error::InternalSnafu {
485                    err_msg: "missing prepared batch",
486                })?;
487                if mappings[batch].is_none() {
488                    let output = *output_indices
489                        .entry(request.table_name.clone())
490                        .or_insert_with(|| {
491                            let index = outputs.len();
492                            outputs.push((
493                                request.table_name.clone(),
494                                TableData::new(rows.schema.len(), rows.rows.len()),
495                            ));
496                            index
497                        });
498                    let columns = std::mem::take(&mut rows.schema)
499                        .into_iter()
500                        .map(|column| outputs[output].1.ensure_column(column))
501                        .collect::<Result<Vec<_>>>()?;
502                    let identity = columns
503                        .iter()
504                        .enumerate()
505                        .all(|(index, &column)| index == column);
506                    mappings[batch] = Some((output, columns, identity));
507                }
508                let (output, columns, identity) =
509                    mappings[batch].as_ref().context(error::InternalSnafu {
510                        err_msg: "missing output column mapping",
511                    })?;
512                let table = &mut outputs[*output].1;
513                let values = std::mem::take(&mut rows.rows[row].values);
514                if *identity {
515                    table.add_row(values);
516                } else {
517                    let mut output = table.alloc_one_row();
518                    for (value, &column) in values.into_iter().zip(columns) {
519                        output[column] = value;
520                    }
521                    table.add_row(output);
522                }
523            }
524            add_accepted_data_points(&mut self.outcome, 1)?;
525            if !self.resources[point.resource].is_empty() {
526                resource_times[point.resource].push(point.timestamp);
527            }
528        }
529        for (attrs, times) in self.resources.iter().zip(resource_times) {
530            if !times.is_empty() {
531                resource_info.observe_times(attrs, times);
532            }
533        }
534        let requests = RowInsertRequests {
535            inserts: if reuse_batches {
536                std::mem::take(&mut self.requests)
537            } else {
538                outputs
539                    .into_iter()
540                    .map(|(table_name, table)| RowInsertRequest {
541                        table_name,
542                        rows: Some(table.into_rows()),
543                    })
544                    .collect()
545            },
546        };
547        let rows = requests
548            .inserts
549            .iter()
550            .filter_map(|r| r.rows.as_ref())
551            .map(|rows| rows.rows.len())
552            .sum();
553        let resource_info = if !resource_info.is_empty()
554            && requests
555                .inserts
556                .iter()
557                .any(|r| r.table_name == OTEL_RESOURCE_INFO_TABLE_NAME)
558        {
559            common_telemetry::warn!(
560                "Skipping OTLP resource descriptor synthesis: a metric uses its table name"
561            );
562            None
563        } else {
564            resource_info.into_row_insert_requests()?
565        };
566        if resource_info.is_some() {
567            semantic_index.record_scalar(
568                OTEL_RESOURCE_INFO_TABLE_NAME,
569                SEMANTIC_METRIC_TYPE,
570                crate::semantic::METRIC_TYPE_INFO,
571            );
572            semantic_index.record_scalar(
573                OTEL_RESOURCE_INFO_TABLE_NAME,
574                SEMANTIC_METRIC_METADATA_QUALITY,
575                METADATA_QUALITY_DECLARED,
576            );
577        }
578        Ok(MetricsConversion {
579            exemplars,
580            requests,
581            rows,
582            semantic_index,
583            resource_info,
584            outcome: self.outcome,
585        })
586    }
587}
588
589fn conflicting_points(
590    requests: &[RowInsertRequest],
591    points: &[PreparedPoint],
592    row_locations: &[(usize, usize)],
593    rejections: &[Option<String>],
594    physical_tables: &HashMap<String, String>,
595) -> HashSet<usize> {
596    let mut contributors = vec![Vec::new(); requests.len()];
597    for (index, point) in points
598        .iter()
599        .enumerate()
600        .filter(|(index, _)| rejections[*index].is_none())
601    {
602        for &(batch, _) in &row_locations[point.rows.clone()] {
603            if contributors[batch].last() != Some(&index) {
604                contributors[batch].push(index);
605            }
606        }
607    }
608    let mut columns = AHashMap::<_, Vec<(usize, &ColumnSchema)>>::new();
609    let mut kinds: AHashMap<&str, Vec<(usize, bool)>> = AHashMap::new();
610    for (batch, request) in requests
611        .iter()
612        .enumerate()
613        .filter(|(batch, _)| !contributors[*batch].is_empty())
614    {
615        let Some(rows) = &request.rows else { continue };
616        for column in &rows.schema {
617            columns
618                .entry((false, &request.table_name, &column.column_name))
619                .or_default()
620                .push((batch, column));
621            if let Some(physical) = physical_tables.get(&request.table_name) {
622                columns
623                    .entry((true, physical, &column.column_name))
624                    .or_default()
625                    .push((batch, column));
626            }
627            if column.semantic_type == SemanticType::Field as i32 {
628                let native = api::helper::is_column_type_value_eq(
629                    column.datatype,
630                    column.datatype_extension.clone(),
631                    native_histogram_value_type(),
632                );
633                kinds
634                    .entry(&request.table_name)
635                    .or_default()
636                    .push((batch, native));
637            }
638        }
639    }
640    let mut rejected = HashSet::new();
641    for group in columns.values() {
642        let first = group[0].1;
643        if group.iter().any(|(_, column)| {
644            column.datatype != first.datatype
645                || column.datatype_extension != first.datatype_extension
646                || column.semantic_type != first.semantic_type
647        }) {
648            for (batch, _) in group {
649                rejected.extend(contributors[*batch].iter().copied());
650            }
651        }
652    }
653    for group in kinds.values() {
654        if group.iter().any(|(_, native)| *native) && group.iter().any(|(_, native)| !*native) {
655            for (batch, _) in group {
656                rejected.extend(contributors[*batch].iter().copied());
657            }
658        }
659    }
660    rejected
661}
servers/otlp/metrics/prepare.rs

servers/otlp/metrics/
prepare.rs
