Skip to content
Sign up

flagon-io/g1t

public
0
182 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Projects have Observability: their requests, logs, traces, metrics and errors, from apps on g1t.page with nothing to set up, and from software anywhere else over OpenTelemetry.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 */
7import { TELEMETRY_OTLP_ENDPOINT, type AttributeValue, type Attributes, type TelemetryExport, type TelemetryLog, type TelemetryOrigin, type TelemetryRequest, type TelemetrySpan } from "@g1t/contracts/telemetry";
8
9import type { Batch, MetricPoint } from "./batch.ts";
10
11type KeyValue = { key: string; value: unknown };
12
13function 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
22function kvs(a: Attributes): KeyValue[] {
23 return Object.entries(a).map(([key, value]) => ({ key, value: anyValue(value) }));
24}
25
26const nanos = (us: number) => `${Math.round(us)}000`;
27
28function 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
36const 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. */
39function 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
48const SCOPE = { name: "g1t", version: "1" };
49const KIND = { unspecified: 0, internal: 1, server: 2, client: 3, producer: 4, consumer: 5 } as const;
50const STATUS = { unset: 0, ok: 1, error: 2 } as const;
51
52function 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
75function 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
91function 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
105function 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. */
127export 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 */
150export 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. */
167export 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}