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/// Serializes a flat `(key → value)` map (with optional extra entries layered on top)
86/// to JSONB bytes. Output ordering is alphabetical, matching `serialize_properties_to_jsonb`.
87pub fn attrs_to_jsonb(attrs: &[KeyValue], extras: &[(String, JsonbValue<'static>)]) -> Vec<u8> {
88 let mut map: BTreeMap<String, JsonbValue<'static>> = BTreeMap::new();
89 for kv in attrs {
90 let value = kv
91 .value
92 .as_ref()
93 .map(any_value_to_jsonb)
94 .unwrap_or(JsonbValue::Null);
95 // `kv.key_strindex` (profiling-only) is intentionally ignored — see `any_value_to_jsonb`.
96 map.insert(kv.key.clone(), value);
97 }
98 for (k, v) in extras {
99 map.insert(k.clone(), v.clone());
100 }
101 to_jsonb_bytes(JsonbValue::Object(map))
102}
103
104/// Renders `AnyValue` to a flat string for fields that need a textual form
105/// (e.g., the `msg` column when an OTel log body is structured).
106pub fn any_value_to_string(v: &AnyValue) -> String {
107 match v.value.as_ref() {
108 Some(Av::StringValue(s)) => s.clone(),
109 Some(Av::IntValue(i)) => i.to_string(),
110 Some(Av::DoubleValue(d)) => d.to_string(),
111 Some(Av::BoolValue(b)) => b.to_string(),
112 Some(Av::BytesValue(b)) => base64::engine::general_purpose::STANDARD.encode(b),
113 Some(Av::ArrayValue(arr)) => {
114 // Render via JSONB to keep round-trippable representations.
115 let bytes = to_jsonb_bytes(JsonbValue::Array(
116 arr.values.iter().map(any_value_to_jsonb).collect(),
117 ));
118 jsonb::RawJsonb::new(&bytes).to_string()
119 }
120 Some(Av::KvlistValue(kvs)) => {
121 let mut map: BTreeMap<String, JsonbValue<'static>> = BTreeMap::new();
122 for kv in &kvs.values {
123 let value = kv
124 .value
125 .as_ref()
126 .map(any_value_to_jsonb)
127 .unwrap_or(JsonbValue::Null);
128 map.insert(kv.key.clone(), value);
129 }
130 let bytes = to_jsonb_bytes(JsonbValue::Object(map));
131 jsonb::RawJsonb::new(&bytes).to_string()
132 }
133 // Profiling-only string-table reference: no dictionary exists for this signal, so the
134 // index has no meaning here. Treat as absent (per OTLP spec) — never as data.
135 Some(Av::StringValueStrindex(_)) => {
136 warn_unexpected_strindex();
137 String::new()
138 }
139 None => String::new(),
140 }
141}
142
143/// Maps OTel `severity_number` (1–24) to micromegas `Level` (1–6).
144///
145/// Per the plan:
146/// - TRACE 1–4 → 6
147/// - DEBUG 5–8 → 5
148/// - INFO 9–12 → 4
149/// - WARN 13–16 → 3
150/// - ERROR 17–20 → 2
151/// - FATAL 21–24 → 1
152///
153/// `severity_number = 0` (UNSPECIFIED) → 4 (Info), so the default
154/// `WHERE level <= 4` filter keeps them visible — the SDK didn't tell us they were
155/// low-priority, so we don't bury them. Out-of-range (negative or `> 24`) → 4 (Info)
156/// as well; promoting an unknown severity to Fatal would silently pollute alerting
157/// when a buggy SDK is off-by-one on the FATAL range.
158pub fn severity_number_to_level(sev: i32) -> i32 {
159 match sev {
160 1..=4 => 6, // TRACE
161 5..=8 => 5, // DEBUG
162 9..=12 => 4, // INFO
163 13..=16 => 3, // WARN
164 17..=20 => 2, // ERROR
165 21..=24 => 1, // FATAL
166 _ => 4, // UNSPECIFIED (0) or out-of-range → Info (don't fake-Fatal-alert)
167 }
168}
169
170/// Builds the per-row `otel.scope.*` properties (`name`, `version`, `attr.*`,
171/// `schema_url`) that ride alongside row attributes in the JSONB `properties`
172/// column. Skips empty fields so absent scopes don't pollute the output.
173pub fn scope_extras(
174 scope: Option<&InstrumentationScope>,
175 schema_url: &str,
176) -> Vec<(String, JsonbValue<'static>)> {
177 let mut extras: Vec<(String, JsonbValue<'static>)> = Vec::new();
178 if let Some(s) = scope {
179 if !s.name.is_empty() {
180 extras.push((
181 "otel.scope.name".to_string(),
182 JsonbValue::String(Cow::Owned(s.name.clone())),
183 ));
184 }
185 if !s.version.is_empty() {
186 extras.push((
187 "otel.scope.version".to_string(),
188 JsonbValue::String(Cow::Owned(s.version.clone())),
189 ));
190 }
191 for kv in &s.attributes {
192 if let Some(v) = kv.value.as_ref() {
193 extras.push((format!("otel.scope.attr.{}", kv.key), any_value_to_jsonb(v)));
194 }
195 }
196 }
197 if !schema_url.is_empty() {
198 extras.push((
199 "otel.scope.schema_url".to_string(),
200 JsonbValue::String(Cow::Owned(schema_url.to_string())),
201 ));
202 }
203 extras
204}