Skip to content

Commit bfe3ac6

Browse files
platypiiplatypii
committed
Add metric decoders and tests
1 parent 24d7e99 commit bfe3ac6

6 files changed

Lines changed: 749 additions & 34 deletions

File tree

‎src/otlp/datapoints.js‎

Lines changed: 291 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,291 @@
1+
/**
2+
* Decoders for OTLP metric data-point message types: NumberDataPoint,
3+
* HistogramDataPoint, ExponentialHistogramDataPoint (+ Buckets),
4+
* SummaryDataPoint (+ ValueAtQuantile), and Exemplar. Also holds the
5+
* packed-scalar readers these data points need.
6+
*
7+
* Repeated scalar fields (bucket_counts, explicit_bounds) accept both
8+
* the proto3-default packed LEN form and the legacy unpacked form.
9+
*/
10+
11+
import {
12+
WIRE_LEN,
13+
readBytes,
14+
readDouble,
15+
readFixed32,
16+
readFixed64,
17+
readSFixed64,
18+
readTag,
19+
readVarBigInt,
20+
readVarint,
21+
skipField,
22+
zigzagDecode,
23+
} from '../protobuf.js'
24+
import { decodeKeyValue, makeReader, toHex } from './common.js'
25+
26+
/**
27+
* Iterate a packed LEN blob of fixed64 values, returning each as a string.
28+
*
29+
* @param {Uint8Array} bytes
30+
* @returns {string[]}
31+
*/
32+
function readPackedFixed64Strings(bytes) {
33+
const r = makeReader(bytes)
34+
/** @type {string[]} */
35+
const out = []
36+
while (r.offset < bytes.byteLength) out.push(readFixed64(r).toString())
37+
return out
38+
}
39+
40+
/**
41+
* @param {Uint8Array} bytes
42+
* @returns {number[]}
43+
*/
44+
function readPackedDoubles(bytes) {
45+
const r = makeReader(bytes)
46+
/** @type {number[]} */
47+
const out = []
48+
while (r.offset < bytes.byteLength) out.push(readDouble(r))
49+
return out
50+
}
51+
52+
/**
53+
* @param {Uint8Array} bytes
54+
* @returns {string[]}
55+
*/
56+
function readPackedVarBigIntStrings(bytes) {
57+
const r = makeReader(bytes)
58+
/** @type {string[]} */
59+
const out = []
60+
while (r.offset < bytes.byteLength) out.push(readVarBigInt(r).toString())
61+
return out
62+
}
63+
64+
/**
65+
* @param {Uint8Array} bytes
66+
* @returns {object}
67+
*/
68+
export function decodeExemplar(bytes) {
69+
const r = makeReader(bytes)
70+
const end = bytes.byteLength
71+
/** @type {Record<string, unknown>} */
72+
const out = {}
73+
/** @type {object[]} */
74+
const filteredAttributes = []
75+
while (r.offset < end) {
76+
const { fieldNumber, wireType } = readTag(r)
77+
switch (fieldNumber) {
78+
case 2: out.timeUnixNano = readFixed64(r).toString(); break
79+
case 3: out.asDouble = readDouble(r); break
80+
case 4: out.spanId = toHex(readBytes(r)); break
81+
case 5: out.traceId = toHex(readBytes(r)); break
82+
case 6: out.asInt = readSFixed64(r).toString(); break
83+
case 7: filteredAttributes.push(decodeKeyValue(readBytes(r))); break
84+
default: skipField(r, wireType)
85+
}
86+
}
87+
if (filteredAttributes.length) out.filteredAttributes = filteredAttributes
88+
return out
89+
}
90+
91+
/**
92+
* @param {Uint8Array} bytes
93+
* @returns {object}
94+
*/
95+
export function decodeNumberDataPoint(bytes) {
96+
const r = makeReader(bytes)
97+
const end = bytes.byteLength
98+
/** @type {Record<string, unknown>} */
99+
const out = {}
100+
/** @type {object[]} */
101+
const attributes = []
102+
/** @type {object[]} */
103+
const exemplars = []
104+
while (r.offset < end) {
105+
const { fieldNumber, wireType } = readTag(r)
106+
switch (fieldNumber) {
107+
case 2: out.startTimeUnixNano = readFixed64(r).toString(); break
108+
case 3: out.timeUnixNano = readFixed64(r).toString(); break
109+
case 4: out.asDouble = readDouble(r); break
110+
case 5: exemplars.push(decodeExemplar(readBytes(r))); break
111+
case 6: out.asInt = readSFixed64(r).toString(); break
112+
case 7: attributes.push(decodeKeyValue(readBytes(r))); break
113+
case 8: out.flags = readFixed32(r); break
114+
default: skipField(r, wireType)
115+
}
116+
}
117+
if (attributes.length) out.attributes = attributes
118+
if (exemplars.length) out.exemplars = exemplars
119+
return out
120+
}
121+
122+
/**
123+
* @param {Uint8Array} bytes
124+
* @returns {object}
125+
*/
126+
export function decodeHistogramDataPoint(bytes) {
127+
const r = makeReader(bytes)
128+
const end = bytes.byteLength
129+
/** @type {Record<string, unknown>} */
130+
const out = {}
131+
/** @type {object[]} */
132+
const attributes = []
133+
/** @type {string[]} */
134+
const bucketCounts = []
135+
/** @type {number[]} */
136+
const explicitBounds = []
137+
/** @type {object[]} */
138+
const exemplars = []
139+
while (r.offset < end) {
140+
const { fieldNumber, wireType } = readTag(r)
141+
switch (fieldNumber) {
142+
case 9: attributes.push(decodeKeyValue(readBytes(r))); break
143+
case 2: out.startTimeUnixNano = readFixed64(r).toString(); break
144+
case 3: out.timeUnixNano = readFixed64(r).toString(); break
145+
case 4: out.count = readFixed64(r).toString(); break
146+
case 5: out.sum = readDouble(r); break
147+
case 6:
148+
if (wireType === WIRE_LEN) {
149+
for (const v of readPackedFixed64Strings(readBytes(r))) bucketCounts.push(v)
150+
} else {
151+
bucketCounts.push(readFixed64(r).toString())
152+
}
153+
break
154+
case 7:
155+
if (wireType === WIRE_LEN) {
156+
for (const v of readPackedDoubles(readBytes(r))) explicitBounds.push(v)
157+
} else {
158+
explicitBounds.push(readDouble(r))
159+
}
160+
break
161+
case 8: exemplars.push(decodeExemplar(readBytes(r))); break
162+
case 10: out.flags = readFixed32(r); break
163+
case 11: out.min = readDouble(r); break
164+
case 12: out.max = readDouble(r); break
165+
default: skipField(r, wireType)
166+
}
167+
}
168+
if (attributes.length) out.attributes = attributes
169+
if (bucketCounts.length) out.bucketCounts = bucketCounts
170+
if (explicitBounds.length) out.explicitBounds = explicitBounds
171+
if (exemplars.length) out.exemplars = exemplars
172+
return out
173+
}
174+
175+
/**
176+
* @param {Uint8Array} bytes
177+
* @returns {object}
178+
*/
179+
export function decodeBuckets(bytes) {
180+
const r = makeReader(bytes)
181+
const end = bytes.byteLength
182+
/** @type {Record<string, unknown>} */
183+
const out = {}
184+
/** @type {string[]} */
185+
const bucketCounts = []
186+
while (r.offset < end) {
187+
const { fieldNumber, wireType } = readTag(r)
188+
switch (fieldNumber) {
189+
case 1: out.offset = zigzagDecode(readVarint(r)); break
190+
case 2:
191+
if (wireType === WIRE_LEN) {
192+
for (const v of readPackedVarBigIntStrings(readBytes(r))) bucketCounts.push(v)
193+
} else {
194+
bucketCounts.push(readVarBigInt(r).toString())
195+
}
196+
break
197+
default: skipField(r, wireType)
198+
}
199+
}
200+
if (bucketCounts.length) out.bucketCounts = bucketCounts
201+
return out
202+
}
203+
204+
/**
205+
* @param {Uint8Array} bytes
206+
* @returns {object}
207+
*/
208+
export function decodeExponentialHistogramDataPoint(bytes) {
209+
const r = makeReader(bytes)
210+
const end = bytes.byteLength
211+
/** @type {Record<string, unknown>} */
212+
const out = {}
213+
/** @type {object[]} */
214+
const attributes = []
215+
/** @type {object[]} */
216+
const exemplars = []
217+
while (r.offset < end) {
218+
const { fieldNumber, wireType } = readTag(r)
219+
switch (fieldNumber) {
220+
case 1: attributes.push(decodeKeyValue(readBytes(r))); break
221+
case 2: out.startTimeUnixNano = readFixed64(r).toString(); break
222+
case 3: out.timeUnixNano = readFixed64(r).toString(); break
223+
case 4: out.count = readFixed64(r).toString(); break
224+
case 5: out.sum = readDouble(r); break
225+
case 6: out.scale = zigzagDecode(readVarint(r)); break
226+
case 7: out.zeroCount = readFixed64(r).toString(); break
227+
case 8: out.positive = decodeBuckets(readBytes(r)); break
228+
case 9: out.negative = decodeBuckets(readBytes(r)); break
229+
case 10: out.flags = readFixed32(r); break
230+
case 11: exemplars.push(decodeExemplar(readBytes(r))); break
231+
case 12: out.min = readDouble(r); break
232+
case 13: out.max = readDouble(r); break
233+
case 14: out.zeroThreshold = readDouble(r); break
234+
default: skipField(r, wireType)
235+
}
236+
}
237+
if (attributes.length) out.attributes = attributes
238+
if (exemplars.length) out.exemplars = exemplars
239+
return out
240+
}
241+
242+
/**
243+
* @param {Uint8Array} bytes
244+
* @returns {object}
245+
*/
246+
export function decodeValueAtQuantile(bytes) {
247+
const r = makeReader(bytes)
248+
const end = bytes.byteLength
249+
/** @type {Record<string, unknown>} */
250+
const out = {}
251+
while (r.offset < end) {
252+
const { fieldNumber, wireType } = readTag(r)
253+
switch (fieldNumber) {
254+
case 1: out.quantile = readDouble(r); break
255+
case 2: out.value = readDouble(r); break
256+
default: skipField(r, wireType)
257+
}
258+
}
259+
return out
260+
}
261+
262+
/**
263+
* @param {Uint8Array} bytes
264+
* @returns {object}
265+
*/
266+
export function decodeSummaryDataPoint(bytes) {
267+
const r = makeReader(bytes)
268+
const end = bytes.byteLength
269+
/** @type {Record<string, unknown>} */
270+
const out = {}
271+
/** @type {object[]} */
272+
const attributes = []
273+
/** @type {object[]} */
274+
const quantileValues = []
275+
while (r.offset < end) {
276+
const { fieldNumber, wireType } = readTag(r)
277+
switch (fieldNumber) {
278+
case 7: attributes.push(decodeKeyValue(readBytes(r))); break
279+
case 2: out.startTimeUnixNano = readFixed64(r).toString(); break
280+
case 3: out.timeUnixNano = readFixed64(r).toString(); break
281+
case 4: out.count = readFixed64(r).toString(); break
282+
case 5: out.sum = readDouble(r); break
283+
case 6: quantileValues.push(decodeValueAtQuantile(readBytes(r))); break
284+
case 8: out.flags = readFixed32(r); break
285+
default: skipField(r, wireType)
286+
}
287+
}
288+
if (attributes.length) out.attributes = attributes
289+
if (quantileValues.length) out.quantileValues = quantileValues
290+
return out
291+
}

0 commit comments

Comments
 (0)