Skip to content
Sign up

flagon-io/g1t

public
0
120 linesCodeBlameRaw
1/**
2 * Reading an OTLP/HTTP request: its token, its encoding (JSON or
3 * protobuf, gzipped or not) and its signal, into a batch. And what OTLP
4 * expects back.
5 */
6import { TELEMETRY_TOKEN_PREFIX } from "@g1t/contracts/telemetry";
7
8import type { Batch } from "../batch.ts";
9import { readLogs, readMetrics, readTraces, type Door } from "./normalize.ts";
10import { ProtoError, decodeLogs, decodeMetrics, decodeTraces } from "./proto.ts";
11
12export type Signal = "traces" | "logs" | "metrics";
13
14/** A request body is at most this many bytes as sent, and this many once unzipped. */
15export const MAX_BODY = 8 * 1024 * 1024;
16export const MAX_DECODED = 32 * 1024 * 1024;
17
18export class IngestError extends Error {
19 readonly status: number;
20
21 constructor(status: number, message: string) {
22 super(message);
23 this.status = status;
24 }
25}
26
27/**
28 * The token a request carries: `Authorization: Bearer <token>` (what
29 * `OTEL_EXPORTER_OTLP_HEADERS` sets), or `x-g1t-token`.
30 */
31export function tokenOf(request: Request): string | null {
32 const auth = request.headers.get("authorization") ?? "";
33 const bearer = /^bearer\s+(\S+)$/i.exec(auth.trim())?.[1];
34 return bearer ?? request.headers.get("x-g1t-token");
35}
36
37/**
38 * A project id as g1t makes them (`newId("prj")`): `prj_` and a UUIDv7 in
39 * lowercase Crockford base32. Anything else names no project, so it never
40 * reaches (or makes) a store.
41 */
42const PROJECT_ID = /^prj_[0-9abcdefghjkmnpqrstvwxyz]{26}$/;
43
44/** The project a token is for: tokens are `g1to_<project id>_<secret>`. */
45export function projectOfToken(token: string): string | null {
46 if (!token.startsWith(TELEMETRY_TOKEN_PREFIX)) return null;
47 const rest = token.slice(TELEMETRY_TOKEN_PREFIX.length);
48 const at = rest.lastIndexOf("_");
49 if (at <= 0 || at === rest.length - 1) return null;
50 const project = rest.slice(0, at);
51 return PROJECT_ID.test(project) ? project : null;
52}
53
54async function readBody(request: Request): Promise<Uint8Array> {
55 const declared = Number(request.headers.get("content-length") ?? 0);
56 if (declared > MAX_BODY) throw new IngestError(413, `A request may be at most ${MAX_BODY / 1024 / 1024} MB; send smaller batches.`);
57 const raw = new Uint8Array(await request.arrayBuffer());
58 if (raw.length > MAX_BODY) throw new IngestError(413, `A request may be at most ${MAX_BODY / 1024 / 1024} MB; send smaller batches.`);
59 const encoding = (request.headers.get("content-encoding") ?? "").toLowerCase();
60 if (!encoding || encoding === "identity") return raw;
61 if (encoding !== "gzip" && encoding !== "deflate") throw new IngestError(415, `Content-Encoding ${encoding} is not one this endpoint reads: send gzip or none.`);
62 const stream = new Blob([raw]).stream().pipeThrough(new DecompressionStream(encoding));
63 const reader = stream.getReader();
64 const chunks: Uint8Array[] = [];
65 let size = 0;
66 for (;;) {
67 const { done, value } = await reader.read();
68 if (done) break;
69 size += value.length;
70 if (size > MAX_DECODED) throw new IngestError(413, "The request is too large once unzipped; send smaller batches.");
71 chunks.push(value);
72 }
73 const out = new Uint8Array(size);
74 let at = 0;
75 for (const c of chunks) {
76 out.set(c, at);
77 at += c.length;
78 }
79 return out;
80}
81
82/** The request read into a batch for the store. */
83export async function readOtlp(request: Request, signal: Signal, door: Door): Promise<{ batch: Batch; protobuf: boolean }> {
84 const type = (request.headers.get("content-type") ?? "").split(";")[0].trim().toLowerCase();
85 const protobuf = type === "application/x-protobuf" || type === "application/protobuf";
86 if (!protobuf && type !== "application/json" && type !== "") {
87 throw new IngestError(415, `Content-Type ${type} is not OTLP: send application/json or application/x-protobuf.`);
88 }
89 const bytes = await readBody(request);
90 let body: unknown;
91 try {
92 if (protobuf) {
93 body = signal === "traces" ? decodeTraces(bytes) : signal === "logs" ? decodeLogs(bytes) : decodeMetrics(bytes);
94 } else {
95 body = bytes.length ? JSON.parse(new TextDecoder().decode(bytes)) : {};
96 }
97 } catch (error) {
98 if (error instanceof ProtoError || error instanceof SyntaxError) throw new IngestError(400, `The body is not OTLP ${protobuf ? "protobuf" : "JSON"}: ${error.message}`);
99 throw error;
100 }
101 if (!body || typeof body !== "object") throw new IngestError(400, "The body is not an OTLP export request.");
102 const batch =
103 signal === "traces" ? readTraces(body as never, door, bytes.length) : signal === "logs" ? readLogs(body as never, door, bytes.length) : readMetrics(body as never, door, bytes.length);
104 return { batch, protobuf };
105}
106
107/** What OTLP expects back for a success: an empty export response, in the encoding it sent. */
108export function accepted(protobuf: boolean): Response {
109 return protobuf
110 ? new Response(new Uint8Array(0), { status: 200, headers: { "content-type": "application/x-protobuf" } })
111 : Response.json({}, { status: 200 });
112}
113
114/**
115 * A failure, as OTLP describes one (a `Status` with a message). 4xx other
116 * than 429 tells the exporter not to retry.
117 */
118export function refused(status: number, message: string): Response {
119 return Response.json({ code: status === 401 ? 16 : status === 429 ? 8 : 3, message }, { status });
120}