Skip to content
Sign up

flagon-io/g1t

public
0
451 linesCodeBlameRaw
1/**
2 * OTLP over HTTP in protobuf (`application/x-protobuf`), the encoding most
3 * OpenTelemetry SDKs send by default, read into the shape OTLP/JSON has
4 * (opentelemetry-proto's JSON mapping: ids as hex, 64-bit integers and
5 * nanosecond times as decimal strings, enums as numbers). Both encodings
6 * then go through one reader (./normalize.ts).
7 *
8 * Only the messages of the three export requests are read; fields this
9 * does not know are skipped, as protobuf allows, so a newer SDK still
10 * gets through. No code generation, no dependency: a protobuf message is
11 * a list of (field number, wire type, value), and each reader below picks
12 * the fields it wants.
13 */
14
15/** A value read off the wire, before it is known what the field means. */
16type Field =
17 | { no: number; wire: 0; value: bigint }
18 | { no: number; wire: 1; value: Uint8Array }
19 | { no: number; wire: 2; value: Uint8Array }
20 | { no: number; wire: 5; value: Uint8Array };
21
22export class ProtoError extends Error {}
23
24const utf8 = new TextDecoder("utf-8");
25
26/** Splits one message into its fields, in order. */
27export function fields(bytes: Uint8Array): Field[] {
28 const out: Field[] = [];
29 let at = 0;
30 const varint = (): bigint => {
31 let result = 0n;
32 let shift = 0n;
33 for (;;) {
34 if (at >= bytes.length) throw new ProtoError("The message ends inside a number.");
35 const byte = bytes[at++];
36 result |= BigInt(byte & 0x7f) << shift;
37 if ((byte & 0x80) === 0) return result;
38 shift += 7n;
39 if (shift > 70n) throw new ProtoError("A number is longer than ten bytes.");
40 }
41 };
42 const take = (n: number): Uint8Array => {
43 if (n < 0 || at + n > bytes.length) throw new ProtoError("A field runs past the end of the message.");
44 const slice = bytes.subarray(at, at + n);
45 at += n;
46 return slice;
47 };
48 while (at < bytes.length) {
49 const key = Number(varint());
50 const no = key >>> 3;
51 const wire = key & 7;
52 if (no === 0) throw new ProtoError("A field numbered 0.");
53 switch (wire) {
54 case 0:
55 out.push({ no, wire, value: varint() });
56 break;
57 case 1:
58 out.push({ no, wire, value: take(8) });
59 break;
60 case 2:
61 out.push({ no, wire, value: take(Number(varint())) });
62 break;
63 case 5:
64 out.push({ no, wire, value: take(4) });
65 break;
66 default:
67 throw new ProtoError(`Wire type ${wire} is not one OTLP uses.`);
68 }
69 }
70 return out;
71}
72
73// ---- Scalars -------------------------------------------------------------
74
75function view(b: Uint8Array): DataView {
76 return new DataView(b.buffer, b.byteOffset, b.byteLength);
77}
78
79const fixed64 = (b: Uint8Array): bigint => view(b).getBigUint64(0, true);
80const sfixed64 = (b: Uint8Array): bigint => view(b).getBigInt64(0, true);
81const double = (b: Uint8Array): number => view(b).getFloat64(0, true);
82const str = (b: Uint8Array): string => utf8.decode(b);
83const hex = (b: Uint8Array): string => [...b].map((x) => x.toString(16).padStart(2, "0")).join("");
84/** A varint as a signed 64-bit two's-complement number (int64). */
85const int64 = (v: bigint): bigint => BigInt.asIntN(64, v);
86
87/** Base64, for `bytes` values, as the JSON mapping writes them. */
88function base64(b: Uint8Array): string {
89 let s = "";
90 for (const x of b) s += String.fromCharCode(x);
91 return btoa(s);
92}
93
94/** A repeated fixed64 or double field, packed (wire type 2) or not. */
95function packed64(list: Field[], no: number, read: (b: Uint8Array) => number | bigint): (number | bigint)[] {
96 const out: (number | bigint)[] = [];
97 for (const f of list) {
98 if (f.no !== no) continue;
99 if (f.wire === 1) out.push(read(f.value));
100 else if (f.wire === 2) for (let i = 0; i + 8 <= f.value.length; i += 8) out.push(read(f.value.subarray(i, i + 8)));
101 }
102 return out;
103}
104
105// ---- Common --------------------------------------------------------------
106
107export type AnyValueJson =
108 | { stringValue: string }
109 | { boolValue: boolean }
110 | { intValue: string }
111 | { doubleValue: number }
112 | { arrayValue: { values: AnyValueJson[] } }
113 | { kvlistValue: { values: KeyValueJson[] } }
114 | { bytesValue: string }
115 | Record<string, never>;
116
117export type KeyValueJson = { key: string; value?: AnyValueJson };
118
119function anyValue(b: Uint8Array): AnyValueJson {
120 for (const f of fields(b)) {
121 if (f.no === 1 && f.wire === 2) return { stringValue: str(f.value) };
122 if (f.no === 2 && f.wire === 0) return { boolValue: f.value !== 0n };
123 if (f.no === 3 && f.wire === 0) return { intValue: int64(f.value).toString() };
124 if (f.no === 4 && f.wire === 1) return { doubleValue: double(f.value) };
125 if (f.no === 5 && f.wire === 2) return { arrayValue: { values: repeated(fields(f.value), 1, anyValue) } };
126 if (f.no === 6 && f.wire === 2) return { kvlistValue: { values: repeated(fields(f.value), 1, keyValue) } };
127 if (f.no === 7 && f.wire === 2) return { bytesValue: base64(f.value) };
128 }
129 return {};
130}
131
132function keyValue(b: Uint8Array): KeyValueJson {
133 const kv: KeyValueJson = { key: "" };
134 for (const f of fields(b)) {
135 if (f.no === 1 && f.wire === 2) kv.key = str(f.value);
136 else if (f.no === 2 && f.wire === 2) kv.value = anyValue(f.value);
137 }
138 return kv;
139}
140
141function repeated<T>(list: Field[], no: number, read: (b: Uint8Array) => T): T[] {
142 const out: T[] = [];
143 for (const f of list) if (f.no === no && f.wire === 2) out.push(read(f.value));
144 return out;
145}
146
147function one<T>(list: Field[], no: number, read: (b: Uint8Array) => T): T | undefined {
148 let found: T | undefined;
149 for (const f of list) if (f.no === no && f.wire === 2) found = read(f.value);
150 return found;
151}
152
153function string(list: Field[], no: number): string | undefined {
154 return one(list, no, str);
155}
156
157function varintField(list: Field[], no: number): bigint | undefined {
158 let found: bigint | undefined;
159 for (const f of list) if (f.no === no && f.wire === 0) found = f.value;
160 return found;
161}
162
163function fixed64Field(list: Field[], no: number): string | undefined {
164 let found: string | undefined;
165 for (const f of list) if (f.no === no && f.wire === 1) found = fixed64(f.value).toString();
166 return found;
167}
168
169function doubleField(list: Field[], no: number): number | undefined {
170 let found: number | undefined;
171 for (const f of list) if (f.no === no && f.wire === 1) found = double(f.value);
172 return found;
173}
174
175function bytesHex(list: Field[], no: number): string | undefined {
176 return one(list, no, hex);
177}
178
179export type ResourceJson = { attributes?: KeyValueJson[] };
180export type ScopeJson = { name?: string; version?: string; attributes?: KeyValueJson[] };
181
182function resource(b: Uint8Array): ResourceJson {
183 return { attributes: repeated(fields(b), 1, keyValue) };
184}
185
186function scope(b: Uint8Array): ScopeJson {
187 const f = fields(b);
188 return { name: string(f, 1), version: string(f, 2), attributes: repeated(f, 3, keyValue) };
189}
190
191// ---- Traces --------------------------------------------------------------
192
193export type SpanJson = {
194 traceId?: string;
195 spanId?: string;
196 traceState?: string;
197 parentSpanId?: string;
198 name?: string;
199 kind?: number;
200 startTimeUnixNano?: string;
201 endTimeUnixNano?: string;
202 attributes?: KeyValueJson[];
203 events?: { timeUnixNano?: string; name?: string; attributes?: KeyValueJson[] }[];
204 links?: { traceId?: string; spanId?: string; attributes?: KeyValueJson[] }[];
205 status?: { message?: string; code?: number };
206};
207
208export type TracesRequestJson = {
209 resourceSpans?: { resource?: ResourceJson; scopeSpans?: { scope?: ScopeJson; spans?: SpanJson[] }[] }[];
210};
211
212function span(b: Uint8Array): SpanJson {
213 const f = fields(b);
214 const kind = varintField(f, 6);
215 return {
216 traceId: bytesHex(f, 1),
217 spanId: bytesHex(f, 2),
218 traceState: string(f, 3),
219 parentSpanId: bytesHex(f, 4),
220 name: string(f, 5),
221 kind: kind === undefined ? undefined : Number(kind),
222 startTimeUnixNano: fixed64Field(f, 7),
223 endTimeUnixNano: fixed64Field(f, 8),
224 attributes: repeated(f, 9, keyValue),
225 events: repeated(f, 11, (e) => {
226 const ef = fields(e);
227 return { timeUnixNano: fixed64Field(ef, 1), name: string(ef, 2), attributes: repeated(ef, 3, keyValue) };
228 }),
229 links: repeated(f, 13, (l) => {
230 const lf = fields(l);
231 return { traceId: bytesHex(lf, 1), spanId: bytesHex(lf, 2), attributes: repeated(lf, 4, keyValue) };
232 }),
233 status: one(f, 15, (s) => {
234 const sf = fields(s);
235 const code = varintField(sf, 3);
236 return { message: string(sf, 2), code: code === undefined ? undefined : Number(code) };
237 }),
238 };
239}
240
241export function decodeTraces(bytes: Uint8Array): TracesRequestJson {
242 return {
243 resourceSpans: repeated(fields(bytes), 1, (rs) => {
244 const f = fields(rs);
245 return {
246 resource: one(f, 1, resource),
247 scopeSpans: repeated(f, 2, (ss) => {
248 const sf = fields(ss);
249 return { scope: one(sf, 1, scope), spans: repeated(sf, 2, span) };
250 }),
251 };
252 }),
253 };
254}
255
256// ---- Logs ----------------------------------------------------------------
257
258export type LogRecordJson = {
259 timeUnixNano?: string;
260 observedTimeUnixNano?: string;
261 severityNumber?: number;
262 severityText?: string;
263 body?: AnyValueJson;
264 attributes?: KeyValueJson[];
265 traceId?: string;
266 spanId?: string;
267 eventName?: string;
268};
269
270export type LogsRequestJson = {
271 resourceLogs?: { resource?: ResourceJson; scopeLogs?: { scope?: ScopeJson; logRecords?: LogRecordJson[] }[] }[];
272};
273
274function logRecord(b: Uint8Array): LogRecordJson {
275 const f = fields(b);
276 const severity = varintField(f, 2);
277 return {
278 timeUnixNano: fixed64Field(f, 1),
279 observedTimeUnixNano: fixed64Field(f, 11),
280 severityNumber: severity === undefined ? undefined : Number(severity),
281 severityText: string(f, 3),
282 body: one(f, 5, anyValue),
283 attributes: repeated(f, 6, keyValue),
284 traceId: bytesHex(f, 9),
285 spanId: bytesHex(f, 10),
286 eventName: string(f, 12),
287 };
288}
289
290export function decodeLogs(bytes: Uint8Array): LogsRequestJson {
291 return {
292 resourceLogs: repeated(fields(bytes), 1, (rl) => {
293 const f = fields(rl);
294 return {
295 resource: one(f, 1, resource),
296 scopeLogs: repeated(f, 2, (sl) => {
297 const sf = fields(sl);
298 return { scope: one(sf, 1, scope), logRecords: repeated(sf, 2, logRecord) };
299 }),
300 };
301 }),
302 };
303}
304
305// ---- Metrics -------------------------------------------------------------
306
307export type NumberPointJson = {
308 attributes?: KeyValueJson[];
309 startTimeUnixNano?: string;
310 timeUnixNano?: string;
311 asDouble?: number;
312 asInt?: string;
313};
314
315export type HistogramPointJson = {
316 attributes?: KeyValueJson[];
317 startTimeUnixNano?: string;
318 timeUnixNano?: string;
319 count?: string;
320 sum?: number;
321 bucketCounts?: string[];
322 explicitBounds?: number[];
323 min?: number;
324 max?: number;
325};
326
327export type SummaryPointJson = {
328 attributes?: KeyValueJson[];
329 startTimeUnixNano?: string;
330 timeUnixNano?: string;
331 count?: string;
332 sum?: number;
333 quantileValues?: { quantile?: number; value?: number }[];
334};
335
336export type MetricJson = {
337 name?: string;
338 description?: string;
339 unit?: string;
340 gauge?: { dataPoints?: NumberPointJson[] };
341 sum?: { dataPoints?: NumberPointJson[]; aggregationTemporality?: number; isMonotonic?: boolean };
342 histogram?: { dataPoints?: HistogramPointJson[]; aggregationTemporality?: number };
343 /** Read as its count, sum, min and max: the buckets are not kept. */
344 exponentialHistogram?: { dataPoints?: HistogramPointJson[]; aggregationTemporality?: number };
345 summary?: { dataPoints?: SummaryPointJson[] };
346};
347
348export type MetricsRequestJson = {
349 resourceMetrics?: { resource?: ResourceJson; scopeMetrics?: { scope?: ScopeJson; metrics?: MetricJson[] }[] }[];
350};
351
352function numberPoint(b: Uint8Array): NumberPointJson {
353 const f = fields(b);
354 const point: NumberPointJson = {
355 attributes: repeated(f, 7, keyValue),
356 startTimeUnixNano: fixed64Field(f, 2),
357 timeUnixNano: fixed64Field(f, 3),
358 };
359 for (const x of f) {
360 if (x.no === 4 && x.wire === 1) point.asDouble = double(x.value);
361 if (x.no === 6 && x.wire === 1) point.asInt = sfixed64(x.value).toString();
362 }
363 return point;
364}
365
366function histogramPoint(b: Uint8Array): HistogramPointJson {
367 const f = fields(b);
368 return {
369 attributes: repeated(f, 9, keyValue),
370 startTimeUnixNano: fixed64Field(f, 2),
371 timeUnixNano: fixed64Field(f, 3),
372 count: fixed64Field(f, 4),
373 sum: doubleField(f, 5),
374 bucketCounts: packed64(f, 6, fixed64).map(String),
375 explicitBounds: packed64(f, 7, double) as number[],
376 min: doubleField(f, 11),
377 max: doubleField(f, 12),
378 };
379}
380
381function exponentialPoint(b: Uint8Array): HistogramPointJson {
382 const f = fields(b);
383 return {
384 attributes: repeated(f, 1, keyValue),
385 startTimeUnixNano: fixed64Field(f, 2),
386 timeUnixNano: fixed64Field(f, 3),
387 count: fixed64Field(f, 4),
388 sum: doubleField(f, 5),
389 min: doubleField(f, 12),
390 max: doubleField(f, 13),
391 };
392}
393
394function summaryPoint(b: Uint8Array): SummaryPointJson {
395 const f = fields(b);
396 return {
397 attributes: repeated(f, 7, keyValue),
398 startTimeUnixNano: fixed64Field(f, 2),
399 timeUnixNano: fixed64Field(f, 3),
400 count: fixed64Field(f, 4),
401 sum: doubleField(f, 5),
402 quantileValues: repeated(f, 6, (q) => {
403 const qf = fields(q);
404 return { quantile: doubleField(qf, 1), value: doubleField(qf, 2) };
405 }),
406 };
407}
408
409function metric(b: Uint8Array): MetricJson {
410 const f = fields(b);
411 const out: MetricJson = { name: string(f, 1), description: string(f, 2), unit: string(f, 3) };
412 const temporality = (list: Field[]) => {
413 const t = varintField(list, 2);
414 return t === undefined ? undefined : Number(t);
415 };
416 const g = one(f, 5, (x) => x);
417 if (g) out.gauge = { dataPoints: repeated(fields(g), 1, numberPoint) };
418 const s = one(f, 7, (x) => x);
419 if (s) {
420 const sf = fields(s);
421 out.sum = { dataPoints: repeated(sf, 1, numberPoint), aggregationTemporality: temporality(sf), isMonotonic: varintField(sf, 3) === 1n };
422 }
423 const h = one(f, 9, (x) => x);
424 if (h) {
425 const hf = fields(h);
426 out.histogram = { dataPoints: repeated(hf, 1, histogramPoint), aggregationTemporality: temporality(hf) };
427 }
428 const e = one(f, 10, (x) => x);
429 if (e) {
430 const ef = fields(e);
431 out.exponentialHistogram = { dataPoints: repeated(ef, 1, exponentialPoint), aggregationTemporality: temporality(ef) };
432 }
433 const q = one(f, 11, (x) => x);
434 if (q) out.summary = { dataPoints: repeated(fields(q), 1, summaryPoint) };
435 return out;
436}
437
438export function decodeMetrics(bytes: Uint8Array): MetricsRequestJson {
439 return {
440 resourceMetrics: repeated(fields(bytes), 1, (rm) => {
441 const f = fields(rm);
442 return {
443 resource: one(f, 1, resource),
444 scopeMetrics: repeated(f, 2, (sm) => {
445 const sf = fields(sm);
446 return { scope: one(sf, 1, scope), metrics: repeated(sf, 2, metric) };
447 }),
448 };
449 }),
450 };
451}