You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by "seekskyworld (via GitHub)" <gi...@apache.org> on 2026/09/05 02:45:51 UTC
[PR] feat(telemetry): add opt-in OTLP usage traces [maka]
seekskyworld opened a new pull request, #4821:
URL: https://github.com/apache/maka/pull/4821
## Summary
Fixes #3757
Add an opt-in OTLP/HTTP exporter at the Runtime Host Usage boundary. It discovers the standard OTEL endpoint, header, service, and resource environment variables, batches model/tool spans, flushes during shutdown, and keeps prompts, message contents, tool arguments/results, credentials, and session paths out of exported data. Without an endpoint configured, the exporter remains a no-op.
## Verification
- `npm --workspace @maka/core run build`
- `npm --workspace @maka/storage run build`
- `npm --workspace @maka/storage run test:dist` (1,130 tests: 1,122 passed, 8 skipped)
- `node --test dist/__tests__/otlp-telemetry-exporter.test.js` (4 passed after final exporter fixes)
- `npx biome check` on changed source, test, and README files
- `npm run check:asf-headers`
## AI use
- [ ] No generative tool made a substantive contribution
- [x] Generative tooling made a substantive contribution
Tool(s) and scope: OpenAI Codex authored the exporter implementation, tests, and documentation under maintainer review.
## Checklist
- [x] Tests cover the change and fail without it
- [x] Lint, format, typecheck and the affected suites pass locally
Does this PR entail a change in behavior?
- [x] Yes — described under Summary above
- [ ] No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
Re: [PR] feat(telemetry): add opt-in OTLP usage traces [maka]
Posted by "hqhq1025 (via GitHub)" <gi...@apache.org>.
hqhq1025 commented on code in PR #4821:
URL: https://github.com/apache/maka/pull/4821#discussion_r3957476686
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ async close(): Promise<void> {
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ await this.flush();
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ try {
+ const response = await this.#fetch(this.#endpoint, {
Review Comment:
[P1] Bound each export request before joining it into shutdown. This `fetch` has no `AbortSignal` and the exporter ignores both OTLP timeout variables. I reproduced this through the real file-backed Usage writer with a local HTTP server that accepted the request but never returned headers: `stores.close()` was still pending after 750 ms and completed only after the socket was destroyed. `usage-stores.ts:447-452` awaits this close, the Runtime Host composition awaits storage close, and the Host then reaches its 10-second forced-termination deadline, skipping the rest of graceful cleanup. Please add a finite request timeout/cancellation path, honor the signal-specific/generic timeout configuration as appropriate, and ensure `close()` cannot wait past the Host shutdown budget.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
Review Comment:
[P2] Honor `OTEL_EXPORTER_OTLP_TRACES_HEADERS` before the generic header variable. It is the standard per-signal override, but this reads only `OTEL_EXPORTER_OTLP_HEADERS`. A probe with a trace endpoint plus `OTEL_EXPORTER_OTLP_TRACES_HEADERS=authorization=Bearer%20trace-token` produced a request with no authorization header, so a standards-compliant authenticated configuration silently drops every export. Parse the trace-specific value when present and fall back to the generic value otherwise, with the same validation.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ async close(): Promise<void> {
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ await this.flush();
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ try {
+ const response = await this.#fetch(this.#endpoint, {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...this.#headers },
+ body: JSON.stringify({
+ resourceSpans: [
+ {
+ resource: { attributes: this.#resourceAttributes },
+ scopeSpans: [{ scope: { name: 'maka.storage' }, spans }],
+ },
+ ],
+ }),
+ });
+ if (!response.ok) {
+ console.error(`[telemetry] OTLP export failed: HTTP ${response.status}`);
+ }
+ } catch {
+ console.error('[telemetry] OTLP export failed: request error');
+ }
+ }
+}
+
+function toSpan(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+}): OtlpSpan {
+ const traceId = randomUUID().replaceAll('-', '');
+ const startedAt = normalizeTimestamp(input.startedAt);
+ const endedAt = normalizeTimestamp(input.startedAt + Math.max(0, input.durationMs));
+ return {
+ traceId,
+ spanId: traceId.slice(0, 16),
+ name: input.name,
+ kind: 1,
+ startTimeUnixNano: unixNanoseconds(startedAt),
+ endTimeUnixNano: unixNanoseconds(endedAt),
+ attributes: input.attributes,
+ status: { code: input.status === 'success' ? 1 : input.status === 'error' ? 2 : 0 },
+ };
+}
+
+function stringAttribute(key: string, value: string): OtlpAttribute {
+ return { key, value: { stringValue: value } };
+}
+
+function numberAttribute(key: string, value: number): OtlpAttribute {
+ return { key, value: { doubleValue: value } };
+}
+
+function resolveEndpoint(env: Environment): string | undefined {
+ const configured = env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? env.OTEL_EXPORTER_OTLP_ENDPOINT;
+ if (!configured) return undefined;
+ try {
+ const url = new URL(configured);
+ const pathname = url.pathname.replace(/\/+$/u, '');
+ url.pathname = pathname.endsWith('/v1/traces') ? pathname : `${pathname}/v1/traces`;
Review Comment:
[P2] Do not append `/v1/traces` to `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`. The OTLP exporter specification requires a per-signal endpoint to be used as-is; only `OTEL_EXPORTER_OTLP_ENDPOINT` is a base URL that gets the signal path appended. On this head, `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=https://collector.example/custom/traces` sent to `https://collector.example/custom/traces/v1/traces`, so valid collectors with a custom trace route receive nothing. The current test covers only a URL that already ends in `/v1/traces`, which masks the branch. Resolve the two variables separately and add custom-path/root-path cases.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ async close(): Promise<void> {
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ await this.flush();
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ try {
+ const response = await this.#fetch(this.#endpoint, {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...this.#headers },
+ body: JSON.stringify({
+ resourceSpans: [
+ {
+ resource: { attributes: this.#resourceAttributes },
+ scopeSpans: [{ scope: { name: 'maka.storage' }, spans }],
+ },
+ ],
+ }),
+ });
+ if (!response.ok) {
+ console.error(`[telemetry] OTLP export failed: HTTP ${response.status}`);
Review Comment:
[P3] Suppress or back off repeated collector failures. With a continuing one-record-per-second workload and a collector returning 503, two consecutive batches produced two identical `console.error` lines; this continues once per flush interval while the collector is down. In the CLI/TUI process that can continuously pollute stderr. Log once until recovery or apply bounded exponential backoff.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ async close(): Promise<void> {
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ await this.flush();
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ try {
+ const response = await this.#fetch(this.#endpoint, {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...this.#headers },
+ body: JSON.stringify({
+ resourceSpans: [
+ {
+ resource: { attributes: this.#resourceAttributes },
+ scopeSpans: [{ scope: { name: 'maka.storage' }, spans }],
+ },
+ ],
+ }),
+ });
+ if (!response.ok) {
+ console.error(`[telemetry] OTLP export failed: HTTP ${response.status}`);
+ }
+ } catch {
+ console.error('[telemetry] OTLP export failed: request error');
+ }
+ }
+}
+
+function toSpan(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+}): OtlpSpan {
+ const traceId = randomUUID().replaceAll('-', '');
+ const startedAt = normalizeTimestamp(input.startedAt);
+ const endedAt = normalizeTimestamp(input.startedAt + Math.max(0, input.durationMs));
+ return {
+ traceId,
+ spanId: traceId.slice(0, 16),
+ name: input.name,
+ kind: 1,
+ startTimeUnixNano: unixNanoseconds(startedAt),
+ endTimeUnixNano: unixNanoseconds(endedAt),
+ attributes: input.attributes,
+ status: { code: input.status === 'success' ? 1 : input.status === 'error' ? 2 : 0 },
+ };
+}
+
+function stringAttribute(key: string, value: string): OtlpAttribute {
+ return { key, value: { stringValue: value } };
+}
+
+function numberAttribute(key: string, value: number): OtlpAttribute {
+ return { key, value: { doubleValue: value } };
+}
+
+function resolveEndpoint(env: Environment): string | undefined {
+ const configured = env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? env.OTEL_EXPORTER_OTLP_ENDPOINT;
+ if (!configured) return undefined;
+ try {
+ const url = new URL(configured);
+ const pathname = url.pathname.replace(/\/+$/u, '');
+ url.pathname = pathname.endsWith('/v1/traces') ? pathname : `${pathname}/v1/traces`;
+ return url.toString();
+ } catch {
+ return undefined;
+ }
+}
+
+function parseHeaders(value: string | undefined): Record<string, string> {
+ if (!value) return {};
+ const headers = Object.create(null) as Record<string, string>;
+ const names = new Set<string>();
+ let encodedBytes = 0;
+ for (const item of value.split(',')) {
+ const separator = item.indexOf('=');
+ if (separator <= 0) continue;
+ const key = item.slice(0, separator).trim();
+ const raw = item.slice(separator + 1).trim();
+ if (
+ !key ||
+ key.length > MAX_HEADER_NAME_LENGTH ||
+ !HEADER_NAME.test(key) ||
+ PROTECTED_HEADERS.has(key.toLowerCase())
+ ) {
+ continue;
+ }
+ const lowerKey = key.toLowerCase();
+ if (names.has(lowerKey) || names.size >= MAX_HEADER_COUNT) continue;
+ const decoded = decodeValue(raw);
+ if (
+ !raw ||
+ decoded.length > MAX_HEADER_VALUE_LENGTH ||
+ /[^\t\u0020-\u007e\u0080-\u00ff]/u.test(decoded)
+ ) {
+ continue;
+ }
+ const entryBytes = new TextEncoder().encode(`${key}:${decoded}`).byteLength;
+ if (encodedBytes + entryBytes > MAX_HEADER_BYTES) break;
+ encodedBytes += entryBytes;
+ names.add(lowerKey);
+ headers[key] = decoded;
+ }
+ return headers;
+}
+
+function warnForInsecureAuthorization(endpoint: string, headers: Record<string, string>): void {
+ let url: URL;
+ try {
+ url = new URL(endpoint);
+ } catch {
+ return;
+ }
+ if (url.protocol === 'https:') return;
+ const hasAuthorization = Object.keys(headers).some(
+ (name) => name.toLowerCase() === 'authorization',
+ );
+ if (hasAuthorization && !insecureAuthorizationWarnings.has(endpoint)) {
+ insecureAuthorizationWarnings.add(endpoint);
+ console.warn(
+ '[telemetry] OTLP endpoint is not using HTTPS while an authorization header is configured; credentials and telemetry will be sent without transport encryption',
+ );
+ }
+}
+
+function boundedErrorClass(value: string | undefined): string | undefined {
+ if (!value) return undefined;
+ const sanitized = value.trim().replace(/[\u0000-\u001f\u007f-\u009f]/gu, '');
Review Comment:
[P2] Truncation does not enforce the documented promise that credentials are never exported. The production tool failure path uses `classifyError(err)`, whose fallback preserves an arbitrary `Error.name`; with `error.name = 'sk-live-secret-value'`, the resulting `errorClass` was exported verbatim as `maka.error.class`. The public Usage record type is also an unrestricted string. Please project this field onto an allowlist of known classes (or redact credential-shaped values) at this outbound boundary rather than relying on length/control-character filtering.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
Re: [PR] feat(telemetry): add opt-in OTLP usage traces [maka]
Posted by "seekskyworld (via GitHub)" <gi...@apache.org>.
seekskyworld commented on code in PR #4821:
URL: https://github.com/apache/maka/pull/4821#discussion_r4017453275
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ async close(): Promise<void> {
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ await this.flush();
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ try {
+ const response = await this.#fetch(this.#endpoint, {
Review Comment:
Addressed in `f9d793381`. Each request receives an AbortSignal and a finite timeout; the trace-specific timeout overrides the generic timeout. Closing shares a separate 1-second cancellation grace across in-flight and pending batches, even for a 60-second configured request timeout. A real never-responding HTTP collector test drives the file-backed Usage writer and proves stores.close() settles; it fails with the old exporter. Cancellation also bounds a transport that ignores its signal. Storage 1,132 passed / 8 skipped; focused exporter and Usage tests 30/30.
Automated follow-up by OpenAI Codex for @seekskyworld.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ async close(): Promise<void> {
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ await this.flush();
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ try {
+ const response = await this.#fetch(this.#endpoint, {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...this.#headers },
+ body: JSON.stringify({
+ resourceSpans: [
+ {
+ resource: { attributes: this.#resourceAttributes },
+ scopeSpans: [{ scope: { name: 'maka.storage' }, spans }],
+ },
+ ],
+ }),
+ });
+ if (!response.ok) {
+ console.error(`[telemetry] OTLP export failed: HTTP ${response.status}`);
+ }
+ } catch {
+ console.error('[telemetry] OTLP export failed: request error');
+ }
+ }
+}
+
+function toSpan(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+}): OtlpSpan {
+ const traceId = randomUUID().replaceAll('-', '');
+ const startedAt = normalizeTimestamp(input.startedAt);
+ const endedAt = normalizeTimestamp(input.startedAt + Math.max(0, input.durationMs));
+ return {
+ traceId,
+ spanId: traceId.slice(0, 16),
+ name: input.name,
+ kind: 1,
+ startTimeUnixNano: unixNanoseconds(startedAt),
+ endTimeUnixNano: unixNanoseconds(endedAt),
+ attributes: input.attributes,
+ status: { code: input.status === 'success' ? 1 : input.status === 'error' ? 2 : 0 },
+ };
+}
+
+function stringAttribute(key: string, value: string): OtlpAttribute {
+ return { key, value: { stringValue: value } };
+}
+
+function numberAttribute(key: string, value: number): OtlpAttribute {
+ return { key, value: { doubleValue: value } };
+}
+
+function resolveEndpoint(env: Environment): string | undefined {
+ const configured = env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? env.OTEL_EXPORTER_OTLP_ENDPOINT;
+ if (!configured) return undefined;
+ try {
+ const url = new URL(configured);
+ const pathname = url.pathname.replace(/\/+$/u, '');
+ url.pathname = pathname.endsWith('/v1/traces') ? pathname : `${pathname}/v1/traces`;
Review Comment:
Addressed in `f9d793381`. Trace-specific endpoints are now used as-is, preserving custom paths, query strings, root paths and trailing slashes. Only the generic base endpoint gets /v1/traces appended. Updated the old trailing-slash expectation and added independent custom/root/generic-path cases; the old implementation fails them.
Automated follow-up by OpenAI Codex for @seekskyworld.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
Review Comment:
Addressed in `f9d793381`. OTEL_EXPORTER_OTLP_TRACES_HEADERS now overrides OTEL_EXPORTER_OTLP_HEADERS, including an explicit empty override. Tests cover authenticated trace headers, generic fallback, empty override, protected content-type and rejected control characters using the same parser.
Automated follow-up by OpenAI Codex for @seekskyworld.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ async close(): Promise<void> {
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ await this.flush();
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ try {
+ const response = await this.#fetch(this.#endpoint, {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...this.#headers },
+ body: JSON.stringify({
+ resourceSpans: [
+ {
+ resource: { attributes: this.#resourceAttributes },
+ scopeSpans: [{ scope: { name: 'maka.storage' }, spans }],
+ },
+ ],
+ }),
+ });
+ if (!response.ok) {
+ console.error(`[telemetry] OTLP export failed: HTTP ${response.status}`);
+ }
+ } catch {
+ console.error('[telemetry] OTLP export failed: request error');
+ }
+ }
+}
+
+function toSpan(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+}): OtlpSpan {
+ const traceId = randomUUID().replaceAll('-', '');
+ const startedAt = normalizeTimestamp(input.startedAt);
+ const endedAt = normalizeTimestamp(input.startedAt + Math.max(0, input.durationMs));
+ return {
+ traceId,
+ spanId: traceId.slice(0, 16),
+ name: input.name,
+ kind: 1,
+ startTimeUnixNano: unixNanoseconds(startedAt),
+ endTimeUnixNano: unixNanoseconds(endedAt),
+ attributes: input.attributes,
+ status: { code: input.status === 'success' ? 1 : input.status === 'error' ? 2 : 0 },
+ };
+}
+
+function stringAttribute(key: string, value: string): OtlpAttribute {
+ return { key, value: { stringValue: value } };
+}
+
+function numberAttribute(key: string, value: number): OtlpAttribute {
+ return { key, value: { doubleValue: value } };
+}
+
+function resolveEndpoint(env: Environment): string | undefined {
+ const configured = env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT ?? env.OTEL_EXPORTER_OTLP_ENDPOINT;
+ if (!configured) return undefined;
+ try {
+ const url = new URL(configured);
+ const pathname = url.pathname.replace(/\/+$/u, '');
+ url.pathname = pathname.endsWith('/v1/traces') ? pathname : `${pathname}/v1/traces`;
+ return url.toString();
+ } catch {
+ return undefined;
+ }
+}
+
+function parseHeaders(value: string | undefined): Record<string, string> {
+ if (!value) return {};
+ const headers = Object.create(null) as Record<string, string>;
+ const names = new Set<string>();
+ let encodedBytes = 0;
+ for (const item of value.split(',')) {
+ const separator = item.indexOf('=');
+ if (separator <= 0) continue;
+ const key = item.slice(0, separator).trim();
+ const raw = item.slice(separator + 1).trim();
+ if (
+ !key ||
+ key.length > MAX_HEADER_NAME_LENGTH ||
+ !HEADER_NAME.test(key) ||
+ PROTECTED_HEADERS.has(key.toLowerCase())
+ ) {
+ continue;
+ }
+ const lowerKey = key.toLowerCase();
+ if (names.has(lowerKey) || names.size >= MAX_HEADER_COUNT) continue;
+ const decoded = decodeValue(raw);
+ if (
+ !raw ||
+ decoded.length > MAX_HEADER_VALUE_LENGTH ||
+ /[^\t\u0020-\u007e\u0080-\u00ff]/u.test(decoded)
+ ) {
+ continue;
+ }
+ const entryBytes = new TextEncoder().encode(`${key}:${decoded}`).byteLength;
+ if (encodedBytes + entryBytes > MAX_HEADER_BYTES) break;
+ encodedBytes += entryBytes;
+ names.add(lowerKey);
+ headers[key] = decoded;
+ }
+ return headers;
+}
+
+function warnForInsecureAuthorization(endpoint: string, headers: Record<string, string>): void {
+ let url: URL;
+ try {
+ url = new URL(endpoint);
+ } catch {
+ return;
+ }
+ if (url.protocol === 'https:') return;
+ const hasAuthorization = Object.keys(headers).some(
+ (name) => name.toLowerCase() === 'authorization',
+ );
+ if (hasAuthorization && !insecureAuthorizationWarnings.has(endpoint)) {
+ insecureAuthorizationWarnings.add(endpoint);
+ console.warn(
+ '[telemetry] OTLP endpoint is not using HTTPS while an authorization header is configured; credentials and telemetry will be sent without transport encryption',
+ );
+ }
+}
+
+function boundedErrorClass(value: string | undefined): string | undefined {
+ if (!value) return undefined;
+ const sanitized = value.trim().replace(/[\u0000-\u001f\u007f-\u009f]/gu, '');
Review Comment:
Addressed in `f9d793381`. The outbound boundary now projects errorClass onto a fixed set of known runtime classes and maps all other names to Other. Both model and tool exports use this projection. Regression coverage includes sk-live-secret-value and an arbitrary password-shaped name, while known Timeout/InvalidArguments remain useful. The former truncation-only implementation fails these cases.
Automated follow-up by OpenAI Codex for @seekskyworld.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,348 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const MAX_ERROR_CLASS_LENGTH = 128;
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(env.OTEL_EXPORTER_OTLP_HEADERS);
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ async close(): Promise<void> {
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ await this.flush();
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ try {
+ const response = await this.#fetch(this.#endpoint, {
+ method: 'POST',
+ headers: { 'content-type': 'application/json', ...this.#headers },
+ body: JSON.stringify({
+ resourceSpans: [
+ {
+ resource: { attributes: this.#resourceAttributes },
+ scopeSpans: [{ scope: { name: 'maka.storage' }, spans }],
+ },
+ ],
+ }),
+ });
+ if (!response.ok) {
+ console.error(`[telemetry] OTLP export failed: HTTP ${response.status}`);
Review Comment:
Addressed in `f9d793381`. Collector failures are reported once until a successful export resets the warning state. A 503,503,200,503,503 sequence produces two logs; the previous implementation produces four. Requests remain best-effort without replaying failed batches.
Automated follow-up by OpenAI Codex for @seekskyworld.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
Re: [PR] feat(telemetry): add opt-in OTLP usage traces [maka]
Posted by "Astro-Han (via GitHub)" <gi...@apache.org>.
Astro-Han commented on code in PR #4821:
URL: https://github.com/apache/maka/pull/4821#discussion_r4237693132
##########
packages/storage/src/usage-stores.ts:
##########
@@ -500,9 +508,15 @@ function createWriterFacade(
latestLlmRuntimeProbe: (connectionSlug, modelId) =>
read(() => telemetry.latestLlmRuntimeProbe(connectionSlug, modelId)),
recordLlmCall: (record) =>
- admitSessionUsageMutation(record.sessionId, () => telemetry.insertLlmCall(record)),
+ admitSessionUsageMutation(record.sessionId, async () => {
+ await telemetry.insertLlmCall(record);
+ void exporter?.exportLlmCall(record);
Review Comment:
[P2] This exports only legacy LLM records, not normal conversation calls. The production Backend now records physical ModelCallAttempt commits through execution-model-composition.ts:266–280, then modelCalls.catchUpModelCallProjection; the old send-level recordLlmCall was deliberately removed. Through a real file-backed writer, one canonical attempt appeared in the local Usage summary but produced zero HTTP requests after flush, while tool records reached the same collector. Auxiliary calls still use this legacy hook and can mask the omission. Export canonical attempt settlements without reinstating a second send-level meter, and cover that actual accounting path.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,458 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import { setTimeout as delay } from 'node:timers/promises';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const DEFAULT_TIMEOUT_MS = 10_000;
+const INITIAL_RETRY_DELAY_MS = 1_000;
+const MAX_RETRY_DELAY_MS = 30_000;
+const RETRYABLE_HTTP_STATUSES = new Set([429, 502, 503, 504]);
+// Host 在 10 秒后强制终止；导出器关闭最多占用 1 秒，为其余资源清理保留时间。
+const CLOSE_TIMEOUT_MS = 1_000;
+const ERROR_CLASSES = new Set([
+ 'Abort',
+ 'Auth',
+ 'ContextLength',
+ 'Network',
+ 'Timeout',
+ 'RateLimit',
+ 'ProviderBilling',
+ 'ProviderCapacity',
+ 'ProviderUnavailable',
+ 'Other',
+ 'ExclusiveStepConflict',
+ 'InvalidArguments',
+ 'AmbiguousComputerTarget',
+ 'LoopGate',
+ 'DeferredNotLoaded',
+ 'ExecutionBoundaryUnavailable',
+ 'ClientCapabilityBoundary',
+ 'ClientCapabilityPreparation',
+ 'RuntimeLimit',
+ 'OutcomeUnknown',
+]);
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(
+ env.OTEL_EXPORTER_OTLP_TRACES_HEADERS || env.OTEL_EXPORTER_OTLP_HEADERS,
+ );
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ resolveTimeout(env),
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ readonly #timeoutMs: number;
+ readonly #shutdown = new AbortController();
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+ #closePromise: Promise<void> | undefined;
+ #failureReported = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ timeoutMs: number,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ this.#timeoutMs = timeoutMs;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
Review Comment:
[P2] Share a promise for the entire drain, not only the current send. After concurrent callers await batch A, one starts batch B and empties the queue; close's caller can then return and disarm the one-second shutdown timer even though B is active. With 64 persisted tool records and a real local collector, close returned in about 18 ms, but the isolated writer process remained alive another 10,005 ms under the default request timeout. A generated-code comparison sharing the whole drain cancelled B after one second and exited about 11 ms later; restoring this code reproduced the delay. Add a two-batch shutdown regression that verifies cancellation and natural process exit.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,458 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import { setTimeout as delay } from 'node:timers/promises';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const DEFAULT_TIMEOUT_MS = 10_000;
+const INITIAL_RETRY_DELAY_MS = 1_000;
+const MAX_RETRY_DELAY_MS = 30_000;
+const RETRYABLE_HTTP_STATUSES = new Set([429, 502, 503, 504]);
+// Host 在 10 秒后强制终止；导出器关闭最多占用 1 秒，为其余资源清理保留时间。
+const CLOSE_TIMEOUT_MS = 1_000;
+const ERROR_CLASSES = new Set([
+ 'Abort',
+ 'Auth',
+ 'ContextLength',
+ 'Network',
+ 'Timeout',
+ 'RateLimit',
+ 'ProviderBilling',
+ 'ProviderCapacity',
+ 'ProviderUnavailable',
+ 'Other',
+ 'ExclusiveStepConflict',
+ 'InvalidArguments',
+ 'AmbiguousComputerTarget',
+ 'LoopGate',
+ 'DeferredNotLoaded',
+ 'ExecutionBoundaryUnavailable',
+ 'ClientCapabilityBoundary',
+ 'ClientCapabilityPreparation',
+ 'RuntimeLimit',
+ 'OutcomeUnknown',
+]);
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(
+ env.OTEL_EXPORTER_OTLP_TRACES_HEADERS || env.OTEL_EXPORTER_OTLP_HEADERS,
+ );
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ resolveTimeout(env),
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ readonly #timeoutMs: number;
+ readonly #shutdown = new AbortController();
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+ #closePromise: Promise<void> | undefined;
+ #failureReported = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ timeoutMs: number,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ this.#timeoutMs = timeoutMs;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ close(): Promise<void> {
+ if (this.#closePromise) return this.#closePromise;
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ const timer = setTimeout(() => this.#shutdown.abort(), CLOSE_TIMEOUT_MS);
+ this.#closePromise = this.flush().finally(() => clearTimeout(timer));
+ return this.#closePromise;
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
Review Comment:
[P3] BATCH_SIZE is only a trigger, not a queue or batch ceiling. Storage callers deliberately do not await export, so a stalled request permits continued appends; splice(0) subsequently takes the entire backlog. A controlled 4,096-record backlog produced one 4,096-span request. Define explicit queue/batch limits and an overflow policy while keeping the local ledger authoritative. I did not reproduce memory exhaustion or collector rejection, so those are potential consequences, not demonstrated failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
Re: [PR] feat(telemetry): add opt-in OTLP usage traces [maka]
Posted by "seekskyworld (via GitHub)" <gi...@apache.org>.
seekskyworld commented on code in PR #4821:
URL: https://github.com/apache/maka/pull/4821#discussion_r4018101689
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,423 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const DEFAULT_TIMEOUT_MS = 10_000;
+// Host 在 10 秒后强制终止；导出器关闭最多占用 1 秒，为其余资源清理保留时间。
+const CLOSE_TIMEOUT_MS = 1_000;
+const ERROR_CLASSES = new Set([
+ 'Abort',
+ 'Auth',
+ 'ContextLength',
+ 'Network',
+ 'Timeout',
+ 'RateLimit',
+ 'ProviderBilling',
+ 'ProviderCapacity',
+ 'ProviderUnavailable',
+ 'Other',
+ 'ExclusiveStepConflict',
+ 'InvalidArguments',
+ 'AmbiguousComputerTarget',
+ 'LoopGate',
+ 'DeferredNotLoaded',
+ 'ExecutionBoundaryUnavailable',
+ 'ClientCapabilityBoundary',
+ 'ClientCapabilityPreparation',
+ 'RuntimeLimit',
+ 'OutcomeUnknown',
+]);
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(
+ env.OTEL_EXPORTER_OTLP_TRACES_HEADERS ?? env.OTEL_EXPORTER_OTLP_HEADERS,
Review Comment:
Addressed in 8dba766f4. Empty trace-specific endpoint, headers and timeout values now fall back to their generic counterparts; the generic endpoint still receives /v1/traces. Updated the README and regressions. The previous implementation fails all three empty-value cases. OpenAI Codex assisted with this follow-up.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,423 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const DEFAULT_TIMEOUT_MS = 10_000;
+// Host 在 10 秒后强制终止；导出器关闭最多占用 1 秒，为其余资源清理保留时间。
+const CLOSE_TIMEOUT_MS = 1_000;
+const ERROR_CLASSES = new Set([
+ 'Abort',
+ 'Auth',
+ 'ContextLength',
+ 'Network',
+ 'Timeout',
+ 'RateLimit',
+ 'ProviderBilling',
+ 'ProviderCapacity',
+ 'ProviderUnavailable',
+ 'Other',
+ 'ExclusiveStepConflict',
+ 'InvalidArguments',
+ 'AmbiguousComputerTarget',
+ 'LoopGate',
+ 'DeferredNotLoaded',
+ 'ExecutionBoundaryUnavailable',
+ 'ClientCapabilityBoundary',
+ 'ClientCapabilityPreparation',
+ 'RuntimeLimit',
+ 'OutcomeUnknown',
+]);
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(
+ env.OTEL_EXPORTER_OTLP_TRACES_HEADERS ?? env.OTEL_EXPORTER_OTLP_HEADERS,
+ );
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ resolveTimeout(env),
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ readonly #timeoutMs: number;
+ readonly #shutdown = new AbortController();
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+ #closePromise: Promise<void> | undefined;
+ #failureReported = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ timeoutMs: number,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ this.#timeoutMs = timeoutMs;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ close(): Promise<void> {
+ if (this.#closePromise) return this.#closePromise;
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ const timer = setTimeout(() => this.#shutdown.abort(), CLOSE_TIMEOUT_MS);
+ this.#closePromise = this.flush().finally(() => clearTimeout(timer));
+ return this.#closePromise;
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ if (this.#shutdown.signal.aborted) return;
+ const controller = new AbortController();
+ const abort = () => controller.abort();
+ const timer = setTimeout(abort, this.#timeoutMs);
+ this.#shutdown.signal.addEventListener('abort', abort, { once: true });
+ const cancelled = new Promise<never>((_resolve, reject) => {
+ controller.signal.addEventListener(
+ 'abort',
+ () => reject(new Error('OTLP request cancelled')),
+ { once: true },
+ );
+ });
+ try {
+ // 同时约束请求及响应释放；即使注入的 transport 不响应 AbortSignal，也不阻塞关闭。
+ const response = await Promise.race([this.sendRequest(spans, controller.signal), cancelled]);
+ if (response.ok) {
+ this.#failureReported = false;
+ } else {
+ this.reportFailure(`HTTP ${response.status}`);
Review Comment:
Addressed in 8dba766f4. Connection failures and HTTP 429/502/503/504 retry the identical batch with exponential backoff and jitter, honoring Retry-After seconds/dates. All attempts and backoff waits share the original batch timeout and shutdown cancellation. Permanent HTTP failures remain non-retryable. Tests cover retry payload identity, response-body cancellation, exponential waits, and cancellation during Retry-After. All 14 exporter tests and the Storage suite (1135 pass, 8 platform skips), full repository build/typecheck/lint/format and Desktop/UI knip pass. OpenAI Codex assisted with this follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
Re: [PR] feat(telemetry): add opt-in OTLP usage traces [maka]
Posted by "seekskyworld (via GitHub)" <gi...@apache.org>.
seekskyworld commented on PR #4821:
URL: https://github.com/apache/maka/pull/4821#issuecomment-6095650803
Confirmed by inspecting exportLlmCall/buildLlmSpan: spans are constructed from an explicit usage/identity/status attribute allowlist; the exporter does not serialize the input record or accept prompt/response bodies. The existing payload assertions exclude argsSummary, and arbitrary error classes are normalized to an allowlist (the secret-value regression also checks the serialized request). This is a verification of the current implementation, not a claim that user-configured OTEL_RESOURCE_ATTRIBUTES cannot contain user-supplied text.
Automated update by OpenAI Codex under the contributor’s direction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
Re: [PR] feat(telemetry): add opt-in OTLP usage traces [maka]
Posted by "seekskyworld (via GitHub)" <gi...@apache.org>.
seekskyworld commented on PR #4821:
URL: https://github.com/apache/maka/pull/4821#issuecomment-5564870232
Addressed the review items in commit `6080c4fc4`:
- Removed the nested storage-root lease around `insertLlmCall` and `insertToolInvocation`; the admission lease now covers the durable write and exporter enqueue directly.
- Hardened `OTEL_EXPORTER_OTLP_HEADERS`: token grammar and size limits, duplicate-name rejection, control-character rejection, and protected transport headers (including `content-type`) are enforced. `authorization` remains supported for collectors.
- Added a one-time-per-configured-endpoint warning when authorization is sent to a non-HTTPS endpoint; HTTP remains allowed for local/LAN collectors.
- Bounded exported `errorClass` values at the OTLP boundary to 128 Unicode characters and removed control characters.
- Switched Unix-nanosecond conversion to integer `BigInt` arithmetic and documented the independent-event semantics.
Validation: `@maka/storage` build/typecheck, 1,134 storage tests (1,126 passed, 8 platform skips), OTLP focused tests (6/6), Biome, ASF header audit, and `git diff --check` all pass. The backoff/kill-switch suggestion remains a separate non-blocking follow-up because it changes exporter delivery policy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
Re: [PR] feat(telemetry): add opt-in OTLP usage traces [maka]
Posted by "hqhq1025 (via GitHub)" <gi...@apache.org>.
hqhq1025 commented on code in PR #4821:
URL: https://github.com/apache/maka/pull/4821#discussion_r4017781071
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,423 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const DEFAULT_TIMEOUT_MS = 10_000;
+// Host 在 10 秒后强制终止；导出器关闭最多占用 1 秒，为其余资源清理保留时间。
+const CLOSE_TIMEOUT_MS = 1_000;
+const ERROR_CLASSES = new Set([
+ 'Abort',
+ 'Auth',
+ 'ContextLength',
+ 'Network',
+ 'Timeout',
+ 'RateLimit',
+ 'ProviderBilling',
+ 'ProviderCapacity',
+ 'ProviderUnavailable',
+ 'Other',
+ 'ExclusiveStepConflict',
+ 'InvalidArguments',
+ 'AmbiguousComputerTarget',
+ 'LoopGate',
+ 'DeferredNotLoaded',
+ 'ExecutionBoundaryUnavailable',
+ 'ClientCapabilityBoundary',
+ 'ClientCapabilityPreparation',
+ 'RuntimeLimit',
+ 'OutcomeUnknown',
+]);
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(
+ env.OTEL_EXPORTER_OTLP_TRACES_HEADERS ?? env.OTEL_EXPORTER_OTLP_HEADERS,
+ );
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ resolveTimeout(env),
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ readonly #timeoutMs: number;
+ readonly #shutdown = new AbortController();
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+ #closePromise: Promise<void> | undefined;
+ #failureReported = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ timeoutMs: number,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ this.#timeoutMs = timeoutMs;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ close(): Promise<void> {
+ if (this.#closePromise) return this.#closePromise;
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ const timer = setTimeout(() => this.#shutdown.abort(), CLOSE_TIMEOUT_MS);
+ this.#closePromise = this.flush().finally(() => clearTimeout(timer));
+ return this.#closePromise;
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
+ if (this.#pending.length >= BATCH_SIZE) {
+ await this.flush();
+ return;
+ }
+ if (!this.#timer) {
+ this.#timer = setTimeout(() => {
+ this.#timer = undefined;
+ void this.flush();
+ }, FLUSH_DELAY_MS);
+ }
+ }
+
+ private async send(spans: OtlpSpan[]): Promise<void> {
+ if (this.#shutdown.signal.aborted) return;
+ const controller = new AbortController();
+ const abort = () => controller.abort();
+ const timer = setTimeout(abort, this.#timeoutMs);
+ this.#shutdown.signal.addEventListener('abort', abort, { once: true });
+ const cancelled = new Promise<never>((_resolve, reject) => {
+ controller.signal.addEventListener(
+ 'abort',
+ () => reject(new Error('OTLP request cancelled')),
+ { once: true },
+ );
+ });
+ try {
+ // 同时约束请求及响应释放；即使注入的 transport 不响应 AbortSignal，也不阻塞关闭。
+ const response = await Promise.race([this.sendRequest(spans, controller.signal), cancelled]);
+ if (response.ok) {
+ this.#failureReported = false;
+ } else {
+ this.reportFailure(`HTTP ${response.status}`);
Review Comment:
[P2] Retry transient OTLP failures instead of dropping this batch after the first attempt. `flush()` removes the spans from `#pending`, and this branch only logs, so retryable 429/502/503/504 responses and connection failures are permanently lost. A probe where the first response was `503 Retry-After: 0` and the next would be 200 made exactly one request even after another `flush()` and `close()`. The stable OTLP exporter specification requires transient errors to use exponential backoff with jitter; best-effort delivery can remain non-fatal while still retrying within the configured batch timeout and honoring shutdown cancellation. Spec: https://opentelemetry.io/docs/specs/otel/protocol/exporter/#retry
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,423 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const DEFAULT_TIMEOUT_MS = 10_000;
+// Host 在 10 秒后强制终止；导出器关闭最多占用 1 秒，为其余资源清理保留时间。
+const CLOSE_TIMEOUT_MS = 1_000;
+const ERROR_CLASSES = new Set([
+ 'Abort',
+ 'Auth',
+ 'ContextLength',
+ 'Network',
+ 'Timeout',
+ 'RateLimit',
+ 'ProviderBilling',
+ 'ProviderCapacity',
+ 'ProviderUnavailable',
+ 'Other',
+ 'ExclusiveStepConflict',
+ 'InvalidArguments',
+ 'AmbiguousComputerTarget',
+ 'LoopGate',
+ 'DeferredNotLoaded',
+ 'ExecutionBoundaryUnavailable',
+ 'ClientCapabilityBoundary',
+ 'ClientCapabilityPreparation',
+ 'RuntimeLimit',
+ 'OutcomeUnknown',
+]);
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(
+ env.OTEL_EXPORTER_OTLP_TRACES_HEADERS ?? env.OTEL_EXPORTER_OTLP_HEADERS,
Review Comment:
[P2] Treat empty signal-specific environment values as unset before applying precedence. The OpenTelemetry environment-variable specification requires an empty value to behave exactly like an unset variable, but `??` selects `''` here. On this head, `OTEL_EXPORTER_OTLP_TRACES_HEADERS=''` removes a valid generic authorization header; the same pattern at `resolveEndpoint` disables the exporter when the generic endpoint is valid, and at `resolveTimeout` ignores a valid generic timeout. I reproduced the endpoint case as `exporter === undefined` and the header case as a request containing only `content-type`. Please fall back to the generic value for empty signal-specific strings and update the README/test that currently defines empty headers as an override. Spec: https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#parsing-empty-value
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org
Re: [PR] feat(telemetry): add opt-in OTLP usage traces [maka]
Posted by "seekskyworld (via GitHub)" <gi...@apache.org>.
seekskyworld commented on code in PR #4821:
URL: https://github.com/apache/maka/pull/4821#discussion_r4246213405
##########
packages/storage/src/usage-stores.ts:
##########
@@ -500,9 +508,15 @@ function createWriterFacade(
latestLlmRuntimeProbe: (connectionSlug, modelId) =>
read(() => telemetry.latestLlmRuntimeProbe(connectionSlug, modelId)),
recordLlmCall: (record) =>
- admitSessionUsageMutation(record.sessionId, () => telemetry.insertLlmCall(record)),
+ admitSessionUsageMutation(record.sessionId, async () => {
+ await telemetry.insertLlmCall(record);
+ void exporter?.exportLlmCall(record);
Review Comment:
Fixed in `63a3d2b1f`. The Usage facade observes newly inserted canonical attempts only after the projection transaction commits. The callback does not await network I/O or create a second send-level record. Real file-backed authority events now reach a local collector; repeated catch-up/reopen and duplicate attempt events do not re-export rows. Missing usage/unpriced cost stays absent, and rollback does not export. The old merged implementation fails with zero spans; current Node 22 focused tests (75) and full Storage (1,654 pass / 8 skip) pass.
Automated update by OpenAI Codex under the contributor’s direction.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,458 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import { setTimeout as delay } from 'node:timers/promises';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const DEFAULT_TIMEOUT_MS = 10_000;
+const INITIAL_RETRY_DELAY_MS = 1_000;
+const MAX_RETRY_DELAY_MS = 30_000;
+const RETRYABLE_HTTP_STATUSES = new Set([429, 502, 503, 504]);
+// Host 在 10 秒后强制终止；导出器关闭最多占用 1 秒，为其余资源清理保留时间。
+const CLOSE_TIMEOUT_MS = 1_000;
+const ERROR_CLASSES = new Set([
+ 'Abort',
+ 'Auth',
+ 'ContextLength',
+ 'Network',
+ 'Timeout',
+ 'RateLimit',
+ 'ProviderBilling',
+ 'ProviderCapacity',
+ 'ProviderUnavailable',
+ 'Other',
+ 'ExclusiveStepConflict',
+ 'InvalidArguments',
+ 'AmbiguousComputerTarget',
+ 'LoopGate',
+ 'DeferredNotLoaded',
+ 'ExecutionBoundaryUnavailable',
+ 'ClientCapabilityBoundary',
+ 'ClientCapabilityPreparation',
+ 'RuntimeLimit',
+ 'OutcomeUnknown',
+]);
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(
+ env.OTEL_EXPORTER_OTLP_TRACES_HEADERS || env.OTEL_EXPORTER_OTLP_HEADERS,
+ );
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ resolveTimeout(env),
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ readonly #timeoutMs: number;
+ readonly #shutdown = new AbortController();
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+ #closePromise: Promise<void> | undefined;
+ #failureReported = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ timeoutMs: number,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ this.#timeoutMs = timeoutMs;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
Review Comment:
Fixed in `63a3d2b1f`: concurrent flush callers share the promise for the entire drain, including later batches. Added an isolated real writer process with 64 persisted tool records and an HTTP collector that releases batch one during close and stalls batch two. The test asserts cancellation and natural process exit within five seconds under the default ten-second request timeout. It fails on the old implementation; the fixed writer exits after the one-second grace on Node 24 and Node 22.
Automated update by OpenAI Codex under the contributor’s direction.
##########
packages/storage/src/otlp-telemetry-exporter.ts:
##########
@@ -0,0 +1,458 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { randomUUID } from 'node:crypto';
+import { setTimeout as delay } from 'node:timers/promises';
+import type {
+ PersistedLlmCallRecord,
+ PersistedToolInvocationRecord,
+} from './telemetry-file-schema.js';
+
+const BATCH_SIZE = 32;
+const FLUSH_DELAY_MS = 1_000;
+const DEFAULT_SERVICE_NAME = 'maka';
+const MAX_HEADER_COUNT = 32;
+const MAX_HEADER_NAME_LENGTH = 128;
+const MAX_HEADER_VALUE_LENGTH = 8_192;
+const MAX_HEADER_BYTES = 64 * 1_024;
+const DEFAULT_TIMEOUT_MS = 10_000;
+const INITIAL_RETRY_DELAY_MS = 1_000;
+const MAX_RETRY_DELAY_MS = 30_000;
+const RETRYABLE_HTTP_STATUSES = new Set([429, 502, 503, 504]);
+// Host 在 10 秒后强制终止；导出器关闭最多占用 1 秒，为其余资源清理保留时间。
+const CLOSE_TIMEOUT_MS = 1_000;
+const ERROR_CLASSES = new Set([
+ 'Abort',
+ 'Auth',
+ 'ContextLength',
+ 'Network',
+ 'Timeout',
+ 'RateLimit',
+ 'ProviderBilling',
+ 'ProviderCapacity',
+ 'ProviderUnavailable',
+ 'Other',
+ 'ExclusiveStepConflict',
+ 'InvalidArguments',
+ 'AmbiguousComputerTarget',
+ 'LoopGate',
+ 'DeferredNotLoaded',
+ 'ExecutionBoundaryUnavailable',
+ 'ClientCapabilityBoundary',
+ 'ClientCapabilityPreparation',
+ 'RuntimeLimit',
+ 'OutcomeUnknown',
+]);
+const HEADER_NAME = /^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/u;
+const PROTECTED_HEADERS = new Set([
+ 'connection',
+ 'content-length',
+ 'content-type',
+ 'host',
+ 'proxy-connection',
+ 'transfer-encoding',
+]);
+const insecureAuthorizationWarnings = new Set<string>();
+
+type FetchLike = (input: string, init?: RequestInit) => Promise<Response>;
+type Environment = Record<string, string | undefined>;
+type OtlpAttributeValue = { stringValue: string } | { intValue: string } | { doubleValue: number };
+type OtlpAttribute = { key: string; value: OtlpAttributeValue };
+type OtlpSpan = {
+ traceId: string;
+ spanId: string;
+ name: string;
+ kind: 1;
+ startTimeUnixNano: string;
+ endTimeUnixNano: string;
+ attributes: OtlpAttribute[];
+ status: { code: 0 | 1 | 2 };
+};
+
+export interface OtlpTelemetryExporter {
+ exportLlmCall(record: PersistedLlmCallRecord): Promise<void>;
+ exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void>;
+ flush(): Promise<void>;
+ close(): Promise<void>;
+}
+
+export interface OtlpTelemetryExporterOptions {
+ readonly env?: Environment;
+ readonly fetch?: FetchLike;
+}
+
+export function createOtlpTelemetryExporter(
+ options: OtlpTelemetryExporterOptions = {},
+): OtlpTelemetryExporter | undefined {
+ const env = options.env ?? process.env;
+ const endpoint = resolveEndpoint(env);
+ if (!endpoint) return undefined;
+ const headers = parseHeaders(
+ env.OTEL_EXPORTER_OTLP_TRACES_HEADERS || env.OTEL_EXPORTER_OTLP_HEADERS,
+ );
+ warnForInsecureAuthorization(endpoint, headers);
+ return new OtlpTelemetryExporterImpl(
+ endpoint,
+ headers,
+ resourceAttributes(env),
+ options.fetch ?? fetch,
+ resolveTimeout(env),
+ );
+}
+
+class OtlpTelemetryExporterImpl implements OtlpTelemetryExporter {
+ readonly #endpoint: string;
+ readonly #headers: Record<string, string>;
+ readonly #resourceAttributes: OtlpAttribute[];
+ readonly #fetch: FetchLike;
+ readonly #pending: OtlpSpan[] = [];
+ readonly #timeoutMs: number;
+ readonly #shutdown = new AbortController();
+ #timer: NodeJS.Timeout | undefined;
+ #flushPromise: Promise<void> | undefined;
+ #closed = false;
+ #closePromise: Promise<void> | undefined;
+ #failureReported = false;
+
+ constructor(
+ endpoint: string,
+ headers: Record<string, string>,
+ resourceAttributes: OtlpAttribute[],
+ fetchFn: FetchLike,
+ timeoutMs: number,
+ ) {
+ this.#endpoint = endpoint;
+ this.#headers = headers;
+ this.#resourceAttributes = resourceAttributes;
+ this.#fetch = fetchFn;
+ this.#timeoutMs = timeoutMs;
+ }
+
+ async exportLlmCall(record: PersistedLlmCallRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.llm.call',
+ startedAt: record.startedAt,
+ durationMs: record.latencyMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'llm'),
+ stringAttribute('maka.provider.id', record.providerId),
+ stringAttribute('maka.model.id', record.modelId),
+ ...(record.callKind ? [stringAttribute('maka.call.kind', record.callKind)] : []),
+ numberAttribute('maka.usage.input_tokens', record.inputTokens),
+ numberAttribute('maka.usage.output_tokens', record.outputTokens),
+ numberAttribute('maka.usage.total_tokens', record.totalTokens),
+ numberAttribute('maka.usage.cost_usd', record.costUsd),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async exportToolInvocation(record: PersistedToolInvocationRecord): Promise<void> {
+ const errorClass = boundedErrorClass(record.errorClass);
+ await this.enqueue({
+ name: 'maka.tool.invocation',
+ startedAt: record.startedAt,
+ durationMs: record.durationMs,
+ status: record.status,
+ attributes: [
+ stringAttribute('maka.telemetry.kind', 'tool'),
+ stringAttribute('maka.tool.name', record.toolName),
+ ...(record.providerId ? [stringAttribute('maka.provider.id', record.providerId)] : []),
+ ...(record.modelId ? [stringAttribute('maka.model.id', record.modelId)] : []),
+ numberAttribute('maka.tool.bytes_in', record.bytesIn),
+ numberAttribute('maka.tool.bytes_out', record.bytesOut),
+ ...(errorClass ? [stringAttribute('maka.error.class', errorClass)] : []),
+ ],
+ });
+ }
+
+ async flush(): Promise<void> {
+ if (this.#flushPromise) await this.#flushPromise;
+ while (this.#pending.length > 0) {
+ const spans = this.#pending.splice(0);
+ if (this.#timer) {
+ clearTimeout(this.#timer);
+ this.#timer = undefined;
+ }
+ this.#flushPromise = this.send(spans).finally(() => {
+ this.#flushPromise = undefined;
+ });
+ await this.#flushPromise;
+ }
+ }
+
+ close(): Promise<void> {
+ if (this.#closePromise) return this.#closePromise;
+ this.#closed = true;
+ if (this.#timer) clearTimeout(this.#timer);
+ this.#timer = undefined;
+ const timer = setTimeout(() => this.#shutdown.abort(), CLOSE_TIMEOUT_MS);
+ this.#closePromise = this.flush().finally(() => clearTimeout(timer));
+ return this.#closePromise;
+ }
+
+ private async enqueue(input: {
+ name: string;
+ startedAt: number;
+ durationMs: number;
+ status: PersistedLlmCallRecord['status'];
+ attributes: OtlpAttribute[];
+ }): Promise<void> {
+ if (this.#closed) return;
+ this.#pending.push(toSpan(input));
Review Comment:
Fixed in `63a3d2b1f`: requests are capped at 32 spans, pending spans at 1,024 plus one in-flight batch, and overflow drops newest optional exports. The local ledger remains authoritative. A stalled 4,096-record producer regression verifies the cap and retained order; the old implementation fails. Limits and loss semantics are documented in the CLI README and PR description.
Automated update by OpenAI Codex under the contributor’s direction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org