| 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 | */ |
| 6 | import { TELEMETRY_TOKEN_PREFIX } from "@g1t/contracts/telemetry"; |
| 7 | |
| 8 | import type { Batch } from "../batch.ts"; |
| 9 | import { readLogs, readMetrics, readTraces, type Door } from "./normalize.ts"; |
| 10 | import { ProtoError, decodeLogs, decodeMetrics, decodeTraces } from "./proto.ts"; |
| 11 | |
| 12 | export type Signal = "traces" | "logs" | "metrics"; |
| 13 | |
| 14 | /** A request body is at most this many bytes as sent, and this many once unzipped. */ |
| 15 | export const MAX_BODY = 8 * 1024 * 1024; |
| 16 | export const MAX_DECODED = 32 * 1024 * 1024; |
| 17 | |
| 18 | export 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 | */ |
| 31 | export 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 | /** The project a token is for: tokens are `g1to_<project id>_<secret>`. */ |
| 38 | export function projectOfToken(token: string): string | null { |
| 39 | if (!token.startsWith(TELEMETRY_TOKEN_PREFIX)) return null; |
| 40 | const rest = token.slice(TELEMETRY_TOKEN_PREFIX.length); |
| 41 | const at = rest.lastIndexOf("_"); |
| 42 | if (at <= 0 || at === rest.length - 1) return null; |
| 43 | const project = rest.slice(0, at); |
| 44 | return /^[A-Za-z0-9_-]{1,64}$/.test(project) ? project : null; |
| 45 | } |
| 46 | |
| 47 | async function readBody(request: Request): Promise<Uint8Array> { |
| 48 | const declared = Number(request.headers.get("content-length") ?? 0); |
| 49 | if (declared > MAX_BODY) throw new IngestError(413, `A request may be at most ${MAX_BODY / 1024 / 1024} MB; send smaller batches.`); |
| 50 | const raw = new Uint8Array(await request.arrayBuffer()); |
| 51 | if (raw.length > MAX_BODY) throw new IngestError(413, `A request may be at most ${MAX_BODY / 1024 / 1024} MB; send smaller batches.`); |
| 52 | const encoding = (request.headers.get("content-encoding") ?? "").toLowerCase(); |
| 53 | if (!encoding || encoding === "identity") return raw; |
| 54 | if (encoding !== "gzip" && encoding !== "deflate") throw new IngestError(415, `Content-Encoding ${encoding} is not one this endpoint reads: send gzip or none.`); |
| 55 | const stream = new Blob([raw]).stream().pipeThrough(new DecompressionStream(encoding)); |
| 56 | const reader = stream.getReader(); |
| 57 | const chunks: Uint8Array[] = []; |
| 58 | let size = 0; |
| 59 | for (;;) { |
| 60 | const { done, value } = await reader.read(); |
| 61 | if (done) break; |
| 62 | size += value.length; |
| 63 | if (size > MAX_DECODED) throw new IngestError(413, "The request is too large once unzipped; send smaller batches."); |
| 64 | chunks.push(value); |
| 65 | } |
| 66 | const out = new Uint8Array(size); |
| 67 | let at = 0; |
| 68 | for (const c of chunks) { |
| 69 | out.set(c, at); |
| 70 | at += c.length; |
| 71 | } |
| 72 | return out; |
| 73 | } |
| 74 | |
| 75 | /** The request read into a batch for the store. */ |
| 76 | export async function readOtlp(request: Request, signal: Signal, door: Door): Promise<{ batch: Batch; protobuf: boolean }> { |
| 77 | const type = (request.headers.get("content-type") ?? "").split(";")[0].trim().toLowerCase(); |
| 78 | const protobuf = type === "application/x-protobuf" || type === "application/protobuf"; |
| 79 | if (!protobuf && type !== "application/json" && type !== "") { |
| 80 | throw new IngestError(415, `Content-Type ${type} is not OTLP: send application/json or application/x-protobuf.`); |
| 81 | } |
| 82 | const bytes = await readBody(request); |
| 83 | let body: unknown; |
| 84 | try { |
| 85 | if (protobuf) { |
| 86 | body = signal === "traces" ? decodeTraces(bytes) : signal === "logs" ? decodeLogs(bytes) : decodeMetrics(bytes); |
| 87 | } else { |
| 88 | body = bytes.length ? JSON.parse(new TextDecoder().decode(bytes)) : {}; |
| 89 | } |
| 90 | } catch (error) { |
| 91 | if (error instanceof ProtoError || error instanceof SyntaxError) throw new IngestError(400, `The body is not OTLP ${protobuf ? "protobuf" : "JSON"}: ${error.message}`); |
| 92 | throw error; |
| 93 | } |
| 94 | if (!body || typeof body !== "object") throw new IngestError(400, "The body is not an OTLP export request."); |
| 95 | const batch = |
| 96 | 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); |
| 97 | return { batch, protobuf }; |
| 98 | } |
| 99 | |
| 100 | /** What OTLP expects back for a success: an empty export response, in the encoding it sent. */ |
| 101 | export function accepted(protobuf: boolean): Response { |
| 102 | return protobuf |
| 103 | ? new Response(new Uint8Array(0), { status: 200, headers: { "content-type": "application/x-protobuf" } }) |
| 104 | : Response.json({}, { status: 200 }); |
| 105 | } |
| 106 | |
| 107 | /** |
| 108 | * A failure, as OTLP describes one (a `Status` with a message). 4xx other |
| 109 | * than 429 tells the exporter not to retry. |
| 110 | */ |
| 111 | export function refused(status: number, message: string): Response { |
| 112 | return Response.json({ code: status === 401 ? 16 : status === 429 ? 8 : 3, message }, { status }); |
| 113 | } |