Skip to content

[OPIK-8262] [BE] fix: classify provider errors by status and stop collapsing trace-thread fan-out #8162

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
thiagohora wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[OPIK-8262] [BE] fix: classify provider errors by status and stop collapsing trace-thread fan-out #8162

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions ...backend/src/main/java/com/comet/opik/api/resources/v1/events/OnlineScoringBaseScorer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import com.comet.opik.api.FeedbackScoreItem;
import com.comet.opik.api.Span;
import com.comet.opik.api.Trace;
import com.comet.opik.api.Visibility;
import com.comet.opik.api.evaluators.AutomationRuleEvaluatorType;
import com.comet.opik.api.events.RedisSubscriberMessage;
import com.comet.opik.api.filter.Operator;
Expand All @@ -12,6 +13,7 @@
import com.comet.opik.domain.SpanService;
import com.comet.opik.domain.TraceSearchCriteria;
import com.comet.opik.domain.TraceService;
import com.comet.opik.domain.evaluators.OnlineScorePublisher;
import com.comet.opik.infrastructure.OnlineScoringConfig;
import com.comet.opik.infrastructure.OnlineScoringStreamConfigurationAdapter;
import com.comet.opik.infrastructure.auth.RequestContext;
Expand All @@ -31,6 +33,7 @@
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.comet.opik.api.FeedbackScoreItem.FeedbackScoreBatchItem;
Expand Down Expand Up @@ -252,4 +255,44 @@ protected Flux<Trace> retrieveFullThreadContext(@NotNull String threadId,
.defer(() -> retrieveFullThreadContext(threadId, lastReceivedIdRef, projectId)));
}));
}

/**
* Routes a trace-thread entry down one of two branches: an entry carrying several thread ids was
* written by an older build and is <b>migrated</b> — republished as one entry per id, never scored —
* while the single-id entries this build writes are <b>scored</b> normally. Completion therefore does
* not mean "scored", so per-outcome logging belongs inside {@code scoreThread} or in the migrate
* branch, never on the returned Mono.
*
* <p>Migrating rather than scoring avoids needing a rule for reducing N per-thread outcomes into the
* one verdict a stream entry gets, which must mis-serve somebody when a permanent failure and a
* retryable one land together. Each replacement entry gets its own retry budget instead.
*
* <p>The ack is implicit: this returns the republish, so a failure leaves the entry unacked and it
* redelivers. If the republish succeeds but the ack fails, the entry splits twice and some threads are
* scored twice — tolerable because {@code feedback_scores} is a {@code ReplacingMergeTree} versioned on
* {@code last_updated_at}, so the second score overwrites rather than duplicating.
*
* <p>The migrate branch is a temporary shim, deletable once no pre-split entry can be in flight.
*/
protected Mono<Void> migrateOrScoreThreadIds(@NonNull M message, @NonNull List<String> threadIds,
@NonNull OnlineScorePublisher publisher, @NonNull Function<String, M> singleThreadIdCopy,
@NonNull Function<String, Mono<Void>> scoreThread) {
if (threadIds.size() > 1) {
// Logged on success, and worded so it can never be mistaken for a scoring log.
return publisher.enqueueMessage(threadIds.stream().map(singleThreadIdCopy).toList(), type)
.doOnSuccess(unused -> log.info(
"Migrated '{}' legacy thread ids to single-id entries for workspace '{}'; "
+ "no scoring performed for this entry",
threadIds.size(), message.workspaceId()));
}
// @NotEmpty says this cannot happen; if it does, no retry would help, so let it be acked away.
if (threadIds.isEmpty()) {
log.warn("Discarding trace-thread entry with no thread ids for workspace '{}'", message.workspaceId());
return Mono.empty();
}
Comment thread
thiagohora marked this conversation as resolved.
return scoreThread.apply(threadIds.getFirst())
.contextWrite(context -> context.put(RequestContext.WORKSPACE_ID, message.workspaceId())
.put(RequestContext.USER_NAME, message.userName())
.put(RequestContext.VISIBILITY, Visibility.PRIVATE));
}
}
35 changes: 13 additions & 22 deletions ...java/com/comet/opik/api/resources/v1/events/OnlineScoringTraceThreadLlmAsJudgeScorer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
import com.comet.opik.api.ScoreSource;
import com.comet.opik.api.Span;
import com.comet.opik.api.Trace;
import com.comet.opik.api.Visibility;
import com.comet.opik.api.attachment.EntityType;
import com.comet.opik.api.evaluators.AutomationRuleEvaluator;
import com.comet.opik.api.evaluators.LlmAsJudgeMessage;
Expand All @@ -19,6 +18,7 @@
import com.comet.opik.domain.evaluation.EvaluationRecorder;
import com.comet.opik.domain.evaluation.OnlineEvaluationRecorder;
import com.comet.opik.domain.evaluators.AutomationRuleEvaluatorService;
import com.comet.opik.domain.evaluators.OnlineScorePublisher;
import com.comet.opik.domain.evaluators.UserLog;
import com.comet.opik.domain.llm.ChatCompletionService;
import com.comet.opik.domain.llm.LlmProviderFactory;
Expand All @@ -35,7 +35,6 @@
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonReactiveClient;
import org.slf4j.Logger;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import ru.vyarus.dropwizard.guice.module.installer.feature.eager.EagerSingleton;
Expand Down Expand Up @@ -70,6 +69,7 @@ public class OnlineScoringTraceThreadLlmAsJudgeScorer extends OnlineScoringBaseS
private final ServiceTogglesConfig serviceTogglesConfig;
private final OnlineEvaluationRecorder onlineEvaluationRecorder;
private final AttachmentService attachmentService;
private final OnlineScorePublisher onlineScorePublisher;

