Skip to content

Commit b74db69

Browse files
miguelgrcmiguelgrc
authored
[OPIK-7274] [BE] fix: make Ollie report stale timeout configurable and add pipeline metrics (#7576)
* [OPIK-7274] [BE] fix: extend Ollie report stale timeout to 30m and add pipeline metrics Raise the stale-report sweep threshold from 10 to 30 minutes: reports queued behind orchestrator pod provisioning routinely take longer than 10 minutes, so the old threshold marked still-in-flight reports as failed. Also instrument the daily-report flow with OTel metrics so failures are observable from opik-backend instead of only through logs: - triggered / trigger_error (OllieDailyReportJob), counting only reports actually created (skips no longer inflate the triggered count) - finished{result=completed|failed|trigger_failed} + end_to_end and scheduled_to_completion duration histograms (ReportService); the async trigger-failure callback now records trigger_failed - stale_swept (StaleReportCleanupJob) as the safety-net signal Completion metrics are recorded after the write transaction commits and never fail the callback. * fix(reports): dimension daily-report metrics and count manual triggers Address PR review on the daily-report metrics (metrics-instrumentation contract): - Dimension triggered, trigger_error, finished, both duration histograms, and stale_swept by workspace_id + workspace_name (paired, name->id fallback); trigger_error also carries error_type. - Emit the triggered counter from createAndTriggerReport so manual /generate triggers are counted alongside scheduled ones, keeping the funnel consistent with finished/stale_swept (which already count both). - Per-workspace stale_swept: failStaleReports returns per-workspace counts; the 30-minute cutoff is a single STALE_THRESHOLD_MINUTES constant bound into both stale queries. - Hoist result/workspace AttributeKeys to constants; add workspaceId/projectId to markReportFailed logs; drop the duplicate job-side stale-sweep INFO log. * fix(reports): configurable stale timeout, drop scheduled_to_completion Address review (thiagohora): - Make the stale-report sweep timeout configurable via reportGeneration.staleReportTimeoutMinutes (default 10 = unchanged behavior; tunable per-env without redeploy). - Drop the scheduled_to_completion_duration histogram. It only added the bounded cron-pickup delay (<=10m) over end_to_end, required a blocking preference read + schedule parse on the completion callback, and was polluted by manual triggers on daily-enabled projects. Removing it makes recordCompletionMetrics a cheap non-blocking side-effect (finished + end_to_end only), so updateReport uses doOnNext again and there's no callback DB read / parse-failure path.
1 parent d5b6bf4 commit b74db69

5 files changed

Lines changed: 197 additions & 41 deletions

File tree

‎apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/jobs/OllieDailyReportJob.java‎

Lines changed: 33 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,16 @@
55
import com.comet.opik.infrastructure.lock.LockService;
66
import io.dropwizard.jobs.Job;
77
import io.dropwizard.jobs.annotations.On;
8+
import io.opentelemetry.api.GlobalOpenTelemetry;
9+
import io.opentelemetry.api.common.AttributeKey;
10+
import io.opentelemetry.api.common.Attributes;
11+
import io.opentelemetry.api.metrics.LongCounter;
12+
import io.opentelemetry.api.metrics.Meter;
813
import jakarta.inject.Inject;
914
import jakarta.inject.Singleton;
1015
import lombok.NonNull;
11-
import lombok.RequiredArgsConstructor;
1216
import lombok.extern.slf4j.Slf4j;
17+
import org.apache.commons.lang3.StringUtils;
1318
import org.quartz.DisallowConcurrentExecution;
1419
import org.quartz.JobExecutionContext;
1520
import reactor.core.publisher.Mono;
@@ -20,22 +25,38 @@
2025

2126
import static com.comet.opik.infrastructure.lock.LockService.Lock;
2227

23-
/**
24-
* Runs every 10 minutes and triggers report generation for projects
25-
* whose schedule_time falls within the previous 10-minute window.
26-
*/
2728
@Slf4j
2829
@Singleton
2930
@DisallowConcurrentExecution
3031
@On(value = "0 0/10 * * * ?", timeZone = "UTC")
31-
@RequiredArgsConstructor(onConstructor_ = @Inject)
3232
public class OllieDailyReportJob extends Job {
3333

3434
private static final int WINDOW_MINUTES = 10;
3535
private static final Lock JOB_LOCK = new Lock("daily_report_job:lock");
3636

37-
private final @NonNull ReportService reportService;
38-
private final @NonNull LockService lockService;
37+
private static final AttributeKey<String> WORKSPACE_ID_KEY = AttributeKey.stringKey("workspace_id");
38+
private static final AttributeKey<String> WORKSPACE_NAME_KEY = AttributeKey.stringKey("workspace_name");
39+
private static final AttributeKey<String> ERROR_TYPE_KEY = AttributeKey.stringKey("error_type");
40+
41+
private final ReportService reportService;
42+
private final LockService lockService;
43+
44+
private final LongCounter triggerErrorCounter;
45+
46+
@Inject
47+
public OllieDailyReportJob(
48+
@NonNull ReportService reportService,
49+
@NonNull LockService lockService) {
50+
this.reportService = reportService;
51+
this.lockService = lockService;
52+
53+
Meter meter = GlobalOpenTelemetry.get().getMeter("opik.daily_report");
54+
55+
this.triggerErrorCounter = meter
56+
.counterBuilder("opik.daily_report.trigger_error")
57+
.setDescription("Number of report trigger failures")
58+
.build();
59+
}
3960

4061
record TimeWindow(String start, String end) {
4162
}
@@ -82,6 +103,10 @@ private void triggerReports(String windowStart, String windowEnd) {
82103
reportService.createAndTriggerReport(pref.workspaceId(), pref.workspaceName(), pref.projectId());
83104
} catch (Exception e) {
84105
log.error("Failed to trigger report for project '{}'", pref.projectId(), e);
106+
triggerErrorCounter.add(1, Attributes.of(
107+
WORKSPACE_ID_KEY, pref.workspaceId(),
108+
WORKSPACE_NAME_KEY, StringUtils.defaultIfBlank(pref.workspaceName(), pref.workspaceId()),
109+
ERROR_TYPE_KEY, e.getClass().getSimpleName()));
85110
}
86111
}
87112
}

