fix: trace opencode provider requests
This commit is contained in:
UniDesk Codex
@@ -95,9 +95,10 @@ export function proxyCloudApiRequest({
|
||||
headers: upstreamRequestHeaders(request, body)
|
||||
},
|
||||
(upstreamResponse) => {
|
||||
const upstreamStatusCode = upstreamResponse.statusCode || 502;
|
||||
const streaming = forceStream || isEventStreamResponse(upstreamResponse.headers);
|
||||
if (streaming) {
|
||||
response.writeHead(upstreamResponse.statusCode || 502, mergeResponseHeaders(copyProxyResponseHeaders(upstreamResponse.headers, { streaming: true }), extraResponseHeaders));
|
||||
response.writeHead(upstreamStatusCode, mergeResponseHeaders(copyProxyResponseHeaders(upstreamResponse.headers, { streaming: true }), extraResponseHeaders));
|
||||
response.flushHeaders?.();
|
||||
upstreamResponse.on("data", (chunk) => {
|
||||
armTimeout();
|
||||
@@ -105,19 +106,19 @@ export function proxyCloudApiRequest({
|
||||
});
|
||||
upstreamResponse.on("end", () => {
|
||||
response.end();
|
||||
settle(resolve);
|
||||
settle(resolve, { statusCode: upstreamStatusCode, streaming: true });
|
||||
});
|
||||
upstreamResponse.on("error", (error) => {
|
||||
if (response.headersSent) {
|
||||
response.destroy(error);
|
||||
settle(resolve);
|
||||
settle(resolve, { statusCode: upstreamStatusCode, streaming: true, errorCode: error?.code || "upstream_response_error" });
|
||||
return;
|
||||
}
|
||||
settle(reject, error);
|
||||
});
|
||||
upstreamResponse.on("close", () => {
|
||||
if (!response.writableEnded) response.end();
|
||||
settle(resolve);
|
||||
settle(resolve, { statusCode: upstreamStatusCode, streaming: true });
|
||||
});
|
||||
response.on("close", () => upstream.destroy());
|
||||
return;
|
||||
@@ -130,12 +131,12 @@ export function proxyCloudApiRequest({
|
||||
});
|
||||
upstreamResponse.on("end", () => {
|
||||
const responseBody = Buffer.concat(chunks);
|
||||
response.writeHead(upstreamResponse.statusCode || 502, {
|
||||
response.writeHead(upstreamStatusCode, {
|
||||
...mergeResponseHeaders(copyProxyResponseHeaders(upstreamResponse.headers), extraResponseHeaders),
|
||||
"content-length": responseBody.length
|
||||
});
|
||||
response.end(responseBody);
|
||||
settle(resolve);
|
||||
settle(resolve, { statusCode: upstreamStatusCode, streaming: false, bodyBytes: responseBody.length });
|
||||
});
|
||||
upstreamResponse.on("error", (error) => settle(reject, error));
|
||||
}
|
||||
@@ -146,7 +147,7 @@ export function proxyCloudApiRequest({
|
||||
if (timedOut) error.timedOut = true;
|
||||
if (response.headersSent) {
|
||||
response.destroy(error);
|
||||
settle(resolve);
|
||||
settle(resolve, { statusCode: 0, streaming: true, errorCode: error?.code || "upstream_request_error" });
|
||||
return;
|
||||
}
|
||||
settle(reject, error);
|
||||
|
||||
@@ -665,16 +665,20 @@ export async function proxyCloudApi({ request, response, url, cloudApiBaseUrl, c
|
||||
|
||||
async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, cloudApiProxyTimeoutMs, opencodeUpstreamUrl, opencodeProxyTimeoutMs, opencodeUsername, opencodePassword, serviceId, sendJson }) {
|
||||
const traceContext = cloudWebTraceContext(request);
|
||||
const startedAtMs = Date.now();
|
||||
if (!cloudApiBaseUrl) {
|
||||
emitOpencodeProxySpanAsync({ traceContext, serviceId, request, url, startedAtMs, statusCode: 503, errorCode: "cloud_api_base_url_missing" });
|
||||
sendJsonWithTraceContext(response, 503, { error: "cloud_api_base_url_missing", serviceId, path: url.pathname, diagnostic: opencodeTraceDiagnostic(traceContext) }, traceContext, sendJson);
|
||||
return;
|
||||
}
|
||||
if (!opencodeUpstreamUrl) {
|
||||
emitOpencodeProxySpanAsync({ traceContext, serviceId, request, url, startedAtMs, statusCode: 503, errorCode: "opencode_upstream_url_missing" });
|
||||
sendJsonWithTraceContext(response, 503, { error: "opencode_upstream_url_missing", serviceId, path: url.pathname, diagnostic: opencodeTraceDiagnostic(traceContext) }, traceContext, sendJson);
|
||||
return;
|
||||
}
|
||||
const authorization = opencodeBasicAuthorization(opencodeUsername, opencodePassword);
|
||||
if (!authorization) {
|
||||
emitOpencodeProxySpanAsync({ traceContext, serviceId, request, url, startedAtMs, statusCode: 503, errorCode: "opencode_credentials_missing" });
|
||||
sendJsonWithTraceContext(response, 503, { error: "opencode_credentials_missing", serviceId, path: url.pathname, diagnostic: opencodeTraceDiagnostic(traceContext) }, traceContext, sendJson);
|
||||
return;
|
||||
}
|
||||
@@ -702,6 +706,17 @@ async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, c
|
||||
valuesPrinted: false
|
||||
}
|
||||
});
|
||||
emitOpencodeProxySpanAsync({
|
||||
traceContext,
|
||||
serviceId,
|
||||
request,
|
||||
url,
|
||||
startedAtMs,
|
||||
statusCode: session.statusCode === 403 ? 403 : 401,
|
||||
errorCode: "opencode_auth_required",
|
||||
ticketAuth,
|
||||
sessionStatusCode: session.statusCode || 0
|
||||
});
|
||||
sendJsonWithTraceContext(response, session.statusCode === 403 ? 403 : 401, {
|
||||
authenticated: false,
|
||||
error: "opencode_auth_required",
|
||||
@@ -727,7 +742,7 @@ async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, c
|
||||
headers: opencodeUpstreamHeaders({ headers: { ...request.headers, ...cloudWebTraceHeaders(traceContext, serviceId) } }, body, authorization)
|
||||
};
|
||||
try {
|
||||
await proxyCloudApiRequest({
|
||||
const proxyResult = await proxyCloudApiRequest({
|
||||
target,
|
||||
request: upstreamRequest,
|
||||
response,
|
||||
@@ -735,11 +750,36 @@ async function proxyOpencodeRequest({ request, response, url, cloudApiBaseUrl, c
|
||||
timeoutMs: opencodeProxyTimeoutMs,
|
||||
extraResponseHeaders
|
||||
});
|
||||
emitOpencodeProxySpanAsync({
|
||||
traceContext,
|
||||
serviceId,
|
||||
request,
|
||||
url,
|
||||
target,
|
||||
startedAtMs,
|
||||
statusCode: proxyResult?.statusCode || 0,
|
||||
errorCode: proxyResult?.errorCode || "",
|
||||
ticketAuth,
|
||||
sessionStatusCode: session.statusCode || 0,
|
||||
streaming: proxyResult?.streaming === true
|
||||
});
|
||||
} catch (error) {
|
||||
if (response.headersSent || response.writableEnded) {
|
||||
if (!response.writableEnded) response.destroy(error);
|
||||
return;
|
||||
}
|
||||
emitOpencodeProxySpanAsync({
|
||||
traceContext,
|
||||
serviceId,
|
||||
request,
|
||||
url,
|
||||
target,
|
||||
startedAtMs,
|
||||
statusCode: 502,
|
||||
errorCode: error?.timedOut ? "opencode_proxy_timeout" : "opencode_proxy_unavailable",
|
||||
ticketAuth,
|
||||
sessionStatusCode: session.statusCode || 0
|
||||
});
|
||||
sendJsonWithTraceContext(response, 502, {
|
||||
status: "failed",
|
||||
error: "opencode_proxy_unavailable",
|
||||
@@ -876,6 +916,131 @@ function opencodeTraceDiagnostic(traceContext, extra = {}) {
|
||||
};
|
||||
}
|
||||
|
||||
function emitOpencodeProxySpanAsync({ traceContext, serviceId, request, url, target = null, startedAtMs, statusCode = 0, errorCode = "", ticketAuth = null, sessionStatusCode = 0, streaming = false } = {}) {
|
||||
const endpoint = otelTracesEndpoint(process.env);
|
||||
if (!endpoint || !traceContext?.traceId || !traceContext?.spanId) return;
|
||||
const endedAtMs = Date.now();
|
||||
const route = opencodeProxyRoute(url?.pathname || "");
|
||||
const span = {
|
||||
traceId: traceContext.traceId,
|
||||
spanId: traceContext.spanId,
|
||||
parentSpanId: traceContext.parentSpanId || "",
|
||||
name: "opencode.proxy.request",
|
||||
kind: 2,
|
||||
startTimeMs: startedAtMs || endedAtMs,
|
||||
endTimeMs: endedAtMs,
|
||||
statusCode,
|
||||
errorCode,
|
||||
attributes: {
|
||||
"http.request.method": request?.method || "GET",
|
||||
"http.route": route,
|
||||
"http.response.status_code": statusCode || undefined,
|
||||
"http.response.status_class": statusCode ? `${Math.floor(statusCode / 100)}xx` : "unknown",
|
||||
"url.scheme": target?.protocol ? target.protocol.replace(":", "") : undefined,
|
||||
"server.address": target?.hostname || undefined,
|
||||
"server.port": target ? Number(target.port || (target.protocol === "https:" ? 443 : 80)) : undefined,
|
||||
"opencode.proxy.duration_ms": Math.max(0, endedAtMs - (startedAtMs || endedAtMs)),
|
||||
"opencode.proxy.ticket_present": ticketAuth?.ticketPresent === true,
|
||||
"opencode.proxy.ticket_accepted": Boolean(ticketAuth?.ticket),
|
||||
"opencode.proxy.session_status_code": sessionStatusCode || undefined,
|
||||
"opencode.proxy.streaming": streaming === true,
|
||||
valuesPrinted: false
|
||||
}
|
||||
};
|
||||
setTimeout(() => {
|
||||
emitOtelSpan(endpoint, span, serviceId || "hwlab-cloud-web", "hwlab.cloud_web.opencode_proxy").catch((error) => {
|
||||
process.stderr.write(JSON.stringify({
|
||||
event: "cloud-web-opencode-otel-emit-failed",
|
||||
serviceId: serviceId || "hwlab-cloud-web",
|
||||
traceId: traceContext.traceId,
|
||||
requestId: traceContext.requestId,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
valuesPrinted: false
|
||||
}) + "\n");
|
||||
});
|
||||
}, 0);
|
||||
}
|
||||
|
||||
function opencodeProxyRoute(pathname) {
|
||||
const path = String(pathname || "/");
|
||||
if (path === "/global/event") return "/global/event";
|
||||
if (path === "/project" || path === "/project/current" || path === "/project/git/init") return path;
|
||||
if (path === "/provider" || path === "/path" || path === "/global/health") return path;
|
||||
if (/^\/session\/[^/]+\/prompt_async$/u.test(path)) return "/session/:sessionID/prompt_async";
|
||||
if (/^\/session\/[^/]+\/message$/u.test(path)) return "/session/:sessionID/message";
|
||||
if (/^\/session\/[^/]+\/abort$/u.test(path)) return "/session/:sessionID/abort";
|
||||
if (/^\/session\/[^/]+(?:\/.*)?$/u.test(path)) return "/session/:sessionID/*";
|
||||
if (path.startsWith("/assets/")) return "/assets/*";
|
||||
return path;
|
||||
}
|
||||
|
||||
async function emitOtelSpan(endpoint, span, serviceId, scopeName) {
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), 1500);
|
||||
try {
|
||||
await fetch(endpoint, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(otelPayload(span, serviceId, scopeName)),
|
||||
signal: controller.signal
|
||||
});
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
function otelPayload(span, serviceId, scopeName) {
|
||||
return {
|
||||
resourceSpans: [{
|
||||
resource: { attributes: otelAttributes({ "service.name": serviceId, "deployment.environment": process.env.HWLAB_ENVIRONMENT || "v03" }) },
|
||||
scopeSpans: [{
|
||||
scope: { name: scopeName, version: "1" },
|
||||
spans: [{
|
||||
traceId: span.traceId,
|
||||
spanId: span.spanId,
|
||||
parentSpanId: span.parentSpanId || undefined,
|
||||
name: span.name,
|
||||
kind: span.kind,
|
||||
startTimeUnixNano: unixNano(span.startTimeMs),
|
||||
endTimeUnixNano: unixNano(span.endTimeMs),
|
||||
attributes: otelAttributes({ ...span.attributes, "otel.trace_id": span.traceId }),
|
||||
status: span.errorCode || (span.statusCode && span.statusCode >= 500) ? { code: 2, message: span.errorCode || `http_${span.statusCode}` } : { code: 1 }
|
||||
}]
|
||||
}]
|
||||
}]
|
||||
};
|
||||
}
|
||||
|
||||
function otelAttributes(values) {
|
||||
const result = [];
|
||||
for (const [key, value] of Object.entries(values || {})) {
|
||||
if (value === undefined || value === null || value === "") continue;
|
||||
if (typeof value === "number" && Number.isFinite(value)) result.push({ key, value: Number.isInteger(value) ? { intValue: String(value) } : { doubleValue: value } });
|
||||
else if (typeof value === "boolean") result.push({ key, value: { boolValue: value } });
|
||||
else result.push({ key, value: { stringValue: String(value) } });
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function unixNano(ms) {
|
||||
return String(BigInt(Math.max(0, Math.trunc(ms))) * 1000000n);
|
||||
}
|
||||
|
||||
function otelTracesEndpoint(env) {
|
||||
const explicit = firstNonEmpty(env.HWLAB_OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT);
|
||||
if (explicit) return explicit;
|
||||
const base = firstNonEmpty(env.HWLAB_OTEL_EXPORTER_OTLP_ENDPOINT, env.OTEL_EXPORTER_OTLP_ENDPOINT);
|
||||
return base ? `${base.replace(/\/+$/u, "")}/v1/traces` : "";
|
||||
}
|
||||
|
||||
function firstNonEmpty(...values) {
|
||||
for (const value of values) {
|
||||
const text = String(value || "").trim();
|
||||
if (text) return text;
|
||||
}
|
||||
return "";
|
||||
}
|
||||
|
||||
function parseOtelTraceparent(value) {
|
||||
const match = firstHeaderValue(value).trim().match(OTEL_TRACEPARENT_PATTERN);
|
||||
if (!match) return null;
|
||||
|
||||
@@ -705,6 +705,74 @@ test("cloud web OpenCode proxy injects upstream Basic Auth without forwarding HW
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud web OpenCode proxy emits bounded OTLP span", async () => {
|
||||
const otelBodies = [];
|
||||
const otel = createServer(async (request, response) => {
|
||||
let body = "";
|
||||
for await (const chunk of request) body += chunk;
|
||||
otelBodies.push(JSON.parse(body));
|
||||
response.writeHead(200, { "content-type": "application/json" });
|
||||
response.end("{}\n");
|
||||
});
|
||||
await listen(otel);
|
||||
const restoreEnv = withEnv({ OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: `${serverUrl(otel)}/v1/traces` });
|
||||
|
||||
const cloudApi = createServer((request, response) => {
|
||||
request.resume();
|
||||
response.writeHead(200, { "content-type": "application/json" });
|
||||
response.end(JSON.stringify({ authenticated: request.headers.cookie === "hwlab_session=session-a" }));
|
||||
});
|
||||
const opencode = createServer((request, response) => {
|
||||
request.resume();
|
||||
response.writeHead(200, { "content-type": "application/json" });
|
||||
response.end(JSON.stringify({ ok: true }));
|
||||
});
|
||||
await listen(cloudApi);
|
||||
await listen(opencode);
|
||||
|
||||
const cloudWeb = createCloudWebServer({
|
||||
serviceId: "hwlab-cloud-web",
|
||||
roots: [],
|
||||
cloudApiBaseUrl: serverUrl(cloudApi),
|
||||
cloudApiProxyTimeoutMs: 1000,
|
||||
opencodeUpstreamUrl: serverUrl(opencode),
|
||||
opencodeProxyHost: "127.0.0.1",
|
||||
opencodeProxyTimeoutMs: 1000,
|
||||
opencodeUsername: "oc_user",
|
||||
opencodePassword: "oc_password",
|
||||
healthPayload: () => ({ status: "ok" }),
|
||||
sendJson(response, statusCode, body) {
|
||||
const payload = JSON.stringify(body);
|
||||
response.writeHead(statusCode, { "content-type": "application/json", "content-length": Buffer.byteLength(payload) });
|
||||
response.end(payload);
|
||||
}
|
||||
});
|
||||
await listen(cloudWeb);
|
||||
|
||||
try {
|
||||
const response = await fetch(`${serverUrl(cloudWeb)}/session/ses_otel/prompt_async`, {
|
||||
method: "POST",
|
||||
headers: { accept: "application/json", "content-type": "application/json", cookie: "hwlab_session=session-a" },
|
||||
body: JSON.stringify({ parts: [{ type: "text", text: "hi" }] })
|
||||
});
|
||||
assert.equal(response.status, 200);
|
||||
await waitFor(() => otelBodies.length > 0, "expected OpenCode proxy OTel span");
|
||||
const spans = otelBodies.flatMap((body) => body.resourceSpans?.flatMap((resourceSpan) => resourceSpan.scopeSpans?.flatMap((scopeSpan) => scopeSpan.spans ?? []) ?? []) ?? []);
|
||||
const span = spans.find((item) => item.name === "opencode.proxy.request");
|
||||
assert.ok(span, "expected opencode.proxy.request span");
|
||||
const attrs = new Map((span.attributes ?? []).map((entry) => [entry.key, entry.value?.stringValue ?? entry.value?.intValue ?? entry.value?.boolValue]));
|
||||
assert.equal(attrs.get("http.route"), "/session/:sessionID/prompt_async");
|
||||
assert.equal(attrs.get("http.response.status_class"), "2xx");
|
||||
assert.equal(attrs.get("valuesPrinted"), false);
|
||||
} finally {
|
||||
restoreEnv();
|
||||
await close(cloudWeb);
|
||||
await close(opencode);
|
||||
await close(cloudApi);
|
||||
await close(otel);
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud web OpenCode proxy accepts short-lived tickets minted by the shell", async () => {
|
||||
const restoreEnv = withEnv({
|
||||
HWLAB_CLOUD_WEB_DISPLAY_TIME_ZONE: "Asia/Shanghai",
|
||||
@@ -829,6 +897,15 @@ function close(server) {
|
||||
});
|
||||
}
|
||||
|
||||
async function waitFor(predicate, message, timeoutMs = 1000) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (predicate()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
assert.fail(message);
|
||||
}
|
||||
|
||||
function disconnectDuringResponse(baseUrl, pathname) {
|
||||
const url = new URL(pathname, baseUrl);
|
||||
return new Promise((resolve, reject) => {
|
||||
|
||||
@@ -0,0 +1,318 @@
|
||||
#!/usr/bin/env node
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { createServer, request as httpRequest } from "node:http";
|
||||
import { request as httpsRequest } from "node:https";
|
||||
|
||||
const serviceId = process.env.OTEL_SERVICE_NAME || process.env.HWLAB_SERVICE_ID || "opencode-provider-proxy";
|
||||
const port = parsePositiveInteger(process.env.PORT || process.env.HWLAB_OPENCODE_PROVIDER_PROXY_PORT, 4097);
|
||||
const upstreamBaseUrl = new URL(requiredEnv("HWLAB_OPENCODE_PROVIDER_PROXY_UPSTREAM_BASE_URL"));
|
||||
const publicBasePath = normalizeBasePath(process.env.HWLAB_OPENCODE_PROVIDER_PROXY_PUBLIC_BASE_PATH || "/v1");
|
||||
const timeoutMs = parsePositiveInteger(process.env.HWLAB_OPENCODE_PROVIDER_PROXY_TIMEOUT_MS, 600000);
|
||||
const otelEndpoint = otelTracesEndpoint(process.env);
|
||||
const startedAt = new Date();
|
||||
|
||||
const hopByHopHeaders = new Set(["connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "te", "trailer", "transfer-encoding", "upgrade"]);
|
||||
|
||||
const server = createServer((clientReq, clientRes) => {
|
||||
const url = new URL(clientReq.url || "/", "http://opencode-provider-proxy.local");
|
||||
if (url.pathname === "/health" || url.pathname === "/health/live" || url.pathname === "/health/readiness") {
|
||||
sendJson(clientRes, 200, healthPayload());
|
||||
return;
|
||||
}
|
||||
proxyProviderRequest(clientReq, clientRes, url).catch((error) => {
|
||||
if (!clientRes.headersSent) {
|
||||
sendJson(clientRes, 502, {
|
||||
error: "opencode_provider_proxy_failed",
|
||||
message: "OpenCode provider proxy failed",
|
||||
reason: error instanceof Error ? error.message : String(error)
|
||||
});
|
||||
} else {
|
||||
clientRes.destroy(error);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
server.listen(port, "0.0.0.0", () => {
|
||||
process.stdout.write(JSON.stringify({ serviceId, status: "listening", port, upstreamHost: upstreamBaseUrl.host, publicBasePath, valuesPrinted: false }) + "\n");
|
||||
});
|
||||
|
||||
async function proxyProviderRequest(clientReq, clientRes, url) {
|
||||
const trace = traceContextFromHeaders(clientReq.headers);
|
||||
const startSpanId = randomOtelSpanId();
|
||||
const started = process.hrtime.bigint();
|
||||
const startTime = Date.now();
|
||||
const target = providerTargetUrl(url);
|
||||
const route = providerRoute(url.pathname);
|
||||
const attrs = {
|
||||
"http.request.method": clientReq.method || "GET",
|
||||
"http.route": route,
|
||||
"url.scheme": target.protocol.replace(":", ""),
|
||||
"server.address": target.hostname,
|
||||
"server.port": Number(target.port || (target.protocol === "https:" ? 443 : 80)),
|
||||
"opencode.provider.upstream_host": upstreamBaseUrl.host,
|
||||
"opencode.provider.public_base_path": publicBasePath,
|
||||
valuesPrinted: false
|
||||
};
|
||||
|
||||
emitOtelSpanAsync({
|
||||
name: "opencode.provider.request.start",
|
||||
traceId: trace.traceId,
|
||||
spanId: startSpanId,
|
||||
parentSpanId: trace.parentSpanId,
|
||||
kind: 1,
|
||||
startTimeMs: startTime,
|
||||
endTimeMs: startTime,
|
||||
attributes: { ...attrs, "opencode.provider.phase": "start" }
|
||||
});
|
||||
|
||||
clientRes.setHeader("traceparent", traceparent(trace.traceId, startSpanId));
|
||||
clientRes.setHeader("x-hwlab-otel-trace-id", trace.traceId);
|
||||
|
||||
await new Promise((resolve) => {
|
||||
let settled = false;
|
||||
let statusCode = 0;
|
||||
let errorCode = "";
|
||||
let timeout = null;
|
||||
const settle = () => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
if (timeout) clearTimeout(timeout);
|
||||
const elapsedMs = Number((process.hrtime.bigint() - started) / 1000000n);
|
||||
emitOtelSpanAsync({
|
||||
name: "opencode.provider.request",
|
||||
traceId: trace.traceId,
|
||||
spanId: randomOtelSpanId(),
|
||||
parentSpanId: startSpanId,
|
||||
kind: 3,
|
||||
startTimeMs: startTime,
|
||||
endTimeMs: startTime + Math.max(elapsedMs, 0),
|
||||
statusCode,
|
||||
errorCode,
|
||||
attributes: {
|
||||
...attrs,
|
||||
"http.response.status_code": statusCode || undefined,
|
||||
"http.response.status_class": statusCode ? `${Math.floor(statusCode / 100)}xx` : "unknown",
|
||||
"opencode.provider.elapsed_ms": elapsedMs,
|
||||
"opencode.provider.timeout_ms": timeoutMs,
|
||||
"opencode.provider.phase": errorCode ? "failed" : "complete"
|
||||
}
|
||||
});
|
||||
resolve();
|
||||
};
|
||||
|
||||
const requestImpl = target.protocol === "https:" ? httpsRequest : httpRequest;
|
||||
const upstreamReq = requestImpl({
|
||||
protocol: target.protocol,
|
||||
hostname: target.hostname,
|
||||
port: target.port || (target.protocol === "https:" ? 443 : 80),
|
||||
method: clientReq.method,
|
||||
path: `${target.pathname}${target.search}`,
|
||||
headers: providerRequestHeaders(clientReq.headers, target, trace, startSpanId)
|
||||
}, (upstreamRes) => {
|
||||
statusCode = upstreamRes.statusCode || 502;
|
||||
clientRes.writeHead(statusCode, {
|
||||
...providerResponseHeaders(upstreamRes.headers),
|
||||
traceparent: traceparent(trace.traceId, startSpanId),
|
||||
"x-hwlab-otel-trace-id": trace.traceId
|
||||
});
|
||||
upstreamRes.pipe(clientRes);
|
||||
upstreamRes.on("end", settle);
|
||||
upstreamRes.on("close", settle);
|
||||
upstreamRes.on("error", (error) => {
|
||||
errorCode = error?.code || "upstream_response_error";
|
||||
if (!clientRes.writableEnded) clientRes.destroy(error);
|
||||
settle();
|
||||
});
|
||||
});
|
||||
|
||||
timeout = setTimeout(() => {
|
||||
errorCode = "upstream_timeout";
|
||||
upstreamReq.destroy(new Error(`opencode provider proxy timed out after ${timeoutMs}ms`));
|
||||
}, timeoutMs);
|
||||
|
||||
upstreamReq.on("error", (error) => {
|
||||
if (!errorCode) errorCode = error?.code || "upstream_request_error";
|
||||
if (!clientRes.headersSent) {
|
||||
statusCode = errorCode === "upstream_timeout" ? 504 : 502;
|
||||
sendJson(clientRes, statusCode, {
|
||||
error: "opencode_provider_upstream_error",
|
||||
message: "OpenCode provider upstream request failed",
|
||||
reason: error instanceof Error ? error.message : String(error),
|
||||
traceId: trace.traceId,
|
||||
valuesPrinted: false
|
||||
});
|
||||
} else if (!clientRes.writableEnded) {
|
||||
clientRes.destroy(error);
|
||||
}
|
||||
settle();
|
||||
});
|
||||
|
||||
clientReq.on("error", (error) => {
|
||||
errorCode = error?.code || "client_request_error";
|
||||
upstreamReq.destroy(error);
|
||||
});
|
||||
clientReq.pipe(upstreamReq);
|
||||
});
|
||||
}
|
||||
|
||||
function healthPayload() {
|
||||
return { serviceId, status: "ok", port, upstreamHost: upstreamBaseUrl.host, publicBasePath, startedAt: startedAt.toISOString(), uptimeSeconds: Math.round((Date.now() - startedAt.getTime()) / 1000), otelEnabled: Boolean(otelEndpoint), valuesPrinted: false };
|
||||
}
|
||||
|
||||
function providerTargetUrl(url) {
|
||||
const suffix = stripBasePath(url.pathname, publicBasePath);
|
||||
const basePath = upstreamBaseUrl.pathname.replace(/\/+$/u, "");
|
||||
const nextPath = suffix ? `${basePath}/${suffix.replace(/^\/+/, "")}` : basePath || "/";
|
||||
const target = new URL(upstreamBaseUrl.toString());
|
||||
target.pathname = nextPath;
|
||||
target.search = url.search;
|
||||
return target;
|
||||
}
|
||||
|
||||
function stripBasePath(pathname, basePath) {
|
||||
if (!basePath || basePath === "/") return pathname;
|
||||
if (pathname === basePath) return "/";
|
||||
if (pathname.startsWith(`${basePath}/`)) return pathname.slice(basePath.length) || "/";
|
||||
return pathname;
|
||||
}
|
||||
|
||||
function providerRoute(pathname) {
|
||||
const stripped = stripBasePath(pathname, publicBasePath);
|
||||
if (/^\/chat\/completions\/?$/u.test(stripped)) return `${publicBasePath}/chat/completions`;
|
||||
if (/^\/models\/?$/u.test(stripped)) return `${publicBasePath}/models`;
|
||||
return `${publicBasePath}${stripped.startsWith("/") ? stripped : `/${stripped}`}`;
|
||||
}
|
||||
|
||||
function providerRequestHeaders(headers, target, trace, parentSpanId) {
|
||||
const result = {};
|
||||
for (const [name, value] of Object.entries(headers)) {
|
||||
const lower = name.toLowerCase();
|
||||
if (hopByHopHeaders.has(lower) || lower === "host") continue;
|
||||
if (value !== undefined) result[lower] = value;
|
||||
}
|
||||
result.host = target.host;
|
||||
result.traceparent = traceparent(trace.traceId, parentSpanId);
|
||||
result["x-hwlab-otel-trace-id"] = trace.traceId;
|
||||
result["x-source-service-id"] = serviceId;
|
||||
return result;
|
||||
}
|
||||
|
||||
function providerResponseHeaders(headers) {
|
||||
const result = {};
|
||||
for (const [name, value] of Object.entries(headers)) {
|
||||
const lower = name.toLowerCase();
|
||||
if (hopByHopHeaders.has(lower) || lower === "content-length") continue;
|
||||
if (value !== undefined) result[name] = value;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function sendJson(response, statusCode, body) {
|
||||
const payload = JSON.stringify(body);
|
||||
response.writeHead(statusCode, { "content-type": "application/json; charset=utf-8", "content-length": Buffer.byteLength(payload) });
|
||||
response.end(payload);
|
||||
}
|
||||
|
||||
function emitOtelSpanAsync(span) {
|
||||
if (!otelEndpoint) return;
|
||||
setTimeout(() => {
|
||||
emitOtelSpan(span).catch((error) => {
|
||||
process.stderr.write(JSON.stringify({ event: "opencode-provider-proxy-otel-emit-failed", serviceId, error: error instanceof Error ? error.message : String(error), valuesPrinted: false }) + "\n");
|
||||
});
|
||||
}, 0);
|
||||
}
|
||||
|
||||
async function emitOtelSpan(span) {
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), 1500);
|
||||
try {
|
||||
await fetch(otelEndpoint, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(otelPayload(span)), signal: controller.signal });
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
function otelPayload(span) {
|
||||
return { resourceSpans: [{ resource: { attributes: otelAttributes({ "service.name": serviceId, "deployment.environment": process.env.HWLAB_ENVIRONMENT || "v03" }) }, scopeSpans: [{ scope: { name: "hwlab.opencode.provider_proxy", version: "1" }, spans: [{ traceId: span.traceId, spanId: span.spanId, parentSpanId: span.parentSpanId || undefined, name: span.name, kind: span.kind, startTimeUnixNano: unixNano(span.startTimeMs), endTimeUnixNano: unixNano(span.endTimeMs), attributes: otelAttributes({ ...span.attributes, "otel.trace_id": span.traceId }), status: span.errorCode || (span.statusCode && span.statusCode >= 500) ? { code: 2, message: span.errorCode || `http_${span.statusCode}` } : { code: 1 } }] }] }] };
|
||||
}
|
||||
|
||||
function otelAttributes(values) {
|
||||
const result = [];
|
||||
for (const [key, value] of Object.entries(values || {})) {
|
||||
if (value === undefined || value === null || value === "") continue;
|
||||
if (typeof value === "number" && Number.isFinite(value)) result.push({ key, value: Number.isInteger(value) ? { intValue: String(value) } : { doubleValue: value } });
|
||||
else if (typeof value === "boolean") result.push({ key, value: { boolValue: value } });
|
||||
else result.push({ key, value: { stringValue: String(value) } });
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function unixNano(ms) {
|
||||
return String(BigInt(Math.max(0, Math.trunc(ms))) * 1000000n);
|
||||
}
|
||||
|
||||
function traceContextFromHeaders(headers) {
|
||||
const parsed = parseTraceparent(headers.traceparent);
|
||||
return { traceId: parsed?.traceId || normalizedTraceId(headers["x-hwlab-otel-trace-id"]) || randomOtelTraceId(), parentSpanId: parsed?.spanId || "" };
|
||||
}
|
||||
|
||||
function parseTraceparent(value) {
|
||||
const match = firstHeaderValue(value).trim().match(/^00-([0-9a-f]{32})-([0-9a-f]{16})-[0-9a-f]{2}$/u);
|
||||
if (!match) return null;
|
||||
const traceId = normalizedTraceId(match[1]);
|
||||
const spanId = normalizedSpanId(match[2]);
|
||||
return traceId && spanId ? { traceId, spanId } : null;
|
||||
}
|
||||
|
||||
function traceparent(traceId, spanId) {
|
||||
return `00-${traceId}-${spanId}-01`;
|
||||
}
|
||||
|
||||
function randomOtelTraceId() {
|
||||
const value = randomBytes(16).toString("hex");
|
||||
return value === "00000000000000000000000000000000" ? "00000000000000000000000000000001" : value;
|
||||
}
|
||||
|
||||
function randomOtelSpanId() {
|
||||
const value = randomBytes(8).toString("hex");
|
||||
return value === "0000000000000000" ? "0000000000000001" : value;
|
||||
}
|
||||
|
||||
function normalizedTraceId(value) {
|
||||
const text = firstHeaderValue(value).trim().toLowerCase();
|
||||
return /^[0-9a-f]{32}$/u.test(text) && text !== "00000000000000000000000000000000" ? text : "";
|
||||
}
|
||||
|
||||
function normalizedSpanId(value) {
|
||||
const text = firstHeaderValue(value).trim().toLowerCase();
|
||||
return /^[0-9a-f]{16}$/u.test(text) && text !== "0000000000000000" ? text : "";
|
||||
}
|
||||
|
||||
function firstHeaderValue(value) {
|
||||
if (Array.isArray(value)) return String(value[0] ?? "");
|
||||
return String(value ?? "");
|
||||
}
|
||||
|
||||
function otelTracesEndpoint(env) {
|
||||
const explicit = String(env.HWLAB_OTEL_EXPORTER_OTLP_TRACES_ENDPOINT || env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT || "").trim();
|
||||
if (explicit) return explicit;
|
||||
const base = String(env.HWLAB_OTEL_EXPORTER_OTLP_ENDPOINT || env.OTEL_EXPORTER_OTLP_ENDPOINT || "").trim();
|
||||
return base ? `${base.replace(/\/+$/u, "")}/v1/traces` : "";
|
||||
}
|
||||
|
||||
function normalizeBasePath(value) {
|
||||
const text = String(value || "/").trim();
|
||||
const normalized = `/${text.replace(/^\/+|\/+$/gu, "")}`;
|
||||
return normalized === "/" ? "/" : normalized;
|
||||
}
|
||||
|
||||
function parsePositiveInteger(value, fallback) {
|
||||
const parsed = Number.parseInt(String(value ?? ""), 10);
|
||||
return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
|
||||
}
|
||||
|
||||
function requiredEnv(name) {
|
||||
const value = String(process.env[name] || "").trim();
|
||||
if (!value) throw new Error(`${name} is required`);
|
||||
return value;
|
||||
}
|
||||
Reference in New Issue
Block a user