Add OTLP payload byte-size chunking to Agent365Exporter

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';

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<string, OTLPSpan[]>();
-    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<string, unknown> {
+    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<string, unknown>): OTLPExportRequest {
+    const scopeMap = new Map<string, OTLPSpan[]>();
+    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<string, unknown> = {};
-    if (spans.length > 0) {
-      const firstSpanResource = spans[0].resource?.attributes;
-      if (firstSpanResource) {
-        resourceAttrs = { ...firstSpanResource };
-      }
-    }
-
     return {
       resourceSpans: [
         {

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;
 }

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<string, unknown> | null;
+  events?: Array<{ name: string; attributes?: Record<string, unknown> | 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<T>(
+  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<T extends OTLPSpanLike>(spanDict: T): T {
   try {
     let currentSize = getSerializedSize(spanDict);