micromegas_analytics/lakehouse/otel/attrs.rs
1//! Converts OTel `KeyValue` arrays + scalar `AnyValue` instances to JSONB bytes.
2//!
3//! Mapping follows the plan's "Attribute value encoding" table:
4//! - string → JSON string
5//! - int / double / bool → JSON number / bool
6//! - bytes → base64-encoded string (existing properties consumers expect text)
7//! - array / kvlist → recursive JSON
8//!
9//! The output is a JSONB-encoded `{key → value}` blob suitable for the
10//! `properties` columns across `log_entries`, `measures`, and `otel_spans`.
11
12use base64::Engine;
13use jsonb::{Number as JsonbNumber, Value as JsonbValue};
14use micromegas_tracing::prelude::*;
15use opentelemetry_proto::tonic::common::v1::{
16 AnyValue, InstrumentationScope, KeyValue, any_value::Value as Av,
17};
18use std::borrow::Cow;
19use std::collections::BTreeMap;
20use std::sync::Once;
21
22/// `AnyValue.string_value_strindex` / `KeyValue.key_strindex` reference a
23/// `ProfilesDictionary.string_table` that exists **only** for the Profiling signal. Per the
24/// OTLP spec, receivers of logs/metrics/traces MUST treat these as a non-fatal issue, log a
25/// warning, and process the data as if the value were absent. We warn once per process so a
26/// misconfigured profiling producer pointed at a non-profiling endpoint is noticeable without
27/// flooding the logs.
28fn warn_unexpected_strindex() {
29 static WARNED: Once = Once::new();
30 WARNED.call_once(|| {
31 warn!(
32 "ignoring profiling-only string-interning field on a non-profiling OTLP signal; \
33 treating it as absent (OTLP spec)"
34 );
35 });
36}
37
38/// Converts an `AnyValue` to a `jsonb::Value`. Recursively handles arrays and kvlists.
39pub fn any_value_to_jsonb(v: &AnyValue) -> JsonbValue<'static> {
40 match v.value.as_ref() {
41 Some(Av::StringValue(s)) => JsonbValue::String(Cow::Owned(s.clone())),
42 Some(Av::BoolValue(b)) => JsonbValue::Bool(*b),
43 Some(Av::IntValue(i)) => JsonbValue::Number(JsonbNumber::Int64(*i)),
44 Some(Av::DoubleValue(d)) => JsonbValue::Number(JsonbNumber::Float64(*d)),
45 Some(Av::BytesValue(b)) => {
46 // existing JSONB readers (`jsonb_extract_path`, etc.) expect strings,
47 // so we base64-encode bytes rather than emitting a JSON binary type.
48 let encoded = base64::engine::general_purpose::STANDARD.encode(b);
49 JsonbValue::String(Cow::Owned(encoded))
50 }
51 Some(Av::ArrayValue(arr)) => {
52 JsonbValue::Array(arr.values.iter().map(any_value_to_jsonb).collect())
53 }
54 Some(Av::KvlistValue(kvs)) => {
55 let mut map: BTreeMap<String, JsonbValue<'static>> = BTreeMap::new();
56 for kv in &kvs.values {
57 let value = kv
58 .value
59 .as_ref()
60 .map(any_value_to_jsonb)
61 .unwrap_or(JsonbValue::Null);
62 // `kv.key_strindex` (profiling-only) is intentionally ignored: keying off `kv.key`
63 // means an interned key (empty `key`) becomes an empty-key entry, i.e. absent.
64 map.insert(kv.key.clone(), value);
65 }
66 JsonbValue::Object(map)
67 }
68 // Profiling-only string-table reference: no dictionary exists for this signal, so the
69 // index has no meaning here. Treat as absent (per OTLP spec) — never as data.
70 Some(Av::StringValueStrindex(_)) => {
71 warn_unexpected_strindex();
72 JsonbValue::Null
73 }
74 None => JsonbValue::Null,
75 }
76}
77
78/// Encodes a `JsonbValue` to its on-wire JSONB bytes.
79pub fn to_jsonb_bytes(value: JsonbValue<'_>) -> Vec<u8> {
80 let mut bytes = Vec::new();
81 value.write_to_vec(&mut bytes);
82 bytes
83}
84
85/// Builds a flat `(key → value)` `JsonbValue::Object` (with optional extra entries
86/// layered on top). Output ordering is alphabetical (a `BTreeMap` underneath),
87/// matching `serialize_properties_to_jsonb`.
88pub fn attrs_to_jsonb_value(
89 attrs: &[KeyValue],
90 extras: &[(String, JsonbValue<'static>)],
91) -> JsonbValue<'static> {
92 let mut map: BTreeMap<String, JsonbValue<'static>> = BTreeMap::new();
93 for kv in attrs {
94 let value = kv
95 .value
96 .as_ref()
97 .map(any_value_to_jsonb)
98 .unwrap_or(JsonbValue::Null);
99 // `kv.key_strindex` (profiling-only) is intentionally ignored — see `any_value_to_jsonb`.
100 map.insert(kv.key.clone(), value);
101 }
102 for (k, v) in extras {
103 map.insert(k.clone(), v.clone());
104 }
105 JsonbValue::Object(map)
106}
107
108/// Serializes a flat `(key → value)` map (with optional extra entries layered on top)
109/// to JSONB bytes. Output ordering is alphabetical, matching `serialize_properties_to_jsonb`.
110pub fn attrs_to_jsonb(attrs: &[KeyValue], extras: &[(String, JsonbValue<'static>)]) -> Vec<u8> {
111 to_jsonb_bytes(attrs_to_jsonb_value(attrs, extras))
112}
113
114/// Renders `AnyValue` to a flat string for fields that need a textual form
115/// (e.g., the `msg` column when an OTel log body is structured).
116pub fn any_value_to_string(v: &AnyValue) -> String {
117 match v.value.as_ref() {
118 Some(Av::StringValue(s)) => s.clone(),
119 Some(Av::IntValue(i)) => i.to_string(),
120 Some(Av::DoubleValue(d)) => d.to_string(),
121 Some(Av::BoolValue(b)) => b.to_string(),
122 Some(Av::BytesValue(b)) => base64::engine::general_purpose::STANDARD.encode(b),
123 Some(Av::ArrayValue(arr)) => {
124 // Render via JSONB to keep round-trippable representations.
125 let bytes = to_jsonb_bytes(JsonbValue::Array(
126 arr.values.iter().map(any_value_to_jsonb).collect(),
127 ));
128 jsonb::RawJsonb::new(&bytes).to_string()
129 }
130 Some(Av::KvlistValue(kvs)) => {
131 let mut map: BTreeMap<String, JsonbValue<'static>> = BTreeMap::new();
132 for kv in &kvs.values {
133 let value = kv
134 .value
135 .as_ref()
136 .map(any_value_to_jsonb)
137 .unwrap_or(JsonbValue::Null);
138 map.insert(kv.key.clone(), value);
139 }
140 let bytes = to_jsonb_bytes(JsonbValue::Object(map));
141 jsonb::RawJsonb::new(&bytes).to_string()
142 }
143 // Profiling-only string-table reference: no dictionary exists for this signal, so the
144 // index has no meaning here. Treat as absent (per OTLP spec) — never as data.
145 Some(Av::StringValueStrindex(_)) => {
146 warn_unexpected_strindex();
147 String::new()
148 }
149 None => String::new(),
150 }
151}
152
153/// Maps OTel `severity_number` (1–24) to micromegas `Level` (1–6).
154///
155/// Per the plan:
156/// - TRACE 1–4 → 6
157/// - DEBUG 5–8 → 5
158/// - INFO 9–12 → 4
159/// - WARN 13–16 → 3
160/// - ERROR 17–20 → 2
161/// - FATAL 21–24 → 1
162///
163/// `severity_number = 0` (UNSPECIFIED) → 4 (Info), so the default
164/// `WHERE level <= 4` filter keeps them visible — the SDK didn't tell us they were
165/// low-priority, so we don't bury them. Out-of-range (negative or `> 24`) → 4 (Info)
166/// as well; promoting an unknown severity to Fatal would silently pollute alerting
167/// when a buggy SDK is off-by-one on the FATAL range.
168pub fn severity_number_to_level(sev: i32) -> i32 {
169 match sev {
170 1..=4 => 6, // TRACE
171 5..=8 => 5, // DEBUG
172 9..=12 => 4, // INFO
173 13..=16 => 3, // WARN
174 17..=20 => 2, // ERROR
175 21..=24 => 1, // FATAL
176 _ => 4, // UNSPECIFIED (0) or out-of-range → Info (don't fake-Fatal-alert)
177 }
178}
179
180/// Builds the per-row `otel.scope.*` properties (`name`, `version`, `attr.*`,
181/// `schema_url`) that ride alongside row attributes in the JSONB `properties`
182/// column. Skips empty fields so absent scopes don't pollute the output.
183pub fn scope_extras(
184 scope: Option<&InstrumentationScope>,
185 schema_url: &str,
186) -> Vec<(String, JsonbValue<'static>)> {
187 let mut extras: Vec<(String, JsonbValue<'static>)> = Vec::new();
188 if let Some(s) = scope {
189 if !s.name.is_empty() {
190 extras.push((
191 "otel.scope.name".to_string(),
192 JsonbValue::String(Cow::Owned(s.name.clone())),
193 ));
194 }
195 if !s.version.is_empty() {
196 extras.push((
197 "otel.scope.version".to_string(),
198 JsonbValue::String(Cow::Owned(s.version.clone())),
199 ));
200 }
201 for kv in &s.attributes {
202 if let Some(v) = kv.value.as_ref() {
203 extras.push((format!("otel.scope.attr.{}", kv.key), any_value_to_jsonb(v)));
204 }
205 }
206 }
207 if !schema_url.is_empty() {
208 extras.push((
209 "otel.scope.schema_url".to_string(),
210 JsonbValue::String(Cow::Owned(schema_url.to_string())),
211 ));
212 }
213 extras
214}