-
[OPIK-7262] [BE] observability: metrics for project last-updated flush + events consumed by listeners (#7384)
frostbyte_neo 发布于
2026-07-08 12:53:53 +00:00 - [OPIK-7246] fix: buffer projects.last_updated_trace_at in Redis, flush to MySQL periodically
ProjectEventListener wrote projects.last_updated_trace_at synchronously on every
TracesCreated/TracesUpdated event (via ProjectService.recordLastUpdatedTrace) on an
unbounded virtual-thread AsyncEventBus, funneling concurrent WRITE transactions onto a
single hot projects row and producing MySQLTransactionRollbackException (deadlock /
lock-wait) and EOFException / ConnectException on the ingestion path.Buffer the per-project maximum timestamp in a Redis ZSET (addIfGreater, member
"workspaceId:projectId", score = epoch millis) and flush the accumulated maxima to MySQL
under a best-effort distributed lock every 30s (ProjectLastUpdatedFlushJob). Many trace
writes per project collapse into one batched UPDATE per flush from a single flusher,
removing the hot-row contention and connection churn. Gated behind projectLastUpdatedFlush
(default off); the synchronous MySQL path remains the fallback. Monotonic (forward-only
guard preserved), idempotent, and fails open on Redis errors.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7246] fix: eliminate flush read-then-remove race via atomic rename-drain
Addresses PR review: flushLastUpdatedTraces read entries then removeAll'd them, so a
concurrent recordLastUpdatedTrace (addIfGreater) between the read and the removal could be
dropped, leaving the marker stale. Atomically renamenx the live buffer into a snapshot key
(project:last-updated-trace:flushing); writers immediately continue on a fresh live key and
the drain owns the snapshot exclusively, so its removeAll cannot race a new write. A snapshot
left by an interrupted flush is drained on the next cycle (idempotent; the DB write only moves
forward). Uses only Redisson APIs (isExists/renamenx/entryRange/removeAll) — no raw Redis/Lua.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7246] fix flush drain re-read, use builder for parsed record, Javadoc
- Drain the flushing snapshot one page at a time, reading the next page only after the
current page's write + removeAll complete. The previous Flux.expand pre-fetched the next
entryRange(0, batchSize-1) before removal, so the same top entries could be read and
written to MySQL repeatedly until the iteration cap stopped the loop (idempotent but
wasteful). Removal now gates pagination, guaranteeing forward progress. - ParsedLastUpdatedTrace: @Builder(toBuilder = true) + @NonNull, built via builder.
- Convert explanatory block comments to Javadoc.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7246] ProjectServiceImpl: explicit @Inject constructor for @Config param
@Config is placed directly on the constructor parameter rather than relying on Lombok's
generated constructor to carry it. Replaces @RequiredArgsConstructor(onConstructor_ = @Inject)
with an explicit constructor (same pattern as ProjectLastUpdatedFlushJob).Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7246] separate concerns: buffer-write service, MySQL-only ProjectService, drain in job
Refactor per review:
- ProjectServiceImpl now only does the MySQL project update (recordLastUpdatedTrace);
dropped the Redis/config dependencies, the buffering branch and the drain. Reverted to
the Lombok constructor (no @Config here anymore). - New ProjectLastUpdatedTraceBufferService (domain) owns the ingestion-path Redis buffer
write (addIfGreater) and the enabled/disabled decision; ProjectEventListener uses it. - ProjectLastUpdatedFlushJob owns ProjectLastUpdatedFlushConfig and the Redis read/drain
(renamenx snapshot + paged entryRange/removeAll + parse) and calls ProjectService for the
MySQL write. - Tests: add ProjectLastUpdatedTraceBufferServiceTest; move drain coverage into
ProjectLastUpdatedFlushJobTest; remove ProjectServiceLastUpdatedTraceTest.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7246] address review: flush logic in service, thin job, sync writes, validation
Per andrescrz review:
- Move all flush/Redis logic into ProjectLastUpdatedTraceBufferService (now owns record()
- flush() + the ZSET key constants). The job is thin: scheduling, distributed lock and
interruption only, delegating to bufferService.flush().
- flush() + the ZSET key constants). The job is thin: scheduling, distributed lock and
- Use the synchronous StringRedisClient so record() no longer hides an async side-effect
behind a void signature; the drain becomes a simple sequential loop. - interrupt() now disposes the in-flight flush subscription, not just the schedule.
- Null-safe argument handling on public methods via CollectionUtils.isEmpty.
- Config: remove Java defaults (single source of truth is config.yml), add @Max bounds;
key constants moved out of config into the service.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7246] add projectLastUpdatedFlush block to config-test.yml
Removing the Java defaults made the config fields @NotNull with no fallback, so every full
config file must define the block. config-test.yml lacked it, failing config validation at
resource-test bootstrap (TestHttpClientUtils static init). Add the block with test values.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7246] address review: interface validation, flush log wording, interrupt test
- Remove @NonNull from the ProjectLastUpdatedTraceBufferService interface method (validation
is an impl detail); keep it on the implementation. - flush() count is markers processed, not rows updated (the MySQL write is a monotonic UPDATE
that skips already-ahead rows). Relabel the job log "wrote" -> "processed" and document it. - Add a job test asserting interrupt() disposes the in-flight flush subscription.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [NA] [BE] observability: metrics for project last-updated flush + events consumed
Adds OpenTelemetry metrics (meter opik.project_last_updated_flush):
- buffer: opik.project_last_updated_flush.buffered (counter, tag result=ok|error) —
markers buffered to Redis on the ingestion path (surfaces the previously log-only,
fail-open Redis write failures). - flush: .markers (counter) processed per flush; .run (counter, tag
result=success|skipped_lock|error) and .duration (histogram, ms) for the job.
Also adds opik.event.consumed (counter, tag listener) in EventInterceptor so every event
consumed by a listener is counted, not just errors (opik.event.error).Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7246] address review: propagate record() errors, decouple job scheduling, integration tests
- record() lets Redis failures bubble to the caller instead of swallowing them,
so ProjectEventListener owns the handling like the synchronous fallback branch. - flush()'s drain loop now conditions on the page-size check instead of an
internal while(true)/break, avoiding the appearance of a potential infinite loop. - Split "buffering enabled" from "job scheduled" via a new jobEnabled flag, so
tests can exercise the buffer/flush and job wiring deterministically without
racing the Quartz schedule. - Replaced the mock-heavy unit tests with black-box/integration coverage:
ProjectLastUpdatedTraceBufferServiceTest drives record()/flush() against real
Redis/MySQL and asserts through the project API; ProjectLastUpdatedFlushJobTest
is a single happy-path smoke test mirroring ExperimentProjectMigrationJobTest.
Co-Authored-By: Claude Opus 4.8 noreply@anthropic.com
- [OPIK-7246] move ProjectLastUpdatedFlushJobTest to its job's package
The Job class lives in api.resources.v1.jobs, matching its sibling
ExperimentDenormalizationJob/Test (same periodic-flush-job shape). The domain
package is for migration jobs tightly coupled to a domain migration service,
which this isn't.Dropped the dependency on PENDING_SET_KEY (package-private in the domain
package) by polling via repeated doJob() invocations until the project API
reflects the flush, instead of peeking at the buffer's internal Redis key.Co-Authored-By: Claude Opus 4.8 noreply@anthropic.com
- [NA] address review: rename opik.event.consumed to reflect actual semantics
Several listeners (TraceThreadListener, TraceDeletedListener,
EnvironmentAutoCreateListener) subscribe to a reactive chain internally and
return before it completes, so the interceptor only observes dispatch of the
synchronous portion, not end-to-end processing. Renamed to opik.event.invoked
and clarified the description accordingly; safe since the metric is new and
unmerged, no consumers depend on the old name.Co-Authored-By: Claude Opus 4.8 noreply@anthropic.com
- [NA] address review: count buffered markers per-item, not whole-batch on error
A mid-batch addIfGreater failure previously counted the entire batch as
result=error (and, on success, the whole batch as ok), overcounting the
opik.project_last_updated_flush.buffered metric. Count each marker once by its
actual outcome: succeeded markers stay ok, the failing one is error, and the
un-attempted remainder isn't counted (we rethrow on first failure).Co-Authored-By: Claude Opus 4.8 noreply@anthropic.com
- [NA] address review: reuse METER_NAME, drop overlapping run counter
- Extract METER_NAME constant in ProjectLastUpdatedFlushJob and build metric
names from it; reuse it for the buffered/markers metric names in
ProjectLastUpdatedTraceBufferServiceImpl (no repeated literal prefix). - Drop runCounter in favor of the runDuration histogram tagged by result
(success|error|skipped_lock) — the histogram's per-attribute _count already
gives run counts, so the separate counter was redundant. - Clarify markers metric is a monotonic cumulative throughput counter (not a
gauge) in its description/comment.
Co-Authored-By: Claude Opus 4.8 noreply@anthropic.com
Co-authored-by: Claude Opus 4.8 (1M context) noreply@anthropic.com
下载附件