diff --git a/deno.json b/deno.json index 937a8f394..831508ec8 100644 --- a/deno.json +++ b/deno.json @@ -1,5 +1,5 @@ { - "version": "3.1.0", + "version": "3.1.1", "deno_version": "2.4.1", "tasks": { "checks": { diff --git a/deno.lock b/deno.lock index fbaa16c8c..4fbeb1bc7 100644 --- a/deno.lock +++ b/deno.lock @@ -48,6 +48,7 @@ "jsr:@std/path@0.214": "0.214.0", "jsr:@std/path@0.214.0": "0.214.0", "jsr:@std/path@0.221": "0.221.0", + "jsr:@std/path@1": "1.1.1", "jsr:@std/path@^1.1.0": "1.1.1", "jsr:@std/path@^1.1.1": "1.1.1", "jsr:@std/path@~1.0.6": "1.0.9", @@ -56,6 +57,8 @@ "jsr:@std/text@~1.0.7": "1.0.15", "jsr:@std/yaml@^1.0.8": "1.0.8", "npm:@octokit/types@^13.4.1": "13.10.0", + "npm:@opentelemetry/api@*": "1.9.0", + "npm:@opentelemetry/core@1": "1.30.1_@opentelemetry+api@1.9.0", "npm:@peculiar/x509@1.11.0": "1.11.0", "npm:@types/node@*": "22.15.15", "npm:crypto-js@4.2.0": "4.2.0", @@ -517,6 +520,19 @@ "aggregate-error" ] }, + "@opentelemetry/api@1.9.0": { + "integrity": "sha512-3giAOQvZiH5F9bMlMiv8+GSPMeqg0dbaeo58/0SlA9sxSqZhnUtxzX9/2FzyhS9sWQf5S0GJE0AKBrFqjpeYcg==" + }, + "@opentelemetry/core@1.30.1_@opentelemetry+api@1.9.0": { + "integrity": "sha512-OOCM2C/QIURhJMuKaekP3TRBxBKxG/TWWA0TL2J6nXUtDnuCtccy49LUJF8xPFXMX+0LMcxFpCo8M9cGY1W6rQ==", + "dependencies": [ + "@opentelemetry/api", + "@opentelemetry/semantic-conventions" + ] + }, + "@opentelemetry/semantic-conventions@1.28.0": { + "integrity": "sha512-lp4qAiMTD4sNWW4DbKLBkfiMZ4jbAboJIGOQr5DvciMRI494OapieI9qiODpOt0XBr1LjIDy1xAGAnVs5supTA==" + }, "@peculiar/asn1-cms@2.3.15": { "integrity": "sha512-B+DoudF+TCrxoJSTjjcY8Mmu+lbv8e7pXGWrhNp2/EGJp9EEcpzjBCar7puU57sGifyzaRVM03oD5L7t7PghQg==", "dependencies": [ @@ -850,6 +866,12 @@ "integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==" } }, + "redirects": { + "https://deno.land/std/http/status.ts": "https://deno.land/std@0.224.0/http/status.ts" + }, + "remote": { + "https://deno.land/std@0.224.0/http/status.ts": "ed61b4882af2514a81aefd3245e8df4c47b9a8e54929a903577643d2d1ebf514" + }, "workspace": { "dependencies": [ "jsr:@cliffy/ansi@^1.0.0-rc.8", diff --git a/src/outputs/functions/main-function.ts b/src/outputs/functions/main-function.ts index c22b6a943..54c7f9923 100644 --- a/src/outputs/functions/main-function.ts +++ b/src/outputs/functions/main-function.ts @@ -1,4 +1,4 @@ -// import { STATUS_CODE } from "https://deno.land/std@0.224.0/http/status.ts"; +import { context, propagation } from "npm:@opentelemetry/api"; const MAX_MEMORY_LIMIT_MB = 150; const WORKER_TIMEOUT_MS = 30000; @@ -7,156 +7,150 @@ const CPU_TIME_SOFT_LIMIT_MS = 10000; const CPU_TIME_HARD_LIMIT_MS = 20000; function mainContent() { + console.log("main function started"); + console.log(Deno.version); + + addEventListener("beforeunload", () => { + console.log("main worker exiting"); + }); + + addEventListener("unhandledrejection", (ev) => { + console.log(ev); + ev.preventDefault(); + }); + Deno.serve(async (req: Request) => { + const ctx = propagation.extract(context.active(), req.headers, { + get(carrier, key) { + return carrier.get(key) ?? void 0; + }, + keys(carrier) { + return [...carrier.keys()]; + }, + }); + + const baggage = propagation.getBaggage(ctx); + const requestId = baggage?.getEntry("cndi-request-id")?.value ?? null; + const headers = new Headers({ "Content-Type": "application/json", }); + const url = new URL(req.url); const { pathname } = url; + // handle health checks if (pathname === "/_internal/health") { return new Response( - JSON.stringify({ - "message": "ok", - }), + JSON.stringify({ "message": "ok" }), { - // @ts-ignore - downstream import provides STATUS_CODE - status: STATUS_CODE.OK, + status: 200, headers, }, ); } + if (pathname === "/_internal/metric") { - // @ts-ignore - EdgeRuntime global is provided downstream + // @ts-ignore - EdgeRuntime defined in runtime const metric = await EdgeRuntime.getRuntimeMetrics(); return Response.json(metric); } - // NOTE: You can test WebSocket in the main worker by uncommenting below. - // if (pathname === '/_internal/ws') { - // const upgrade = req.headers.get("upgrade") || ""; - // if (upgrade.toLowerCase() != "websocket") { - // return new Response("request isn't trying to upgrade to websocket."); - // } - // const { socket, response } = Deno.upgradeWebSocket(req); - // socket.onopen = () => console.log("socket opened"); - // socket.onmessage = (e) => { - // console.log("socket message:", e.data); - // socket.send(new Date().toString()); - // }; - // socket.onerror = e => console.log("socket errored:", e.message); - // socket.onclose = () => console.log("socket closed"); - // return response; // 101 (Switching Protocols) - // } - + let servicePath = pathname; const path_parts = pathname.split("/"); const service_name = path_parts[1]; + if (!service_name || service_name === "") { - const error = { - msg: "missing function name in request", - }; - return new Response(JSON.stringify(error), { - // @ts-ignore - downstream import provides STATUS_CODE - status: STATUS_CODE.BadRequest, - headers: { - "Content-Type": "application/json", + const error = { msg: "missing function name in request" }; + return new Response( + JSON.stringify(error), + { + status: 400, + headers: { "Content-Type": "application/json" }, }, - }); + ); } - const servicePath = `./${service_name}`; - // console.error(`serving the request with ${servicePath}`); - const createWorker = async () => { + + // route functions traffic by pathname to the functions in ./cndi/functions/src/${service_name}/index.ts + servicePath = `./${service_name}`; + + const createWorker = async (otelAttributes?: { [_: string]: string }) => { const memoryLimitMb = MAX_MEMORY_LIMIT_MB; const workerTimeoutMs = WORKER_TIMEOUT_MS; const noModuleCache = NO_MODULE_CACHE; const cpuTimeSoftLimitMs = CPU_TIME_SOFT_LIMIT_MS; const cpuTimeHardLimitMs = CPU_TIME_HARD_LIMIT_MS; - // you can provide an import map inline - // const inlineImportMap = { - // imports: { - // "std/": "https://deno.land/std@0.131.0/", - // "cors": "./examples/_shared/cors.ts" - // } - // } - // const importMapPath = `data:${encodeURIComponent(JSON.stringify(importMap))}?${encodeURIComponent('/home/deno/functions/test')}`; - const importMapPath = null; - const envVarsObj = Deno.env.toObject(); - const envVars = Object.keys(envVarsObj).map((k) => [ - k, - envVarsObj[k], - ]); const forceCreate = false; - const netAccessDisabled = false; - // load source from an eszip - // const maybeEszip = await Deno.readFile('./bin.eszip'); - // const maybeEntrypoint = 'file:///src/index.ts'; - // const maybeEntrypoint = 'file:///src/index.ts'; - // or load module source from an inline module - // const maybeModuleCode = 'Deno.serve((req) => new Response("Hello from Module Code"));'; + const envVarsObj = Deno.env.toObject(); + const envVars = Object.keys(envVarsObj).map((k) => [k, envVarsObj[k]]); - // @ts-ignore - EdgeRuntime global is provided downstream + // @ts-ignore - EdgeRuntime patched return await EdgeRuntime.userWorkers.create({ servicePath, memoryLimitMb, workerTimeoutMs, noModuleCache, - importMapPath, envVars, forceCreate, - netAccessDisabled, cpuTimeSoftLimitMs, cpuTimeHardLimitMs, + staticPatterns: [], + context: { + useReadSyncFileAPI: true, + otel: otelAttributes, + }, + otelConfig: { + tracing_enabled: true, + propagators: ["TraceContext", "Baggage"], + }, }); }; const callWorker = async () => { try { - // If a worker for the given service path already exists, + // If a worker for the given service path already exists // it will be reused by default. // Update forceCreate option in createWorker to force create a new worker for each request. - const worker = await createWorker(); + const worker = await createWorker( + requestId + ? { + "cndi-request-id": requestId, + } + : void 0, + ); + const controller = new AbortController(); + const signal = controller.signal; - // Optional: abort the request after a timeout - // setTimeout(() => controller.abort(), 2 * 60 * 1000); - return await worker.fetch(req, { - signal, - }); - } catch (err) { - const e = err as Error; - console.error(e); - // @ts-ignore - EdgeRuntime seems to patch Deno.errors + // hard abort: setTimeout(() => controller.abort(), 2 * 60 * 1000); + + return await worker.fetch(req, { signal }); + } catch (e) { + // @ts-ignore - Patched by Runtime + if (e instanceof Deno.errors.WorkerAlreadyRetired) { + return await callWorker(); + } + // @ts-ignore - Patched by Runtime if (e instanceof Deno.errors.WorkerRequestCancelled) { headers.append("Connection", "close"); - // XXX(Nyannyacha): I can't think right now how to re-poll - // inside the worker pool without exposing the error to the - // surface. - // It is satisfied when the supervisor that handled the original - // request terminated due to reaches such as CPU time limit or - // Wall-clock limit. - // - // The current request to the worker has been canceled due to - // some internal reasons. We should repoll the worker and call - // `fetch` again. - // return await callWorker(); } - const error = { - msg: e?.toString(), - }; - return new Response(JSON.stringify(error), { - // @ts-ignore - downstream import provides STATUS_CODE - status: STATUS_CODE.InternalServerError, - headers, - }); + + const error = { msg: (e as Error).toString() }; + return new Response( + JSON.stringify(error), + { + status: 500, + headers, + }, + ); } }; + return callWorker(); }); } -// it's possible that this is all silly, and we should instead just fetch this from a URL - type getFunctionsMainContentOptions = { noModuleCache?: boolean; maxMemoryLimitMb?: number; @@ -167,8 +161,8 @@ type getFunctionsMainContentOptions = { export function getFunctionsMainContent( { - noModuleCache = NO_MODULE_CACHE, maxMemoryLimitMb = MAX_MEMORY_LIMIT_MB, + noModuleCache = NO_MODULE_CACHE, workerTimeoutMs = WORKER_TIMEOUT_MS, cpuTimeHardLimitMs = CPU_TIME_HARD_LIMIT_MS, cpuTimeSoftLimitMs = CPU_TIME_SOFT_LIMIT_MS, @@ -176,11 +170,11 @@ export function getFunctionsMainContent( ) { // divide typescript code into headings, imports and content const headings = [ - "// https://github.com/supabase/edge-runtime/blob/main/examples/main/index.ts", + "// https://github.com/polyseam/cndi/blob/main/src/outputs/functions/main-function.ts", ]; const imports = [ - `import { STATUS_CODE } from "https://deno.land/std@0.224.0/http/status.ts";`, + 'import { context, propagation } from "npm:@opentelemetry/api";', ]; const constants = [ diff --git a/src/outputs/functions/runtime-dockerfile.ts b/src/outputs/functions/runtime-dockerfile.ts index 452a29f62..bffe2ce42 100644 --- a/src/outputs/functions/runtime-dockerfile.ts +++ b/src/outputs/functions/runtime-dockerfile.ts @@ -1,11 +1,13 @@ -import { EDGE_RUNTIME_IMAGE_TAG } from "versions"; +import { EDGE_RUNTIME_IMAGE_VERSION } from "versions"; export function getFunctionsDockerfileContent( - version = EDGE_RUNTIME_IMAGE_TAG, + version = EDGE_RUNTIME_IMAGE_VERSION, ) { return ` FROM ghcr.io/supabase/edge-runtime:v${version} +ENV OTEL_DENO=true + COPY ./src /home/deno/functions WORKDIR /home/deno/functions CMD [ "start", "--main-service", "/home/deno/functions/main" ] diff --git a/src/versions.ts b/src/versions.ts index 61e37c48e..843743b5f 100644 --- a/src/versions.ts +++ b/src/versions.ts @@ -10,7 +10,7 @@ export const ARGOCD_RELEASE_VERSION = "2.11.2"; export const LARSTOBI_MULTIPASS_PROVIDER_VERSION = "1.4.2"; // Edge Runtime that powers CNDI Functions -export const EDGE_RUNTIME_IMAGE_TAG = "1.67.4"; +export const EDGE_RUNTIME_IMAGE_VERSION = "1.68.3"; // used in terraform output to create clusters export const DEFAULT_K8S_VERSION = "1.33";