From 797f196b265b9a05486512d05903e2ff5b8164a5 Mon Sep 17 00:00:00 2001 From: Peng Fan Date: Wed, 22 Apr 2026 13:35:37 -0700 Subject: [PATCH 1/3] Add OTLP payload byte-size chunking to Agent365Exporter A365 OTLP traces endpoint enforces a 1 MB request body limit. The exporter previously batched by span count (512) with no byte-level enforcement, so batches of gen-AI spans could exceed 1 MB and get rejected (403/502/500). This adds two layers of defense: - Heuristic span size estimator (estimateSpanBytes) with generous over-estimation - Byte-size chunking (chunkBySize) that splits batches into sub-batches under maxPayloadBytes (default 900 KB, configurable via Agent365ExporterOptions) - All-or-nothing semantics: if any chunk fails, the entire batch fails and retries Per-span truncation (250 KB cap) already existed; this complements it at the batch level. Co-Authored-By: Claude Opus 4.6 (1M context) --- .../agents-a365-observability/src/index.ts | 2 +- .../src/tracing/exporter/Agent365Exporter.ts | 105 +++++++++---- .../exporter/Agent365ExporterOptions.ts | 3 + .../src/tracing/exporter/utils.ts | 125 +++++++++++++++ .../core/payload-chunking.test.ts | 145 ++++++++++++++++++ 5 files changed, 349 insertions(+), 31 deletions(-) create mode 100644 tests/observability/core/payload-chunking.test.ts diff --git a/packages/agents-a365-observability/src/index.ts b/packages/agents-a365-observability/src/index.ts index 6931e03f..ffdbd5c8 100644 --- a/packages/agents-a365-observability/src/index.ts +++ b/packages/agents-a365-observability/src/index.ts @@ -86,7 +86,7 @@ export { safeSerializeToJson } from './tracing/util'; export { serializeMessages, normalizeInputMessages, normalizeOutputMessages } from './tracing/message-utils'; // Exporter utilities -export { isPerRequestExportEnabled, MAX_SPAN_SIZE_BYTES } from './tracing/exporter/utils'; +export { isPerRequestExportEnabled, MAX_SPAN_SIZE_BYTES, estimateSpanBytes, estimateValueBytes, chunkBySize } from './tracing/exporter/utils'; // Configuration export * from './configuration'; diff --git a/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts b/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts index 0fc42e56..33afbe6c 100644 --- a/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts +++ b/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts @@ -19,6 +19,8 @@ import { getAgent365ObservabilityDomainOverride, isPerRequestExportEnabled, truncateSpan, + estimateSpanBytes, + chunkBySize, } from './utils'; import { getExportToken } from '../context/token-context'; import logger, { formatError } from '../../utils/logging'; @@ -77,6 +79,13 @@ interface OTLPStatus { message?: string; } +interface MappedSpan { + span: OTLPSpan; + scopeKey: string; + scopeName: string; + scopeVersion?: string; +} + /** * Observability span exporter for Agent365: * - Partitions spans by (tenantId, agentId) @@ -167,8 +176,19 @@ export class Agent365Exporter implements SpanExporter { const startTime = Date.now(); - const payload = this.buildExportRequest(spans); - const body = JSON.stringify(payload); + // Map, truncate, and chunk spans by estimated byte size + const mappedSpans = this.mapAndTruncateSpans(spans); + const resourceAttrs = this.getResourceAttributes(spans); + const chunks = chunkBySize( + mappedSpans, + (ms) => estimateSpanBytes(ms.span), + this.options.maxPayloadBytes, + ); + + if (chunks.length > 1) { + logger.info(`[Agent365Exporter] Split ${spans.length} spans into ${chunks.length} chunks for tenantId: ${tenantId}, agentId: ${agentId}`); + } + // Select endpoint path based on S2S flag (includes tenantId in path) const servicePrefix = this.options.useS2SEndpoint ? '/observabilityService' : '/observability'; const endpointRelativePath = `${servicePrefix}/tenants/${encodeURIComponent(tenantId)}/otlp/agents/${encodeURIComponent(agentId)}/traces`; @@ -222,14 +242,25 @@ export class Agent365Exporter implements SpanExporter { // Always include tenant id header headers['x-ms-tenant-id'] = tenantId; - // Basic retry loop - const { ok, correlationId } = await this.postWithRetries(url, body, headers); - const duration = Date.now() - startTime; - if (!ok) { - logger.event(ExporterEventNames.EXPORT_GROUP, false, duration, undefined, { tenantId, agentId, correlationId }); - throw new Error('Failed to export spans'); + // Send each chunk (all-or-nothing: fail on first chunk failure) + let lastCorrelationId = 'unknown'; + for (let i = 0; i < chunks.length; i++) { + const chunk = chunks[i]; + const payload = this.buildEnvelope(chunk, resourceAttrs); + const body = JSON.stringify(payload); + logger.info(`[Agent365Exporter] Sending chunk ${i + 1} of ${chunks.length} (${chunk.length} spans, ${body.length} bytes)`); + + const { ok, correlationId } = await this.postWithRetries(url, body, headers); + lastCorrelationId = correlationId; + if (!ok) { + const duration = Date.now() - startTime; + logger.event(ExporterEventNames.EXPORT_GROUP, false, duration, `chunk ${i + 1} of ${chunks.length} failed`, { tenantId, agentId, correlationId }); + throw new Error(`Failed to export spans (chunk ${i + 1} of ${chunks.length})`); + } } - logger.event(ExporterEventNames.EXPORT_GROUP, true, duration, 'Spans exported successfully', { tenantId, agentId, correlationId }); + + const duration = Date.now() - startTime; + logger.event(ExporterEventNames.EXPORT_GROUP, true, duration, `${chunks.length} chunk(s) exported successfully`, { tenantId, agentId, correlationId: lastCorrelationId }); } /** @@ -289,42 +320,56 @@ export class Agent365Exporter implements SpanExporter { } /** - * Build OTLP export request payload + * Map ReadableSpans to OTLP format and apply per-span truncation. */ - private buildExportRequest(spans: ReadableSpan[]): OTLPExportRequest { - // Group by instrumentation scope (name, version) - const scopeMap = new Map(); - logger.info('[Agent365Exporter] Building OTLP export request payload'); - for (const sp of spans) { + private mapAndTruncateSpans(spans: ReadableSpan[]): MappedSpan[] { + logger.info('[Agent365Exporter] Mapping and truncating spans'); + return spans.map(sp => { const scope = sp.instrumentationScope || (sp as ReadableSpan & { instrumentationLibrary?: { name?: string; version?: string } }).instrumentationLibrary; - const scopeKey = `${scope?.name || 'unknown'}:${scope?.version || ''}`; + const scopeName = scope?.name || 'unknown'; + const scopeVersion = scope?.version || ''; + return { + span: truncateSpan(this.mapSpan(sp)), + scopeKey: `${scopeName}:${scopeVersion}`, + scopeName, + scopeVersion: scopeVersion || undefined, + }; + }); + } - const existing = scopeMap.get(scopeKey) || []; - existing.push(truncateSpan(this.mapSpan(sp))); - scopeMap.set(scopeKey, existing); + /** + * Extract resource attributes from the first span in the batch. + */ + private getResourceAttributes(spans: ReadableSpan[]): Record { + if (spans.length > 0 && spans[0].resource?.attributes) { + return { ...spans[0].resource.attributes }; + } + return {}; + } + + /** + * Build an OTLP export request envelope from pre-mapped spans. + */ + private buildEnvelope(mappedSpans: MappedSpan[], resourceAttrs: Record): OTLPExportRequest { + const scopeMap = new Map(); + for (const ms of mappedSpans) { + const existing = scopeMap.get(ms.scopeKey) || []; + existing.push(ms.span); + scopeMap.set(ms.scopeKey, existing); } const scopeSpans: ScopeSpan[] = []; - for (const [scopeKey, mappedSpans] of scopeMap) { + for (const [scopeKey, spans] of scopeMap) { const [name, version] = scopeKey.split(':'); scopeSpans.push({ scope: { name, version: version || undefined, }, - spans: mappedSpans, + spans, }); } - // Resource attributes (from the first span - all spans in a batch usually share resource) - let resourceAttrs: Record = {}; - if (spans.length > 0) { - const firstSpanResource = spans[0].resource?.attributes; - if (firstSpanResource) { - resourceAttrs = { ...firstSpanResource }; - } - } - return { resourceSpans: [ { diff --git a/packages/agents-a365-observability/src/tracing/exporter/Agent365ExporterOptions.ts b/packages/agents-a365-observability/src/tracing/exporter/Agent365ExporterOptions.ts index f99f9d09..828f84fa 100644 --- a/packages/agents-a365-observability/src/tracing/exporter/Agent365ExporterOptions.ts +++ b/packages/agents-a365-observability/src/tracing/exporter/Agent365ExporterOptions.ts @@ -52,4 +52,7 @@ export class Agent365ExporterOptions { /** Maximum number of spans per export batch. */ public maxExportBatchSize: number = 512; + + /** Upper bound on HTTP request body size in bytes. 100 KB headroom under the 1 MB server limit for estimator error and header overhead. */ + public maxPayloadBytes: number = 900_000; } diff --git a/packages/agents-a365-observability/src/tracing/exporter/utils.ts b/packages/agents-a365-observability/src/tracing/exporter/utils.ts index a3429dc4..7f760eaa 100644 --- a/packages/agents-a365-observability/src/tracing/exporter/utils.ts +++ b/packages/agents-a365-observability/src/tracing/exporter/utils.ts @@ -511,6 +511,131 @@ function runShrinkPhase( * priority, remeasuring after each step. * Phase 2 (fallback): replace remaining string attributes with overlimit sentinel. */ +// --------------------------------------------------------------------------- +// Span size estimation and byte-level chunking +// --------------------------------------------------------------------------- + +/** + * Overhead constant for OTLP JSON span fixed fields + * (traceId, spanId, parentSpanId, kind, timestamps, status, scope wrapper, etc.). + * Intentionally generous to account for serializer variance. + * @internal + */ +const SPAN_BASE_OVERHEAD = 2000; + +/** + * Overhead per attribute in OTLP JSON format. + * Covers key/value JSON wrapping overhead. + * @internal + */ +const ATTR_OVERHEAD = 80; + +/** + * Overhead per event in OTLP JSON. + * @internal + */ +const EVENT_OVERHEAD = 200; + +/** + * Estimate the serialized byte size of a single attribute value in OTLP/HTTP JSON. + */ +export function estimateValueBytes(value: unknown): number { + if (typeof value === 'string') { + return 40 + Buffer.byteLength(value, 'utf8'); + } + if (Array.isArray(value)) { + if (value.length === 0) return 60; + if (typeof value[0] === 'string') { + let sum = 60; + for (const s of value) { + sum += 40 + Buffer.byteLength(String(s), 'utf8'); + } + return sum; + } + return 60 + 50 * value.length; + } + return 40; // bool/int/float/null/other +} + +/** + * Heuristic estimator for the serialized size of an OTLP span in HTTP JSON. + * + * Uses generous constants tuned to over-estimate by ~25-50%, providing headroom + * for JSON serializer variance (whitespace, enum representation, integer-as-string). + */ +export function estimateSpanBytes(span: { + name?: string; + attributes?: Record | null; + events?: Array<{ name: string; attributes?: Record | null }> | null; +}): number { + let total = SPAN_BASE_OVERHEAD; + if (span.name) { + total += Buffer.byteLength(span.name, 'utf8'); + } + if (span.attributes) { + for (const [key, value] of Object.entries(span.attributes)) { + total += ATTR_OVERHEAD; + total += Buffer.byteLength(key, 'utf8'); + total += estimateValueBytes(value); + } + } + if (span.events) { + for (const ev of span.events) { + total += EVENT_OVERHEAD; + total += Buffer.byteLength(ev.name, 'utf8'); + if (ev.attributes) { + for (const [key, value] of Object.entries(ev.attributes)) { + total += ATTR_OVERHEAD; + total += Buffer.byteLength(key, 'utf8'); + total += estimateValueBytes(value); + } + } + } + } + return total; +} + +/** + * Split items into sub-batches whose cumulative estimated size stays under maxChunkBytes. + * + * Invariants: + * - Input order is preserved across chunks. + * - Empty input produces empty output. + * - A single item whose size exceeds maxChunkBytes forms its own single-item chunk + * (never silently dropped). + * - No chunk is ever empty. + * + * @param items The items to chunk. + * @param getSize Estimator function returning approximate serialized byte size of one item. + * @param maxChunkBytes Upper bound on cumulative estimated size per chunk. + */ +export function chunkBySize( + items: T[], + getSize: (item: T) => number, + maxChunkBytes: number, +): T[][] { + const chunks: T[][] = []; + let current: T[] = []; + let currentBytes = 0; + + for (const item of items) { + const itemBytes = getSize(item); + if (current.length > 0 && currentBytes + itemBytes > maxChunkBytes) { + chunks.push(current); + current = []; + currentBytes = 0; + } + current.push(item); + currentBytes += itemBytes; + } + + if (current.length > 0) { + chunks.push(current); + } + + return chunks; +} + export function truncateSpan(spanDict: T): T { try { let currentSize = getSerializedSize(spanDict); diff --git a/tests/observability/core/payload-chunking.test.ts b/tests/observability/core/payload-chunking.test.ts new file mode 100644 index 00000000..ba8d5029 --- /dev/null +++ b/tests/observability/core/payload-chunking.test.ts @@ -0,0 +1,145 @@ +// ------------------------------------------------------------------------------ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. +// ------------------------------------------------------------------------------ + +import { describe, it, expect } from '@jest/globals'; +import { + estimateSpanBytes, + estimateValueBytes, + chunkBySize, + truncateSpan, +} from '@microsoft/agents-a365-observability/src/tracing/exporter/utils'; + +function makeOTLPSpan(attrs: Record, name = 'test') { + return { + traceId: '00000000000000000000000000000001', + spanId: '0000000000000002', + name, + kind: 'INTERNAL', + startTimeUnixNano: 1000000000, + endTimeUnixNano: 2000000000, + attributes: attrs, + events: null, + links: null, + status: { code: 'UNSET', message: '' }, + }; +} + +// --------------------------------------------------------------------------- +// estimateSpanBytes +// --------------------------------------------------------------------------- + +describe('estimateSpanBytes', () => { + it('over-estimates relative to actual JSON size', () => { + const span = makeOTLPSpan({ + 'gen_ai.system': 'openai', + 'gen_ai.tool.arguments': 'x'.repeat(1000), + 'gen_ai.tool.call_result': 'y'.repeat(1000), + }); + const actual = Buffer.byteLength(JSON.stringify(span), 'utf8'); + expect(estimateSpanBytes(span)).toBeGreaterThanOrEqual(actual); + }); + + it('grows with attribute content', () => { + const small = makeOTLPSpan({ key: 'val' }); + const large = makeOTLPSpan({ key: 'x'.repeat(10000) }); + expect(estimateSpanBytes(large)).toBeGreaterThan(estimateSpanBytes(small)); + }); + + it('accounts for events', () => { + const withEvents = { ...makeOTLPSpan({}), events: [{ name: 'ev', attributes: { k: 'v' } }] }; + expect(estimateSpanBytes(withEvents)).toBeGreaterThan(estimateSpanBytes(makeOTLPSpan({}))); + }); +}); + +// --------------------------------------------------------------------------- +// estimateValueBytes +// --------------------------------------------------------------------------- + +describe('estimateValueBytes', () => { + it('handles all value types', () => { + expect(estimateValueBytes('hello')).toBe(40 + 5); + expect(estimateValueBytes([])).toBe(60); + expect(estimateValueBytes(['a', 'bb'])).toBe(60 + (40 + 1) + (40 + 2)); + expect(estimateValueBytes([1, 2])).toBe(60 + 50 * 2); + expect(estimateValueBytes(true)).toBe(40); + expect(estimateValueBytes(42)).toBe(40); + expect(estimateValueBytes(null)).toBe(40); + }); +}); + +// --------------------------------------------------------------------------- +// chunkBySize +// --------------------------------------------------------------------------- + +describe('chunkBySize', () => { + type SizedItem = { id: string; size: number }; + const sizedItem = (id: string, size: number): SizedItem => ({ id, size }); + const getSize = (item: SizedItem) => item.size; + + it('empty input returns empty output', () => { + expect(chunkBySize([], getSize, 900_000)).toEqual([]); + }); + + it('small items fit in one chunk', () => { + const items = Array.from({ length: 10 }, (_, i) => sizedItem(`s${i}`, 100)); + const chunks = chunkBySize(items, getSize, 900_000); + expect(chunks).toHaveLength(1); + expect(chunks[0]).toHaveLength(10); + }); + + it('splits when cumulative exceeds limit and preserves order', () => { + const items = Array.from({ length: 5 }, (_, i) => sizedItem(`s${i}`, 300_000)); + const chunks = chunkBySize(items, getSize, 900_000); + expect(chunks.length).toBeGreaterThanOrEqual(2); + expect(chunks.flat().map(i => i.id)).toEqual(['s0', 's1', 's2', 's3', 's4']); + }); + + it('oversize single item gets its own chunk', () => { + const chunks = chunkBySize([sizedItem('big', 2_000_000)], getSize, 900_000); + expect(chunks).toHaveLength(1); + expect(chunks[0][0].id).toBe('big'); + }); + + it('multi-item chunks respect limit; no chunk is empty', () => { + const items = Array.from({ length: 5 }, (_, i) => sizedItem(`s${i}`, 200_000)); + const chunks = chunkBySize(items, getSize, 500_000); + for (const chunk of chunks) { + expect(chunk.length).toBeGreaterThan(0); + if (chunk.length > 1) { + expect(chunk.reduce((s, i) => s + i.size, 0)).toBeLessThanOrEqual(500_000); + } + } + }); +}); + +// --------------------------------------------------------------------------- +// truncateSpan verification +// --------------------------------------------------------------------------- + +describe('truncateSpan verification', () => { + const MAX = 250 * 1024; + + it('leaves small span unchanged', () => { + const span = makeOTLPSpan({ k: 'small' }); + expect(truncateSpan(span).attributes!['k']).toBe('small'); + }); + + it('shrinks oversize span below limit, largest string first', () => { + const span = makeOTLPSpan({ small: 'ok', fat: 'x'.repeat(300_000) }); + const result = truncateSpan(span); + expect(Buffer.byteLength(JSON.stringify(result), 'utf8')).toBeLessThanOrEqual(MAX); + expect(result.attributes!['small']).toBe('ok'); + expect((result.attributes!['fat'] as string).length).toBeLessThan(300_000); + }); + + it('does not loop forever on already-sentinel values', () => { + const span = makeOTLPSpan({ + a: '[overlimit]', + b: '[overlimit]', + big: new Array(100000).fill(42), + }); + expect(truncateSpan(span)).toBeDefined(); + }); +}); From 7ee0d2bbec503412d4f7836529050b491a72f17e Mon Sep 17 00:00:00 2001 From: PengF <126631706+fpfp100@users.noreply.github.com> Date: Fri, 24 Apr 2026 11:37:20 -0700 Subject: [PATCH 2/3] Update packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../src/tracing/exporter/Agent365Exporter.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts b/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts index 33afbe6c..4a362967 100644 --- a/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts +++ b/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts @@ -248,7 +248,8 @@ export class Agent365Exporter implements SpanExporter { const chunk = chunks[i]; const payload = this.buildEnvelope(chunk, resourceAttrs); const body = JSON.stringify(payload); - logger.info(`[Agent365Exporter] Sending chunk ${i + 1} of ${chunks.length} (${chunk.length} spans, ${body.length} bytes)`); + const bodyBytes = Buffer.byteLength(body, 'utf8'); + logger.info(`[Agent365Exporter] Sending chunk ${i + 1} of ${chunks.length} (${chunk.length} spans, ${bodyBytes} bytes)`); const { ok, correlationId } = await this.postWithRetries(url, body, headers); lastCorrelationId = correlationId; From 7cc6ba9cc359e77c0fe68d1f75895d5c91487536 Mon Sep 17 00:00:00 2001 From: Peng Fan Date: Sun, 26 Apr 2026 21:49:06 -0700 Subject: [PATCH 3/3] Address PR review comments for payload chunking - Use Buffer.byteLength for accurate UTF-8 byte logging - Use MappedSpan fields instead of splitting scopeKey by ':' - Remove internal utils from public exports - Fix JSDoc wording (envelope overhead, not header overhead) - Reduce oversized test array from 100k to 5.2k elements Co-Authored-By: Claude Opus 4.6 (1M context) --- packages/agents-a365-observability/src/index.ts | 2 +- .../src/tracing/exporter/Agent365Exporter.ts | 6 +++--- .../src/tracing/exporter/Agent365ExporterOptions.ts | 2 +- tests/observability/core/payload-chunking.test.ts | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/packages/agents-a365-observability/src/index.ts b/packages/agents-a365-observability/src/index.ts index ffdbd5c8..6931e03f 100644 --- a/packages/agents-a365-observability/src/index.ts +++ b/packages/agents-a365-observability/src/index.ts @@ -86,7 +86,7 @@ export { safeSerializeToJson } from './tracing/util'; export { serializeMessages, normalizeInputMessages, normalizeOutputMessages } from './tracing/message-utils'; // Exporter utilities -export { isPerRequestExportEnabled, MAX_SPAN_SIZE_BYTES, estimateSpanBytes, estimateValueBytes, chunkBySize } from './tracing/exporter/utils'; +export { isPerRequestExportEnabled, MAX_SPAN_SIZE_BYTES } from './tracing/exporter/utils'; // Configuration export * from './configuration'; diff --git a/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts b/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts index 4a362967..fffc5648 100644 --- a/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts +++ b/packages/agents-a365-observability/src/tracing/exporter/Agent365Exporter.ts @@ -361,11 +361,11 @@ export class Agent365Exporter implements SpanExporter { const scopeSpans: ScopeSpan[] = []; for (const [scopeKey, spans] of scopeMap) { - const [name, version] = scopeKey.split(':'); + const representative = mappedSpans.find(ms => ms.scopeKey === scopeKey)!; scopeSpans.push({ scope: { - name, - version: version || undefined, + name: representative.scopeName, + version: representative.scopeVersion, }, spans, }); diff --git a/packages/agents-a365-observability/src/tracing/exporter/Agent365ExporterOptions.ts b/packages/agents-a365-observability/src/tracing/exporter/Agent365ExporterOptions.ts index 828f84fa..340f462a 100644 --- a/packages/agents-a365-observability/src/tracing/exporter/Agent365ExporterOptions.ts +++ b/packages/agents-a365-observability/src/tracing/exporter/Agent365ExporterOptions.ts @@ -53,6 +53,6 @@ export class Agent365ExporterOptions { /** Maximum number of spans per export batch. */ public maxExportBatchSize: number = 512; - /** Upper bound on HTTP request body size in bytes. 100 KB headroom under the 1 MB server limit for estimator error and header overhead. */ + /** Upper bound on HTTP request body size in bytes. 100 KB headroom under the 1 MB server limit for estimator error and JSON/envelope overhead (for example, resource attributes and scope wrappers). */ public maxPayloadBytes: number = 900_000; } diff --git a/tests/observability/core/payload-chunking.test.ts b/tests/observability/core/payload-chunking.test.ts index ba8d5029..788c688a 100644 --- a/tests/observability/core/payload-chunking.test.ts +++ b/tests/observability/core/payload-chunking.test.ts @@ -138,7 +138,7 @@ describe('truncateSpan verification', () => { const span = makeOTLPSpan({ a: '[overlimit]', b: '[overlimit]', - big: new Array(100000).fill(42), + big: new Array(5200).fill(42), }); expect(truncateSpan(span)).toBeDefined(); });