Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

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