AWS Lambda + AOP로 분산 캐시 무효화하기

sequenceDiagram
    participant 관리자
    participant LB as 로드밸런서
    participant A as 서버 A
    participant B as 서버 B
    participant 사용자

    관리자->>LB: 건물 정보 수정
    LB->>A: 요청 전달
    A->>A: DB 업데이트 + 캐시 무효화
    Note over B: 캐시 그대로 유지 (문제!)

    사용자->>LB: 건물 정보 조회
    LB->>B: 요청 전달
    B->>사용자: 오래된 캐시 반환

// 이론적으로는 이게 정답입니다
@Service
public class CacheInvalidationService {
    private final RedisMessagePublisher publisher;

    public void invalidateCache(String key) {
        publisher.publish("cache:invalidate", key);
    }
}

@Component
public class CacheInvalidationSubscriber implements MessageListener {
    private final CacheManager cacheManager;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String key = message.toString();
        cacheManager.getCache("building").evict(key);
    }
}

flowchart LR
    A[Admin API<br/>데이터 수정] --> B[AOP Interceptor<br/>변경 감지]
    B --> C[AWS Lambda<br/>트리거]
    C --> D[API Server 1<br/>/cache/invalidate]
    C --> E[API Server 2<br/>/cache/invalidate]
    C --> F[API Server N<br/>/cache/invalidate]

    style A fill:#e8f5e9
    style C fill:#fff3e0
    style D fill:#e3f2fd
    style E fill:#e3f2fd
    style F fill:#e3f2fd

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface InvalidateCache {
    String cacheName();
    String key();
}

@Aspect
@Component
@Slf4j
public class CacheInvalidationAspect {

    private final LambdaClient lambdaClient;
    private final ObjectMapper objectMapper;

    @AfterReturning("@annotation(invalidateCache)")
    public void invalidateCacheAfterUpdate(
        JoinPoint joinPoint,
        InvalidateCache invalidateCache
    ) {
        try {
            String cacheKey = extractCacheKey(joinPoint, invalidateCache);
            String cacheName = invalidateCache.cacheName();

            // Lambda 비동기 호출
            InvocationRequest request = InvocationRequest.builder()
                .functionName("cache-invalidation-broadcaster")
                .payload(SdkBytes.fromUtf8String(
                    objectMapper.writeValueAsString(
                        Map.of("cacheName", cacheName, "key", cacheKey)
                    )
                ))
                .build();

            lambdaClient.invoke(request);

            log.info("Cache invalidation triggered: {}:{}", cacheName, cacheKey);

        } catch (Exception e) {
            // 캐시 무효화 실패는 서비스 실패로 전파하지 않음
            log.error("Failed to invalidate cache", e);
        }
    }

    private String extractCacheKey(JoinPoint joinPoint, InvalidateCache annotation) {
        Object[] args = joinPoint.getArgs();
        String keyExpression = annotation.key();

        // SpEL 평가 로직 (실제로는 SpelExpressionParser 사용)
        return evaluateKey(keyExpression, args);
    }
}

@Service
public class BuildingService {

    @InvalidateCache(cacheName = "building", key = "#buildingId")
    @Transactional
    public void updateBuilding(Long buildingId, BuildingUpdateDto dto) {
        Building building = buildingRepository.findById(buildingId)
            .orElseThrow();

        building.update(dto);
        buildingRepository.save(building);

        // AOP가 트랜잭션 커밋 후 자동으로 Lambda 호출
    }
}

import json
import boto3
import requests

ec2 = boto3.client('ec2')

def lambda_handler(event, context):
    cache_name = event['cacheName']
    cache_key = event['key']

    # 현재 실행 중인 API 서버 인스턴스 목록 가져오기
    instances = get_api_server_instances()

    success_count = 0
    for instance in instances:
        private_ip = instance['PrivateIpAddress']
        url = f"http://{private_ip}:8080/internal/cache/invalidate"

        try:
            response = requests.post(url, json={
                'cacheName': cache_name,
                'key': cache_key
            }, timeout=3)

            if response.status_code == 200:
                success_count += 1

        except Exception as e:
            print(f"Failed to invalidate on {private_ip}: {e}")
            # 한 서버 실패해도 나머지는 계속 진행
            continue

    return {
        'statusCode': 200,
        'body': json.dumps(f'Invalidated on {success_count}/{len(instances)} instances')
    }

def get_api_server_instances():
    # Auto Scaling Group의 인스턴스 목록 조회
    response = ec2.describe_instances(Filters=[
        {'Name': 'tag:Name', 'Values': ['api-server']},
        {'Name': 'instance-state-name', 'Values': ['running']}
    ])

    instances = []
    for reservation in response['Reservations']:
        for instance in reservation['Instances']:
            instances.append(instance)

    return instances

@RestController
@RequestMapping("/internal/cache")
@Slf4j
public class CacheInvalidationController {

    private final CacheManager cacheManager;

    @PostMapping("/invalidate")
    public ResponseEntity<Void> invalidateCache(
        @RequestBody CacheInvalidationRequest request
    ) {
        String cacheName = request.getCacheName();
        String key = request.getKey();

        Cache cache = cacheManager.getCache(cacheName);
        if (cache != null) {
            cache.evict(key);
            log.info("Cache evicted: {}:{}", cacheName, key);
        }

        return ResponseEntity.ok().build();
    }
}

@Aspect
@Component
public class CacheInvalidationAspect {

    private final StatsDClient statsd;

    @AfterReturning("@annotation(invalidateCache)")
    public void invalidateCacheAfterUpdate(...) {
        long startTime = System.currentTimeMillis();

        try {
            lambdaClient.invoke(request);

            long duration = System.currentTimeMillis() - startTime;

            // Datadog 메트릭 전송
            statsd.increment("cache.invalidation.trigger");
            statsd.histogram("cache.invalidation.duration", duration);
            statsd.increment("cache.invalidation.success");

        } catch (Exception e) {
            statsd.increment("cache.invalidation.failure");
            log.error("Failed to invalidate cache", e);
        }
    }
}