From 48d2f877447417eacd59343f01611959ba20b3a8 Mon Sep 17 00:00:00 2001 From: Daniel Dimenshtein Date: Wed, 2 Sep 2026 11:20:55 +0300 Subject: [PATCH 1/2] [OPIK-8202] [BE] fix: harden ClickHouseAppender against dropped user logs Investigating the intermittent Integration Group 14 failures in AutomationRuleEvaluatorsResourceTest surfaced two ways the appender can lose user-facing log rows outright. Both are fixed here. flushLogs drains the batch out of the queue before inserting, so any failure past that point loses those events: - a failed insert was only logged, never retried or re-queued - a throw escaping flushLogs cancelled the scheduleAtFixedRate task permanently, silently stopping log persistence for the rest of the JVM's life Retry the insert, re-queue the batch when the retries are exhausted (including on a synchronous throw), and keep the periodic flush alive across failures. The two new unit tests fail without these changes and pass with them; they need no containers. The awaits in AutomationRuleEvaluatorsResourceTest now state their timeout explicitly instead of inheriting Awaitility's implicit 10s default. This is a readability change, not a fix: the flush interval is 500ms, and the failing tests still fail with a 30s wait, so the rows are not merely late. This does NOT confirm a fix for the OPIK-8202 flake. Locally the affected tests fail on unmodified main and still fail with these changes, at rates too close to distinguish over three runs each. Local runs are also a poor proxy here -- they fail far more often than CI does, so something environmental may be involved. The root cause of the flake is still unidentified: in failing runs the log write path reports no errors, and the sampler scores the rule the test expects it to skip, so the awaited row is never produced rather than lost. Co-Authored-By: Claude Opus 5 (1M context) --- .../log/ClickHouseAppender.java | 59 ++++++++-- .../AutomationRuleEvaluatorsResourceTest.java | 28 +++-- .../log/ClickHouseAppenderTest.java | 101 ++++++++++++++++++ 3 files changed, 168 insertions(+), 20 deletions(-) create mode 100644 apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java diff --git a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java index 4c954f10369..735a3a733b6 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java @@ -8,6 +8,7 @@ import lombok.NonNull; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import reactor.util.retry.Retry; import java.time.Duration; import java.util.ArrayList; @@ -27,6 +28,9 @@ @Slf4j class ClickHouseAppender extends AppenderBase { + private static final long INSERT_RETRY_ATTEMPTS = 3; + private static final Duration INSERT_RETRY_MIN_BACKOFF = Duration.ofMillis(100); + private static ClickHouseAppender instance; public static synchronized ClickHouseAppender init(@NonNull UserLogTableFactory userLogTableFactory, int batchSize, @@ -58,13 +62,23 @@ private static void setInstance(ClickHouseAppender instance) { @Override public void start() { - // Background flush thread - scheduler.get().scheduleAtFixedRate(this::flushLogs, flushIntervalDuration.toMillis(), + // Background flush thread. Wrapped in safeFlushLogs because scheduleAtFixedRate silently + // cancels the task forever if it ever throws, which would stop persisting user logs for the + // rest of the JVM's life. + scheduler.get().scheduleAtFixedRate(this::safeFlushLogs, flushIntervalDuration.toMillis(), flushIntervalDuration.toMillis(), TimeUnit.MILLISECONDS); super.start(); } + private void safeFlushLogs() { + try { + flushLogs(); + } catch (Exception e) { + log.error("Failed to flush logs", e); + } + } + private void flushLogs() { if (logQueue.isEmpty()) return; @@ -87,16 +101,43 @@ private void flushLogs() { UserLogTableFactory.UserLogTableDAO tableDAO = userLogTableFactory .getDAO(UserLog.valueOf(userLog)); - tableDAO - .saveAll(events) - .subscribe( - noop -> { - }, - e -> log.error("Failed to insert logs", e)); + // The batch is already drained out of the queue, so a failed insert would lose + // these events outright. Retry transient failures, and put the events back on + // the queue if the retries are exhausted so a later flush can pick them up. + // saveAll is called inside the try so that a synchronous throw re-queues too, + // rather than only errors signalled through the returned Mono. + try { + tableDAO + .saveAll(events) + .retryWhen(Retry.backoff(INSERT_RETRY_ATTEMPTS, INSERT_RETRY_MIN_BACKOFF)) + .subscribe( + noop -> { + }, + e -> { + log.error("Failed to insert logs", e); + requeue(events); + }); + } catch (Exception e) { + log.error("Failed to insert logs", e); + requeue(events); + } } }); } + private void requeue(List events) { + if (!running) { + log.warn("ClickHouseAppender is stopped, dropping '{}' logs after failed insert", events.size()); + return; + } + + for (ILoggingEvent event : events) { + if (!logQueue.offer(event)) { + log.warn("Log queue is full, dropping log: {}", event.getFormattedMessage()); + } + } + } + @Override protected void append(ILoggingEvent event) { if (!running) { @@ -110,7 +151,7 @@ protected void append(ILoggingEvent event) { } if (logQueue.size() >= batchSize) { - scheduler.get().execute(this::flushLogs); + scheduler.get().execute(this::safeFlushLogs); } } diff --git a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/AutomationRuleEvaluatorsResourceTest.java b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/AutomationRuleEvaluatorsResourceTest.java index 502f528bdee..9b17977e01c 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/AutomationRuleEvaluatorsResourceTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/priv/AutomationRuleEvaluatorsResourceTest.java @@ -99,6 +99,7 @@ import java.util.Optional; import java.util.Set; import java.util.UUID; +import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.function.Predicate; import java.util.stream.Collectors; @@ -244,6 +245,11 @@ def score( private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + // The evaluator log rows these tests wait for are written by the async ClickHouseAppender, which + // flushes every 500ms. Stated explicitly rather than relying on Awaitility's implicit 10s default, + // which is easy to miss when reading the waits below. + private static final int AWAIT_TIMEOUT_SECONDS = 30; + private final RedisContainer redis = RedisContainerUtils.newRedisContainer(); private final MySQLContainer mysql = MySQLContainerUtils.newMySQLContainer(); private final GenericContainer zookeeper = ClickHouseContainerUtils.newZookeeperContainer(); @@ -641,7 +647,7 @@ void getLogsPerRuleEvaluators__whenSessionTokenIsPresent__thenReturnProperRespon .build(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { try (var actualResponse = evaluatorsResourceClient.getLogsWithSessionToken( id, sessionToken, workspaceName)) { if (isAuthorized) { @@ -1435,7 +1441,7 @@ void getLogsLlmAsJudgeScorer(LlmProviderFactory llmProviderFactory) throws JsonP .build(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { var logPage = evaluatorsResourceClient.getLogs(id, WORKSPACE_NAME, API_KEY); assertTraceLogResponse(logPage, id, trace); }); @@ -1472,7 +1478,7 @@ void getLogsTraceThreadLlmAsJudgeScorer(LlmProviderFactory llmProviderFactory) t Instant createdAt = trace.createdAt(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { TraceThread traceThread = traceResourceClient.getTraceThread(trace.threadId(), projectId, API_KEY, WORKSPACE_NAME); @@ -1531,7 +1537,7 @@ void getLogsUserDefinedMetricPythonScorer() throws JsonProcessingException { .build(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { var logPage = evaluatorsResourceClient.getLogs(id, WORKSPACE_NAME, API_KEY); assertTraceLogResponse(logPage, id, trace); }); @@ -1590,7 +1596,7 @@ void getLogsTraceThreadUserDefinedMetricPythonScorer() throws JsonProcessingExce traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); // Then - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { TraceThread traceThread = traceResourceClient.getTraceThread(trace.threadId(), projectId, API_KEY, WORKSPACE_NAME); @@ -1641,7 +1647,7 @@ void getLogsTraceSkipped() { .build(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { var logPagePython = evaluatorsResourceClient.getLogs(idPython, WORKSPACE_NAME, API_KEY); assertLogResponse(logPagePython, idPython, evaluatorPython, trace); var logPageLlm = evaluatorsResourceClient.getLogs(idLlm, WORKSPACE_NAME, API_KEY); @@ -1693,7 +1699,7 @@ void getLogsTraceThreadSkipped() { Instant createdAt = trace.createdAt(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { TraceThread traceThread = traceResourceClient.getTraceThread(trace.threadId(), projectId, API_KEY, WORKSPACE_NAME); @@ -1783,7 +1789,7 @@ void getLogsTraceSkippedDueToDisabledRule() { .build(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { var logPagePython = evaluatorsResourceClient.getLogs(idPython, WORKSPACE_NAME, API_KEY); assertDisabledRuleLogResponse(logPagePython, idPython, evaluatorPython, trace); var logPageLlm = evaluatorsResourceClient.getLogs(idLlm, WORKSPACE_NAME, API_KEY); @@ -1835,7 +1841,7 @@ void getLogsTraceThreadSkippedDueToDisabledRule() { Instant createdAt = trace.createdAt(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { TraceThread traceThread = traceResourceClient.getTraceThread(trace.threadId(), projectId, API_KEY, WORKSPACE_NAME); @@ -1894,7 +1900,7 @@ void mixedEnabledAndDisabledRules() { .build(); traceResourceClient.createTrace(trace, API_KEY, WORKSPACE_NAME); - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { // Enabled rule should generate sampling rate message (skipped due to 0% rate) var enabledLogPage = evaluatorsResourceClient.getLogs(enabledId, WORKSPACE_NAME, API_KEY); assertLogResponse(enabledLogPage, enabledId, enabledRule, trace); @@ -1933,7 +1939,7 @@ void disabledRuleDoesNotConsumeResources() { } // All traces should be skipped with "disabled" message, none with sampling rate message - Awaitility.await().untilAsserted(() -> { + Awaitility.await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).untilAsserted(() -> { var logPage = evaluatorsResourceClient.getLogs(disabledId, WORKSPACE_NAME, API_KEY); assertLogPage(logPage, 5); // Should have 5 log entries diff --git a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java new file mode 100644 index 00000000000..73a5cecea87 --- /dev/null +++ b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java @@ -0,0 +1,101 @@ +package com.comet.opik.infrastructure.log; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.LoggerContext; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.classic.spi.LoggingEvent; +import com.comet.opik.domain.evaluators.UserLog; +import com.comet.opik.infrastructure.log.tables.UserLogTableFactory; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; + +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +@DisplayName("ClickHouse Appender") +class ClickHouseAppenderTest { + + private static final Duration FLUSH_INTERVAL = Duration.ofMillis(50); + private static final int BATCH_SIZE = 1000; + private static final int AWAIT_TIMEOUT_SECONDS = 10; + + private ClickHouseAppender appender; + + @AfterEach + void tearDown() { + if (appender != null) { + appender.stop(); + } + } + + @Test + @DisplayName("when the insert fails transiently, then the events are retried and still persisted") + void whenInsertFailsTransiently__thenEventsAreRetriedAndPersisted() { + var attempts = new AtomicInteger(); + var persisted = new ConcurrentLinkedQueue(); + + // Fail the first attempt, then accept. Without the retry, the drained batch would be lost. + startAppender(events -> Mono.defer(() -> { + if (attempts.incrementAndGet() == 1) { + return Mono.error(new IllegalStateException("transient failure")); + } + persisted.addAll(events); + return Mono.empty(); + })); + + appender.doAppend(event("a message")); + + await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(persisted).hasSize(1)); + } + + @Test + @DisplayName("when a flush throws, then subsequent flushes still run") + void whenFlushThrows__thenSubsequentFlushesStillRun() { + var persisted = new ConcurrentLinkedQueue(); + var failNext = new AtomicInteger(1); + + // A throw escaping into scheduleAtFixedRate would cancel the periodic flush permanently, + // silently stopping log persistence for the rest of the JVM's life. + startAppender(events -> { + if (failNext.getAndSet(0) == 1) { + throw new IllegalStateException("failure inside flush"); + } + persisted.addAll(events); + return Mono.empty(); + }); + + appender.doAppend(event("first")); + + await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(persisted).isNotEmpty()); + } + + private void startAppender(SaveAll saveAll) { + UserLogTableFactory tableFactory = userLog -> saveAll::apply; + + appender = ClickHouseAppender.init(tableFactory, BATCH_SIZE, FLUSH_INTERVAL, new LoggerContext()); + } + + private ILoggingEvent event(String message) { + var event = new LoggingEvent(); + event.setLevel(Level.INFO); + event.setMessage(message); + event.setMDCPropertyMap(Map.of(UserLog.MARKER, UserLog.AUTOMATION_RULE_EVALUATOR.name())); + return event; + } + + @FunctionalInterface + private interface SaveAll { + Mono apply(List events); + } +} From a360c1d86c17df62927e2f3d72abf3c7d30edadb Mon Sep 17 00:00:00 2001 From: Daniel Dimenshtein Date: Thu, 3 Sep 2026 22:20:43 +0300 Subject: [PATCH 2/2] fix(logs): drop batches rejected before insert instead of requeueing Review feedback on the retry path. A synchronous throw from saveAll means the batch is rejected before any I/O -- an event missing workspace_id or rule_id throws IllegalStateException out of the DAO. Requeueing such a batch made every later flush throw identically, so the appender would spin on a poison batch forever. Drop it instead, and keep requeueing only for errors signalled through the returned Mono, which are the genuinely transient ones. Also hoist the fixed Retry.backoff policy into a constant rather than building one per flushed batch, and assert the retry attempt count in the transient test -- it previously passed on requeue-then-reflush alone, so it did not actually pin Retry.backoff. Co-Authored-By: Claude Opus 5 (1M context) --- .../log/ClickHouseAppender.java | 11 +++--- .../log/ClickHouseAppenderTest.java | 39 ++++++++++++++++++- 2 files changed, 43 insertions(+), 7 deletions(-) diff --git a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java index 735a3a733b6..3874641a3ce 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/infrastructure/log/ClickHouseAppender.java @@ -30,6 +30,7 @@ class ClickHouseAppender extends AppenderBase { private static final long INSERT_RETRY_ATTEMPTS = 3; private static final Duration INSERT_RETRY_MIN_BACKOFF = Duration.ofMillis(100); + private static final Retry INSERT_RETRY = Retry.backoff(INSERT_RETRY_ATTEMPTS, INSERT_RETRY_MIN_BACKOFF); private static ClickHouseAppender instance; @@ -104,12 +105,10 @@ private void flushLogs() { // The batch is already drained out of the queue, so a failed insert would lose // these events outright. Retry transient failures, and put the events back on // the queue if the retries are exhausted so a later flush can pick them up. - // saveAll is called inside the try so that a synchronous throw re-queues too, - // rather than only errors signalled through the returned Mono. try { tableDAO .saveAll(events) - .retryWhen(Retry.backoff(INSERT_RETRY_ATTEMPTS, INSERT_RETRY_MIN_BACKOFF)) + .retryWhen(INSERT_RETRY) .subscribe( noop -> { }, @@ -118,8 +117,10 @@ private void flushLogs() { requeue(events); }); } catch (Exception e) { - log.error("Failed to insert logs", e); - requeue(events); + // A synchronous throw from saveAll means the batch itself is rejected before + // any I/O (e.g. an event missing workspace_id or rule_id). Requeueing would + // fail identically on every later flush, so drop it instead of looping. + log.error("Dropping '{}' logs rejected before insert", events.size(), e); } } }); diff --git a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java index 73a5cecea87..05e136c5824 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java @@ -54,8 +54,36 @@ void whenInsertFailsTransiently__thenEventsAreRetriedAndPersisted() { appender.doAppend(event("a message")); + // The attempt count is what pins Retry.backoff: without it a batch requeued and picked up by a + // later flush would also persist eventually, so hasSize(1) alone would not prove a retry ran. await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) - .untilAsserted(() -> assertThat(persisted).hasSize(1)); + .untilAsserted(() -> { + assertThat(persisted).hasSize(1); + assertThat(attempts).hasValue(2); + }); + } + + @Test + @DisplayName("when the batch is rejected before insert, then it is dropped rather than requeued") + void whenBatchRejectedBeforeInsert__thenItIsDroppedRatherThanRequeued() { + var attempts = new AtomicInteger(); + + // A synchronous throw means the batch is malformed (e.g. missing workspace_id), so it would + // fail identically forever. Requeueing it would spin the flush thread on a poison batch. + startAppender(events -> { + attempts.incrementAndGet(); + throw new IllegalStateException("workspace_id is not set"); + }); + + appender.doAppend(event("a malformed message")); + + await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(attempts).hasValue(1)); + + // Give several further flush intervals a chance to re-attempt the dropped batch. + await().during(FLUSH_INTERVAL.multipliedBy(10)) + .atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(attempts).hasValue(1)); } @Test @@ -76,8 +104,15 @@ void whenFlushThrows__thenSubsequentFlushesStillRun() { appender.doAppend(event("first")); + // The first batch is dropped (it was rejected before insert), so the surviving scheduler is + // proven by a later event still being persisted rather than by the failed batch reappearing. await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) - .untilAsserted(() -> assertThat(persisted).isNotEmpty()); + .untilAsserted(() -> assertThat(failNext).hasValue(0)); + + appender.doAppend(event("second")); + + await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(persisted).hasSize(1)); } private void startAppender(SaveAll saveAll) {