From 74e2bc1457bcb1046036ed1aee185b92e9751ab7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=83=E4=B9=98=E5=A6=8D=20=28Xiaoyaner=29?= <258399167+xiaoyaner0201@users.noreply.github.com> Date: Thu, 29 Oct 2026 01:41:45 +0800 Subject: [PATCH] fix(observability): bound business log cleanup and suppress unsafe SDK diagnostics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: 千乘妍 (Xiaoyaner) <258399167+xiaoyaner0201@users.noreply.github.com> --- docs/business-event-logs.md | 12 +- .../acceptance.md | 20 ++- .../add-otlp-business-event-logs/design.md | 6 +- .../specs/business-event-logs/spec.md | 25 +++- .../add-otlp-business-event-logs/tasks.md | 10 ++ .../business/BusinessLogsPipeline.java | 42 +++++-- .../src/main/resources/logback-spring.xml | 6 + .../business/BusinessLogsDiagnosticsTest.java | 117 ++++++++++++++++++ .../business/BusinessLogsHttpTest.java | 39 ++++++ .../business/BusinessLogsPipelineTest.java | 68 ++++++++++ .../business/BusinessLogsTestReceiver.java | 73 +++++++++++ .../business/BusinessLogsTransactionTest.java | 88 ++++++++++++- 12 files changed, 489 insertions(+), 17 deletions(-) create mode 100644 server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsDiagnosticsTest.java create mode 100644 server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsPipelineTest.java create mode 100644 server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTestReceiver.java diff --git a/docs/business-event-logs.md b/docs/business-event-logs.md index bb2ff56c6..95cf43c5f 100644 --- a/docs/business-event-logs.md +++ b/docs/business-event-logs.md @@ -20,7 +20,9 @@ Every row requires `skillhub.event.name`, `skillhub.skill.id`, `skillhub.namespa Publication covers all four producers: `SkillPublishService`, `SkillReviewSubmitService`, `ReviewService`, `PromotionService`. The reserved `builtin-skill-publisher` identity is classified as `system` even through the ordinary privileged auto-publication path (bootstrap verifies that it is a system account). Approval may correctly create TWO different facts: publication plus review/promotion approval. These are not duplicate records. Promotion approval's publication references the TARGET skill/version/namespace; its promotion fact references the SOURCE. Namespace filtering may therefore keep only one. / 发布覆盖四条生产路径；内置技能的保留系统身份即便走普通特权直发，也标记为 system，启动器会校验其系统账号属性；批准同时产生发布及审核/提升两个不同事实，不是重复。提升的发布事件指目标，提升决定事件指来源，空间过滤可能只保留其一。 -Timestamp is captured immediately before the producer publishes its domain event, not `publishedAt`, transaction commit time or async-send time. Transaction synchronization captures the record and valid span identity at event dispatch, admits only after commit, and never emits on rollback. Outside a transaction (notably the shared download service), admission is immediate. No entity lookup happens in the telemetry listener. / 时间是生产点发布领域事件前的时刻，不是提交或发送时刻；事务内仅提交后入队，回滚不发；无事务下载直接入队。 +Timestamp is captured immediately before the producer publishes its domain event, not `publishedAt`, transaction commit time or async-send time. Transaction synchronization captures the record and valid span identity at event dispatch, admits only after the enclosing physical transaction commits, and suppresses records when that physical transaction rolls back. Outside a transaction (notably the shared download service), admission is immediate. No entity lookup happens in the telemetry listener. / 时间是生产点发布领域事件前的时刻，不是提交或发送时刻；事务内仅物理事务提交后入队，物理事务回滚不发；无事务下载直接入队。 + +Savepoint exclusion: rollback suppression does NOT cover `PROPAGATION_NESTED` savepoint rollback. Its synchronization belongs to the enclosing physical transaction: a nested event can still export when the outer transaction commits, even though the nested database write was rolled back. Do not use this pipeline as evidence of savepoint-level persistence; producers requiring that guarantee must not dispatch from nested/savepoint work. No current included production path configures NESTED propagation. `REQUIRES_NEW` is a separate physical transaction: its committed event exports even if the suspended outer transaction later rolls back; the outer event is suppressed. / 保存点例外：不保证 `PROPAGATION_NESTED` 回滚时抑制事件；外层物理事务提交仍可导出已回滚保存点内的事件。需要该保证的生产点不得在保存点内发布；当前纳入路径未配置 NESTED。`REQUIRES_NEW` 独立提交不受外层之后回滚影响。 Download means prepared, NOT delivered, complete, installed or executed. Web/v1/CLI share the producer; authorized previews of unpublished versions and review packages do not emit. Compat redirects alone, Suite plans, resolve/detail polling, direct object-storage access, URL reuse and client caches are not new registry download-prepared facts. / 下载准备不等于送达、传输完成、安装或执行；未发布预览、审核包、套装计划、查询轮询不计入。对象存储直取、URL 重用及缓存可绕过注册表。 @@ -127,6 +129,8 @@ Collector is optional. Production operators choose exporters, retention, TLS/aut Business threads only map fixed bounded fields, capture context, register a transaction callback and attempt nonwaiting SDK admission. No DB enrichment or network IO is performed by the adapter on business threads. Standard BatchLogRecordProcessor handles a bounded record queue; a conservative admission permit counts queued PLUS in-flight records to expose exact queue-full drops without SDK internals. Only SDK exporters implement network retry (retryable HTTP status/IO); no caller-runs fallback. Batch export budget is `(timeout + retry-max-backoff) * retry-max-attempts + 1s`; receiver `Retry-After` and sender behavior still depend on the managed SDK. / 业务线程仅做固定字段映射及非等待入队，不发送网络请求；队列预算包含在途，重试交标准 SDK。 +Each export has an owned completion guard: SDK completion OR the export-budget deadline, whichever wins, releases admission permits exactly once. An abandoned SDK result (including 1.31's truncated non-2xx response-body callback failure) becomes one failed batch at the deadline. Late SDK success/failure cannot add permits or change counters. Deadlines run on one daemon scheduler, canceled tasks are removed, and closing fails outstanding guards and stops the scheduler. This bounds local accounting, not remote cancellation or delivery: a timed-out request might still reach the receiver. / 每批导出由 SDK 完成或预算到期中先发生者结束，许可只释放一次；SDK 遗失完成信号也会计一次失败，迟到结果不重复释放或计数。关闭时结束守卫并停止调度器；本地超时不代表远端已撤销。 + Loss is expected on commit-to-listener/admission gaps, crashes, forced exits, queue saturation, timeout or exhausted retry, and incomplete shutdown. No disk spool, outbox, replay, backfill or exactly-once. Ambiguous acknowledgement can duplicate the same UID; deduplicate by `log.record.uid`, not `(skill,version)` (multiple legitimate downloads), request ID or trace ID. Mapping a new domain-event dispatch creates a new UID; idempotency of repeated business operations is unchanged. / 提交到入队间隙、崩溃、满队列及重试耗尽都可能漏报；超时重试可能重复，接收方按 UID 去重，不按技能版本或 Trace 去重。 The queue budget starts at after-commit admission. Before commit, transaction synchronizations retain one bounded mapped snapshot per dispatched event until completion, outside that SDK budget. Avoid unbounded bulk work in one outer transaction; this configuration is not a process-wide heap cap. / 队列预算从提交后接纳开始；提交前每个事件的有界快照由事务回调保留，不计入 SDK 预算。不要在单个外层事务里执行无界批量操作，这不是整个进程的堆内存上限。 @@ -143,10 +147,12 @@ Micrometer names on the existing protected Actuator metrics surface (Prometheus | `skillhub.business.logs.dropped{reason=queue_full}` | Records refused because outstanding budget is full / 满队列丢弃 | | `skillhub.business.logs.dropped{reason=shutdown}` | New records refused during closing / 退出时拒绝新记录 | | `skillhub.business.logs.errors` | Adapter mapping/admission/lifecycle exceptions, swallowed / 适配器异常（不传播） | -| `skillhub.business.logs.export.failures` | Batches whose SDK export fails/throws after retries / 最终失败批次 | +| `skillhub.business.logs.export.failures` | Batches whose SDK export fails/throws, exceeds the export budget, or is still outstanding at close / 最终失败、超时或关闭时仍未结束的批次 | | `skillhub.business.logs.exported` | Records in SDK-successful batches; may include receiver partial rejections / SDK 成功记录，含部分拒收盲区 | -No per-user/skill/namespace metric tags. Adapter diagnostics are fixed text, at most once per minute per pipeline, without exception messages, headers or event payloads; SDK diagnostics use its own throttling. / 指标不带高基数用户/资源标签，适配器诊断每分钟最多一次固定文字。 +No per-user/skill/namespace metric tags. Adapter diagnostics are fixed text, at most once per minute per pipeline, without exception messages, headers or event payloads: `Business Logs telemetry failed; inspect skillhub.business.logs metrics and receiver health`. Shipped `logback-spring.xml` sets `io.opentelemetry.exporter` and `io.opentelemetry.sdk.logs` to OFF by default, including their child loggers routed from JUL by Spring Boot. Raw SDK diagnostics can echo receiver error bodies or transport exception text containing credentials; throttling or ERROR level alone is NOT safe. This also silences shared OTel exporter diagnostics for Trace; Trace export behavior is unchanged. Custom logging configurations must retain these OFF rules. / 指标无高基数标签；适配器每分钟最多一次固定文字。默认关闭 OTel exporter 与 sdk.logs 原始诊断（包括子记录器及 JUL 桥接），其响应或异常文本可能回显凭证，ERROR 级别也不安全；同时静默共享 Trace exporter 诊断但不改变导出。自定义日志配置需保留这些规则。 + +Troubleshoot using the protected Actuator counters above, this fixed warning, receiver-side status/health, and independent DNS/network/TLS/auth checks with secret-safe tools. If raw SDK diagnostics are indispensable, first remove production credentials or reproduce with synthetic credentials in an isolated environment, restrict log access/retention, and knowingly accept disclosure risk. Temporarily set the exact Spring property `logging.level.io.opentelemetry.exporter.internal.http.HttpExporter=DEBUG` (e.g. a deployment-owned config file or `SPRING_APPLICATION_JSON`); `logging.level.io.opentelemetry.sdk.logs=DEBUG` opts into Logs processor diagnostics separately. Logger names are case-sensitive; do not rely on relaxed uppercase env names for the class logger. Restore both package defaults to OFF and remove any child-logger override after diagnosis, then restart; do not share unredacted SDK output. Raising root logging alone does not override explicit package OFF. / 排障优先使用受保护指标、固定告警及接收端健康，不输出 Secret。确需 SDK 原始细节时，在隔离环境用假凭证复现，限制访问与保留并明确接受泄露风险；临时提高精确日志属性（类名区分大小写），结束后恢复 OFF、删除子记录器覆盖并重启，勿分享未脱敏输出。 | Symptom / 现象 | Check / 排查 | | --- | --- | diff --git a/openspec/changes/add-otlp-business-event-logs/acceptance.md b/openspec/changes/add-otlp-business-event-logs/acceptance.md index bfac53fb7..798a62085 100644 --- a/openspec/changes/add-otlp-business-event-logs/acceptance.md +++ b/openspec/changes/add-otlp-business-event-logs/acceptance.md @@ -4,7 +4,7 @@ Base: `9ebb99226dd1b3e1d4fd8dd3e36baae70a4ed273`; branch `feat/otlp-business-eve ## Event/producer matrix -`BusinessLogsProducerPostgresTest` uses actual Spring service proxies, real JPA repositories, Flyway migrations and PostgreSQL 16. Storage/scanner/avatar/notification/audit/compensation collaborators and package-validation/metadata-parsing collaborators are isolated mocks; business state and transaction boundaries are real. Each row has a committed scenario and a rollback scenario. Rollback emits zero records. / 服务代理、仓储、迁移和事务为真实依赖；存储、扫描、头像、通知、审计、补偿及包校验/解析协作组件用 mock 隔离，不宣称验证它们的端到端副作用。 +`BusinessLogsProducerPostgresTest` uses actual Spring service proxies, real JPA repositories, Flyway migrations and PostgreSQL 16. Storage/scanner/avatar/notification/audit/compensation collaborators and package-validation/metadata-parsing collaborators are isolated mocks; business state and transaction boundaries are real. Each row has a committed scenario and a physical-transaction rollback scenario. Physical rollback emits zero records; `PROPAGATION_NESTED` savepoint rollback is excluded (it can export after outer commit). `REQUIRES_NEW` commit exports independently of outer rollback. / 服务代理、仓储、迁移和物理事务为真实依赖；物理事务回滚不导出，但保存点回滚明确不保证抑制；独立新事务提交不受外层回滚影响。外部协作组件用 mock 隔离，不宣称验证它们的端到端副作用。 | Producer path | Committed facts (exactly) | Rollback | Actor | | --- | --- | --- | --- | @@ -32,7 +32,7 @@ Result: 39 real-service/JPA scenarios passed in `final-focused-3.log`: the 26 co | Mapper boundary | Exact fixed keys; no rejection reason/submitter/unknown event serialization; namespace/event allowlists; anonymous omission; system actor; length limits; reserved built-in publisher | PASS `BusinessEventMapperTest` | | Default off | No pipeline/listener/provider bean; no default receiver | PASS `BusinessLogsConfigurationTest` | | Invalid configuration | Enabled+empty URL rejected; metadata/userinfo/query/file URL and bad budgets/event names rejected without echoing supplied URL; URL-encoded headers and control-character rejection | PASS `BusinessLogsConfigurationTest` | -| Capture before commit | Valid unsampled span survives scope end before PostgreSQL commit; implementation reads standard SpanContext only, not request/MDC strings | PASS `BusinessLogsTransactionTest`; mapper/listener source boundary | +| Capture before commit | Valid unsampled span survives scope end before PostgreSQL commit; Repair 1 additionally seeds distinct MDC/Baggage secrets and checks exported attributes/body/resource | PASS `BusinessLogsTransactionTest`; see Repair 1 evidence below | | Trace coexistence | Actual production configuration exports 9 records each with none, otel-sdk (sampling=0), external-agent mode | PASS, actual Collector output 27 records / 9 names / 27 unique UIDs | | Actual OTLP receive/query | Pinned Collector contrib 0.111.0 file exporter; documented jq query run against real received JSON | PASS `collector-received.jsonl`, `collector-query.tsv`, `collector-query-summary.json` | | Queue full, deterministic | 1000 caller events with blocked exporter; exactly 999 drops; bounded shutdown | PASS final focused run: 12 ms caller loop, 169 ms close for 100 ms wait budget (includes scheduling/cleanup overhead) | @@ -49,6 +49,22 @@ Result: 39 real-service/JPA scenarios passed in `final-focused-3.log`: the 26 co The raw measurement values above are from the named focused run, not promised production latency percentiles. Shutdown has a bounded SDK wait budget plus local cleanup/scheduling overhead. / 延迟数字是该次测试实测，不是生产百分位承诺；退出等待预算之外有本地清理和调度开销。 +## Repair 1 — independent review RETURN + +Repair base: `894e5860ce19059a27dab8cf94642aba56d33a29`; changes remain uncommitted. Evidence filenames in this section are relative to `evidence/repair1/` in the worker report, not the earlier evidence directory. The original results above remain historical; the current focused result is **299 tests, 0 failures, 0 errors, 0 skips** (193 domain + 106 app; 73 business-Logs tests). The same full focused selector/environment was rerun, including the new `Business*Test` classes. / 本轮在独立评审退回后修复，未提交；本节路径属于 repair1 子目录。之前结果保留为历史，当前定向结果为 299 项全部通过，其中业务日志 73 项。 + +| Finding / boundary | Executed assertion and change | RED → GREEN evidence | +| --- | --- | --- | +| Abandoned SDK result / permit leak | Production exporter receives HTTP 400, declares 100 body bytes but sends one and closes; export-budget guard fails once, then two HTTP 200 exports succeed. Requests=3, enqueued=3, exported=2, failures=1, drops=0; no restart | RED `red-permit-http.log` (1 test, failure-counter await times out at zero); GREEN `green-repair-focused.log` and `final-focused-green.log`, `BusinessLogsHttpTest` | +| Exactly-once cleanup | Deadline wins over an unfinished result; a second export occupies the recovered capacity; late SDK success OR failure cannot admit a third record or alter the failed outcome. Two parameter cases pass | RED `red-guard-diagnostics-boundaries.log` (2 guard timeouts); GREEN `final-focused-green.log`, `BusinessLogsPipelineTest` | +| SDK credential disclosure | Real 401 protobuf Status echoes synthetic Authorization; real malformed HTTP status line produces a credential-bearing ProtocolException. Shipped Logback package OFF rules suppress raw SDK output; fixed warning + failure counter remain. Captured stdout/stderr checked in BOTH text and async JSON modes; no synthetic credential | RED `red-guard-diagnostics-boundaries.log` (2 log assertions fail with credential output); GREEN `final-focused-green.log`, 4 `BusinessLogsDiagnosticsTest` cases | +| Savepoint contract | Explicitly exclude `PROPAGATION_NESTED` rollback suppression in operator guide/design/spec/acceptance. Real PostgreSQL: nested rollback + outer commit leaves 0 rows but exports 1 record; `REQUIRES_NEW` commit + outer rollback leaves only the independent row and exports only its record. A documentation consistency test covers all four contract files | RED `red-guard-diagnostics-boundaries.log` (documentation assertion fails; both runtime boundary tests already pass); GREEN `final-focused-green.log`, `BusinessLogsTransactionTest` | +| Non-vacuous MDC/Baggage exclusion | Assert distinct sensitive keys/values are actually present at dispatch, retain unsampled span across scope end, then assert neither keys nor values in exported attributes, body or resource | RED `red-guard-diagnostics-boundaries.log` (fixture precondition fails on missing MDC sentinel, not a product disclosure); GREEN `final-focused-green.log`, `BusinessLogsTransactionTest` | +| Deployment regression | New deployment contracts 4/4, no skips; existing Helm configuration and release validation regression suites both pass; no Helm/Compose setting changed | `deployment-business-logs.log`, `deployment-helm-contracts.log`, `deployment-release-contracts.log` | +| Live receiver regression | Real pinned Collector receives 27 records / 9 names / 27 distinct UIDs again. Paused receiver: 1000 caller events in 11 ms, 998 drops; stopped receiver: caller 0 ms, one failure. Blocked exporter: 12 ms caller loop, 999 drops, close 216 ms for 100 ms wait budget plus overhead | `final-focused-green.log`, current `collector-received.jsonl`; recorded values are observations, not SLOs | + +The first full rerun (`final-focused.log`) executed 299 tests but had 4 diagnostic-test setup errors: querying an uncreated Logback child returned null. The fixture now creates the child before checking its inherited OFF level; the subsequent complete rerun above is green. `green-repair-focused.log` is an earlier 15-test pass before extending diagnostics to both log formats; it is not the final count. No broad-backend/base comparison or docs-site build was rerun in Repair 1; the limitations below are unchanged. / 首次全定向检查中诊断测试的子 logger 尚未创建，4 项 setup 报错；修正测试 fixture 后完整重跑通过。不将中间失败或早期 15 项绿色记录冒充最终结果；本轮未重跑全量后端/基线对比或文档站构建。 + ## Broad suite and environment limits Broad `make test-backend` runs are NOT green. Initial candidate: 2732 tests, 2 failures, 1 error, 30 skips. Final candidate broad run: 2762 tests, 3 failures, 11 errors, 30 skips. Frozen base: 2712 tests, 2 failures, 2 errors, 30 skips. Failures are in GitHub integration tests; exact failing methods are retained in the count JSONs. Four methods occur in both the final candidate's and base's broad failure sets (Go executor, credential privacy, two budget cases), but the full failure sets and error modes vary. Do NOT infer that every candidate failure was proven pre-existing. Subsequent identical five-class GitHub focused commands PASS all 37 tests on BOTH the candidate and frozen base. An independent SQL/host clock probe found the Docker database clock ahead of the host in every sample, consistent with timing-sensitive eligibility checks, but it is NOT asserted as the proven cause of every failure. No clock/global configuration was changed. The authorized detached base worktree was removed. / 全量检查不是绿色；基线与 candidate 有四个相同失败方法，但集合及错误形式不同，不冒称全部故障已证实为基线问题。随后相同五类 GitHub 定向测试在两边均 37 项通过。独立探测发现 Docker 数据库时钟领先宿主机，与时间筛选类故障相符，但未证明所有故障根因；未改系统时钟/全局配置。基线临时工作树已移除。 diff --git a/openspec/changes/add-otlp-business-event-logs/design.md b/openspec/changes/add-otlp-business-event-logs/design.md index 5a5ffd3e0..b20789074 100644 --- a/openspec/changes/add-otlp-business-event-logs/design.md +++ b/openspec/changes/add-otlp-business-event-logs/design.md @@ -8,11 +8,15 @@ The conditional app configuration owns a dedicated `SdkLoggerProvider`, BatchLog A synchronous Spring `@EventListener` maps fixed scalar data and captures only valid OTel SpanContext when the producer dispatches. It registers `TransactionSynchronization.afterCommit` for actual transactions, otherwise admits immediately. This is equivalent to AFTER_COMMIT + fallbackExecution but preserves the producer's span even if the span ends before commit. Each boundary catches telemetry RuntimeExceptions and counts them with bounded fixed-text diagnostics. No network IO, enrichment queries or space-waiting occurs in the listener. Disabled configuration creates no listener, provider, worker, exporter or OTel context objects. +The commit/rollback contract is physical-transaction scoped, not savepoint aware. `PROPAGATION_NESTED` savepoint rollback is explicitly excluded: an event dispatched inside the rolled-back savepoint still exports if the outer physical transaction commits. No current included producer configures NESTED. Producers needing savepoint-level suppression must not dispatch within nested/savepoint work. `REQUIRES_NEW` commits independently and exports even when the suspended outer transaction rolls back, suppressing only the outer event. PostgreSQL boundary tests assert both database rows and exact exported records; this is an honest exclusion, not new savepoint support. + Occurrence is `Instant.now()` immediately before producer publication, not the independently clocked persisted publishedAt or commit time. Sources identify domain branch, not caller-controlled Web/CLI labels. Forced internal auto-publication is a system actor; privileged human auto-publication remains a human actor. Anonymous downloads have actor type user and no user.id (user means user-request origin, not proof of an identified human). Review and promotion use the actual approver, not owner/uploader. Promotion decisions use source skill/namespace; effective publication uses the derived target. ## Bounds and observability -Standard SDK batching is retained. A nonwaiting semaphore around processor admission conservatively bounds queued PLUS in-flight records to `max-queue-size`; SDK queue capacity is at least that large, so accepted records cannot hit its hidden overflow boundary. A measured exporter releases permits when the SDK result completes and counts exported records or failed batches. This supplies exact admission/full metrics without reflection or a custom network queue/retry engine. SDK MeterProvider is explicitly noop; exporter health uses the existing Micrometer registry without global registration. +Standard SDK batching is retained. A nonwaiting semaphore around processor admission conservatively bounds queued PLUS in-flight records to `max-queue-size`; SDK queue capacity is at least that large, so accepted records cannot hit its hidden overflow boundary. A measured exporter owns a first-completion-wins result: SDK completion or the existing `(timeout + retry-max-backoff) * retry-max-attempts + 1s` deadline releases permits and updates counters exactly once. This handles OTel 1.31 async response-body exceptions that abandon their result; deadline/close completion counts one failed batch and late SDK completion is ignored. One daemon scheduler removes canceled deadlines; close stops it and fails outstanding guards. It bounds local accounting, not remote delivery/cancellation. This supplies exact admission/full metrics without reflection or a custom network queue/retry engine. SDK MeterProvider is explicitly noop; exporter health uses the existing Micrometer registry without global registration. + +Shipped Logback configuration sets `io.opentelemetry.exporter` and `io.opentelemetry.sdk.logs` (and inherited child loggers) OFF. These JUL diagnostics are bridged by Boot and may otherwise disclose arbitrary receiver Status messages or transport exception text echoing headers; ERROR-only is insufficient. SkillHub's fixed warning (at most once/minute/pipeline) and Micrometer counters remain the operator signal. This also suppresses shared Trace exporter diagnostics, not its exports. Custom logging must retain these defaults; a deliberate case-sensitive `logging.level` override can temporarily enable unsafe details, preferably only with synthetic credentials and restricted logs, then restore OFF/remove child overrides. The operator guide documents that troubleshooting path. A single mapper allows only known event classes, fixed sources/actor types, scalar IDs, enums and bounded user ID. The SDK also limits 16 attributes and 256 code units. Three Resource strings are separately bounded. Configuration has hard count/duration/header/list bounds. Admitted pipeline memory is bounded by record count and fixed record shape, not a promised byte-accurate or zero-allocation budget. Before commit, transaction synchronizations retain one bounded mapped snapshot per dispatched event outside the SDK admission window; callers must not wrap unbounded bulk work in one transaction. This is not a process-wide heap cap. Saturation latency and receiver container limits are measured in tests; this is not a production throughput benchmark. diff --git a/openspec/changes/add-otlp-business-event-logs/specs/business-event-logs/spec.md b/openspec/changes/add-otlp-business-event-logs/specs/business-event-logs/spec.md index dc9aa3f7c..5fba11930 100644 --- a/openspec/changes/add-otlp-business-event-logs/specs/business-event-logs/spec.md +++ b/openspec/changes/add-otlp-business-event-logs/specs/business-event-logs/spec.md @@ -18,9 +18,16 @@ The application SHALL create a dedicated LoggerProvider only when `skillhub.obse The event catalog in `docs/business-event-logs.md` SHALL define all included events, producer paths, occurrence boundary, actor and fields. Publication SHALL cover auto-publication, PRIVATE confirmation, review approval and promotion approval. Download preparation SHALL NOT mean transfer, installation or execution success. #### Scenario: Commit and rollback -- **WHEN** a supported event is published within a transaction +- **WHEN** a supported event is published within a physical transaction (excluding savepoint-level rollback semantics) - **THEN** the adapter captures a scalar snapshot and trace context before handoff -- **AND** export admission occurs only after commit; rollback emits no record +- **AND** export admission occurs only after that physical transaction commits; its rollback emits no record + +#### Scenario: Savepoint exclusion and independent transaction +- **WHEN** `PROPAGATION_NESTED` work dispatches an event and rolls back its savepoint, but the outer physical transaction commits +- **THEN** savepoint-level rollback suppression is explicitly outside the contract; the event can still export despite the nested write rolling back +- **AND** producers requiring savepoint suppression SHALL NOT dispatch within nested/savepoint work +- **WHEN** `REQUIRES_NEW` work commits and the suspended outer transaction later rolls back +- **THEN** the independently committed event is admitted and the outer transaction's event is suppressed #### Scenario: Nontransactional download - **WHEN** an authorized PUBLISHED download successfully builds its result without an outer transaction @@ -41,6 +48,12 @@ The adapter SHALL use asynchronous SDK batching and standard bounded exporter re - **THEN** the same `log.record.uid` survives retries - **AND** receivers may deduplicate, without an exactly-once promise +#### Scenario: Abandoned export result +- **WHEN** an SDK export result never completes, including a truncated non-2xx response-body callback failure +- **THEN** the export-budget deadline SHALL fail the owned completion guard, release admission permits and count one failed batch +- **AND** result completion, deadline and close SHALL compete for exactly one cleanup; late results SHALL NOT release again or change counters +- **AND** later events SHALL be admitted and exported when the receiver recovers + ### Requirement: Data export SHALL use a single explicit allowlist A single mapper SHALL export only fixed documented scalar fields, including IDs of private resources. Namespace IDs and event names SHALL be operator-filterable. Human actors SHALL use only local opaque `user.id`; anonymous/unknown and system actors SHALL omit it. No entity serialization, enrichment queries, resource autodetection, content, credentials, URLs, IP/UA, request IDs or raw headers is permitted. @@ -54,6 +67,14 @@ A single mapper SHALL export only fixed documented scalar fields, including IDs - **THEN** only IDs, visibility and documented operational fields are exported - **AND** names, slugs and content remain excluded +#### Scenario: Context and diagnostic credential boundaries +- **WHEN** MDC and OTel Baggage contain distinct sensitive sentinels +- **THEN** neither keys nor values SHALL appear in exported attributes, body or resource +- **WHEN** receiver Status messages or transport exceptions echo transport credentials +- **THEN** shipped logging SHALL suppress raw `io.opentelemetry.exporter` and `io.opentelemetry.sdk.logs` diagnostics, including child loggers, by default +- **AND** fixed-text rate-bounded SkillHub diagnostics and counters SHALL remain available +- **AND** operator documentation SHALL warn that explicitly raising SDK logging opts into possible credential disclosure + ### Requirement: Deployment and delivery limitations SHALL be explicit User documentation SHALL include every configuration key/default/bound, direct and Collector deployment, Secret injection, loss/duplicate/shutdown semantics, SDK compatibility limits, troubleshooting, a reproducible real receiver/query example and an acceptance matrix separating executed from unverified scenarios. diff --git a/openspec/changes/add-otlp-business-event-logs/tasks.md b/openspec/changes/add-otlp-business-event-logs/tasks.md index c38dc9706..74502841f 100644 --- a/openspec/changes/add-otlp-business-event-logs/tasks.md +++ b/openspec/changes/add-otlp-business-event-logs/tasks.md @@ -24,3 +24,13 @@ - [x] Verify cleanup of every owned temporary container/network/volume and the authorized baseline worktree; leave changes uncommitted. See acceptance.md for final executed results and explicit unverified deployment scenarios; a task checkbox is not itself acceptance evidence. + +## Repair 1 + +- [x] Reproduce truncated HTTP error-body permit leak before production changes; add bounded first-completion-wins export accounting and late success/failure regression cases. +- [x] Reproduce synthetic credential disclosure in receiver Status and transport-error diagnostics; disable raw exporter/Logs SDK loggers by default and verify both shipped log formats retain safe operator signals. +- [x] Explicitly exclude NESTED savepoint rollback from suppression; verify PostgreSQL NESTED rollback/outer commit and REQUIRES_NEW commit/outer rollback plus documentation consistency. +- [x] Make MDC/Baggage exclusion non-vacuous with distinct seeded sentinels, dispatch preconditions and exported attributes/body/resource checks. +- [x] Retain RED/GREEN evidence and rerun full focused command: 299 tests green; deployment contracts: 4 Python cases plus Helm/release shell suites green. + +Repair 1 results and intermediate failures are recorded in acceptance.md and the appended worker report; earlier broad-suite/site-build limitations remain explicit. diff --git a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsPipeline.java b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsPipeline.java index 5851cb128..869debde6 100644 --- a/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsPipeline.java +++ b/server/skillhub-app/src/main/java/com/iflytek/skillhub/observability/business/BusinessLogsPipeline.java @@ -14,6 +14,9 @@ import io.opentelemetry.sdk.logs.export.LogRecordExporter; import io.opentelemetry.sdk.resources.Resource; import java.time.Duration; import java.util.Collection; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -30,31 +33,49 @@ public final class BusinessLogsPipeline implements AutoCloseable { private final BusinessLogsMetrics metrics; private final Duration shutdownTimeout; private final AtomicBoolean accepting = new AtomicBoolean(true); + private final ScheduledThreadPoolExecutor exportDeadlines; + private final Set pendingExports = ConcurrentHashMap.newKeySet(); BusinessLogsPipeline(BusinessLogsProperties properties, Resource resource, LogRecordExporter exporter, BusinessLogsMetrics metrics) { this.exporter = exporter; this.metrics = metrics; this.shutdownTimeout = properties.getShutdownTimeout(); + exportDeadlines = new ScheduledThreadPoolExecutor(1, task -> { + Thread thread = new Thread(task, "skillhub-business-logs-deadlines"); + thread.setDaemon(true); + return thread; + }); + exportDeadlines.setRemoveOnCancelPolicy(true); Semaphore capacity = new Semaphore(properties.getMaxQueueSize()); LogRecordExporter measured = new LogRecordExporter() { @Override public CompletableResultCode export(Collection records) { int count = records.size(); + // Own the completion boundary: an SDK async callback can throw without completing + // its result (e.g. 1.31 reading a truncated error body). The first outcome wins; + // late completion cannot release permits or update counters a second time. + CompletableResultCode bounded = new CompletableResultCode(); + pendingExports.add(bounded); + bounded.whenComplete(() -> { + pendingExports.remove(bounded); + capacity.release(count); + if (bounded.isSuccess()) { metrics.exported.increment(count); } + else { metrics.exportFailures.increment(); metrics.warn(); } + }); try { + var deadline = exportDeadlines.schedule(bounded::fail, + properties.exportBudget().toNanos(), TimeUnit.NANOSECONDS); + bounded.whenComplete(() -> deadline.cancel(false)); CompletableResultCode result = exporter.export(records); result.whenComplete(() -> { - capacity.release(count); - if (result.isSuccess()) { metrics.exported.increment(count); } - else { metrics.exportFailures.increment(); metrics.warn(); } + if (result.isSuccess()) { bounded.succeed(); } + else { bounded.fail(); } }); - return result; } catch (RuntimeException ex) { - capacity.release(count); - metrics.exportFailures.increment(); - metrics.warn(); - return CompletableResultCode.ofFailure(); + bounded.fail(); } + return bounded; } @Override public CompletableResultCode flush() { return exporter.flush(); } @Override public CompletableResultCode shutdown() { return exporter.shutdown(); } @@ -111,6 +132,11 @@ public final class BusinessLogsPipeline implements AutoCloseable { metrics.error(); // Also cancel network work if shutdown itself unexpectedly fails. try { exporter.shutdown(); } catch (RuntimeException shutdownException) { metrics.error(); } + } finally { + // No deadline thread/tasks survive this lifecycle. A concurrent late export either + // appears here or is rejected by the stopped scheduler and fails its own guard. + exportDeadlines.shutdownNow(); + pendingExports.forEach(CompletableResultCode::fail); } } } diff --git a/server/skillhub-app/src/main/resources/logback-spring.xml b/server/skillhub-app/src/main/resources/logback-spring.xml index c9372168c..803ac9180 100644 --- a/server/skillhub-app/src/main/resources/logback-spring.xml +++ b/server/skillhub-app/src/main/resources/logback-spring.xml @@ -45,6 +45,12 @@ + + + + diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsDiagnosticsTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsDiagnosticsTest.java new file mode 100644 index 000000000..6027cee9c --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsDiagnosticsTest.java @@ -0,0 +1,117 @@ +package com.iflytek.skillhub.observability.business; + +import com.iflytek.skillhub.domain.event.BusinessEventSnapshot; +import com.iflytek.skillhub.domain.event.SkillPublishedEvent; +import com.iflytek.skillhub.domain.skill.SkillVisibility; +import com.sun.net.httpserver.HttpServer; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.springframework.boot.logging.LoggingInitializationContext; +import org.springframework.boot.logging.LoggingSystem; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.core.env.StandardEnvironment; +import org.springframework.core.env.MapPropertySource; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Load the shipped logging config, including JUL-to-SLF4J routing used by Boot. */ +@ExtendWith(OutputCaptureExtension.class) +class BusinessLogsDiagnosticsTest { + private static final String CREDENTIAL = "SYNTHETIC-1009-CREDENTIAL"; + private LoggingSystem logging; + + private void configureShippedLogging(String format) { + logging = LoggingSystem.get(getClass().getClassLoader()); + logging.beforeInitialize(); + var environment = new StandardEnvironment(); + environment.getPropertySources().addFirst(new MapPropertySource("diagnostic-test", + java.util.Map.of("skillhub.observability.log-format", format))); + logging.initialize(new LoggingInitializationContext(environment), "classpath:logback-spring.xml", null); + for (String name : java.util.List.of("io.opentelemetry.exporter.internal.http.HttpExporter", + "io.opentelemetry.exporter.sender.okhttp.internal.OkHttpHttpSender", + "io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor")) { + org.slf4j.LoggerFactory.getLogger(name); + assertThat(logging.getLoggerConfiguration(name).getEffectiveLevel()) + .isEqualTo(org.springframework.boot.logging.LogLevel.OFF); + } + } + + @AfterEach + void cleanupLogging() { + if (logging != null) { + logging.cleanUp(); + logging.initialize(new LoggingInitializationContext(new StandardEnvironment()), "classpath:logback-test.xml", null); + logging.cleanUp(); + } + } + + @ParameterizedTest + @ValueSource(strings = {"text", "json"}) + void receiverStatusEchoingAuthorizationCannotLeakToAnyConsoleLog(String format, CapturedOutput output) throws Exception { + configureShippedLogging(format); + var server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/v1/logs", exchange -> { + exchange.getRequestBody().readAllBytes(); + String message = "auth failed for " + exchange.getRequestHeaders().getFirst("Authorization"); + byte[] text = message.getBytes(StandardCharsets.UTF_8); + // google.rpc.Status { code: 16, message: } + byte[] body = new byte[text.length + 4]; + body[0] = 0x08; body[1] = 0x10; body[2] = 0x12; body[3] = (byte) text.length; + System.arraycopy(text, 0, body, 4, text.length); + exchange.getResponseHeaders().add("Content-Type", "application/x-protobuf"); + exchange.sendResponseHeaders(401, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + server.start(); + try { + exportFailure("http://127.0.0.1:" + server.getAddress().getPort() + "/v1/logs"); + org.awaitility.Awaitility.await().untilAsserted(() -> assertThat(output.getAll()).contains("Business Logs telemetry failed")); + assertThat(output.getAll()).contains("Business Logs telemetry failed") + .doesNotContain(CREDENTIAL, "auth failed for", "Failed to export logs"); + } finally { server.stop(0); } + } + + @ParameterizedTest + @ValueSource(strings = {"text", "json"}) + void transportExceptionEchoingSensitiveTextCannotLeakToAnyConsoleLog(String format, CapturedOutput output) throws Exception { + configureShippedLogging(format); + // OkHttp reports an invalid status line verbatim in ProtocolException (transport onError). + try (var receiver = new BusinessLogsTestReceiver( + "HTTP/1.1 " + CREDENTIAL + "\r\nConnection: close\r\n\r\n")) { + exportFailure(receiver.endpoint()); + assertThat(receiver.requests.get()).isEqualTo(1); + org.awaitility.Awaitility.await().untilAsserted(() -> assertThat(output.getAll()).contains("Business Logs telemetry failed")); + assertThat(output.getAll()).contains("Business Logs telemetry failed") + .doesNotContain(CREDENTIAL, "Unexpected status line", "Failed to export logs"); + } + } + + private static void exportFailure(String endpoint) { + new ApplicationContextRunner().withUserConfiguration(BusinessLogsConfiguration.class) + .withBean(SimpleMeterRegistry.class, SimpleMeterRegistry::new) + .withPropertyValues("skillhub.observability.business-logs.enabled=true", + "skillhub.observability.business-logs.endpoint=" + endpoint, + "skillhub.observability.business-logs.headers=Authorization=Bearer%20" + CREDENTIAL, + "skillhub.observability.business-logs.retry-max-attempts=1", + "skillhub.observability.business-logs.timeout=500ms") + .run(context -> { + assertThat(context).hasNotFailed(); + var snapshot = new BusinessEventSnapshot(7L, SkillVisibility.PUBLIC, "reviewer", "user", "review", Instant.now()); + context.publishEvent(new SkillPublishedEvent(1L, 2L, "reviewer", snapshot)); + context.getBean(BusinessLogsPipeline.class).forceFlush().join(5, TimeUnit.SECONDS); + var metrics = context.getBean(BusinessLogsMetrics.class); + assertThat(metrics.exportFailures.count()).isEqualTo(1); + assertThat(metrics.exported.count()).isZero(); + }); + } +} diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsHttpTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsHttpTest.java index 73749ae00..e7b31b3cc 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsHttpTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsHttpTest.java @@ -17,10 +17,49 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; import static org.assertj.core.api.Assertions.assertThat; class BusinessLogsHttpTest { + @Test + void truncatedErrorBodyCannotPermanentlyExhaustAdmission() throws Exception { + try (var receiver = new BusinessLogsTestReceiver( + "HTTP/1.1 400 Bad Request\r\nContent-Length: 100\r\nConnection: close\r\n\r\nx")) { + new ApplicationContextRunner().withUserConfiguration(BusinessLogsConfiguration.class) + .withBean(SimpleMeterRegistry.class, SimpleMeterRegistry::new) + .withPropertyValues("skillhub.observability.business-logs.enabled=true", + "skillhub.observability.business-logs.endpoint=" + receiver.endpoint(), + "skillhub.observability.business-logs.max-queue-size=1", + "skillhub.observability.business-logs.max-export-batch-size=1", + "skillhub.observability.business-logs.timeout=100ms", + "skillhub.observability.business-logs.retry-max-attempts=1", + "skillhub.observability.business-logs.retry-max-backoff=1ms", + "skillhub.observability.business-logs.retry-initial-backoff=1ms") + .run(context -> { + assertThat(context).hasNotFailed(); + var pipeline = context.getBean(BusinessLogsPipeline.class); + var metrics = context.getBean(BusinessLogsMetrics.class); + var snapshot = new BusinessEventSnapshot(7L, SkillVisibility.PUBLIC, "reviewer", "user", "review", Instant.now()); + var event = new SkillPublishedEvent(1L, 2L, "reviewer", snapshot); + context.publishEvent(event); + pipeline.forceFlush().join(5, TimeUnit.SECONDS); + // Wait beyond the bounded export budget, not just the HTTP timeout. + org.awaitility.Awaitility.await().atMost(Duration.ofSeconds(5)) + .untilAsserted(() -> assertThat(metrics.exportFailures.count()).isEqualTo(1)); + for (int i = 0; i < 2; i++) { + context.publishEvent(event); + pipeline.forceFlush().join(5, TimeUnit.SECONDS); + } + assertThat(receiver.requests.get()).isEqualTo(3); + assertThat(metrics.enqueued.count()).isEqualTo(3); + assertThat(metrics.exported.count()).isEqualTo(2); + assertThat(metrics.queueFull.count()).isZero(); + assertThat(metrics.exportFailures.count()).isEqualTo(1); + }); + } + } + @Test void boundedRetryReusesIdenticalUidAndPayloadAndAuthenticationIsTransportOnly() throws Exception { var server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsPipelineTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsPipelineTest.java new file mode 100644 index 000000000..f0f94df33 --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsPipelineTest.java @@ -0,0 +1,68 @@ +package com.iflytek.skillhub.observability.business; + +import com.iflytek.skillhub.domain.event.BusinessEventSnapshot; +import com.iflytek.skillhub.domain.event.SkillPublishedEvent; +import com.iflytek.skillhub.domain.skill.SkillVisibility; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.opentelemetry.context.Context; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.data.LogRecordData; +import io.opentelemetry.sdk.logs.export.LogRecordExporter; +import io.opentelemetry.sdk.resources.Resource; +import java.time.Duration; +import java.time.Instant; +import java.util.Collection; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +class BusinessLogsPipelineTest { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void lateCompletionAfterDeadlineCannotReleaseTwiceOrChangeOutcome(boolean lateSuccess) throws Exception { + var properties = new BusinessLogsProperties(); + properties.setMaxQueueSize(1); + properties.setMaxExportBatchSize(1); + properties.setTimeout(Duration.ofMillis(100)); + properties.setRetryMaxAttempts(1); + properties.setRetryMaxBackoff(Duration.ofMillis(1)); + var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry()); + var exports = new LinkedBlockingQueue(); + var exporter = new LogRecordExporter() { + @Override public CompletableResultCode export(Collection records) { + var result = new CompletableResultCode(); + exports.add(result); + return result; + } + @Override public CompletableResultCode flush() { return CompletableResultCode.ofSuccess(); } + @Override public CompletableResultCode shutdown() { return CompletableResultCode.ofSuccess(); } + }; + try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics)) { + var snapshot = new BusinessEventSnapshot(7L, SkillVisibility.PUBLIC, "reviewer", "user", "review", Instant.now()); + var record = new BusinessEventMapper(properties).map(new SkillPublishedEvent(1L, 2L, "reviewer", snapshot)).orElseThrow(); + pipeline.emit(record, Context.root()); + var abandoned = exports.poll(5, TimeUnit.SECONDS); + assertThat(abandoned).isNotNull(); + pipeline.forceFlush().join(5, TimeUnit.SECONDS); + await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> assertThat(metrics.exportFailures.count()).isEqualTo(1)); + pipeline.emit(record, Context.root()); + var recovering = exports.poll(5, TimeUnit.SECONDS); + assertThat(recovering).isNotNull(); + if (lateSuccess) { abandoned.succeed(); } else { abandoned.fail(); } + // The recovered permit is now occupied again. Late completion must not add capacity. + pipeline.emit(record, Context.root()); + assertThat(metrics.enqueued.count()).isEqualTo(2); + assertThat(metrics.queueFull.count()).isEqualTo(1); + assertThat(metrics.exported.count()).isZero(); + assertThat(metrics.exportFailures.count()).isEqualTo(1); + recovering.succeed(); + pipeline.forceFlush().join(5, TimeUnit.SECONDS); + assertThat(metrics.exported.count()).isEqualTo(1); + assertThat(metrics.exportFailures.count()).isEqualTo(1); + } + } +} diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTestReceiver.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTestReceiver.java new file mode 100644 index 000000000..d78eea22e --- /dev/null +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTestReceiver.java @@ -0,0 +1,73 @@ +package com.iflytek.skillhub.observability.business; + +import java.io.BufferedInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; + +/** Raw local HTTP peer for responses that HttpServer deliberately cannot produce. */ +final class BusinessLogsTestReceiver implements AutoCloseable { + private final ServerSocket server; + private final Thread worker; + private final String firstResponse; + final AtomicInteger requests = new AtomicInteger(); + private volatile Throwable failure; + + BusinessLogsTestReceiver(String firstResponse) throws IOException { + this.firstResponse = firstResponse; + server = new ServerSocket(0, 10, InetAddress.getByName("127.0.0.1")); + worker = new Thread(this::receive, "skillhub-1009-http-test"); + worker.setDaemon(true); + worker.start(); + } + + String endpoint() { return "http://127.0.0.1:" + server.getLocalPort() + "/v1/logs"; } + + private void receive() { + try { + while (!server.isClosed()) { + try (Socket socket = server.accept()) { + socket.setSoTimeout(5000); + var input = new BufferedInputStream(socket.getInputStream()); + int length = 0; + String line; + while (!(line = line(input)).isEmpty()) { + if (line.toLowerCase(java.util.Locale.ROOT).startsWith("content-length:")) { + length = Integer.parseInt(line.substring(line.indexOf(':') + 1).trim()); + } + } + if (input.readNBytes(length).length != length) { throw new IOException("Incomplete test request"); } + int count = requests.incrementAndGet(); + String response = count == 1 ? firstResponse + : "HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"; + socket.getOutputStream().write(response.getBytes(StandardCharsets.ISO_8859_1)); + socket.getOutputStream().flush(); + } + } + } catch (IOException ex) { + if (!server.isClosed()) { failure = ex; } + } catch (RuntimeException ex) { failure = ex; } + } + + private static String line(BufferedInputStream input) throws IOException { + var bytes = new ByteArrayOutputStream(); + int value; + while ((value = input.read()) != -1) { + if (value == '\n') { return bytes.toString(StandardCharsets.ISO_8859_1).trim(); } + bytes.write(value); + } + throw new IOException("Incomplete test headers"); + } + + @Override + public void close() throws Exception { + server.close(); + worker.join(6000); + if (worker.isAlive()) { throw new AssertionError("HTTP test peer did not stop"); } + if (failure != null) { throw new AssertionError("HTTP test peer failed", failure); } + } +} diff --git a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTransactionTest.java b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTransactionTest.java index bd07d0426..c8f800de7 100644 --- a/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTransactionTest.java +++ b/server/skillhub-app/src/test/java/com/iflytek/skillhub/observability/business/BusinessLogsTransactionTest.java @@ -13,6 +13,7 @@ import com.iflytek.skillhub.domain.skill.SkillStatus; import com.iflytek.skillhub.domain.skill.SkillVisibility; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.baggage.Baggage; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanContext; import io.opentelemetry.api.trace.TraceFlags; @@ -29,11 +30,13 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; +import org.slf4j.MDC; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.datasource.DriverManagerDataSource; import org.springframework.jdbc.datasource.DataSourceTransactionManager; import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.transaction.TransactionDefinition; import org.testcontainers.containers.PostgreSQLContainer; import org.testcontainers.junit.jupiter.Container; import org.testcontainers.junit.jupiter.Testcontainers; @@ -102,6 +105,79 @@ class BusinessLogsTransactionTest { } } + @Test + void nestedSavepointRollbackIsExplicitlyOutsideSuppressionContract() { + var dataSource = new DriverManagerDataSource(POSTGRES.getJdbcUrl(), POSTGRES.getUsername(), POSTGRES.getPassword()); + var jdbc = new JdbcTemplate(dataSource); + jdbc.execute("create table nested_analytics_probe (id int primary key)"); + var manager = new DataSourceTransactionManager(dataSource); + var outer = new TransactionTemplate(manager); + var nested = new TransactionTemplate(manager); + nested.setPropagationBehavior(TransactionDefinition.PROPAGATION_NESTED); + var exporter = new CapturingExporter(); + var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry()); + var properties = new BusinessLogsProperties(); + try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics)) { + var listener = new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics); + outer.executeWithoutResult(status -> { + nested.executeWithoutResult(savepoint -> { + assertThat(savepoint.hasSavepoint()).isTrue(); + jdbc.update("insert into nested_analytics_probe values (1)"); + listener.onEvent(new SkillPublishedEvent(1L, 2L, "approver", snapshot)); + savepoint.setRollbackOnly(); + }); + assertThat(jdbc.queryForObject("select count(*) from nested_analytics_probe", Integer.class)).isZero(); + assertThat(metrics.enqueued.count()).isZero(); + }); + pipeline.forceFlush().join(5, TimeUnit.SECONDS); + assertThat(jdbc.queryForObject("select count(*) from nested_analytics_probe", Integer.class)).isZero(); + // Synchronization follows the physical transaction, not the rolled-back savepoint. + assertThat(exporter.records).hasSize(1); + } + } + + @Test + void requiresNewCommitExportsEvenWhenOuterTransactionRollsBack() { + var dataSource = new DriverManagerDataSource(POSTGRES.getJdbcUrl(), POSTGRES.getUsername(), POSTGRES.getPassword()); + var jdbc = new JdbcTemplate(dataSource); + jdbc.execute("create table requires_new_analytics_probe (id int primary key)"); + var manager = new DataSourceTransactionManager(dataSource); + var outer = new TransactionTemplate(manager); + var independent = new TransactionTemplate(manager); + independent.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + var exporter = new CapturingExporter(); + var metrics = new BusinessLogsMetrics(new SimpleMeterRegistry()); + var properties = new BusinessLogsProperties(); + try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics)) { + var listener = new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics); + outer.executeWithoutResult(status -> { + jdbc.update("insert into requires_new_analytics_probe values (1)"); + listener.onEvent(new SkillPublishedEvent(1L, 2L, "approver", snapshot)); + independent.executeWithoutResult(inner -> { + jdbc.update("insert into requires_new_analytics_probe values (2)"); + listener.onEvent(new SkillPublishedEvent(3L, 4L, "approver", snapshot)); + }); + assertThat(metrics.enqueued.count()).isEqualTo(1); + status.setRollbackOnly(); + }); + pipeline.forceFlush().join(5, TimeUnit.SECONDS); + assertThat(jdbc.queryForList("select id from requires_new_analytics_probe", Integer.class)).containsExactly(2); + assertThat(exporter.records).hasSize(1); + assertThat(exporter.records.getFirst().getAttributes().get(AttributeKey.longKey("skillhub.skill.id"))).isEqualTo(3L); + } + } + + @Test + void transactionDocumentationStatesSavepointExclusionAndIndependentCommit() throws Exception { + for (String path : List.of("docs/business-event-logs.md", + "openspec/changes/add-otlp-business-event-logs/design.md", + "openspec/changes/add-otlp-business-event-logs/specs/business-event-logs/spec.md", + "openspec/changes/add-otlp-business-event-logs/acceptance.md")) { + assertThat(java.nio.file.Files.readString(java.nio.file.Path.of("../..", path))) + .as(path).contains("PROPAGATION_NESTED", "REQUIRES_NEW"); + } + } + @Test void unsampledSpanCapturedBeforeCommitAndNoMdcOrBaggageExported() { var dataSource = new DriverManagerDataSource(POSTGRES.getJdbcUrl(), POSTGRES.getUsername(), POSTGRES.getPassword()); @@ -113,7 +189,11 @@ class BusinessLogsTransactionTest { try (var pipeline = new BusinessLogsPipeline(properties, Resource.empty(), exporter, metrics)) { var listener = new BusinessEventListener(new BusinessEventMapper(properties), pipeline, metrics); transaction.executeWithoutResult(status -> { - try (var ignored = Span.wrap(spanContext).makeCurrent()) { + try (var mdc = MDC.putCloseable("sensitive.mdc", "SYNTHETIC-1009-MDC-SECRET"); + var baggage = Baggage.builder().put("sensitive.baggage", "SYNTHETIC-1009-BAGGAGE-SECRET").build().makeCurrent(); + var ignored = Span.wrap(spanContext).makeCurrent()) { + assertThat(MDC.get("sensitive.mdc")).isEqualTo("SYNTHETIC-1009-MDC-SECRET"); + assertThat(Baggage.current().getEntryValue("sensitive.baggage")).isEqualTo("SYNTHETIC-1009-BAGGAGE-SECRET"); listener.onEvent(new SkillPublishedEvent(1L, 2L, "approver", snapshot)); } assertThat(Span.current().getSpanContext().isValid()).isFalse(); @@ -122,6 +202,12 @@ class BusinessLogsTransactionTest { assertThat(exporter.records).hasSize(1); assertThat(exporter.records.getFirst().getSpanContext()).isEqualTo(spanContext); assertThat(exporter.records.getFirst().getSpanContext().isSampled()).isFalse(); + var exported = exporter.records.getFirst(); + for (String value : List.of(exported.getAttributes().toString(), String.valueOf(exported.getBody().asString()), + exported.getResource().getAttributes().toString())) { + assertThat(value).doesNotContain("sensitive.mdc", "sensitive.baggage", + "SYNTHETIC-1009-MDC-SECRET", "SYNTHETIC-1009-BAGGAGE-SECRET"); + } } }