1use anyhow::{bail, ensure, Context, Result};
3use opentelemetry_proto::tonic::{
4 collector::{
5 logs::v1::ExportLogsServiceRequest, metrics::v1::ExportMetricsServiceRequest,
6 trace::v1::ExportTraceServiceRequest,
7 },
8 common::v1::{any_value, AnyValue, InstrumentationScope, KeyValue},
9 logs::v1::{LogRecord, ResourceLogs, ScopeLogs},
10 metrics::v1::{
11 metric, number_data_point, Gauge, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics,
12 },
13 resource::v1::Resource,
14 trace::v1::{ResourceSpans, ScopeSpans, Span, Status},
15};
16use prost::Message;
17use serde_json::Value;
18pub const MAX_ITEMS: usize = 128;
19pub const MAX_ATTRIBUTES: usize = 32;
20pub const MAX_EXPORT_BYTES: usize = 1024 * 1024;
21#[derive(Clone, Debug)]
22pub enum Export {
23 Traces(ExportTraceServiceRequest),
24 Metrics(ExportMetricsServiceRequest),
25 Logs(ExportLogsServiceRequest),
26}
27impl Export {
28 pub fn signal(&self) -> crate::server::otlp::codec::Signal {
29 match self {
30 Self::Traces(_) => crate::server::otlp::codec::Signal::Traces,
31 Self::Metrics(_) => crate::server::otlp::codec::Signal::Metrics,
32 Self::Logs(_) => crate::server::otlp::codec::Signal::Logs,
33 }
34 }
35 pub fn encoded_len(&self) -> usize {
36 match self {
37 Self::Traces(v) => v.encoded_len(),
38 Self::Metrics(v) => v.encoded_len(),
39 Self::Logs(v) => v.encoded_len(),
40 }
41 }
42 pub fn encode(&self) -> Vec<u8> {
43 match self {
44 Self::Traces(v) => v.encode_to_vec(),
45 Self::Metrics(v) => v.encode_to_vec(),
46 Self::Logs(v) => v.encode_to_vec(),
47 }
48 }
49}
50#[derive(Default)]
51struct Budget(usize);
52impl Budget {
53 fn charge(&mut self, n: usize) -> Result<()> {
54 self.0 = self.0.checked_add(n).context("telemetry size overflow")?;
55 ensure!(self.0 <= MAX_EXPORT_BYTES, "typed telemetry exceeds 1 MiB");
56 Ok(())
57 }
58 fn text(&mut self, value: &Value, field: &str, max: usize, required: bool) -> Result<String> {
59 let Some(raw) = value.get(field) else {
60 ensure!(!required, "missing {field}");
61 return Ok(String::new());
62 };
63 let text = raw
64 .as_str()
65 .with_context(|| format!("{field} must be a string"))?;
66 ensure!(
67 text.len() <= max && (!required || !text.is_empty()),
68 "{field} must be {}..{max} bytes",
69 usize::from(required)
70 );
71 self.charge(text.len())?;
72 Ok(text.to_owned())
73 }
74 fn attributes(&mut self, value: Option<&Value>) -> Result<Vec<KeyValue>> {
75 let Some(value) = value else {
76 return Ok(Vec::new());
77 };
78 let attrs = value
79 .as_object()
80 .context("attributes must be a flat object")?;
81 ensure!(attrs.len() <= MAX_ATTRIBUTES, "at most 32 attributes");
82 let mut out = Vec::with_capacity(attrs.len());
83 for (key, value) in attrs {
84 ensure!(
85 !key.is_empty() && key.len() <= 256,
86 "attribute keys must be 1..256 bytes"
87 );
88 self.charge(key.len())?;
89 let value = match value {
90 Value::String(text) => {
91 ensure!(text.len() <= 1024, "attribute strings exceed 1024 bytes");
92 self.charge(text.len())?;
93 any_value::Value::StringValue(text.clone())
94 }
95 Value::Bool(value) => any_value::Value::BoolValue(*value),
96 Value::Number(value) => {
97 if let Some(n) = value.as_i64() {
98 any_value::Value::IntValue(n)
99 } else {
100 let n = value.as_f64().context("invalid attribute number")?;
101 ensure!(n.is_finite(), "attribute number must be finite");
102 ensure!(!value.is_u64(), "attribute integer exceeds i64");
103 any_value::Value::DoubleValue(n)
104 }
105 }
106 _ => bail!("attributes accept only string, bool, i64 or finite float"),
107 };
108 out.push(KeyValue {
109 key: key.clone(),
110 value: Some(AnyValue { value: Some(value) }),
111 });
112 }
113 Ok(out)
114 }
115}
116fn timestamp(v: &Value, key: &str) -> Result<u64> {
117 let n = v[key]
118 .as_u64()
119 .with_context(|| format!("{key} must be a positive u64 Unix timestamp in nanoseconds"))?;
120 ensure!(n > 0, "{key} must be positive");
121 Ok(n)
122}
123fn id(v: &Value, key: &str, bytes: usize, required: bool) -> Result<Vec<u8>> {
124 let Some(value) = v.get(key) else {
125 ensure!(!required, "missing {key}");
126 return Ok(Vec::new());
127 };
128 let value = value
129 .as_str()
130 .with_context(|| format!("{key} must be hex"))?;
131 ensure!(
132 value.len() == bytes * 2,
133 "{key} must be {} hex digits",
134 bytes * 2
135 );
136 let data = hex::decode(value).with_context(|| format!("invalid {key}"))?;
137 ensure!(data.iter().any(|b| *b != 0), "{key} cannot be all zero");
138 Ok(data)
139}
140fn items<'a>(v: &'a Value, key: &str) -> Result<&'a [Value]> {
141 let items = v[key]
142 .as_array()
143 .with_context(|| format!("{key} must be an array"))?;
144 ensure!(
145 !items.is_empty() && items.len() <= MAX_ITEMS,
146 "{key} must contain 1..128 items"
147 );
148 ensure!(
149 items.iter().all(Value::is_object),
150 "{key} items must be objects"
151 );
152 Ok(items)
153}
154pub fn build(action: &Value) -> Result<Export> {
155 let mut budget = Budget::default();
156 let service = budget.text(action, "service_name", 256, true)?;
157 let mut attrs = budget.attributes(action.get("resource_attributes"))?;
158 ensure!(
159 !attrs.iter().any(|a| a.key == "service.name"),
160 "use service_name rather than resource_attributes.service.name"
161 );
162 attrs.push(KeyValue {
163 key: "service.name".into(),
164 value: Some(AnyValue {
165 value: Some(any_value::Value::StringValue(service)),
166 }),
167 });
168 let resource = Some(Resource {
169 attributes: attrs,
170 ..Default::default()
171 });
172 let scope_name = budget.text(action, "scope_name", 256, false)?;
173 let scope = Some(InstrumentationScope {
174 name: if scope_name.is_empty() {
175 "netget".into()
176 } else {
177 scope_name
178 },
179 ..Default::default()
180 });
181 let export = match action["type"].as_str().context("missing type")? {
182 "export_otlp_traces" => {
183 let mut spans = Vec::new();
184 for v in items(action, "spans")? {
185 let start = timestamp(v, "start_time_unix_nano")?;
186 let end = timestamp(v, "end_time_unix_nano")?;
187 ensure!(end >= start, "span end must not precede start");
188 let kind = match v
189 .get("kind")
190 .map(|v| v.as_str().context("kind must be a string"))
191 .transpose()?
192 .unwrap_or("internal")
193 {
194 "internal" => 1,
195 "server" => 2,
196 "client" => 3,
197 "producer" => 4,
198 "consumer" => 5,
199 _ => bail!("invalid span kind"),
200 };
201 let status = match v
202 .get("status")
203 .map(|v| v.as_str().context("status must be a string"))
204 .transpose()?
205 .unwrap_or("unset")
206 {
207 "unset" => 0,
208 "ok" => 1,
209 "error" => 2,
210 _ => bail!("invalid span status"),
211 };
212 spans.push(Span {
213 trace_id: id(v, "trace_id", 16, true)?,
214 span_id: id(v, "span_id", 8, true)?,
215 parent_span_id: id(v, "parent_span_id", 8, false)?,
216 name: budget.text(v, "name", 256, true)?,
217 kind,
218 start_time_unix_nano: start,
219 end_time_unix_nano: end,
220 attributes: budget.attributes(v.get("attributes"))?,
221 status: Some(Status {
222 code: status,
223 message: budget.text(v, "status_message", 512, false)?,
224 }),
225 ..Default::default()
226 });
227 }
228 Export::Traces(ExportTraceServiceRequest {
229 resource_spans: vec![ResourceSpans {
230 resource,
231 scope_spans: vec![ScopeSpans {
232 scope,
233 spans,
234 ..Default::default()
235 }],
236 ..Default::default()
237 }],
238 })
239 }
240 "export_otlp_logs" => {
241 let mut log_records = Vec::new();
242 for v in items(action, "logs")? {
243 let severity = v
244 .get("severity_number")
245 .map(|v| v.as_u64().context("severity_number must be 0..24"))
246 .transpose()?
247 .unwrap_or(9);
248 ensure!(severity <= 24, "severity_number must be 0..24");
249 ensure!(
250 v.get("trace_id").is_some() == v.get("span_id").is_some(),
251 "log trace_id and span_id must be supplied together"
252 );
253 log_records.push(LogRecord {
254 time_unix_nano: timestamp(v, "time_unix_nano")?,
255 severity_number: severity as i32,
256 severity_text: budget.text(v, "severity_text", 64, false)?,
257 body: Some(AnyValue {
258 value: Some(any_value::Value::StringValue(
259 budget.text(v, "body", 4096, true)?,
260 )),
261 }),
262 attributes: budget.attributes(v.get("attributes"))?,
263 trace_id: id(v, "trace_id", 16, false)?,
264 span_id: id(v, "span_id", 8, false)?,
265 ..Default::default()
266 });
267 }
268 Export::Logs(ExportLogsServiceRequest {
269 resource_logs: vec![ResourceLogs {
270 resource,
271 scope_logs: vec![ScopeLogs {
272 scope,
273 log_records,
274 ..Default::default()
275 }],
276 ..Default::default()
277 }],
278 })
279 }
280 "export_otlp_gauge" => {
281 let mut data_points = Vec::new();
282 for v in items(action, "data_points")? {
283 let number = v["value"]
284 .as_number()
285 .context("gauge value must be a number")?;
286 let value = if let Some(n) = number.as_i64() {
287 number_data_point::Value::AsInt(n)
288 } else {
289 ensure!(!number.is_u64(), "gauge integer exceeds i64");
290 let n = number.as_f64().context("invalid gauge value")?;
291 ensure!(n.is_finite(), "gauge value must be finite");
292 number_data_point::Value::AsDouble(n)
293 };
294 data_points.push(NumberDataPoint {
295 time_unix_nano: timestamp(v, "time_unix_nano")?,
296 attributes: budget.attributes(v.get("attributes"))?,
297 value: Some(value),
298 ..Default::default()
299 });
300 }
301 let metric = Metric {
302 name: budget.text(action, "name", 256, true)?,
303 description: budget.text(action, "description", 256, false)?,
304 unit: budget.text(action, "unit", 64, false)?,
305 data: Some(metric::Data::Gauge(Gauge { data_points })),
306 ..Default::default()
307 };
308 Export::Metrics(ExportMetricsServiceRequest {
309 resource_metrics: vec![ResourceMetrics {
310 resource,
311 scope_metrics: vec![ScopeMetrics {
312 scope,
313 metrics: vec![metric],
314 ..Default::default()
315 }],
316 ..Default::default()
317 }],
318 })
319 }
320 _ => bail!("unknown OTLP export action"),
321 };
322 ensure!(
323 export.encoded_len() <= MAX_EXPORT_BYTES,
324 "encoded export exceeds 1 MiB"
325 );
326 Ok(export)
327}