mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add code agent otel spans

Browse Source
This commit is contained in:
lyon
2026-06-19 23:05:50 +08:00
parent bfceacaee1
commit 69b8e03661
4 changed files with 200 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/deploy.yaml
+2
View File
@@ -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
internal/cloud/code-agent-agentrun-adapter.ts
+14
View File
@@ -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,
internal/cloud/otel-trace.ts
+133
View File
@@ -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");
}
internal/cloud/server-code-agent-http.ts
+51 -1
View File
@@ -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) {