fix: Parallel Call Observation Add And DB/Redis Observation Add #105

Author: oOccasio

build.gradle

@@ -103,6 +103,8 @@ dependencies {
     implementation 'io.micrometer:micrometer-tracing-bridge-otel'
     // OTLP Exporter — Alloy로 트레이스 전송
     implementation 'io.opentelemetry:opentelemetry-exporter-otlp'
+    // DB (JDBC/JPA) 쿼리 단위 Span — 각 SQL 쿼리를 child Span으로 계측
+    implementation 'net.ttddyy.observation:datasource-micrometer-spring-boot:1.0.6'
 
     //resilience4j
     implementation 'io.github.resilience4j:resilience4j-spring-boot3'

src/main/java/sopt/comfit/global/config/RedisConfig.java

@@ -1,5 +1,8 @@
 package sopt.comfit.global.config;
 
+import io.lettuce.core.tracing.MicrometerTracing;
+import io.micrometer.observation.ObservationRegistry;
+import org.springframework.boot.autoconfigure.data.redis.ClientResourcesBuilderCustomizer;
 import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -17,6 +20,15 @@
 @Configuration
 public class RedisConfig {
 
+    /**
+     * Lettuce Redis 커맨드를 Micrometer Observation으로 계측 → Tempo에 child Span으로 전송
+     * includeCommandArgsInSpanTags=true : GET key, SET key value 등 인자를 span tag로 기록
+     */
+    @Bean
+    public ClientResourcesBuilderCustomizer lettuceTracingCustomizer(ObservationRegistry observationRegistry) {
+        return builder -> builder.tracing(new MicrometerTracing(observationRegistry, "redis", true));
+    }
+
     @Bean
     public RedisTemplate<String, String> redisTemplate(RedisConnectionFactory factory) {
         RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();

src/main/java/sopt/comfit/report/job/AIReportJobService.java

@@ -1,5 +1,7 @@
 package sopt.comfit.report.job;
 
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.Context;
 import lombok.RequiredArgsConstructor;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.data.redis.core.StringRedisTemplate;
@@ -12,6 +14,9 @@
 import sopt.comfit.report.dto.command.MatchExperienceCommandDto;
 import sopt.comfit.report.exception.AIReportErrorCode;
 
+import java.util.HashMap;
+import java.util.Map;
+
 @Service
 @RequiredArgsConstructor
 public class AIReportJobService {
@@ -36,7 +41,13 @@ public Long createJob(MatchExperienceCommandDto command) {
 
         reportJobRepository.save(job);
 
-        eventPublisher.publishEvent(new JobCreatedEvent(job.getId()));
+        // 현재 HTTP 요청의 trace context를 W3C traceparent 포맷으로 추출
+        // Worker에서 복원해 job.process span을 HTTP 요청 trace의 child로 연결
+        Map<String, String> carrier = new HashMap<>();
+        W3CTraceContextPropagator.getInstance().inject(Context.current(), carrier, Map::put);
+        String traceparent = carrier.get("traceparent");
+
+        eventPublisher.publishEvent(new JobCreatedEvent(job.getId(), traceparent));
 
         return job.getId();
     }

src/main/java/sopt/comfit/report/job/AIReportJobWorker.java

@@ -2,6 +2,9 @@
 
 import io.micrometer.observation.Observation;
 import io.micrometer.observation.ObservationRegistry;
+import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapGetter;
 import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 import lombok.RequiredArgsConstructor;
@@ -86,10 +89,16 @@ private void guardedListen(int index) {
     private void listen() {
         while (!Thread.currentThread().isInterrupted() && running.get()) {
             try {
-                String jobId = redisTemplate.opsForList()
+                String raw = redisTemplate.opsForList()
                         .rightPop(Constants.JOB_QUEUE_KEY, Duration.ofSeconds(30));
-                if (jobId == null) continue;
-                processJob(Long.parseLong(jobId));
+                if (raw == null) continue;
+
+                // "jobId|traceparent" 또는 "jobId" 형태 파싱
+                String[] parts = raw.split("\\|", 2);
+                Long jobId = Long.parseLong(parts[0]);
+                String traceparent = parts.length > 1 ? parts[1] : null;
+
+                processJob(jobId, traceparent);
 
             } catch (QueryTimeoutException e) {
                 log.debug("BRPOP timeout, 재시도");
@@ -109,14 +118,29 @@ private void sleep(int seconds) {
         }
     }
 
-    private void processJob(Long jobId) {
-        // 루트 Span 생성 — Worker는 HTTP 요청이 아니라 Micrometer가 자동 생성 안 함
-        // Observation을 시작하면 traceId/spanId가 MDC에 자동 주입되고
-        // 내부의 Feign 호출(perspectives, density 등)이 child Span으로 자동 연결됨
-        Observation observation = Observation.createNotStarted("job.process", observationRegistry)
-                .lowCardinalityKeyValue("jobId", String.valueOf(jobId));
-
-        observation.observe(() -> {
+    private void processJob(Long jobId, String traceparent) {
+        // traceparent가 있으면 HTTP 요청 trace의 child span으로 연결
+        // 없으면 새 root trace 시작 (이전 버전 Redis 값 대비 방어)
+        Context parentContext = traceparent != null
+                ? W3CTraceContextPropagator.getInstance().extract(
+                        Context.root(), traceparent,
+                        new TextMapGetter<String>() {
+                            @Override
+                            public Iterable<String> keys(String carrier) {
+                                return java.util.List.of("traceparent");
+                            }
+                            @Override
+                            public String get(String carrier, String key) {
+                                return "traceparent".equals(key) ? carrier : null;
+                            }
+                        })
+                : Context.current();
+
+        try (io.opentelemetry.context.Scope ignored = parentContext.makeCurrent()) {
+            Observation observation = Observation.createNotStarted("job.process", observationRegistry)
+                    .lowCardinalityKeyValue("jobId", String.valueOf(jobId));
+
+            observation.observe(() -> {
             try {
                 // jobId는 Micrometer가 자동으로 안 넣어주므로 직접 설정
                 MdcUtils.setJobId(jobId);
@@ -151,6 +175,7 @@ private void processJob(Long jobId) {
                 MdcUtils.clear();
             }
         });
+        } // try (parentContext.makeCurrent())
     }
 
     private MatchExperienceCommandDto buildCommand(Long jobId) {

src/main/java/sopt/comfit/report/job/JobCreatedEvent.java

@@ -1,4 +1,4 @@
 package sopt.comfit.report.job;
 
-public record JobCreatedEvent(Long jobId) {
+public record JobCreatedEvent(Long jobId, String traceparent) {
 }

src/main/java/sopt/comfit/report/job/JobEventListener.java

@@ -15,9 +15,10 @@ public class JobEventListener {
 
     @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
     public void handleJobCreated(JobCreatedEvent event) {
-        stringRedisTemplate.opsForList().leftPush(
-                Constants.JOB_QUEUE_KEY,
-                String.valueOf(event.jobId())
-        );
+        // traceparent가 있으면 "jobId|traceparent" 형태로 저장해 Worker가 parent trace에 연결할 수 있도록
+        String value = event.traceparent() != null
+                ? event.jobId() + "|" + event.traceparent()
+                : String.valueOf(event.jobId());
+        stringRedisTemplate.opsForList().leftPush(Constants.JOB_QUEUE_KEY, value);
     }
 }