| 1 | /** |
| 2 | * OTLP export requests (in their JSON shape, which ./proto.ts also gives |
| 3 | * for protobuf) as rows for a project's store. |
| 4 | * |
| 5 | * Each record is filed by its resource: `service.name` names the service, |
| 6 | * `deployment.environment.name` (or the older `deployment.environment`) |
| 7 | * the environment, `g1t.deployment.id` the deployment on g1t.page, and |
| 8 | * `vcs.ref.head.revision` the commit. Every server span is also a request, |
| 9 | * so software hosted anywhere gets the same request numbers apps on |
| 10 | * g1t.page get from their tail events; exceptions recorded on spans (the |
| 11 | * `exception` event) and in logs (`exception.*` attributes) are errors. |
| 12 | */ |
| 13 | import type { AttributeValue, Attributes, SpanKind, SpanStatus, TelemetryOrigin, TelemetrySource } from "@g1t/contracts/telemetry"; |
| 14 | |
| 15 | import { emptyBatch, type Batch, type MetricPoint } from "../batch.ts"; |
| 16 | import { MAX_VALUE, clip, hexId, hostPath, nanosToMicros, randomId, route, severityNumber, severityOf } from "../shape.ts"; |
| 17 | import type { |
| 18 | AnyValueJson, |
| 19 | HistogramPointJson, |
| 20 | KeyValueJson, |
| 21 | LogsRequestJson, |
| 22 | MetricJson, |
| 23 | MetricsRequestJson, |
| 24 | NumberPointJson, |
| 25 | ResourceJson, |
| 26 | SummaryPointJson, |
| 27 | TracesRequestJson, |
| 28 | } from "./proto.ts"; |
| 29 | |
| 30 | /** What the door the records came through says about them, over what they say themselves. */ |
| 31 | export type Door = { |
| 32 | source: TelemetrySource; |
| 33 | /** |
| 34 | * For Cloudflare's own export: the app's deployment, found from its |
| 35 | * script name. `undefined` keeps what the resource says; `null` skips |
| 36 | * the resource (another project's). |
| 37 | */ |
| 38 | origin?: (resource: Attributes) => Partial<TelemetryOrigin> | null | undefined; |
| 39 | /** Whether server spans count as requests: not when the tail event already counted them. */ |
| 40 | countRequests: boolean; |
| 41 | }; |
| 42 | |
| 43 | export const OTLP_DOOR: Door = { source: "otlp", countRequests: true }; |
| 44 | |
| 45 | const MAX_ATTRIBUTES = 128; |
| 46 | |
| 47 | /** An OTLP value as a plain one. Integers beyond a double's range keep their digits as a string. */ |
| 48 | export function value(v: AnyValueJson | undefined): AttributeValue { |
| 49 | if (!v) return null; |
| 50 | if ("stringValue" in v) return clip(String(v.stringValue), MAX_VALUE); |
| 51 | if ("boolValue" in v) return Boolean(v.boolValue); |
| 52 | if ("intValue" in v) { |
| 53 | const n = Number(v.intValue); |
| 54 | return Number.isSafeInteger(n) ? n : String(v.intValue); |
| 55 | } |
| 56 | if ("doubleValue" in v) return Number(v.doubleValue); |
| 57 | if ("arrayValue" in v) return (v.arrayValue?.values ?? []).slice(0, 64).map(value); |
| 58 | if ("kvlistValue" in v) return attributes(v.kvlistValue?.values); |
| 59 | if ("bytesValue" in v) return String(v.bytesValue); |
| 60 | return null; |
| 61 | } |
| 62 | |
| 63 | export function attributes(list: KeyValueJson[] | undefined): Attributes { |
| 64 | const out: Attributes = {}; |
| 65 | for (const kv of (list ?? []).slice(0, MAX_ATTRIBUTES)) if (kv?.key) out[kv.key] = value(kv.value); |
| 66 | return out; |
| 67 | } |
| 68 | |
| 69 | function str(a: Attributes, ...keys: string[]): string | null { |
| 70 | for (const key of keys) { |
| 71 | const v = a[key]; |
| 72 | if (typeof v === "string" && v !== "") return v; |
| 73 | if (typeof v === "number") return String(v); |
| 74 | } |
| 75 | return null; |
| 76 | } |
| 77 | |
| 78 | function num(a: Attributes, ...keys: string[]): number | null { |
| 79 | for (const key of keys) { |
| 80 | const v = a[key]; |
| 81 | if (typeof v === "number" && Number.isFinite(v)) return v; |
| 82 | if (typeof v === "string" && v !== "" && Number.isFinite(Number(v))) return Number(v); |
| 83 | } |
| 84 | return null; |
| 85 | } |
| 86 | |
| 87 | /** Where a resource's records are filed. */ |
| 88 | export function originOf(resource: ResourceJson | undefined, door: Door): { origin: TelemetryOrigin; resource: Attributes } | null { |
| 89 | const r = attributes(resource?.attributes); |
| 90 | const origin: TelemetryOrigin = { |
| 91 | service: str(r, "service.name") ?? "unknown_service", |
| 92 | environment: str(r, "deployment.environment.name", "deployment.environment"), |
| 93 | deployment_id: str(r, "g1t.deployment.id"), |
| 94 | commit_sha: str(r, "vcs.ref.head.revision", "g1t.commit.sha"), |
| 95 | source: door.source, |
| 96 | }; |
| 97 | const over = door.origin ? door.origin(r) : undefined; |
| 98 | if (over === null) return null; |
| 99 | return { origin: { ...origin, ...(over ?? {}) }, resource: r }; |
| 100 | } |
| 101 | |
| 102 | const KINDS: SpanKind[] = ["unspecified", "internal", "server", "client", "producer", "consumer"]; |
| 103 | const STATUSES: SpanStatus[] = ["unset", "ok", "error"]; |
| 104 | |
| 105 | function kindOf(kind: number | string | undefined): SpanKind { |
| 106 | if (typeof kind === "string") { |
| 107 | const name = kind.replace(/^SPAN_KIND_/, "").toLowerCase() as SpanKind; |
| 108 | return KINDS.includes(name) ? name : "unspecified"; |
| 109 | } |
| 110 | return KINDS[kind ?? 0] ?? "unspecified"; |
| 111 | } |
| 112 | |
| 113 | function statusOf(code: number | string | undefined): SpanStatus { |
| 114 | if (typeof code === "string") { |
| 115 | const name = code.replace(/^STATUS_CODE_/, "").toLowerCase() as SpanStatus; |
| 116 | return STATUSES.includes(name) ? name : "unset"; |
| 117 | } |
| 118 | return STATUSES[code ?? 0] ?? "unset"; |
| 119 | } |
| 120 | |
| 121 | export function readTraces(body: TracesRequestJson, door: Door, bytes: number): Batch { |
| 122 | const batch = emptyBatch(); |
| 123 | batch.bytes = bytes; |
| 124 | for (const rs of body.resourceSpans ?? []) { |
| 125 | const found = originOf(rs.resource, door); |
| 126 | if (!found) continue; |
| 127 | const { origin, resource } = found; |
| 128 | for (const ss of rs.scopeSpans ?? []) { |
| 129 | for (const s of ss.spans ?? []) { |
| 130 | const traceId = hexId(s.traceId, 32); |
| 131 | const spanId = hexId(s.spanId, 16); |
| 132 | const start = nanosToMicros(s.startTimeUnixNano); |
| 133 | if (!traceId || !spanId || start === null) continue; |
| 134 | const end = nanosToMicros(s.endTimeUnixNano) ?? start; |
| 135 | const attrs = attributes(s.attributes); |
| 136 | const kind = kindOf(s.kind); |
| 137 | const status = statusOf(s.status?.code); |
| 138 | const events = (s.events ?? []).slice(0, 128).map((e) => ({ |
| 139 | time_us: nanosToMicros(e.timeUnixNano) ?? start, |
| 140 | name: e.name ?? "", |
| 141 | attributes: attributes(e.attributes), |
| 142 | })); |
| 143 | const span = { |
| 144 | ...origin, |
| 145 | trace_id: traceId, |
| 146 | span_id: spanId, |
| 147 | parent_span_id: hexId(s.parentSpanId, 16), |
| 148 | name: clip(s.name || "(unnamed)", 512), |
| 149 | kind, |
| 150 | start_us: start, |
| 151 | duration_us: Math.max(0, end - start), |
| 152 | status, |
| 153 | status_message: s.status?.message ? clip(s.status.message, 2048) : null, |
| 154 | attributes: attrs, |
| 155 | resource, |
| 156 | events, |
| 157 | links: (s.links ?? []).flatMap((l) => { |
| 158 | const t = hexId(l.traceId, 32); |
| 159 | const p = hexId(l.spanId, 16); |
| 160 | return t && p ? [{ trace_id: t, span_id: p }] : []; |
| 161 | }), |
| 162 | }; |
| 163 | batch.spans.push(span); |
| 164 | |
| 165 | for (const e of events) { |
| 166 | if (e.name !== "exception") continue; |
| 167 | batch.errors.push({ |
| 168 | ...origin, |
| 169 | time_us: e.time_us, |
| 170 | name: str(e.attributes, "exception.type") ?? "Error", |
| 171 | message: str(e.attributes, "exception.message") ?? span.status_message ?? "", |
| 172 | stack: str(e.attributes, "exception.stacktrace"), |
| 173 | request_id: kind === "server" ? spanId : null, |
| 174 | trace_id: traceId, |
| 175 | }); |
| 176 | } |
| 177 | |
| 178 | // An app on g1t.page has its requests counted from its tail events. |
| 179 | if (door.countRequests && kind === "server" && !origin.deployment_id) { |
| 180 | const status = num(attrs, "http.response.status_code", "http.status_code"); |
| 181 | const full = str(attrs, "url.full", "http.url"); |
| 182 | const where = full ? hostPath(full) : null; |
| 183 | const path = str(attrs, "url.path", "http.target") ?? where?.path ?? null; |
| 184 | const host = str(attrs, "server.address", "http.host", "net.host.name"); |
| 185 | const failed = span.status === "error" || (status !== null && status >= 500); |
| 186 | batch.requests.push({ |
| 187 | ...origin, |
| 188 | id: spanId, |
| 189 | time_us: start, |
| 190 | duration_us: span.duration_us, |
| 191 | trigger: str(attrs, "faas.trigger") ?? (path || status !== null ? "fetch" : "other"), |
| 192 | method: str(attrs, "http.request.method", "http.method"), |
| 193 | url: where?.url ?? (path ? `${host ?? ""}${path.split("?")[0]}` : null), |
| 194 | route: str(attrs, "http.route") ?? (path ? route(path) : span.name), |
| 195 | status: status === null ? null : Math.trunc(status), |
| 196 | outcome: failed ? "error" : "ok", |
| 197 | cpu_ms: null, |
| 198 | region: str(resource, "cloud.region") ?? str(attrs, "cloud.region"), |
| 199 | country: null, |
| 200 | trace_id: traceId, |
| 201 | log_count: 0, |
| 202 | error: failed ? { name: str(attrs, "error.type") ?? "Error", message: span.status_message ?? (status ? `HTTP ${status}` : "") } : null, |
| 203 | }); |
| 204 | } |
| 205 | } |
| 206 | } |
| 207 | } |
| 208 | return batch; |
| 209 | } |
| 210 | |
| 211 | export function readLogs(body: LogsRequestJson, door: Door, bytes: number): Batch { |
| 212 | const batch = emptyBatch(); |
| 213 | batch.bytes = bytes; |
| 214 | for (const rl of body.resourceLogs ?? []) { |
| 215 | const found = originOf(rl.resource, door); |
| 216 | if (!found) continue; |
| 217 | const { origin } = found; |
| 218 | for (const sl of rl.scopeLogs ?? []) { |
| 219 | for (const r of sl.logRecords ?? []) { |
| 220 | const time = nanosToMicros(r.timeUnixNano) ?? nanosToMicros(r.observedTimeUnixNano) ?? Date.now() * 1000; |
| 221 | const attrs = attributes(r.attributes); |
| 222 | const number = r.severityNumber && r.severityNumber > 0 ? r.severityNumber : severityNumber(r.severityText); |
| 223 | const raw = value(r.body); |
| 224 | const body = typeof raw === "string" ? raw : raw === null ? (r.eventName ?? "") : JSON.stringify(raw); |
| 225 | const traceId = hexId(r.traceId, 32); |
| 226 | const spanId = hexId(r.spanId, 16); |
| 227 | batch.logs.push({ |
| 228 | ...origin, |
| 229 | id: randomId(), |
| 230 | time_us: time, |
| 231 | severity: severityOf(number), |
| 232 | severity_number: number, |
| 233 | body: clip(body), |
| 234 | trace_id: traceId, |
| 235 | span_id: spanId, |
| 236 | request_id: null, |
| 237 | attributes: attrs, |
| 238 | }); |
| 239 | const type = str(attrs, "exception.type"); |
| 240 | const message = str(attrs, "exception.message"); |
| 241 | if (type || message) { |
| 242 | batch.errors.push({ |
| 243 | ...origin, |
| 244 | time_us: time, |
| 245 | name: type ?? "Error", |
| 246 | message: message ?? body, |
| 247 | stack: str(attrs, "exception.stacktrace"), |
| 248 | request_id: null, |
| 249 | trace_id: traceId, |
| 250 | }); |
| 251 | } |
| 252 | } |
| 253 | } |
| 254 | } |
| 255 | return batch; |
| 256 | } |
| 257 | |
| 258 | export function readMetrics(body: MetricsRequestJson, door: Door, bytes: number): Batch { |
| 259 | const batch = emptyBatch(); |
| 260 | batch.bytes = bytes; |
| 261 | for (const rm of body.resourceMetrics ?? []) { |
| 262 | const found = originOf(rm.resource, door); |
| 263 | if (!found) continue; |
| 264 | const { origin } = found; |
| 265 | for (const sm of rm.scopeMetrics ?? []) { |
| 266 | for (const m of sm.metrics ?? []) batch.metrics.push(...points(m, origin)); |
| 267 | } |
| 268 | } |
| 269 | return batch; |
| 270 | } |
| 271 | |
| 272 | function points(m: MetricJson, origin: TelemetryOrigin): MetricPoint[] { |
| 273 | const name = m.name ?? ""; |
| 274 | if (!name) return []; |
| 275 | const base = { ...origin, name, unit: m.unit ?? "", description: clip(m.description ?? "", 512) }; |
| 276 | const at = (p: { timeUnixNano?: string }) => nanosToMicros(p.timeUnixNano) ?? Date.now() * 1000; |
| 277 | const empty = { count: null, sum: null, min: null, max: null, bounds: null, counts: null, quantiles: null }; |
| 278 | const number = (p: NumberPointJson) => (p.asDouble !== undefined ? Number(p.asDouble) : p.asInt !== undefined ? Number(p.asInt) : null); |
| 279 | if (m.gauge) { |
| 280 | return (m.gauge.dataPoints ?? []).map((p) => ({ ...base, ...empty, kind: "gauge", time_us: at(p), value: number(p), monotonic: false, attributes: attributes(p.attributes) })); |
| 281 | } |
| 282 | if (m.sum) { |
| 283 | // A delta sum is the amount in its interval; a cumulative one, the |
| 284 | // total since its start. Both are kept as sent, with which it was. |
| 285 | const cumulative = m.sum.aggregationTemporality === 2; |
| 286 | return (m.sum.dataPoints ?? []).map((p) => ({ |
| 287 | ...base, |
| 288 | ...empty, |
| 289 | kind: "sum", |
| 290 | time_us: at(p), |
| 291 | value: number(p), |
| 292 | monotonic: cumulative && Boolean(m.sum!.isMonotonic), |
| 293 | attributes: attributes(p.attributes), |
| 294 | })); |
| 295 | } |
| 296 | const histogram = (p: HistogramPointJson): MetricPoint => { |
| 297 | const count = p.count !== undefined ? Number(p.count) : null; |
| 298 | const sum = p.sum ?? null; |
| 299 | return { |
| 300 | ...base, |
| 301 | kind: "histogram", |
| 302 | time_us: at(p), |
| 303 | value: count && sum !== null ? sum / count : null, |
| 304 | count, |
| 305 | sum, |
| 306 | min: p.min ?? null, |
| 307 | max: p.max ?? null, |
| 308 | bounds: p.explicitBounds?.length ? p.explicitBounds.map(Number) : null, |
| 309 | counts: p.bucketCounts?.length ? p.bucketCounts.map(Number) : null, |
| 310 | quantiles: null, |
| 311 | monotonic: false, |
| 312 | attributes: attributes(p.attributes), |
| 313 | }; |
| 314 | }; |
| 315 | if (m.histogram) return (m.histogram.dataPoints ?? []).map(histogram); |
| 316 | if (m.exponentialHistogram) return (m.exponentialHistogram.dataPoints ?? []).map(histogram); |
| 317 | if (m.summary) { |
| 318 | return (m.summary.dataPoints ?? []).map((p: SummaryPointJson) => { |
| 319 | const count = p.count !== undefined ? Number(p.count) : null; |
| 320 | const sum = p.sum ?? null; |
| 321 | return { |
| 322 | ...base, |
| 323 | ...empty, |
| 324 | kind: "summary", |
| 325 | time_us: at(p), |
| 326 | value: count && sum !== null ? sum / count : null, |
| 327 | count, |
| 328 | sum, |
| 329 | quantiles: (p.quantileValues ?? []).map((q) => ({ q: Number(q.quantile ?? 0), v: Number(q.value ?? 0) })), |
| 330 | monotonic: false, |
| 331 | attributes: attributes(p.attributes), |
| 332 | }; |
| 333 | }); |
| 334 | } |
| 335 | return []; |
| 336 | } |