@Inject
public OnlineScoringTraceThreadLlmAsJudgeScorer(@NonNull @Config("onlineScoring") OnlineScoringConfig config,
Expand All @@ -85,7 +85,8 @@ public OnlineScoringTraceThreadLlmAsJudgeScorer(@NonNull @Config("onlineScoring"
@NonNull AgenticScoringService agenticScoringService,
@NonNull SpanService spanService,
@NonNull OnlineEvaluationRecorder onlineEvaluationRecorder,
@NonNull AttachmentService attachmentService) {
@NonNull AttachmentService attachmentService,
@NonNull OnlineScorePublisher onlineScorePublisher) {
super(config, redisson, feedbackScoreService, traceService, spanService, TRACE_THREAD_LLM_AS_JUDGE,
Constants.TRACE_THREAD_LLM_AS_JUDGE);
this.aiProxyService = aiProxyService;
Expand All @@ -97,6 +98,7 @@ public OnlineScoringTraceThreadLlmAsJudgeScorer(@NonNull @Config("onlineScoring"
this.serviceTogglesConfig = serviceTogglesConfig;
this.onlineEvaluationRecorder = onlineEvaluationRecorder;
this.attachmentService = attachmentService;
this.onlineScorePublisher = onlineScorePublisher;
this.userFacingLogger = UserFacingLoggingFactory.getLogger(OnlineScoringTraceThreadLlmAsJudgeScorer.class);
}

Expand All @@ -112,25 +114,14 @@ protected Mono<Void> score(@NonNull TraceThreadToScoreLlmAsJudge message) {
log.info("Message received with projectId: '{}', ruleId: '{}', threadIds: '{}' for workspace '{}'",
message.projectId(), message.ruleId(), message.threadIds(), message.workspaceId());

return Flux.fromIterable(message.threadIds())
// Score each thread id independently: a single thread's failure must not stop scoring the
// sibling thread ids. Per-thread errors are materialized (onErrorResume) so the flatMap
// completes for every thread; the batch's first failure is then re-surfaced below. This keeps
// the failure on the Mono error path handled by BaseRedisSubscriber.processMessage's
// onErrorResume — classified as a processing error, following the normal retryable/
// non-retryable path — instead of leaking into the enclosing onErrorContinue via Flux.flatMap
// (which would drop the element and count it as an "unexpected" error).
.flatMap(threadId -> processThreadScores(message, threadId)
.then(Mono.<Throwable>empty())
.onErrorResume(Mono::just))
.collectList()
.flatMap(errors -> errors.isEmpty() ? Mono.<Void>empty() : Mono.error(errors.getFirst()))
.contextWrite(context -> context.put(RequestContext.WORKSPACE_ID, message.workspaceId())
.put(RequestContext.USER_NAME, message.userName())
.put(RequestContext.VISIBILITY, Visibility.PRIVATE))
.doOnSuccess(unused -> log.info(
"Processed trace threads for projectId '{}', ruleId '{}' for workspace '{}'",
message.projectId(), message.ruleId(), message.workspaceId()))
// The success log sits inside the scoring callback: a migrated entry completes this chain without
// scoring anything, so a doOnSuccess out here would claim work that never happened.
return migrateOrScoreThreadIds(message, message.threadIds(), onlineScorePublisher,
threadId -> message.toBuilder().threadIds(List.of(threadId)).build(),
threadId -> processThreadScores(message, threadId)
.doOnSuccess(unused -> log.info(
"Processed trace thread '{}' for projectId '{}', ruleId '{}' for workspace '{}'",
threadId, message.projectId(), message.ruleId(), message.workspaceId())))
.doOnError(error -> log.error(
"Error processing trace thread for projectId '{}', ruleId '{}' for workspace '{}'",
message.projectId(), message.ruleId(), message.workspaceId(), error))
Expand Down
35 changes: 13 additions & 22 deletions ...t/opik/api/resources/v1/events/OnlineScoringTraceThreadUserDefinedMetricPythonScorer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,14 @@
import com.comet.opik.api.ScoreSource;
import com.comet.opik.api.Span;
import com.comet.opik.api.Trace;
import com.comet.opik.api.Visibility;
import com.comet.opik.api.evaluators.AutomationRuleEvaluator;
import com.comet.opik.api.events.TraceThreadToScoreUserDefinedMetricPython;
import com.comet.opik.domain.FeedbackScoreService;
import com.comet.opik.domain.ProjectService;
import com.comet.opik.domain.SpanService;
import com.comet.opik.domain.TraceService;
import com.comet.opik.domain.evaluators.AutomationRuleEvaluatorService;
import com.comet.opik.domain.evaluators.OnlineScorePublisher;
import com.comet.opik.domain.evaluators.UserLog;
import com.comet.opik.domain.evaluators.python.PythonEvaluatorService;
import com.comet.opik.domain.threads.TraceThreadService;
Expand All @@ -26,7 +26,6 @@
import org.apache.commons.lang3.tuple.Pair;
import org.redisson.api.RedissonReactiveClient;
import org.slf4j.Logger;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import ru.vyarus.dropwizard.guice.module.installer.feature.eager.EagerSingleton;
Expand Down Expand Up @@ -62,6 +61,7 @@ public class OnlineScoringTraceThreadUserDefinedMetricPythonScorer
private final ProjectService projectService;
private final AutomationRuleEvaluatorService automationRuleEvaluatorService;
private final AgenticScoringService agenticScoringService;
private final OnlineScorePublisher onlineScorePublisher;

@Inject
public OnlineScoringTraceThreadUserDefinedMetricPythonScorer(
Expand All @@ -75,7 +75,8 @@ public OnlineScoringTraceThreadUserDefinedMetricPythonScorer(
@NonNull ProjectService projectService,
@NonNull AutomationRuleEvaluatorService automationRuleEvaluatorService,
@NonNull SpanService spanService,
@NonNull AgenticScoringService agenticScoringService) {
@NonNull AgenticScoringService agenticScoringService,
@NonNull OnlineScorePublisher onlineScorePublisher) {
super(config, redisson, feedbackScoreService, traceService, spanService,
TRACE_THREAD_USER_DEFINED_METRIC_PYTHON,
Constants.TRACE_THREAD_USER_DEFINED_METRIC_PYTHON);
Expand All @@ -85,6 +86,7 @@ public OnlineScoringTraceThreadUserDefinedMetricPythonScorer(
this.projectService = projectService;
this.automationRuleEvaluatorService = automationRuleEvaluatorService;
this.agenticScoringService = agenticScoringService;
this.onlineScorePublisher = onlineScorePublisher;
this.userFacingLogger = UserFacingLoggingFactory
.getLogger(OnlineScoringTraceThreadUserDefinedMetricPythonScorer.class);
}
Expand All @@ -104,25 +106,14 @@ protected Mono<Void> score(@NonNull TraceThreadToScoreUserDefinedMetricPython me
log.info("Message received with projectId '{}', ruleId '{}' for workspace '{}'",
message.projectId(), message.ruleId(), message.workspaceId());

return Flux.fromIterable(message.threadIds())
// Score each thread id independently: a single thread's failure must not stop scoring the
// sibling thread ids. Per-thread errors are materialized (onErrorResume) so the flatMap
// completes for every thread; the batch's first failure is then re-surfaced below. This keeps
// the failure on the Mono error path handled by BaseRedisSubscriber.processMessage's
// onErrorResume — classified as a processing error, following the normal retryable/
// non-retryable path — instead of leaking into the enclosing onErrorContinue via Flux.flatMap
// (which would drop the element and count it as an "unexpected" error).
.flatMap(threadId -> processThreadScores(message, threadId)
.then(Mono.<Throwable>empty())
.onErrorResume(Mono::just))
.collectList()
.flatMap(errors -> errors.isEmpty() ? Mono.<Void>empty() : Mono.error(errors.getFirst()))
.contextWrite(context -> context.put(RequestContext.WORKSPACE_ID, message.workspaceId())
.put(RequestContext.USER_NAME, message.userName())
.put(RequestContext.VISIBILITY, Visibility.PRIVATE))
.doOnSuccess(unused -> log.info(
"Processed trace threads for projectId '{}', ruleId '{}' for workspace '{}'",
message.projectId(), message.ruleId(), message.workspaceId()))
// The success log sits inside the scoring callback: a migrated entry completes this chain without
// scoring anything, so a doOnSuccess out here would claim work that never happened.
return migrateOrScoreThreadIds(message, message.threadIds(), onlineScorePublisher,
threadId -> message.toBuilder().threadIds(List.of(threadId)).build(),
threadId -> processThreadScores(message, threadId)
.doOnSuccess(unused -> log.info(
"Processed trace thread '{}' for projectId '{}', ruleId '{}' for workspace '{}'",
threadId, message.projectId(), message.ruleId(), message.workspaceId())))
.doOnError(error -> log.error(
"Error processing trace thread for projectId '{}', ruleId '{}' for workspace '{}'",
message.projectId(), message.ruleId(), message.workspaceId(), error))
Expand Down
8 changes: 4 additions & 4 deletions .../opik-backend/src/main/java/com/comet/opik/domain/evaluators/ManualEvaluationService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,8 +185,8 @@ private Mono<Integer> evaluateTraces(List<UUID> traceIds, List<AutomationRuleEva
.toList();
Mono<Void> traceThreadMono = Flux.fromIterable(traceThreadRules)
.flatMap(rule -> {
log.info("Enqueueing trace-thread evaluation for rule '{}' with '{}' trace IDs", rule.getId(),
traceIdStrings.size());
log.info("Enqueueing '{}' trace-thread evaluation messages, one per trace ID, for rule '{}'",
traceIdStrings.size(), rule.getId());
return onlineScorePublisher.enqueueThreadMessage(traceIdStrings, rule, projectId, workspaceId,
userName);
})
Expand Down Expand Up @@ -384,8 +384,8 @@ private Mono<Integer> evaluateThreads(List<UUID> threadModelIds, List<Automation
// reactive context. enqueueThreadMessage does a blocking rule lookup, so defer onto boundedElastic.
return Flux.fromIterable(rules)
.flatMap(rule -> {
log.info("Enqueueing evaluation for rule '{}' with '{}' thread IDs", rule.getId(),
threadIds.size());
log.info("Enqueueing '{}' evaluation messages, one per thread ID, for rule '{}'",
threadIds.size(), rule.getId());
return onlineScorePublisher.enqueueThreadMessage(threadIds, rule, projectId,
workspaceId, userName);
})
Expand Down
Loading
Loading