From 69b8e03661d15cfc6f034ceb1d0bff7238c50127 Mon Sep 17 00:00:00 2001 From: lyon Date: Fri, 19 Jun 2026 23:05:50 +0800 Subject: [PATCH] feat: add code agent otel spans --- deploy/deploy.yaml | 2 + internal/cloud/code-agent-agentrun-adapter.ts | 14 ++ internal/cloud/otel-trace.ts | 133 ++++++++++++++++++ internal/cloud/server-code-agent-http.ts | 52 ++++++- 4 files changed, 200 insertions(+), 1 deletion(-) create mode 100644 internal/cloud/otel-trace.ts diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index 78e6be0d..85020d22 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -457,6 +457,8 @@ lanes: HWLAB_KEYCLOAK_CLIENT_ID: hwlab-cloud-web HWLAB_KEYCLOAK_CLIENT_SECRET: secretRef:hwlab-cloud-web-client/client-secret HWLAB_CODE_AGENT_ADAPTER: agentrun-v01 + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: http://otel-collector.platform-infra.svc.cluster.local:4318/v1/traces + OTEL_SERVICE_NAME: hwlab-cloud-api AGENTRUN_MGR_URL: http://agentrun-mgr.agentrun-v02.svc.cluster.local:8080 AGENTRUN_API_KEY: secretRef:hwlab-v03-master-server-admin-api-key/api-key HWLAB_CODE_AGENT_AGENTRUN_RUNNER_NAMESPACE: agentrun-v02 diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index e5ecfba9..4cd4bf29 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -16,6 +16,7 @@ import { } from "./server-http-utils.ts"; import { backendPerformanceRouteTemplate, currentBackendPerformanceContext } from "./backend-performance.ts"; import { agentRunResultSyncDeferred, buildAgentRunProjectionEventsFetchPlan, buildWorkbenchProjectionStateUpdate } from "./workbench-projection-cursor.ts"; +import { codeAgentOtelTraceFields, codeAgentOtelTraceContext, emitCodeAgentOtelSpan } from "./otel-trace.ts"; const ADAPTER_ID = "agentrun-v01"; const DEFAULT_AGENTRUN_MGR_URL = "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080"; @@ -226,6 +227,7 @@ export function initialAgentRunChatResult({ params = {}, options = {}, traceId } status: "running", shortConnection: true, traceId, + ...codeAgentOtelTraceFields(traceId, env), projectId: firstNonEmpty(params.projectId) || null, conversationId: safeConversationId(params.conversationId) || null, sessionId: safeSessionId(params.sessionId) || agentRunSessionId(traceId), @@ -276,6 +278,7 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI waitingFor: "agentrun-run-reuse-or-create", adapter: ADAPTER_ID, managerHost: new URL(managerUrl).hostname, + ...codeAgentOtelTraceFields(traceId, env), ...agentRunPromptEventFields(promptEvidence), valuesPrinted: false }, backendProfile), agentRunTraceMeta(env, params)); @@ -386,8 +389,10 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI throw error; } const runInput = buildAgentRunCreateRunInput({ params, env, traceId, backendProfile, sessionId, toolCapabilities }); + const runCreateStartedAt = Date.now(); run = await agentRunJson(fetchImpl, managerUrl, "/api/v1/runs", { method: "POST", body: runInput, timeoutMs, env }); const runId = requiredString(run?.id, "run.id"); + void emitCodeAgentOtelSpan("agentrun_run_create", traceId, env, { startTimeMs: runCreateStartedAt, attributes: { runId, backendProfile, managerUrl } }); traceStore.append(traceId, agentRunTraceEvent({ type: "backend", status: "running", label: "agentrun:run:created", @@ -395,8 +400,10 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI runId, backendProfile, waitingFor: "agentrun-command-create", valuesPrinted: false, }, backendProfile)); const commandInput = buildAgentRunCommandInput({ params, traceId, backendProfile, sessionId }); + const commandCreateStartedAt = Date.now(); command = await agentRunJson(fetchImpl, managerUrl, "/api/v1/runs/" + encodeURIComponent(runId) + "/commands", { method: "POST", body: commandInput, timeoutMs, env }); const commandId = requiredString(command?.id, "command.id"); + void emitCodeAgentOtelSpan("agentrun_command_create", traceId, env, { startTimeMs: commandCreateStartedAt, attributes: { runId, commandId, backendProfile, managerUrl } }); traceStore.append(traceId, agentRunTraceEvent({ type: "backend", status: "running", label: "agentrun:command:created", @@ -405,7 +412,9 @@ export async function submitAgentRunChatTurn({ params = {}, options = {}, traceI }, backendProfile)); const runnerJobInput = buildAgentRunRunnerJobInput({ env, traceId, commandId, ownerApiKey, toolCapabilities, backendProfile }); try { + const runnerJobStartedAt = Date.now(); runnerJob = await agentRunJson(fetchImpl, managerUrl, "/api/v1/runs/" + encodeURIComponent(runId) + "/runner-jobs", { method: "POST", body: runnerJobInput, timeoutMs, env }); + void emitCodeAgentOtelSpan("agentrun_runner_job_create", traceId, env, { startTimeMs: runnerJobStartedAt, attributes: { runId, commandId, backendProfile, managerUrl, jobName: runnerJob?.jobName ?? runnerJob?.jobIdentity?.name ?? null } }); } catch (error) { if (attempt === 0 && await shouldResetSessionAfterEviction("session-store-evicted", error?.message)) { sessionReset = true; @@ -1017,6 +1026,8 @@ function buildAgentRunCreateRunInput({ params, env, traceId, backendProfile, ses metadata: { adapter: ADAPTER_ID, hwlabTraceId: traceId, + otelTraceId: codeAgentOtelTraceContext(traceId).traceId, + traceparent: codeAgentOtelTraceContext(traceId).traceparent, hwlabApi: "/v1/agent/chat", hwlabProjectId, hwlabSessionId: safeSessionId(params.sessionId) || null, @@ -1044,6 +1055,7 @@ function buildAgentRunCreateRunInput({ params, env, traceId, backendProfile, ses traceSink: { kind: "hwlab-cloud-api", traceId, + ...codeAgentOtelTraceFields(traceId, env), resultUrl: `/v1/agent/chat/result/${encodeURIComponent(traceId)}`, traceUrl: `/v1/agent/traces/${encodeURIComponent(traceId)}`, valuesPrinted: false @@ -1946,6 +1958,7 @@ async function fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, e const limit = parsePositiveInteger(mapping.eventsPageLimit ?? env?.HWLAB_CODE_AGENT_AGENTRUN_EVENTS_PAGE_LIMIT, 500); const path = `/api/v1/runs/${encodeURIComponent(runId)}/events?afterSeq=${encodeURIComponent(String(afterSeq))}&limit=${encodeURIComponent(String(limit))}`; const response = await agentRunJson(fetchImpl, managerUrl, path, { method: "GET", timeoutMs, env }); + void emitCodeAgentOtelSpan("projection_sync", mapping.traceId ?? mapping.hwlabTraceId ?? mapping.businessTraceId, env, { attributes: { runId, commandId: currentCommandId || null, afterSeq, rawEventCount: Array.isArray(response?.items) ? response.items.length : 0 } }); const rawEvents = Array.isArray(response?.items) ? response.items : []; const events = rawEvents.filter((event) => agentRunEventBelongsToTrace(event, { currentCommandId, afterSeq, endSeq })); return { @@ -2480,6 +2493,7 @@ function agentRunMapping({ env, managerUrl, backendProfile, run, command, runner terminalStatus: null, lastSeq: 0, traceId, + ...codeAgentOtelTraceFields(traceId, env), sessionId: run?.sessionRef?.sessionId ?? null, conversationId: run?.sessionRef?.conversationId ?? null, threadId: run?.sessionRef?.threadId ?? null, diff --git a/internal/cloud/otel-trace.ts b/internal/cloud/otel-trace.ts new file mode 100644 index 00000000..0652bfec --- /dev/null +++ b/internal/cloud/otel-trace.ts @@ -0,0 +1,133 @@ +import { createHash, randomBytes } from "node:crypto"; + +const OTLP_TIMEOUT_MS = 1500; +const ZERO_TRACE_ID = "00000000000000000000000000000000"; +const ZERO_SPAN_ID = "0000000000000000"; + +export function codeAgentOtelTraceFields(traceId, env = process.env) { + const context = codeAgentOtelTraceContext(traceId); + return { + otelTraceId: context.traceId, + traceparent: context.traceparent, + otelTrace: { + traceId: context.traceId, + businessTraceId: context.businessTraceId, + tempoPath: `/api/traces/${context.traceId}`, + exporterConfigured: Boolean(resolveOtlpTracesEndpoint(env)), + valuesPrinted: false + } + }; +} + +export function codeAgentOtelTraceContext(traceId) { + const businessTraceId = String(traceId ?? "").trim() || "trc_unassigned"; + const otelTraceId = nonZeroHex(createHash("sha256").update(`hwlab-code-agent:${businessTraceId}`).digest("hex").slice(0, 32), ZERO_TRACE_ID); + const parentSpanId = nonZeroHex(createHash("sha256").update(`hwlab-code-agent-parent:${businessTraceId}`).digest("hex").slice(0, 16), ZERO_SPAN_ID); + return { + businessTraceId, + traceId: otelTraceId, + parentSpanId, + traceparent: `00-${otelTraceId}-${parentSpanId}-01`, + valuesPrinted: false + }; +} + +export async function emitCodeAgentOtelSpan(name, traceId, env = process.env, options = {}) { + const endpoint = resolveOtlpTracesEndpoint(env); + if (!endpoint || typeof fetch !== "function") return { ok: false, skipped: true, reason: "otlp-endpoint-missing", valuesPrinted: false }; + const context = codeAgentOtelTraceContext(traceId); + const startedAtMs = Number.isFinite(Number(options.startTimeMs)) ? Number(options.startTimeMs) : Date.now(); + const endedAtMs = Number.isFinite(Number(options.endTimeMs)) ? Number(options.endTimeMs) : Date.now(); + const spanId = nonZeroHex(String(options.spanId ?? randomBytes(8).toString("hex")).slice(0, 16), ZERO_SPAN_ID); + const parentSpanId = typeof options.parentSpanId === "string" && /^[0-9a-f]{16}$/u.test(options.parentSpanId) ? options.parentSpanId : context.parentSpanId; + const statusCode = options.status === "error" || options.error ? 2 : 1; + const body = { + resourceSpans: [{ + resource: { attributes: attributesFromRecord(resourceAttributes(env)) }, + scopeSpans: [{ + scope: { name: "hwlab.code-agent", version: "1" }, + spans: [{ + traceId: context.traceId, + spanId, + parentSpanId, + name, + kind: Number(options.kind ?? 1), + startTimeUnixNano: unixNano(startedAtMs), + endTimeUnixNano: unixNano(Math.max(startedAtMs, endedAtMs)), + attributes: attributesFromRecord({ + traceId: context.businessTraceId, + "otel.trace_id": context.traceId, + "code_agent.stage": name, + ...options.attributes + }), + status: { + code: statusCode, + ...(options.error ? { message: String(options.error?.message ?? options.error).slice(0, 300) } : {}) + } + }] + }] + }] + }; + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), OTLP_TIMEOUT_MS); + try { + const response = await fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + signal: controller.signal + }); + return { ok: response.ok, status: response.status, valuesPrinted: false }; + } catch (error) { + return { ok: false, error: error?.name === "AbortError" ? "otlp-timeout" : "otlp-send-failed", valuesPrinted: false }; + } finally { + clearTimeout(timeout); + } +} + +function resolveOtlpTracesEndpoint(env = process.env) { + const explicit = firstNonEmpty(env.HWLAB_OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT); + if (explicit) return explicit.replace(/\/+$/u, ""); + const base = firstNonEmpty(env.HWLAB_OTEL_EXPORTER_OTLP_ENDPOINT, env.OTEL_EXPORTER_OTLP_ENDPOINT); + return base ? `${base.replace(/\/+$/u, "")}/v1/traces` : null; +} + +function resourceAttributes(env = process.env) { + return { + "service.name": firstNonEmpty(env.OTEL_SERVICE_NAME, "hwlab-cloud-api"), + "deployment.environment": firstNonEmpty(env.HWLAB_ENVIRONMENT, env.HWLAB_GITOPS_PROFILE, env.HWLAB_RUNTIME_LANE, "unknown"), + "unidesk.node": firstNonEmpty(env.HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID, env.HWLAB_NODE_ID, env.UNIDESK_NODE_ID, "unknown"), + "hwlab.lane": firstNonEmpty(env.HWLAB_RUNTIME_LANE, env.HWLAB_GITOPS_PROFILE, env.HWLAB_BOOT_REF, "unknown"), + "k8s.namespace.name": firstNonEmpty(env.HWLAB_RUNTIME_NAMESPACE, env.POD_NAMESPACE, env.HWLAB_NAMESPACE, "unknown"), + "git.commit": firstNonEmpty(env.HWLAB_COMMIT_ID, env.HWLAB_GITOPS_SOURCE_COMMIT, env.HWLAB_REVISION, "unknown") + }; +} + +function attributesFromRecord(record = {}) { + return Object.entries(record) + .filter(([, value]) => value !== undefined && value !== null) + .map(([key, value]) => ({ key, value: otlpAnyValue(value) })); +} + +function otlpAnyValue(value) { + if (typeof value === "boolean") return { boolValue: value }; + if (typeof value === "number" && Number.isInteger(value)) return { intValue: String(value) }; + if (typeof value === "number" && Number.isFinite(value)) return { doubleValue: value }; + return { stringValue: typeof value === "string" ? value : JSON.stringify(value) }; +} + +function unixNano(ms) { + return String(Math.trunc(ms * 1_000_000)); +} + +function firstNonEmpty(...values) { + for (const value of values) { + const text = String(value ?? "").trim(); + if (text) return text; + } + return null; +} + +function nonZeroHex(value, zero) { + return /^[0-9a-f]+$/u.test(value) && value !== zero ? value : zero.replace(/0$/u, "1"); +} diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 6b9421fd..5dffb8e9 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -21,6 +21,7 @@ import { scheduleWorkbenchProjectionFinalizer } from "./workbench-projection-fin import { writeWorkbenchProjectionSession } from "./workbench-projection-writer.ts"; import { buildAgentRunProjectionEventsFetchPlan } from "./workbench-projection-cursor.ts"; import { createWorkbenchReadModel } from "./workbench-read-model.ts"; +import { codeAgentOtelTraceFields, emitCodeAgentOtelSpan } from "./otel-trace.ts"; import { firstHeaderValue, getHeader, @@ -127,12 +128,18 @@ export async function handleCodeAgentChatHttp(request, response, options) { } const traceUrl = `/v1/agent/traces/${encodeURIComponent(traceId)}`; const turnUrl = `/v1/agent/turns/${encodeURIComponent(traceId)}`; + const statusCode = 202; + void emitCodeAgentOtelSpan("POST /v1/agent/chat", traceId, options.env ?? process.env, { + kind: 2, + attributes: { "http.method": "POST", "http.route": "/v1/agent/chat", "http.status_code": statusCode, sessionId: safeSessionId(nativeSessionChatParams.sessionId) || null, turnId: lifecycle.turnId } + }); sendJson(response, 202, { accepted: true, status: "running", shortConnection: true, controlSemantics: "submit-and-poll", traceId, + ...codeAgentOtelTraceFields(traceId, options.env ?? process.env), ...lifecycle, sessionId: safeSessionId(nativeSessionChatParams.sessionId) || null, traceUrl, @@ -749,6 +756,7 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { accepted: true, status: "running", traceId, + ...codeAgentOtelTraceFields(traceId, process.env), ...codeAgentTurnLifecycleFields(traceId, params), conversationId: safeConversationId(params.conversationId) || null, sessionId: safeSessionId(params.sessionId) || null, @@ -765,7 +773,14 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { try { initial = initialAgentRunChatResult({ params, options, traceId }); results.set(traceId, annotateOwner(initial, params)); + const billingStartedAt = Date.now(); const billingPreflight = await preflightCodeAgentBilling({ params, options, traceId, sendResponse: false }); + void emitCodeAgentOtelSpan("billing_preflight", traceId, options.env ?? process.env, { + startTimeMs: billingStartedAt, + attributes: { sessionId: safeSessionId(params.sessionId) || null, turnId: codeAgentTurnLifecycleFields(traceId, params).turnId, blocked: Boolean(billingPreflight?.blocked) }, + status: billingPreflight?.blocked ? "error" : "ok", + error: billingPreflight?.blocked ? billingPreflight?.payload?.error?.message ?? billingPreflight?.blocker?.summary ?? "billing preflight blocked" : null + }); if (billingPreflight?.blocked) { await settleAdmittedCodeAgentFailure({ base: initial, @@ -782,14 +797,27 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { initial = withCodeAgentBillingReservation(initial, dispatchParams); results.set(traceId, annotateOwner(initial, dispatchParams)); executionOptions = { ...options, ...(await codeAgentChatExecutionOptions(options, dispatchParams)) }; + const dispatchStartedAt = Date.now(); const payload = await submitAgentRunChatTurn({ params: dispatchParams, options: executionOptions, traceId, traceStore, results }); + void emitCodeAgentOtelSpan("agentrun_dispatch", traceId, executionOptions.env ?? process.env, { + startTimeMs: dispatchStartedAt, + attributes: { sessionId: safeSessionId(dispatchParams.sessionId) || null, turnId: codeAgentTurnLifecycleFields(traceId, dispatchParams).turnId, runId: payload?.agentRun?.runId ?? null, commandId: payload?.agentRun?.commandId ?? null, providerProfile: dispatchParams.providerProfile ?? null } + }); if (isCodeAgentResultCanceled(results.get(traceId))) return; const owned = annotateOwner(withCodeAgentBillingReservation(payload, dispatchParams), dispatchParams); await recordCodeAgentSessionOwner({ payload: owned, params: dispatchParams, options: executionOptions, status: "running" }); results.set(traceId, owned); scheduleAgentRunProjectionSync({ traceId, params: dispatchParams, options: executionOptions, traceStore, currentResult: owned }); + void emitCodeAgentOtelSpan("projection_write", traceId, executionOptions.env ?? process.env, { + attributes: { sessionId: safeSessionId(dispatchParams.sessionId) || null, turnId: codeAgentTurnLifecycleFields(traceId, dispatchParams).turnId, runId: owned?.agentRun?.runId ?? null, commandId: owned?.agentRun?.commandId ?? null, projection: "scheduled" } + }); } catch (error) { if (isCodeAgentResultCanceled(results.get(traceId))) return; + void emitCodeAgentOtelSpan("agentrun_dispatch", traceId, executionOptions.env ?? process.env, { + status: "error", + error, + attributes: { sessionId: safeSessionId(params.sessionId) || null, turnId: codeAgentTurnLifecycleFields(traceId, params).turnId } + }); await settleAdmittedCodeAgentFailure({ base: initial ?? codeAgentAdmittedFailureBasePayload({ params, options, traceId }), params, @@ -821,7 +849,14 @@ async function submitCodeAgentChatTurn({ params, options, traceId }) { const run = async () => { try { + const billingStartedAt = Date.now(); const billingPreflight = await preflightCodeAgentBilling({ params, options, traceId, sendResponse: false }); + void emitCodeAgentOtelSpan("billing_preflight", traceId, options.env ?? process.env, { + startTimeMs: billingStartedAt, + attributes: { sessionId: safeSessionId(params.sessionId) || null, turnId: codeAgentTurnLifecycleFields(traceId, params).turnId, blocked: Boolean(billingPreflight?.blocked) }, + status: billingPreflight?.blocked ? "error" : "ok", + error: billingPreflight?.blocked ? billingPreflight?.payload?.error?.message ?? billingPreflight?.blocker?.summary ?? "billing preflight blocked" : null + }); if (billingPreflight?.blocked) { await settleAdmittedCodeAgentFailure({ base: codeAgentAdmittedFailureBasePayload({ params, options, traceId }), @@ -899,9 +934,13 @@ async function recordCodeAgentTurnAdmission({ payload = {}, params = {}, options message: "Code Agent turn was durably admitted before billing or dispatch preflight.", sessionId: safeSessionId(params.sessionId) || null, threadId: safeOpaqueId(params.threadId) || null, + ...codeAgentOtelTraceFields(traceId, options.env ?? process.env), waitingFor: "billing-or-dispatch", valuesPrinted: false }); + void emitCodeAgentOtelSpan("durable_admission", traceId, options.env ?? process.env, { + attributes: { sessionId: safeSessionId(params.sessionId) || null, threadId: safeOpaqueId(params.threadId) || null, turnId: codeAgentTurnLifecycleFields(traceId, params).turnId } + }); return ownerRecord; } @@ -926,6 +965,7 @@ function codeAgentAdmissionUnavailablePayload({ error, params = {}, traceId } = accepted: false, status: "failed", traceId: safeTraceId(traceId) || null, + ...codeAgentOtelTraceFields(traceId, options.env ?? process.env), ...codeAgentTurnLifecycleFields(traceId, params), conversationId: safeConversationId(params.conversationId) || null, sessionId: safeSessionId(params.sessionId) || null, @@ -1768,6 +1808,10 @@ export async function handleCodeAgentTurnHttp(request, response, url, options) { } const resolved = await measureCodeAgentHttpPhase(requestOptions, "turn_resolve_status", () => resolveCodeAgentTurnStatusSnapshot(traceId, requestOptions)); recordCodeAgentTurnReadMetric(options, url, resolved.statusCode, resolved.body, startedAt); + void emitCodeAgentOtelSpan("turn_status_read", traceId, requestOptions.env ?? process.env, { + startTimeMs: startedAt, + attributes: { "http.method": "GET", "http.route": "/v1/agent/turns/:traceId", "http.status_code": resolved.statusCode, status: resolved.body?.status ?? null, terminal: resolved.body?.terminal ?? null } + }); sendJson(response, resolved.statusCode, resolved.body); } @@ -3123,6 +3167,7 @@ function cancelBlockedPayload({ } export async function handleCodeAgentTraceHttp(request, response, url, options) { + const startedAt = nowMs(); const parts = url.pathname.split("/").filter(Boolean); const traceId = decodeURIComponent(parts[3] ?? ""); const streamSegment = parts[4]; @@ -3157,7 +3202,12 @@ export async function handleCodeAgentTraceHttp(request, response, url, options) const body = await measureCodeAgentHttpPhase(requestOptions, "trace_paginate", () => { return paginateTraceSnapshot(context.trace, pageOptions); }); - sendJson(response, context.found ? 200 : 404, codeAgentCompatProjectionPayload(body, context)); + const statusCode = context.found ? 200 : 404; + void emitCodeAgentOtelSpan("trace_events_read", traceId, requestOptions.env ?? process.env, { + startTimeMs: startedAt, + attributes: { "http.method": "GET", "http.route": "/v1/agent/traces/:traceId", "http.status_code": statusCode, returnedEvents: Array.isArray(body?.events) ? body.events.length : 0, found: Boolean(context.found) } + }); + sendJson(response, statusCode, codeAgentCompatProjectionPayload(body, context)); } function codeAgentTracePageOptions(url) {