servers/
semantic.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//! Per-table semantic metadata accumulated during one ingest pass (OTLP
16//! metrics, Prometheus remote write v2).
17//!
18//! Each written table collects the scalar semantic keys its wire metadata
19//! declares (an OTLP histogram fans out into `_bucket`/`_sum`/`_count`
20//! companions; a remote write series names one metric family). The resulting
21//! index is serialized onto the `greptime.internal.semantic.per_table_index`
22//! context extension as `{schema -> {table -> {key -> value}}}` and folded into
23//! each table's options at auto-create time.
24//!
25//! Conflict handling follows the RFC: when two sources disagree on a
26//! single-valued key the value collapses to `mixed` (or `unknown` for keys whose
27//! domain has no `mixed`).
28
29use std::collections::{BTreeMap, HashMap};
30
31use table::requests::{SEMANTIC_VALUE_MIXED, SEMANTIC_VALUE_UNKNOWN, validate_semantic_option};
32
33// `greptime.semantic.metric.type` values stamped per emitted table. Must stay
34// within the domain accepted by `validate_semantic_option`; the drift-guard test
35// asserts this.
36pub const METRIC_TYPE_COUNTER: &str = "counter";
37pub const METRIC_TYPE_UPDOWN_COUNTER: &str = "updown_counter";
38pub const METRIC_TYPE_GAUGE: &str = "gauge";
39pub const METRIC_TYPE_HISTOGRAM: &str = "histogram";
40pub const METRIC_TYPE_GAUGE_HISTOGRAM: &str = "gauge_histogram";
41pub const METRIC_TYPE_SUMMARY: &str = "summary";
42pub const METRIC_TYPE_INFO: &str = "info";
43pub const METRIC_TYPE_STATESET: &str = "stateset";
44
45/// Maps an OpenMetrics unit name (open vocabulary full words: `seconds`,
46/// `bytes`) to the UCUM code the `greptime.semantic.metric.unit` option is
47/// defined in. Only the OpenMetrics base units are mapped; anything else is
48/// dropped — a missing unit beats a corrupted cross-protocol vocabulary.
49pub fn openmetrics_unit_to_ucum(unit: &str) -> Option<&'static str> {
50    Some(match unit {
51        "seconds" => "s",
52        "celsius" => "Cel",
53        "meters" => "m",
54        "bytes" => "By",
55        "ratios" => "1",
56        "volts" => "V",
57        "amperes" => "A",
58        "joules" => "J",
59        "grams" => "g",
60        _ => return None,
61    })
62}
63
64/// Index of `{table_name -> {semantic_key -> value}}` for one target schema.
65#[derive(Debug, Default)]
66pub struct SemanticIndex {
67    /// Per-table scalar keys; conflicting values collapse to `mixed`/`unknown`.
68    tables: HashMap<String, BTreeMap<&'static str, String>>,
69}
70
71impl SemanticIndex {
72    pub fn is_empty(&self) -> bool {
73        self.tables.is_empty()
74    }
75
76    /// Records a scalar semantic key for `table`. A value conflicting with one
77    /// already recorded collapses the key to `mixed`/`unknown`; once collapsed
78    /// it stays collapsed.
79    pub fn record_scalar(&mut self, table: &str, key: &'static str, value: &str) {
80        // Avoid allocating the table name (and an empty map) on the common path
81        // where the table is already present.
82        if let Some(scalars) = self.tables.get_mut(table) {
83            match scalars.get(key).map(String::as_str) {
84                Some(existing) if existing == value => {}
85                Some(SEMANTIC_VALUE_MIXED) | Some(SEMANTIC_VALUE_UNKNOWN) => {}
86                Some(_) => {
87                    scalars.insert(key, collapse_value(key));
88                }
89                None => {
90                    scalars.insert(key, value.to_string());
91                }
92            }
93        } else {
94            self.tables.insert(
95                table.to_string(),
96                BTreeMap::from([(key, value.to_string())]),
97            );
98        }
99    }
100
101    /// Serializes to the JSON `{schema -> {table -> {key -> value}}}` carried
102    /// on the context extension, with every table under `schema`. `None` when
103    /// nothing was recorded.
104    pub fn encode(&self, schema: &str) -> Option<String> {
105        if self.tables.is_empty() {
106            return None;
107        }
108        serde_json::to_string(&BTreeMap::from([(schema, &self.tables)])).ok()
109    }
110
111    fn merge_from(&mut self, other: &SemanticIndex) {
112        for (table, scalars) in &other.tables {
113            for (key, value) in scalars {
114                self.record_scalar(table, key, value);
115            }
116        }
117    }
118
119    #[cfg(test)]
120    fn options_of(&self, table: &str) -> Option<&BTreeMap<&'static str, String>> {
121        self.tables.get(table)
122    }
123}
124
125/// Schema-aware collection of [`SemanticIndex`]es: Prometheus remote write lets
126/// each series override the target schema with a special label, so one request
127/// may write the same metric name into several schemas — their metadata must
128/// not collapse into each other.
129#[derive(Debug, Default)]
130pub struct SemanticIndexes {
131    /// Tables written into the request's default schema (no override).
132    default: SemanticIndex,
133    /// Tables written under a per-series schema override.
134    overrides: HashMap<String, SemanticIndex>,
135}
136
137impl SemanticIndexes {
138    pub fn is_empty(&self) -> bool {
139        self.default.is_empty() && self.overrides.values().all(SemanticIndex::is_empty)
140    }
141
142    /// The index for `schema` (`None` = the request's default schema).
143    pub fn index_for(&mut self, schema: Option<&str>) -> &mut SemanticIndex {
144        match schema {
145            None => &mut self.default,
146            Some(schema) => {
147                if !self.overrides.contains_key(schema) {
148                    self.overrides
149                        .insert(schema.to_string(), SemanticIndex::default());
150                }
151                self.overrides.get_mut(schema).expect("just inserted")
152            }
153        }
154    }
155
156    /// Serializes to the JSON `{schema -> {table -> {key -> value}}}` carried on
157    /// the context extension, resolving the default index to `default_schema`.
158    /// An override explicitly naming `default_schema` merges into the default
159    /// with the usual conflict collapse.
160    pub fn encode(&self, default_schema: &str) -> Option<String> {
161        if self.is_empty() {
162            return None;
163        }
164        let mut by_schema: BTreeMap<&str, &HashMap<String, BTreeMap<&'static str, String>>> =
165            BTreeMap::new();
166        let mut merged_default;
167        if let Some(aliased) = self.overrides.get(default_schema) {
168            merged_default = SemanticIndex::default();
169            merged_default.merge_from(&self.default);
170            merged_default.merge_from(aliased);
171            by_schema.insert(default_schema, &merged_default.tables);
172        } else if !self.default.is_empty() {
173            by_schema.insert(default_schema, &self.default.tables);
174        }
175        for (schema, index) in &self.overrides {
176            if schema != default_schema && !index.is_empty() {
177                by_schema.insert(schema, &index.tables);
178            }
179        }
180        serde_json::to_string(&by_schema).ok()
181    }
182}
183
184/// The collapsed value for a conflicting scalar key: `mixed` when the key's
185/// domain accepts it, else `unknown`. Uses the vocabulary validator as the
186/// single source of truth for which keys allow `mixed`.
187fn collapse_value(key: &str) -> String {
188    if validate_semantic_option(key, SEMANTIC_VALUE_MIXED) {
189        SEMANTIC_VALUE_MIXED.to_string()
190    } else {
191        SEMANTIC_VALUE_UNKNOWN.to_string()
192    }
193}
194
195#[cfg(test)]
196mod tests {
197    use table::requests::{
198        SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_METRIC_TYPE, SEMANTIC_METRIC_UNIT,
199    };
200
201    use super::*;
202
203    type Decoded = BTreeMap<String, BTreeMap<String, BTreeMap<String, String>>>;
204
205    #[test]
206    fn test_scalar_recording_keeps_first_then_collapses_on_conflict() {
207        let mut index = SemanticIndex::default();
208        index.record_scalar("t", SEMANTIC_METRIC_TYPE, "counter");
209        index.record_scalar("t", SEMANTIC_METRIC_TYPE, "counter");
210        assert_eq!(
211            index
212                .options_of("t")
213                .unwrap()
214                .get(SEMANTIC_METRIC_TYPE)
215                .map(String::as_str),
216            Some("counter")
217        );
218
219        // Conflict on a key whose domain has `mixed` collapses to `mixed`.
220        index.record_scalar("t", SEMANTIC_METRIC_TYPE, "gauge");
221        assert_eq!(
222            index
223                .options_of("t")
224                .unwrap()
225                .get(SEMANTIC_METRIC_TYPE)
226                .map(String::as_str),
227            Some("mixed")
228        );
229        // Further writes stay collapsed.
230        index.record_scalar("t", SEMANTIC_METRIC_TYPE, "histogram");
231        assert_eq!(
232            index
233                .options_of("t")
234                .unwrap()
235                .get(SEMANTIC_METRIC_TYPE)
236                .map(String::as_str),
237            Some("mixed")
238        );
239    }
240
241    #[test]
242    fn test_scalar_conflict_without_mixed_domain_collapses_to_unknown() {
243        let mut index = SemanticIndex::default();
244        index.record_scalar("t", SEMANTIC_METRIC_METADATA_QUALITY, "declared");
245        index.record_scalar("t", SEMANTIC_METRIC_METADATA_QUALITY, "inferred");
246        // metadata_quality accepts only declared/inferred/unknown, so a conflict
247        // is `unknown`.
248        assert_eq!(
249            index
250                .options_of("t")
251                .unwrap()
252                .get(SEMANTIC_METRIC_METADATA_QUALITY)
253                .map(String::as_str),
254            Some("unknown")
255        );
256    }
257
258    #[test]
259    fn test_encode_is_none_when_empty_and_round_trips() {
260        let index = SemanticIndex::default();
261        assert!(index.is_empty());
262        assert_eq!(index.encode("public"), None);
263
264        let mut index = SemanticIndex::default();
265        index.record_scalar("metric_a", SEMANTIC_METRIC_TYPE, "counter");
266        index.record_scalar("metric_a", SEMANTIC_METRIC_UNIT, "By");
267        let json = index.encode("public").unwrap();
268        let parsed: Decoded = serde_json::from_str(&json).unwrap();
269        let table = parsed.get("public").unwrap().get("metric_a").unwrap();
270        assert_eq!(
271            table.get(SEMANTIC_METRIC_TYPE).map(String::as_str),
272            Some("counter")
273        );
274        assert_eq!(
275            table.get(SEMANTIC_METRIC_UNIT).map(String::as_str),
276            Some("By")
277        );
278    }
279
280    #[test]
281    fn test_indexes_keep_schemas_apart_and_merge_default_alias() {
282        let mut indexes = SemanticIndexes::default();
283        assert!(indexes.is_empty());
284        assert_eq!(indexes.encode("public"), None);
285
286        // The same metric name in two schemas must not collapse to `mixed`.
287        indexes
288            .index_for(None)
289            .record_scalar("cpu_usage", SEMANTIC_METRIC_TYPE, "counter");
290        indexes.index_for(Some("tenant_b")).record_scalar(
291            "cpu_usage",
292            SEMANTIC_METRIC_TYPE,
293            "gauge",
294        );
295        let parsed: Decoded = serde_json::from_str(&indexes.encode("public").unwrap()).unwrap();
296        assert_eq!(
297            parsed["public"]["cpu_usage"][SEMANTIC_METRIC_TYPE],
298            "counter"
299        );
300        assert_eq!(
301            parsed["tenant_b"]["cpu_usage"][SEMANTIC_METRIC_TYPE],
302            "gauge"
303        );
304
305        // An override naming the default schema merges into it — and a real
306        // conflict then collapses.
307        indexes
308            .index_for(Some("public"))
309            .record_scalar("cpu_usage", SEMANTIC_METRIC_TYPE, "gauge");
310        let parsed: Decoded = serde_json::from_str(&indexes.encode("public").unwrap()).unwrap();
311        assert_eq!(parsed["public"]["cpu_usage"][SEMANTIC_METRIC_TYPE], "mixed");
312    }
313
314    #[test]
315    fn test_openmetrics_unit_mapping() {
316        assert_eq!(openmetrics_unit_to_ucum("seconds"), Some("s"));
317        assert_eq!(openmetrics_unit_to_ucum("bytes"), Some("By"));
318        assert_eq!(openmetrics_unit_to_ucum("ratios"), Some("1"));
319        // Outside the OpenMetrics base set: dropped, not passed through.
320        assert_eq!(openmetrics_unit_to_ucum("requests"), None);
321        assert_eq!(openmetrics_unit_to_ucum(""), None);
322        // No fuzzy matching: UCUM codes are not OpenMetrics names.
323        assert_eq!(openmetrics_unit_to_ucum("By"), None);
324    }
325}
servers/semantic.rs

servers/
semantic.rs