‎apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/jobs/StaleReportCleanupJob.java‎

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,14 @@
44
import com.comet.opik.infrastructure.lock.LockService;
55
import io.dropwizard.jobs.Job;
66
import io.dropwizard.jobs.annotations.Every;
7+
import io.opentelemetry.api.GlobalOpenTelemetry;
8+
import io.opentelemetry.api.common.AttributeKey;
9+
import io.opentelemetry.api.common.Attributes;
10+
import io.opentelemetry.api.metrics.LongCounter;
11+
import io.opentelemetry.api.metrics.Meter;
712
import jakarta.inject.Inject;
813
import jakarta.inject.Singleton;
914
import lombok.NonNull;
10-
import lombok.RequiredArgsConstructor;
1115
import lombok.extern.slf4j.Slf4j;
1216
import org.quartz.DisallowConcurrentExecution;
1317
import org.quartz.JobExecutionContext;
@@ -21,19 +25,41 @@
2125
@Singleton
2226
@DisallowConcurrentExecution
2327
@Every("15min")
24-
@RequiredArgsConstructor(onConstructor_ = @Inject)
2528
public class StaleReportCleanupJob extends Job {
2629

2730
private static final Lock JOB_LOCK = new Lock("stale_report_cleanup:lock");
2831

29-
private final @NonNull ReportService reportService;
30-
private final @NonNull LockService lockService;
32+
private static final AttributeKey<String> WORKSPACE_ID_KEY = AttributeKey.stringKey("workspace_id");
33+
private static final AttributeKey<String> WORKSPACE_NAME_KEY = AttributeKey.stringKey("workspace_name");
34+
35+
private final ReportService reportService;
36+
private final LockService lockService;
37+
38+
private final LongCounter staleReportsCounter;
39+
40+
@Inject
41+
public StaleReportCleanupJob(
42+
@NonNull ReportService reportService,
43+
@NonNull LockService lockService) {
44+
this.reportService = reportService;
45+
this.lockService = lockService;
46+
47+
Meter meter = GlobalOpenTelemetry.get().getMeter("opik.daily_report");
48+
49+
this.staleReportsCounter = meter
50+
.counterBuilder("opik.daily_report.stale_swept")
51+
.setDescription("Number of stale reports marked as failed")
52+
.build();
53+
}
3154

3255
@Override
3356
public void doJob(JobExecutionContext context) {
3457
lockService.bestEffortLock(
3558
JOB_LOCK,
36-
Mono.fromRunnable(reportService::failStaleReports),
59+
Mono.fromRunnable(() -> reportService.failStaleReports()
60+
.forEach((workspaceId, count) -> staleReportsCounter.add(count, Attributes.of(
61+
WORKSPACE_ID_KEY, workspaceId,
62+
WORKSPACE_NAME_KEY, workspaceId)))),
3763
Mono.defer(() -> {
3864
log.debug("Could not acquire lock for stale report cleanup, another instance is running");
3965
return Mono.empty();

‎apps/opik-backend/src/main/java/com/comet/opik/domain/OllieReportDAO.java‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import org.jdbi.v3.sqlobject.statement.SqlQuery;
1313
import org.jdbi.v3.sqlobject.statement.SqlUpdate;
1414

15+
import java.time.Instant;
1516
import java.util.List;
1617
import java.util.UUID;
1718

@@ -65,9 +66,21 @@ SELECT COUNT(*) FROM ollie_reports
6566
long countByProjectId(@Bind("workspaceId") String workspaceId,
6667
@Bind("projectId") UUID projectId);
6768

69+
@SqlQuery("""
70+
SELECT created_at FROM ollie_reports
71+
WHERE id = :id AND workspace_id = :workspaceId
72+
""")
73+
Instant getCreatedAt(@Bind("id") UUID id, @Bind("workspaceId") String workspaceId);
74+
75+
@SqlQuery("""
76+
SELECT workspace_id FROM ollie_reports
77+
WHERE status = 'pending' AND created_at < DATE_SUB(NOW(), INTERVAL :staleMinutes MINUTE)
78+
""")
79+
List<String> findStalePendingWorkspaceIds(@Bind("staleMinutes") int staleMinutes);
80+
6881
@SqlUpdate("""
6982
UPDATE ollie_reports SET status = 'failed'
70-
WHERE status = 'pending' AND created_at < DATE_SUB(NOW(), INTERVAL 10 MINUTE)
83+
WHERE status = 'pending' AND created_at < DATE_SUB(NOW(), INTERVAL :staleMinutes MINUTE)
7184
""")
72-
int failStaleReports();
85+
int failStaleReports(@Bind("staleMinutes") int staleMinutes);
7386
}

‎apps/opik-backend/src/main/java/com/comet/opik/domain/ReportService.java‎

Lines changed: 115 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -3,36 +3,93 @@
33
import com.comet.opik.api.OllieReport.OllieReportPage;
44
import com.comet.opik.api.OllieReport.ReportStatus;
55
import com.comet.opik.api.ReportPreference;
6+
import com.comet.opik.infrastructure.ReportGenerationConfig;
67
import com.comet.opik.infrastructure.auth.RequestContext;
78
import com.fasterxml.jackson.databind.JsonNode;
9+
import io.opentelemetry.api.GlobalOpenTelemetry;
10+
import io.opentelemetry.api.common.AttributeKey;
11+
import io.opentelemetry.api.common.Attributes;
12+
import io.opentelemetry.api.metrics.LongCounter;
13+
import io.opentelemetry.api.metrics.LongHistogram;
14+
import io.opentelemetry.api.metrics.Meter;
815
import jakarta.inject.Inject;
916
import jakarta.inject.Provider;
1017
import jakarta.inject.Singleton;
1118
import jakarta.ws.rs.NotFoundException;
1219
import lombok.NonNull;
13-
import lombok.RequiredArgsConstructor;
1420
import lombok.extern.slf4j.Slf4j;
21+
import org.apache.commons.lang3.StringUtils;
1522
import reactor.core.publisher.Mono;
1623
import reactor.core.scheduler.Schedulers;
24+
import ru.vyarus.dropwizard.guice.module.yaml.bind.Config;
1725
import ru.vyarus.guicey.jdbi3.tx.TransactionTemplate;
1826

27+
import java.time.Instant;
1928
import java.util.List;
29+
import java.util.Map;
2030
import java.util.Set;
2131
import java.util.UUID;
32+
import java.util.stream.Collectors;
2233

2334
import static com.comet.opik.infrastructure.db.TransactionTemplateAsync.READ_ONLY;
2435
import static com.comet.opik.infrastructure.db.TransactionTemplateAsync.WRITE;
36+
import static io.opentelemetry.api.common.AttributeKey.stringKey;
2537

2638
@Singleton
27-
@RequiredArgsConstructor(onConstructor_ = @Inject)
2839
@Slf4j
2940
public class ReportService {
3041

31-
private final @NonNull TransactionTemplate transactionTemplate;
32-
private final @NonNull IdGenerator idGenerator;
33-
private final @NonNull Provider<RequestContext> requestContext;
34-
private final @NonNull OrchestratorClient orchestratorClient;
35-
private final @NonNull ProjectService projectService;
42+
private static final AttributeKey<String> RESULT_KEY = stringKey("result");
43+
private static final AttributeKey<String> WORKSPACE_ID_KEY = stringKey("workspace_id");
44+
private static final AttributeKey<String> WORKSPACE_NAME_KEY = stringKey("workspace_name");
45+
46+
private final TransactionTemplate transactionTemplate;
47+
private final IdGenerator idGenerator;
48+
private final Provider<RequestContext> requestContext;
49+
private final OrchestratorClient orchestratorClient;
50+
private final ProjectService projectService;
51+
private final ReportGenerationConfig reportGenerationConfig;
52+
53+
private final LongCounter triggeredCounter;
54+
private final LongCounter finishedCounter;
55+
private final LongHistogram endToEndDuration;
56+
57+
@Inject
58+
public ReportService(
59+
@NonNull TransactionTemplate transactionTemplate,
60+
@NonNull IdGenerator idGenerator,
61+
@NonNull Provider<RequestContext> requestContext,
62+
@NonNull OrchestratorClient orchestratorClient,
63+
@NonNull ProjectService projectService,
64+
@NonNull @Config("reportGeneration") ReportGenerationConfig reportGenerationConfig) {
65+
this.transactionTemplate = transactionTemplate;
66+
this.idGenerator = idGenerator;
67+
this.requestContext = requestContext;
68+
this.orchestratorClient = orchestratorClient;
69+
this.projectService = projectService;
70+
this.reportGenerationConfig = reportGenerationConfig;
71+
72+
Meter meter = GlobalOpenTelemetry.get().getMeter("opik.daily_report");
73+
74+
this.triggeredCounter = meter
75+
.counterBuilder("opik.daily_report.triggered")
76+
.setDescription("Number of reports triggered for generation (scheduled and manual)")
77+
.build();
78+
79+
this.finishedCounter = meter
80+
.counterBuilder("opik.daily_report.finished")
81+
.setDescription("Number of reports finalized via the completion callback or trigger failure, "
82+
+ "by result (completed / failed / trigger_failed); stale sweeps are counted separately "
83+
+ "by opik.daily_report.stale_swept")
84+
.build();
85+
86+
this.endToEndDuration = meter
87+
.histogramBuilder("opik.daily_report.end_to_end_duration")
88+
.setDescription("Time from report creation to completion callback")
89+
.setUnit("ms")
90+
.ofLongs()
91+
.build();
92+
}
3693

3794
public Mono<UUID> generateReport(@NonNull UUID projectId) {
3895
var ctx = requestContext.get();
@@ -70,25 +127,36 @@ public UUID createAndTriggerReport(@NonNull String workspaceId, @NonNull String
70127
orchestratorClient.triggerReportGeneration(
71128
reportId.toString(), projectId.toString(), projectName,
72129
workspaceName, customPrompt,
73-
() -> markReportFailed(reportId, workspaceId, projectId));
130+
() -> markReportFailed(reportId, workspaceId, workspaceName, projectId));
131+
132+
triggeredCounter.add(1, Attributes.of(
133+
WORKSPACE_ID_KEY, workspaceId,
134+
WORKSPACE_NAME_KEY, StringUtils.defaultIfBlank(workspaceName, workspaceId)));
74135

75136
return reportId;
76137
}
77138

78139
public Mono<Void> updateReport(@NonNull UUID projectId, @NonNull UUID reportId,
79140
@NonNull ReportStatus status, String content, String sessionId,
80141
JsonNode recommendedActions) {
81-
String workspaceId = requestContext.get().getWorkspaceId();
142+
var ctx = requestContext.get();
143+
String workspaceId = ctx.getWorkspaceId();
144+
String workspaceName = ctx.getWorkspaceName();
82145

83146
return Mono.fromCallable(() -> transactionTemplate.inTransaction(WRITE, handle -> {
84-
int updated = handle.attach(OllieReportDAO.class)
85-
.update(reportId, workspaceId, projectId, content, sessionId, recommendedActions,
86-
status.getValue());
147+
var dao = handle.attach(OllieReportDAO.class);
148+
149+
int updated = dao.update(reportId, workspaceId, projectId, content, sessionId, recommendedActions,
150+
status.getValue());
87151
if (updated == 0) {
88152
throw new NotFoundException("Report not found or already processed: " + reportId);
89153
}
90-
return null;
91-
})).subscribeOn(Schedulers.boundedElastic()).then();
154+
155+
return dao.getCreatedAt(reportId, workspaceId);
156+
}))
157+
.doOnNext(createdAt -> recordCompletionMetrics(workspaceId, workspaceName, status, createdAt))
158+
.subscribeOn(Schedulers.boundedElastic())
159+
.then();
92160
}
93161

94162
public Mono<OllieReportPage> getReports(@NonNull UUID projectId, int page, int size) {
@@ -134,26 +202,47 @@ public List<ReportPreference> findEnabledPreferencesInTimeWindow(String windowSt
134202
.findAllEnabledInTimeWindow(windowStart, windowEnd));
135203
}
136204

137-
private void markReportFailed(UUID reportId, String workspaceId, UUID projectId) {
205+
private void recordCompletionMetrics(String workspaceId, String workspaceName, ReportStatus status,
206+
Instant createdAt) {
207+
String result = status == ReportStatus.COMPLETED ? "completed" : "failed";
208+
Attributes attrs = Attributes.of(
209+
RESULT_KEY, result,
210+
WORKSPACE_ID_KEY, workspaceId,
211+
WORKSPACE_NAME_KEY, StringUtils.defaultIfBlank(workspaceName, workspaceId));
212+
213+
finishedCounter.add(1, attrs);
214+
endToEndDuration.record(Instant.now().toEpochMilli() - createdAt.toEpochMilli(), attrs);
215+
}
216+
217+
private void markReportFailed(UUID reportId, String workspaceId, String workspaceName, UUID projectId) {
138218
try {
139-
transactionTemplate.inTransaction(WRITE, handle -> {
140-
handle.attach(OllieReportDAO.class)
141-
.update(reportId, workspaceId, projectId, null, null, null, ReportStatus.FAILED.getValue());
142-
return null;
143-
});
144-
log.info("Marked report '{}' as failed", reportId);
219+
int updated = transactionTemplate.inTransaction(WRITE, handle -> handle.attach(OllieReportDAO.class)
220+
.update(reportId, workspaceId, projectId, null, null, null, ReportStatus.FAILED.getValue()));
221+
if (updated > 0) {
222+
finishedCounter.add(1, Attributes.of(
223+
RESULT_KEY, "trigger_failed",
224+
WORKSPACE_ID_KEY, workspaceId,
225+
WORKSPACE_NAME_KEY, StringUtils.defaultIfBlank(workspaceName, workspaceId)));
226+
log.info("Marked report as failed reportId='{}' workspaceId='{}' projectId='{}'",
227+
reportId, workspaceId, projectId);
228+
}
145229
} catch (Exception e) {
146-
log.error("Failed to mark report '{}' as failed", reportId, e);
230+
log.error("Failed to mark report as failed reportId='{}' workspaceId='{}' projectId='{}'",
231+
reportId, workspaceId, projectId, e);
147232
}
148233
}
149234

150-
public void failStaleReports() {
151-
transactionTemplate.inTransaction(WRITE, handle -> {
152-
int failed = handle.attach(OllieReportDAO.class).failStaleReports();
235+
public Map<String, Long> failStaleReports() {
236+
return transactionTemplate.inTransaction(WRITE, handle -> {
237+
var dao = handle.attach(OllieReportDAO.class);
238+
Map<String, Long> sweptByWorkspace = dao
239+
.findStalePendingWorkspaceIds(reportGenerationConfig.getStaleReportTimeoutMinutes()).stream()
240+
.collect(Collectors.groupingBy(id -> id, Collectors.counting()));
241+
int failed = dao.failStaleReports(reportGenerationConfig.getStaleReportTimeoutMinutes());
153242
if (failed > 0) {
154243
log.info("Marked {} stale pending reports as failed", failed);
155244
}
156-
return null;
245+
return sweptByWorkspace;
157246
});
158247
}
159248
}

‎apps/opik-backend/src/main/java/com/comet/opik/infrastructure/ReportGenerationConfig.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,4 +8,7 @@ public class ReportGenerationConfig {
88

99
@JsonProperty
1010
private String url = "";
11+
12+
@JsonProperty
13+
private int staleReportTimeoutMinutes = 10;
1114
}

0 commit comments

Comments
 (0)