[Feat] Embedding Circuit 상태와 지연 재시도 대기열 관측성 추가

backend/src/main/java/com/opensource/docgrid/domain/embedding/client/EmbeddingProviderCircuitBreaker.java

@@ -16,7 +16,7 @@
  *
  * <p>HTTP 실행 전 Permission만 발급하고 실제 요청·응답 계약에는 관여하지 않는다. 모든 상태 변경은
  * synchronized 경계 안에서 직렬화하며, Open 전 이미 시작된 요청의 늦은 결과가 새 상태를 덮어쓰지
- * 못하도록 세대 번호를 함께 검증한다.
+ * 못하도록 세대 번호를 함께 검증한다. 확정된 상태 전이는 같은 경계 안에서 운영 Metric에 반영한다.
  */
 @Slf4j
 @Component
@@ -26,6 +26,7 @@ public class EmbeddingProviderCircuitBreaker {
 
     private final EmbeddingProviderCircuitBreakerProperties properties;
     private final Clock clock;
+    private final EmbeddingProviderCircuitMetrics metrics;
 
     private CircuitState state = CircuitState.CLOSED;
     private int consecutiveFailures;
@@ -34,14 +35,16 @@ public class EmbeddingProviderCircuitBreaker {
     private boolean halfOpenProbeInProgress;
 
     /**
-     * Circuit 임계값·개방 시간 설정과 테스트 가능한 기준 Clock을 연결한다.
+     * Circuit 임계값·개방 시간 설정, 기준 Clock과 상태 전이 Metric을 연결한다.
      */
     public EmbeddingProviderCircuitBreaker(
         EmbeddingProviderCircuitBreakerProperties properties,
-        Clock clock
+        Clock clock,
+        EmbeddingProviderCircuitMetrics metrics
     ) {
         this.properties = properties;
         this.clock = clock;
+        this.metrics = metrics;
     }
 
     /**
@@ -61,6 +64,7 @@ public synchronized CallPermission acquirePermission() {
             }
             state = CircuitState.HALF_OPEN;
             halfOpenProbeInProgress = false;
+            metrics.recordHalfOpen();
         }
 
         // 3. HALF_OPEN에서는 한 호출에만 Probe 소유권을 주고 나머지는 빠르게 거절한다.
@@ -84,6 +88,8 @@ public synchronized void recordSuccess(CallPermission permission) {
             return;
         }
         if (permission.halfOpenProbe() && state == CircuitState.HALF_OPEN) {
+            // Circuit 닫힘과 함께 회복을 확인한 Probe 결과를 한 번만 기록한다.
+            metrics.recordProbeSuccess();
             closeCircuit();
             return;
         }
@@ -122,6 +128,8 @@ public synchronized Duration recordFailure(
             return Duration.ZERO;
         }
         if (permission.halfOpenProbe() && state == CircuitState.HALF_OPEN) {
+            // 최초 OPEN과 구분해 장애 지속을 입증하는 Probe 실패를 별도로 기록한다.
+            metrics.recordProbeFailure();
             return openCircuit();
         }
         if (state != CircuitState.CLOSED) {
@@ -146,6 +154,7 @@ private Duration openCircuit() {
         consecutiveFailures = 0;
         halfOpenProbeInProgress = false;
         generation++;
+        metrics.recordOpen();
         log.warn(
             "Embedding Provider Circuit을 열었습니다. openDurationMs={}",
             properties.getOpenDuration().toMillis()
@@ -162,6 +171,7 @@ private void closeCircuit() {
         consecutiveFailures = 0;
         halfOpenProbeInProgress = false;
         generation++;
+        metrics.recordClosed();
         log.info("Embedding Provider Circuit이 정상 호출로 닫혔습니다.");
     }
 

backend/src/main/java/com/opensource/docgrid/domain/embedding/client/EmbeddingProviderCircuitMetrics.java

@@ -0,0 +1,96 @@
+package com.opensource.docgrid.domain.embedding.client;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.springframework.stereotype.Component;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.MeterRegistry;
+
+/**
+ * Embedding Provider Circuit의 현재 보호 활성 여부, 상태 전환과 Probe 결과를 Micrometer에
+ * 기록한다.
+ *
+ * <p>Circuit 상태 결정은 하지 않고 {@link EmbeddingProviderCircuitBreaker}가 확정한 전이만 반영한다.
+ * 상태 label 세 값과 Probe 결과 label 두 값만 사용해 시계열 Cardinality가 입력 데이터에 따라
+ * 증가하지 않게 한다.
+ */
+@Component
+public class EmbeddingProviderCircuitMetrics {
+
+    private static final String TRANSITIONS = "docgrid.embedding.provider.circuit.transitions";
+    private static final String PROBES = "docgrid.embedding.provider.circuit.probe";
+
+    private final AtomicInteger open = new AtomicInteger();
+    private final AtomicInteger probeFailed = new AtomicInteger();
+    private final Counter openTransitions;
+    private final Counter halfOpenTransitions;
+    private final Counter closedTransitions;
+    private final Counter successfulProbes;
+    private final Counter failedProbes;
+
+    /** Circuit Gauge와 상태 전환·Probe 결과 Counter를 애플리케이션 시작 시 한 번 등록한다. */
+    public EmbeddingProviderCircuitMetrics(MeterRegistry meterRegistry) {
+        Gauge.builder("docgrid.embedding.provider.circuit.open", open, AtomicInteger::get)
+            .description("Whether the Embedding Provider circuit is open or half-open")
+            .register(meterRegistry);
+        Gauge.builder(
+            "docgrid.embedding.provider.circuit.probe.failed",
+            probeFailed,
+            AtomicInteger::get
+        )
+            .description("Whether the current circuit protection episode has a failed probe")
+            .register(meterRegistry);
+        openTransitions = transitionCounter(meterRegistry, "open");
+        halfOpenTransitions = transitionCounter(meterRegistry, "half_open");
+        closedTransitions = transitionCounter(meterRegistry, "closed");
+        // Counter를 0에서 미리 등록해 첫 Probe 결과도 Prometheus increase()가 관측할 수 있게 한다.
+        successfulProbes = probeCounter(meterRegistry, "success");
+        failedProbes = probeCounter(meterRegistry, "failed");
+    }
+
+    /** Circuit OPEN 전이를 기록하고 현재 차단 상태를 게시한다. */
+    void recordOpen() {
+        open.set(1);
+        openTransitions.increment();
+    }
+
+    /** Open 유예 종료 뒤 단일 Probe를 허용하는 HALF_OPEN 전이를 기록한다. */
+    void recordHalfOpen() {
+        // Probe 성공 전까지 일반 호출은 계속 차단되므로 장기 장애 경보를 끊지 않는다.
+        halfOpenTransitions.increment();
+    }
+
+    /** Probe 성공으로 정상 호출을 재개한 CLOSED 전이를 기록한다. */
+    void recordClosed() {
+        open.set(0);
+        closedTransitions.increment();
+    }
+
+    /** Half-open Probe 성공 횟수를 기록하고 현재 보호 주기의 실패 표시를 해제한다. */
+    void recordProbeSuccess() {
+        successfulProbes.increment();
+        probeFailed.set(0);
+    }
+
+    /** Half-open Probe 실패 횟수를 기록하고 현재 보호 주기에 확인된 장애를 표시한다. */
+    void recordProbeFailure() {
+        failedProbes.increment();
+        probeFailed.set(1);
+    }
+
+    private Counter transitionCounter(MeterRegistry meterRegistry, String state) {
+        return Counter.builder(TRANSITIONS)
+            .description("Embedding Provider circuit transitions by destination state")
+            .tag("state", state)
+            .register(meterRegistry);
+    }
+
+    private Counter probeCounter(MeterRegistry meterRegistry, String outcome) {
+        return Counter.builder(PROBES)
+            .description("Embedding Provider half-open probe results by outcome")
+            .tag("outcome", outcome)
+            .register(meterRegistry);
+    }
+}

backend/src/main/java/com/opensource/docgrid/global/observability/OperationalMetrics.java

@@ -33,6 +33,9 @@ public OperationalMetrics(MeterRegistry meterRegistry, Clock clock) {
         registerGauge(meterRegistry, "docgrid.embedding.claimable.jobs",
             "Embedding jobs that can be claimed now",
             snapshot -> snapshot.embeddingClaimableJobs());
+        registerGauge(meterRegistry, "docgrid.embedding.delayed.retry.jobs",
+            "Embedding retry jobs waiting for their next retry time",
+            snapshot -> snapshot.embeddingDelayedRetryJobs());
         registerGauge(meterRegistry, "docgrid.embedding.processing.jobs",
             "Embedding jobs currently processing",
             snapshot -> snapshot.embeddingProcessingJobs());

backend/src/main/java/com/opensource/docgrid/global/observability/OperationalMetricsSnapshot.java

@@ -6,11 +6,12 @@
 /**
  * 한 시점에 읽은 비동기 Queue와 Worker의 운영 상태를 불변 값으로 보관한다.
  *
- * <p>DB 집계 결과와 Prometheus scrape 사이의 경계다. 오래된 항목이 없으면 해당 시각은 {@code null}이며,
- * 모든 개수는 음수가 될 수 없다.
+ * <p>DB 집계 결과와 Prometheus scrape 사이의 경계다. Embedding의 즉시 실행 가능 항목과 미래 재시도
+ * 항목을 분리하며, 오래된 항목이 없으면 해당 시각은 {@code null}이다. 모든 개수는 음수가 될 수 없다.
  */
 public record OperationalMetricsSnapshot(
     long embeddingClaimableJobs,
+    long embeddingDelayedRetryJobs,
     long embeddingProcessingJobs,
     LocalDateTime embeddingOldestClaimableAt,
     long embeddingActiveWorkers,
@@ -23,7 +24,8 @@ public record OperationalMetricsSnapshot(
 
     /** 집계 Query의 잘못된 결과가 Gauge에 게시되지 않도록 불변 조건을 확인한다. */
     public OperationalMetricsSnapshot {
-        if (embeddingClaimableJobs < 0 || embeddingProcessingJobs < 0 || embeddingActiveWorkers < 0
+        if (embeddingClaimableJobs < 0 || embeddingDelayedRetryJobs < 0
+            || embeddingProcessingJobs < 0 || embeddingActiveWorkers < 0
             || ragProcessingJobs < 0 || syncClaimableEvents < 0 || syncProcessingEvents < 0) {
             throw new IllegalArgumentException("운영 상태 개수는 음수일 수 없습니다.");
         }

backend/src/main/java/com/opensource/docgrid/global/observability/OperationalMetricsSnapshotQueryService.java

@@ -13,7 +13,7 @@
 import lombok.RequiredArgsConstructor;
 
 /**
- * 비동기 Pipeline의 현재 운영 상태를 행 로딩 없이 목적별 aggregate query로 읽는다.
+ * 비동기 Pipeline의 즉시 실행 가능 상태와 미래 재시도 대기 상태를 행 로딩 없이 aggregate query로 읽는다.
  *
  * <p>호출자는 이 짧은 읽기 Transaction의 결과만 메모리에 보관한다. Prometheus scrape 경로에서는 이
  * 서비스를 호출하지 않아, 감시 트래픽이 DB 부하로 이어지지 않게 한다.
@@ -31,6 +31,12 @@ SELECT COUNT(*)
                 WHERE job.status = 'PENDING'
                   AND (job.next_retry_at IS NULL OR job.next_retry_at <= ?)
             ) AS claimable_jobs,
+            (
+                SELECT COUNT(*)
+                FROM embedding_jobs job
+                WHERE job.status = 'PENDING'
+                  AND job.next_retry_at > ?
+            ) AS delayed_retry_jobs,
             (
                 SELECT COUNT(*)
                 FROM embedding_jobs job
@@ -103,6 +109,7 @@ public OperationalMetricsSnapshot load(LocalDateTime observedAt) {
             this::mapEmbeddingSnapshot,
             observedAt,
             observedAt,
+            observedAt,
             heartbeatDeadline
         );
         RagSnapshot rag = jdbcTemplate.queryForObject(RAG_SNAPSHOT_SQL, this::mapRagSnapshot);
@@ -116,6 +123,7 @@ public OperationalMetricsSnapshot load(LocalDateTime observedAt) {
         // 3. 세 Query가 모두 성공한 경우에만 하나의 원자적 Snapshot 후보를 반환한다.
         return new OperationalMetricsSnapshot(
             embedding.claimableJobs(),
+            embedding.delayedRetryJobs(),
             embedding.processingJobs(),
             embedding.oldestClaimableAt(),
             embedding.activeWorkers(),
@@ -130,6 +138,7 @@ public OperationalMetricsSnapshot load(LocalDateTime observedAt) {
     private EmbeddingSnapshot mapEmbeddingSnapshot(ResultSet resultSet, int rowNumber) throws SQLException {
         return new EmbeddingSnapshot(
             resultSet.getLong("claimable_jobs"),
+            resultSet.getLong("delayed_retry_jobs"),
             resultSet.getLong("processing_jobs"),
             resultSet.getObject("oldest_claimable_at", LocalDateTime.class),
             resultSet.getLong("active_workers")
@@ -151,9 +160,10 @@ private SyncSnapshot mapSyncSnapshot(ResultSet resultSet, int rowNumber) throws
         );
     }
 
-    /** Embedding Queue와 유효 Worker를 한 SQL에서 읽은 내부 집계 값이다. */
+    /** Embedding 즉시 실행·지연 재시도 Queue와 유효 Worker를 한 SQL에서 읽은 내부 집계 값이다. */
     private record EmbeddingSnapshot(
         long claimableJobs,
+        long delayedRetryJobs,
         long processingJobs,
         LocalDateTime oldestClaimableAt,
         long activeWorkers

backend/src/test/java/com/opensource/docgrid/domain/document/benchmark/ChunkQualityPerformanceBenchmark.java

@@ -30,10 +30,13 @@
 import com.opensource.docgrid.domain.document.benchmark.ChunkQualityBenchmarkSupport.TimingSummary;
 import com.opensource.docgrid.domain.embedding.client.EmbeddingClient;
 import com.opensource.docgrid.domain.embedding.client.EmbeddingProviderCircuitBreaker;
+import com.opensource.docgrid.domain.embedding.client.EmbeddingProviderCircuitMetrics;
 import com.opensource.docgrid.domain.embedding.config.EmbeddingProviderCircuitBreakerProperties;
 import com.opensource.docgrid.domain.embedding.dto.response.EmbedBatchItemResponse;
 import com.opensource.docgrid.domain.embedding.dto.response.EmbedBatchServerResponse;
 
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -63,7 +66,11 @@ void compareChunkSizeAndOverlapQuality() throws IOException {
         EmbeddingClient embeddingClient = new EmbeddingClient(
             restClient,
             restClient,
-            new EmbeddingProviderCircuitBreaker(circuitProperties, Clock.systemUTC())
+            new EmbeddingProviderCircuitBreaker(
+                circuitProperties,
+                Clock.systemUTC(),
+                new EmbeddingProviderCircuitMetrics(new SimpleMeterRegistry())
+            )
         );
         List<QueryCase> corpus = ChunkQualityBenchmarkSupport.createCorpus();
         List<ChunkProfile> profiles = ChunkQualityBenchmarkSupport.profiles();