diff --git a/actions/experiment/README.md b/actions/experiment/README.md index 19edfe2d..57fee9be 100644 --- a/actions/experiment/README.md +++ b/actions/experiment/README.md @@ -70,17 +70,18 @@ changes of each release. | `command` | A shell command that runs the eval script, for example `uv run python evals/reply_eval.py` or `pnpm eval`. | | `dataset` | The dataset to run on. The action puts it in `TELEMETRY_DEV_EXPERIMENT_DATASET`. When the script also names a dataset, the two must be the same. | | `min-scores` | Pass marks, as `name=value` pairs that commas or new lines separate. A pass rate is 0 to 1, and a mean is on the scale of its score. The action puts them in `TELEMETRY_DEV_EXPERIMENT_MIN_SCORES`, and they win over the `minScores` of the script for the same name. | -| `api-key` | A telemetry.dev API key with the `datasets:read` and `experiments:write` scopes, for all environments or for `experiment`. Default: the `TELEMETRY_DEV_API_KEY` variable of the job. | +| `api-key` | A telemetry.dev API key with the `datasets:read` and `experiments:write` scopes (Run experiments) and, for the traces of the items, Send telemetry. Default: the `TELEMETRY_DEV_API_KEY` variable of the job. | | `base-url` | The telemetry.dev API. Default: `https://api.telemetry.dev`. | | `working-directory` | The directory that the script or the command runs in. Default: `.`. | | `comment` | `true` or `false`. Set to `false` to write the job summary only. Another value gets a warning and counts as `true`. Default: `true`. | | `github-token` | The token that writes the comment. Default: `github.token`, which needs the `pull-requests: write` permission. When it is empty on a pull request, the action writes no comment and logs a warning. | The action sets `TELEMETRY_DEV_ENVIRONMENT` to `experiment` when the job does not set it, so the -traces of the items stay out of your production pages. Thus the key must be for all environments -or for `experiment`. A key for another environment, for example your production key, cannot send -the traces of the items: the experiment then completes with its outputs, but with no traces and no -cost. `CI` is `true` in GitHub Actions, thus the SDK runs in CI mode: it records the repository, +traces of the items stay out of your production pages. A key with Run experiments sends the +`experiment` environment even when it is for another environment, but it needs Send telemetry for +the traces of the items. Without it, the experiment completes with its outputs, but with no traces +and no cost, and the job summary says that the runs have no traces, with the reason of the ingest. +`CI` is `true` in GitHub Actions, thus the SDK runs in CI mode: it records the repository, the commit, the run, and the pull request with the experiment, and it exits with an error when the check fails. diff --git a/actions/experiment/report.mjs b/actions/experiment/report.mjs index 972435cb..f6693cbc 100644 --- a/actions/experiment/report.mjs +++ b/actions/experiment/report.mjs @@ -1,7 +1,9 @@ // Writes the job summary of the experiment action, and the comment with the result on the pull // request. runExperiment and run_experiment add one line of JSON to the results file for each // experiment: the body of GET /v1/experiments/{id}, with `ci`, false when the CI mode was off, and -// `runsPerItem`. Lines from SDKs before those fields have neither. +// `runsPerItem`, and `traced`, false without a telemetry client or when the ingest refused its +// exports (401 or 403), with `tracesError` for a refusal. Lines from SDKs before those fields do +// not have them. import { createHash } from "node:crypto"; import { appendFileSync, readFileSync } from "node:fs"; @@ -180,6 +182,25 @@ function render(experiments, code) { lines.push("", "The CI mode of this experiment was off, thus it does not fail the job."); } + if (experiment.traced === false) { + const missing = + "The runs have no traces: the items have no trace to open, and Cost leaves out the model calls of your code."; + + if (typeof experiment.tracesError === "string" && experiment.tracesError !== "") { + lines.push( + "", + `${missing} The ingest refused them:`, + "", + `> ${text(experiment.tracesError)}`, + ); + } else { + lines.push( + "", + `${missing} No telemetry client sent them: before the experiment runs, call \`init({ environment: "experiment" })\` in TypeScript, or \`init(environment="experiment")\` in Python, with an API key.`, + ); + } + } + if (typeof experiment.error === "string" && experiment.error !== "") { lines.push("", `> ${text(experiment.error)}`); } diff --git a/docs/sdk-conformance.md b/docs/sdk-conformance.md index 65b6d052..5ad243ac 100644 --- a/docs/sdk-conformance.md +++ b/docs/sdk-conformance.md @@ -32,8 +32,8 @@ repository, not this SDK workspace. Paths below refer to that product repository | `finishReason` | `gen_ai.response.finish_reasons = [value]` | single-element array | | sampling params | `gen_ai.request.{temperature,top_p,top_k,max_tokens,stop_sequences,seed,frequency_penalty,presence_penalty}` | flat fields | | `timeToFirstChunkMs` / `time_to_first_chunk_ms` | `gen_ai.response.time_to_first_chunk` | **seconds** on the wire (key has no `ms` suffix; ingest multiplies by 1000) | -| `toolName`, `toolCallId`, `toolDescription` | `gen_ai.tool.{name,call.id,description}` | | -| `agentName`, `agentId` | `gen_ai.agent.{name,id}` | | +| `toolName`, `toolCallId`, `toolDescription` | `gen_ai.tool.{name,call.id,description}` | `gen_ai.tool.name` defaults to the span name of a `type: "tool"` span, unless it is `execute_tool` or starts with `execute_tool ` | +| `agentName`, `agentId` | `gen_ai.agent.{name,id}` | `gen_ai.agent.name` defaults to the span name of a `type: "agent"` span, unless it is `invoke_agent` or starts with `invoke_agent ` | | `metadata.` (per span) | `td.metadata.` | reserved keys `userId`/`sessionId`/`user_id`/`session_id` dropped with a debug warning; never passed through the mask | | `propagateAttributes userId` | `user.id` on EVERY span/log in scope | stamped by the span processor's onStart | | `propagateAttributes sessionId` | `gen_ai.conversation.id` on EVERY span/log in scope | never `session.id` | @@ -367,8 +367,10 @@ the `dataset` option has no value, and a different `dataset` option is a `TypeEr The value must match `-?\d+(\.\d+)?` (ASCII digits), else it is a `TypeError`. These marks win over the `minScores` option for the same name. When `TELEMETRY_DEV_EXPERIMENT_RESULTS` names a file, the SDK adds the JSON of the `GET /v1/experiments/{id}` body to it as one line, then `ci`, the CI mode -of the run (a boolean), and `runsPerItem`. The action fails the job for a failed experiment only -when its line has `ci` true. +of the run (a boolean), `runsPerItem`, and `traced`: false when there is no enabled client, or +when the ingest refused an export of the client with 401 or 403 since an export last reached it, +and then `tracesError`, the message of that export error. The action fails the job for a failed +experiment only when its line has `ci` true, and notes runs without traces in the job summary. The SDKs print one line when the experiment starts and the result when it ends. The result and the `check_failed` error show a rate as a whole percent and any other score, with its pass mark, to one @@ -420,7 +422,7 @@ eventName: "e2e.inside")` inside it `prompted-private` (the same with captureInput false) 2. plain span `background-job` started after the agent ended, with `parent = ` -3. agent span `follow-up-turn` (`agentName: "support"`) started after the agent ended with no +3. agent span `follow-up-turn` (no `agentName`) started after the agent ended with no active parent: a second root call of the same session 4. standalone `log("outside spans", level: "warn", eventName: "e2e.outside")` 5. `flush()` + `shutdown()` @@ -462,7 +464,7 @@ do not sanitize raw OpenTelemetry events or arbitrary attributes. | C17 | Exact span census (20). Without a key the SDK sends nothing and does not throw; the census cannot see that, so the TypeScript test counts captured requests and the Python scenario checks only that nothing raised | | C18 | Streaming generation: `timeToFirstChunkMs` stored as 250 ms; wire attribute `gen_ai.response.time_to_first_chunk` = 0.25 s | | C19 | OpenRouter generation: provider stored as `openrouter` | -| C20 | Session trace: `follow-up-turn` (a second root call of the same session) lands in the same trace, parented to the session parent, started at or after the agent ended | +| C20 | Session trace: `follow-up-turn` (a second root call of the same session) lands in the same trace, parented to the session parent, started at or after the agent ended, with the agent name `follow-up-turn` | | C21 | Process mode: two otherwise independent roots and a standalone log share one generated UUID session; a nested explicit-session span/log subtree consistently uses the explicit ID instead | | C22 | Streaming truncation: `telemetry.dev.capture.truncated = true` survives OTLP transport and ingest on the streamed generation | | C23 | Captured JSON matches an independently specified JSON.stringify byte sequence, including number spelling, key ordering, non-finite values and Unicode | diff --git a/packages/otel/src/index.ts b/packages/otel/src/index.ts index fa93edbe..8c87c910 100644 --- a/packages/otel/src/index.ts +++ b/packages/otel/src/index.ts @@ -76,6 +76,7 @@ export { createMetricExporter, createTraceExporter, type ExportFailures, + IngestError, maybeGzip, otlpHeaders, type OtlpTarget, diff --git a/packages/otel/src/transport.ts b/packages/otel/src/transport.ts index cac5f4a5..0e3723c7 100644 --- a/packages/otel/src/transport.ts +++ b/packages/otel/src/transport.ts @@ -34,6 +34,11 @@ export interface ExportFailures { report(cause: unknown): void; /** Records that an export reached the ingest, so that report() sends the next failure. */ clear(): void; + /** + * The last export that the ingest refused with 401 (the API key is not valid) or 403 (the key + * cannot send telemetry) since an export reached the ingest. + */ + readonly rejected?: IngestError; } /** @@ -42,15 +47,24 @@ export interface ExportFailures { */ export function createExportFailures(onError?: (error: Error) => void): ExportFailures { let reported = false; + let rejected: IngestError | undefined; return { report(cause) { + if (cause instanceof IngestError && (cause.status === 401 || cause.status === 403)) { + rejected = cause; + } + if (reported) return; reported = true; reportError(onError, cause); }, clear() { reported = false; + rejected = undefined; + }, + get rejected() { + return rejected; }, }; } @@ -210,19 +224,25 @@ const cancelBody = async (res: Response) => { } catch {} }; -export const postOtlp = async ({ - fetchImpl, - url, - headers, - body, - signal, -}: { +type OtlpPost = { fetchImpl: typeof fetch; url: string; headers: Record; body: Uint8Array; signal?: AbortSignal; -}) => { +}; + +/** Posts with retries, and returns the last response with its body cancelled. */ +export const postOtlp = async (post: OtlpPost) => { + const res = await postWithRetries(post); + + await cancelBody(res); + + return res; +}; + +/** Posts with retries. The body of the last response is left to the caller. */ +const postWithRetries = async ({ fetchImpl, url, headers, body, signal }: OtlpPost) => { for (let attempt = 0; ; attempt += 1) { let retryAfter: number | undefined; @@ -242,8 +262,6 @@ export const postOtlp = async ({ attempt === RETRY_DELAYS_MS.length || (retryAfter !== undefined && retryAfter > MAX_RETRY_AFTER_MS) ) { - await cancelBody(res); - return res; } @@ -292,7 +310,7 @@ async function postSerialized( ? { ...target.headers, "content-encoding": contentEncoding } : target.headers; - const res = await postOtlp({ + const res = await postWithRetries({ fetchImpl: transport.fetchImpl, url: target.url, headers, @@ -309,8 +327,127 @@ async function postSerialized( ? `; retry-after ${retryAfterHeader} exceeds ${MAX_RETRY_AFTER_MS / 1000}s retry cap` : ""; - throw new Error(`telemetry.dev ${label} ingest failed: ${res.status}${retryDetail}`); + const reason = await statusMessage(res); + + throw new IngestError( + `telemetry.dev ${label} ingest failed: ${res.status}${reason ? ` ${reason}` : ""}${retryDetail}`, + res.status, + ); + } else { + await cancelBody(res); + } +} + +/** A failed export, with the HTTP status that the ingest answered. */ +export class IngestError extends Error { + constructor( + message: string, + readonly status: number, + ) { + super(message); + this.name = "IngestError"; + } +} + +const MAX_ERROR_BODY_BYTES = 4096; + +/** + * The message of an OTLP error response: the `message` of its JSON or protobuf `Status`, as the + * telemetry.dev ingest sends it, such as "This API key cannot send telemetry. Create a key with + * Send telemetry access." Undefined for another content type, or a body that cannot be read. + */ +async function statusMessage(res: Response): Promise { + try { + const type = res.headers.get("content-type") ?? ""; + const parse = type.includes("json") + ? jsonStatusMessage + : type.includes("protobuf") + ? protobufStatusMessage + : undefined; + + if (!parse) { + await cancelBody(res); + + return undefined; + } + + const message = parse(await readPrefix(res, MAX_ERROR_BODY_BYTES)); + + return message?.trim() || undefined; + } catch { + await cancelBody(res); + + return undefined; + } +} + +/** The first `max` bytes of the body. The rest is not downloaded: the body is cancelled. */ +async function readPrefix(res: Response, max: number): Promise { + const reader = res.body?.getReader(); + const prefix = new Uint8Array(max); + let size = 0; + + if (!reader) return prefix.subarray(0, 0); + + try { + while (size < max) { + const { done, value } = await reader.read(); + + if (done) break; + const take = value.subarray(0, max - size); + prefix.set(take, size); + size += take.length; + } + } finally { + await reader.cancel().catch(() => {}); + } + + return prefix.subarray(0, size); +} + +function jsonStatusMessage(bytes: Uint8Array): string | undefined { + const status: unknown = JSON.parse(new TextDecoder().decode(bytes)); + const message = (status as { message?: unknown } | null)?.message; + + return typeof message === "string" ? message : undefined; +} + +/** Field 2 of google.rpc.Status, read with the protobuf wire format. */ +function protobufStatusMessage(bytes: Uint8Array): string | undefined { + let at = 0; + + const varint = () => { + let value = 0; + + for (let shift = 0; shift < 35; shift += 7) { + const byte = bytes[at++]; + + if (byte === undefined) throw new Error("truncated"); + value += (byte & 0x7f) * 2 ** shift; + + if (byte < 0x80) return value; + } + + throw new Error("varint too long"); + }; + + while (at < bytes.length) { + const tag = varint(); + const wire = tag & 7; + + if (wire === 0) varint(); + else if (wire === 2) { + const length = varint(); + const end = at + length; + + if (end > bytes.length) return undefined; + + if (tag >>> 3 === 2) return new TextDecoder().decode(bytes.subarray(at, end)); + at = end; + } else return undefined; } + + return undefined; } const createExportLifecycle = (transport: Transport) => { diff --git a/packages/otel/tests/transport.test.ts b/packages/otel/tests/transport.test.ts index ed4e5926..9c2118a7 100644 --- a/packages/otel/tests/transport.test.ts +++ b/packages/otel/tests/transport.test.ts @@ -24,6 +24,7 @@ import { import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { + createExportFailures, createMetricExporter, createLogExporter, createTraceExporter, @@ -186,6 +187,96 @@ test("reports retry-after values beyond the local cap", async () => { expect(String(errors[0])).toContain("retry-after 60 exceeds 2s retry cap"); }); +/** A google.rpc.Status with a code and a message, as the ingest encodes it for protobuf. */ +function protobufStatus(code: number, message: string): Uint8Array { + const text = new TextEncoder().encode(message); + const varint = (value: number) => { + const bytes = []; + + for (; value >= 0x80; value >>>= 7) bytes.push((value & 0x7f) | 0x80); + bytes.push(value); + + return bytes; + }; + + return new Uint8Array([0x08, ...varint(code), 0x12, ...varint(text.length), ...text]); +} + +const PROTOBUF = { "content-type": "application/x-protobuf" }; +const NO_TELEMETRY = "This API key cannot send telemetry. Create a key with Send telemetry access."; +// Over 127 bytes, so that its length takes two bytes. +const LONG_MESSAGE = `API key was not accepted. ${"x".repeat(120)}`; + +test.each([ + { + name: "a 403 with a protobuf Status", + response: () => + new Response(protobufStatus(7, NO_TELEMETRY), { status: 403, headers: PROTOBUF }), + message: `telemetry.dev trace ingest failed: 403 ${NO_TELEMETRY}`, + rejected: 403, + }, + { + name: "a 401 with a protobuf Status of a long message", + response: () => + new Response(protobufStatus(16, LONG_MESSAGE), { status: 401, headers: PROTOBUF }), + message: `telemetry.dev trace ingest failed: 401 ${LONG_MESSAGE}`, + rejected: 401, + }, + { + name: "a 403 with a Status in a body that does not end", + // Only the first 4096 bytes are read; the rest of the body is not waited for. + response: () => + new Response( + new ReadableStream({ + start(controller) { + const status = protobufStatus(7, NO_TELEMETRY); + const padded = new Uint8Array(5000); + padded.set(status); + // Zero padding after the Status; the message is read before it. + controller.enqueue(padded); + }, + }), + { status: 403, headers: PROTOBUF }, + ), + message: `telemetry.dev trace ingest failed: 403 ${NO_TELEMETRY}`, + rejected: 403, + }, + { + name: "a 401 with a JSON Status", + response: () => + Response.json({ code: 16, message: "API key was not accepted" }, { status: 401 }), + message: "telemetry.dev trace ingest failed: 401 API key was not accepted", + rejected: 401, + }, + { + name: "a 400 with a body of another type", + // Bytes that read as a protobuf Status, from a proxy that answers with text. + response: () => + new Response("\x12\x04oops", { status: 400, headers: { "content-type": "text/plain" } }), + message: "telemetry.dev trace ingest failed: 400", + rejected: undefined, + }, +])("$name reports the message of the ingest", async ({ response, message, rejected }) => { + const errors: Error[] = []; + const failures = createExportFailures((error) => errors.push(error)); + let reached = false; + const fetchImpl: typeof fetch = () => + Promise.resolve(reached ? new Response(null, { status: 200 }) : response()); + const exporter = createTraceExporter( + { url: "https://ingest.example/v1/traces", headers: otlpHeaders("td_live_x") }, + { fetchImpl, failures }, + ); + + expect(await exportSpans(exporter, makeSpans(1))).toBe(ExportResultCode.FAILED); + expect(errors.map((error) => error.message)).toEqual([message]); + expect(failures.rejected?.status).toBe(rejected); + + // An export that reaches the ingest ends the rejection. + reached = true; + expect(await exportSpans(exporter, makeSpans(1))).toBe(ExportResultCode.SUCCESS); + expect(failures.rejected).toBeUndefined(); +}); + describe("retry-after", () => { beforeEach(() => { vi.useFakeTimers(); diff --git a/packages/sdk/README.md b/packages/sdk/README.md index 6edafae9..5f06caa8 100644 --- a/packages/sdk/README.md +++ b/packages/sdk/README.md @@ -131,6 +131,11 @@ Explicit and propagated session IDs take precedence, and valid parent trace cont The generated session ends when the client is shut down or replaced; a later `init()` generates a new ID. It is not persisted across processes. +The root spans of a session join one trace for each API key, and the trace keeps the environment +of its first turn. When one key sends turns of the same session id from two environments, the +later turns show under the first environment, or get a 403 when the key may not send that +environment. Use a key for each environment, or give each environment its own session ids. + `sampler` accepts an OpenTelemetry `Sampler` and overrides `OTEL_TRACES_SAMPLER` / `OTEL_TRACES_SAMPLER_ARG`. The default is parent-based always-on. Session roots keep the configured root policy. Nested spans and explicitly parented spans keep the real parent's policy. @@ -307,9 +312,16 @@ const result = await runExperiment({ - The key needs Run experiments in its Access on the API keys page (the `datasets:read` and `experiments:write` scopes). runExperiment uses the same API key, base URL, and `fetch` as getPrompt, or its own `apiKey`, `baseUrl`, and `fetch` - options. With `environment: "experiment"`, the key of the telemetry client must be for all - environments or for `experiment`: a key for another environment cannot send the traces of the - items, and the items then have no trace and no cost. + options. +- The items send their traces with the key of the telemetry client, which needs Send telemetry. A + key for one environment also sends the `experiment` environment when it has Run experiments. + Without a client, without Send telemetry, with a key for another environment and no Run + experiments, or when Accept telemetry is off in the project settings, the items have no trace + and no cost. The export error says why, and the job summary of the experiment action says that + the runs have no traces, with the reason of the ingest. +- When the ingest refuses a trace export because the key is not valid (401), for example after + the key was revoked or its project was deleted, and the client sends with the key and the base + URL of the run, runExperiment starts no more items and rejects with `invalid_api_key`. The outputs, the errors of failed tasks, the scorer reasons, and `metadata` go to telemetry.dev and stay with the experiment, because they are its result. `captureOutput`, `mask`, and the @@ -358,6 +370,8 @@ await score({ ``` - Give `traceId`, `sessionId`, or the two, and a `spanId` of the trace for a score on one span. + With a `traceId`, the score belongs to that trace and shows in the session of the trace, even + when the `sessionId` names another session. Give `value`, `label`, or the two. `user_feedback` takes the value 1 for up or 0 for down, and no label. A pass/fail score has the label `pass` with the value 1, or `fail` with 0. - Like the telemetry functions, score never rejects. It resolves with the id of the score, or with diff --git a/packages/sdk/src/client.ts b/packages/sdk/src/client.ts index d0b6cabc..66590c26 100644 --- a/packages/sdk/src/client.ts +++ b/packages/sdk/src/client.ts @@ -28,6 +28,7 @@ import { createTraceExporter, diag, DORMANT_INTERVAL_MS, + type ExportFailures, ExportQueueBudget, otlpHeaders, reportError, @@ -61,6 +62,7 @@ export interface ClientCore { config: ResolvedConfig; tracer: Tracer; ensureLogger(): Logger | undefined; + failures: ExportFailures; } export interface ClientHandle extends TelemetryClient { @@ -349,7 +351,7 @@ function initInner(options?: TelemetryOptions, overrides?: ClientOverrides): Tel }, }; - handle.core = { config, tracer, ensureLogger }; + handle.core = { config, tracer, ensureLogger, failures }; activeClient = handle; return handle; diff --git a/packages/sdk/src/experiments.ts b/packages/sdk/src/experiments.ts index 4748dd85..dc6e9c06 100644 --- a/packages/sdk/src/experiments.ts +++ b/packages/sdk/src/experiments.ts @@ -2,6 +2,7 @@ import { type Context, trace } from "@opentelemetry/api"; import { activeContext, diag, + type IngestError, jsonAttr, PROPAGATED_KEY, propagatedFromContext, @@ -296,7 +297,12 @@ export async function runExperiment(options: RunExperimentOptions): Promise), ci: run.inCi, runsPerItem: run.runsPerItem }, + { + ...(body as Record), + ci: run.inCi, + runsPerItem: run.runsPerItem, + ...traces(), + }, api.onError, ); console.log(describe(result, run.runsPerItem)); @@ -535,6 +541,21 @@ async function runItems( const worker = async () => { while (failure === undefined) { + const rejected = keyRejected(api); + + if (rejected) { + failure = { + error: new ExperimentError( + "invalid_api_key", + `The API key was not accepted, so no more items run: ${rejected.message}`, + rejected.status, + { cause: rejected }, + ), + }; + + return; + } + const next = await runs.next().catch((error: unknown) => { failure ??= { error }; @@ -566,6 +587,39 @@ async function runItems( } } +/** + * The refusal of an export of the client with 401 since an export last reached the ingest, when + * the client sends with the key of the run to the same server. The key then stopped working, for + * example because it was revoked or its project was deleted, and the API will refuse the results + * of the items too. + */ +function keyRejected(api: Authed): IngestError | undefined { + const core = currentClient().core; + const rejected = core?.failures.rejected; + + return rejected?.status === 401 && + core?.config.apiKey === api.apiKey && + core.config.baseUrl === api.baseUrl + ? rejected + : undefined; +} + +/** + * For the results line of the experiment action: `traced` is false without an enabled client, or + * when the ingest refused an export of the client with 401 or 403 since an export last reached + * it, and `tracesError` then gives its reason. Exports that failed for another reason, such as an + * outage, are not counted. + */ +function traces(): { traced: boolean; tracesError?: string } { + const core = currentClient().core; + + if (core === undefined) return { traced: false }; + + const rejected = core.failures.rejected; + + return rejected ? { traced: false, tracesError: rejected.message } : { traced: true }; +} + /** Collects item results and posts them in order, one post at a time. */ function itemBatch(post: (results: ItemResult[]) => Promise) { let pending: ItemResult[] = []; diff --git a/packages/sdk/src/span.ts b/packages/sdk/src/span.ts index 3ed9cd94..877f74c0 100644 --- a/packages/sdk/src/span.ts +++ b/packages/sdk/src/span.ts @@ -250,6 +250,14 @@ export function createSpanHandle( if (meta.scopedPrompt) dropPromptAttributes(attrs); + // An agent or a tool is named after its span unless a field names it, as in the Python SDK. A + // span named after its operation, such as "invoke_agent" or "invoke_agent AGENT123" from an + // integration that names it by id, names no agent or tool. + const spanName = options.name || name; + const typeOperation = SPAN_TYPE_OPERATIONS[meta.type]; + const defaultName = + spanName === typeOperation || spanName.startsWith(`${typeOperation} `) ? undefined : spanName; + const span = core.tracer.startSpan( name, { @@ -258,6 +266,8 @@ export function createSpanHandle( links: options.links, attributes: { "gen_ai.operation.name": SPAN_TYPE_OPERATIONS[meta.type], + ...(meta.type === "agent" && defaultName ? { "gen_ai.agent.name": defaultName } : {}), + ...(meta.type === "tool" && defaultName ? { "gen_ai.tool.name": defaultName } : {}), ...propagatedFromContext(parentCtx), ...prompt, ...attrs, diff --git a/packages/sdk/tests/experiment-action.test.ts b/packages/sdk/tests/experiment-action.test.ts index 3003ee54..51df4aca 100644 --- a/packages/sdk/tests/experiment-action.test.ts +++ b/packages/sdk/tests/experiment-action.test.ts @@ -396,12 +396,20 @@ test("report.mjs writes the summary, posts it on the pull request, and sets the test("report.mjs fails the job for a failed experiment before the last, when the script exits with 0", async () => { // The lines of an SDK that writes the CI mode. - const failed = { ...result, ci: true, runsPerItem: 1 }; + const failed = { ...result, ci: true, runsPerItem: 1, traced: true }; const passed = { ...failed, name: "reply-short", checkStatus: "passed", itemsFailed: 0 }; const stopped = { ...passed, name: "reply-stopped", status: "stopped", checkStatus: null }; - // A failed check and a stopped experiment of a script that runs with the CI mode off. - const notInCi = { ...result, name: "reply-local", ci: false, runsPerItem: 3 }; - const stoppedNotInCi = { ...stopped, name: "reply-local-stopped", ci: false }; + // A failed check of a script that runs with the CI mode off, and without a telemetry client. + const notInCi = { ...result, name: "reply-local", ci: false, runsPerItem: 3, traced: false }; + // A stopped experiment whose traces the ingest refused. + const stoppedNotInCi = { + ...stopped, + name: "reply-local-stopped", + ci: false, + traced: false, + tracesError: + "telemetry.dev trace ingest failed: 403 This project does not accept telemetry. Turn on Accept telemetry in its settings.", + }; const failedFirst = await report({ EXPERIMENT_EXIT_CODE: "0", INPUT_COMMENT: "false" }, [ failed, @@ -427,17 +435,33 @@ test("report.mjs fails the job for a failed experiment before the last, when the expect(failedFirst.outputs["check-status"]).toBe("passed"); expect(failedFirst.summary).not.toContain("runs each"); expect(failedFirst.summary).not.toContain("does not fail the job"); + expect(failedFirst.summary).not.toContain("no traces"); expect(olderSdk.code).toBe(0); expect(olderSdk.stdout).not.toContain("::error"); expect(olderSdk.summary).not.toContain("does not fail the job"); + expect(olderSdk.summary).not.toContain("no traces"); expect(allPassed.code).toBe(0); expect(allPassed.outputs["check-status"]).toBe("passed"); expect(offCi.code).toBe(0); expect(offCi.stdout).not.toContain("::error"); expect(offCi.summary).toContain( "Check failed. 40 items, 3 runs each: 118 of 120 runs completed, 2 failed.\n\n" + - "The CI mode of this experiment was off, thus it does not fail the job.", + "The CI mode of this experiment was off, thus it does not fail the job.\n\n" + + "The runs have no traces: the items have no trace to open, and Cost leaves out the model " + + "calls of your code. No telemetry client sent them: before the experiment runs, call " + + '`init({ environment: "experiment" })` in TypeScript, or `init(environment="experiment")` ' + + "in Python, with an API key.", ); + // A refusal names the reason of the ingest in place of the init() advice. + expect(offCi.summary).toContain( + "Cost leaves out the model calls of your code. The ingest refused them:\n\n" + + // Escaped, as the action escapes all text from the API. + "> telemetry\\.dev trace ingest failed\\: 403 This project does not accept telemetry\\. Turn " + + "on Accept telemetry in its settings\\.", + ); + // Only the note without a reason gives the init() advice, in TypeScript and in Python. + expect(offCi.summary.match(/init\(/g)).toHaveLength(2); + expect(offCi.summary.match(/no traces/g)).toHaveLength(2); }); /** The body of the comment that the action posts for the result. */ diff --git a/packages/sdk/tests/experiments.test.ts b/packages/sdk/tests/experiments.test.ts index 93bacc8c..63e6491b 100644 --- a/packages/sdk/tests/experiments.test.ts +++ b/packages/sdk/tests/experiments.test.ts @@ -7,6 +7,8 @@ import { afterEach, beforeEach, expect, test, vi } from "vitest"; import { type ExperimentItem, ExperimentError, + flush, + init, log, propagateAttributes, runExperiment, @@ -513,6 +515,84 @@ test("a failed post of items stops new items, completes the experiment, and reje ]); }); +test.each<{ name: string; status: number; key: string; started: number; clientUrl?: string }>([ + { + name: "a 401 for the key of the run stops new items", + status: 401, + key: "td_live_test", + started: 2, + }, + { + name: "a 401 for another key does not stop the run", + status: 401, + key: "td_live_other", + started: 5, + }, + { name: "a 403 does not stop the run", status: 403, key: "td_live_test", started: 5 }, + { + name: "a 401 from another server does not stop the run", + status: 401, + key: "td_live_test", + started: 5, + clientUrl: "https://staging.ingest.test", + }, +])( + "a trace export refused with $status: $name", + async ({ status, key, started: expected, clientUrl }) => { + const results = join(mkdtempSync(join(tmpdir(), "experiment-")), "results.jsonl"); + writeFileSync(results, ""); + vi.stubEnv("TELEMETRY_DEV_EXPERIMENT_RESULTS", results); + const api = fakeApi({ + items: Array.from({ length: 5 }, (_, index) => datasetItem(index + 1)), + override: (request) => + request.url.pathname.startsWith("/v1/traces") + ? Response.json({ message: `refused with ${status}` }, { status }) + : undefined, + }); + init({ + apiKey: key, + baseUrl: clientUrl ?? "https://ingest.test", + environment: "experiment", + exportMode: "immediate", + logLevel: "silent", + fetch: api.fetch, + }); + let started = 0; + + const outcome = await experiment(api, { + concurrency: 1, + task: async () => { + started += 1; + // Waits for the export of the trace of the item before, so that its refusal has arrived. + await flush(); + + return "ok"; + }, + }).catch((caught: unknown) => caught); + + // The item that saw the refusal still ends; no item starts after it. + expect(started).toBe(expected); + + if (expected === 2) { + expect(outcome).toBeInstanceOf(ExperimentError); + expect(outcome).toMatchObject({ + code: "invalid_api_key", + status: 401, + message: + "The API key was not accepted, so no more items run: telemetry.dev trace ingest failed: 401 refused with 401", + }); + expect(routes(api.requests).at(-1)).toBe(`POST /v1/experiments/${EXPERIMENT_ID}/complete`); + } else { + expect(outcome).not.toBeInstanceOf(Error); + // The action notes that the runs have no traces, with the reason of the ingest. + expect(JSON.parse(readFileSync(results, "utf8"))).toMatchObject({ + traced: false, + tracesError: `telemetry.dev trace ingest failed: ${status} refused with ${status}`, + }); + } + }, +); + test("a complete that fails after a failed run goes to onError, and the run rejects with the first error", async () => { const errors: Error[] = []; setup({ environment: "experiment", onError: (error) => errors.push(error) }); @@ -976,7 +1056,10 @@ test("in GitHub Actions the experiment records the pull request, its head commit vi.stubEnv("TELEMETRY_DEV_EXPERIMENT_RESULTS", results); const api = fakeApi(); + // With a client the runs send their traces; the second experiment runs without one. + setup({ environment: "experiment" }); await experiment(api, { dataset: undefined, minScores: { correctness: 0.5, tone: 2 } }); + await shutdown(); expect(api.requests[1]!.body).toMatchObject({ dataset: "support-golden", @@ -997,8 +1080,8 @@ test("in GitHub Actions the experiment records the pull request, its head commit await experiment(fakeApi(), { ci: false, runsPerItem: 2 }); expect(readFileSync(results, "utf8")).toBe( - `${JSON.stringify({ ...result(), ci: true, runsPerItem: 1 })}\n` + - `${JSON.stringify({ ...result(), ci: false, runsPerItem: 2 })}\n`, + `${JSON.stringify({ ...result(), ci: true, runsPerItem: 1, traced: true })}\n` + + `${JSON.stringify({ ...result(), ci: false, runsPerItem: 2, traced: false })}\n`, ); }); diff --git a/packages/sdk/tests/span.test.ts b/packages/sdk/tests/span.test.ts index b569db57..d748d891 100644 --- a/packages/sdk/tests/span.test.ts +++ b/packages/sdk/tests/span.test.ts @@ -190,6 +190,25 @@ test("tool spans capture input/output under tool call keys", async () => { expect(span.attributes["gen_ai.input.messages"]).toBeUndefined(); }); +test("an agent or a tool span without a name field takes the name of the span, as in Python", async () => { + const { spans } = setup(); + startSpan("turn", { type: "agent" }).end(); + startSpan("lookup", { type: "tool" }).end(); + startSpan("step").end(); + // A span named after its operation names no agent or tool. + startSpan("invoke_agent", { type: "agent" }).end(); + startSpan("execute_tool lookup", { type: "tool" }).end(); + await flush(); + const [agent, tool, step, unnamedAgent, unnamedTool] = spans.getFinishedSpans(); + expect(agent!.attributes["gen_ai.agent.name"]).toBe("turn"); + expect(tool!.attributes["gen_ai.tool.name"]).toBe("lookup"); + expect(tool!.attributes["gen_ai.agent.name"]).toBeUndefined(); + expect(step!.attributes["gen_ai.agent.name"]).toBeUndefined(); + expect(step!.attributes["gen_ai.tool.name"]).toBeUndefined(); + expect(unnamedAgent!.attributes["gen_ai.agent.name"]).toBeUndefined(); + expect(unnamedTool!.attributes["gen_ai.tool.name"]).toBeUndefined(); +}); + test("agent fields map and update() applies later fields", async () => { const { spans } = setup(); const handle = startSpan("agent-run", { type: "agent", agentName: "support" }); diff --git a/sdks/python-anthropic/uv.lock b/sdks/python-anthropic/uv.lock index 4956ec6d..3643ddc6 100644 --- a/sdks/python-anthropic/uv.lock +++ b/sdks/python-anthropic/uv.lock @@ -748,17 +748,21 @@ name = "telemetry-dev" version = "0.3.2" source = { editable = "../python" } dependencies = [ + { name = "googleapis-common-protos" }, { name = "opentelemetry-api" }, { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-sdk" }, + { name = "requests" }, { name = "rfc8785" }, ] [package.metadata] requires-dist = [ + { name = "googleapis-common-protos", specifier = ">=1.63.1,<2" }, { name = "opentelemetry-api", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-sdk", specifier = ">=1.39.0,<2" }, + { name = "requests", specifier = ">=2.28,<3" }, { name = "rfc8785", specifier = ">=0.1.4" }, ] diff --git a/sdks/python-bedrock/uv.lock b/sdks/python-bedrock/uv.lock index 92d00ce2..71ad1c37 100644 --- a/sdks/python-bedrock/uv.lock +++ b/sdks/python-bedrock/uv.lock @@ -454,17 +454,21 @@ name = "telemetry-dev" version = "0.3.2" source = { editable = "../python" } dependencies = [ + { name = "googleapis-common-protos" }, { name = "opentelemetry-api" }, { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-sdk" }, + { name = "requests" }, { name = "rfc8785" }, ] [package.metadata] requires-dist = [ + { name = "googleapis-common-protos", specifier = ">=1.63.1,<2" }, { name = "opentelemetry-api", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-sdk", specifier = ">=1.39.0,<2" }, + { name = "requests", specifier = ">=2.28,<3" }, { name = "rfc8785", specifier = ">=0.1.4" }, ] diff --git a/sdks/python-google-genai/uv.lock b/sdks/python-google-genai/uv.lock index f6f8594c..cdeea0c8 100644 --- a/sdks/python-google-genai/uv.lock +++ b/sdks/python-google-genai/uv.lock @@ -824,17 +824,21 @@ name = "telemetry-dev" version = "0.3.2" source = { editable = "../python" } dependencies = [ + { name = "googleapis-common-protos" }, { name = "opentelemetry-api" }, { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-sdk" }, + { name = "requests" }, { name = "rfc8785" }, ] [package.metadata] requires-dist = [ + { name = "googleapis-common-protos", specifier = ">=1.63.1,<2" }, { name = "opentelemetry-api", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-sdk", specifier = ">=1.39.0,<2" }, + { name = "requests", specifier = ">=2.28,<3" }, { name = "rfc8785", specifier = ">=0.1.4" }, ] diff --git a/sdks/python-litellm/uv.lock b/sdks/python-litellm/uv.lock index acdff9d4..03ccabcb 100644 --- a/sdks/python-litellm/uv.lock +++ b/sdks/python-litellm/uv.lock @@ -2892,20 +2892,24 @@ name = "telemetry-dev" version = "0.3.2" source = { editable = "../python" } dependencies = [ + { name = "googleapis-common-protos" }, { name = "opentelemetry-api", version = "1.43.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, { name = "opentelemetry-api", version = "1.45.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, { name = "opentelemetry-exporter-otlp-proto-http", version = "1.43.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, { name = "opentelemetry-exporter-otlp-proto-http", version = "1.45.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, { name = "opentelemetry-sdk", version = "1.43.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, { name = "opentelemetry-sdk", version = "1.45.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "requests" }, { name = "rfc8785" }, ] [package.metadata] requires-dist = [ + { name = "googleapis-common-protos", specifier = ">=1.63.1,<2" }, { name = "opentelemetry-api", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-sdk", specifier = ">=1.39.0,<2" }, + { name = "requests", specifier = ">=2.28,<3" }, { name = "rfc8785", specifier = ">=0.1.4" }, ] diff --git a/sdks/python-openai/uv.lock b/sdks/python-openai/uv.lock index ffd4d32f..032089e3 100644 --- a/sdks/python-openai/uv.lock +++ b/sdks/python-openai/uv.lock @@ -738,17 +738,21 @@ name = "telemetry-dev" version = "0.3.2" source = { editable = "../python" } dependencies = [ + { name = "googleapis-common-protos" }, { name = "opentelemetry-api" }, { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-sdk" }, + { name = "requests" }, { name = "rfc8785" }, ] [package.metadata] requires-dist = [ + { name = "googleapis-common-protos", specifier = ">=1.63.1,<2" }, { name = "opentelemetry-api", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-sdk", specifier = ">=1.39.0,<2" }, + { name = "requests", specifier = ">=2.28,<3" }, { name = "rfc8785", specifier = ">=0.1.4" }, ] diff --git a/sdks/python-openrouter/uv.lock b/sdks/python-openrouter/uv.lock index 05f8f675..69622d39 100644 --- a/sdks/python-openrouter/uv.lock +++ b/sdks/python-openrouter/uv.lock @@ -606,17 +606,21 @@ name = "telemetry-dev" version = "0.3.2" source = { editable = "../python" } dependencies = [ + { name = "googleapis-common-protos" }, { name = "opentelemetry-api" }, { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-sdk" }, + { name = "requests" }, { name = "rfc8785" }, ] [package.metadata] requires-dist = [ + { name = "googleapis-common-protos", specifier = ">=1.63.1,<2" }, { name = "opentelemetry-api", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-sdk", specifier = ">=1.39.0,<2" }, + { name = "requests", specifier = ">=2.28,<3" }, { name = "rfc8785", specifier = ">=0.1.4" }, ] diff --git a/sdks/python/README.md b/sdks/python/README.md index 2c80d5b8..a0cf725c 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -74,6 +74,11 @@ initialization. Explicit and propagated session IDs take precedence, and valid p context is preserved. The generated session ends when the client is shut down or replaced; a later `init()` generates a new ID. It is not persisted across processes. +The root spans of a session join one trace for each API key, and the trace keeps the environment +of its first turn. When one key sends turns of the same session id from two environments, the +later turns show under the first environment, or get a 403 when the key may not send that +environment. Use a key for each environment, or give each environment its own session ids. + ## API reference | Name | Description | @@ -388,10 +393,16 @@ result = telemetry_dev.run_experiment( it when it got none. - The key needs Run experiments in its Access on the API keys page (the `datasets:read` and `experiments:write` scopes). `run_experiment` uses the same - API key and base URL as `get_prompt`, or its own `api_key` and `base_url`. With - `environment="experiment"`, the key of the telemetry client must be for all environments or for - `experiment`: a key for another environment cannot send the traces of the items, and the items - then have no trace and no cost. + API key and base URL as `get_prompt`, or its own `api_key` and `base_url`. +- The items send their traces with the key of the telemetry client, which needs Send telemetry. A + key for one environment also sends the `experiment` environment when it has Run experiments. + Without a client, without Send telemetry, with a key for another environment and no Run + experiments, or when Accept telemetry is off in the project settings, the items have no trace + and no cost. The export error says why, and the job summary of the experiment action says that + the runs have no traces, with the reason of the ingest. +- When the ingest refuses a trace export because the key is not valid (401), for example after + the key was revoked or its project was deleted, and the client sends with the key and the base + URL of the run, `run_experiment` starts no more items and raises `ExperimentError` with `invalid_api_key`. The outputs, the exceptions of failed tasks, the scorer reasons, and `metadata` go to telemetry.dev and stay with the experiment, because they are its result. `capture_output`, `mask`, and the @@ -435,6 +446,8 @@ telemetry_dev.score( ``` - Give `trace_id`, `session_id`, or the two, and a `span_id` of the trace for a score on one span. + With a `trace_id`, the score belongs to that trace and shows in the session of the trace, even + when the `session_id` names another session. Give `value`, `label`, or the two. `user_feedback` takes the value 1 for up or 0 for down, and no label. A pass/fail score has the label `pass` with the value 1, or `fail` with 0. - Like the telemetry functions, `score` never raises. It returns the id of the score, or `None` diff --git a/sdks/python/pyproject.toml b/sdks/python/pyproject.toml index 1cffcd58..063c0e66 100644 --- a/sdks/python/pyproject.toml +++ b/sdks/python/pyproject.toml @@ -22,6 +22,12 @@ dependencies = [ "opentelemetry-api>=1.39.0,<2", "opentelemetry-sdk>=1.39.0,<2", "opentelemetry-exporter-otlp-proto-http>=1.39.0,<2", + # The OTLP exporter sends with requests, and the ingest answers a refused export with a + # google.rpc.Status. The exporter requires both; the SDK reads the Status of the response. + # 1.63.1 is the first release that imports with protobuf 5, which opentelemetry-proto needs. + # The requests floor of the exporter, 2.7, does not import on Python 3.10; 2.28 does. + "googleapis-common-protos>=1.63.1,<2", + "requests>=2.28,<3", "rfc8785>=0.1.4", ] diff --git a/sdks/python/src/telemetry_dev/_client.py b/sdks/python/src/telemetry_dev/_client.py index 4eb54dcc..969e98f6 100644 --- a/sdks/python/src/telemetry_dev/_client.py +++ b/sdks/python/src/telemetry_dev/_client.py @@ -50,6 +50,7 @@ ReportingLogExporter, ReportingSpanExporter, after_export, + otlp_session, resolve_byte_bound, ) from ._metrics import ( @@ -156,6 +157,7 @@ def __init__( self._logger_provider: LoggerProvider | None = None self._meter_provider: MeterProvider | None = None self._output_chunks: OutputChunkAggregation | None = None + self.export_failures: ExportFailures = ExportFailures(on_error) if not enabled: return @@ -185,7 +187,7 @@ def __init__( ), ) self._tracer_provider.sampler = SessionSampler(self._tracer_provider.sampler) - failures = ExportFailures(on_error) + failures = self.export_failures # Without an api key (enabled via a test seam), never construct real network # exporters — they would POST to the ingest with a bogus Authorization header. @@ -218,6 +220,7 @@ def __init__( headers=headers, compression=Compression.Gzip, timeout=timeout, + session=otlp_session(), ), failures, on_error, @@ -242,6 +245,7 @@ def __init__( headers=headers, compression=Compression.Gzip, timeout=timeout, + session=otlp_session(), ), failures, on_error, diff --git a/sdks/python/src/telemetry_dev/_experiments.py b/sdks/python/src/telemetry_dev/_experiments.py index 1c364e78..f3b618be 100644 --- a/sdks/python/src/telemetry_dev/_experiments.py +++ b/sdks/python/src/telemetry_dev/_experiments.py @@ -621,6 +621,7 @@ def _run_items(api: Api, run: _Run, created: _Created, runs: _Runs) -> None: def worker() -> None: while not failures: try: + _check_key(api) item = runs.next() if item is None: return @@ -648,6 +649,7 @@ async def _run_items_async(api: Api, run: _Run, created: _Created, runs: _Runs) async def worker() -> None: while not failures: try: + _check_key(api) item = await asyncio.to_thread(runs.next) if item is None: return @@ -661,6 +663,43 @@ async def worker() -> None: await asyncio.to_thread(_end_items, batch, runs, failures) +def _check_key(api: Api) -> None: + """Raise when the ingest refused an export of the client with 401 since an export last + reached it, and the client sends with the key of the run to the same server. The key then + stopped working, for example because it was revoked or its project was deleted, and the API + will refuse the results of the items too.""" + client = get_client() + if ( + client is None + or client.config.api_key != api.api_key + or client.config.base_url != api.base_url + ): + return + rejected = client.export_failures.rejected + if rejected is not None and rejected.status == 401: + error = ExperimentError( + "invalid_api_key", + f"The API key was not accepted, so no more items run: {rejected}", + 401, + ) + error.__cause__ = rejected + raise error + + +def _traces() -> dict[str, Any]: + """For the results line of the experiment action: ``traced`` is False without an enabled + client, or when the ingest refused an export of the client with 401 or 403 since an export + last reached it, and ``tracesError`` then gives its reason. Exports that failed for another + reason, such as an outage, are not counted.""" + client = get_client() + if client is None or not client.enabled: + return {"traced": False} + rejected = client.export_failures.rejected + if rejected is not None: + return {"traced": False, "tracesError": str(rejected)} + return {"traced": True} + + def _end_items(batch: _Batch, runs: _Runs, failures: list[BaseException]) -> None: if not failures: try: @@ -899,7 +938,12 @@ def _finish(api: Api, run: _Run, created: _Created) -> ExperimentResult: _complete(api, created) result, body = _wait_for_result(api, created.id) # The action skips the check of a result with the CI mode off, and counts the items from runs. - line = {**cast("dict[str, Any]", body), "ci": run.in_ci, "runsPerItem": run.runs_per_item} + line = { + **cast("dict[str, Any]", body), + "ci": run.in_ci, + "runsPerItem": run.runs_per_item, + **_traces(), + } write_result(os.environ, line, api.on_error) print(_describe(result, run.runs_per_item), flush=True) if run.in_ci: diff --git a/sdks/python/src/telemetry_dev/_export.py b/sdks/python/src/telemetry_dev/_export.py index 99b0d9b8..a77fa308 100644 --- a/sdks/python/src/telemetry_dev/_export.py +++ b/sdks/python/src/telemetry_dev/_export.py @@ -2,6 +2,7 @@ import atexit import copy +import json import math import os import threading @@ -11,6 +12,7 @@ from enum import Enum from typing import Any, TypeVar, cast +import requests from opentelemetry.exporter.otlp.proto.common._log_encoder import encode_logs from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans from opentelemetry.sdk._logs.export import LogRecordExporter, LogRecordExportResult @@ -434,6 +436,8 @@ def shutdown(self) -> None: @contextmanager def exporting() -> Generator[None]: """Marks the current thread as inside an export of this SDK while the block runs.""" + if not getattr(_export_state, "depth", 0): + _export_state.refused = None _export_state.depth = getattr(_export_state, "depth", 0) + 1 try: yield @@ -441,6 +445,67 @@ def exporting() -> Generator[None]: _export_state.depth -= 1 +class IngestError(RuntimeError): + """A failed export, with the HTTP status that the ingest answered.""" + + def __init__(self, message: str, status: int) -> None: + super().__init__(message) + self.status = status + + +def otlp_session() -> requests.Session: + """A session for an OTLP exporter of this SDK. It keeps the last response of an export that + the ingest refused, so that the failure names the reason that the ingest gave. The OTLP + exporter sends on the thread of the export, thus the response is kept for that thread.""" + session = requests.Session() + session.hooks["response"].append(_keep_refused) + return session + + +def _keep_refused(response: requests.Response, *args: Any, **kwargs: Any) -> None: + _export_state.refused = None if response.ok else response + + +def export_error(label: str) -> RuntimeError: + """The error of a failed export of this thread: the status and the message of the response + that the ingest refused it with, when there was one.""" + response: requests.Response | None = getattr(_export_state, "refused", None) + if response is None: + return RuntimeError(f"OTLP {label} export failed (check the API key and ingest URL)") + reason = _status_message(response) + return IngestError( + f"telemetry.dev {_LABELS.get(label, label)} ingest failed: {response.status_code}" + + (f" {reason}" if reason else ""), + response.status_code, + ) + + +# The labels of the TypeScript transport, so that both SDKs report a failure the same way. +_LABELS = {"span": "trace"} + + +def _status_message(response: requests.Response) -> str | None: + """The message of the JSON or protobuf Status of an OTLP error response, as the telemetry.dev + ingest sends it, such as "This API key cannot send telemetry. Create a key with Send + telemetry access." None for another content type, or a body that cannot be read.""" + content_type = response.headers.get("content-type", "") + try: + # The exporter has read the whole body; a cut could split the Status. + body = response.content + if "json" in content_type: + message = json.loads(body).get("message") + elif "protobuf" in content_type: + # Imported here, so that a protobuf stack that cannot load it loses only the message. + from google.rpc.status_pb2 import Status + + message = Status.FromString(body).message + else: + return None + except Exception: + return None + return message.strip() or None if isinstance(message, str) else None + + # The keys of the work items whose thread has not finished, each with the process of its thread. # A fork that runs no hooks, as uWSGI's fork of its workers by default, copies the keys but not # the threads, so a key of another process does not count. @@ -527,9 +592,18 @@ def __init__(self, on_error: Callable[[BaseException], None] | None) -> None: self._on_error = on_error self._lock = ProcessLock(threading.Lock) self._reported = False + self._rejected: IngestError | None = None + + @property + def rejected(self) -> IngestError | None: + """The last export that the ingest refused with 401 (the API key is not valid) or 403 + (the key cannot send telemetry) since an export reached the ingest.""" + return self._rejected def report(self, message: str, exc: BaseException) -> None: with self._lock(): + if isinstance(exc, IngestError) and exc.status in (401, 403): + self._rejected = exc if self._reported: return self._reported = True @@ -541,6 +615,7 @@ def clear(self) -> None: """Records that an export reached the ingest, so that report() sends the next failure.""" with self._lock(): self._reported = False + self._rejected = None class ReportingSpanExporter(SpanExporter): @@ -577,10 +652,7 @@ def export(self, spans: Any) -> SpanExportResult: # Every record was dropped: the export did not fail, and the series goes on. return SpanExportResult.SUCCESS if result is SpanExportResult.FAILURE: - self._failures.report( - "span export failed", - RuntimeError("OTLP span export failed (check the API key and ingest URL)"), - ) + self._failures.report("span export failed", export_error("span")) else: self._failures.clear() return result @@ -625,10 +697,7 @@ def export(self, batch: Any) -> LogRecordExportResult: # Every record was dropped: the export did not fail, and the series goes on. return LogRecordExportResult.SUCCESS if result is LogRecordExportResult.FAILURE: - self._failures.report( - "log export failed", - RuntimeError("OTLP log export failed (check the API key and ingest URL)"), - ) + self._failures.report("log export failed", export_error("log")) else: self._failures.clear() return result diff --git a/sdks/python/src/telemetry_dev/_metrics.py b/sdks/python/src/telemetry_dev/_metrics.py index 8c0e9aaa..94375e80 100644 --- a/sdks/python/src/telemetry_dev/_metrics.py +++ b/sdks/python/src/telemetry_dev/_metrics.py @@ -27,7 +27,7 @@ from opentelemetry.trace import INVALID_SPAN_CONTEXT, NonRecordingSpan, set_span_in_context from ._config import report_error -from ._export import ExportFailures, exporting +from ._export import ExportFailures, export_error, exporting, otlp_session from ._process_lock import ProcessLock from ._semconv import ( ATTR_ERROR_TYPE, @@ -177,6 +177,7 @@ def __init__( failures: ExportFailures, **kwargs: Any, ) -> None: + kwargs.setdefault("session", otlp_session()) super().__init__(*args, **kwargs) # pyright: ignore[reportUnknownMemberType] self._td_failures = failures self.output_chunks: OutputChunkAggregation | None = None @@ -201,10 +202,7 @@ def export( self._td_failures.report("metric export failed", exc) return MetricExportResult.FAILURE if result is MetricExportResult.FAILURE: - self._td_failures.report( - "metric export failed", - RuntimeError("OTLP metric export failed (check the API key and ingest URL)"), - ) + self._td_failures.report("metric export failed", export_error("metric")) else: self._td_failures.clear() return result diff --git a/sdks/python/src/telemetry_dev/_spans.py b/sdks/python/src/telemetry_dev/_spans.py index 0dd277bb..7bb2215c 100644 --- a/sdks/python/src/telemetry_dev/_spans.py +++ b/sdks/python/src/telemetry_dev/_spans.py @@ -740,9 +740,12 @@ def start_span( ) attrs = _fields_to_attributes(client, state, fields) initial: dict[str, otel_types.AttributeValue] = {ATTR_OPERATION: operation} - if operation == "execute_tool": + # A span named after its operation, such as "invoke_agent" or "invoke_agent AGENT123" from + # an integration that names it by id, names no agent or tool. + named = name != operation and not name.startswith(f"{operation} ") + if operation == "execute_tool" and named: initial[ATTR_TOOL_NAME] = name - elif operation == "invoke_agent": + elif operation == "invoke_agent" and named: initial[ATTR_AGENT_NAME] = name ctx = context_from_parent(parent) propagated = propagated_attributes(ctx) diff --git a/sdks/python/src/telemetry_dev/otel.py b/sdks/python/src/telemetry_dev/otel.py index 8c1f3162..3272811a 100644 --- a/sdks/python/src/telemetry_dev/otel.py +++ b/sdks/python/src/telemetry_dev/otel.py @@ -28,6 +28,7 @@ ExportFailures, ReportingSpanExporter, after_export, + otlp_session, resolve_byte_bound, ) from ._metrics import GuardedOTLPMetricExporter, MetricsRecorder, sdk_meter_provider @@ -70,6 +71,7 @@ def _otlp_span_exporter( }, compression=Compression.Gzip, timeout=timeout, + session=otlp_session(), ) diff --git a/sdks/python/tests/e2e_scenarios.py b/sdks/python/tests/e2e_scenarios.py index 77ad9d30..5f49c15a 100644 --- a/sdks/python/tests/e2e_scenarios.py +++ b/sdks/python/tests/e2e_scenarios.py @@ -308,8 +308,9 @@ def main() -> int: # C13: continue the trace from a serialized W3C traceparent. start_span("background-job", parent=agent_traceparent).end() - # C20: a second root call of the same session joins the session trace. - start_span("follow-up-turn", type="agent", agent_name="support").end() + # C20: a second root call of the same session joins the session trace. With no agent_name, + # the agent takes the span name. + start_span("follow-up-turn", type="agent").end() # C10: log outside any span (standalone record, no trace correlation). log("outside spans", level="warn", event_name="e2e.outside") diff --git a/sdks/python/tests/fake_api.py b/sdks/python/tests/fake_api.py index 37484699..9f4bb3a1 100644 --- a/sdks/python/tests/fake_api.py +++ b/sdks/python/tests/fake_api.py @@ -189,7 +189,10 @@ def _answer(self) -> None: path=target.path.removeprefix(prefix), query=dict(parse_qsl(target.query)), headers={key.lower(): value for key, value in self.headers.items()}, - body=json.loads(raw) if raw else None, + # The OTLP exports of a client send protobuf. + body=json.loads(raw) + if raw and "json" in self.headers["content-type"] + else None, ) ) if isinstance(reply, Drop): diff --git a/sdks/python/tests/otlp_capture.py b/sdks/python/tests/otlp_capture.py index def242c4..e85d4352 100644 --- a/sdks/python/tests/otlp_capture.py +++ b/sdks/python/tests/otlp_capture.py @@ -37,6 +37,9 @@ class CaptureServer: status: int = 200 # The HTTP status of POST responses to one path, in place of status. path_status: dict[str, int] = field(default_factory=lambda: {}) + # The content type and the body of POST responses with an error status. + error_type: str = "application/x-protobuf" + error_body: bytes = b"" # The (status, JSON body) replies to GET requests, in order; the last one repeats. replies: list[tuple[int, object]] = field(default_factory=lambda: []) # Seconds that the server waits before it answers a GET request. The server reads it before the @@ -85,10 +88,15 @@ def do_POST(self) -> None: captured.append(request) time.sleep(capture.post_delay) # Not captured[-1]: requests to other paths can arrive on other threads meanwhile. - self.send_response(capture.path_status.get(request.path, capture.status)) - self.send_header("content-type", "application/x-protobuf") - self.send_header("content-length", "0") + status = capture.path_status.get(request.path, capture.status) + body = capture.error_body if status >= 400 else b"" + self.send_response(status) + self.send_header( + "content-type", capture.error_type if status >= 400 else "application/x-protobuf" + ) + self.send_header("content-length", str(len(body))) self.end_headers() + self.wfile.write(body) def do_GET(self) -> None: path = self.own_path() diff --git a/sdks/python/tests/test_experiments.py b/sdks/python/tests/test_experiments.py index 0ec9c2ac..7702f708 100644 --- a/sdks/python/tests/test_experiments.py +++ b/sdks/python/tests/test_experiments.py @@ -348,6 +348,85 @@ def task(item: ExperimentItem) -> str: ] +@pytest.mark.parametrize( + ("status", "key", "expected", "client_path"), + [ + # A 401 for the key of the run stops new items. In immediate mode an item exports its + # trace when it ends, so the refusal arrives before the next item starts. + (401, "td_live_test", 1, ""), + # A 401 for another key does not stop the run. + (401, "td_live_other", 5, ""), + # A 403 does not stop the run. + (403, "td_live_test", 5, ""), + # A 401 from another server does not stop the run. + (401, "td_live_test", 5, "/staging"), + ], +) +@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"]) +def test_a_trace_export_refused_with_401_for_the_key_of_the_run_stops_new_items( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: int, + key: str, + expected: int, + client_path: str, + asynchronous: bool, +) -> None: + results = tmp_path / "results.jsonl" + monkeypatch.setenv("TELEMETRY_DEV_EXPERIMENT_RESULTS", str(results)) + started = 0 + + def refuse(request: ApiRequest) -> Reply | None: + if request.path.endswith("/v1/traces"): + return answer_with(status, {"message": f"refused with {status}"}) + return None + + def task(item: ExperimentItem) -> str: + nonlocal started + started += 1 + return "ok" + + async def async_task(item: ExperimentItem) -> str: + return task(item) + + with fake_api(items=[dataset_item(n) for n in range(1, 6)], override=refuse) as api: + telemetry_dev.init( + api_key=key, + base_url=api.url + client_path, + environment="experiment", + export_mode="immediate", + log_level="silent", + disable_atexit=True, + ) + try: + outcome: object = run_both( + api, asynchronous, concurrency=1, task=async_task if asynchronous else task + ) + except ExperimentError as error: + outcome = error + finally: + telemetry_dev.shutdown() + + assert started == expected + if expected == 1: + assert isinstance(outcome, ExperimentError) + assert (outcome.code, outcome.status, str(outcome)) == ( + "invalid_api_key", + 401, + "The API key was not accepted, so no more items run: " + "telemetry.dev trace ingest failed: 401 refused with 401", + ) + assert api.routes()[-1] == f"POST /v1/experiments/{EXPERIMENT_ID}/complete" + else: + assert not isinstance(outcome, BaseException) + # The action notes that the runs have no traces, with the reason of the ingest. + line = json.loads(results.read_text()) + assert (line["traced"], line["tracesError"]) == ( + False, + f"telemetry.dev trace ingest failed: {status} refused with {status}", + ) + + def test_a_complete_that_fails_after_a_failed_run_goes_to_on_error(make: MakeClient) -> None: errors: list[BaseException] = [] make(environment="experiment", on_error=errors.append) @@ -864,7 +943,7 @@ def test_in_ci_mode_an_experiment_that_fails_raises_with_its_error( def test_in_github_actions_it_records_the_pull_request_its_head_commit_and_the_run( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, api: FakeApi + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, api: FakeApi, make: MakeClient ) -> None: event = tmp_path / "event.json" results = tmp_path / "results.jsonl" @@ -893,7 +972,10 @@ def test_in_github_actions_it_records_the_pull_request_its_head_commit_and_the_r }.items(): monkeypatch.setenv(name, value) + # With a client the runs send their traces; the second experiment runs without one. + make(environment="experiment") experiment(api, dataset=None, min_scores={"correctness": 0.5, "tone": 2}) + telemetry_dev.shutdown() create = api.requests[1].body assert create["dataset"] == "support-golden" @@ -911,12 +993,12 @@ def test_in_github_actions_it_records_the_pull_request_its_head_commit_and_the_r # for it, and counts its items from runsPerItem. experiment(api, ci=False, runs_per_item=2) - # The same line as the TypeScript SDK writes: the body, then ci and runsPerItem. + # The same line as the TypeScript SDK writes: the body, then ci, runsPerItem, and traced. assert results.read_text() == "".join( json.dumps(line, ensure_ascii=False, separators=(",", ":")) + "\n" for line in [ - {**result(), "ci": True, "runsPerItem": 1}, - {**result(), "ci": False, "runsPerItem": 2}, + {**result(), "ci": True, "runsPerItem": 1, "traced": True}, + {**result(), "ci": False, "runsPerItem": 2, "traced": False}, ] ) assert results.stat().st_mode & 0o777 == 0o600 diff --git a/sdks/python/tests/test_export_on_error.py b/sdks/python/tests/test_export_on_error.py index bbf01f02..a8eb17d0 100644 --- a/sdks/python/tests/test_export_on_error.py +++ b/sdks/python/tests/test_export_on_error.py @@ -3,6 +3,7 @@ # pyright: reportPrivateUsage=false import atexit import ctypes +import json import os import random import signal @@ -11,9 +12,11 @@ import threading import time from collections.abc import Callable, Mapping, Sequence +from concurrent.futures import ThreadPoolExecutor from typing import Any import pytest +from google.rpc.status_pb2 import Status from opentelemetry import trace from opentelemetry.exporter.otlp.proto.common._log_encoder import encode_logs from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans @@ -375,12 +378,102 @@ def test_an_ingest_outage_is_one_report_until_an_export_reaches_the_ingest_again telemetry_dev.shutdown() assert len(errors) == 2 - assert "span export failed" in str(errors[0]) - assert "log export failed" in str(errors[1]) + assert str(errors[0]) == "telemetry.dev trace ingest failed: 503" + assert str(errors[1]) == "telemetry.dev log ingest failed: 503" logged = [record for record in caplog.records if record.name == "telemetry_dev"] assert len(logged) == 2 +_NO_TELEMETRY = "This API key cannot send telemetry. Create a key with Send telemetry access." + + +@pytest.mark.parametrize( + ("status", "content_type", "body", "message", "rejected"), + [ + ( + 403, + "application/x-protobuf", + Status(code=7, message=_NO_TELEMETRY).SerializeToString(), + f"telemetry.dev trace ingest failed: 403 {_NO_TELEMETRY}", + 403, + ), + ( + 401, + "application/json", + json.dumps({"code": 16, "message": "API key was not accepted"}).encode(), + "telemetry.dev trace ingest failed: 401 API key was not accepted", + 401, + ), + # Bytes that read as a protobuf Status, from a proxy that answers with text. + (400, "text/plain", b"\x12\x04oops", "telemetry.dev trace ingest failed: 400", None), + ], +) +def test_a_refused_export_reports_the_message_of_the_ingest( + status: int, content_type: str, body: bytes, message: str, rejected: int | None +) -> None: + errors: list[BaseException] = [] + + with otlp_capture_server() as ingest: + ingest.status = status + ingest.error_type = content_type + ingest.error_body = body + client = telemetry_dev.init( + api_key="[REDACTED:api-key]", + base_url=ingest.url, + export_mode="immediate", + log_level="silent", + disable_atexit=True, + on_error=errors.append, + ) + telemetry_dev.start_span("step").end() + telemetry_dev.flush() + assert [str(error) for error in errors] == [message] + failures = client.export_failures + assert (failures.rejected and failures.rejected.status) == rejected + + # An export that reaches the ingest ends the rejection. + ingest.status = 200 + telemetry_dev.start_span("step").end() + telemetry_dev.flush() + assert failures.rejected is None + telemetry_dev.shutdown() + + +def test_an_export_with_no_response_does_not_report_an_earlier_refusal_of_its_thread() -> None: + errors: list[BaseException] = [] + + with otlp_capture_server() as ingest: + ingest.path_status["/v1/traces"] = 401 + client = telemetry_dev.init( + api_key="[REDACTED:api-key]", + base_url=ingest.url, + log_level="silent", + disable_atexit=True, + timeout=0.5, + on_error=errors.append, + ) + # A flush exports on the thread that calls it. The span exports run on one thread, and the + # log export on another. + with ThreadPoolExecutor(1) as span_thread: + telemetry_dev.start_span("refused").end() + span_thread.submit(telemetry_dev.flush).result() + # A log reaches the ingest on another thread, which ends the refusal. + telemetry_dev.log("reaches the ingest") + telemetry_dev.flush() + assert client.export_failures.rejected is None + # The next span export on the span thread gets no response before its timeout. + ingest.post_delay = 1.0 + telemetry_dev.start_span("times out").end() + span_thread.submit(telemetry_dev.flush).result() + ingest.post_delay = 0 + telemetry_dev.shutdown() + + assert str(errors[0]) == "telemetry.dev trace ingest failed: 401" + # The generic message, or the timeout itself from the exporter versions that raise it. + assert len(errors) == 2 + assert "401" not in str(errors[1]) + + def test_a_failure_during_a_slow_report_on_another_thread_is_quiet_and_does_not_wait() -> None: errors: list[BaseException] = [] first_report_started = threading.Event() @@ -430,7 +523,7 @@ def test_an_outage_that_stops_byo_traces_and_metrics_is_one_report() -> None: assert sorted(request.path for request in ingest.requests) == ["/v1/metrics", "/v1/traces"] assert len(errors) == 1 - assert "span export failed" in str(errors[0]) + assert str(errors[0]) == "telemetry.dev trace ingest failed: 400" _MAX_BODY_BYTES = 16 * 1024 * 1024 @@ -714,7 +807,9 @@ def test_a_split_batch_stops_at_its_first_failed_request(kind: str) -> None: telemetry_dev.shutdown() assert len(ingest.requests_for("/v1/traces" if kind == "span" else "/v1/logs")) == 1 - assert len([error for error in errors if f"{kind} export failed" in str(error)]) == 1 + label = "trace" if kind == "span" else kind + failed = f"telemetry.dev {label} ingest failed: 400" + assert len([error for error in errors if failed in str(error)]) == 1 @pytest.mark.parametrize("kind", ["span", "log"]) @@ -745,7 +840,9 @@ def record(name: str, attributes: dict[str, Any] | None = None) -> None: assert len(ingest.requests_for("/v1/traces" if kind == "span" else "/v1/logs")) == 2 assert any("exceeds max export size, dropped" in str(error) for error in errors) - assert len([error for error in errors if f"{kind} export failed" in str(error)]) == 1 + label = "trace" if kind == "span" else kind + failed = f"telemetry.dev {label} ingest failed: 400" + assert len([error for error in errors if failed in str(error)]) == 1 # The processors ignore an export's result, so check that the drop alone does not fail. spans = InMemorySpanExporter() diff --git a/sdks/python/tests/test_spans.py b/sdks/python/tests/test_spans.py index 238e9d86..b57aed67 100644 --- a/sdks/python/tests/test_spans.py +++ b/sdks/python/tests/test_spans.py @@ -84,6 +84,14 @@ def test_agent_span_defaults_agent_name(memory: SimpleNamespace) -> None: assert a["gen_ai.agent.id"] == "agent-7" +def test_a_span_named_after_its_operation_names_no_agent_or_tool(memory: SimpleNamespace) -> None: + telemetry_dev.start_span("invoke_agent AGENT123", type="agent").end() + telemetry_dev.start_span("execute_tool", type="tool").end() + agent, tool = (attrs(span) for span in memory.span_exporter.get_finished_spans()) + assert "gen_ai.agent.name" not in agent + assert "gen_ai.tool.name" not in tool + + def test_full_field_mapping(memory: SimpleNamespace) -> None: telemetry_dev.start_span( "gen", diff --git a/sdks/python/uv.lock b/sdks/python/uv.lock index ed63e154..724cf03e 100644 --- a/sdks/python/uv.lock +++ b/sdks/python/uv.lock @@ -407,9 +407,11 @@ name = "telemetry-dev" version = "0.3.2" source = { editable = "." } dependencies = [ + { name = "googleapis-common-protos" }, { name = "opentelemetry-api" }, { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-sdk" }, + { name = "requests" }, { name = "rfc8785" }, ] @@ -423,9 +425,11 @@ dev = [ [package.metadata] requires-dist = [ + { name = "googleapis-common-protos", specifier = ">=1.63.1,<2" }, { name = "opentelemetry-api", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = ">=1.39.0,<2" }, { name = "opentelemetry-sdk", specifier = ">=1.39.0,<2" }, + { name = "requests", specifier = ">=2.28,<3" }, { name = "rfc8785", specifier = ">=0.1.4" }, ]