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..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 @@ -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,10 @@ @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 final Retry INSERT_RETRY = Retry.backoff(INSERT_RETRY_ATTEMPTS, INSERT_RETRY_MIN_BACKOFF); + private static ClickHouseAppender instance; public static synchronized ClickHouseAppender init(@NonNull UserLogTableFactory userLogTableFactory, int batchSize, @@ -58,13 +63,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 +102,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. + try { + tableDAO + .saveAll(events) + .retryWhen(INSERT_RETRY) + .subscribe( + noop -> { + }, + e -> { + log.error("Failed to insert logs", e); + requeue(events); + }); + } catch (Exception e) { + // 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); + } } }); } + 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 +152,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..05e136c5824 --- /dev/null +++ b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/log/ClickHouseAppenderTest.java @@ -0,0 +1,136 @@ +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")); + + // 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); + 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 + @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")); + + // 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(failNext).hasValue(0)); + + appender.doAppend(event("second")); + + await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(persisted).hasSize(1)); + } + + 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); + } +}