From e3884f8d1b14bb54aafd8de899022aeb2470b172 Mon Sep 17 00:00:00 2001 From: akash-vijay-kv Date: Mon, 13 Jul 2026 11:08:56 +0530 Subject: [PATCH 1/4] chore: Update CHANGELOG with all latest changes --- CHANGELOG.md | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 07e3434..e2aad5a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -55,6 +55,56 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **CommonJS (CJS) Support for LLM Provider Instrumentations**: Fixed instrumentation for Anthropic, Google GenAI, Groq, and Mistral AI so they patch correctly under CommonJS module resolution, resolving instrumentation failures in CJS applications. +## [1.5.0] - 2026-07-10 + +### Added + +- **Google GenAI Instrumentation**: Added instrumentation support for the Google GenAI (`@google/genai`) SDK, with a shared `base-instrumentor` abstraction and a separate `google-generative-ai` module. + +### Fixed + +- **OpenAI Agents Cache Tokens**: Captured cache token details (`cached_tokens`, `cache_write_tokens`, `reasoning_tokens`) for Claude models in OpenAI Agents instrumentation. +- **Raw Chat Completions Response**: Added a helper to handle usage from the Chat Completions API naming (`prompt_tokens`/`completion_tokens` and their token details) alongside the Responses API naming, so `OpenAIChatCompletionsModel` usage is captured correctly. +- **Span Input/Output on Active and Root Spans**: Updated utility functions so input and output attributes are set correctly on both the active span and the root span. + +## [1.4.0] - 2026-07-02 + +### Added + +- **Anthropic Tool Runner Support**: Added span capture for the Anthropic tool-runner flow, so tool-execution turns within an agentic Anthropic run are traced. +- **Attribute Size Limit Processor**: New `AttributeSizeLimitProcessor` enforces a hard per-attribute size cap (default 32KB, configurable via `NETRA_SPAN_ATTRIBUTE_MAX_SIZE`) by wrapping `setAttribute` on span start, preventing "entity too large" errors during export. +- **Resource Attributes Propagation**: `Config` now sets `OTEL_RESOURCE_ATTRIBUTES` so the `TracerProvider` resource carries `service.name` and `deployment.environment`, with precedence order: pre-existing env value > config `resourceAttributes` > Netra defaults. + +### Changed + +- **SDK Version Source**: `Config.LIBRARY_VERSION` now derives from `SDK_VERSION` in `src/version.ts` instead of a hardcoded string, keeping the reported library version in sync with the package version. + +## [1.3.0] - 2026-07-01 + +### Added + +- **NativeTracingMode**: Replaced the boolean `disableNativeTracing` option with a `NativeTracingMode` (`"both" | "netra" | "netra-strict"`) for granular control over where OpenAI Agents traces are sent. Configurable per-instrument or via the `NATIVE_TRACING_MODE` environment variable. `"netra"` routes traces to Netra only (falling back to additive mode with a warning if SDK APIs are unavailable), `"netra-strict"` (default) never falls back to native, and `"both"` sends to Netra and native. + +### Fixed + +- **Netra-only OpenAI Agents Tracing**: Added `canSet` gating and a restore fallback so the OpenAI Agents processor can be swapped safely, and included `OPENAI_AGENTS` in the root-span instrument allowlist (`DEFAULT_INSTRUMENTS_FOR_ROOT`) so agent runs are captured as root spans. + +## [1.2.0] - 2026-06-23 + +### Added + +- **Simulation File Handling**: The simulation workflow now supports file attachments on user messages. The SDK parses `attachments` metadata from the API (`FileData`), downloads the referenced content via pre-signed URLs, and delivers it to user tasks as base64-encoded `ProcessedFile` objects. `ProcessedFile` is now exported from the package root. + +## [1.1.0] - 2026-06-16 + +### Added + +- **Context Propagation Helpers**: Exported `netraExpressMiddleware` and `runWithExtractedContext` for distributed tracing. These utilities extract incoming W3C Trace Context from HTTP headers and run code within that context, covering cases where auto-instrumentation is unavailable (ESM load-order issues, missing peer dependencies, or non-Express frameworks). + +### Fixed + +- **CommonJS (CJS) Support for LLM Provider Instrumentations**: Fixed instrumentation for Anthropic, Google GenAI, Groq, and Mistral AI so they patch correctly under CommonJS module resolution, resolving instrumentation failures in CJS applications. + ## [1.1.0-beta.1] - 2026-06-09 ### Fixed From efb3e03ab6eed75dddcf67367563a92c42a39020 Mon Sep 17 00:00:00 2001 From: Akash Vijay Date: Fri, 17 Jul 2026 00:39:22 +0530 Subject: [PATCH 2/4] [NET-1367] feat: Reparent children of blocked root instruments instead of dropping subtree (#115) --- src/config.ts | 20 + src/exporters/filteringSpanExporter.ts | 220 ++++++++--- src/exporters/utils.ts | 254 +++++++++++++ .../root-instrument-filter-processor.ts | 342 +++++++++++++----- 4 files changed, 690 insertions(+), 146 deletions(-) diff --git a/src/config.ts b/src/config.ts index 9bf08b5..b99266a 100644 --- a/src/config.ts +++ b/src/config.ts @@ -18,6 +18,26 @@ export interface NetraConfig { blockedSpans?: string[]; instruments?: Set; blockInstruments?: Set; + /** + * Set of instruments allowed to produce root-level spans. Independent of + * `instruments`. Defaults to `DEFAULT_INSTRUMENTS_FOR_ROOT` when omitted. + * + * When a root span comes from an instrumentation outside this set, only that + * span is dropped — its children are reparented onto its parent (a true + * root's children become the new roots). The peel then repeats: if a promoted + * child is itself from a non-root instrumentation it is dropped too, + * recursively, until a surviving span is reached (an allowed instrument, a + * Netra decorator / manual span, or any non-instrumentation span). This keeps + * LLM/vector spans that originate under an unwanted server root (e.g. + * Express/HTTP) without emitting the server span itself. A span whose parent + * link is remote (cross-process) is kept as a root so an upstream distributed + * trace is not severed. + * + * Note: when `enableRootSpan` is true, Netra attaches its own root span and + * every auto-instrumentation span becomes its child, so no reparenting + * occurs. Pass a set containing `NetraInstruments.ALL` to allow all + * instrumentations to produce root spans (legacy behaviour). + */ rootInstruments?: Set; } diff --git a/src/exporters/filteringSpanExporter.ts b/src/exporters/filteringSpanExporter.ts index 498b526..7addcc4 100644 --- a/src/exporters/filteringSpanExporter.ts +++ b/src/exporters/filteringSpanExporter.ts @@ -8,9 +8,11 @@ import { } from "../utils/pattern-matching"; import { addBlockedTraceId, + findRootSpansBlocked, getTraceId, isTraceIdBlocked, isTrialBlocked, + normalizeParent, } from "./utils"; /** @@ -49,6 +51,20 @@ class ReparentedSpan implements ReadableSpan { get droppedLinksCount() { return this.delegate.droppedLinksCount; } } +/** + * SpanExporter wrapper that drops spans by name and by root-instrument policy. + * + * A span is dropped when any of the following holds: + * - its trace id was blocked while a trial/quota block was active; + * - its name matches a globally configured block pattern; + * - its name matches a per-span local block pattern set by + * `LocalFilteringSpanProcessor`; + * - it is a root-connected span from an instrumentation not allowed to emit + * root spans (resolved by `findRootSpansBlocked`). + * + * Children of a dropped span are reparented onto the dropped span's parent so + * subtrees are never silently discarded. + */ export class FilteringSpanExporter implements SpanExporter { private static readonly MAX_REMEMBERED_ENTRIES = 10_000; @@ -76,89 +92,180 @@ export class FilteringSpanExporter implements SpanExporter { this.compiled = compilePatterns(globalPatterns); } + /** + * Filter `spans`, reparent survivors, and forward them to the wrapped exporter. + * + * While a trial/quota block is active the whole batch is dropped and its trace + * ids are remembered so their later spans are dropped too. Otherwise the batch + * is classified into name/local drops and root-instrument drops, the survivors + * are reparented past any dropped ancestor, and only they are forwarded. + */ export( spans: ReadableSpan[], resultCallback: (result: { code: ExportResultCode }) => void, ): void { if (isTrialBlocked()) { - for (const span of spans) { - const traceId = getTraceId(span); - if (traceId) addBlockedTraceId(traceId); - } + this.recordBlockedTraceIds(spans); resultCallback({ code: ExportResultCode.SUCCESS }); return; } - const filtered: ReadableSpan[] = []; - const batchBlockedMap = new Map(); - // Snapshot the shared processor map to avoid iteration/mutation races // when BatchSpanProcessor invokes export() concurrently with onStart(). const localBlockedSnapshot = this.localBlockedMap ? new Map(this.localBlockedMap) : undefined; + // Find unconditional name/local drops first, then resolve which root-block + // candidates are root-connected. Feeding the name/local drops into that walk + // is what stops a disallowed candidate from being promoted to a root when + // its only surviving ancestor is itself name-blocked. + const parentsOfSpansBlockedByName = this.findSpansBlockedByName(spans); + const { rootConnectedIds: rootSpansBlocked, parentsOfRootSpansBlocked } = + findRootSpansBlocked(spans, parentsOfSpansBlockedByName); + + const allBlockedSpanIds = new Set([ + ...parentsOfSpansBlockedByName.keys(), + ...rootSpansBlocked, + ]); + + const survivingSpans = this.collectSurvivors(spans, allBlockedSpanIds); + + const reparentMap = this.buildReparentMap( + parentsOfRootSpansBlocked, + parentsOfSpansBlockedByName, + localBlockedSnapshot, + ); + if (reparentMap.size > 0) { + this.reparentSpans(survivingSpans, reparentMap); + } + + this.evictRememberedIfNeeded(); + + if (survivingSpans.length === 0) { + resultCallback({ code: ExportResultCode.SUCCESS }); + return; + } + + this.exporter.export(survivingSpans, resultCallback); + } + + shutdown(): Promise { + this.rememberedBlockedParentMap.clear(); + this.localPatternCache.clear(); + return this.exporter.shutdown(); + } + + forceFlush(): Promise { + return this.exporter.forceFlush?.() ?? Promise.resolve(); + } + + /** + * Remember the trace ids seen during a block so their later spans are dropped + * too, even after the block expires. + */ + private recordBlockedTraceIds(spans: ReadableSpan[]): void { + for (const span of spans) { + const traceId = getTraceId(span); + if (traceId) addBlockedTraceId(traceId); + } + } + + /** + * Find the spans dropped unconditionally by a global or local name rule. + * + * Trace-blocked spans are skipped entirely (dropped without reparenting). Each + * dropped span is also recorded in the cross-batch remembered map so a child + * arriving in a later batch can still be reparented past it. The returned map + * feeds both the root-connected candidate walk — as transparent dropped + * ancestors — and the reparent map. + * + * @returns A `{droppedSpanId -> normalizedParent}` map for the name/locally + * blocked spans in this batch. + */ + private findSpansBlockedByName( + spans: ReadableSpan[], + ): Map { + const parentMap = new Map(); for (const span of spans) { const traceId = getTraceId(span); if (traceId && isTraceIdBlocked(traceId)) continue; const name = span.name; - const globallyBlocked = matchesAnyPattern(name, this.compiled); - - const localPatterns = this.getLocalPatterns(span); - const locallyBlocked = - (localPatterns.length > 0 && - matchesAnyPattern(name, this.getCompiledLocalPatterns(localPatterns))) || - this.hasLocalBlockFlag(span); - - if (!globallyBlocked && !locallyBlocked) { - filtered.push(span); - continue; - } + if (!globallyBlocked && !this.isLocallyBlocked(span, name)) continue; const spanId = span.spanContext().spanId; - const parentCtx = span.parentSpanContext; - batchBlockedMap.set(spanId, parentCtx); + const parentCtx = normalizeParent(span.parentSpanContext); + parentMap.set(spanId, parentCtx); this.rememberedBlockedParentMap.set(spanId, parentCtx); } + return parentMap; + } - const merged = new Map(); + /** + * Check whether `span` is blocked by its per-span local rules — either its + * name matches a local pattern or it carries the local-block flag. + */ + private isLocallyBlocked(span: ReadableSpan, name: string): boolean { + const localPatterns = this.readLocalBlockPatterns(span); + if ( + localPatterns.length > 0 && + matchesAnyPattern(name, this.getCompiledLocalPatterns(localPatterns)) + ) { + return true; + } + return this.hasLocalBlockFlag(span); + } + /** + * Return the spans that survive filtering: not trace-blocked and not in + * `allBlockedSpanIds` (name/local drops or the root-instrument policy). + */ + private collectSurvivors( + spans: ReadableSpan[], + allBlockedSpanIds: Set, + ): ReadableSpan[] { + const survivingSpans: ReadableSpan[] = []; + for (const span of spans) { + const traceId = getTraceId(span); + if (traceId && isTraceIdBlocked(traceId)) continue; + if (allBlockedSpanIds.has(span.spanContext().spanId)) continue; + survivingSpans.push(span); + } + return survivingSpans; + } + + /** + * Merge the cross-batch remembered map with this batch's dropped-parent maps. + * + * Ordering matters — later wins on conflict: the remembered map and the + * injected local-block snapshot are the base, overlaid by root-blocked + * parents, then this batch's name/local blocks, so in-batch drops win. + * + * @returns The merged `{droppedSpanId -> parent}` map used for reparenting. + */ + private buildReparentMap( + parentsOfRootSpansBlocked: Map, + parentsOfSpansBlockedByName: Map, + localBlockedSnapshot?: Map, + ): Map { + const reparentMap = new Map(); for (const [k, v] of this.rememberedBlockedParentMap) { - merged.set(k, v); + reparentMap.set(k, v); } if (localBlockedSnapshot) { for (const [k, v] of localBlockedSnapshot) { - merged.set(k, v); + reparentMap.set(k, v); } } - for (const [k, v] of batchBlockedMap) { - merged.set(k, v); + for (const [k, v] of parentsOfRootSpansBlocked) { + reparentMap.set(k, v); } - - if (merged.size > 0) { - this.reparentBlockedChildren(filtered, merged); + for (const [k, v] of parentsOfSpansBlockedByName) { + reparentMap.set(k, v); } - - this.evictRememberedIfNeeded(); - - if (filtered.length === 0) { - resultCallback({ code: ExportResultCode.SUCCESS }); - return; - } - - this.exporter.export(filtered, resultCallback); - } - - shutdown(): Promise { - this.rememberedBlockedParentMap.clear(); - this.localPatternCache.clear(); - return this.exporter.shutdown(); - } - - forceFlush(): Promise { - return this.exporter.forceFlush?.() ?? Promise.resolve(); + return reparentMap; } private getCompiledLocalPatterns(patterns: string[]): CompiledPatterns { @@ -190,7 +297,11 @@ export class FilteringSpanExporter implements SpanExporter { } } - private getLocalPatterns(span: ReadableSpan): string[] { + /** + * Read the per-span local-block patterns set by `LocalFilteringSpanProcessor`, + * or an empty list when none are set. + */ + private readLocalBlockPatterns(span: ReadableSpan): string[] { const value = span.attributes?.["netra.local_blocked_spans"]; if (Array.isArray(value) && value.every((v) => typeof v === "string")) { return (value as string[]).filter((v) => v !== ""); @@ -198,17 +309,20 @@ export class FilteringSpanExporter implements SpanExporter { return []; } + /** Check whether the processor explicitly flagged `span` as locally blocked. */ private hasLocalBlockFlag(span: ReadableSpan): boolean { return span.attributes?.["netra.local_blocked"] === true; } /** - * Walk up the blocked-parent chain for each surviving span and wrap it - * with a ReparentedSpan pointing to the nearest non-blocked ancestor. + * Reparent each span past any dropped ancestor onto its first surviving one. * - * A cycle guard (visited set) prevents infinite loops in malformed traces. + * Walks the chain of dropped parents (following `blockedMap`) until a surviving + * parent — or `undefined` (promote to root) — is reached, then wraps the span + * in a `ReparentedSpan` pointing there. A cycle guard (visited set) prevents + * infinite loops in malformed traces. */ - private reparentBlockedChildren( + private reparentSpans( spans: ReadableSpan[], blockedMap: Map, ): void { diff --git a/src/exporters/utils.ts b/src/exporters/utils.ts index 4d769f7..5965c24 100644 --- a/src/exporters/utils.ts +++ b/src/exporters/utils.ts @@ -1,10 +1,27 @@ import { ReadableSpan } from "@opentelemetry/sdk-trace-base"; +import type { SpanContext } from "@opentelemetry/api"; import { Config } from "../config"; import { Logger } from "../logger"; +import { + getRootBlockCandidatesFor, + isRootBlockCandidate, + rootBlockCandidatesContains, + INVALID_SPAN_ID, +} from "../processors/root-instrument-filter-processor"; + +export { INVALID_SPAN_ID }; let trialBlockedAt: number | null = null; const blockedTraceIds = new Set(); +/** + * Set the trial-blocked status, with automatic expiration. + * + * When called with `blocked=true`, starts a timer: all span exports are blocked + * for `Config.TRIAL_BLOCK_DURATION_SECONDS`. After that window, exports resume + * automatically even if this function is never called again. Calling with + * `blocked=false` clears the block immediately. + */ export function setTrialBlocked(blocked: boolean) { if (blocked) { if (!trialBlockedAt) { @@ -18,6 +35,13 @@ export function setTrialBlocked(blocked: boolean) { } } +/** + * Check whether the trial is currently blocked. + * + * Returns `false` automatically once `Config.TRIAL_BLOCK_DURATION_SECONDS` have + * elapsed since the block started, even if `setTrialBlocked(true)` was never + * called again. + */ export function isTrialBlocked(): boolean { if (trialBlockedAt === null) return false; @@ -31,10 +55,18 @@ export function isTrialBlocked(): boolean { return true; } +/** + * Add a trace ID to the blocked list. + * + * Trace IDs seen during a block are remembered so their later spans are dropped + * too — even after the block window expires. Only trace IDs created after the + * block clears will be exported again. + */ export function addBlockedTraceId(traceId: string) { blockedTraceIds.add(traceId); } +/** Check whether a trace ID is in the blocked list. */ export function isTraceIdBlocked(traceId: string): boolean { return blockedTraceIds.has(traceId); } @@ -46,3 +78,225 @@ export function getTraceId(span: ReadableSpan): string { return ""; } } + +/** Return a span's own span id, or `null` when unavailable. */ +export function getSpanId(span: ReadableSpan): string | null { + try { + const spanId = span.spanContext().spanId; + return typeof spanId === "string" && spanId ? spanId : null; + } catch { + return null; + } +} + +/** + * Normalize a span's parent link, collapsing an invalid/sentinel parent to + * `undefined` (= trace root). Mirrors the processor's `_getParentSpanContext` + * so the in-batch overlay and the registry store parents identically. + */ +export function normalizeParent( + parent: SpanContext | undefined, +): SpanContext | undefined { + if (!parent) { + return undefined; + } + if (!parent.spanId || parent.spanId === INVALID_SPAN_ID) { + return undefined; + } + return parent; +} + +/** + * Find the root-block candidates in `spans` that must be dropped. + * + * A candidate is dropped when it is *root-connected*: it is a trace root (no + * local parent), or every ancestor between it and the trace root is itself + * dropped. Walking up the ancestor chain stops at the first surviving ancestor + * — an allowed instrument, a netra/manual span, or a non-instrumentation span — + * and never crosses a remote (cross-process) parent link. + * + * `parentsOfSpansBlockedByName` are spans dropped for *other* reasons (a global + * or per-span local name block) that are removed from the exported tree just + * like candidates. Folding them in lets the walk "see through" them: a candidate + * whose only surviving ancestor was a name-blocked span becomes root-connected + * once that span is dropped, and so must be dropped too rather than promoted to + * a root it is not allowed to be. + * + * The candidate set is the registry snapshot overlaid with any span in the + * current batch that carries the durable candidacy marker. The overlay makes + * the decision robust: a blocked root still in the batch is dropped even if its + * registry entry was evicted (TTL/overflow) or cleared, because the marker + * travels with the span. Only ancestor chains reachable from this batch are + * evaluated, so the cost is proportional to the batch (plus its ancestry) + * rather than to the whole registry. + * + * @returns `{ rootConnectedIds, parentsOfRootSpansBlocked }` — the dropped span + * ids and a map of each dropped span id to its parent `SpanContext` + * (`undefined` for a true root) for reparenting. + */ +export function findRootSpansBlocked( + spans: ReadableSpan[], + parentsOfSpansBlockedByName?: Map, +): { + rootConnectedIds: Set; + parentsOfRootSpansBlocked: Map; +} { + const candidates = collectRootBlockCandidates(spans); + // Only genuine root-block candidates trigger a chain walk; name/local drops + // alone (with no candidate in play) are handled by the caller directly. + if (candidates.size === 0) { + return { rootConnectedIds: new Set(), parentsOfRootSpansBlocked: new Map() }; + } + + if (parentsOfSpansBlockedByName) { + for (const [spanId, parentCtx] of parentsOfSpansBlockedByName) { + if (!candidates.has(spanId)) { + candidates.set(spanId, parentCtx); + } + } + } + + const rootConnectedIds = findRootConnectedCandidates(spans, candidates); + const parentsOfRootSpansBlocked = new Map(); + for (const spanId of rootConnectedIds) { + parentsOfRootSpansBlocked.set(spanId, candidates.get(spanId)); + } + return { rootConnectedIds, parentsOfRootSpansBlocked }; +} + +/** + * Merge in-batch candidacy markers with the cross-batch registry. + * + * The in-batch markers let a marked span be recognised even if its registry + * entry was evicted or cleared before export. The registry is only consulted + * for *cross-batch* ancestry — a batch parent whose candidacy was recorded in + * an earlier batch. When no batch parent references such an entry, the full + * registry snapshot is skipped entirely, and otherwise only the reachable + * ancestry is walked. + * + * @returns A `{spanId -> parentSpanContext}` map of every relevant candidate, + * with in-batch markers overriding the registry snapshot. + */ +function collectRootBlockCandidates( + spans: ReadableSpan[], +): Map { + const inBatch = new Map(); + const parentIds = new Set(); + + for (const span of spans) { + const spanId = getSpanId(span); + if (isRootBlockCandidate(span) && spanId !== null && spanId !== INVALID_SPAN_ID) { + inBatch.set(spanId, normalizeParent(span.parentSpanContext)); + } + const parent = span.parentSpanContext; + const parentId = parent?.spanId; + if (parentId && parentId !== INVALID_SPAN_ID) { + parentIds.add(parentId); + } + } + + // A registry entry is only walked if it is the parent of some batch span + // (every in-batch candidate contributes its own parent id here too, covering + // multi-hop cross-batch chains). Parents resolvable in-batch never need it. + const unresolvedParents = new Set(); + for (const parentId of parentIds) { + if (!inBatch.has(parentId)) { + unresolvedParents.add(parentId); + } + } + if (!rootBlockCandidatesContains(unresolvedParents)) { + return inBatch; + } + + // Walk only the ancestry reachable from the batch's unresolved parents rather + // than copying the whole registry. + const candidates = getRootBlockCandidatesFor(unresolvedParents); + for (const [spanId, parentCtx] of inBatch) { + candidates.set(spanId, parentCtx); + } + return candidates; +} + +/** + * Return the candidate span ids that are root-connected (and so dropped). + * + * Walks each candidate's ancestry chain iteratively and memoized — no recursion, + * so a deeply nested trace cannot blow the stack. Every node on a linear + * candidate chain shares the terminal verdict (the chain is dropped iff it walks + * all the way up to a true root), so the resolved result is written back to the + * whole walked path in one pass. The walk is seeded only from batch spans and + * their parents, so unrelated registry entries are never walked. + */ +function findRootConnectedCandidates( + spans: ReadableSpan[], + candidates: Map, +): Set { + const memo = new Map(); + const rootConnectedIds = new Set(); + + const resolvesToRoot = (startId: string): boolean => { + const walkedIds: string[] = []; + const seenIds = new Set(); + let currentId = startId; + let reachesRoot = false; + + while (true) { + const memoized = memo.get(currentId); + if (memoized !== undefined) { + reachesRoot = memoized; + break; + } + if (!candidates.has(currentId)) { + // Not a candidate -> a surviving ancestor. Stops the walk. + reachesRoot = false; + break; + } + if (seenIds.has(currentId)) { + // Cycle guard: treat as non-dropped to avoid looping forever. + reachesRoot = false; + break; + } + + // Fresh candidate node: it shares the chain's terminal verdict. + walkedIds.push(currentId); + seenIds.add(currentId); + + const parentCtx = candidates.get(currentId); + if (!parentCtx) { + reachesRoot = true; // candidate is a true trace root + break; + } + if (parentCtx.isRemote) { + reachesRoot = false; // do not walk across a process boundary + break; + } + const parentId = parentCtx.spanId; + if (!parentId || parentId === INVALID_SPAN_ID) { + reachesRoot = true; + break; + } + currentId = parentId; + } + + for (const spanId of walkedIds) { + memo.set(spanId, reachesRoot); + if (reachesRoot) { + rootConnectedIds.add(spanId); + } + } + return reachesRoot; + }; + + for (const span of spans) { + const spanId = getSpanId(span); + if (spanId !== null && spanId !== INVALID_SPAN_ID) { + resolvesToRoot(spanId); + } + const parentId = span.parentSpanContext?.spanId; + if (parentId && parentId !== INVALID_SPAN_ID) { + resolvesToRoot(parentId); + } + } + + return rootConnectedIds; +} diff --git a/src/processors/root-instrument-filter-processor.ts b/src/processors/root-instrument-filter-processor.ts index a059f55..1233d24 100644 --- a/src/processors/root-instrument-filter-processor.ts +++ b/src/processors/root-instrument-filter-processor.ts @@ -1,19 +1,35 @@ /** * Root Instrument Filter Processor * - * Blocks root spans (and their entire subtree) from instrumentations not in - * the allowed root_instruments set. Tracking is trace-ID-based: when a root - * span is blocked, its trace_id is recorded and all subsequent spans sharing - * that trace_id are blocked as well. + * Records spans from auto-instrumentation libraries that are not permitted to + * produce root-level spans, so the exporter can *drop-and-reparent* them rather + * than discarding a whole subtree. * - * This guarantees correct propagation even in async frameworks where a parent - * span may end before all children have started. + * When an auto-instrumentation span (e.g. Express, HTTP, FastAPI-equivalent) + * comes from a library outside the allowed root-instrument set, this processor + * marks it as a **root-block candidate** with a durable instance-level marker + * (`ROOT_BLOCK_CANDIDATE_FIELD`) and records it — along with its parent + * `SpanContext` — in a module-global, TTL-evicted registry used for cross-batch + * reparenting. + * + * The actual drop decision is made at export time by `FilteringSpanExporter`: + * a candidate is dropped only when it is *root-connected* — it is a trace root, + * or every ancestor up to the trace root is also a dropped candidate. Dropped + * candidates have their children reparented onto the dropped span's parent + * (`undefined` for a true root, so the child becomes the new root). This peel + * repeats recursively until a survivor is reached (an allowed instrument, a + * Netra decorator / manual span, or any non-instrumentation span). + * + * Spans created directly through Netra decorators or `Netra.startSpan` are + * never candidates — only spans from recognised auto-instrumentation libraries + * are subject to the allow-list. */ -import { Context, Span, trace } from "@opentelemetry/api"; +import { Context, Span } from "@opentelemetry/api"; +import type { SpanContext } from "@opentelemetry/api"; import { ReadableSpan, SpanProcessor } from "@opentelemetry/sdk-trace-base"; -const LOCAL_BLOCKED_ATTR = "netra.local_blocked"; +export const INVALID_SPAN_ID = "0000000000000000"; const INSTRUMENTATION_PREFIXES = [ "opentelemetry.instrumentation.", @@ -22,24 +38,143 @@ const INSTRUMENTATION_PREFIXES = [ "@traceloop/instrumentation-", ]; -const MAX_BLOCKED_TRACES = 4096; -const BLOCKED_TRACE_TTL_MS = 600_000; // 600 seconds +const MAX_ROOT_CANDIDATES = 4096; +const ROOT_CANDIDATE_TTL_MS = 600_000; // 600 seconds +/** + * Minimum spacing between TTL sweeps. `onEnd` fires for every span, so the + * (unbounded-order) full sweep is time-throttled to keep the hot path cheap; + * the hard `MAX_ROOT_CANDIDATES` overflow bound still applies between sweeps. + */ +const EVICTION_INTERVAL_MS = 10_000; + +/** + * Canonicalize an instrumentation name for allow-list comparison. Instrument + * scope names arrive concatenated (e.g. `@traceloop/instrumentation-llamaindex` + * → `"llamaindex"`, `...-mistralai` → `"mistralai"`) while the + * `NetraInstruments` values use snake_case (`"llama_index"`, `"mistral_ai"`). + * The only discrepancy is underscore placement, so lower-casing and stripping + * underscores on both sides makes the two representations match without a + * hand-maintained alias table. + */ +export function canonicalInstrumentName(name: string): string { + return name.toLowerCase().replace(/_/g, ""); +} + +/** + * Durable per-span marker set the moment a span is classified as a root-block + * candidate. It is a plain *instance property* — deliberately NOT an OTel span + * attribute — so it travels with the span into its export batch without + * consuming the span's bounded attribute capacity and without being evicted by + * `AttributeSizeLimitProcessor`. Being off the attribute map, it is never + * serialised to the backend. The exporter can still recognise a blocked root + * even if the registry entry was evicted (TTL/overflow) or cleared (shutdown) + * between `onStart` and export. + */ +export const ROOT_BLOCK_CANDIDATE_FIELD = "_netraRootBlockCandidate"; + +/** + * `endedAt` is `null` while the span is still active; the TTL clock starts once + * the span ends. A long-lived active candidate is therefore never evicted while + * still open. + */ +interface CandidateEntry { + parentCtx: SpanContext | undefined; + endedAt: number | null; +} + +/** + * Module-global registry of root-block candidates: `spanId -> CandidateEntry`. + * + * Shared between this processor (writer, on span start/end) and the exporter's + * peel (reader, on export). It enables *cross-batch* ancestry resolution — + * reparenting a child that exports in a later batch than its dropped ancestor. + * It is a supplement, not the source of truth: candidacy is carried durably on + * the span via `ROOT_BLOCK_CANDIDATE_FIELD`. + * + * JS is single-threaded — `onStart` and `export()` never run concurrently — so + * no locking is required (unlike the Python equivalent). + */ +const ROOT_BLOCK_CANDIDATES = new Map(); + +/** + * Resolve only the registry ancestry reachable from `seedIds` — the cross-batch + * parent ids (and their parents, transitively) referenced by the current export + * batch. Returns a fresh `{ spanId -> parentCtx }` map, decoupled from later + * writes. Walking only the reachable subset avoids copying the entire registry + * (up to `MAX_ROOT_CANDIDATES` entries) on every export. + */ +export function getRootBlockCandidatesFor( + seedIds: Set, +): Map { + const out = new Map(); + const stack = [...seedIds]; + while (stack.length > 0) { + const spanId = stack.pop(); + if (spanId === undefined || out.has(spanId)) { + continue; + } + const entry = ROOT_BLOCK_CANDIDATES.get(spanId); + if (!entry) { + continue; + } + out.set(spanId, entry.parentCtx); + const parentId = entry.parentCtx?.spanId; + if (parentId && parentId !== INVALID_SPAN_ID) { + stack.push(parentId); + } + } + return out; +} + +/** + * Clear the module-global candidate registry. Called when a new processor is + * constructed so a fresh Netra session (or test) does not inherit stale + * cross-batch state. Deliberately NOT called on shutdown — see `shutdown()`. + */ +export function resetRootBlockCandidates(): void { + ROOT_BLOCK_CANDIDATES.clear(); +} + +/** + * Cheap membership probe so the exporter can skip the full snapshot when a + * batch references no cross-batch candidate ancestor. + */ +export function rootBlockCandidatesContains(spanIds: Set): boolean { + if (spanIds.size === 0) { + return false; + } + for (const spanId of spanIds) { + if (ROOT_BLOCK_CANDIDATES.has(spanId)) { + return true; + } + } + return false; +} -interface BlockedEntry { - timestamp: number; +/** + * Whether a span carries the durable root-block-candidate marker. Robust + * against registry eviction/clear because the marker lives on the span object. + */ +export function isRootBlockCandidate(span: unknown): boolean { + return Boolean((span as Record | null | undefined)?.[ROOT_BLOCK_CANDIDATE_FIELD]); } export class RootInstrumentFilterProcessor implements SpanProcessor { private readonly _allowed: Set; - private readonly _blockedTraceIds: Map = new Map(); + private _lastEvictionAt = 0; constructor(allowedRootInstrumentNames: Set) { - this._allowed = new Set(allowedRootInstrumentNames); + this._allowed = new Set( + [...allowedRootInstrumentNames].map(canonicalInstrumentName), + ); + // A new processor marks the start of a fresh session; drop any registry + // state left behind by a previous init (or a prior test). + resetRootBlockCandidates(); } - onStart(span: Span, parentContext: Context): void { + onStart(span: Span, _parentContext: Context): void { try { - this._processSpanStart(span, parentContext); + this._processSpanStart(span); } catch { // Best-effort; never break span pipeline } @@ -47,14 +182,26 @@ export class RootInstrumentFilterProcessor implements SpanProcessor { onEnd(span: ReadableSpan): void { try { - this._evictStaleTraces(); + const now = Date.now(); + this._markEnded(span, now); + this._evictStaleCandidates(now); } catch { // no-op } } + /** + * No-op — the candidate registry is deliberately NOT cleared here. + * + * `MultiSpanProcessor.shutdown()` invokes every registered processor's + * `shutdown()` concurrently (`Promise.all`), so clearing the shared registry + * here could race with the exporter's final flush still reading it to + * reparent buffered blocked roots. Fresh sessions instead reset the registry + * on construction (see the constructor); it is also bounded + * (`MAX_ROOT_CANDIDATES`) and TTL-evicted, so leaving it intact on shutdown + * is safe. + */ shutdown(): Promise { - this._blockedTraceIds.clear(); return Promise.resolve(); } @@ -62,86 +209,116 @@ export class RootInstrumentFilterProcessor implements SpanProcessor { return Promise.resolve(); } - private _processSpanStart(span: Span, parentContext: Context): void { - const parentSpanId = this._resolveParentSpanId(parentContext) - ?? this._getParentSpanIdFromSpan(span); - - const isChild = parentSpanId !== null && parentSpanId !== "0000000000000000"; - - if (isChild) { - this._maybeBlockChild(span); - } else { - this._maybeBlockRoot(span); + private _processSpanStart(span: Span): void { + if (!this._isFromInstrumentationLibrary(span)) { + return; } - } - private _maybeBlockChild(span: Span): void { - const traceId = this._getTraceId(span); - if (!traceId) { + const instrName = this._extractInstrumentationName(span); + if (instrName === null || this._allowed.has(canonicalInstrumentName(instrName))) { return; } - if (this._blockedTraceIds.has(traceId)) { - this._markBlocked(span); - } - } - private _maybeBlockRoot(span: Span): void { - if (!this._isFromInstrumentationLibrary(span)) { + const spanId = this._getOwnSpanId(span); + if (spanId === null || spanId === INVALID_SPAN_ID) { return; } - const instrName = this._extractInstrumentationName(span); - if (instrName === null || this._allowed.has(instrName)) { - return; + const parentCtx = this._getParentSpanContext(span); + // Mark durably first: the marker must survive even if the registry entry is + // later evicted/cleared before this span reaches the exporter. + this._markCandidate(span); + this._recordCandidate(spanId, parentCtx); + } + + private _markCandidate(span: Span): void { + try { + (span as unknown as Record)[ROOT_BLOCK_CANDIDATE_FIELD] = true; + } catch { + // no-op } + } - const traceId = this._getTraceId(span); - if (traceId) { - this._blockedTraceIds.set(traceId, { timestamp: Date.now() }); - this._evictOverflow(); + /** + * Register `spanId` as an active candidate. `move-to-end` semantics + * (delete+set) refresh insertion order; overflow drops the oldest entry. + */ + private _recordCandidate(spanId: string, parentCtx: SpanContext | undefined): void { + ROOT_BLOCK_CANDIDATES.delete(spanId); + ROOT_BLOCK_CANDIDATES.set(spanId, { parentCtx, endedAt: null }); + while (ROOT_BLOCK_CANDIDATES.size > MAX_ROOT_CANDIDATES) { + const oldest = ROOT_BLOCK_CANDIDATES.keys().next().value; + if (oldest === undefined) { + break; + } + ROOT_BLOCK_CANDIDATES.delete(oldest); } - this._markBlocked(span); } - private _resolveParentSpanId(parentContext: Context): string | null { - const parentSpan = trace.getSpan(parentContext); - if (!parentSpan) { - return null; + /** Start the TTL clock for a just-ended candidate, if it has an entry. */ + private _markEnded(span: ReadableSpan, now: number): void { + const spanId = this._getOwnSpanId(span); + if (spanId === null) { + return; } - const sc = parentSpan.spanContext(); - if (!sc || sc.spanId === "0000000000000000") { - return null; + const entry = ROOT_BLOCK_CANDIDATES.get(spanId); + if (entry) { + entry.endedAt = now; } - return sc.spanId; } - private _getParentSpanIdFromSpan(span: Span): string | null { - const spanAny = span as any; - const parent = spanAny.parentSpanId ?? spanAny.parent?.spanId ?? null; - if (!parent || parent === "0000000000000000") { - return null; + /** + * Evict entries whose span ended more than `ROOT_CANDIDATE_TTL_MS` ago. + * Active candidates (`endedAt === null`) are skipped — never evicted while the + * span is still open. + * + * The scan cannot break early: the registry is ordered by span *start* + * (insertion) whereas staleness is measured from span *end*, and those two + * orders do not coincide (a long-lived active root sits at the head while + * short spans behind it may already be stale). A full sweep is therefore + * required, but it is time-throttled (`EVICTION_INTERVAL_MS`) so it does not + * run on every span end. + */ + private _evictStaleCandidates(now: number): void { + if (now - this._lastEvictionAt < EVICTION_INTERVAL_MS) { + return; + } + this._lastEvictionAt = now; + const cutoff = now - ROOT_CANDIDATE_TTL_MS; + for (const [spanId, entry] of ROOT_BLOCK_CANDIDATES) { + if (entry.endedAt !== null && entry.endedAt <= cutoff) { + ROOT_BLOCK_CANDIDATES.delete(spanId); + } } - return parent; } - private _getTraceId(span: Span | ReadableSpan): string | null { + private _getOwnSpanId(span: Span | ReadableSpan): string | null { try { - const ctx = (span as any).spanContext?.() ?? (span as any).context; - if (ctx) { - return ctx.traceId ?? null; - } + const sc = (span as Span).spanContext?.() ?? (span as unknown as { spanContext?: SpanContext }).spanContext; + const spanId = (sc as SpanContext | undefined)?.spanId; + return typeof spanId === "string" && spanId ? spanId : null; } catch { - // no-op + return null; } - return null; } - private _markBlocked(span: Span): void { - try { - span.setAttribute(LOCAL_BLOCKED_ATTR, true); - } catch { - // no-op + /** + * Resolve the parent `SpanContext` recorded on the span, normalizing an + * invalid/sentinel parent to `undefined` (= trace root). + */ + private _getParentSpanContext(span: Span): SpanContext | undefined { + const spanAny = span as unknown as { + parentSpanContext?: SpanContext; + parent?: SpanContext; + }; + const parent = spanAny.parentSpanContext ?? spanAny.parent; + if (!parent) { + return undefined; } + if (!parent.spanId || parent.spanId === INVALID_SPAN_ID) { + return undefined; + } + return parent; } private _isFromInstrumentationLibrary(span: Span): boolean { @@ -180,25 +357,4 @@ export class RootInstrumentFilterProcessor implements SpanProcessor { } return name; } - - private _evictStaleTraces(): void { - const cutoff = Date.now() - BLOCKED_TRACE_TTL_MS; - for (const [traceId, entry] of this._blockedTraceIds) { - if (entry.timestamp > cutoff) { - break; - } - this._blockedTraceIds.delete(traceId); - } - } - - private _evictOverflow(): void { - while (this._blockedTraceIds.size > MAX_BLOCKED_TRACES) { - const first = this._blockedTraceIds.keys().next().value; - if (first !== undefined) { - this._blockedTraceIds.delete(first); - } else { - break; - } - } - } } From ec74a4a6cd4e7c7268b4de7eb40d21aa05d26ba4 Mon Sep 17 00:00:00 2001 From: Nithish-KV Date: Fri, 17 Jul 2026 10:15:53 +0530 Subject: [PATCH 3/4] [NET-1333] fix: Standardize span attribute serialization and fix addConversation persistence (#114) --- package-lock.json | 27 +-- package.json | 16 +- src/config.ts | 25 ++- src/decorators.ts | 6 +- src/index.ts | 4 +- src/instrumentation/anthropic/wrappers.ts | 9 +- src/instrumentation/google-genai/utils.ts | 13 +- .../google-generative-ai/utils.ts | 5 +- src/instrumentation/index.ts | 90 ++++---- .../openai-agents/attributes.ts | 21 +- .../openai-agents/constants.ts | 1 - .../openai-agents/processor.ts | 8 +- src/instrumentation/openai-agents/utils.ts | 19 +- src/instrumentation/utils.ts | 9 +- .../attribute-size-limit-processor.ts | 75 ------- src/processors/index.ts | 2 +- .../instrumentation-span-processor.ts | 93 +------- src/processors/scrubbing-span-processor.ts | 210 ++++++++---------- .../serialization-span-processor.ts | 70 ++++++ src/session-manager.ts | 59 +++-- src/utils/serialization.ts | 84 ------- src/utils/serialization/index.ts | 10 + src/utils/serialization/safe-stringify.ts | 67 ++++++ .../serialization/truncating-serializer.ts | 209 +++++++++++++++++ src/version.ts | 2 +- 25 files changed, 608 insertions(+), 526 deletions(-) delete mode 100644 src/processors/attribute-size-limit-processor.ts create mode 100644 src/processors/serialization-span-processor.ts delete mode 100644 src/utils/serialization.ts create mode 100644 src/utils/serialization/index.ts create mode 100644 src/utils/serialization/safe-stringify.ts create mode 100644 src/utils/serialization/truncating-serializer.ts diff --git a/package-lock.json b/package-lock.json index e5d293f..f847de1 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "netra-sdk", - "version": "1.5.0", + "version": "1.6.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "netra-sdk", - "version": "1.5.0", + "version": "1.6.0", "license": "Apache-2.0", "dependencies": { "@opentelemetry/api": "^1.9.0", @@ -19,7 +19,7 @@ "@opentelemetry/sdk-trace-base": "^2.0.1", "@traceloop/node-server-sdk": "^0.22.5", "axios": "^1.13.4", - "dotenv": "^17.2.3", + "jsonrepair": "^3.15.0", "p-limit": "^7.2.0", "shimmer": "^1.2.1" }, @@ -4031,18 +4031,6 @@ "node": ">=0.3.1" } }, - "node_modules/dotenv": { - "version": "17.2.3", - "resolved": "https://registry.npmjs.org/dotenv/-/dotenv-17.2.3.tgz", - "integrity": "sha512-JVUnt+DUIzu87TABbhPmNfVdBDt18BLOWjMUFJMSi/Qqg7NTYtabbvSNJGOJ7afbRuv9D/lngizHtP7QyLQ+9w==", - "license": "BSD-2-Clause", - "engines": { - "node": ">=12" - }, - "funding": { - "url": "https://dotenvx.com" - } - }, "node_modules/dunder-proto": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/dunder-proto/-/dunder-proto-1.0.1.tgz", @@ -4731,6 +4719,15 @@ "bignumber.js": "^9.0.0" } }, + "node_modules/jsonrepair": { + "version": "3.15.0", + "resolved": "https://registry.npmjs.org/jsonrepair/-/jsonrepair-3.15.0.tgz", + "integrity": "sha512-wy8OTjwsJwQRnQJkKnMJJ9vcytRdBPAgIF/Hy6+s1dAj42BHMKiyL8JzEieIl3JY7idt8eyHwBWTO8mh/+mtwA==", + "license": "ISC", + "bin": { + "jsonrepair": "bin/cli.js" + } + }, "node_modules/jwa": { "version": "2.0.1", "resolved": "https://registry.npmjs.org/jwa/-/jwa-2.0.1.tgz", diff --git a/package.json b/package.json index c263922..0e737d3 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "netra-sdk", - "version": "1.5.0", + "version": "1.6.0", "description": "A comprehensive TypeScript/JavaScript SDK for AI application observability built on top of OpenTelemetry and Traceloop", "type": "module", "main": "./dist/index.cjs", @@ -50,33 +50,33 @@ "dependencies": { "@opentelemetry/api": "^1.9.0", "@opentelemetry/core": "^2.0.1", - "@opentelemetry/instrumentation-undici": "^0.6.0", "@opentelemetry/instrumentation-express": "^0.67.0", "@opentelemetry/instrumentation-http": "^0.219.0", + "@opentelemetry/instrumentation-undici": "^0.6.0", "@opentelemetry/otlp-transformer": "^0.211.0", "@opentelemetry/resources": "^2.0.1", "@opentelemetry/sdk-trace-base": "^2.0.1", "@traceloop/node-server-sdk": "^0.22.5", "axios": "^1.13.4", - "dotenv": "^17.2.3", + "jsonrepair": "^3.15.0", "p-limit": "^7.2.0", "shimmer": "^1.2.1" }, "peerDependencies": { + "@anthropic-ai/sdk": ">=0.60.0", + "@google/genai": ">=1.0.0", + "@google/generative-ai": "^0.24.1", "@langchain/langgraph": "^1.1.1", "@langchain/ollama": "^1.2.1", "@mistralai/mistralai": ">=1.0.0", - "@google/generative-ai": "^0.24.1", - "@google/genai": ">=1.0.0", + "@openai/agents": ">=0.10.0 <1.0.0", "@opentelemetry/instrumentation": ">=0.53.0", "@opentelemetry/instrumentation-express": "^0.67.0", "@opentelemetry/instrumentation-http": "^0.219.0", "@opentelemetry/instrumentation-undici": "^0.6.0", "@prisma/instrumentation": "^5.0.0", "groq-sdk": ">=0.3.0", - "openai": "^4.0.0 || ^5.0.0 || ^6.0.0", - "@anthropic-ai/sdk": ">=0.60.0", - "@openai/agents": ">=0.10.0 <1.0.0" + "openai": "^4.0.0 || ^5.0.0 || ^6.0.0" }, "peerDependenciesMeta": { "@opentelemetry/instrumentation": { diff --git a/src/config.ts b/src/config.ts index b99266a..f297a7e 100644 --- a/src/config.ts +++ b/src/config.ts @@ -126,15 +126,22 @@ export class Config { static readonly LIBRARY_NAME = "netra"; static readonly LIBRARY_VERSION = SDK_VERSION; static readonly TRIAL_BLOCK_DURATION_SECONDS = 900; // 15 minutes - static readonly ATTRIBUTE_MAX_LEN = parseInt( - process.env.NETRA_ATTRIBUTE_MAX_LEN || "50000", - ); - static readonly CONVERSATION_MAX_LEN = parseInt( - process.env.NETRA_CONVERSATION_CONTENT_MAX_LEN || "50000", - ); - static readonly SPAN_ATTRIBUTE_MAX_SIZE = parseInt( - process.env.NETRA_SPAN_ATTRIBUTE_MAX_SIZE || "30000", - ); + + private static _spanAttributeMaxSize: number | undefined; + + /** + * Lazily reads NETRA_SPAN_ATTRIBUTE_MAX_SIZE from process.env on first + * access so that env vars set before Netra.init() (via dotenv, shell + * export, Docker, etc.) are always respected regardless of import order. + */ + static get SPAN_ATTRIBUTE_MAX_SIZE(): number { + if (Config._spanAttributeMaxSize === undefined) { + Config._spanAttributeMaxSize = parseInt( + process.env.NETRA_SPAN_ATTRIBUTE_MAX_SIZE || "30000", + ); + } + return Config._spanAttributeMaxSize; + } appName: string; otlpEndpoint?: string; diff --git a/src/decorators.ts b/src/decorators.ts index 193c5dd..5d8f4fb 100644 --- a/src/decorators.ts +++ b/src/decorators.ts @@ -8,7 +8,7 @@ import { Logger } from "./logger"; import { SessionManager } from "./session-manager"; import { SpanType, DecoratorOptions } from "./types"; import { wrapResponse } from "./utils/response-handler"; -import { safeStringify, serializeValue } from "./utils/serialization"; +import { safeStringify } from "./utils/serialization"; type AnyFunction = (...args: any[]) => any; type AsyncFunction = (...args: any[]) => Promise; @@ -41,7 +41,7 @@ function spanHasOutput(span: Span): boolean { function addInputAttributes(span: Span, args: any[], entityType: string): void { span.setAttribute(`${Config.LIBRARY_NAME}.entity.type`, entityType); if (args.length > 0) { - span.setAttribute("input", safeStringify(args, Config.ATTRIBUTE_MAX_LEN)); + span.setAttribute("input", safeStringify(args)); } } @@ -49,7 +49,7 @@ function addOutputAttributes(span: Span, result: any): void { // Skip if the user already set output explicitly inside the decorated function if (result === undefined || spanHasOutput(span)) return; try { - span.setAttribute("output", serializeValue(result, Config.ATTRIBUTE_MAX_LEN)); + span.setAttribute("output", safeStringify(result)); } catch (e) { span.setAttribute("output_error", String(e)); } diff --git a/src/index.ts b/src/index.ts index 1cd80b1..4d768d2 100644 --- a/src/index.ts +++ b/src/index.ts @@ -26,12 +26,14 @@ export { } from "./config"; export { agent, span, task, workflow } from "./decorators"; export { - AttributeSizeLimitProcessor, InstrumentationSpanProcessor, ScrubbingSpanProcessor, + SerializationSpanProcessor, SessionSpanProcessor, SpanIOProcessor, } from "./processors"; +export { safeStringify, TruncatingSerializer } from "./utils/serialization"; +export type { SafeStringifyOptions } from "./utils/serialization"; export { ConversationType } from "./session-manager"; export { SpanType } from "./types"; export type { ActionModel, UsageModel, SpanOptions } from "./types"; diff --git a/src/instrumentation/anthropic/wrappers.ts b/src/instrumentation/anthropic/wrappers.ts index 382e210..73267b0 100644 --- a/src/instrumentation/anthropic/wrappers.ts +++ b/src/instrumentation/anthropic/wrappers.ts @@ -63,8 +63,6 @@ const TRACKED_STREAM_EVENTS = new Set([ "finalMessage", ]); -const MAX_TOOL_ATTR_LENGTH = 4096; - /** * Wrap each tool's `run()` method so every invocation produces a TOOL span. * Preserves the tool's prototype chain — only `run` is replaced. @@ -94,7 +92,7 @@ export function wrapRunnableTools( }; if (toolUseId) attrs[SpanAttributes.LLM_REQUEST_TOOL_ID] = toolUseId; if (traceContent) { - attrs.input = safeStringify(input, MAX_TOOL_ATTR_LENGTH); + attrs.input = safeStringify(input); } const span = tracer.startSpan( @@ -109,10 +107,7 @@ export function wrapRunnableTools( originalRun.call(this, input, runContext), ); if (traceContent) { - span.setAttribute( - "output", - safeStringify(result, MAX_TOOL_ATTR_LENGTH), - ); + span.setAttribute("output", safeStringify(result)); } span.setStatus({ code: SpanStatusCode.OK }); return result; diff --git a/src/instrumentation/google-genai/utils.ts b/src/instrumentation/google-genai/utils.ts index 6b43e29..e2308ab 100644 --- a/src/instrumentation/google-genai/utils.ts +++ b/src/instrumentation/google-genai/utils.ts @@ -12,7 +12,6 @@ */ import { Span } from "@opentelemetry/api"; -import { Config } from "../../config"; import { Logger } from "../../logger"; import { safeStringify } from "../../utils/serialization"; import { SpanAttributes } from "../span-attributes"; @@ -146,7 +145,7 @@ function buildGoogleInputMessages( const sysContent = typeof cfg.systemInstruction === "string" ? cfg.systemInstruction - : safeStringify(cfg.systemInstruction, Config.CONVERSATION_MAX_LEN); + : safeStringify(cfg.systemInstruction); messages.push({ role: "system", content: sysContent }); } @@ -186,7 +185,7 @@ function buildGoogleInputMessages( } else { messages.push({ role: "user", - content: safeStringify(item, Config.CONVERSATION_MAX_LEN), + content: safeStringify(item), }); } } @@ -201,7 +200,7 @@ function buildGoogleInputMessages( } else { messages.push({ role: "user", - content: safeStringify(contents, Config.CONVERSATION_MAX_LEN), + content: safeStringify(contents), }); } @@ -220,10 +219,10 @@ function extractTextFromParts(parts: unknown[]): string { if (typeof part.text === "string") { texts.push(part.text); } else if (part.functionCall) { - texts.push(safeStringify(part.functionCall, Config.CONVERSATION_MAX_LEN)); + texts.push(safeStringify(part.functionCall)); } else if (part.functionResponse) { texts.push( - safeStringify(part.functionResponse, Config.CONVERSATION_MAX_LEN), + safeStringify(part.functionResponse), ); } } @@ -402,7 +401,6 @@ function buildGoogleOutputMessages( role: "tool", content: safeStringify( { name: fc.name, arguments: fc.args }, - Config.CONVERSATION_MAX_LEN, ), }); } @@ -429,7 +427,6 @@ function buildGoogleOutputMessages( role: "tool", content: safeStringify( { name: fc.name, arguments: fc.args }, - Config.CONVERSATION_MAX_LEN, ), }); } diff --git a/src/instrumentation/google-generative-ai/utils.ts b/src/instrumentation/google-generative-ai/utils.ts index c17a291..679ce13 100644 --- a/src/instrumentation/google-generative-ai/utils.ts +++ b/src/instrumentation/google-generative-ai/utils.ts @@ -311,10 +311,7 @@ function _setPromptAttributes( } if (inlineDataParts.length > 0) { - span.setAttribute( - `gen_ai.prompt.${slotIndex}.has_inline_data`, - true, - ); + span.setAttribute(`gen_ai.prompt.${slotIndex}.has_inline_data`, true); span.setAttribute( `gen_ai.prompt.${slotIndex}.inline_data_count`, inlineDataParts.length, diff --git a/src/instrumentation/index.ts b/src/instrumentation/index.ts index c2d0089..3142785 100644 --- a/src/instrumentation/index.ts +++ b/src/instrumentation/index.ts @@ -18,12 +18,12 @@ import { } from "../config"; import { Logger } from "../logger"; import { - AttributeSizeLimitProcessor, InstrumentationSpanProcessor, LlmTraceIdentifierSpanProcessor, RootSpanProcessor, RootInstrumentFilterProcessor, ScrubbingSpanProcessor, + SerializationSpanProcessor, SessionSpanProcessor, SpanIOProcessor, } from "../processors"; @@ -804,21 +804,32 @@ function addCustomSpanProcessors( } // ─── Processor registration order matters ─── - // Processors that wrap span.setAttribute form an implicit call chain: - // InstrumentationSpanProcessor (#1) wraps setAttribute for truncation, - // then SpanIOProcessor (#3) wraps the already-wrapped setAttribute for - // IO normalisation. Reordering these will break the chain silently. // - // onEnd runs in registration order: SpanIOProcessor (#3) promotes - // netra.user.input/output → input/output BEFORE RootSpanProcessor (#5) - // cleans up its root span map. + // Processors that wrap span.setAttribute in onStart form an implicit call + // chain. The LAST registered wrapper is the OUTERMOST (first to intercept + // an external setAttribute call). The chain is: + // + // External setAttribute call + // → SpanIOProcessor (outermost: routes/assembles IO attributes) + // → SerializationSpanProcessor (middle: serializes & truncates values) + // → ScrubbingSpanProcessor (innermost: scrubs sensitive data) + // → OTel original setAttribute + // + // Serialization runs before scrubbing so complex objects are converted to + // JSON strings first, allowing the regex-based scrubber to detect sensitive + // patterns in the serialized text. When SpanIOProcessor assembles a derived + // value (e.g. combined `input` JSON) and writes it via its saved `original` + // reference, that reference points to the Serialization wrapper — so + // assembled values still get serialized, scrubbed, and stored. + // + // Processors that only CALL setAttribute (not wrap) must be registered AFTER + // all wrapping processors so their writes flow through the full chain. // 0. Local Filtering Span Processor - filters spans based on local context const localFilteringSpanProcessor = new LocalFilteringSpanProcessor(); provider.addSpanProcessor(localFilteringSpanProcessor); // 0.5. Root Instrument Filter Processor - blocks root spans from non-allowed instrumentations - // When rootInstrumentNames is null, all instrumentations may produce root spans (no filtering). if (rootInstrumentNames !== null) { const rootFilterProcessor = new RootInstrumentFilterProcessor( rootInstrumentNames, @@ -826,49 +837,48 @@ function addCustomSpanProcessors( provider.addSpanProcessor(rootFilterProcessor); } - // 1. Instrumentation Span Processor - truncates attributes and adds instrumentation name. - // MUST run before SpanIOProcessor so the IO processor's setAttribute wrapper - // captures the truncation wrapper in its call chain. + // 1. Scrubbing Span Processor (wraps setAttribute — innermost layer) + // Scrubs sensitive data from already-serialized string values just + // before they reach OTel storage. + if (config.enableScrubbing) { + const scrubbingProcessor = new ScrubbingSpanProcessor(); + provider.addSpanProcessor(scrubbingProcessor); + } + + // 2. Serialization Span Processor (wraps setAttribute — middle layer) + // Serializes objects to JSON strings and truncates values at write time. + // Runs before scrubbing so the scrubber receives string values. + const serializationProcessor = new SerializationSpanProcessor(); + provider.addSpanProcessor(serializationProcessor); + + // 3. Span I/O Processor (wraps setAttribute — outermost layer) + // Routes IO-related attributes, assembles input/output from gen_ai.* + // and traceloop.* keys. Assembled values flow through serialize → scrub. + const spanIOProcessor = new SpanIOProcessor(); + provider.addSpanProcessor(spanIOProcessor); + + // 4. Instrumentation Span Processor - tags instrumentation name. + // Runs AFTER wrappers are in place so its setAttribute calls flow + // through the full chain. const instrumentationProcessor = new InstrumentationSpanProcessor(); provider.addSpanProcessor(instrumentationProcessor); - // 2. Session Span Processor - adds session context (session_id, user_id, etc.) + // 5. Session Span Processor - adds session context (session_id, user_id, etc.) + // Runs AFTER wrappers so session attributes flow through scrub→serialize. const sessionProcessor = new SessionSpanProcessor(config.environment); provider.addSpanProcessor(sessionProcessor); - // 3. Span I/O Processor - normalises input/output from gen_ai.prompt/completion - // and traceloop.entity attributes; remaps traceloop.* → netra.* - // MUST run after InstrumentationSpanProcessor (captures its setAttribute wrapper). - // MUST run before RootSpanProcessor (promotes netra.user.* in onEnd first). - const spanIOProcessor = new SpanIOProcessor(); - provider.addSpanProcessor(spanIOProcessor); - - // 4. LLM Trace Identifier Span Processor - marks root spans that contain LLM calls + // 6. LLM Trace Identifier Span Processor - marks root spans containing LLM calls. + // Uses setAttribute on root span in onEnd (root span is still recording). const llmTraceProcessor = new LlmTraceIdentifierSpanProcessor(); provider.addSpanProcessor(llmTraceProcessor); - // 5. Root Span Processor - tracks the root span per trace by traceId. - // Registered AFTER LlmTraceIdentifierSpanProcessor so that on_end cleanup - // happens after the LLM processor has finished annotating the root span. - // Registered AFTER SpanIOProcessor so root spans have setAttribute wrapped - // before being stored in the root span map. + // 7. Root Span Processor - tracks root span per trace. + // Registered AFTER LlmTraceIdentifierSpanProcessor so onEnd cleanup + // runs after LLM annotation is complete. const rootSpanProcessor = new RootSpanProcessor(); provider.addSpanProcessor(rootSpanProcessor); - // 6. Scrubbing Span Processor - scrubs sensitive data (if enabled) - if (config.enableScrubbing) { - const scrubbingProcessor = new ScrubbingSpanProcessor(); - provider.addSpanProcessor(scrubbingProcessor); - } - - // 7. Attribute Size Limit Processor - enforces hard size limits on span - // attributes before export to prevent "entity too large" errors. - // MUST be last so it acts as final gate after all other processors. - const sizeLimitProcessor = new AttributeSizeLimitProcessor( - Config.SPAN_ATTRIBUTE_MAX_SIZE, - ); - provider.addSpanProcessor(sizeLimitProcessor); - Logger.debug("Custom span processors registered successfully"); return provider; diff --git a/src/instrumentation/openai-agents/attributes.ts b/src/instrumentation/openai-agents/attributes.ts index 3bb3a4f..22210aa 100644 --- a/src/instrumentation/openai-agents/attributes.ts +++ b/src/instrumentation/openai-agents/attributes.ts @@ -20,8 +20,9 @@ import { NETRA_SPAN_NAME, } from "./constants"; import { Logger } from "../../logger"; +import { safeStringify } from "../../utils/serialization"; import { AgentSpan, SpanData } from "./types"; -import { extractContentString, extractInstructions, safeJsonStringify } from "./utils"; +import { extractContentString, extractInstructions } from "./utils"; export function setSpanDataAttributes( otelSpan: OTelSpan, @@ -123,12 +124,12 @@ function setGenerationAttributes(span: OTelSpan, data: SpanData, systemName: str span.setAttribute(SpanAttributes.LLM_PRESENCE_PENALTY, Number(config.presence_penalty)); } if (config.stop !== undefined) { - span.setAttribute(SpanAttributes.LLM_CHAT_STOP_SEQUENCES, safeJsonStringify(config.stop)); + span.setAttribute(SpanAttributes.LLM_CHAT_STOP_SEQUENCES, safeStringify(config.stop)); } if (config.reasoning_effort !== undefined) { span.setAttribute(SpanAttributes.LLM_REQUEST_REASONING_EFFORT, String(config.reasoning_effort)); } else if (config.reasoning !== undefined) { - span.setAttribute(SpanAttributes.LLM_REQUEST_REASONING_EFFORT, safeJsonStringify(config.reasoning)); + span.setAttribute(SpanAttributes.LLM_REQUEST_REASONING_EFFORT, safeStringify(config.reasoning)); } } @@ -202,11 +203,11 @@ function setFunctionAttributes(span: OTelSpan, data: SpanData): void { } if (data.input !== undefined) { span.setAttribute(`${SpanAttributes.LLM_PROMPTS}.0.role`, "tool"); - span.setAttribute(`${SpanAttributes.LLM_PROMPTS}.0.content`, safeJsonStringify(data.input)); + span.setAttribute(`${SpanAttributes.LLM_PROMPTS}.0.content`, safeStringify(data.input)); } if (data.output !== undefined) { span.setAttribute(`${SpanAttributes.LLM_COMPLETIONS}.0.role`, "tool"); - span.setAttribute(`${SpanAttributes.LLM_COMPLETIONS}.0.content`, safeJsonStringify(data.output)); + span.setAttribute(`${SpanAttributes.LLM_COMPLETIONS}.0.content`, safeStringify(data.output)); } } @@ -240,7 +241,7 @@ function setCustomAttributes(span: OTelSpan, data: SpanData): void { span.setAttribute(NETRA_SPAN_NAME, data.name); } if (data.data !== undefined) { - span.setAttribute(NETRA_CUSTOM_DATA, safeJsonStringify(data.data)); + span.setAttribute(NETRA_CUSTOM_DATA, safeStringify(data.data)); } } @@ -342,7 +343,7 @@ function setResponseObjectAttributes(span: OTelSpan, response: unknown): void { span.setAttribute(SpanAttributes.LLM_PRESENCE_PENALTY, resp.presence_penalty); } if (resp.reasoning !== undefined) { - span.setAttribute(SpanAttributes.LLM_REQUEST_REASONING_EFFORT, safeJsonStringify(resp.reasoning)); + span.setAttribute(SpanAttributes.LLM_REQUEST_REASONING_EFFORT, safeStringify(resp.reasoning)); } if (resp.usage) { @@ -398,7 +399,7 @@ function setIndexedPromptAttributes(span: OTelSpan, input: unknown, startIndex = span.setAttribute(`${SpanAttributes.LLM_PROMPTS}.${index}.role`, "assistant"); span.setAttribute( `${SpanAttributes.LLM_PROMPTS}.${index}.content`, - safeJsonStringify({ name: msg.name, arguments: msg.arguments }), + safeStringify({ name: msg.name, arguments: msg.arguments }), ); } else if (msgType === "function_call_output" || msgType === "function_call_result") { span.setAttribute(`${SpanAttributes.LLM_PROMPTS}.${index}.role`, "tool"); @@ -456,7 +457,7 @@ function setIndexedCompletionAttributes(span: OTelSpan, responseData: unknown): span.setAttribute(`${SpanAttributes.LLM_COMPLETIONS}.${index}.role`, "assistant"); span.setAttribute( `${SpanAttributes.LLM_COMPLETIONS}.${index}.content`, - safeJsonStringify({ name: element.name, arguments: element.arguments }), + safeStringify({ name: element.name, arguments: element.arguments }), ); const toolCallId = element.call_id || element.id; if (toolCallId) { @@ -496,7 +497,7 @@ function setIndexedCompletionAttributes(span: OTelSpan, responseData: unknown): span.setAttribute(`${SpanAttributes.LLM_COMPLETIONS}.${index}.role`, "assistant"); span.setAttribute( `${SpanAttributes.LLM_COMPLETIONS}.${index}.content`, - safeJsonStringify({ name: func.name ?? "", arguments: func.arguments ?? "" }), + safeStringify({ name: func.name ?? "", arguments: func.arguments ?? "" }), ); if (tc.id) { span.setAttribute(`${SpanAttributes.LLM_COMPLETIONS}.${index}.tool_call_id`, String(tc.id)); diff --git a/src/instrumentation/openai-agents/constants.ts b/src/instrumentation/openai-agents/constants.ts index 20d21e4..3fc364c 100644 --- a/src/instrumentation/openai-agents/constants.ts +++ b/src/instrumentation/openai-agents/constants.ts @@ -1,7 +1,6 @@ export const INSTRUMENTATION_NAME = "netra.instrumentation.openai_agents"; export const DEFAULT_LLM_SYSTEM = "openai"; export const MAX_INDEXED_MESSAGES = 128; -export const MAX_STRINGIFY_LENGTH = 65_536; export const NETRA_SPAN_TYPE_ATTR = "netra.span.type"; diff --git a/src/instrumentation/openai-agents/processor.ts b/src/instrumentation/openai-agents/processor.ts index 5e67096..fc677ad 100644 --- a/src/instrumentation/openai-agents/processor.ts +++ b/src/instrumentation/openai-agents/processor.ts @@ -8,6 +8,7 @@ import { } from "@opentelemetry/api"; import { SpanType } from "../../types"; import { SpanAttributes } from "../span-attributes"; +import { safeStringify } from "../../utils/serialization"; import { DEFAULT_LLM_SYSTEM, NETRA_AGENTS_GROUP_ID, @@ -17,7 +18,7 @@ import { NETRA_SPAN_TYPE_ATTR, NETRA_WORKFLOW_NAME, } from "./constants"; -import { getNetraSpanType, getSpanName, safeJsonStringify } from "./utils"; +import { getNetraSpanType, getSpanName } from "./utils"; import { setSpanDataAttributes } from "./attributes"; import { Logger } from "../../logger"; import type { AgentSpan, AgentTrace, TracingProcessor } from "./types"; @@ -118,7 +119,7 @@ export class NetraAgentsTracingProcessor implements TracingProcessor { span.setAttribute(NETRA_AGENTS_GROUP_ID, agentTrace.groupId); } if (agentTrace.metadata) { - span.setAttribute(NETRA_AGENTS_METADATA, safeJsonStringify(agentTrace.metadata)); + span.setAttribute(NETRA_AGENTS_METADATA, safeStringify(agentTrace.metadata)); } this._rootSpans.set(agentTrace.traceId, span); @@ -227,8 +228,9 @@ export class NetraAgentsTracingProcessor implements TracingProcessor { } if (agentSpan.error) { + const MAX_ERROR_DATA_LEN = 4096; const errorMsg = agentSpan.error.data - ? `${agentSpan.error.message}: ${safeJsonStringify(agentSpan.error.data)}` + ? `${agentSpan.error.message}: ${safeStringify(agentSpan.error.data, { maxLength: MAX_ERROR_DATA_LEN })}` : agentSpan.error.message; span.setStatus({ code: SpanStatusCode.ERROR, message: errorMsg }); span.recordException(new Error(agentSpan.error.message)); diff --git a/src/instrumentation/openai-agents/utils.ts b/src/instrumentation/openai-agents/utils.ts index 5ba67c5..f43eeb2 100644 --- a/src/instrumentation/openai-agents/utils.ts +++ b/src/instrumentation/openai-agents/utils.ts @@ -1,18 +1,6 @@ import { SpanType } from "../../types"; -import { MAX_STRINGIFY_LENGTH } from "./constants"; import { AgentSpan, SpanData } from "./types"; - -export function safeJsonStringify(obj: unknown, maxLength: number = MAX_STRINGIFY_LENGTH): string { - try { - if (typeof obj === "string") { - return maxLength > 0 && obj.length > maxLength ? obj.slice(0, maxLength) : obj; - } - const s = JSON.stringify(obj); - return maxLength > 0 && s.length > maxLength ? s.slice(0, maxLength) : s; - } catch { - return String(obj); - } -} +import { safeStringify } from "../../utils/serialization"; export function getNetraSpanType(data: SpanData): SpanType { switch (data.type) { @@ -36,7 +24,8 @@ export function getNetraSpanType(data: SpanData): SpanType { export function getSpanName(agentSpan: AgentSpan): string { const data = agentSpan.spanData; if (data.name) return data.name; - if (data.type === "handoff" && data.to_agent) return `handoff to ${data.to_agent}`; + if (data.type === "handoff" && data.to_agent) + return `handoff to ${data.to_agent}`; return `openai.agents.${data.type}`; } @@ -53,7 +42,7 @@ export function extractContentString(content: unknown): string { } } } - return texts.length > 0 ? texts.join(" ") : safeJsonStringify(content); + return texts.length > 0 ? texts.join(" ") : safeStringify(content); } return String(content); } diff --git a/src/instrumentation/utils.ts b/src/instrumentation/utils.ts index ba5d43b..53db4ae 100644 --- a/src/instrumentation/utils.ts +++ b/src/instrumentation/utils.ts @@ -4,7 +4,6 @@ */ import { Span, context, type Context as OTelContext } from "@opentelemetry/api"; -import { Config } from "../config"; import { Logger } from "../logger"; import { safeStringify } from "../utils/serialization"; import { SpanAttributes } from "./span-attributes"; @@ -212,7 +211,7 @@ export function buildInputMessages( tool_use_id: block.tool_use_id, content: block.content, is_error: block.is_error ?? false, - }, Config.CONVERSATION_MAX_LEN), + }), }); } } @@ -244,7 +243,7 @@ export function buildInputMessages( // Embeddings / genericinput const input = kwargs.input ?? kwargs.inputs; if (hasContent(input)) { - const content = safeStringify(input, Config.CONVERSATION_MAX_LEN); + const content = safeStringify(input); messages.push({ role: "user", content }); } } @@ -456,7 +455,7 @@ export function setRequestAttributes( // Tool definitions if (Array.isArray(kwargs.tools)) { - span.setAttribute("tools", safeStringify(kwargs.tools, Config.ATTRIBUTE_MAX_LEN)); + span.setAttribute("tools", safeStringify(kwargs.tools)); } // Reasoning config @@ -476,7 +475,7 @@ export function setRequestAttributes( if (kwargs.suffix !== undefined) { span.setAttribute( "llm.request.suffix", - safeStringify(kwargs.suffix, Config.CONVERSATION_MAX_LEN), + safeStringify(kwargs.suffix), ); } } diff --git a/src/processors/attribute-size-limit-processor.ts b/src/processors/attribute-size-limit-processor.ts deleted file mode 100644 index f274d64..0000000 --- a/src/processors/attribute-size-limit-processor.ts +++ /dev/null @@ -1,75 +0,0 @@ -/** - * Attribute Size Limit Span Processor - * - * Enforces a hard max length on every span attribute value via setAttribute - * wrapping in onStart. Prevents "entity too large" errors during export. - */ - -import { AttributeValue, Context, Span } from "@opentelemetry/api"; -import { ReadableSpan, SpanProcessor } from "@opentelemetry/sdk-trace-base"; -import { Logger } from "../logger"; - -const DEFAULT_MAX_ATTRIBUTE_SIZE = 32_000; // 32KB per attribute - -type SetAttributeFn = (key: string, value: AttributeValue) => Span; - -export class AttributeSizeLimitProcessor implements SpanProcessor { - private maxAttributeSize: number; - - constructor(maxAttributeSize?: number) { - this.maxAttributeSize = maxAttributeSize ?? DEFAULT_MAX_ATTRIBUTE_SIZE; - } - - onStart(span: Span, _parentContext: Context): void { - try { - this._wrapSetAttribute(span); - } catch (e) { - Logger.debug("AttributeSizeLimitProcessor.onStart error:", e); - } - } - - onEnd(_span: ReadableSpan): void {} - - shutdown(): Promise { - return Promise.resolve(); - } - - forceFlush(): Promise { - return Promise.resolve(); - } - - private _wrapSetAttribute(span: Span): void { - const original: SetAttributeFn = span.setAttribute.bind(span); - const maxLen = this.maxAttributeSize; - - const patched = (key: string, value: AttributeValue): Span => { - try { - return original(key, truncateValue(value, maxLen) as AttributeValue); - } catch { - try { - return original(key, value); - } catch { - return span; - } - } - }; - - (span as any).setAttribute = patched; - } -} - -function truncateValue(value: unknown, maxLen: number): unknown { - if (typeof value === "string") { - return value.length > maxLen ? value.substring(0, maxLen) : value; - } - - if (Array.isArray(value)) { - const serialized = JSON.stringify(value); - if (serialized.length > maxLen) { - return serialized.substring(0, maxLen); - } - return value; - } - - return value; -} diff --git a/src/processors/index.ts b/src/processors/index.ts index 70df132..687e243 100644 --- a/src/processors/index.ts +++ b/src/processors/index.ts @@ -5,12 +5,12 @@ * attribute management, session tracking, and sensitive data scrubbing. */ -export { AttributeSizeLimitProcessor } from "./attribute-size-limit-processor"; export { InstrumentationSpanProcessor } from "./instrumentation-span-processor"; export { LlmTraceIdentifierSpanProcessor } from "./llm-trace-identifier-span-processor"; export { withBlockedSpansLocal } from "./localfiltering-span-processor"; export { RootSpanProcessor } from "./root-span-processor"; export { ScrubbingSpanProcessor } from "./scrubbing-span-processor"; +export { SerializationSpanProcessor } from "./serialization-span-processor"; export { SessionSpanProcessor, setSessionBaggage, diff --git a/src/processors/instrumentation-span-processor.ts b/src/processors/instrumentation-span-processor.ts index 1fc6e24..3d1a26c 100644 --- a/src/processors/instrumentation-span-processor.ts +++ b/src/processors/instrumentation-span-processor.ts @@ -1,9 +1,9 @@ /** * Instrumentation Span Processor - * - * OpenTelemetry span processor that: - * 1. Records the raw instrumentation scope name for each span - * 2. Truncates attribute values to prevent oversized payloads + * + * OpenTelemetry span processor that records the raw instrumentation scope + * name for each span. This enables downstream processors and the dashboard + * to identify which provider/framework produced a given span. */ import { Context, Span } from "@opentelemetry/api"; @@ -45,19 +45,8 @@ const ALLOWED_INSTRUMENTATION_NAMES = new Set([ ]); export class InstrumentationSpanProcessor implements SpanProcessor { - private maxAttributeLength: number; - - constructor(maxAttributeLength?: number) { - this.maxAttributeLength = maxAttributeLength ?? Config.ATTRIBUTE_MAX_LEN; - } - - /** - * Detect the raw instrumentation name from the span's instrumentation scope. - */ private detectRawInstrumentationName(span: Span): string | null { try { - // Access the instrumentation scope from the span - // In JS SDK, this is typically available via the span's internal properties const spanAny = span as any; const scope = spanAny.instrumentationLibrary || spanAny.instrumentationScope; @@ -80,73 +69,20 @@ export class InstrumentationSpanProcessor implements SpanProcessor { return base; } } catch { - // Fall through to return the full name + Logger.debug("InstrumentationSpanProcessor: Error extracting base name:", name); } } return name; } } } catch { - // Ignore errors + Logger.debug("InstrumentationSpanProcessor: Error detecting instrumentation name"); } return null; } - /** - * Truncate a value to the maximum allowed length. - * Handles strings, bytes, and recursively handles lists and objects. - */ - private truncateValue(value: unknown): unknown { + onStart(span: Span, _parentContext: Context): void { try { - if (typeof value === "string") { - return value.length <= this.maxAttributeLength - ? value - : value.substring(0, this.maxAttributeLength); - } - - if (Array.isArray(value)) { - return value.map((v) => - typeof v === "string" ? this.truncateValue(v) : v - ); - } - - if (typeof value === "object" && value !== null) { - const result: Record = {}; - for (const [k, v] of Object.entries(value)) { - result[k] = typeof v === "string" ? this.truncateValue(v) : v; - } - return result; - } - } catch { - return value; - } - return value; - } - - /** - * Called when a span starts. Wraps setAttribute to truncate values - * and records the instrumentation name. - */ - onStart(span: Span, parentContext: Context): void { - try { - // Wrap setAttribute to truncate values - const originalSetAttribute = span.setAttribute.bind(span); - - span.setAttribute = (key: string, value: unknown): Span => { - try { - const truncated = this.truncateValue(value); - return originalSetAttribute(key, truncated as any); - } catch { - // Best-effort; never break span - try { - return originalSetAttribute(key, value as any); - } catch { - return span; - } - } - }; - - // Set instrumentation name attribute const name = this.detectRawInstrumentationName(span); if (name && ALLOWED_INSTRUMENTATION_NAMES.has(name.toLowerCase())) { span.setAttribute(`${Config.LIBRARY_NAME}.instrumentation.name`, name); @@ -154,28 +90,17 @@ export class InstrumentationSpanProcessor implements SpanProcessor { } catch (e) { Logger.error( "InstrumentationSpanProcessor: Error on span start:", - e + e, ); } } - /** - * Called when a span ends. No-op for this processor. - */ - onEnd(span: ReadableSpan): void { - // No-op - } + onEnd(_span: ReadableSpan): void {} - /** - * Shuts down the processor. - */ shutdown(): Promise { return Promise.resolve(); } - /** - * Forces a flush of any pending spans. - */ forceFlush(): Promise { return Promise.resolve(); } diff --git a/src/processors/scrubbing-span-processor.ts b/src/processors/scrubbing-span-processor.ts index 99e9ccb..2f0bc52 100644 --- a/src/processors/scrubbing-span-processor.ts +++ b/src/processors/scrubbing-span-processor.ts @@ -1,16 +1,27 @@ /** * Scrubbing Span Processor * - * OpenTelemetry span processor that scrubs sensitive data from span attributes. - * This includes API keys, emails, phone numbers, SSNs, credit cards, passwords, - * bearer tokens, and other sensitive information. + * OpenTelemetry span processor that scrubs sensitive data from span attributes + * at write time via setAttribute wrapping. This includes API keys, emails, + * phone numbers, SSNs, credit cards, passwords, bearer tokens, and other + * sensitive information. + * + * Handles both pre-serialized string values (the normal path when + * SerializationSpanProcessor is active) and raw objects/arrays (defensive + * fallback for custom pipelines that skip serialization). + * + * Wraps in onStart to avoid unreliable _attributes mutation in onEnd. + * MUST be registered BEFORE SerializationSpanProcessor so it forms the + * innermost layer (closest to OTel storage). Because serialization wraps + * outside this layer, values are serialized to strings first, then this + * processor scrubs sensitive patterns from the resulting string values. */ -import { Context, Span } from "@opentelemetry/api"; +import { AttributeValue, Context, Span } from "@opentelemetry/api"; import { ReadableSpan, SpanProcessor } from "@opentelemetry/sdk-trace-base"; import { Logger } from "../logger"; -// Sensitive patterns for data detection (based on pydantic logfire scrubbing) +// Sensitive patterns for data detection const SENSITIVE_PATTERNS: Record = { // API keys - scrub "Token: " or sk-... tokens api_key: new RegExp( @@ -63,149 +74,106 @@ const SENSITIVE_KEYS = new Set([ const SCRUB_REPLACEMENT = "[SCRUBBED]"; -export class ScrubbingSpanProcessor implements SpanProcessor { - /** - * Check if a key is considered sensitive. - */ - private isSensitiveKey(key: string): boolean { - const keyLower = key.toLowerCase(); - for (const sensitiveKey of SENSITIVE_KEYS) { - if (keyLower.includes(sensitiveKey)) { - return true; - } +type SetAttributeFn = (key: string, value: AttributeValue) => Span; + +function isSensitiveKey(key: string): boolean { + const keyLower = key.toLowerCase(); + for (const sensitiveKey of SENSITIVE_KEYS) { + if (keyLower.includes(sensitiveKey)) { + return true; } - return false; } + return false; +} - /** - * Scrub sensitive patterns from a string value. - */ - private scrubStringValue(value: string): string { - let scrubbed = value; +function scrubStringValue(value: string): string { + let scrubbed = value; - // Early catch-all for contiguous 13-19 digit sequences (credit/debit cards) - scrubbed = scrubbed.replace(LONG_DIGIT_PATTERN, SCRUB_REPLACEMENT); + // Early catch-all for contiguous 13-19 digit sequences (credit/debit cards) + scrubbed = scrubbed.replace(LONG_DIGIT_PATTERN, SCRUB_REPLACEMENT); - // Apply all sensitive patterns - for (const pattern of Object.values(SENSITIVE_PATTERNS)) { - // Reset lastIndex for global patterns + // Apply all sensitive patterns + for (const pattern of Object.values(SENSITIVE_PATTERNS)) { + // Reset lastIndex for global patterns + pattern.lastIndex = 0; + if (pattern.test(scrubbed)) { pattern.lastIndex = 0; - if (pattern.test(scrubbed)) { - pattern.lastIndex = 0; - scrubbed = scrubbed.replace(pattern, SCRUB_REPLACEMENT); - } + scrubbed = scrubbed.replace(pattern, SCRUB_REPLACEMENT); } - - return scrubbed; } - /** - * Recursively scrub sensitive data from a dictionary value. - */ - private scrubDictValue( - value: Record, - ): Record { - const scrubbed: Record = {}; - for (const [k, v] of Object.entries(value)) { - const [, scrubbedValue] = this.scrubKeyValue(k, v); - scrubbed[k] = scrubbedValue; - } - return scrubbed; - } + return scrubbed; +} - /** - * Recursively scrub sensitive data from a list value. - */ - private scrubListValue(value: unknown[]): unknown[] { - return value.map((item) => { - if (typeof item === "string") { - return this.scrubStringValue(item); - } else if (Array.isArray(item)) { - return this.scrubListValue(item); - } else if (typeof item === "object" && item !== null) { - return this.scrubDictValue(item as Record); - } - return item; - }); +function scrubDict(obj: Record): Record { + const result: Record = {}; + for (const [k, v] of Object.entries(obj)) { + result[k] = scrubAny(k, v); } + return result; +} - /** - * Scrub sensitive data from a key-value pair. - */ - private scrubKeyValue(key: string, value: unknown): [string, unknown] { - // Check if key itself is sensitive and value is a simple type - if ( - this.isSensitiveKey(key) && - typeof value !== "object" && - !Array.isArray(value) - ) { - return [key, SCRUB_REPLACEMENT]; - } - - // Scrub value based on its type - if (typeof value === "string") { - return [key, this.scrubStringValue(value)]; - } else if (Array.isArray(value)) { - return [key, this.scrubListValue(value)]; - } else if (typeof value === "object" && value !== null) { - return [key, this.scrubDictValue(value as Record)]; +function scrubList(arr: unknown[]): unknown[] { + return arr.map((item) => { + if (typeof item === "string") return scrubStringValue(item); + if (Array.isArray(item)) return scrubList(item); + if (typeof item === "object" && item !== null) { + return scrubDict(item as Record); } + return item; + }); +} - return [key, value]; +function scrubAny(key: string, value: unknown): unknown { + if (isSensitiveKey(key) && typeof value !== "object" && !Array.isArray(value)) { + return SCRUB_REPLACEMENT; } - - /** - * Called when a span starts. No-op for this processor. - */ - onStart(span: Span, parentContext: Context): void { - // No-op - scrubbing happens on span end + if (typeof value === "string") return scrubStringValue(value); + if (Array.isArray(value)) return scrubList(value); + if (typeof value === "object" && value !== null) { + return scrubDict(value as Record); } + return value; +} - /** - * Called when a span ends. Scrubs sensitive data from span attributes. - */ - onEnd(span: ReadableSpan): void { - try { - // Get span attributes - we need to access the internal attributes - const spanAny = span as any; - const attributes = spanAny.attributes || spanAny._attributes; - - if (attributes && typeof attributes === "object") { - const scrubbedAttributes: Record = {}; - - for (const [key, value] of Object.entries(attributes)) { - const [, scrubbedValue] = this.scrubKeyValue(key, value); - scrubbedAttributes[key] = scrubbedValue; - } +function scrubValue(key: string, value: AttributeValue): AttributeValue { + return scrubAny(key, value) as AttributeValue; +} - // Replace attributes with scrubbed versions - // This works because ReadableSpan is actually a reference to the span object - if (spanAny._attributes) { - spanAny._attributes = scrubbedAttributes; - } else if (spanAny.attributes) { - // Some implementations use different property names - Object.keys(spanAny.attributes).forEach( - (k) => delete spanAny.attributes[k], - ); - Object.assign(spanAny.attributes, scrubbedAttributes); - } - } +export class ScrubbingSpanProcessor implements SpanProcessor { + onStart(span: Span, _parentContext: Context): void { + try { + this.wrapSetAttribute(span); } catch (e) { - Logger.error("ScrubbingSpanProcessor: Error scrubbing attributes:", e); + Logger.error("ScrubbingSpanProcessor.onStart error:", e); } } - /** - * Shuts down the processor. - */ + onEnd(_span: ReadableSpan): void {} + shutdown(): Promise { return Promise.resolve(); } - /** - * Forces a flush of any pending spans. - */ forceFlush(): Promise { return Promise.resolve(); } + + private wrapSetAttribute(span: Span): void { + const original: SetAttributeFn = span.setAttribute.bind(span); + + const patched = (key: string, value: AttributeValue): Span => { + try { + return original(key, scrubValue(key, value)); + } catch { + try { + return original(key, value); + } catch { + return span; + } + } + }; + + (span as any).setAttribute = patched; + } } diff --git a/src/processors/serialization-span-processor.ts b/src/processors/serialization-span-processor.ts new file mode 100644 index 0000000..3ca7881 --- /dev/null +++ b/src/processors/serialization-span-processor.ts @@ -0,0 +1,70 @@ +/** + * Serialization Span Processor + * + * Single enforcement point for attribute serialization and truncation. + * Wraps span.setAttribute in onStart so every value is serialized and + * truncated at write time. Uses jsonrepair-based truncation for JSON + * values and `...[TRUNCATED]` suffix for plain strings, aligned with + * backend behavior. + * + * MUST be registered AFTER ScrubbingSpanProcessor so it wraps outside + * the scrubbing layer. In the resulting call chain, serialization runs + * before scrubbing — converting objects to JSON strings so the + * regex-based scrubber can detect sensitive patterns in string form. + * All values, including those assembled by SpanIOProcessor, pass + * through serialization → scrubbing before reaching OTel storage. + */ + +import { AttributeValue, Context, Span } from "@opentelemetry/api"; +import { ReadableSpan, SpanProcessor } from "@opentelemetry/sdk-trace-base"; +import { serializeAttribute } from "../utils/serialization/truncating-serializer"; +import { Config } from "../config"; +import { Logger } from "../logger"; + +type SetAttributeFn = (key: string, value: AttributeValue) => Span; + +export class SerializationSpanProcessor implements SpanProcessor { + private readonly maxAttributeSize: number; + + constructor(maxAttributeSize?: number) { + this.maxAttributeSize = maxAttributeSize ?? Config.SPAN_ATTRIBUTE_MAX_SIZE; + } + + onStart(span: Span, _parentContext: Context): void { + try { + this.wrapSetAttribute(span); + } catch (e) { + Logger.debug("SerializationSpanProcessor.onStart error:", e); + } + } + + onEnd(_span: ReadableSpan): void {} + + shutdown(): Promise { + return Promise.resolve(); + } + + forceFlush(): Promise { + return Promise.resolve(); + } + + private wrapSetAttribute(span: Span): void { + const original: SetAttributeFn = span.setAttribute.bind(span); + const maxSize = this.maxAttributeSize; + + const patched = (key: string, value: AttributeValue): Span => { + try { + const truncated = serializeAttribute(value, maxSize); + return original(key, truncated as AttributeValue); + } catch { + try { + return original(key, value); + } catch { + return span; + } + } + }; + + (span as any).setAttribute = patched; + } +} diff --git a/src/session-manager.ts b/src/session-manager.ts index 9ed7c80..e8f81c0 100644 --- a/src/session-manager.ts +++ b/src/session-manager.ts @@ -13,7 +13,7 @@ import { Config } from "./config"; import { Logger } from "./logger"; import { RootSpanProcessor } from "./processors/root-span-processor"; import { ConversationType } from "./types"; -import { serializeValue } from "./utils/serialization"; +import { safeStringify } from "./utils/serialization"; export { ConversationType }; @@ -31,6 +31,13 @@ interface EntityContext { spansByName: Map; } +type ConversationEntry = { + type: string; + role: string; + content: string | Record; + format: string; +}; + const entityStorage = new AsyncLocalStorage(); // Global fallback for single-threaded / non-async entry points @@ -163,7 +170,7 @@ export class SessionManager { try { const span = trace.getActiveSpan(); if (span?.isRecording()) { - span.setAttribute(key, typeof value === "string" ? value : JSON.stringify(value)); + span.setAttribute(key, typeof value === "string" ? value : safeStringify(value)); } else { Logger.warn(`setAttributeOnActiveSpan: no recording span for key '${key}'`); } @@ -181,7 +188,7 @@ export class SessionManager { try { SessionManager.setAttributeOnActiveSpan( "netra.user.input", - serializeValue(value, Config.ATTRIBUTE_MAX_LEN), + safeStringify(value), ); } catch (e) { Logger.error("setInput failed:", e); @@ -197,7 +204,7 @@ export class SessionManager { try { SessionManager.setAttributeOnActiveSpan( "netra.user.output", - serializeValue(value, Config.ATTRIBUTE_MAX_LEN), + safeStringify(value), ); } catch (e) { Logger.error("setOutput failed:", e); @@ -213,7 +220,7 @@ export class SessionManager { try { RootSpanProcessor.setAttributeOnRootSpan( "netra.root.input", - serializeValue(value, Config.ATTRIBUTE_MAX_LEN), + safeStringify(value), ); } catch (e) { Logger.error("setRootInput failed:", e); @@ -229,7 +236,7 @@ export class SessionManager { try { RootSpanProcessor.setAttributeOnRootSpan( "netra.root.output", - serializeValue(value, Config.ATTRIBUTE_MAX_LEN), + safeStringify(value), ); } catch (e) { Logger.error("setRootOutput failed:", e); @@ -266,8 +273,8 @@ export class SessionManager { * * Reads and re-serialises the existing JSON array so entries accumulate * rather than overwrite — matching the Python SDK's behaviour. - * Writes directly to the span's internal attribute store to bypass OTel's - * per-attribute length truncation on the final payload. + * Uses span.setAttribute so the value flows through the processor pipeline + * including the SerializationSpanProcessor for proper truncation. */ static addConversation( conversationType: ConversationType, @@ -286,43 +293,33 @@ export class SessionManager { return; } - // Read existing entries from the span's internal attribute store - let existing: Array<{ - type: string; - role: string; - content: string | Record; - format: string; - }> = []; + let existing: ConversationEntry[] = []; try { - const raw = (span as any)._attributes?.["conversation"]; + const raw = (span as any).attributes?.["conversation"]; if (typeof raw === "string") { const parsed = JSON.parse(raw); - if (Array.isArray(parsed)) existing = parsed; + if (Array.isArray(parsed)) { + // Filter out truncation markers injected by the serialization processor + existing = parsed.filter( + (entry: any) => + !(entry && typeof entry === "object" && entry.__truncated__ === true && + Object.keys(entry).length === 1), + ); + } } } catch (e) { Logger.warn("addConversation: failed to parse existing conversation, starting fresh:", e); } - const maxLen = Config.CONVERSATION_MAX_LEN; - const processedContent = - typeof content === "string" ? content.substring(0, maxLen) : content; - existing.push({ type: conversationType, role, - content: processedContent, - format: typeof processedContent === "string" ? "text" : "json", + content, + format: typeof content === "string" ? "text" : "json", }); const payload = JSON.stringify(existing); - - // Bypass per-attribute truncation by writing to the internal store directly - const internalAttrs = (span as any)._attributes; - if (internalAttrs && typeof internalAttrs === "object") { - internalAttrs["conversation"] = payload; - } else { - span.setAttribute("conversation", payload); - } + span.setAttribute("conversation", payload); } catch (e) { Logger.error("addConversation failed:", e); } diff --git a/src/utils/serialization.ts b/src/utils/serialization.ts deleted file mode 100644 index 264e1d6..0000000 --- a/src/utils/serialization.ts +++ /dev/null @@ -1,84 +0,0 @@ -const ELLIPSIS = "..."; - -function truncate(s: string, maxLength: number): string { - if (s.length <= maxLength) return s; - if (maxLength <= ELLIPSIS.length) return s.slice(0, maxLength); - return s.slice(0, maxLength - ELLIPSIS.length) + ELLIPSIS; -} - -/** - * Circular-reference-safe JSON.stringify with optional length truncation. - * - * - Strings pass through as-is (only truncated if over maxLength) - * - Functions, symbols, bigints get descriptive placeholders - * - Circular references become "[Circular]" - * - Large class instances (>20 keys) become "[ClassName]" - * - Result is truncated to `maxLength` when provided - */ -export function safeStringify( - value: unknown, - maxLength?: number, -): string { - if (typeof value === "string") { - if (maxLength && value.length > maxLength) { - return truncate(value, maxLength); - } - return value; - } - - const seen = new WeakSet(); - let result: string; - try { - result = JSON.stringify(value, (_key, val) => { - if (typeof val === "string" && maxLength && val.length > maxLength) { - return truncate(val, maxLength); - } - if (typeof val === "function") - return `[Function: ${val.name || "anonymous"}]`; - if (typeof val === "symbol") return val.toString(); - if (typeof val === "bigint") return val.toString(); - if (val !== null && typeof val === "object") { - if (seen.has(val)) return "[Circular]"; - seen.add(val); - const name = val.constructor?.name; - if ( - name && - name !== "Object" && - name !== "Array" && - Object.keys(val).length > 20 - ) { - return `[${name}]`; - } - } - return val; - }) ?? String(value); - } catch { - result = - (value as any)?.constructor?.name - ? `[${(value as any).constructor.name}]` - : String(value); - } - - if (maxLength && result.length > maxLength) { - return truncate(result, maxLength); - } - return result; -} - -/** - * Convert any value to a string suitable for span attributes. - * Primitives use `String()`, objects use `safeStringify`. - */ -export function serializeValue( - value: unknown, - maxLength?: number, -): string { - if (value === null || value === undefined) return String(value); - const t = typeof value; - if (t === "string" || t === "number" || t === "boolean") { - const s = String(value); - if (maxLength && s.length > maxLength) return truncate(s, maxLength); - return s; - } - return safeStringify(value, maxLength); -} diff --git a/src/utils/serialization/index.ts b/src/utils/serialization/index.ts new file mode 100644 index 0000000..42906bb --- /dev/null +++ b/src/utils/serialization/index.ts @@ -0,0 +1,10 @@ +export { safeStringify } from "./safe-stringify"; +export type { SafeStringifyOptions } from "./safe-stringify"; +export { + TruncatingSerializer, + serializeAttribute, +} from "./truncating-serializer"; +export type { + TruncatingSerializerConfig, + SerializedAttributeValue, +} from "./truncating-serializer"; diff --git a/src/utils/serialization/safe-stringify.ts b/src/utils/serialization/safe-stringify.ts new file mode 100644 index 0000000..54c18ac --- /dev/null +++ b/src/utils/serialization/safe-stringify.ts @@ -0,0 +1,67 @@ +export interface SafeStringifyOptions { + maxLength?: number; +} + +/** + * Circular-reference-safe JSON serialization. + * + * Handles circular references, functions, symbols, bigints, and + * large class instances gracefully. Never throws. + * + * When called without options, performs pure serialization with no truncation. + * Pass `{ maxLength }` to enable hard truncation of the final result. + */ +export function safeStringify(value: unknown, options?: SafeStringifyOptions): string { + const maxLength = options?.maxLength; + + if (typeof value === "string") { + if (maxLength && value.length > maxLength) { + return value.slice(0, maxLength); + } + return value; + } + + if (value === null) return "null"; + if (value === undefined) return "undefined"; + + const t = typeof value; + if (t === "number" || t === "boolean") return String(value); + if (t === "bigint") return value.toString(); + if (t === "symbol") return (value as symbol).toString(); + if (t === "function") return `[Function: ${(value as Function).name || "anonymous"}]`; + + const seen = new WeakSet(); + let result: string; + + try { + result = JSON.stringify(value, (_key, val) => { + if (typeof val === "function") + return `[Function: ${val.name || "anonymous"}]`; + if (typeof val === "symbol") return val.toString(); + if (typeof val === "bigint") return val.toString(); + if (val !== null && typeof val === "object") { + if (seen.has(val)) return "[Circular]"; + seen.add(val); + const name = val.constructor?.name; + if ( + name && + name !== "Object" && + name !== "Array" && + Object.keys(val).length > 20 + ) { + return `[${name}]`; + } + } + return val; + }) ?? "null"; + } catch { + const name = (value as any)?.constructor?.name; + result = name ? `[Unserializable: ${name}]` : "[Unserializable]"; + } + + if (maxLength && result.length > maxLength) { + return result.slice(0, maxLength); + } + + return result; +} diff --git a/src/utils/serialization/truncating-serializer.ts b/src/utils/serialization/truncating-serializer.ts new file mode 100644 index 0000000..f35964f --- /dev/null +++ b/src/utils/serialization/truncating-serializer.ts @@ -0,0 +1,209 @@ +import { jsonrepair } from "jsonrepair"; +import { safeStringify } from "./safe-stringify"; +import { Logger } from "../../logger"; + +const TRUNCATION_SUFFIX = "...[TRUNCATED]"; + +const MINIMAL_TRUNCATED_OBJECT = '{"__truncated__":true}'; +const MINIMAL_TRUNCATED_ARRAY = '[{"__truncated__":true}]'; +const MIN_JSON_TRUNCATION_LENGTH = MINIMAL_TRUNCATED_ARRAY.length; // 24 + +const RETRY_RATIOS = [0.7, 0.5, 0.3]; + +export interface TruncatingSerializerConfig { + maxAttributeSize: number; +} + +const DEFAULT_CONFIG: TruncatingSerializerConfig = { + maxAttributeSize: 30000, +}; + +function isJsonLike(str: string): boolean { + const trimmed = str.trimStart(); + return trimmed.startsWith("{") || trimmed.startsWith("["); +} + +function minimalTruncatedJson(json: string): string { + return json.trimStart().startsWith("[") + ? MINIMAL_TRUNCATED_ARRAY + : MINIMAL_TRUNCATED_OBJECT; +} + +function injectTruncationMarker(parsed: unknown): void { + if (Array.isArray(parsed)) { + parsed.push({ __truncated__: true }); + } else if (typeof parsed === "object" && parsed !== null) { + (parsed as Record).__truncated__ = true; + } +} + +function tryRepairAndFit( + json: string, + cutLen: number, + budget: number, +): string | null { + if (cutLen <= 0) return null; + try { + const repaired = jsonrepair(json.slice(0, cutLen)); + const parsed: unknown = JSON.parse(repaired); + injectTruncationMarker(parsed); + const result = JSON.stringify(parsed); + if (result.length <= budget) return result; + } catch { + Logger.debug("Failed to repair and fit JSON at cut length", { + length: json.length, + cutLength: cutLen, + }); + } + return null; +} + +function truncateJsonValue(json: string, budget: number): string { + const minimal = minimalTruncatedJson(json); + if (budget < minimal.length) { + return truncatePlainString(json, budget); + } + + // Estimate overhead: the marker itself plus structural chars added by jsonrepair. + // Use a generous estimate to leave room for repair expansion. + const overhead = Math.max(30, Math.ceil(budget * 0.05)); + const initialCut = budget - overhead; + + // Phase 1: fixed ratio candidates (coarse) + const candidateLengths = [ + initialCut, + ...RETRY_RATIOS.map((ratio) => Math.floor(initialCut * ratio)), + ]; + + for (const cutLen of candidateLengths) { + const result = tryRepairAndFit(json, cutLen, budget); + if (result !== null) return result; + } + + // Phase 2: binary search for the largest cut that fits + let lo = 1; + let hi = Math.max(1, Math.floor(initialCut * 0.1)); + let bestResult: string | null = null; + + while (lo <= hi) { + const mid = Math.floor((lo + hi) / 2); + const result = tryRepairAndFit(json, mid, budget); + if (result !== null) { + bestResult = result; + lo = mid + 1; + } else { + hi = mid - 1; + } + } + if (bestResult !== null) return bestResult; + + return minimal; +} + +function truncatePlainString(str: string, maxLen: number): string { + if (maxLen <= TRUNCATION_SUFFIX.length) { + return str.slice(0, maxLen); + } + return str.slice(0, maxLen - TRUNCATION_SUFFIX.length) + TRUNCATION_SUFFIX; +} + +function truncateString(value: string, budget: number): string { + if (value.length <= budget) return value; + + if (isJsonLike(value) && budget >= MIN_JSON_TRUNCATION_LENGTH) { + return truncateJsonValue(value, budget); + } + + return truncatePlainString(value, budget); +} + +export type SerializedAttributeValue = + | string + | number + | boolean + | Array + | Array + | Array; + +/** + * Serializes and truncates a single span attribute value. + * + * - Numbers and booleans pass through unchanged (they're always small). + * - Strings are truncated using jsonrepair-based truncation (for JSON) + * or `...[TRUNCATED]` suffix (for plain strings). + * - Homogeneous primitive arrays (number[], boolean[], string[]) are + * kept as native OTel arrays when their serialized form fits within + * the budget; otherwise they are serialized to a JSON string and truncated. + * - Everything else is serialized via safeStringify then truncated. + * + * This is the single source of truth for attribute size enforcement. + */ +export function serializeAttribute( + value: unknown, + maxSize: number, +): SerializedAttributeValue { + if (typeof value === "number" || typeof value === "boolean") { + return value; + } + + if (typeof value === "string") { + return truncateString(value, maxSize); + } + + if (Array.isArray(value)) { + if (value.every((v) => v == null || typeof v === "number")) { + const serializedLen = estimateArraySize(value); + if (serializedLen <= maxSize) { + return value as Array; + } + } else if (value.every((v) => v == null || typeof v === "boolean")) { + const serializedLen = estimateArraySize(value); + if (serializedLen <= maxSize) { + return value as Array; + } + } else if (value.every((v) => v == null || typeof v === "string")) { + // `s == null` intentionally matches both `null` and `undefined`. + // In arrays, both `null` and `undefined` are serialized as `null`, + // so both contribute 4 characters. + const serializedLen = value.reduce( + (sum, s) => sum + (s == null ? 4 : (s as string).length + 2), + 2 + Math.max(0, value.length - 1), + ); + if (serializedLen <= maxSize) { + return value as Array; + } + } + } + + const serialized = safeStringify(value); + return truncateString(serialized, maxSize); +} + +function estimateArraySize(arr: unknown[]): number { + let size = 2; // [ ] + for (let i = 0; i < arr.length; i++) { + if (i > 0) size += 1; // comma + const v = arr[i]; + // Check to handle null and undefined, since both are serialized as null. + // Without this check, length of undefined is 9, which is longer than the length of null. + if (v == null) size += 4; + else size += String(v).length; + } + return size; +} + +export class TruncatingSerializer { + private readonly maxAttributeSize: number; + + constructor(config?: Partial) { + const merged = { ...DEFAULT_CONFIG, ...config }; + this.maxAttributeSize = merged.maxAttributeSize; + } + + serializeAttribute( + value: unknown, + budget?: number, + ): SerializedAttributeValue { + return serializeAttribute(value, budget ?? this.maxAttributeSize); + } +} diff --git a/src/version.ts b/src/version.ts index 68b66ac..2c8ca3b 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1 +1 @@ -export const SDK_VERSION = "1.5.0"; +export const SDK_VERSION = "1.6.0"; From 0121b00f2195ce2741e97e7d0cfd7cbffc6ed01d Mon Sep 17 00:00:00 2001 From: akash-vijay-kv Date: Fri, 17 Jul 2026 10:29:25 +0530 Subject: [PATCH 4/4] chore: Add CHANGELOG for v1.6.0 --- CHANGELOG.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index e2aad5a..af14836 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.6.0] - 2026-07-17 + +### Added + +- **Reparent Children of Blocked Root Instruments**: When a root-instrument span is blocked, its children are no longer dropped along with the subtree. Instead they are reparented onto the blocked span's parent (a true root's children become the new roots), and the peel repeats for promoted spans that are themselves blocked. +### Changed + +- **Standardized Attribute Serialization**: Introduced a single `SerializationSpanProcessor` that serializes and truncates every span attribute at write time, replacing the removed `AttributeSizeLimitProcessor`. Truncation uses a `jsonrepair`-based serializer for JSON values (and a `...[TRUNCATED]` suffix for plain strings) to stay aligned with backend behavior, and runs before the scrubbing layer so the regex scrubber sees values in string form. The per-attribute cap is read lazily from `NETRA_SPAN_ATTRIBUTE_MAX_SIZE` (default 30000). Serialization helpers were consolidated under `src/utils/serialization/`. + +### Fixed + +- **`addConversation` Persistence**: Fixed appending conversation entries to the active span's `conversation` attribute so existing entries are parsed and preserved (rather than overwritten) when a new entry is added, with defensive fallback when the stored value cannot be parsed. + ## [1.5.0] - 2026-07-10 ### Added