스프링 프로젝트에 Langfuse 도입하기 [2] — 프로덕션 안정화

@Getter
@NoArgsConstructor
@AllArgsConstructor(staticName = "of")
public class CachedPrompt {
    private String promptText;
    private PromptConfigData config;
    private Integer version;
}

@Bean
public RedisTemplate<String, Object> langfuseRedisTemplate(
    RedisConnectionFactory connectionFactory) {
 
    RedisTemplate<String, Object> template = new RedisTemplate<>();
    template.setConnectionFactory(connectionFactory);
    template.setKeySerializer(new StringRedisSerializer());
    template.setValueSerializer(
        new GenericJackson2JsonRedisSerializer(objectMapper()));
    template.setHashKeySerializer(new StringRedisSerializer());
    template.setHashValueSerializer(
        new GenericJackson2JsonRedisSerializer(objectMapper()));
    return template;
}
 
private ObjectMapper objectMapper() {
    ObjectMapper mapper = new ObjectMapper();
    mapper.registerModule(new Jdk8Module());
    mapper.activateDefaultTypingAsProperty(
        mapper.getPolymorphicTypeValidator(),
        ObjectMapper.DefaultTyping.NON_FINAL,
        "type"
    );
    return mapper;
}

@Service
@RequiredArgsConstructor
public class LangfusePromptCacheService {
 
    private static final String CACHE_PREFIX = "langfuse-prompt::";
 
    private final RedisTemplate<String, Object> langfuseRedisTemplate;
    private final LangfuseClient langfuseClient;
 
    @Value("${langfuse.cache.ttl-seconds:1800}")
    private int ttlSeconds;
 
    public CachedPrompt getCachedPromptByLabel(String promptName, PromptLabel label) {
        String cacheKey = CACHE_PREFIX + promptName + ":" + label.getValue();
 
        // 캐시에서 조회
        CachedPrompt cached = getFromCache(cacheKey);
        if (cached != null) {
            return cached;
        }
 
        // 캐시 미스 → Langfuse API 호출
        CachedPrompt fetched = fetchFromLangfuse(promptName, label);
        putToCache(cacheKey, fetched);
        return fetched;
    }
 
    private CachedPrompt getFromCache(String key) {
        try {
            return (CachedPrompt) langfuseRedisTemplate.opsForValue().get(key);
        } catch (Exception e) {
            log.warn("[LangfuseCache] 캐시 조회 실패: {}", key, e);
            return null;
        }
    }
 
    private void putToCache(String key, CachedPrompt value) {
        try {
            langfuseRedisTemplate.opsForValue()
                .set(key, value, Duration.ofSeconds(ttlSeconds));
        } catch (Exception e) {
            log.warn("[LangfuseCache] 캐시 저장 실패: {}", key, e);
        }
    }
}

private final ConcurrentHashMap<String, ReentrantLock> lockMap = new ConcurrentHashMap<>();
 
public CachedPrompt getCachedPromptByLabel(String promptName, PromptLabel label) {
    String cacheKey = CACHE_PREFIX + promptName + ":" + label.getValue();
 
    // 1차 체크: 락 없이 빠른 경로
    CachedPrompt cached = getFromCache(cacheKey);
    if (cached != null) {
        return cached;
    }
 
    // 락 획득
    ReentrantLock lock = lockMap.computeIfAbsent(cacheKey, k -> new ReentrantLock());
    lock.lock();
    try {
        // 2차 체크: 다른 스레드가 이미 캐시를 채웠는지 확인
        cached = getFromCache(cacheKey);
        if (cached != null) {
            return cached;
        }
 
        // 이 스레드만 API 호출
        CachedPrompt fetched = fetchFromLangfuse(promptName, label);
        putToCache(cacheKey, fetched);
        return fetched;
    } finally {
        lock.unlock();
    }
}

public class PromptLocalLockManager {
 
    private static final long LOCK_EXPIRY = Duration.ofMinutes(10).toMillis();
    private static final long CLEANUP_INTERVAL = Duration.ofMinutes(5).toMillis();
 
    private final ConcurrentHashMap<String, LockEntry> keyLocks = new ConcurrentHashMap<>();
    private final ScheduledExecutorService cleanupScheduler;
 
    private record LockEntry(ReentrantLock lock, AtomicLong lastUsed) {
        static LockEntry create() {
            return new LockEntry(new ReentrantLock(), new AtomicLong(System.currentTimeMillis()));
        }
 
        void touch() {
            lastUsed.set(System.currentTimeMillis());
        }
 
        boolean isExpired() {
            return System.currentTimeMillis() - lastUsed.get() > LOCK_EXPIRY;
        }
    }
 
    public PromptLocalLockManager() {
        this.cleanupScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "langfuse-lock-cleanup");
            t.setDaemon(true);
            return t;
        });
        cleanupScheduler.scheduleAtFixedRate(
            this::cleanupUnusedLocks,
            CLEANUP_INTERVAL, CLEANUP_INTERVAL, TimeUnit.MILLISECONDS
        );
    }
 
    public <T> T executeWithLock(String key, Supplier<T> action) {
        LockEntry entry = keyLocks.computeIfAbsent(key, k -> LockEntry.create());
        entry.touch();
        entry.lock().lock();
        try {
            return action.get();
        } finally {
            entry.lock().unlock();
        }
    }
 
    private void cleanupUnusedLocks() {
        keyLocks.entrySet().removeIf(e ->
            e.getValue().isExpired() && !e.getValue().lock().isLocked()
        );
    }
}

