diff --git a/docs/content/docs/framework/tracing.en.mdx b/docs/content/docs/framework/tracing.en.mdx index a5e300a9a..1b7a16663 100644 --- a/docs/content/docs/framework/tracing.en.mdx +++ b/docs/content/docs/framework/tracing.en.mdx @@ -56,6 +56,30 @@ Once enabled, the IDs of the related tracing data are printed in the logs: For the full integration steps per platform, see [Observe on Volcengine](/en/docs/framework/ve-tracing). +## Limit local retention + +The following settings ship with this fix and are unavailable in the published 1.1.14 release and earlier versions + +`OpentelemetryTracer` retains up to 1024 recent spans with an estimated data budget of 16 MiB by default. It evicts the oldest spans when either limit is reached. A span exceeding the byte limit is skipped locally and still passed to cloud exporters according to the existing configuration + +Set these environment variables before starting the application to adjust local retention: + +| Environment variable | Type | Default | Description | +| --- | --- | --- | --- | +| `OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED` | Boolean | `true` | Whether to retain local span history | +| `OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_SPANS` | Nonnegative integer | `1024` | Maximum retained span count; `0` disables retention | +| `OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_BYTES` | Nonnegative integer | `16777216` | Estimated data budget in bytes; `0` disables retention | + +For applications that only need cloud observability, set this variable in the terminal used to start the application: + +```bash +export OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED=false +``` + +Disabling local retention preserves span annotations, trace IDs in logs, and configured cloud exports; `tracer.dump()` writes an empty array. With retention enabled, dumps contain only retained session history. If a span used to associate a trace with a session is evicted, that trace may also disappear from local dumps. Query your cloud observability platform for complete history + +Tracers using the same TracerProvider share these limits. Configuration takes effect when the cache is first created; restart the application or create a new provider after changing it. The byte budget estimates cached data and does not limit total process memory + ## Why Tracing With tracing in place, you can: diff --git a/docs/content/docs/framework/tracing.mdx b/docs/content/docs/framework/tracing.mdx index 696075724..a5adaf3fc 100644 --- a/docs/content/docs/framework/tracing.mdx +++ b/docs/content/docs/framework/tracing.mdx @@ -54,6 +54,30 @@ Tracing（链路追踪）对智能体执行过程进行**全链路记录**，是 针对不同平台的完整接入方式，参见[在火山引擎观测](/cn/docs/framework/ve-tracing)。 +## 限制本地缓存 + +以下配置随本次修复提供，已发布的 1.1.14 及更早版本不支持这些配置 + +`OpentelemetryTracer` 默认保留最近的 1024 个 Span，估算数据总量最多为 16 MiB。达到任一上限时会淘汰最早的 Span；单个 Span 超过字节上限时不写入本地缓存，仍按原有配置交给云端 Exporter + +在启动应用前，可通过以下环境变量调整本地缓存： + +| 环境变量 | 类型 | 默认值 | 说明 | +| --- | --- | --- | --- | +| `OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED` | 布尔值 | `true` | 是否保留本地 Span 历史 | +| `OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_SPANS` | 非负整数 | `1024` | 最多保留的 Span 数量，`0` 表示不保留 | +| `OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_BYTES` | 非负整数 | `16777216` | 估算数据量上限，单位为字节，`0` 表示不保留 | + +仅需云端观测时，在启动应用的终端设置： + +```bash +export OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED=false +``` + +关闭本地缓存后，Span 标注、日志中的 Trace ID 和已配置的云端上报保持可用，`tracer.dump()` 写入空数组。启用缓存时，`dump()` 只包含仍保留的会话历史；用于关联会话的 Span 被淘汰后，对应链路也可能不再出现在本地导出中，完整历史应在云端观测平台查询 + +同一个 TracerProvider 下的 tracer 共享上述额度。配置在首次创建缓存时生效，修改后需重启应用或创建新的 Provider；字节数是缓存数据的估算值，不代表整个进程的内存上限 + ## Tracing 的价值 引入 Tracing 后，你可以： diff --git a/prd-spec/bugfixes/tracing-memory/2026-10-13-bounded-cache.md b/prd-spec/bugfixes/tracing-memory/2026-10-13-bounded-cache.md new file mode 100644 index 000000000..bac593fb4 --- /dev/null +++ b/prd-spec/bugfixes/tracing-memory/2026-10-13-bounded-cache.md @@ -0,0 +1,73 @@ +# Bound the local tracing cache + +[中文版](2026-10-13-bounded-cache.zh.md) + +## Metadata + +- Change ID: `tracing-memory` +- Created/revised: 2026-10-13 +- Status: implemented +- Component: [tracing cache](../../../specs/tracing-cache/README.md) + +## Background, goals, and scope + +Published versions 1.0.9 and 1.1.14 append every sampled span to a process-lifetime list. Each tracer also registers another collector with the global provider. The session index grows independently, and `clear()` does not clear it. Large responses attached to repeatedly polled requests amplify retention. Source inspection confirms the defect; the proportion of any production process's RSS attributable to it is not yet measured. + +The fix bounds local retention without disabling cloud exporters or changing span annotation, sampling, or instrumentation suppression. It preserves recent session dumps and the latest trace ID. It does not change AgentKit SDK request payloads, frontend behavior, production configuration, package versions, or deployment. + +## Scenarios and requirements + +- FR-1: Given sustained exports, retain at most 1,024 spans and 16,777,216 estimated bytes by default; evict oldest spans until both limits hold. Reject a single oversized span from the local cache without discarding earlier valid entries or altering cloud export. +- FR-2: Given local caching disabled, retain no spans or session index, but keep the annotation processor, latest trace ID, and cloud pipeline operational. +- FR-3: Given eviction, clear, or shutdown, do not retain stale session mappings. Retrieval returns a consistent snapshot of retained spans; concurrent export/clear/read must be safe. +- FR-4: Given several tracers using one provider, register exactly one shared local collector at the front of its processor pipeline. Each provider has its own cache, with configuration captured when its first collector is created. +- FR-5: Publish validated configuration and document that local dumps contain only retained history; cloud retention is owned by the cloud exporter/service. + +## Design and contract impact + +Use a lock-protected FIFO deque with parallel per-span size accounting. Estimate the known span fields, attribute values, resource, events, links, and instrumentation metadata using Python object sizes without serializing or copying large payloads. This estimate is a cache budget, not an exact process RSS limit; count bounds metadata overhead as well. Skip oversized entries locally. Derive the session-to-trace index from retained `call_llm` spans on read, avoiding an independently growing index. A trace whose session-bearing span was evicted may no longer appear in a local session dump. + +Expose `inmemory_enabled`, `inmemory_max_spans`, and `inmemory_max_bytes` in `OpenTelemetryConfig`, with corresponding `OBSERVABILITY_OPENTELEMETRY_INMEMORY_*` variables. Defaults are true, 1024, and 16777216; nonnegative limits are required and either zero disables storage. Configuration follows existing startup settings/config.yaml handling; changes require a new provider/process. + +The provider owns a shared exporter, initialized under a registration lock. Keep the existing annotation processor even when retention is disabled. Export, read, clear, and shutdown share a cache lock; shutdown is terminal. `force_flush()` does not clear history. No background threads, timers, new dependencies, new network calls, or permission changes are introduced. + +Alternatives rejected: periodic clear leaves peak retention dependent on traffic and loses whole batches; count-only bounds do not adequately handle large responses; removing the processor would remove annotation and suppression behavior. Public tracer/exporter constructor signatures remain compatible. Local history changes from unlimited to bounded by design; no cloud span is truncated or filtered by this cache. + +## Tasks and acceptance + +| Requirement | Task | Acceptance | Verification | +| --- | --- | --- | --- | +| FR-1, FR-3 | T-1: add failing regression tests, then implement bounded cache in `inmemory_exporter.py` | AC-1: count/byte eviction, oversized payload, session cleanup, shutdown, and concurrent operations pass | `uv run --no-sync --extra dev pytest tests/test_inmemory_exporter.py -q` | +| FR-2, FR-4 | T-2: share the collector in `opentelemetry_tracer.py` | AC-2: multiple tracers share one collector; cloud spans, annotations, and trace IDs survive disabled/limited retention | `uv run --no-sync --extra dev pytest tests/test_tracing.py -q` | +| FR-5 | T-3: add config validation and bilingual user docs/spec | AC-3: defaults, overrides, zero, and invalid values are verified | `uv run --no-sync --extra dev pytest tests/config/test_tracing_config.py -q` | +| All | T-4: inspect diff, run focused tracing regressions and Ruff, reconcile both languages | AC-4: checks pass and limitations are recorded | See delivery record | + +## Risks and review + +Finite local history may truncate long session dumps. A shared cache covers the entire provider, so all tracers compete for the same budget and observe the same recent history. Python object estimates exclude allocator fragmentation and memory owned elsewhere; this does not promise an RSS ceiling. No production memory plateau or cloud behavior may be claimed from local tests alone. + +Design reviewed directly: ownership, errors, concurrency, compatibility, configuration, bilingual equivalence, and verification coverage have no open blockers. User approval: request on 2026-10-13 to fix the confirmed leak following the proposed bounded-cache approach. Complex tests and cloud operations require follow-up; neither is needed to implement the scoped regression fix. + +## Delivery record + +- Baseline: `b8038fb16`; verification date: 2026-10-13 +- Scope: bounded exporter, provider registration, configuration, regression tests, and bilingual cache documentation; T-1 through T-4 and AC-1 through AC-4 complete +- Regression before implementation: fail as expected, 20 failed and 3 passed for cache/config tests; includes old-span retention and stale session state +- Core checks: pass, 46 tests via `uv run --no-sync --extra dev pytest tests/test_inmemory_exporter.py tests/config/test_tracing_config.py tests/test_tracing.py -q` +- Focused compatibility checks: pass, 75 tests total including the core checks, content collection, LLM attribute extraction, runtime tracing, and Studio telemetry; one pre-existing invalid-escape warning +- Existing exporter fixtures also attempted metric delivery to `localhost:8000` during shutdown and logged `UNAVAILABLE`; the test process exited 0. New cache/cloud integration tests use a local in-memory stand-in and no cloud service +- Ruff format/check and `git diff --check`: pass for changed Python files and the patch +- Documentation: pass, both tracing MDX pages compile with `@mdx-js/mdx`; bilingual contract/configuration consistency and relative document links reviewed +- Pre-submission checks: pass, reran 75 focused tests after synchronizing `origin/main`; `uv run --no-sync --extra dev pre-commit run --all-files` passed, including Ruff and both secret scanners +- Full regression and documentation site build/type checks: not_run; this pass is limited to focused checks under the requested workflow, with no frontend application change +- Production/cloud validation: not_run, no production access or changes requested +- UI validation: not_applicable, no frontend behavior changes +- Commit and PR: authorized; package publication and production deployment: not_run, outside this scope + +## CI follow-up + +The initial Python 3.12 CI run had 7023 passing tests and two failures caused by runtime tests slicing the private `_spans` list. Update those two consumers to read session snapshots through `get_finished_spans(session_id)`, avoiding assumptions about the queue type or offsets that change after eviction. Preserve the existing span and session assertions. This follow-up changes no production code or approved contract + +- Failed cases: reproduced both failures locally before the update; both pass afterward +- Extended check: `uv run --no-sync --extra dev pytest tests/runtime/differential tests/runtime/piagent/test_piagent_turn_contract.py -q` finishes with 84 passed, 2 xfailed, and 1 failed after the update +- Remaining failure: `test_model_fallbacks_warn_for_external_runtime` sees empty captured logs; it also fails on the original PR commit `9671acd7`, where the same check finishes with 82 passed, 2 xfailed, and 3 failed. It is outside the cache fix and remains unchanged diff --git a/prd-spec/bugfixes/tracing-memory/2026-10-13-bounded-cache.zh.md b/prd-spec/bugfixes/tracing-memory/2026-10-13-bounded-cache.zh.md new file mode 100644 index 000000000..08bbac50c --- /dev/null +++ b/prd-spec/bugfixes/tracing-memory/2026-10-13-bounded-cache.zh.md @@ -0,0 +1,73 @@ +# 限制本地追踪缓存 + +[English](2026-10-13-bounded-cache.md) + +## 元数据 + +- 变更 ID：`tracing-memory` +- 创建及修订日期：2026-10-13 +- 状态：implemented +- 组件：[追踪缓存](../../../specs/tracing-cache/README.zh.md) + +## 背景、目标与范围 + +已发布的 1.0.9 和 1.1.14 会将每个被采样的 span 追加到进程生命周期内的列表中，每个追踪器还会向全局 provider 注册另一个收集器。会话索引独立增长，`clear()` 不会清理它。反复轮询请求携带的大响应会放大内存占用。源码检查已确认缺陷，尚未测量它在具体生产进程 RSS 中的占比 + +本次修复限制本地保留量，保留云端 exporter、span 字段补充、采样及埋点抑制行为，继续支持近期会话导出和最新 trace ID。不修改 AgentKit SDK 请求内容、前端行为、生产配置、包版本或部署 + +## 场景与需求 + +- FR-1：持续导出时，默认最多保留 1,024 个 span 和 16,777,216 估算字节，淘汰最旧 span 直到同时满足两个限制。单个 span 超限时仅跳过本地缓存，不丢弃之前的有效记录，也不改变云端上报 +- FR-2：关闭本地缓存时，不保留 span 或会话索引，仍保留字段补充处理器、最新 trace ID 和云端链路 +- FR-3：淘汰、清空或关闭后，不保留过期会话映射。查询返回保留数据的一致快照，并发导出、清理及读取安全 +- FR-4：同一 provider 的多个追踪器只在处理流水线首部注册一个共享收集器。不同 provider 拥有各自的缓存，首次创建收集器时确定配置 +- FR-5：提供经过校验的配置，明确本地导出只包含仍被保留的历史，云端保留策略由云端 exporter 或服务负责 + +## 设计与契约影响 + +使用加锁的 FIFO 双端队列和对应的逐 span 大小记录。通过 Python 对象大小估算已知 span 字段、属性值、资源、事件、链接和埋点元数据，不序列化或复制大块内容。该估算是缓存预算，并非进程 RSS 精确限制；数量限制同时约束元数据开销。超大记录仅跳过本地保存。读取时从仍被保留的 `call_llm` span 生成会话到 trace 的索引，避免独立索引持续增长。若包含会话信息的 span 已被淘汰，对应 trace 可能不再出现在本地会话导出中 + +在 `OpenTelemetryConfig` 中提供 `inmemory_enabled`、`inmemory_max_spans`、`inmemory_max_bytes`，对应 `OBSERVABILITY_OPENTELEMETRY_INMEMORY_*` 环境变量。默认值为 true、1024、16777216；限制值必须非负，任一限制为零即关闭保存。沿用已有启动 settings/config.yaml 处理方式，修改后需新建 provider 或重启进程 + +provider 拥有共享 exporter，通过注册锁初始化。即使关闭保存，仍保留现有字段补充处理器。导出、读取、清空和关闭共用缓存锁，关闭为终态。`force_flush()` 不清理历史。不引入后台线程、定时任务、新依赖、额外网络调用或权限变更 + +未选择的方案：定时清空的峰值仍由流量决定且会整批丢失历史；只限制数量不足以控制大响应；删除处理器会破坏字段补充与埋点抑制。公开 tracer/exporter 构造签名兼容。按设计将本地历史从无限保留改为有限保留，缓存不会截断或过滤任何云端 span + +## 任务与验收 + +| 需求 | 任务 | 验收 | 验证 | +| --- | --- | --- | --- | +| FR-1、FR-3 | T-1：先增加失败回归，再在 `inmemory_exporter.py` 实现有界缓存 | AC-1：数量及字节淘汰、超大内容、会话清理、关闭和并发操作通过 | `uv run --no-sync --extra dev pytest tests/test_inmemory_exporter.py -q` | +| FR-2、FR-4 | T-2：在 `opentelemetry_tracer.py` 共享收集器 | AC-2：多追踪器共用一个收集器，关闭或限制缓存仍保留云端 span、字段和 trace ID | `uv run --no-sync --extra dev pytest tests/test_tracing.py -q` | +| FR-5 | T-3：增加配置校验和双语用户文档及规范 | AC-3：验证默认值、覆盖、零值和非法值 | `uv run --no-sync --extra dev pytest tests/config/test_tracing_config.py -q` | +| 全部 | T-4：检查差异，运行针对性追踪回归及 Ruff，同步双语记录 | AC-4：检查通过，限制被如实记录 | 见交付记录 | + +## 风险与审查 + +有限本地历史可能导致长会话导出不完整。共享缓存覆盖整个 provider，各追踪器竞争同一预算并观察同一近期历史。Python 对象估算不包含分配器碎片和其他组件拥有的内存，不承诺 RSS 上限。不能用本地测试声称生产内存已稳定或云端行为已验证 + +已直接审查归属、错误、并发、兼容性、配置、双语一致性及验证覆盖，无待解决阻塞项。用户于 2026-10-13 在提出有界缓存方向后要求修复已确认的泄漏。复杂测试及云端操作需后续确认，本次范围内的回归修复不依赖二者 + +## 交付记录 + +- 基线：`b8038fb16`；验证日期：2026-10-13 +- 范围：有界 exporter、provider 注册、配置、回归测试和双语缓存文档；T-1 至 T-4、AC-1 至 AC-4 已完成 +- 实现前回归：按预期 fail，缓存及配置测试 20 项失败、3 项通过，包含旧 span 引用保留和会话状态残留 +- 核心检查：pass，`uv run --no-sync --extra dev pytest tests/test_inmemory_exporter.py tests/config/test_tracing_config.py tests/test_tracing.py -q` 共 46 项通过 +- 针对性兼容检查：pass，连同核心检查共 75 项通过，覆盖内容采集、LLM 属性提取、运行时 tracing 和 Studio 观测；出现一条原有非法转义警告 +- 原有 exporter fixture 在退出时尝试向 `localhost:8000` 发送指标并记录 `UNAVAILABLE`，测试进程退出码为 0；新增缓存及云端导出集成测试使用本地内存替身，不访问云服务 +- Ruff 格式及检查、`git diff --check`：pass，覆盖修改的 Python 文件和差异 +- 文档：pass，两份 tracing MDX 均通过 `@mdx-js/mdx` 编译，已审查双语契约及配置一致性和相对文档链接 +- 提交前检查：pass，同步 `origin/main` 后再次运行 75 项针对性测试，并通过 `uv run --no-sync --extra dev pre-commit run --all-files`，包含 Ruff 和两项密钥扫描 +- 全量回归、文档站构建及类型检查：not_run，按当前流程仅执行针对性检查，未修改前端应用 +- 生产及云端验证：not_run，未要求访问或修改生产环境 +- UI 验证：not_applicable，无前端行为变更 +- 提交及 PR：已获授权；软件包发布及生产部署：not_run，不在本次范围内 + +## CI 后续修正 + +Python 3.12 的首次 CI 有 7023 项通过、2 项失败；失败来自两个运行时测试对私有 `_spans` 列表的切片假设。将这两个使用方改为通过 `get_finished_spans(session_id)` 读取会话快照，避免依赖队列类型或会因淘汰而变化的列表偏移，保留原有 span 及会话断言。本次修正不修改生产代码或已批准契约 + +- 失败用例：修正前本地复现 2 项失败，修正后 2 项通过 +- 扩展检查：`uv run --no-sync --extra dev pytest tests/runtime/differential tests/runtime/piagent/test_piagent_turn_contract.py -q` 在修正后为 84 passed、2 xfailed、1 failed +- 剩余失败：`test_model_fallbacks_warn_for_external_runtime` 的日志捕获断言为空；原始 PR 提交 `9671acd7` 下同样复现，该基线为 82 passed、2 xfailed、3 failed。此项不在本次缓存修复范围，未修改 diff --git a/specs/tracing-cache/README.md b/specs/tracing-cache/README.md new file mode 100644 index 000000000..e8f170be2 --- /dev/null +++ b/specs/tracing-cache/README.md @@ -0,0 +1,29 @@ +# Local tracing cache + +[中文版](README.zh.md) + +Component ID: `tracing-cache` · Status: active · Revision: 2026-10-13 + +## Responsibility and interfaces + +Owns bounded diagnostic span retention in `veadk/tracing/telemetry/exporters/inmemory_exporter.py`, wired by `OpentelemetryTracer`. Cloud transport, SDK request instrumentation, and Studio storage are separate responsibilities. Consumers are `OpentelemetryTracer.dump()` and `trace_id`; constructor signatures remain compatible. + +## Contracts + +- CON-1: One cache per provider, shared by all VeADK tracers. Default limits are 1024 spans and 16777216 estimated bytes. Oldest entries are evicted first. Entries larger than the byte budget are not cached. Other exporters receive unchanged spans. +- CON-2: `OpenTelemetryConfig.inmemory_enabled` defaults to true. `inmemory_max_spans` and `inmemory_max_bytes` are nonnegative integers; either zero disables storage. Environment names are `OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED`, `OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_SPANS`, and `OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_BYTES`. Use existing startup settings and config.yaml-to-environment handling. Invalid values fail configuration validation. First collector creation captures configuration; changes require a new provider/process. +- CON-3: Disabled storage preserves the annotation processor, instrumentation suppression, latest trace ID, and existing cloud export/flush behavior. No retained span or session index remains. `dump()` writes an empty array when no matching history is retained. +- CON-4: The session index is derived from retained `call_llm` spans. Session dumps return retained spans with matching trace IDs in export order, including related spans without session attributes. Evicted session anchors can make a partial trace unavailable to session lookup. `clear()` resets spans, accounting, index, and latest trace ID. `shutdown()` clears data and permanently stops retention. Flush does not clear data. +- CON-5: Export/read/clear/shutdown are serialized by a cache lock; provider registration has its own lock. Reads return snapshots. There are no timers or new background tasks. Retention is process-local and not persistent. Limits cover estimated known span data, not exact RSS, allocator fragmentation, or other components' memory. + +Example: set `OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED=false` before startup to keep cloud telemetry while disabling local history. A bounded enabled cache retains recent diagnostics; use cloud storage for complete history. + +## Security, failures, and compatibility + +The cache introduces no credentials, permissions, external writes, or network operations. Existing content collection/redaction settings remain owned by instrumentation. Eviction changes only diagnostic history, not user sessions or cloud telemetry. Oversized records are skipped locally without logging their content. Finite history and provider-wide sharing replace previous unlimited per-tracer retention. No additional packages or unsupported Python features are required. + +## Verification and change record + +CON-1/4/5 map to `tests/test_inmemory_exporter.py`; CON-2 to `tests/config/test_tracing_config.py`; CON-1/3/5 integration to `tests/test_tracing.py`. Run these using `uv run --no-sync --extra dev pytest -q`, followed by focused tracing compatibility checks and Ruff. Production RSS requires separate workload evidence. + +Design and verification: [tracing-memory](../../prd-spec/bugfixes/tracing-memory/2026-10-13-bounded-cache.md). Contracts above are implemented and covered by the focused regression checks recorded in the linked design; package publication is separate. diff --git a/specs/tracing-cache/README.zh.md b/specs/tracing-cache/README.zh.md new file mode 100644 index 000000000..0f6e2a81a --- /dev/null +++ b/specs/tracing-cache/README.zh.md @@ -0,0 +1,29 @@ +# 本地追踪缓存 + +[English](README.md) + +组件 ID：`tracing-cache` · 状态：active · 修订日期：2026-10-13 + +## 职责与接口 + +负责 `veadk/tracing/telemetry/exporters/inmemory_exporter.py` 中有界的诊断 span 保存，由 `OpentelemetryTracer` 接入。云端传输、SDK 请求埋点和 Studio 存储属于其他组件。使用方为 `OpentelemetryTracer.dump()` 和 `trace_id`，构造签名保持兼容 + +## 契约 + +- CON-1：每个 provider 只有一个缓存，由所有 VeADK 追踪器共用。默认上限为 1024 个 span 和 16777216 估算字节，优先淘汰最旧记录，超过字节预算的单条记录不缓存。其他 exporter 接收未经本缓存修改的 span +- CON-2：`OpenTelemetryConfig.inmemory_enabled` 默认 true。`inmemory_max_spans` 和 `inmemory_max_bytes` 为非负整数，任一为零即关闭保存。环境变量为 `OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED`、`OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_SPANS`、`OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_BYTES`。使用现有启动 settings 及 config.yaml 到环境变量的处理方式，非法值导致配置校验失败。首次创建收集器时确定配置，修改后需新建 provider 或重启进程 +- CON-3：关闭保存时，保留字段补充处理器、埋点抑制、最新 trace ID 和现有云端导出及 flush 行为，不保留 span 或会话索引。没有对应历史时，`dump()` 写出空数组 +- CON-4：会话索引从仍被保留的 `call_llm` span 生成。会话导出按导出顺序返回 trace ID 匹配的保留 span，包括自身没有会话属性的关联 span。会话关联节点被淘汰后，部分 trace 可能无法再通过会话查询。`clear()` 重置 span、大小记录、索引和最新 trace ID。`shutdown()` 清空数据并永久停止保存，flush 不清空数据 +- CON-5：导出、读取、清空、关闭通过缓存锁串行处理，provider 注册使用独立锁。读取返回快照，无定时器或新增后台任务。缓存仅存在当前进程内，不持久化。预算覆盖已知 span 数据的估算量，不代表精确 RSS，不包含分配器碎片或其他组件内存 + +例如：启动前设置 `OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED=false`，保留云端观测并关闭本地历史。有界缓存用于保存近期诊断，完整历史应使用云端存储 + +## 安全、失败与兼容性 + +缓存不引入凭据、权限、外部写入或网络操作，现有内容采集及脱敏设置仍由埋点层负责。淘汰仅影响诊断历史，不影响用户会话或云端观测。超大记录只在本地跳过，不打印其内容。有限历史及 provider 内共享替代此前每个追踪器无限保存的行为。不增加依赖或不兼容的 Python 语法 + +## 验证与变更记录 + +CON-1/4/5 对应 `tests/test_inmemory_exporter.py`；CON-2 对应 `tests/config/test_tracing_config.py`；CON-1/3/5 集成对应 `tests/test_tracing.py`。使用 `uv run --no-sync --extra dev pytest -q` 运行，再进行针对性追踪兼容检查及 Ruff。生产 RSS 需要另行提供负载证据 + +设计与验证：[tracing-memory](../../prd-spec/bugfixes/tracing-memory/2026-10-13-bounded-cache.zh.md)。上述契约已实现，并通过所链接设计记录中的针对性回归检查；尚未发布软件包 diff --git a/tests/config/test_tracing_config.py b/tests/config/test_tracing_config.py index f90563040..9e7ab7185 100644 --- a/tests/config/test_tracing_config.py +++ b/tests/config/test_tracing_config.py @@ -12,7 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. -from veadk.configs.tracing_configs import APMPlusConfig +import pytest +from pydantic import ValidationError + +from veadk.configs.tracing_configs import APMPlusConfig, OpenTelemetryConfig def test_apmplus_endpoint_defaults_to_volcengine_region(monkeypatch): @@ -50,3 +53,34 @@ def test_apmplus_endpoint_environment_override_takes_precedence(monkeypatch): APMPlusConfig().otel_exporter_endpoint == "http://custom-apmplus.example.com:4317" ) + + +def test_inmemory_defaults_and_environment_overrides(monkeypatch): + for key in ("ENABLED", "MAX_SPANS", "MAX_BYTES"): + monkeypatch.delenv(f"OBSERVABILITY_OPENTELEMETRY_INMEMORY_{key}", raising=False) + defaults = OpenTelemetryConfig() + assert defaults.inmemory_enabled is True + assert defaults.inmemory_max_spans == 1024 + assert defaults.inmemory_max_bytes == 16 * 1024 * 1024 + monkeypatch.setenv("OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED", "false") + monkeypatch.setenv("OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_SPANS", "12") + monkeypatch.setenv("OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_BYTES", "8192") + config = OpenTelemetryConfig() + assert config.inmemory_enabled is False + assert config.inmemory_max_spans == 12 + assert config.inmemory_max_bytes == 8192 + + +@pytest.mark.parametrize("key", ["MAX_SPANS", "MAX_BYTES"]) +@pytest.mark.parametrize("value", ["-1", "invalid", "1.5"]) +def test_inmemory_invalid_limits_fail_configuration(monkeypatch, key, value): + monkeypatch.setenv(f"OBSERVABILITY_OPENTELEMETRY_INMEMORY_{key}", value) + with pytest.raises(ValidationError): + OpenTelemetryConfig() + + +def test_inmemory_zero_limits_are_valid(monkeypatch): + monkeypatch.setenv("OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_SPANS", "0") + monkeypatch.setenv("OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_BYTES", "0") + config = OpenTelemetryConfig() + assert config.inmemory_max_spans == config.inmemory_max_bytes == 0 diff --git a/tests/runtime/differential/conftest.py b/tests/runtime/differential/conftest.py index ad86b877c..d054842d8 100644 --- a/tests/runtime/differential/conftest.py +++ b/tests/runtime/differential/conftest.py @@ -307,14 +307,12 @@ async def run( session_id = session_id or f"session-{uuid.uuid4().hex[:8]}" tracer = None - spans_before = 0 if capture_spans: from veadk.tracing.telemetry.opentelemetry_tracer import OpentelemetryTracer # Attaches to the active provider (creating one only if none # exists), so the memoized proxy tracers keep resolving correctly. tracer = OpentelemetryTracer(exporters=[]) - spans_before = len(tracer._inmemory_exporter._exporter._spans) kwargs.setdefault("tracers", [tracer]) if arm == "codex": @@ -373,7 +371,7 @@ async def run( outcome.session_state = dict(getattr(session, "state", None) or {}) outcome.declared_usage = backend.declared_usage_total if tracer is not None: - spans = tracer._inmemory_exporter._exporter._spans[spans_before:] + spans = tracer._inmemory_exporter._exporter.get_finished_spans(session_id) outcome.span_names = frozenset(span.name for span in spans) if arm == "codex" and codex_fault is not None: @@ -439,9 +437,8 @@ def event_classifier() -> Any: # ``ProxyTracer`` objects that memoize the first real provider they resolve, so # installing a second provider mid-session silently sends every later span into # a detached (often already shut down) pipeline -- which looks exactly like "the -# runtime emits no spans". Each ``capture_spans`` row instead attaches its own -# ``InMemoryExporter`` to whichever provider is already active and reads back -# only the spans produced during its own run. +# runtime emits no spans". Each ``capture_spans`` row instead uses the active +# provider's shared cache and reads a snapshot for its unique session ID __all__ = [ diff --git a/tests/runtime/piagent/test_piagent_turn_contract.py b/tests/runtime/piagent/test_piagent_turn_contract.py index 50e862045..9f97663fc 100644 --- a/tests/runtime/piagent/test_piagent_turn_contract.py +++ b/tests/runtime/piagent/test_piagent_turn_contract.py @@ -350,7 +350,6 @@ async def test_turn_emits_an_indexable_call_llm_span(tmp_path, monkeypatch) -> N # provider they resolve, so replacing it makes later spans vanish). tracer = OpentelemetryTracer(exporters=[]) exporter = tracer._inmemory_exporter._exporter - before = len(exporter._spans) agent = _agent(tracers=[tracer]) session_id = f"session-{uuid.uuid4().hex[:8]}" @@ -370,7 +369,7 @@ async def test_turn_emits_an_indexable_call_llm_span(tmp_path, monkeypatch) -> N ] assert events, "the pi run produced no events at all" - spans = exporter._spans[before:] + spans = exporter.get_finished_spans(session_id) call_llm = [s for s in spans if s.name == "call_llm"] assert call_llm, ( f"no call_llm span for a Pi turn: {sorted({s.name for s in spans})}" diff --git a/tests/test_inmemory_exporter.py b/tests/test_inmemory_exporter.py new file mode 100644 index 000000000..2bb2f8291 --- /dev/null +++ b/tests/test_inmemory_exporter.py @@ -0,0 +1,169 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import gc +import weakref +from concurrent.futures import ThreadPoolExecutor + +import pytest +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import Event, ReadableSpan +from opentelemetry.trace import Link, SpanContext, TraceFlags + +from veadk.tracing.telemetry.exporters.inmemory_exporter import _InMemoryExporter + + +def make_span(number, *, session="session", trace_id=None, name="call_llm", **kwargs): + context = SpanContext( + trace_id=trace_id or number, + span_id=number, + is_remote=False, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + attributes = kwargs.pop("attributes", {"gen_ai.session.id": session}) + return ReadableSpan(name=name, context=context, attributes=attributes, **kwargs) + + +def test_default_retention_is_bounded_and_releases_old_spans(): + exporter = _InMemoryExporter() + old = make_span(1) + reference = weakref.ref(old) + exporter.export([old]) + del old + for number in range(2, 1200): + exporter.export([make_span(number)]) + gc.collect() + assert reference() is None + assert len(exporter.get_finished_spans("session")) == 1024 + assert len(exporter.session_trace_dict["session"]) == 1024 + + +def test_fifo_eviction_preserves_session_filtering_and_related_spans(): + exporter = _InMemoryExporter(max_spans=3) + exporter.export([make_span(1, session="old")]) + anchor = make_span(2, trace_id=20) + related = make_span(3, trace_id=20, name="invoke", attributes={}) + other = make_span(4, session="other") + exporter.export([anchor, related, other]) + assert "old" not in exporter.session_trace_dict + assert exporter.get_finished_spans("session") == [anchor, related] + snapshot = exporter.get_finished_spans("session") + snapshot.clear() + assert exporter.get_finished_spans("session") == [anchor, related] + assert exporter.get_finished_spans("other") == [other] + + +def test_session_index_is_deduplicated_and_disappears_with_evicted_anchors(): + exporter = _InMemoryExporter(max_spans=3) + exporter.export([make_span(n, trace_id=1) for n in range(1, 4)]) + assert exporter.session_trace_dict == {"session": [1]} + exporter.export([make_span(n, session="new") for n in range(4, 7)]) + assert "session" not in exporter.session_trace_dict + + +def test_byte_budget_evicts_large_responses_before_count_limit(): + exporter = _InMemoryExporter(max_spans=100, max_bytes=16 * 1024) + refs = [] + for number in range(1, 30): + span = make_span( + number, + attributes={ + "gen_ai.session.id": "session", + "gen_ai.output": str(number) + "x" * 6000, + }, + ) + refs.append(weakref.ref(span)) + exporter.export([span]) + del span + gc.collect() + retained = exporter.get_finished_spans("session") + assert 0 < len(retained) < 3 + assert refs[0]() is None + assert retained[-1].context.span_id == 29 + assert exporter._retained_bytes <= 16 * 1024 + + +@pytest.mark.parametrize("location", ["attributes", "events", "links", "resource"]) +def test_oversized_span_is_skipped_without_mutation_or_eviction(location): + exporter = _InMemoryExporter(max_bytes=16 * 1024) + previous = make_span(1) + exporter.export([previous]) + payload = "large" * 20000 + values = {"gen_ai.output": payload} + kwargs = { + "attributes": {"attributes": values}, + "events": {"events": [Event("event", attributes=values)]}, + "links": {"links": [Link(previous.context, attributes=values)]}, + "resource": {"resource": Resource(values)}, + }[location] + huge = make_span(2, **kwargs) + exporter.export([huge]) + assert list(exporter._spans) == [previous] + assert exporter.trace_id == 2 + if location == "attributes": + assert huge.attributes["gen_ai.output"] == payload + + +@pytest.mark.parametrize("limits", [{"max_spans": 0}, {"max_bytes": 0}]) +def test_zero_limit_preserves_latest_trace_id_without_retention(limits): + exporter = _InMemoryExporter(**limits) + exporter.export([make_span(1), make_span(2)]) + assert exporter.get_finished_spans("session") == [] + assert exporter.session_trace_dict == {} + assert exporter.trace_id == 2 + + +def test_clear_and_shutdown_release_all_cache_state(): + exporter = _InMemoryExporter() + exporter.export([make_span(1)]) + exporter.clear() + assert list(exporter._spans) == [] + assert exporter.session_trace_dict == {} + assert exporter.trace_id == "" + assert exporter._retained_bytes == 0 + exporter.export([make_span(2)]) + assert exporter.get_finished_spans("session") + exporter.shutdown() + exporter.export([make_span(3)]) + assert list(exporter._spans) == [] + assert exporter.session_trace_dict == {} + assert exporter.trace_id == "" + assert exporter._retained_bytes == 0 + + +def test_concurrent_export_read_and_clear_remain_consistent(): + exporter = _InMemoryExporter(max_spans=32, max_bytes=64 * 1024) + + def export_spans(): + for n in range(1, 500): + exporter.export([make_span(n)]) + + def read_spans(): + for _ in range(200): + assert len(exporter.get_finished_spans("session")) <= 32 + + def clear_spans(): + for _ in range(100): + exporter.clear() + + with ThreadPoolExecutor(max_workers=4) as pool: + futures = [ + pool.submit(fn) + for fn in (export_spans, export_spans, read_spans, clear_spans) + ] + for future in futures: + future.result() + assert len(exporter._spans) <= 32 + assert 0 <= exporter._retained_bytes <= 64 * 1024 + assert exporter._retained_bytes == sum(exporter._span_sizes) diff --git a/tests/test_tracing.py b/tests/test_tracing.py index 8c8c7caaa..4ec0b2145 100644 --- a/tests/test_tracing.py +++ b/tests/test_tracing.py @@ -13,9 +13,13 @@ # limitations under the License. +import json +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path from types import SimpleNamespace import pytest +from opentelemetry import context as context_api from opentelemetry import trace as trace_api from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( OTLPSpanExporter, @@ -28,6 +32,7 @@ from opentelemetry.util._once import Once from veadk.agent import Agent +from veadk.config import settings from veadk.tracing.telemetry import ( opentelemetry_tracer as opentelemetry_tracer_module, ) @@ -42,6 +47,7 @@ CozeloopExporter, CozeloopExporterConfig, ) +from veadk.tracing.telemetry.exporters.inmemory_exporter import InMemoryExporter from veadk.tracing.telemetry.exporters.tls_exporter import ( TLSExporter, TLSExporterConfig, @@ -53,6 +59,113 @@ SESSION_ID = "testsession" +def test_tracers_share_one_bounded_provider_cache( + fresh_global_tracer_provider, monkeypatch +): + monkeypatch.setattr(settings.opentelemetry_config, "inmemory_enabled", True) + monkeypatch.setattr(settings.opentelemetry_config, "inmemory_max_spans", 2) + first = OpentelemetryTracer() + provider = trace_api.get_tracer_provider() + tracers = [first] + [OpentelemetryTracer() for _ in range(20)] + assert len(provider._active_span_processor._span_processors) == 1 + assert all(t._inmemory_exporter is first._inmemory_exporter for t in tracers) + otel_tracer = provider.get_tracer(__name__) + for number in range(3): + with otel_tracer.start_as_current_span("call_llm") as span: + span.set_attribute("gen_ai.session.id", SESSION_ID) + span.set_attribute("number", number) + spans = first._inmemory_exporter._exporter.get_finished_spans(SESSION_ID) + assert [span.attributes["number"] for span in spans] == [1, 2] + assert all(t.trace_id == first.trace_id for t in tracers) + + +@pytest.mark.parametrize("enabled", [False, True], ids=["disabled", "bounded"]) +def test_local_cache_limits_preserve_cloud_spans_annotations_and_dumps( + fresh_global_tracer_provider, monkeypatch, tmp_path, enabled +): + monkeypatch.setattr(settings.opentelemetry_config, "inmemory_enabled", enabled) + monkeypatch.setattr(settings.opentelemetry_config, "inmemory_max_spans", 2) + monkeypatch.setattr(settings.opentelemetry_config, "inmemory_max_bytes", 16 * 1024) + monkeypatch.setattr( + opentelemetry_tracer_module, + "portal_metric_recorder", + SimpleNamespace(force_flush=lambda: True), + ) + provider = trace_sdk.TracerProvider() + cloud = OTelInMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(cloud)) + trace_api.set_tracer_provider(provider) + tracer = OpentelemetryTracer() + otel_tracer = provider.get_tracer(__name__) + payload = "large response" * 10000 + token = context_api.attach(context_api.Context()) + try: + with otel_tracer.start_as_current_span("invocation") as invocation: + assert ( + context_api.get_value("suppress_language_model_instrumentation") is True + ) + with otel_tracer.start_as_current_span("invoke_agent test"): + for number in range(3): + with otel_tracer.start_as_current_span("call_llm") as span: + span.set_attribute("gen_ai.session.id", SESSION_ID) + span.set_attribute("number", number) + if number == 2: + span.set_attribute("gen_ai.output", payload) + finally: + context_api.detach(token) + + spans = cloud.get_finished_spans() + assert len(spans) == 5 + assert spans[2].attributes["gen_ai.output"] == payload + assert spans[3].attributes["gen_ai.span.kind"] == "agent" + assert spans[4].attributes["gen_ai.span.kind"] == "workflow" + assert spans[4].attributes["gen_ai.usage.total_tokens"] == 0 + assert tracer.trace_id == format(invocation.context.trace_id, "032x") + cache = tracer._inmemory_exporter._exporter + assert len(cache._spans) == (2 if enabled else 0) + # The call_llm anchors have been evicted, so this session dump is empty + assert ( + json.loads( + Path(tracer.dump(session_id=SESSION_ID, path=str(tmp_path))).read_text() + ) + == [] + ) + if enabled: + with otel_tracer.start_as_current_span("call_llm") as span: + span.set_attribute("gen_ai.session.id", SESSION_ID) + data = json.loads( + Path(tracer.dump(session_id=SESSION_ID, path=str(tmp_path))).read_text() + ) + assert len(data) == 1 + assert data[0]["name"] == "call_llm" + provider.shutdown() + assert cache.get_finished_spans(SESSION_ID) == [] + assert cache.trace_id == "" + + +def test_provider_registration_is_atomic_and_provider_scoped(monkeypatch): + monkeypatch.setattr(settings.opentelemetry_config, "inmemory_enabled", True) + first = trace_sdk.TracerProvider() + second = trace_sdk.TracerProvider() + try: + with ThreadPoolExecutor(max_workers=8) as pool: + collectors = list(pool.map(InMemoryExporter.for_provider, [first] * 32)) + assert all(collector is collectors[0] for collector in collectors) + assert len(first._active_span_processor._span_processors) == 1 + other = InMemoryExporter.for_provider(second) + assert other is not collectors[0] + with first.get_tracer(__name__).start_as_current_span("call_llm") as span: + span.set_attribute("gen_ai.session.id", SESSION_ID) + assert len(collectors[0]._exporter.get_finished_spans(SESSION_ID)) == 1 + assert other._exporter.get_finished_spans(SESSION_ID) == [] + monkeypatch.setattr(settings.opentelemetry_config, "inmemory_enabled", False) + assert InMemoryExporter.for_provider(first) is collectors[0] + assert len(collectors[0]._exporter.get_finished_spans(SESSION_ID)) == 1 + finally: + first.shutdown() + second.shutdown() + + def init_exporters(): cozeloop_exporter = CozeloopExporter( config=CozeloopExporterConfig( diff --git a/veadk/configs/tracing_configs.py b/veadk/configs/tracing_configs.py index 7099b054f..3cbc1c9a5 100644 --- a/veadk/configs/tracing_configs.py +++ b/veadk/configs/tracing_configs.py @@ -45,6 +45,20 @@ class OpenTelemetryConfig(BaseSettings): default=True, alias="OBSERVABILITY_OPENTELEMETRY_TRACE_CONTENT", ) + inmemory_enabled: bool = Field( + default=True, + alias="OBSERVABILITY_OPENTELEMETRY_INMEMORY_ENABLED", + ) + inmemory_max_spans: int = Field( + default=1024, + ge=0, + alias="OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_SPANS", + ) + inmemory_max_bytes: int = Field( + default=16 * 1024 * 1024, + ge=0, + alias="OBSERVABILITY_OPENTELEMETRY_INMEMORY_MAX_BYTES", + ) class APMPlusConfig(BaseSettings): diff --git a/veadk/tracing/telemetry/exporters/inmemory_exporter.py b/veadk/tracing/telemetry/exporters/inmemory_exporter.py index e1f5e13cd..b15a87751 100644 --- a/veadk/tracing/telemetry/exporters/inmemory_exporter.py +++ b/veadk/tracing/telemetry/exporters/inmemory_exporter.py @@ -12,6 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. +import sys +from collections import deque +from collections.abc import Mapping +from threading import Lock from typing import Sequence from opentelemetry.context import ( @@ -20,74 +24,92 @@ detach, set_value, ) -from opentelemetry.sdk.trace import ReadableSpan, export +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider, export from typing_extensions import override from veadk.tracing.telemetry.exporters.base_exporter import BaseExporter from veadk.utils.logger import get_logger logger = get_logger(__name__) +_registration_lock = Lock() + + +def _value_size(value) -> int: + """Estimate OTel attribute data without copying or serializing large strings""" + size = sys.getsizeof(value) + if isinstance(value, Mapping): + size += sum(_value_size(k) + _value_size(v) for k, v in value.items()) + elif isinstance(value, (tuple, list)): + size += sum(_value_size(item) for item in value) + return size + + +def _span_size(span: ReadableSpan) -> int: + """Estimate retained span data; this is a cache budget, not an RSS measurement""" + # Include fixed span/container bookkeeping and all variable-size OTel fields + size = 1024 + _value_size(span.name) + _value_size(span.attributes) + for context in (span.context, span.parent): + if context: + size += _value_size(dict(context.trace_state.items())) + if span.status: + size += _value_size(span.status.description) + if span.resource: + size += _value_size(span.resource.attributes) + size += _value_size(span.resource.schema_url) + scope = span.instrumentation_scope + if scope: + size += _value_size((scope.name, scope.version, scope.schema_url)) + size += _value_size(scope.attributes) + for event in span.events or (): + size += 128 + _value_size(event.name) + _value_size(event.attributes) + for link in span.links or (): + size += 128 + _value_size(link.attributes) + size += _value_size(dict(link.context.trace_state.items())) + return size # ======== Adapted from Google ADK ======== class _InMemoryExporter(export.SpanExporter): - """Internal span exporter that stores spans in memory for local analysis and debugging. - - This exporter collects and stores OpenTelemetry spans in memory rather than - sending them to external services. It's particularly useful for development, - testing, and local trace analysis scenarios where immediate access to span - data is needed. - - Key Features: - - In-memory span storage with session-based organization - - Trace ID tracking for span correlation - - Session-to-trace mapping for efficient filtering - - Support for span retrieval by session ID - - Attributes: - _spans: List of all collected ReadableSpan objects - trace_id: Current trace identifier from the most recent span - session_trace_dict: Mapping of session IDs to their associated trace IDs - """ + """Bounded local span history, independent of cloud exporter delivery""" - def __init__(self) -> None: - """Initialize the in-memory exporter with empty storage containers.""" + def __init__( + self, max_spans: int = 1024, max_bytes: int = 16 * 1024 * 1024 + ) -> None: super().__init__() - self._spans = [] + if max_spans < 0 or max_bytes < 0: + raise ValueError("In-memory tracing limits must be nonnegative") + self._max_spans = max_spans + self._max_bytes = max_bytes + self._spans: deque[ReadableSpan] = deque() + self._span_sizes: deque[int] = deque() + self._retained_bytes = 0 + self._lock = Lock() + self._shutdown = False self.trace_id = "" - self.session_trace_dict = {} @override def export(self, spans: Sequence[ReadableSpan]) -> export.SpanExportResult: - """Export spans to in-memory storage with session tracking. - - Processes and stores spans while maintaining session-to-trace mapping - for efficient retrieval. Extracts session information from LLM call spans - to enable session-based filtering. - - Args: - spans: Sequence of ReadableSpan objects to store - - Returns: - SpanExportResult.SUCCESS: Always returns success for in-memory storage - """ - for span in spans: - if span.context: - self.trace_id = span.context.trace_id - else: - logger.warning( - f"Span context is missing, failed to get `trace_id`. span: {span}" - ) - - if span.name == "call_llm": - attributes = dict(span.attributes or {}) - session_id = attributes.get("gen_ai.session.id", None) - if session_id: - if session_id not in self.session_trace_dict: - self.session_trace_dict[session_id] = [self.trace_id] - else: - self.session_trace_dict[session_id] += [self.trace_id] - self._spans.extend(spans) + """Retain recent spans within both limits without changing their contents""" + with self._lock: + if self._shutdown: + return export.SpanExportResult.SUCCESS + for span in spans: + if span.context: + self.trace_id = span.context.trace_id + if not self._max_spans or not self._max_bytes: + continue + size = _span_size(span) + if size > self._max_bytes: + continue + while self._spans and ( + len(self._spans) >= self._max_spans + or self._retained_bytes + size > self._max_bytes + ): + self._spans.popleft() + self._retained_bytes -= self._span_sizes.popleft() + self._spans.append(span) + self._span_sizes.append(size) + self._retained_bytes += size return export.SpanExportResult.SUCCESS @override @@ -102,31 +124,49 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: """ return True - def get_finished_spans(self, session_id: str): - """Retrieve all spans associated with a specific session ID. - - Filters stored spans to return only those belonging to the specified - session, enabling session-scoped trace analysis and debugging. - - Args: - session_id: Session identifier to filter spans by - - Returns: - list[ReadableSpan]: List of spans associated with the session, - empty list if session not found or no spans available - """ - trace_ids = self.session_trace_dict.get(session_id, None) - if trace_ids is None or not trace_ids: - return [] - return [x for x in self._spans if x.context.trace_id in trace_ids] + def _session_traces(self) -> dict[str, dict[int, None]]: + """Build an index only from retained anchors while holding the cache lock""" + sessions: dict[str, dict[int, None]] = {} + for span in self._spans: + if span.name == "call_llm" and span.context: + session_id = (span.attributes or {}).get("gen_ai.session.id") + if isinstance(session_id, str) and session_id: + sessions.setdefault(session_id, {})[span.context.trace_id] = None + return sessions + + @property + def session_trace_dict(self) -> dict[str, list[int]]: + """Return a snapshot with no independently retained session metadata""" + with self._lock: + return {key: list(traces) for key, traces in self._session_traces().items()} + + def get_finished_spans(self, session_id: str) -> list[ReadableSpan]: + """Return retained session spans; evicted history is unavailable locally""" + with self._lock: + trace_ids = self._session_traces().get(session_id, {}) + return [ + span + for span in self._spans + if span.context and span.context.trace_id in trace_ids + ] def clear(self): - """Clear all stored spans and session mappings. + """Reset local history, accounting, session mappings, and the trace ID""" + with self._lock: + self._clear() - Removes all collected span data from memory, useful for cleanup - between test runs or to free memory in long-running processes. - """ + def _clear(self) -> None: self._spans.clear() + self._span_sizes.clear() + self._retained_bytes = 0 + self.trace_id = "" + + @override + def shutdown(self) -> None: + """Release history and stop retaining spans after provider shutdown""" + with self._lock: + self._shutdown = True + self._clear() class _InMemorySpanProcessor(export.SimpleSpanProcessor): @@ -236,7 +276,8 @@ class InMemoryExporter(BaseExporter): Note: - Cannot be added to exporter lists (validation prevents this) - Automatically managed by OpentelemetryTracer - - Memory usage grows with span count - consider periodic clearing + - Recent history is bounded by configured span and estimated byte limits + - Disabling retention preserves annotation and cloud export behavior - Session tracking requires properly configured session IDs """ @@ -250,5 +291,28 @@ def __init__(self, name: str = "inmemory_exporter") -> None: self.name = name - self._exporter = _InMemoryExporter() + from veadk.config import settings + + config = settings.opentelemetry_config + self._exporter = _InMemoryExporter( + max_spans=config.inmemory_max_spans if config.inmemory_enabled else 0, + max_bytes=config.inmemory_max_bytes, + ) self.processor = _InMemorySpanProcessor(self._exporter) + + @classmethod + def for_provider(cls, provider: TracerProvider) -> "InMemoryExporter": + """Reuse one annotation processor and bounded cache for each provider""" + with _registration_lock: + existing = getattr(provider, "_veadk_inmemory_exporter", None) + if existing is not None: + return existing + exporter = cls() + # Preserve first-processor annotation order without racing registration + processors = provider._active_span_processor + with processors._lock: + processors._span_processors = ( + exporter.processor, + ) + processors._span_processors + provider._veadk_inmemory_exporter = exporter + return exporter diff --git a/veadk/tracing/telemetry/opentelemetry_tracer.py b/veadk/tracing/telemetry/opentelemetry_tracer.py index 046f1b4ab..4575f485b 100644 --- a/veadk/tracing/telemetry/opentelemetry_tracer.py +++ b/veadk/tracing/telemetry/opentelemetry_tracer.py @@ -162,14 +162,8 @@ def _init_global_tracer_provider(self) -> None: for exporter in self.exporters: self._activate_exporter(exporter) - self._inmemory_exporter = InMemoryExporter() + self._inmemory_exporter = InMemoryExporter.for_provider(global_tracer_provider) if self._inmemory_exporter.processor: - # make sure the in memory exporter processor is added at index 0 - # because we use this to record all spans - global_tracer_provider._active_span_processor._span_processors = ( - self._inmemory_exporter.processor, - ) + global_tracer_provider._active_span_processor._span_processors - self._processors.append(self._inmemory_exporter.processor) self.exporters.append(self._inmemory_exporter) else: