diff --git a/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionService.java b/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionService.java index f210459c..8ccbf2cc 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionService.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionService.java @@ -37,6 +37,7 @@ import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -137,9 +138,10 @@ public DocumentIndexingCompletionResponse complete( durationMs ); - // 6. 대시보드가 최신 집계를 다시 계산하도록 상태 전이를 알린다. AFTER_COMMIT 구독자만 - // 반응하므로 이 Transaction이 실제로 커밋된 뒤에만 push로 이어진다. + // 6. 대시보드 갱신과 성공 Counter를 같은 Commit에 결박한다. 두 AFTER_COMMIT 구독자는 + // 이 Transaction이 실제로 커밋된 뒤에만 반응한다. applicationEventPublisher.publishEvent(new EmbeddingJobStatusChangedEvent(embeddingJob.getId())); + applicationEventPublisher.publishEvent(EmbeddingJobAttemptMetricEvent.success()); log.info( "문서 인덱싱 완료: jobId={}, attemptId={}, documentId={}, versionId={}, " diff --git a/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureService.java b/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureService.java index 9c564fac..a3223571 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureService.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureService.java @@ -29,6 +29,7 @@ import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent; import lombok.extern.slf4j.Slf4j; @@ -167,11 +168,15 @@ public DocumentIndexingFailureResponse fail( minimumRetryDelay ); - // 4. 대시보드가 최신 집계를 다시 계산하도록 상태 전이를 알린다. transition()이 재시도 예약 + // 4. 대시보드 갱신과 실패 Counter를 같은 Commit에 결박한다. transition()이 재시도 예약 // (PENDING)과 최종 실패(FAILED) 중 어느 쪽으로 끝났든 embeddingJob은 같은 영속 인스턴스라 // 최종 상태를 그대로 반영한다. AFTER_COMMIT 구독자만 반응하므로 이 Transaction이 실제로 - // 커밋된 뒤에만 push로 이어진다. + // 커밋된 뒤에만 push와 Counter 증가로 이어진다. applicationEventPublisher.publishEvent(new EmbeddingJobStatusChangedEvent(embeddingJob.getId())); + applicationEventPublisher.publishEvent(EmbeddingJobAttemptMetricEvent.failure( + embeddingJob.getStatus(), + request.failureType() + )); log.info( "문서 인덱싱 실패 기록: jobId={}, attemptId={}, failureType={}, retryCount={}, terminal={}", diff --git a/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryService.java b/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryService.java index 610c9f98..e1511f36 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryService.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryService.java @@ -6,6 +6,7 @@ import java.util.Optional; import java.util.UUID; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; @@ -21,6 +22,7 @@ import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent; import lombok.RequiredArgsConstructor; @@ -43,6 +45,7 @@ public class EmbeddingJobLeaseRecoveryService { private final EmbeddingJobAttemptRepository embeddingJobAttemptRepository; private final IndexingEventRepository indexingEventRepository; private final IndexingFailureTransitionService failureTransitionService; + private final ApplicationEventPublisher applicationEventPublisher; /** * 후보 Snapshot 이후에도 만료 상태인 Job만 회수해 Retry 또는 최종 실패로 전환한다. @@ -84,6 +87,11 @@ public RecoveryResult recover(Long jobId, LocalDateTime recoveredAt) { recoveredAt, Duration.ZERO ); + + // 5. 실제 회수 전이가 커밋된 뒤에만 재시도 또는 최종 실패 Counter가 증가하게 한다. + applicationEventPublisher.publishEvent( + EmbeddingJobAttemptMetricEvent.leaseExpired(embeddingJob.getStatus()) + ); return RecoveryResult.recovered(embeddingJob); } diff --git a/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java b/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java index d7e5aa91..41309ad9 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java @@ -2,6 +2,7 @@ import java.util.List; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -21,6 +22,8 @@ import com.opensource.docgrid.domain.search.service.query.SearchConversationQueryService; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.RagJobCompletionMetricEvent; +import com.opensource.docgrid.global.observability.RagJobCompletionMetricEvent.Outcome; import jakarta.persistence.EntityManager; import lombok.RequiredArgsConstructor; @@ -88,6 +91,7 @@ public class RagFacade { private final SearchResultRepository searchResultRepository; private final SearchConversationQueryService searchConversationQueryService; private final EntityManager entityManager; + private final ApplicationEventPublisher applicationEventPublisher; /** * 검색 직후 SearchController가 동기 호출하는 접수 단계. Ollama는 아직 호출하지 않는다. @@ -107,6 +111,8 @@ public RagEnqueueOutcome enqueue( if (candidates.isEmpty()) { RagResponse ragResponse = ragResponseCommandService.createNoContext(queryRef); + // LLM 호출을 생략한 정상 완료도 저장 Transaction이 커밋된 뒤 별도 결과로 집계한다. + applicationEventPublisher.publishEvent(new RagJobCompletionMetricEvent(Outcome.NO_CONTEXT)); log.info("[RAG] no context queryId={} responseId={}", queryId, ragResponse.getId()); return RagEnqueueOutcome.done(RagAnswer.noContext(ragResponse.getAnswerText())); } @@ -169,6 +175,11 @@ public boolean processJob(Long jobId) { String fallbackAnswer = candidates.isEmpty() ? e.getErrorCode().getMessage() : buildExtractiveFallbackAnswer(candidates); boolean completed = ragResponseCommandService.completeFailed(job, fallbackAnswer, e.getMessage()); + if (completed) { + applicationEventPublisher.publishEvent( + new RagJobCompletionMetricEvent(Outcome.PROVIDER_FALLBACK) + ); + } log.warn("[RAG] fallback queryId={} errorCode={}", queryId, e.getErrorCode().getCode()); return completed; } @@ -216,6 +227,8 @@ public boolean processJob(Long jobId) { // 3. 완료 처리가 성공한 답변에만 선택한 후보 순서대로 출처를 저장한다. responseCitationCommandService.saveAll(job, citationCandidates, citationSearchResults); } + // 답변과 citation 저장이 모두 끝난 동일 Transaction의 커밋 이후 성공 Counter를 기록한다. + applicationEventPublisher.publishEvent(new RagJobCompletionMetricEvent(Outcome.SUCCESS)); log.info("[RAG] done queryId={} responseId={} latencyMs={}", queryId, job.getId(), result.latencyMs()); return true; } @@ -232,9 +245,15 @@ public boolean processJob(Long jobId) { * (RagJobWorker)는 이 경우 WebSocket 알림을 보내지 않는다. */ public boolean markUnexpectedFailure(Long jobId, String errorMessage) { - return ragResponseRepository.findById(jobId) + boolean completed = ragResponseRepository.findById(jobId) .map(job -> ragResponseCommandService.completeFailed(job, UNEXPECTED_FAILURE_ANSWER_TEXT, errorMessage)) .orElse(false); + if (completed) { + applicationEventPublisher.publishEvent( + new RagJobCompletionMetricEvent(Outcome.UNEXPECTED_FAILURE) + ); + } + return completed; } /** @@ -252,6 +271,9 @@ public boolean failIfStillProcessing(Long jobId, Long queryId) { ? UNEXPECTED_FAILURE_ANSWER_TEXT : buildExtractiveFallbackAnswer(candidates); int updated = ragResponseRepository.forceFailIfProcessing(jobId, fallbackAnswer, TIMEOUT_ERROR_MESSAGE); + if (updated > 0) { + applicationEventPublisher.publishEvent(new RagJobCompletionMetricEvent(Outcome.TIMEOUT_SWEPT)); + } return updated > 0; } diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java index 54e30848..3e0bbd81 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java @@ -4,6 +4,7 @@ import java.time.LocalDateTime; import java.util.Objects; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; @@ -15,6 +16,8 @@ import com.opensource.docgrid.domain.sync.service.SyncEventHandlerRegistry; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome; import lombok.RequiredArgsConstructor; @@ -33,6 +36,7 @@ public class SyncEventDispatchService { private final SyncEventHandlerRegistry syncEventHandlerRegistry; private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService; private final Clock clock; + private final ApplicationEventPublisher applicationEventPublisher; /** * Claim된 Outbox Event의 Handler 부작용과 완료 전이를 독립 트랜잭션으로 실행한다. @@ -64,6 +68,9 @@ public void dispatch(ClaimedSyncEvent claimedEvent) { completedAt ); completionEvent.complete(claimedEvent.claimToken(), completedAt); + + // 5. Handler 부작용과 완료 상태가 함께 커밋된 뒤에만 처리 성공 Counter를 기록한다. + applicationEventPublisher.publishEvent(new SyncEventAttemptMetricEvent(Outcome.PROCESSED)); } /** diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java index 33205792..c2b8efdf 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java @@ -5,6 +5,7 @@ import java.util.Objects; import java.util.UUID; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; @@ -15,6 +16,8 @@ import com.opensource.docgrid.domain.sync.service.SyncEventRetrySchedule; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome; import lombok.RequiredArgsConstructor; @@ -31,6 +34,7 @@ public class SyncEventFailureService { private final SyncEventRetrySchedule syncEventRetrySchedule; private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService; private final Clock clock; + private final ApplicationEventPublisher applicationEventPublisher; /** * Handler 실패를 현재 Claim의 Delivery Attempt와 Queue 상태에 함께 기록한다. @@ -54,6 +58,9 @@ public void recordFailure(UUID eventId, UUID claimToken, String errorCode, Strin // 4. 이번 실패가 허용 횟수를 채우면 다시 Claim되지 않는 최종 상태로 종결한다. if (event.getRetryCount() + 1 >= event.getMaxRetryCount()) { event.markFailed(claimToken, errorCode, errorMessage, failedAt); + applicationEventPublisher.publishEvent( + new SyncEventAttemptMetricEvent(Outcome.TERMINAL_FAILURE) + ); return; } @@ -65,6 +72,9 @@ public void recordFailure(UUID eventId, UUID claimToken, String errorCode, Strin failedAt, syncEventRetrySchedule.nextAvailableAt(event, failedAt) ); + applicationEventPublisher.publishEvent( + new SyncEventAttemptMetricEvent(Outcome.RETRY_SCHEDULED) + ); } /** diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java index 6d4f72c3..24cc404c 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java @@ -4,6 +4,7 @@ import java.util.Optional; import java.util.UUID; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; @@ -12,6 +13,8 @@ import com.opensource.docgrid.domain.sync.enums.SyncEventStatus; import com.opensource.docgrid.domain.sync.repository.SyncOutboxEventRepository; import com.opensource.docgrid.domain.sync.service.SyncEventRetrySchedule; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome; import lombok.RequiredArgsConstructor; @@ -28,6 +31,7 @@ public class SyncEventLeaseRecoveryService { private final SyncOutboxEventRepository syncOutboxEventRepository; private final SyncEventRetrySchedule syncEventRetrySchedule; private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService; + private final ApplicationEventPublisher applicationEventPublisher; /** * 만료 후보 Event를 다시 확인해 실제로 만료된 현재 Claim만 회수한다. @@ -65,7 +69,13 @@ public RecoveryResult recover(UUID eventId, LocalDateTime recoveredAt) { syncEventRetrySchedule.nextAvailableAt(event, recoveredAt) ); - // 5. Scheduler가 회수 결과를 집계할 수 있도록 변경 후 상태를 반환한다. + // 5. 최종 실패는 즉시 조치 경보 대상이며, 재예약된 Lease 만료는 회수 활동으로 구분한다. + Outcome outcome = event.getStatus() == SyncEventStatus.FAILED + ? Outcome.TERMINAL_FAILURE + : Outcome.LEASE_RECOVERED; + applicationEventPublisher.publishEvent(new SyncEventAttemptMetricEvent(outcome)); + + // 6. Scheduler가 회수 결과를 집계할 수 있도록 변경 후 상태를 반환한다. return new RecoveryResult(eventId, true, event.getStatus()); } diff --git a/backend/src/main/java/com/opensource/docgrid/global/observability/AsyncPipelineMetrics.java b/backend/src/main/java/com/opensource/docgrid/global/observability/AsyncPipelineMetrics.java new file mode 100644 index 00000000..ac5f86f0 --- /dev/null +++ b/backend/src/main/java/com/opensource/docgrid/global/observability/AsyncPipelineMetrics.java @@ -0,0 +1,60 @@ +package com.opensource.docgrid.global.observability; + +import org.springframework.stereotype.Component; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; + +import io.micrometer.core.instrument.MeterRegistry; +import lombok.RequiredArgsConstructor; + +/** + * 비동기 파이프라인의 커밋된 상태 전이를 Micrometer Counter로 변환한다. + * + *

