-
[OPIK-7051] [BE] Reactive online-scoring enqueue + enqueue metric with workspace labels (#7243)
frostbyte_neo 发布于
2026-06-26 07:47:28 +00:00 - [OPIK-7051] [BE] Add enqueue counter for online-scoring publisher
OnlineScorePublisher.enqueueMessage only logged Redis publish outcomes, so the
enqueue stage of the online-scoring pipeline had no metric — push failures were
invisible to monitoring. Add an OTel counter online_scoring_enqueue_total with
attributes evaluator_type and result (success|error); result=error counts the
publish failures that were previously only logged.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7051] [BE] Make enqueue reactive; label enqueue metric with workspace
Follow-up on the enqueue counter:
- enqueueMessage / enqueueThreadMessage now return Mono (reactive) instead
of self-subscribing. enqueueThreadMessage resolves its blocking rule lookup via
Mono.fromCallable on boundedElastic, so it is fully non-blocking and callers
compose it directly. - The enqueue metric online_scoring_enqueue_total now carries workspace_id and
workspace_name (plus evaluator_type, result), read from the reactive context via
deferContextual — the workspace the ingest endpoint / sampler already put on the
context. Falls back to unknown→id when absent. - Callers updated to subscribe/compose the returned Mono (so the enqueue still
runs): samplers seed the context from the event and fire-and-forget with error
logging; ManualEvaluationService and the thread-close publisher compose the
enqueue into their request-scoped reactive chains so it inherits the full
workspace context.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7051] [BE] Fix online-scoring tests for reactive enqueue
enqueueMessage/enqueueThreadMessage now return Mono and are lazy, so the
unit tests had to be adjusted:- Sampler/span-sampler/test-suite-sampler tests: stub enqueueMessage to return
Mono.empty() (the production code chains .contextWrite(...).subscribe()/compose
on the result, which NPE'd on the default null). Replaced
verifyNoInteractions(onlineScorePublisher) with verify(..., never()).enqueueMessage(..)
where a lenient stub is present. - OnlineScorePublisherTest: .block() the returned Mono so the (now lazy) enqueue
actually runs before asserting on the Redis stream add params.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7051] [BE] Address review: workspace_name fallback + skip redundant rule lookup
- enqueueMessage: workspace_name now falls back to workspace_id (not the literal
'unknown') when WORKSPACE_NAME is absent/blank on the reactive context. - enqueueThreadMessage: add an overload taking the already-resolved
AutomationRuleEvaluator so callers that already hold the rule
(ManualEvaluationService) skip the redundant blocking findById; the ruleId
overload (used where only the id is known) resolves once and delegates.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7051] [BE] Carry workspaceName on SpansCreated so span enqueue is labelled
The span sampler only had workspace_id, so span-level enqueue metrics fell back to
the id for workspace_name. Mirror TracesCreated: add a nullable workspaceName to
SpansCreated, populate it from RequestContext.WORKSPACE_NAME at both SpanService
publish sites, and seed it into the reactive context in OnlineScoringSpanSampler
(falling back to the id when absent). Span enqueue errors now carry the real name.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7051] [BE] Extract shared sampler enqueue helper (review)
Both samplers had an identical publishSampled (context seeding + subscribe + error
log), so the error-log change had to be kept in sync across two files. Extract it
into a package-private OnlineScoringSamplerSupport.publishSampled used by the trace
and span samplers, and adopt the structured error log
('...for evaluator={} workspaceId={} workspaceName={}'). Single source of truth; no
behaviour change (still fire-and-forget via OnlineScorePublisher.enqueueMessage).Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7051] [BE] Quote log placeholders in sampler enqueue helper (review)
Match the backend logging convention (.agents/skills/opik-backend/SKILL.md):
evaluator='{}' workspaceId='{}' workspaceName='{}'.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7051] [BE] Address review nits
- SpansCreated: workspaceName field comment as javadoc.
- TestSuiteAssertionSampler: log the message count in the enqueue-error log.
- OnlineScorePublisher.enqueueMessage: subscribeOn(boundedElastic) so the Redis
stream writes run off the caller's (e.g. EventBus) thread. This also restores the
bounded-scheduler offload that the TraceThreadOnlineScorerPublisher refactor
dropped, since both thread paths funnel through enqueueMessage.
Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
- [OPIK-7051] [BE] Log batch size in sampler enqueue-error log too
Apply the 'log messages size' nit consistently to the shared
OnlineScoringSamplerSupport error log (the trace/span sampler path), matching the
TestSuiteAssertionSampler change.Co-Authored-By: Claude Opus 4.8 (1M context) noreply@anthropic.com
Co-authored-by: Claude Opus 4.8 (1M context) noreply@anthropic.com
下载附件