| 1 | /** |
| 2 | * Sending a project's telemetry on, as it arrives, to any OTLP/HTTP |
| 3 | * endpoint it names (a collector, or a vendor that takes OTLP): rows back |
| 4 | * into OTLP/JSON. Requests go as server spans (and their logs as log |
| 5 | * records), so wherever they land shows them as it shows any service. |
| 6 | */ |
| 7 | import { TELEMETRY_OTLP_ENDPOINT, type AttributeValue, type Attributes, type TelemetryExport, type TelemetryLog, type TelemetryOrigin, type TelemetryRequest, type TelemetrySpan } from "@g1t/contracts/telemetry"; |
| 8 | |
| 9 | import type { Batch, MetricPoint } from "./batch.ts"; |
| 10 | |
| 11 | type KeyValue = { key: string; value: unknown }; |
| 12 | |
| 13 | function anyValue(v: AttributeValue): unknown { |
| 14 | if (v === null) return {}; |
| 15 | if (typeof v === "string") return { stringValue: v }; |
| 16 | if (typeof v === "boolean") return { boolValue: v }; |
| 17 | if (typeof v === "number") return Number.isInteger(v) ? { intValue: String(v) } : { doubleValue: v }; |
| 18 | if (Array.isArray(v)) return { arrayValue: { values: v.map(anyValue) } }; |
| 19 | return { kvlistValue: { values: kvs(v) } }; |
| 20 | } |
| 21 | |
| 22 | function kvs(a: Attributes): KeyValue[] { |
| 23 | return Object.entries(a).map(([key, value]) => ({ key, value: anyValue(value) })); |
| 24 | } |
| 25 | |
| 26 | const nanos = (us: number) => `${Math.round(us)}000`; |
| 27 | |
| 28 | function resourceOf(o: TelemetryOrigin): { attributes: KeyValue[] } { |
| 29 | const a: Attributes = { "service.name": o.service || "unknown_service", "g1t.source": o.source }; |
| 30 | if (o.environment) a["deployment.environment.name"] = o.environment; |
| 31 | if (o.deployment_id) a["g1t.deployment.id"] = o.deployment_id; |
| 32 | if (o.commit_sha) a["vcs.ref.head.revision"] = o.commit_sha; |
| 33 | return { attributes: kvs(a) }; |
| 34 | } |
| 35 | |
| 36 | const originKey = (o: TelemetryOrigin) => [o.service, o.environment, o.deployment_id, o.commit_sha, o.source].join("\u0000"); |
| 37 | |
| 38 | /** Rows grouped by where they came from, as OTLP groups them by resource. */ |
| 39 | function byResource<T extends TelemetryOrigin>(rows: T[]): T[][] { |
| 40 | const groups = new Map<string, T[]>(); |
| 41 | for (const r of rows) { |
| 42 | const key = originKey(r); |
| 43 | groups.set(key, [...(groups.get(key) ?? []), r]); |
| 44 | } |
| 45 | return [...groups.values()]; |
| 46 | } |
| 47 | |
| 48 | const SCOPE = { name: "g1t", version: "1" }; |
| 49 | const KIND = { unspecified: 0, internal: 1, server: 2, client: 3, producer: 4, consumer: 5 } as const; |
| 50 | const STATUS = { unset: 0, ok: 1, error: 2 } as const; |
| 51 | |
| 52 | function requestSpan(r: TelemetryRequest) { |
| 53 | const a: Attributes = { "faas.trigger": r.trigger, "g1t.request.id": r.id }; |
| 54 | if (r.method) a["http.request.method"] = r.method; |
| 55 | if (r.url) a["url.full"] = `https://${r.url}`; |
| 56 | if (r.route) a["http.route"] = r.route; |
| 57 | if (r.status !== null) a["http.response.status_code"] = r.status; |
| 58 | if (r.cpu_ms !== null) a["cloudflare.cpu_time_ms"] = r.cpu_ms; |
| 59 | if (r.region) a["cloudflare.colo"] = r.region; |
| 60 | if (r.country) a["client.geo.country_iso_code"] = r.country; |
| 61 | a["cloudflare.outcome"] = r.outcome; |
| 62 | const failed = r.error !== null || (r.status ?? 0) >= 500; |
| 63 | return { |
| 64 | traceId: r.trace_id ?? r.id.padStart(32, "0"), |
| 65 | spanId: r.id, |
| 66 | name: [r.method, r.route].filter(Boolean).join(" ") || r.trigger, |
| 67 | kind: KIND.server, |
| 68 | startTimeUnixNano: nanos(r.time_us), |
| 69 | endTimeUnixNano: nanos(r.time_us + r.duration_us), |
| 70 | attributes: kvs(a), |
| 71 | status: failed ? { code: STATUS.error, message: r.error?.message ?? "" } : { code: STATUS.unset }, |
| 72 | }; |
| 73 | } |
| 74 | |
| 75 | function span(s: TelemetrySpan) { |
| 76 | return { |
| 77 | traceId: s.trace_id, |
| 78 | spanId: s.span_id, |
| 79 | parentSpanId: s.parent_span_id ?? undefined, |
| 80 | name: s.name, |
| 81 | kind: KIND[s.kind], |
| 82 | startTimeUnixNano: nanos(s.start_us), |
| 83 | endTimeUnixNano: nanos(s.start_us + s.duration_us), |
| 84 | attributes: kvs(s.attributes), |
| 85 | events: s.events.map((e) => ({ timeUnixNano: nanos(e.time_us), name: e.name, attributes: kvs(e.attributes) })), |
| 86 | links: s.links.map((l) => ({ traceId: l.trace_id, spanId: l.span_id })), |
| 87 | status: { code: STATUS[s.status], message: s.status_message ?? undefined }, |
| 88 | }; |
| 89 | } |
| 90 | |
| 91 | function log(l: TelemetryLog) { |
| 92 | const a = { ...l.attributes }; |
| 93 | if (l.request_id) a["g1t.request.id"] = l.request_id; |
| 94 | return { |
| 95 | timeUnixNano: nanos(l.time_us), |
| 96 | severityNumber: l.severity_number, |
| 97 | severityText: l.severity.toUpperCase(), |
| 98 | body: { stringValue: l.body }, |
| 99 | attributes: kvs(a), |
| 100 | traceId: l.trace_id ?? undefined, |
| 101 | spanId: l.span_id ?? undefined, |
| 102 | }; |
| 103 | } |
| 104 | |
| 105 | function metric(m: MetricPoint) { |
| 106 | const point = { attributes: kvs(m.attributes), timeUnixNano: nanos(m.time_us) }; |
| 107 | const base = { name: m.name, unit: m.unit, description: m.description }; |
| 108 | switch (m.kind) { |
| 109 | case "gauge": |
| 110 | return { ...base, gauge: { dataPoints: [{ ...point, asDouble: m.value ?? 0 }] } }; |
| 111 | case "sum": |
| 112 | return { ...base, sum: { dataPoints: [{ ...point, asDouble: m.value ?? 0 }], aggregationTemporality: m.monotonic ? 2 : 1, isMonotonic: m.monotonic } }; |
| 113 | case "histogram": |
| 114 | return { |
| 115 | ...base, |
| 116 | histogram: { |
| 117 | aggregationTemporality: 1, |
| 118 | dataPoints: [{ ...point, count: String(m.count ?? 0), sum: m.sum ?? undefined, min: m.min ?? undefined, max: m.max ?? undefined, explicitBounds: m.bounds ?? [], bucketCounts: (m.counts ?? []).map(String) }], |
| 119 | }, |
| 120 | }; |
| 121 | case "summary": |
| 122 | return { ...base, summary: { dataPoints: [{ ...point, count: String(m.count ?? 0), sum: m.sum ?? 0, quantileValues: (m.quantiles ?? []).map((q) => ({ quantile: q.q, value: q.v })) }] } }; |
| 123 | } |
| 124 | } |
| 125 | |
| 126 | /** The OTLP/JSON bodies a batch makes, by signal; a signal with nothing in it is absent. */ |
| 127 | export function otlpBodies(batch: Batch, requests: TelemetryRequest[]): { traces?: object; logs?: object; metrics?: object } { |
| 128 | const out: { traces?: object; logs?: object; metrics?: object } = {}; |
| 129 | type Traced = (TelemetryOrigin & { span: object }); |
| 130 | const traced: Traced[] = [ |
| 131 | ...requests.map((r) => ({ ...r, span: requestSpan(r) })), |
| 132 | ...batch.spans.map((s) => ({ ...s, span: span(s) })), |
| 133 | ]; |
| 134 | if (traced.length) { |
| 135 | out.traces = { resourceSpans: byResource(traced).map((g) => ({ resource: resourceOf(g[0]), scopeSpans: [{ scope: SCOPE, spans: g.map((t) => t.span) }] })) }; |
| 136 | } |
| 137 | if (batch.logs.length) { |
| 138 | out.logs = { resourceLogs: byResource(batch.logs).map((g) => ({ resource: resourceOf(g[0]), scopeLogs: [{ scope: SCOPE, logRecords: g.map(log) }] })) }; |
| 139 | } |
| 140 | if (batch.metrics.length) { |
| 141 | out.metrics = { resourceMetrics: byResource(batch.metrics).map((g) => ({ resource: resourceOf(g[0]), scopeMetrics: [{ scope: SCOPE, metrics: g.map(metric) }] })) }; |
| 142 | } |
| 143 | return out; |
| 144 | } |
| 145 | |
| 146 | /** |
| 147 | * Whether an export's address is one telemetry may be sent to: https, and |
| 148 | * not g1t's own ingest (which would loop) or an address on a private network. |
| 149 | */ |
| 150 | export function exportEndpointProblem(endpoint: string): string | null { |
| 151 | let url: URL; |
| 152 | try { |
| 153 | url = new URL(endpoint); |
| 154 | } catch { |
| 155 | return "That is not a web address."; |
| 156 | } |
| 157 | if (url.protocol !== "https:") return "The address must start with https://."; |
| 158 | const host = url.hostname.toLowerCase(); |
| 159 | if (host === new URL(TELEMETRY_OTLP_ENDPOINT).hostname) return "That is g1t's own endpoint: it would send everything back here."; |
| 160 | if (host === "localhost" || /^(127|10|0)\./.test(host) || /^192\.168\./.test(host) || /^172\.(1[6-9]|2\d|3[01])\./.test(host) || host.endsWith(".internal") || host.startsWith("[")) { |
| 161 | return "The address must be reachable on the internet."; |
| 162 | } |
| 163 | return null; |
| 164 | } |
| 165 | |
| 166 | /** Sends one signal's body to an export; the error, or null. */ |
| 167 | export async function send(target: Pick<TelemetryExport, "endpoint"> & { headers: Record<string, string> }, signal: "traces" | "logs" | "metrics", body: object): Promise<string | null> { |
| 168 | const base = target.endpoint.replace(/\/+$/, ""); |
| 169 | const url = /\/v1\/(traces|logs|metrics)$/.test(base) ? base.replace(/(traces|logs|metrics)$/, signal) : `${base}/v1/${signal}`; |
| 170 | try { |
| 171 | const response = await fetch(url, { |
| 172 | method: "POST", |
| 173 | headers: { ...target.headers, "content-type": "application/json" }, |
| 174 | body: JSON.stringify(body), |
| 175 | signal: AbortSignal.timeout(10_000), |
| 176 | }); |
| 177 | if (response.ok) return null; |
| 178 | return `${url} answered ${response.status}: ${(await response.text()).slice(0, 300)}`; |
| 179 | } catch (error) { |
| 180 | return `${url} could not be reached: ${String(error).replace(/^Error: /, "")}`; |
| 181 | } |
| 182 | } |