도메인 서비스는 DB 상태와 같은 트랜잭션에서 제한된 이벤트만 발행한다. 이 구독자는 + * {@link TransactionPhase#AFTER_COMMIT}에만 실행되므로 롤백된 상태 변화는 Counter에 남지 않는다. + */ +@Component +@RequiredArgsConstructor +public class AsyncPipelineMetrics { + + private static final String EMBEDDING_ATTEMPTS = "docgrid.embedding.job.attempts"; + private static final String RAG_COMPLETIONS = "docgrid.rag.job.completions"; + private static final String SYNC_ATTEMPTS = "docgrid.sync.event.attempts"; + + private final MeterRegistry meterRegistry; + + /** + * 확정된 Embedding 실행 결과를 실패 분류와 재시도 가능 여부별로 기록한다. + */ + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + public void recordEmbeddingAttempt(EmbeddingJobAttemptMetricEvent event) { + meterRegistry.counter( + EMBEDDING_ATTEMPTS, + "outcome", event.outcome().label(), + "failure_type", event.failureType().name(), + "retryable", Boolean.toString(event.failureType().isRetryable()) + ).increment(); + } + + /** + * 확정된 RAG 답변 결과를 정상·fallback·timeout 경로별로 기록한다. + */ + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + public void recordRagCompletion(RagJobCompletionMetricEvent event) { + meterRegistry.counter( + RAG_COMPLETIONS, + "outcome", event.outcome().label() + ).increment(); + } + + /** + * 확정된 Sync Outbox 처리 결과를 완료·재시도·종결·Lease 회수별로 기록한다. + */ + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + public void recordSyncAttempt(SyncEventAttemptMetricEvent event) { + meterRegistry.counter( + SYNC_ATTEMPTS, + "outcome", event.outcome().label() + ).increment(); + } +} diff --git a/backend/src/main/java/com/opensource/docgrid/global/observability/EmbeddingJobAttemptMetricEvent.java b/backend/src/main/java/com/opensource/docgrid/global/observability/EmbeddingJobAttemptMetricEvent.java new file mode 100644 index 00000000..0a663839 --- /dev/null +++ b/backend/src/main/java/com/opensource/docgrid/global/observability/EmbeddingJobAttemptMetricEvent.java @@ -0,0 +1,139 @@ +package com.opensource.docgrid.global.observability; + +import java.util.Objects; + +import com.opensource.docgrid.domain.embedding.enums.EmbeddingJobStatus; +import com.opensource.docgrid.domain.embedding.enums.IndexingFailureType; + +/** + * 커밋이 확정된 Embedding Job 실행 결과를 제한된 Micrometer label 값으로 전달한다. + * + *

Job ID나 오류 메시지를 싣지 않아 시계열 cardinality가 입력 데이터에 따라 증가하지 않는다. + */ +public record EmbeddingJobAttemptMetricEvent( + Outcome outcome, + FailureType failureType +) { + + /** + * 성공한 실행의 단일 label 조합을 생성한다. + */ + public static EmbeddingJobAttemptMetricEvent success() { + return new EmbeddingJobAttemptMetricEvent(Outcome.SUCCESS, FailureType.NONE); + } + + /** + * Worker가 보고한 고정 실패 분류와 전이 후 상태로 재시도 또는 최종 실패를 구분한다. + */ + public static EmbeddingJobAttemptMetricEvent failure( + EmbeddingJobStatus status, + IndexingFailureType failureType + ) { + IndexingFailureType requiredFailureType = Objects.requireNonNull( + failureType, + "failureType은 필수입니다." + ); + return new EmbeddingJobAttemptMetricEvent( + Outcome.fromFailureStatus(status), + FailureType.from(requiredFailureType) + ); + } + + /** + * Worker Lease 만료는 외부 요청 enum과 분리된 고정 운영 실패 분류로 기록한다. + */ + public static EmbeddingJobAttemptMetricEvent leaseExpired(EmbeddingJobStatus status) { + return new EmbeddingJobAttemptMetricEvent( + Outcome.fromFailureStatus(status), + FailureType.WORKER_LEASE_EXPIRED + ); + } + + /** + * 성공과 실패 label 조합이 서로 모순되지 않도록 생성 시점에 검증한다. + */ + public EmbeddingJobAttemptMetricEvent { + Objects.requireNonNull(outcome, "outcome은 필수입니다."); + Objects.requireNonNull(failureType, "failureType은 필수입니다."); + if ((outcome == Outcome.SUCCESS) != (failureType == FailureType.NONE)) { + throw new IllegalArgumentException("성공 결과만 NONE 실패 유형을 사용할 수 있습니다."); + } + } + + /** + * Embedding 실행이 확정된 뒤 노출할 저 cardinality 결과 label이다. + */ + public enum Outcome { + SUCCESS("success"), + RETRY_SCHEDULED("retry_scheduled"), + TERMINAL_FAILURE("terminal_failure"); + + private final String label; + + Outcome(String label) { + this.label = label; + } + + public String label() { + return label; + } + + private static Outcome fromFailureStatus(EmbeddingJobStatus status) { + return switch (Objects.requireNonNull(status, "status는 필수입니다.")) { + case PENDING -> RETRY_SCHEDULED; + case FAILED -> TERMINAL_FAILURE; + default -> throw new IllegalArgumentException("실패 전이 후 상태는 PENDING 또는 FAILED여야 합니다."); + }; + } + } + + /** + * 외부 요청 enum과 내부 Lease 회수 원인을 합친 고정 실패 label 목록이다. + */ + public enum FailureType { + NONE(null), + STORAGE_UNAVAILABLE(IndexingFailureType.STORAGE_UNAVAILABLE), + STORAGE_CONFIGURATION_INVALID(IndexingFailureType.STORAGE_CONFIGURATION_INVALID), + STORAGE_OBJECT_MISSING(IndexingFailureType.STORAGE_OBJECT_MISSING), + DOCUMENT_CONTENT_INVALID(IndexingFailureType.DOCUMENT_CONTENT_INVALID), + EMBEDDING_PROVIDER_UNAVAILABLE(IndexingFailureType.EMBEDDING_PROVIDER_UNAVAILABLE), + EMBEDDING_PROVIDER_OVERLOADED(IndexingFailureType.EMBEDDING_PROVIDER_OVERLOADED), + EMBEDDING_PROVIDER_TIMEOUT(IndexingFailureType.EMBEDDING_PROVIDER_TIMEOUT), + EMBEDDING_PROVIDER_CIRCUIT_OPEN(IndexingFailureType.EMBEDDING_PROVIDER_CIRCUIT_OPEN), + EMBEDDING_REQUEST_INVALID(IndexingFailureType.EMBEDDING_REQUEST_INVALID), + EMBEDDING_RESULT_INVALID(IndexingFailureType.EMBEDDING_RESULT_INVALID), + INDEXING_STATE_INCONSISTENT(IndexingFailureType.INDEXING_STATE_INCONSISTENT), + WORKER_INTERNAL_ERROR(IndexingFailureType.WORKER_INTERNAL_ERROR), + WORKER_LEASE_EXPIRED(null); + + private final IndexingFailureType indexingFailureType; + + FailureType(IndexingFailureType indexingFailureType) { + this.indexingFailureType = indexingFailureType; + } + + /** Retry 정책은 도메인 enum을 단일 출처로 사용하고 내부 Lease 원인만 고정한다. */ + public boolean isRetryable() { + return this == WORKER_LEASE_EXPIRED + || (indexingFailureType != null && indexingFailureType.isRetryable()); + } + + private static FailureType from(IndexingFailureType failureType) { + // 도메인 enum이 늘어나면 이 switch가 컴파일 오류를 내므로 런타임 실패 경로를 막는다. + return switch (failureType) { + case STORAGE_UNAVAILABLE -> STORAGE_UNAVAILABLE; + case STORAGE_CONFIGURATION_INVALID -> STORAGE_CONFIGURATION_INVALID; + case STORAGE_OBJECT_MISSING -> STORAGE_OBJECT_MISSING; + case DOCUMENT_CONTENT_INVALID -> DOCUMENT_CONTENT_INVALID; + case EMBEDDING_PROVIDER_UNAVAILABLE -> EMBEDDING_PROVIDER_UNAVAILABLE; + case EMBEDDING_PROVIDER_OVERLOADED -> EMBEDDING_PROVIDER_OVERLOADED; + case EMBEDDING_PROVIDER_TIMEOUT -> EMBEDDING_PROVIDER_TIMEOUT; + case EMBEDDING_PROVIDER_CIRCUIT_OPEN -> EMBEDDING_PROVIDER_CIRCUIT_OPEN; + case EMBEDDING_REQUEST_INVALID -> EMBEDDING_REQUEST_INVALID; + case EMBEDDING_RESULT_INVALID -> EMBEDDING_RESULT_INVALID; + case INDEXING_STATE_INCONSISTENT -> INDEXING_STATE_INCONSISTENT; + case WORKER_INTERNAL_ERROR -> WORKER_INTERNAL_ERROR; + }; + } + } +} diff --git a/backend/src/main/java/com/opensource/docgrid/global/observability/RagJobCompletionMetricEvent.java b/backend/src/main/java/com/opensource/docgrid/global/observability/RagJobCompletionMetricEvent.java new file mode 100644 index 00000000..147a07d3 --- /dev/null +++ b/backend/src/main/java/com/opensource/docgrid/global/observability/RagJobCompletionMetricEvent.java @@ -0,0 +1,37 @@ +package com.opensource.docgrid.global.observability; + +import java.util.Objects; + +/** + * 커밋이 확정된 RAG Job의 최종 처리 경로를 고정된 결과 label로 전달한다. + */ +public record RagJobCompletionMetricEvent(Outcome outcome) { + + /** + * 자유 형식 결과가 metric label로 유입되지 않도록 enum 값만 허용한다. + */ + public RagJobCompletionMetricEvent { + Objects.requireNonNull(outcome, "outcome은 필수입니다."); + } + + /** + * 정상 답변과 각 fallback·회수 원인을 구분하는 RAG 완료 결과다. + */ + public enum Outcome { + SUCCESS("success"), + NO_CONTEXT("no_context"), + PROVIDER_FALLBACK("provider_fallback"), + TIMEOUT_SWEPT("timeout_swept"), + UNEXPECTED_FAILURE("unexpected_failure"); + + private final String label; + + Outcome(String label) { + this.label = label; + } + + public String label() { + return label; + } + } +} diff --git a/backend/src/main/java/com/opensource/docgrid/global/observability/SyncEventAttemptMetricEvent.java b/backend/src/main/java/com/opensource/docgrid/global/observability/SyncEventAttemptMetricEvent.java new file mode 100644 index 00000000..e78456f7 --- /dev/null +++ b/backend/src/main/java/com/opensource/docgrid/global/observability/SyncEventAttemptMetricEvent.java @@ -0,0 +1,36 @@ +package com.opensource.docgrid.global.observability; + +import java.util.Objects; + +/** + * 커밋이 확정된 Sync Outbox 처리 또는 Lease 회수 결과를 고정된 label로 전달한다. + */ +public record SyncEventAttemptMetricEvent(Outcome outcome) { + + /** + * 자유 형식 오류 코드가 metric label로 유입되지 않도록 enum 값만 허용한다. + */ + public SyncEventAttemptMetricEvent { + Objects.requireNonNull(outcome, "outcome은 필수입니다."); + } + + /** + * Outbox가 실제로 확정한 처리 결과다. + */ + public enum Outcome { + PROCESSED("processed"), + RETRY_SCHEDULED("retry_scheduled"), + TERMINAL_FAILURE("terminal_failure"), + LEASE_RECOVERED("lease_recovered"); + + private final String label; + + Outcome(String label) { + this.label = label; + } + + public String label() { + return label; + } + } +} diff --git a/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionServiceTest.java b/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionServiceTest.java index 02f703ad..707a8f32 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionServiceTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionServiceTest.java @@ -48,6 +48,7 @@ import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent; /** * 문서 인덱싱 최초 완료 Transaction의 잠금 순서, 검증, 상태 전이와 이전 검색 Set 비활성화를 검증한다. @@ -141,6 +142,8 @@ void complete_transitionsFirstVersionAtomically() { assertThat(eventCaptor.getValue().getFromStatus()).isEqualTo("EMBEDDING"); assertThat(eventCaptor.getValue().getToStatus()).isEqualTo("INDEXED"); assertThat(eventCaptor.getValue().getOccurredAt()).isEqualTo(COMPLETED_AT); + then(applicationEventPublisher).should() + .publishEvent(EmbeddingJobAttemptMetricEvent.success()); } @Test diff --git a/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureServiceTest.java b/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureServiceTest.java index c687d133..257513b5 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureServiceTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureServiceTest.java @@ -50,6 +50,7 @@ import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent; /** * 인덱싱 실패 Service의 Retry·최종 종료·멱등 재생 분기와 원자 상태 변경을 검증한다. @@ -159,6 +160,12 @@ void fail_schedulesInitialRetryAndRecordsEvents() { .isEqualTo("{\"attemptId\":103,\"attemptNo\":1,\"failureType\":\"STORAGE_UNAVAILABLE\"}"); assertThat(eventCaptor.getAllValues().get(1).getMetadataJson()) .isEqualTo("{\"retryCount\":1,\"nextRetryAt\":\"2026-08-03T10:30:10\"}"); + then(applicationEventPublisher).should().publishEvent( + EmbeddingJobAttemptMetricEvent.failure( + EmbeddingJobStatus.PENDING, + IndexingFailureType.STORAGE_UNAVAILABLE + ) + ); } @Test @@ -223,6 +230,12 @@ void fail_terminatesFirstVersionForPermanentFailure() { .containsExactly(IndexingEventType.EMBEDDING_FAILED, IndexingEventType.FAILED); assertThat(eventCaptor.getAllValues().get(1).getMetadataJson()) .isEqualTo("{\"retryCount\":0,\"maxRetryCount\":3}"); + then(applicationEventPublisher).should().publishEvent( + EmbeddingJobAttemptMetricEvent.failure( + EmbeddingJobStatus.FAILED, + IndexingFailureType.EMBEDDING_RESULT_INVALID + ) + ); } @Test diff --git a/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryServiceTest.java b/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryServiceTest.java index def62893..56daf575 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryServiceTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryServiceTest.java @@ -3,8 +3,11 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; import java.time.Duration; @@ -18,6 +21,7 @@ import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.test.util.ReflectionTestUtils; import com.opensource.docgrid.domain.document.entity.DocumentVersion; @@ -36,6 +40,7 @@ import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent; /** * EmbeddingJobLeaseRecoveryService의 후보 재검증, 선택적 Attempt와 공통 실패 전이 위임을 검증한다. @@ -59,6 +64,7 @@ class EmbeddingJobLeaseRecoveryServiceTest { @Mock private EmbeddingJobAttemptRepository embeddingJobAttemptRepository; @Mock private IndexingEventRepository indexingEventRepository; @Mock private IndexingFailureTransitionService failureTransitionService; + @Mock private ApplicationEventPublisher applicationEventPublisher; private EmbeddingJobLeaseRecoveryService recoveryService; @@ -68,7 +74,19 @@ void setUp() { embeddingJobRepository, embeddingJobAttemptRepository, indexingEventRepository, - failureTransitionService + failureTransitionService, + applicationEventPublisher + ); + // 공통 실패 전이는 mock이므로, 운영 구현처럼 회수 후 상태가 PENDING이 되도록 반영한다. + lenient().doAnswer(invocation -> { + ReflectionTestUtils.setField( + invocation.getArgument(0), + "status", + EmbeddingJobStatus.PENDING + ); + return null; + }).when(failureTransitionService).transition( + any(), any(), any(), any(), anyBoolean(), any(), any() ); } @@ -104,6 +122,8 @@ void recover_transitionsStartedAttemptAndSavesLeaseExpiredEvent() { ); assertThat(result.recovered()).isTrue(); assertThat(result.jobId()).isEqualTo(JOB_ID); + then(applicationEventPublisher).should() + .publishEvent(EmbeddingJobAttemptMetricEvent.leaseExpired(EmbeddingJobStatus.PENDING)); } @Test @@ -132,6 +152,32 @@ void recover_transitionsJobWithoutCreatingAttempt_when_attemptDoesNotExist() { Duration.ZERO ); assertThat(result.recovered()).isTrue(); + then(applicationEventPublisher).should() + .publishEvent(EmbeddingJobAttemptMetricEvent.leaseExpired(EmbeddingJobStatus.PENDING)); + } + + @Test + @DisplayName("Lease 만료 재시도를 모두 소진하면 최종 실패 결과를 계측한다") + void recover_publishesTerminalFailureMetric_when_retryIsExhausted() { + EmbeddingJob embeddingJob = createExpiredJob(); + EmbeddingJobAttempt attempt = createAttempt(embeddingJob, AttemptStatus.STARTED); + given(embeddingJobRepository.findExpiredByIdForUpdateSkipLocked(JOB_ID, RECOVERED_AT)) + .willReturn(Optional.of(embeddingJob)); + given(embeddingJobAttemptRepository.findByEmbeddingJobIdAndClaimToken(JOB_ID, CLAIM_TOKEN)) + .willReturn(Optional.of(attempt)); + doAnswer(invocation -> { + ReflectionTestUtils.setField(embeddingJob, "status", EmbeddingJobStatus.FAILED); + return null; + }).when(failureTransitionService).transition( + any(), any(), any(), any(), anyBoolean(), any(), any() + ); + + RecoveryResult result = recoveryService.recover(JOB_ID, RECOVERED_AT); + + assertThat(result.recovered()).isTrue(); + assertThat(result.status()).isEqualTo(EmbeddingJobStatus.FAILED); + then(applicationEventPublisher).should() + .publishEvent(EmbeddingJobAttemptMetricEvent.leaseExpired(EmbeddingJobStatus.FAILED)); } @Test @@ -147,6 +193,7 @@ void recover_skips_when_candidateCannotBeLocked() { then(embeddingJobAttemptRepository).shouldHaveNoInteractions(); then(indexingEventRepository).shouldHaveNoInteractions(); then(failureTransitionService).shouldHaveNoInteractions(); + then(applicationEventPublisher).shouldHaveNoInteractions(); } @Test diff --git a/backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagFacadeTest.java b/backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagFacadeTest.java index cc4d098f..ce71966d 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagFacadeTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagFacadeTest.java @@ -27,6 +27,7 @@ import org.mockito.junit.jupiter.MockitoExtension; import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; +import org.springframework.context.ApplicationEventPublisher; import com.opensource.docgrid.domain.rag.dto.OllamaGenerateResult; import com.opensource.docgrid.domain.rag.dto.RagAnswer; @@ -43,6 +44,8 @@ import com.opensource.docgrid.domain.search.service.query.SearchConversationQueryService; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; +import com.opensource.docgrid.global.observability.RagJobCompletionMetricEvent; +import com.opensource.docgrid.global.observability.RagJobCompletionMetricEvent.Outcome; import jakarta.persistence.EntityManager; @@ -82,6 +85,9 @@ class RagFacadeTest { @Mock private EntityManager entityManager; + @Mock + private ApplicationEventPublisher applicationEventPublisher; + private static final Long QUERY_ID = 100L; private static final Long CONVERSATION_ID = 50L; private static final Long JOB_ID = 999L; @@ -107,6 +113,8 @@ void enqueue_noQualifiedCandidates_returnsDoneWithFixedAnswer() { then(promptBuilder).should(never()).build(anyString(), any(), any()); then(ragResponseCommandService).should(times(1)).createNoContext(queryRef); then(ragResponseCommandService).should(never()).createPending(any(), anyString()); + then(applicationEventPublisher).should() + .publishEvent(new RagJobCompletionMetricEvent(Outcome.NO_CONTEXT)); } @Test @@ -241,6 +249,8 @@ void processJob_success_savesResponseAndCitations() { then(ragResponseCommandService).should(times(1)).completeSuccess(eq(job), any()); then(responseCitationCommandService).should(times(1)).saveAll(eq(job), any(), eq(List.of(searchResult))); then(ragResponseCommandService).should(never()).completeFailed(any(), anyString(), anyString()); + then(applicationEventPublisher).should() + .publishEvent(new RagJobCompletionMetricEvent(Outcome.SUCCESS)); } @Test @@ -275,6 +285,7 @@ void processJob_completeSuccessLosesRace_skipsCitationsAndReturnsFalse() { assertThat(completed).isFalse(); then(responseCitationCommandService).should(never()).saveAll(any(), any(), any()); then(searchResultRepository).should(never()).findByQuery_IdOrderByRankNo(any()); + then(applicationEventPublisher).shouldHaveNoInteractions(); } @Test @@ -299,6 +310,8 @@ void processJob_ollamaFails_savesFailedWithExtractiveFallback() { eq(job), argThatFallbackContains("AI 답변 생성이 지연", "청크 내용", "인사규정"), anyString() ); then(responseCitationCommandService).should(never()).saveAll(any(), any(), any()); + then(applicationEventPublisher).should() + .publishEvent(new RagJobCompletionMetricEvent(Outcome.PROVIDER_FALLBACK)); } @Test @@ -315,6 +328,7 @@ void processJob_completeFailedLosesRace_returnsFalse() { boolean completed = ragFacade.processJob(JOB_ID); assertThat(completed).isFalse(); + then(applicationEventPublisher).shouldHaveNoInteractions(); } @Test @@ -364,6 +378,37 @@ void processJob_phraseEmbeddedInAnswer_trimsBeforePersistingAndKeepsCitations() then(responseCitationCommandService).should(times(1)).saveAll(eq(job), any(), eq(List.of(searchResult))); } + // === markUnexpectedFailure() === + + @Test + @DisplayName("markUnexpectedFailure: PROCESSING job을 실제 종료하면 예상 밖 실패 메트릭을 발행한다") + void markUnexpectedFailure_completed_publishesMetricEvent() { + RagResponse job = RagResponse.builder().status(ResultStatus.PROCESSING).build(); + given(ragResponseRepository.findById(JOB_ID)).willReturn(Optional.of(job)); + given(ragResponseCommandService.completeFailed(job, "답변 생성 중 예상치 못한 오류가 발생했습니다.", "bug")) + .willReturn(true); + + boolean completed = ragFacade.markUnexpectedFailure(JOB_ID, "bug"); + + assertThat(completed).isTrue(); + then(applicationEventPublisher).should() + .publishEvent(new RagJobCompletionMetricEvent(Outcome.UNEXPECTED_FAILURE)); + } + + @Test + @DisplayName("markUnexpectedFailure: 조건부 종료 경합에서 지면 메트릭을 발행하지 않는다") + void markUnexpectedFailure_losesRace_doesNotPublishMetricEvent() { + RagResponse job = RagResponse.builder().status(ResultStatus.PROCESSING).build(); + given(ragResponseRepository.findById(JOB_ID)).willReturn(Optional.of(job)); + given(ragResponseCommandService.completeFailed(job, "답변 생성 중 예상치 못한 오류가 발생했습니다.", "bug")) + .willReturn(false); + + boolean completed = ragFacade.markUnexpectedFailure(JOB_ID, "bug"); + + assertThat(completed).isFalse(); + then(applicationEventPublisher).shouldHaveNoInteractions(); + } + // === failIfStillProcessing() === @Test @@ -379,6 +424,8 @@ void failIfStillProcessing_withCandidates_forceFailsWithFallbackAndReturnsTrue() then(ragResponseRepository).should(times(1)).forceFailIfProcessing( eq(JOB_ID), argThatFallbackContains("AI 답변 생성이 지연", "청크 내용", "인사규정"), anyString() ); + then(applicationEventPublisher).should() + .publishEvent(new RagJobCompletionMetricEvent(Outcome.TIMEOUT_SWEPT)); } @Test @@ -391,6 +438,7 @@ void failIfStillProcessing_alreadyFinishedByWorker_returnsFalse() { boolean result = ragFacade.failIfStillProcessing(JOB_ID, QUERY_ID); assertThat(result).isFalse(); + then(applicationEventPublisher).shouldHaveNoInteractions(); } @Test diff --git a/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchServiceTest.java b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchServiceTest.java index e13ceaf0..f2b18e57 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchServiceTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchServiceTest.java @@ -21,6 +21,7 @@ import org.junit.jupiter.api.extension.ExtendWith; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.context.ApplicationEventPublisher; import com.opensource.docgrid.domain.sync.dto.ClaimedSyncEvent; import com.opensource.docgrid.domain.sync.entity.SyncOutboxEvent; @@ -29,6 +30,8 @@ import com.opensource.docgrid.domain.sync.enums.SyncEventType; import com.opensource.docgrid.domain.sync.repository.SyncOutboxEventRepository; import com.opensource.docgrid.domain.sync.service.SyncEventHandlerRegistry; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome; /** * Handler 실행과 Outbox Event 완료 전이가 같은 Dispatch 경계에서 수행되는지 검증한다. @@ -43,6 +46,7 @@ class SyncEventDispatchServiceTest { @Mock private SyncOutboxEventRepository syncOutboxEventRepository; @Mock private SyncEventHandlerRegistry syncEventHandlerRegistry; @Mock private SyncEventDeliveryAttemptService syncEventDeliveryAttemptService; + @Mock private ApplicationEventPublisher applicationEventPublisher; private SyncEventDispatchService service; private SyncOutboxEvent event; @@ -56,7 +60,8 @@ void setUp() { syncOutboxEventRepository, syncEventHandlerRegistry, syncEventDeliveryAttemptService, - clock + clock, + applicationEventPublisher ); LocalDateTime now = LocalDateTime.ofInstant(NOW, ZONE_ID); UUID eventId = UUID.randomUUID(); @@ -85,6 +90,8 @@ void dispatch_reloadsAndCompletesEvent_afterHandlerSucceeds() { assertThat(event.getStatus()).isEqualTo(SyncEventStatus.PROCESSING); assertThat(completionEvent.getStatus()).isEqualTo(SyncEventStatus.PROCESSED); assertThat(completionEvent.getProcessedAt()).isNotNull(); + then(applicationEventPublisher).should() + .publishEvent(new SyncEventAttemptMetricEvent(Outcome.PROCESSED)); } @Test @@ -102,6 +109,7 @@ void dispatch_doesNotComplete_whenHandlerFails() { ); assertThat(event.getStatus()).isEqualTo(SyncEventStatus.PROCESSING); assertThat(event.getProcessedAt()).isNull(); + then(applicationEventPublisher).shouldHaveNoInteractions(); } private SyncOutboxEvent event(UUID eventId, LocalDateTime now) { diff --git a/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureServiceTest.java b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureServiceTest.java new file mode 100644 index 00000000..f19e2edc --- /dev/null +++ b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureServiceTest.java @@ -0,0 +1,103 @@ +package com.opensource.docgrid.domain.sync.service.command; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.then; + +import java.time.Clock; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.Optional; +import java.util.UUID; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.context.ApplicationEventPublisher; + +import com.opensource.docgrid.domain.sync.entity.SyncOutboxEvent; +import com.opensource.docgrid.domain.sync.enums.SyncAggregateType; +import com.opensource.docgrid.domain.sync.enums.SyncEventStatus; +import com.opensource.docgrid.domain.sync.enums.SyncEventType; +import com.opensource.docgrid.domain.sync.repository.SyncOutboxEventRepository; +import com.opensource.docgrid.domain.sync.service.SyncEventRetrySchedule; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome; + +/** + * Sync Handler 실패가 Retry 또는 최종 실패로 확정될 때 대응 metric event를 발행하는지 검증한다. + */ +@ExtendWith(MockitoExtension.class) +@DisplayName("SyncEventFailureService 메트릭 테스트") +class SyncEventFailureServiceTest { + + private static final UUID EVENT_ID = UUID.fromString("68c12639-7c1f-4740-8db9-29a341983251"); + private static final UUID CLAIM_TOKEN = UUID.fromString("a5492111-f005-40cf-9f2f-f72fd63f45f7"); + private static final LocalDateTime FAILED_AT = LocalDateTime.of(2026, 9, 13, 19, 0); + private static final Clock CLOCK = Clock.fixed( + Instant.parse("2026-09-13T19:00:00Z"), + ZoneOffset.UTC + ); + + @Mock private SyncOutboxEventRepository repository; + @Mock private SyncEventRetrySchedule retrySchedule; + @Mock private SyncEventDeliveryAttemptService attemptService; + @Mock private ApplicationEventPublisher applicationEventPublisher; + + @Test + @DisplayName("남은 실행 기회가 있으면 RETRY_SCHEDULED event를 발행한다") + void recordFailure_retryable_publishesRetryScheduledMetric() { + SyncOutboxEvent event = claimedEvent(3); + given(repository.findByEventIdForUpdate(EVENT_ID)).willReturn(Optional.of(event)); + given(retrySchedule.nextAvailableAt(event, FAILED_AT)).willReturn(FAILED_AT.plusSeconds(5)); + SyncEventFailureService service = service(); + + service.recordFailure(EVENT_ID, CLAIM_TOKEN, "HANDLER_ERROR", "temporary failure"); + + assertThat(event.getStatus()).isEqualTo(SyncEventStatus.PENDING); + then(applicationEventPublisher).should() + .publishEvent(new SyncEventAttemptMetricEvent(Outcome.RETRY_SCHEDULED)); + } + + @Test + @DisplayName("마지막 실행 기회를 소진하면 TERMINAL_FAILURE event를 발행한다") + void recordFailure_exhausted_publishesTerminalFailureMetric() { + SyncOutboxEvent event = claimedEvent(1); + given(repository.findByEventIdForUpdate(EVENT_ID)).willReturn(Optional.of(event)); + SyncEventFailureService service = service(); + + service.recordFailure(EVENT_ID, CLAIM_TOKEN, "HANDLER_ERROR", "permanent failure"); + + assertThat(event.getStatus()).isEqualTo(SyncEventStatus.FAILED); + then(applicationEventPublisher).should() + .publishEvent(new SyncEventAttemptMetricEvent(Outcome.TERMINAL_FAILURE)); + } + + private SyncEventFailureService service() { + return new SyncEventFailureService( + repository, + retrySchedule, + attemptService, + CLOCK, + applicationEventPublisher + ); + } + + private SyncOutboxEvent claimedEvent(int maxRetryCount) { + SyncOutboxEvent event = SyncOutboxEvent.builder() + .eventId(EVENT_ID) + .idempotencyKey("sync-failure-service-test-" + maxRetryCount) + .aggregateType(SyncAggregateType.DOCUMENT_VERSION) + .aggregateId(1L) + .eventType(SyncEventType.DOCUMENT_VERSION_CREATED) + .availableAt(FAILED_AT.minusMinutes(1)) + .occurredAt(FAILED_AT.minusMinutes(1)) + .maxRetryCount(maxRetryCount) + .build(); + event.claim("dispatcher", CLAIM_TOKEN, FAILED_AT.minusSeconds(1), FAILED_AT.plusMinutes(1)); + return event; + } +} diff --git a/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryServiceTest.java b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryServiceTest.java new file mode 100644 index 00000000..bb4cafb4 --- /dev/null +++ b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryServiceTest.java @@ -0,0 +1,105 @@ +package com.opensource.docgrid.domain.sync.service.command; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.then; + +import java.time.LocalDateTime; +import java.util.Optional; +import java.util.UUID; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.context.ApplicationEventPublisher; + +import com.opensource.docgrid.domain.sync.entity.SyncOutboxEvent; +import com.opensource.docgrid.domain.sync.enums.SyncAggregateType; +import com.opensource.docgrid.domain.sync.enums.SyncEventStatus; +import com.opensource.docgrid.domain.sync.enums.SyncEventType; +import com.opensource.docgrid.domain.sync.repository.SyncOutboxEventRepository; +import com.opensource.docgrid.domain.sync.service.SyncEventRetrySchedule; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent; +import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome; + +/** + * 만료 Sync Lease 회수가 재예약과 최종 실패 결과에 맞는 metric event를 발행하는지 검증한다. + */ +@ExtendWith(MockitoExtension.class) +@DisplayName("SyncEventLeaseRecoveryService 메트릭 테스트") +class SyncEventLeaseRecoveryServiceTest { + + private static final UUID EVENT_ID = UUID.fromString("8db4bc37-8148-4bb4-b471-d89722d9013f"); + private static final UUID CLAIM_TOKEN = UUID.fromString("4b65a076-0677-47b2-a21f-d3499ef1d09b"); + private static final LocalDateTime RECOVERED_AT = LocalDateTime.of(2026, 9, 13, 19, 0); + + @Mock private SyncOutboxEventRepository repository; + @Mock private SyncEventRetrySchedule retrySchedule; + @Mock private SyncEventDeliveryAttemptService attemptService; + @Mock private ApplicationEventPublisher applicationEventPublisher; + + @Test + @DisplayName("남은 실행 기회가 있는 만료 Lease는 LEASE_RECOVERED event를 발행한다") + void recover_retryable_publishesLeaseRecoveredMetric() { + SyncOutboxEvent event = expiredEvent(3); + given(repository.findExpiredByEventIdForUpdateSkipLocked(EVENT_ID, RECOVERED_AT)) + .willReturn(Optional.of(event)); + given(retrySchedule.nextAvailableAt(event, RECOVERED_AT)).willReturn(RECOVERED_AT.plusSeconds(5)); + SyncEventLeaseRecoveryService service = service(); + + SyncEventLeaseRecoveryService.RecoveryResult result = service.recover(EVENT_ID, RECOVERED_AT); + + assertThat(result.recovered()).isTrue(); + assertThat(result.status()).isEqualTo(SyncEventStatus.PENDING); + then(applicationEventPublisher).should() + .publishEvent(new SyncEventAttemptMetricEvent(Outcome.LEASE_RECOVERED)); + } + + @Test + @DisplayName("마지막 실행 기회를 소진한 만료 Lease는 TERMINAL_FAILURE event를 발행한다") + void recover_exhausted_publishesTerminalFailureMetric() { + SyncOutboxEvent event = expiredEvent(1); + given(repository.findExpiredByEventIdForUpdateSkipLocked(EVENT_ID, RECOVERED_AT)) + .willReturn(Optional.of(event)); + given(retrySchedule.nextAvailableAt(event, RECOVERED_AT)).willReturn(RECOVERED_AT.plusSeconds(5)); + SyncEventLeaseRecoveryService service = service(); + + SyncEventLeaseRecoveryService.RecoveryResult result = service.recover(EVENT_ID, RECOVERED_AT); + + assertThat(result.recovered()).isTrue(); + assertThat(result.status()).isEqualTo(SyncEventStatus.FAILED); + then(applicationEventPublisher).should() + .publishEvent(new SyncEventAttemptMetricEvent(Outcome.TERMINAL_FAILURE)); + } + + private SyncEventLeaseRecoveryService service() { + return new SyncEventLeaseRecoveryService( + repository, + retrySchedule, + attemptService, + applicationEventPublisher + ); + } + + private SyncOutboxEvent expiredEvent(int maxRetryCount) { + SyncOutboxEvent event = SyncOutboxEvent.builder() + .eventId(EVENT_ID) + .idempotencyKey("sync-lease-recovery-test-" + maxRetryCount) + .aggregateType(SyncAggregateType.DOCUMENT_VERSION) + .aggregateId(1L) + .eventType(SyncEventType.DOCUMENT_VERSION_CREATED) + .availableAt(RECOVERED_AT.minusMinutes(2)) + .occurredAt(RECOVERED_AT.minusMinutes(2)) + .maxRetryCount(maxRetryCount) + .build(); + event.claim( + "dispatcher", + CLAIM_TOKEN, + RECOVERED_AT.minusMinutes(1), + RECOVERED_AT.minusSeconds(1) + ); + return event; + } +} diff --git a/backend/src/test/java/com/opensource/docgrid/global/observability/AsyncPipelineMetricsAfterCommitIntegrationTest.java b/backend/src/test/java/com/opensource/docgrid/global/observability/AsyncPipelineMetricsAfterCommitIntegrationTest.java new file mode 100644 index 00000000..1496bc77 --- /dev/null +++ b/backend/src/test/java/com/opensource/docgrid/global/observability/AsyncPipelineMetricsAfterCommitIntegrationTest.java @@ -0,0 +1,120 @@ +package com.opensource.docgrid.global.observability; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.concurrent.atomic.AtomicBoolean; + +import javax.sql.DataSource; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; + +/** + * 상태 전이 이벤트가 실제 Transaction 커밋 뒤에만 Counter에 반영되는지 검증한다. + */ +@SpringJUnitConfig(AsyncPipelineMetricsAfterCommitIntegrationTest.Config.class) +@Tag("integration") +@DisplayName("비동기 파이프라인 Counter AFTER_COMMIT 통합 테스트") +class AsyncPipelineMetricsAfterCommitIntegrationTest { + + @Autowired private ApplicationEventPublisher applicationEventPublisher; + @Autowired private PlatformTransactionManager transactionManager; + @Autowired private SimpleMeterRegistry meterRegistry; + + private TransactionTemplate transactionTemplate; + + @BeforeEach + void setUp() { + transactionTemplate = new TransactionTemplate(transactionManager); + meterRegistry.clear(); + } + + @Test + @DisplayName("이벤트를 발행한 Transaction이 커밋되면 Counter가 한 번 증가한다") + void incrementsCounterAfterCommit() { + AtomicBoolean beforeCommitObserved = new AtomicBoolean(); + transactionTemplate.executeWithoutResult(status -> { + applicationEventPublisher.publishEvent(EmbeddingJobAttemptMetricEvent.success()); + + // 1. Event 발행 직후에는 아직 commit이 아니므로 Counter가 생성되지 않아야 한다. + assertThat(meterRegistry.find("docgrid.embedding.job.attempts").counter()).isNull(); + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void beforeCommit(boolean readOnly) { + // 2. BEFORE_COMMIT phase에서도 값이 없음을 확인해 listener phase 회귀를 잡는다. + assertThat(meterRegistry.find("docgrid.embedding.job.attempts").counter()).isNull(); + beforeCommitObserved.set(true); + } + }); + }); + + assertThat(beforeCommitObserved).isTrue(); + assertThat(successCounter()).isEqualTo(1.0); + } + + @Test + @DisplayName("이벤트를 발행한 Transaction이 롤백되면 Counter가 증가하지 않는다") + void doesNotIncrementCounterAfterRollback() { + transactionTemplate.executeWithoutResult(status -> { + applicationEventPublisher.publishEvent(EmbeddingJobAttemptMetricEvent.success()); + status.setRollbackOnly(); + }); + + assertThat(meterRegistry.find("docgrid.embedding.job.attempts").counter()).isNull(); + } + + private double successCounter() { + return meterRegistry.get("docgrid.embedding.job.attempts") + .tag("outcome", "success") + .tag("failure_type", "NONE") + .tag("retryable", "false") + .counter() + .count(); + } + + /** + * 운영 Metrics 구독자와 실제 Spring Transaction 이벤트 처리만 격리해 구성한다. + */ + @Configuration + @EnableTransactionManagement + @Import(AsyncPipelineMetrics.class) + static class Config { + + @Bean(destroyMethod = "shutdown") + EmbeddedDatabase dataSource() { + return new EmbeddedDatabaseBuilder() + .setType(EmbeddedDatabaseType.H2) + .generateUniqueName(true) + .build(); + } + + @Bean + PlatformTransactionManager transactionManager(DataSource dataSource) { + return new DataSourceTransactionManager(dataSource); + } + + @Bean + SimpleMeterRegistry meterRegistry() { + return new SimpleMeterRegistry(); + } + } +} diff --git a/backend/src/test/java/com/opensource/docgrid/global/observability/AsyncPipelineMetricsTest.java b/backend/src/test/java/com/opensource/docgrid/global/observability/AsyncPipelineMetricsTest.java new file mode 100644 index 00000000..4cc3e24e --- /dev/null +++ b/backend/src/test/java/com/opensource/docgrid/global/observability/AsyncPipelineMetricsTest.java @@ -0,0 +1,87 @@ +package com.opensource.docgrid.global.observability; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import com.opensource.docgrid.domain.embedding.enums.EmbeddingJobStatus; +import com.opensource.docgrid.domain.embedding.enums.IndexingFailureType; +import com.opensource.docgrid.global.observability.RagJobCompletionMetricEvent.Outcome; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; + +/** + * 비동기 파이프라인 이벤트가 정해진 이름과 제한된 label의 Counter로 변환되는지 검증한다. + */ +@DisplayName("비동기 파이프라인 상태 전이 Counter 테스트") +class AsyncPipelineMetricsTest { + + private final SimpleMeterRegistry meterRegistry = new SimpleMeterRegistry(); + private final AsyncPipelineMetrics metrics = new AsyncPipelineMetrics(meterRegistry); + + @Test + @DisplayName("Embedding 재시도와 사용자 입력 최종 실패를 failure_type과 retryable로 구분한다") + void recordsEmbeddingOutcomesWithBoundedFailureLabels() { + metrics.recordEmbeddingAttempt(EmbeddingJobAttemptMetricEvent.failure( + EmbeddingJobStatus.PENDING, + IndexingFailureType.EMBEDDING_PROVIDER_TIMEOUT + )); + metrics.recordEmbeddingAttempt(EmbeddingJobAttemptMetricEvent.failure( + EmbeddingJobStatus.FAILED, + IndexingFailureType.DOCUMENT_CONTENT_INVALID + )); + + assertThat(embeddingCounter("retry_scheduled", "EMBEDDING_PROVIDER_TIMEOUT", "true")) + .isEqualTo(1.0); + assertThat(embeddingCounter("terminal_failure", "DOCUMENT_CONTENT_INVALID", "false")) + .isEqualTo(1.0); + } + + @Test + @DisplayName("RAG fallback과 timeout을 서로 다른 완료 결과로 기록한다") + void recordsRagCompletionPathsSeparately() { + metrics.recordRagCompletion(new RagJobCompletionMetricEvent(Outcome.PROVIDER_FALLBACK)); + metrics.recordRagCompletion(new RagJobCompletionMetricEvent(Outcome.TIMEOUT_SWEPT)); + + assertThat(ragCounter("provider_fallback")).isEqualTo(1.0); + assertThat(ragCounter("timeout_swept")).isEqualTo(1.0); + } + + @Test + @DisplayName("Outbox 재시도와 최종 실패를 서로 다른 처리 결과로 기록한다") + void recordsSyncAttemptOutcomesSeparately() { + metrics.recordSyncAttempt(new SyncEventAttemptMetricEvent( + SyncEventAttemptMetricEvent.Outcome.RETRY_SCHEDULED + )); + metrics.recordSyncAttempt(new SyncEventAttemptMetricEvent( + SyncEventAttemptMetricEvent.Outcome.TERMINAL_FAILURE + )); + + assertThat(syncCounter("retry_scheduled")).isEqualTo(1.0); + assertThat(syncCounter("terminal_failure")).isEqualTo(1.0); + } + + private double embeddingCounter(String outcome, String failureType, String retryable) { + return meterRegistry.get("docgrid.embedding.job.attempts") + .tag("outcome", outcome) + .tag("failure_type", failureType) + .tag("retryable", retryable) + .counter() + .count(); + } + + private double ragCounter(String outcome) { + return meterRegistry.get("docgrid.rag.job.completions") + .tag("outcome", outcome) + .counter() + .count(); + } + + private double syncCounter(String outcome) { + return meterRegistry.get("docgrid.sync.event.attempts") + .tag("outcome", outcome) + .counter() + .count(); + } +} diff --git a/docs/design/gimin-#332-async-pipeline-transition-metrics.md b/docs/design/gimin-#332-async-pipeline-transition-metrics.md new file mode 100644 index 00000000..70952493 --- /dev/null +++ b/docs/design/gimin-#332-async-pipeline-transition-metrics.md @@ -0,0 +1,101 @@ +# 비동기 Pipeline 상태 전이 메트릭 설계 (#332) + +closes #332 + +## 문제 + +Actuator가 제공하는 JVM·HTTP·HikariCP 메트릭만으로는 DocGrid의 비동기 처리 결과를 알 수 없다. +인덱싱, RAG, Sync Outbox는 실패가 누적 상태로 남거나 재시도를 위해 다시 `PENDING`으로 돌아가므로 +현재 `FAILED` 행 수를 경보 기준으로 사용하면 새 장애와 과거 장애를 구분할 수 없다. + +운영 경보에는 일정 시간 동안 새로 확정된 상태 전이 횟수가 필요하다. 이 문서는 Counter 이름과 +label, 증가 시점, 이를 사용하는 Prometheus 규칙을 고정한다. + +## 상태 전이와 Counter의 Commit 경계 + +도메인 서비스는 DB 상태를 바꾸는 Transaction 안에서 제한된 metric event를 발행한다. +`AsyncPipelineMetrics`는 `@TransactionalEventListener(AFTER_COMMIT)`으로 이벤트를 받는다. + +```text +Transactional command + → 상태 전이와 감사 이력 저장 + → 제한된 outcome event 발행 + → commit 성공 + → AFTER_COMMIT listener + → Micrometer Counter 증가 +``` + +Transaction이 롤백되면 listener는 호출되지 않는다. 조건부 UPDATE가 경합에서 0건을 반환한 RAG +완료·실패 경로는 이벤트 자체를 발행하지 않는다. 멱등 재생도 최초 상태를 반환할 뿐 새로운 전이가 +아니므로 Counter를 다시 올리지 않는다. + +## Metric 계약 + +Micrometer 이름은 Prometheus endpoint에서 아래 `_total` Counter로 노출된다. + +| Prometheus metric | label | 허용 값 | +|---|---|---| +| `docgrid_embedding_job_attempts_total` | `outcome` | `success`, `retry_scheduled`, `terminal_failure` | +| | `failure_type` | `NONE`, `IndexingFailureType` 12종, `WORKER_LEASE_EXPIRED` | +| | `retryable` | `true`, `false` | +| `docgrid_rag_job_completions_total` | `outcome` | `success`, `no_context`, `provider_fallback`, `timeout_swept`, `unexpected_failure` | +| `docgrid_sync_event_attempts_total` | `outcome` | `processed`, `retry_scheduled`, `terminal_failure`, `lease_recovered` | + +ID, 사용자 정보, 문서명, 오류 메시지는 label에 넣지 않는다. 실패 유형과 결과는 코드의 enum에서만 +생성해 입력 크기에 따라 시계열 수가 늘어나지 않게 한다. `failure_type`은 `IndexingFailureType`의 +모든 값을 exhaustive switch로 매핑하고, `retryable`은 도메인 enum의 정책값을 그대로 사용한다. +실패 유형 추가 시 누락은 컴파일 오류가 되고 Retry 정책은 한 곳에서만 관리된다. + +## 도메인별 기록 위치 + +### Embedding + +- `DocumentIndexingCompletionService`: 새 실행이 `INDEXED`로 확정되면 `success` +- `DocumentIndexingFailureService`: 전이 후 상태가 `PENDING`이면 `retry_scheduled`, `FAILED`면 + `terminal_failure` +- `EmbeddingJobLeaseRecoveryService`: 같은 결과 구분에 `WORKER_LEASE_EXPIRED` 실패 유형 사용 + +완료·실패 요청의 멱등 재생 분기는 이벤트 발행 전에 반환한다. + +### RAG + +- 검색 후보 없음: `no_context` +- Ollama 정상 응답 저장: `success` +- Ollama 예외 뒤 extractive fallback 저장: `provider_fallback` +- Worker의 예상하지 못한 실패 확정: `unexpected_failure` +- Timeout Sweeper의 조건부 UPDATE 성공: `timeout_swept` + +RAG 이벤트는 `forceFailIfProcessing` 또는 `completeSuccessIfProcessing`의 영향 행이 1건일 때만 +발행한다. Worker와 Sweeper가 경쟁해도 승리한 한 경로만 기록된다. + +### Sync Outbox + +- Handler 부작용과 Event 완료를 함께 커밋: `processed` +- 실패 후 backoff 예약: `retry_scheduled` +- 재시도 소진: `terminal_failure` +- 만료 Lease를 다시 `PENDING`으로 회수: `lease_recovered` + +Lease 회수가 재시도 한도를 소진해 `FAILED`로 끝나면 즉시 대응할 수 있도록 +`terminal_failure`로 기록한다. + +## 경보 정책 + +| 경보 | 목적 | 오탐 방지 | +|---|---|---| +| `DocGridEmbeddingRetryableFailureRatioHigh` | 인프라성 인덱싱 실패 증가 | retryable만 포함, 10분간 최소 10회, 10% 초과가 5분 지속 | +| `DocGridRagProviderFallbackSpike` | Ollama 장애 또는 지연 증가 | 10분간 3회 이상, 1분 지속 | +| `DocGridRagTimeoutSweepSpike` | Worker 정체로 Sweeper 회수 증가 | 10분간 3회 이상, 1분 지속 | +| `DocGridSyncOutboxTerminalFailure` | 새 최종 실패 즉시 발견 | 누적 FAILED 개수 대신 15분 `increase` 사용 | + +`DOCUMENT_CONTENT_INVALID`처럼 retryable이 아닌 사용자 문서 오류는 인덱싱 운영 장애 경보에서 +제외한다. 성공 Counter만 증가하거나 Outbox가 재시도 중인 경우에도 최종 실패 경보는 발생하지 않는다. + +## 선택한 방식과 한계 + +도메인 서비스가 `MeterRegistry`를 직접 사용하면 상태 저장이 롤백돼도 이미 증가한 Counter를 되돌릴 +수 없다. 호출자에서 Transaction 반환 뒤 증가시키는 방식도 가능하지만 HTTP Controller, Scheduler, +내부 Worker마다 같은 규칙을 반복해야 한다. Transaction event를 사용하면 상태 전이와 metric event의 +결합 지점을 도메인 서비스 한 곳에 두고 실제 증가는 commit 이후로 미룰 수 있다. + +Counter는 프로세스 메모리에 있으므로 Backend 재시작 때 0부터 시작한다. Prometheus가 주기적으로 +누적 시계열을 저장하고 `rate`·`increase`에서 Counter reset을 처리한다는 운영 모델을 따른다. diff --git a/docs/test-results/gimin-#332-async-pipeline-transition-metrics.md b/docs/test-results/gimin-#332-async-pipeline-transition-metrics.md new file mode 100644 index 00000000..ee58d062 --- /dev/null +++ b/docs/test-results/gimin-#332-async-pipeline-transition-metrics.md @@ -0,0 +1,80 @@ +# 비동기 Pipeline 상태 전이 메트릭 검증 결과 (#332) + +검증일: 2026-09-13 + +## Java 검증 + +```bash +DB_PORT=55433 \ +JWT_SECRET=docgrid-observability-test-secret-key-2026-with-at-least-32-bytes \ +./backend/gradlew -p backend test +``` + +결과: **1164 tests, 실패 0, errors 0, skipped 0** + +새 검증은 다음 동작을 포함한다. + +- Embedding retryable Provider 실패와 non-retryable 문서 오류의 label 분리 +- Embedding 성공·재시도·최종 실패 Service의 metric event 발행 +- Embedding Lease 회수의 재예약·재시도 소진 최종 실패 event 분기 +- 도메인 실패 유형의 exhaustive metric label 매핑과 Retry 정책 재사용 +- RAG provider fallback과 timeout sweep Counter 분리 +- RAG 예상 밖 실패의 실제 종료와 조건부 UPDATE 경합 패배 분기 +- Sync Outbox 재시도와 최종 실패 Counter 분리 +- Sync Outbox 실패·Lease 회수의 재예약과 최종 실패 event 분기 +- Transaction commit 후 Counter 증가 +- Transaction rollback 시 Counter 미생성 +- RAG 조건부 UPDATE 경합 패배 시 metric event 미발행 +- Sync Handler 실패 시 처리 성공 metric event 미발행 + +## Prometheus 설정과 규칙 + +```bash +docker run --rm --entrypoint=promtool \ + -v "$PWD/monitoring/prometheus:/etc/prometheus:ro" \ + prom/prometheus:v3.5.5 \ + check config /etc/prometheus/prometheus.yml +``` + +결과: **성공** + +- rule file 3개 로드 +- Embedding Provider 규칙 7개 +- Backend 규칙 3개 +- 비동기 Pipeline 규칙 4개 + +```bash +docker run --rm --entrypoint=promtool \ + -v "$PWD/monitoring/prometheus:/etc/prometheus:ro" \ + prom/prometheus:v3.5.5 \ + test rules \ + /etc/prometheus/tests/docgrid-backend-alerts.test.yml \ + /etc/prometheus/tests/docgrid-pipeline-alerts.test.yml +``` + +결과: **두 rule test suite 모두 성공** + +| 시나리오 | 결과 | +|---|---| +| retryable 실패 20%, 최소 표본 충족, `for: 5m` 충족 | 인덱싱 경보 발생 | +| `DOCUMENT_CONTENT_INVALID` 20% | 인덱싱 운영 경보 미발생 | +| RAG provider fallback 3회 | fallback 경보 발생 | +| RAG timeout sweep 3회 | timeout 경보 발생 | +| RAG success만 발생 | RAG 경보 미발생 | +| Outbox `retry_scheduled`만 발생 | 최종 실패 경보 미발생 | +| Outbox `terminal_failure` 발생 | 최종 실패 경보 발생 | + +## Compose 검증 + +```bash +docker compose config +docker compose --profile monitoring config +``` + +결과: **두 구성 모두 정상 렌더링** + +## 테스트 환경 참고 + +격리 checkout에는 Git 제외 `.env`가 없으므로 전체 테스트에 테스트용 `JWT_SECRET`을 명시했다. +기존 PostgreSQL 컨테이너가 Host `55433`에 연결되어 있어 `DB_PORT=55433`을 사용했다. 실제 Secret은 +출력하거나 저장하지 않았으며 테스트 전용 값만 사용했다. diff --git a/monitoring/prometheus/README.md b/monitoring/prometheus/README.md index 03745afd..7e31978f 100644 --- a/monitoring/prometheus/README.md +++ b/monitoring/prometheus/README.md @@ -54,8 +54,8 @@ scrape_configs: `8081`은 사용자 API 포트가 아닌 Management 포트다. 방화벽, Security Group, 컨테이너 network로 Prometheus와 운영자만 접근하도록 제한한다. -Backend 경보를 함께 사용하려면 `monitoring/prometheus/rules/docgrid-backend-alerts.yml`을 기존 -Prometheus의 `rule_files` 경로에 복사한다. +DocGrid 경보를 함께 사용하려면 `monitoring/prometheus/rules/`의 규칙 파일을 기존 Prometheus의 +`rule_files` 경로에 복사한다. ## 기본 Backend 경보 @@ -68,6 +68,19 @@ Prometheus의 `rule_files` 경로에 복사한다. HTTP 오류율에서는 Streamable HTTP 특성이 다른 `/mcp`를 제외한다. 이 경보들은 현재 Prometheus 화면에서 확인하며 외부 전달은 Alertmanager 설정을 추가한 뒤 활성화된다. +## 비동기 Pipeline 경보 + +| 경보 | 조건 | 지속 시간 | +|---|---|---:| +| `DocGridEmbeddingRetryableFailureRatioHigh` | 10분간 10회 이상 실행되고 retryable 실패가 10% 초과 | 5분 | +| `DocGridRagProviderFallbackSpike` | 10분간 provider fallback 3회 이상 | 1분 | +| `DocGridRagTimeoutSweepSpike` | 10분간 timeout 강제 종료 3회 이상 | 1분 | +| `DocGridSyncOutboxTerminalFailure` | 15분간 새로운 최종 실패 1회 이상 | 즉시 | + +Counter는 DB 상태 전이를 수행한 Transaction이 커밋된 뒤에만 증가한다. Embedding 실패의 +`failure_type`은 고정 enum이며 `retryable` label로 사용자 문서 오류와 운영 장애를 구분한다. +Job ID, 오류 메시지와 사용자 입력은 label에 포함하지 않는다. + ## 설정 검증 로컬에 promtool을 설치하지 않아도 고정된 Prometheus 이미지로 검사할 수 있다. @@ -83,12 +96,18 @@ docker run --rm --entrypoint=promtool \ prom/prometheus:v3.5.5 \ check rules \ /etc/prometheus/rules/embedding-provider-alerts.yml \ - /etc/prometheus/rules/docgrid-backend-alerts.yml + /etc/prometheus/rules/docgrid-backend-alerts.yml \ + /etc/prometheus/rules/docgrid-pipeline-alerts.yml docker run --rm --entrypoint=promtool \ -v "$PWD/monitoring/prometheus:/etc/prometheus:ro" \ prom/prometheus:v3.5.5 \ test rules /etc/prometheus/tests/docgrid-backend-alerts.test.yml +docker run --rm --entrypoint=promtool \ + -v "$PWD/monitoring/prometheus:/etc/prometheus:ro" \ + prom/prometheus:v3.5.5 \ + test rules /etc/prometheus/tests/docgrid-pipeline-alerts.test.yml + docker compose config ``` diff --git a/monitoring/prometheus/rules/docgrid-pipeline-alerts.yml b/monitoring/prometheus/rules/docgrid-pipeline-alerts.yml new file mode 100644 index 00000000..5eff92a7 --- /dev/null +++ b/monitoring/prometheus/rules/docgrid-pipeline-alerts.yml @@ -0,0 +1,76 @@ +groups: + - name: docgrid-async-pipelines + rules: + - alert: DocGridEmbeddingRetryableFailureRatioHigh + expr: | + ( + sum by (cluster, environment) ( + rate(docgrid_embedding_job_attempts_total{ + outcome=~"retry_scheduled|terminal_failure", + retryable="true" + }[10m]) + ) + / + clamp_min( + sum by (cluster, environment) ( + rate(docgrid_embedding_job_attempts_total[10m]) + ), + 0.001 + ) + ) > 0.10 + and + sum by (cluster, environment) ( + increase(docgrid_embedding_job_attempts_total[10m]) + ) >= 10 + for: 5m + labels: + severity: warning + service: embedding-worker + annotations: + summary: DocGrid embedding retryable failure ratio is high + description: More than 10% of at least 10 embedding attempts failed with a retryable cause during the last ten minutes. + + - alert: DocGridRagProviderFallbackSpike + expr: | + sum by (cluster, environment) ( + increase(docgrid_rag_job_completions_total{ + outcome="provider_fallback" + }[10m]) + ) >= 3 + for: 1m + labels: + severity: warning + service: rag-worker + dependency: ollama + annotations: + summary: DocGrid RAG provider fallback is increasing + description: At least three RAG jobs used the provider fallback during the last ten minutes. + + - alert: DocGridRagTimeoutSweepSpike + expr: | + sum by (cluster, environment) ( + increase(docgrid_rag_job_completions_total{ + outcome="timeout_swept" + }[10m]) + ) >= 3 + for: 1m + labels: + severity: warning + service: rag-worker + annotations: + summary: DocGrid RAG timeout recovery is increasing + description: At least three RAG jobs were force-failed by the timeout sweeper during the last ten minutes. + + - alert: DocGridSyncOutboxTerminalFailure + expr: | + sum by (cluster, environment) ( + increase(docgrid_sync_event_attempts_total{ + outcome="terminal_failure" + }[15m]) + ) > 0 + labels: + severity: critical + service: sync-outbox + annotations: + summary: A DocGrid Sync Outbox event reached terminal failure + description: At least one Sync Outbox event exhausted retries during the last fifteen minutes. diff --git a/monitoring/prometheus/tests/docgrid-pipeline-alerts.test.yml b/monitoring/prometheus/tests/docgrid-pipeline-alerts.test.yml new file mode 100644 index 00000000..437554b2 --- /dev/null +++ b/monitoring/prometheus/tests/docgrid-pipeline-alerts.test.yml @@ -0,0 +1,104 @@ +rule_files: + - /etc/prometheus/rules/docgrid-pipeline-alerts.yml + +evaluation_interval: 1m + +tests: + - name: retryable embedding failures require ratio sample and duration + interval: 1m + input_series: + - series: 'docgrid_embedding_job_attempts_total{instance="backend-a:8081",environment="production",cluster="docgrid-production",outcome="success",failure_type="NONE",retryable="false"}' + values: '0+8x12' + - series: 'docgrid_embedding_job_attempts_total{instance="backend-a:8081",environment="production",cluster="docgrid-production",outcome="retry_scheduled",failure_type="EMBEDDING_PROVIDER_TIMEOUT",retryable="true"}' + values: '0+2x12' + alert_rule_test: + - eval_time: 5m + alertname: DocGridEmbeddingRetryableFailureRatioHigh + exp_alerts: [] + - eval_time: 6m + alertname: DocGridEmbeddingRetryableFailureRatioHigh + exp_alerts: + - exp_labels: + alertname: DocGridEmbeddingRetryableFailureRatioHigh + cluster: docgrid-production + environment: production + service: embedding-worker + severity: warning + exp_annotations: + summary: DocGrid embedding retryable failure ratio is high + description: More than 10% of at least 10 embedding attempts failed with a retryable cause during the last ten minutes. + + - name: non-retryable document failures do not trigger infrastructure alert + interval: 1m + input_series: + - series: 'docgrid_embedding_job_attempts_total{instance="backend-a:8081",environment="production",cluster="docgrid-production",outcome="success",failure_type="NONE",retryable="false"}' + values: '0+8x12' + - series: 'docgrid_embedding_job_attempts_total{instance="backend-a:8081",environment="production",cluster="docgrid-production",outcome="terminal_failure",failure_type="DOCUMENT_CONTENT_INVALID",retryable="false"}' + values: '0+2x12' + alert_rule_test: + - eval_time: 8m + alertname: DocGridEmbeddingRetryableFailureRatioHigh + exp_alerts: [] + + - name: rag fallback and timeout are evaluated independently + interval: 1m + input_series: + - series: 'docgrid_rag_job_completions_total{instance="backend-a:8081",environment="production",cluster="docgrid-production",outcome="provider_fallback"}' + values: '0+1x8' + - series: 'docgrid_rag_job_completions_total{instance="backend-b:8081",environment="healthy",cluster="docgrid-healthy",outcome="success"}' + values: '0+5x8' + - series: 'docgrid_rag_job_completions_total{instance="backend-c:8081",environment="timeout",cluster="docgrid-timeout",outcome="timeout_swept"}' + values: '0+1x8' + alert_rule_test: + - eval_time: 3m + alertname: DocGridRagProviderFallbackSpike + exp_alerts: [] + - eval_time: 4m + alertname: DocGridRagProviderFallbackSpike + exp_alerts: + - exp_labels: + alertname: DocGridRagProviderFallbackSpike + cluster: docgrid-production + dependency: ollama + environment: production + service: rag-worker + severity: warning + exp_annotations: + summary: DocGrid RAG provider fallback is increasing + description: At least three RAG jobs used the provider fallback during the last ten minutes. + - eval_time: 3m + alertname: DocGridRagTimeoutSweepSpike + exp_alerts: [] + - eval_time: 4m + alertname: DocGridRagTimeoutSweepSpike + exp_alerts: + - exp_labels: + alertname: DocGridRagTimeoutSweepSpike + cluster: docgrid-timeout + environment: timeout + service: rag-worker + severity: warning + exp_annotations: + summary: DocGrid RAG timeout recovery is increasing + description: At least three RAG jobs were force-failed by the timeout sweeper during the last ten minutes. + + - name: outbox retries stay quiet and terminal failures alert + interval: 1m + input_series: + - series: 'docgrid_sync_event_attempts_total{instance="backend-a:8081",environment="retrying",cluster="docgrid-retrying",outcome="retry_scheduled"}' + values: '0+2x4' + - series: 'docgrid_sync_event_attempts_total{instance="backend-b:8081",environment="production",cluster="docgrid-production",outcome="terminal_failure"}' + values: '0+1x4' + alert_rule_test: + - eval_time: 1m + alertname: DocGridSyncOutboxTerminalFailure + exp_alerts: + - exp_labels: + alertname: DocGridSyncOutboxTerminalFailure + cluster: docgrid-production + environment: production + service: sync-outbox + severity: critical + exp_annotations: + summary: A DocGrid Sync Outbox event reached terminal failure + description: At least one Sync Outbox event exhausted retries during the last fifteen minutes.