public CachedPrompt getCachedPromptByLabel(String promptName, PromptLabel label) {
    String cacheKey = CACHE_PREFIX + promptName + ":" + label.getValue();
 
    // 1차 체크
    CachedPrompt cached = getFromCache(cacheKey);
    if (cached != null) return cached;
 
    // 2차 체크 + API 호출 (락 내부)
    return lockManager.executeWithLock(cacheKey, () -> {
        CachedPrompt rechecked = getFromCache(cacheKey);
        if (rechecked != null) return rechecked;
 
        CachedPrompt fetched = fetchFromLangfuse(promptName, label);
        putToCache(cacheKey, fetched);
        return fetched;
    });
}

@Service
@RequiredArgsConstructor
public class LangfuseDistributedLockService {
 
    private final StringRedisTemplate redisTemplate;
 
    @Value("${langfuse.cache.ttl-seconds:1800}")
    private int cacheTtl;
 
    private static final String LOCK_SCRIPT = """
        if redis.call('set', KEYS[1], ARGV[1], 'NX', 'EX', ARGV[2]) then
            return 1
        else
            return 0
        end
        """;
 
    private static final String UNLOCK_SCRIPT = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """;
 
    public String tryLock(String key, int waitTimeSeconds) {
        String lockValue = UUID.randomUUID().toString();
        int lockTtl = cacheTtl + 10;  // 캐시 TTL보다 10초 길게
 
        long deadline = System.currentTimeMillis() + (waitTimeSeconds * 1000L);
 
        while (System.currentTimeMillis() < deadline) {
            Long result = redisTemplate.execute(
                new DefaultRedisScript<>(LOCK_SCRIPT, Long.class),
                List.of(key),
                lockValue, String.valueOf(lockTtl)
            );
 
            if (result != null && result == 1) {
                return lockValue;  // 락 획득 성공
            }
 
            try {
                Thread.sleep(100);  // 100ms 후 재시도
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return null;
            }
        }
        return null;  // 타임아웃
    }
 
    public void unlock(String key, String lockValue) {
        Long result = redisTemplate.execute(
            new DefaultRedisScript<>(UNLOCK_SCRIPT, Long.class),
            List.of(key),
            lockValue
        );
 
        if (result == null || result == 0) {
            log.warn("[DistributedLock] 락 해제 실패 — 이미 만료되었거나 다른 인스턴스가 보유: {}", key);
        }
    }
}

private final ScheduledExecutorService cacheRefreshScheduler;
private final Set<String> scheduledRefreshKeys = ConcurrentHashMap.newKeySet();
 
private void scheduleAsyncCacheRefresh(String cacheKey, String promptName,
                                        PromptLabel label, int ttl) {
    if (ttl <= 5) return;  // TTL이 5초 이하면 비동기 갱신 의미 없음
 
    // 이미 스케줄된 키면 중복 등록 방지
    if (!scheduledRefreshKeys.add(cacheKey)) return;
 
    int delaySeconds = ttl - 5;
 
    cacheRefreshScheduler.schedule(() -> {
        try {
            executeAsyncCacheRefresh(cacheKey, promptName, label);
        } finally {
            scheduledRefreshKeys.remove(cacheKey);
        }
    }, delaySeconds, TimeUnit.SECONDS);
}

private void executeAsyncCacheRefresh(String cacheKey, String promptName,
                                       PromptLabel label) {
    // 분산 락 획득 (1초 대기)
    String lockKey = "lock::" + cacheKey;
    String lockValue = distributedLockService.tryLock(lockKey, 1);
 
    if (lockValue == null) {
        log.debug("[LangfuseCache] 다른 인스턴스가 갱신 중: {}", cacheKey);
        return;  // 다른 인스턴스가 하고 있으니 건너뜀
    }
 
    try {
        CachedPrompt fetched = fetchFromLangfuse(promptName, label);
        putToCache(cacheKey, fetched);
        log.info("[LangfuseCache] 비동기 갱신 완료: {}", cacheKey);
 
        // 다음 갱신도 스케줄
        scheduleAsyncCacheRefresh(cacheKey, promptName, label, ttlSeconds);
    } catch (Exception e) {
        log.warn("[LangfuseCache] 비동기 갱신 실패: {}", cacheKey, e);
        // 실패해도 기존 캐시가 남아있으니 서비스에 영향 없음
    } finally {
        distributedLockService.unlock(lockKey, lockValue);
    }
}

@Bean
public ScheduledExecutorService cacheRefreshScheduler() {
    ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2);
    executor.setThreadFactory(r -> {
        Thread t = new Thread(r, "langfuse-cache-refresh");
        t.setDaemon(true);
        return t;
    });
    executor.setRemoveOnCancelPolicy(true);
    return executor;
}

private CachedPrompt getLastKnownGood(String cacheKey) {
    try {
        // TTL 만료된 캐시에서 읽기 시도
        // (별도의 LKG 저장소를 쓰거나, TTL 없는 백업 키를 유지)
        return (CachedPrompt) langfuseRedisTemplate.opsForValue()
            .get("lkg::" + cacheKey);
    } catch (Exception e) {
        return null;
    }
}