operator/statement/semantic_graph/
conventions.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//! The built-in derivation conventions of the entity graph, embedded as
16//! [`conventions.yaml`](./conventions.yaml): the co-declared edge vocabulary,
17//! the virtual-destination candidates, and the implicit declarations of
18//! well-known Prometheus entity-descriptor metrics. The file is data shipped
19//! with the binary, not an operator-editable configuration surface; explicit
20//! `greptime.semantic.entity.*` declarations always override it.
21
22use std::collections::{BTreeMap, HashMap, HashSet};
23use std::sync::LazyLock;
24
25use serde::Deserialize;
26
27/// A same-row co-declaration rule: a source-table row carrying both entity
28/// identities witnesses `src -rel-> dst`.
29#[derive(Debug, Deserialize)]
30#[serde(deny_unknown_fields)]
31pub struct EdgeRule {
32    pub src: String,
33    pub dst: String,
34    pub rel: String,
35}
36
37/// A span-attribute column that may name an uninstrumented peer, and the
38/// `connection_type` its match implies.
39#[derive(Debug, Deserialize)]
40#[serde(deny_unknown_fields)]
41pub struct VirtualDstCandidate {
42    pub column: String,
43    pub connection_type: String,
44}
45
46/// One implicit entity declaration: of a whitelisted Prometheus or OTel info
47/// metric, or of a trace-v1 table's flattened resource attributes.
48#[derive(Debug, Deserialize)]
49#[serde(deny_unknown_fields)]
50pub struct ImplicitEntity {
51    pub entity: String,
52    /// Identifying label columns, ordered broad to narrow; every one must
53    /// exist on the table for the declaration to apply.
54    pub id: Vec<String>,
55    /// Column qualifying `id[0]` as `<qualifier>/<id[0]>`, so a source
56    /// carrying the parts separately matches one carrying them pre-composed.
57    /// Skipped when the column is absent or empty.
58    #[serde(default)]
59    pub qualified_by: Option<String>,
60    /// A more specific entity type declared next to this one, which takes over
61    /// on the rows carrying its full identity. Naming the type rather than a
62    /// trigger column is what keeps the entity replaceable but never
63    /// droppable: where the specific type is not derivable, this one stands.
64    #[serde(default)]
65    pub superseded_by: Option<String>,
66    /// Descriptive label columns, filtered to those present (kube-state-metrics
67    /// label sets vary across versions).
68    #[serde(default)]
69    pub descriptive: Vec<String>,
70    /// Snapshot every tag column except the id columns as descriptive —
71    /// `target_info`-style enrichment over an open attribute set.
72    #[serde(default)]
73    pub descriptive_rest: bool,
74}
75
76/// The parsed, validated conventions file.
77#[derive(Debug, Deserialize)]
78#[serde(deny_unknown_fields)]
79pub struct Conventions {
80    pub co_declared_edges: Vec<EdgeRule>,
81    pub trace_co_declared_edges: Vec<EdgeRule>,
82    pub virtual_dst_candidates: Vec<VirtualDstCandidate>,
83    pub otlp_trace_entities: Vec<ImplicitEntity>,
84    /// Table name -> the entities that table declares, for Prometheus-sourced
85    /// descriptor metrics (`source = prometheus`).
86    pub prometheus_info_metrics: BTreeMap<String, Vec<ImplicitEntity>>,
87    /// Table name -> the entities that table declares, for OTLP-sourced
88    /// descriptor tables (`source = opentelemetry`).
89    pub otel_info_metrics: BTreeMap<String, Vec<ImplicitEntity>>,
90}
91
92/// The built-in entity-type vocabulary. User-declared types are open-ended;
93/// the embedded conventions file must stay inside this set.
94pub const ENTITY_TYPE_SERVICE: &str = "service";
95pub const ENTITY_TYPE_SERVICE_INSTANCE: &str = "service.instance";
96pub const ENTITY_TYPE_HOST: &str = "host";
97pub const ENTITY_TYPE_CONTAINER: &str = "container";
98pub const ENTITY_TYPE_PROCESS: &str = "process";
99pub const ENTITY_TYPE_K8S_POD: &str = "k8s.pod";
100pub const ENTITY_TYPE_K8S_NODE: &str = "k8s.node";
101pub const ENTITY_TYPE_K8S_CONTAINER: &str = "k8s.container";
102pub const ENTITY_TYPE_K8S_WORKLOAD: &str = "k8s.workload";
103pub const ENTITY_TYPE_K8S_SERVICE: &str = "k8s.service";
104pub const ENTITY_TYPE_GEN_AI_AGENT: &str = "gen_ai.agent";
105pub const ENTITY_TYPE_GEN_AI_MODEL: &str = "gen_ai.model";
106pub const ENTITY_TYPE_GEN_AI_TOOL: &str = "gen_ai.tool";
107
108const ENTITY_TYPES: [&str; 13] = [
109    ENTITY_TYPE_SERVICE,
110    ENTITY_TYPE_SERVICE_INSTANCE,
111    ENTITY_TYPE_HOST,
112    ENTITY_TYPE_CONTAINER,
113    ENTITY_TYPE_PROCESS,
114    ENTITY_TYPE_K8S_POD,
115    ENTITY_TYPE_K8S_NODE,
116    ENTITY_TYPE_K8S_CONTAINER,
117    ENTITY_TYPE_K8S_WORKLOAD,
118    ENTITY_TYPE_K8S_SERVICE,
119    ENTITY_TYPE_GEN_AI_AGENT,
120    ENTITY_TYPE_GEN_AI_MODEL,
121    ENTITY_TYPE_GEN_AI_TOOL,
122];
123
124/// The relationship vocabulary (the RFC's rel_type table) and the edge
125/// provenances.
126pub const REL_TYPE_CALLS: &str = "calls";
127pub const REL_TYPE_RUNS_ON: &str = "runs_on";
128pub const REL_TYPE_CONTAINS: &str = "contains";
129pub const REL_TYPE_PART_OF: &str = "part_of";
130pub const REL_TYPE_USES: &str = "uses";
131pub const REL_TYPE_INVOKES: &str = "invokes";
132pub const REL_TYPE_DEPENDS_ON: &str = "depends_on";
133pub const REL_TYPE_OWNS: &str = "owns";
134pub const PROVENANCE_TRACE: &str = "trace";
135pub const PROVENANCE_ATTRIBUTE: &str = "attribute";
136pub const PROVENANCE_DECLARED: &str = "declared";
137pub const PROVENANCE_AGENT: &str = "agent";
138
139const REL_TYPES: [&str; 8] = [
140    REL_TYPE_CALLS,
141    REL_TYPE_RUNS_ON,
142    REL_TYPE_CONTAINS,
143    REL_TYPE_PART_OF,
144    REL_TYPE_USES,
145    REL_TYPE_INVOKES,
146    REL_TYPE_DEPENDS_ON,
147    REL_TYPE_OWNS,
148];
149
150/// The `connection_type` values a virtual-destination candidate may imply;
151/// the calls derivation branches on them when building edge attributes.
152pub const CONNECTION_TYPE_DATABASE: &str = "database";
153pub const CONNECTION_TYPE_VIRTUAL_NODE: &str = "virtual_node";
154
155const CONNECTION_TYPES: [&str; 2] = [CONNECTION_TYPE_DATABASE, CONNECTION_TYPE_VIRTUAL_NODE];
156
157static CONVENTIONS: LazyLock<Result<Conventions, String>> =
158    LazyLock::new(|| parse(include_str!("conventions.yaml")));
159
160/// The embedded conventions. `Err` means the embedded file is broken — pinned
161/// by unit test, and propagated by the derivation paths rather than panicking.
162pub fn conventions() -> Result<&'static Conventions, String> {
163    CONVENTIONS.as_ref().map_err(Clone::clone)
164}
165
166fn parse(yaml: &str) -> Result<Conventions, String> {
167    let conventions: Conventions =
168        serde_yaml_ng::from_str(yaml).map_err(|e| format!("malformed conventions: {e}"))?;
169    validate(&conventions)?;
170    Ok(conventions)
171}
172
173fn validate(conventions: &Conventions) -> Result<(), String> {
174    let mut seen_edges = HashSet::new();
175    for rule in conventions
176        .co_declared_edges
177        .iter()
178        .chain(&conventions.trace_co_declared_edges)
179    {
180        for ty in [&rule.src, &rule.dst] {
181            if !ENTITY_TYPES.contains(&ty.as_str()) {
182                return Err(format!("unknown entity type `{ty}` in edge vocabulary"));
183            }
184        }
185        if !REL_TYPES.contains(&rule.rel.as_str()) {
186            return Err(format!(
187                "unknown rel_type `{}` in edge vocabulary",
188                rule.rel
189            ));
190        }
191        if !seen_edges.insert((&rule.src, &rule.dst, &rule.rel)) {
192            return Err(format!(
193                "duplicate edge rule `{} -{}-> {}`",
194                rule.src, rule.rel, rule.dst
195            ));
196        }
197    }
198
199    let mut seen_columns = HashSet::new();
200    for candidate in &conventions.virtual_dst_candidates {
201        if candidate.column.is_empty() {
202            return Err("empty virtual destination column".to_string());
203        }
204        if !CONNECTION_TYPES.contains(&candidate.connection_type.as_str()) {
205            return Err(format!(
206                "unknown connection_type `{}` for virtual destination `{}`",
207                candidate.connection_type, candidate.column
208            ));
209        }
210        if !seen_columns.insert(&candidate.column) {
211            return Err(format!(
212                "duplicate virtual destination column `{}`",
213                candidate.column
214            ));
215        }
216    }
217
218    let per_table = conventions
219        .prometheus_info_metrics
220        .iter()
221        .chain(&conventions.otel_info_metrics)
222        .map(|(table, entities)| (table.as_str(), entities))
223        .chain(std::iter::once((
224            "otlp traces",
225            &conventions.otlp_trace_entities,
226        )));
227    // An entity type declared with a different number of id columns by two
228    // sources yields ids that can never match, silently splitting one entity
229    // in two.
230    let mut arity = HashMap::new();
231    for (table, entities) in per_table {
232        let mut seen_types = HashSet::new();
233        for implicit in entities {
234            if !ENTITY_TYPES.contains(&implicit.entity.as_str()) {
235                return Err(format!(
236                    "unknown entity type `{}` for info metric `{table}`",
237                    implicit.entity
238                ));
239            }
240            if !seen_types.insert(&implicit.entity) {
241                return Err(format!(
242                    "duplicate entity type `{}` for info metric `{table}`",
243                    implicit.entity
244                ));
245            }
246            if implicit.id.is_empty() || implicit.id.iter().any(String::is_empty) {
247                return Err(format!(
248                    "entity `{}` of info metric `{table}` needs non-empty id columns",
249                    implicit.entity
250                ));
251            }
252            if let Some(superseding) = &implicit.superseded_by {
253                let superseding = entities
254                    .iter()
255                    .find(|other| &other.entity == superseding)
256                    .ok_or_else(|| {
257                        format!(
258                            "entity `{}` of info metric `{table}` is superseded by `{superseding}`, \
259                             which `{table}` does not declare",
260                            implicit.entity
261                        )
262                    })?;
263                // A chain would let the middle entity withdraw while its own
264                // replacement is absent.
265                if superseding.superseded_by.is_some() {
266                    return Err(format!(
267                        "entity `{}` of info metric `{table}` is superseded by `{}`, which is \
268                         itself superseded",
269                        implicit.entity, superseding.entity
270                    ));
271                }
272            }
273            if implicit.qualified_by.as_ref().is_some_and(String::is_empty) {
274                return Err(format!(
275                    "entity `{}` of info metric `{table}` has an empty qualified_by",
276                    implicit.entity
277                ));
278            }
279            if implicit.descriptive_rest && !implicit.descriptive.is_empty() {
280                return Err(format!(
281                    "entity `{}` of info metric `{table}` sets both descriptive and \
282                     descriptive_rest",
283                    implicit.entity
284                ));
285            }
286            match arity.insert(implicit.entity.as_str(), implicit.id.len()) {
287                Some(previous) if previous != implicit.id.len() => {
288                    return Err(format!(
289                        "entity `{}` is declared with {previous} id columns elsewhere but \
290                         {} for `{table}`",
291                        implicit.entity,
292                        implicit.id.len()
293                    ));
294                }
295                _ => {}
296            }
297        }
298    }
299    Ok(())
300}
301
302#[cfg(test)]
303mod tests {
304    use super::*;
305
306    #[test]
307    fn embedded_conventions_parse_and_validate() {
308        conventions().unwrap();
309    }
310
311    /// Each case must fail on exactly the rule it names, so the shared
312    /// boilerplate is valid and the mutated part is inside the vocabulary.
313    fn broken(edges: &str, info_metrics: &str, otel_metrics: &str) -> String {
314        format!(
315            "co_declared_edges: [{edges}]\ntrace_co_declared_edges: []\n\
316             virtual_dst_candidates: []\notlp_trace_entities: []\n\
317             prometheus_info_metrics: {{{info_metrics}}}\n\
318             otel_info_metrics: {{{otel_metrics}}}"
319        )
320    }
321
322    #[test]
323    fn validation_rejects_broken_conventions() {
324        let err = |edges, info, otel| parse(&broken(edges, info, otel)).unwrap_err();
325
326        assert!(err("{src: host, dst: service, rel: pets}", "", "").contains("unknown rel_type"));
327        assert!(
328            err("{src: k8s.pods, dst: k8s.node, rel: runs_on}", "", "")
329                .contains("unknown entity type")
330        );
331        assert!(
332            err(
333                "{src: host, dst: service, rel: uses}, {src: host, dst: service, rel: uses}",
334                "",
335                ""
336            )
337            .contains("duplicate edge rule")
338        );
339        assert!(
340            err(
341                "",
342                "t: [{entity: host, id: [x], descriptive: [y], descriptive_rest: true}]",
343                ""
344            )
345            .contains("descriptive_rest")
346        );
347        // Superseding a type the same table does not declare leaves the rule
348        // dead and the duplicate node back in the graph.
349        assert!(
350            err(
351                "",
352                "t: [{entity: container, id: [x], superseded_by: k8s.container}]",
353                ""
354            )
355            .contains("does not declare")
356        );
357        // the otel map runs through the same per-table validation
358        assert!(err("", "", "t: [{entity: hosts, id: [x]}]").contains("unknown entity type"));
359        // ids of a different arity for one type can never match each other
360        assert!(
361            err(
362                "",
363                "a: [{entity: host, id: [x]}]",
364                "b: [{entity: host, id: [x, y]}]"
365            )
366            .contains("id columns")
367        );
368        // Unknown YAML keys are rejected, catching typos in the embedded file.
369        assert!(
370            parse(&broken(
371                "{src: host, dst: service, rel: uses, direction: down}",
372                "",
373                ""
374            ))
375            .is_err()
376        );
377    }
378}
operator/statement/semantic_graph/conventions.rs

operator/statement/semantic_graph/
conventions.rs
