span-sink-otlp.js
2.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
/**
* OTLP/HTTP sink —— 把 `CompletedSpan` 编码为 OTLP JSON 并经注入 `fetchImpl` 发一次 POST。
*
* body 结构：`resourceSpans[].scopeSpans[].spans[]`（OTLP/HTTP JSON 编码）。
* 边界纪律：`fetchImpl` 注入，包内无全局 `fetch`、不持有连接、不读 env。
*/
/**
* @typedef {import('./span.js').CompletedSpan} CompletedSpan
* @typedef {import('./span.js').Sink} Sink
*/
/**
* JS 值 → OTLP `AnyValue`。整数取 `intValue`（字符串编码），非整数取 `doubleValue`。
* @param {any} v
* @returns {object}
*/
function toAnyValue(v) {
if (typeof v === 'string') return { stringValue: v }
if (typeof v === 'boolean') return { boolValue: v }
if (Number.isInteger(v)) return { intValue: String(v) }
return { doubleValue: v }
}
/**
* 对象 → OTLP `KeyValue[]`。
* @param {Record<string, any>} obj
* @returns {Array<{ key: string, value: object }>}
*/
function toKeyValues(obj) {
return Object.entries(obj).map(([key, v]) => ({ key, value: toAnyValue(v) }))
}
/**
* `CompletedSpan` → OTLP span。事件时间统一取 span 结束时刻（本包不逐事件采时钟）。
* @param {CompletedSpan} span
* @returns {object}
*/
function toOtlpSpan(span) {
const timeUnixNano = String(span.endUnixNano)
return {
name: span.name,
startTimeUnixNano: String(span.startUnixNano),
endTimeUnixNano: timeUnixNano,
attributes: toKeyValues(span.attributes),
events: span.events.map((e) => ({
name: e.name,
timeUnixNano,
attributes: toKeyValues(e.attributes),
})),
}
}
/**
* OTLP/HTTP sink 工厂。
* @param {object} deps
* @param {typeof fetch} deps.fetchImpl 注入的 fetch。
* @param {string} deps.endpoint OTLP traces 端点（如 http://otel:4318/v1/traces）。
* @param {Record<string, string>} [deps.headers={}] 附加请求头（如鉴权）。
* @param {Record<string, any>} [deps.resourceAttributes={}] 资源属性（如 service.name）。
* @returns {Sink}
*/
export function createOtlpSink({ fetchImpl, endpoint, headers = {}, resourceAttributes = {} }) {
return {
write(span) {
const body = {
resourceSpans: [
{
resource: { attributes: toKeyValues(resourceAttributes) },
scopeSpans: [
{
scope: { name: '@drama/observability' },
spans: [toOtlpSpan(span)],
},
],
},
],
}
fetchImpl(endpoint, {
method: 'POST',
headers: { 'content-type': 'application/json', ...headers },
body: JSON.stringify(body),
})
},
}
}