FoodDonor CDC 도입

배경

FoodDonor의 구축을 어느 정도 완료했고, 푸드뱅크 담당자의 인터뷰 내용을 확인하던 중,

FoodDonor 관계자 같이 책임자들이 매칭 시 발생하는 로그를 종합적으로 확인하면 좋겠다는 의견을 듣게 되었음

또한 매칭 수행 후 기록들을 남겨서 가지고 있으면 좋겠다고 하셨음

 

현재 화면을 보면 관리자들이 한번에 시각화해서 확인할 수 있는 화면은 찾기 어려웠음

여기서 관리자란 매칭을 전반적으로 관리할 수 있는 푸드뱅크의 담당자들과 서비스를 관리할 수 있는 사람들을 말함

 

물론 관리자 전용 화면을 추가 구축하면 될거 같다고 했지만, 성공 및 실패(오류를 확인하기 위한...)로그를 확인할 수 있는 페이지와 매칭 로그만 정상적으로 확인을 하는 관리자들이 구분될 필요는 있다고 생각을 했음

 

또한 서비스를 운영하는 관리자들 역시 분리되어있는 로그들을 확인할때 어디서 나온 오류이고, 어떤 시점에서 발생한 오류를 각 서비스마다 확인하는 것은 비효율적이었음

 

그렇기에 CDC를 통해 DynamoDB에 있는 Table들 변화에 대한 로그를 시각화하면 이 문제들을 해결할 수 있을 것이라 생각했음

현재 서비스들은 이벤트가 일어난 시점에 대해서는 DB에 기록을 하고 있기에 DB에 대해 CDC를 적용해도

위와 같은 고민들에 대해 어느 정도 도움을 줄 수 있을 것임

---

CDC 솔루션

보통 CDC라고 하면 Debezium을 사용해야 하는거라고 생각을 할 것임

CDC는 Change Data Capture이기에, Data의 변화에 대한 감지 솔루션들을 포함하는 거라고 생각을 하면 됨

 

현재 우리 아키텍처는 AWS의 DynamoDB를 사용하고 있기에, 굳이 Debezium과 Kafka 등..을 도입할 필요가 없음

또한 Debezium은 RDBMS의 로그를 읽는데 특화되어 있고(물론 MonogoDB도 지원...)

이미 AWS DynamoDB에서는 DynamoDB Streams 내장형 CDC를 통해 AWS가 직접 커널 레벨에서 최적화 지원을 하고 있음

DynamoDB Streams를 활용해 솔루션을 구축을 해봤음

 

---

구축

먼저 데이터를 받아주기위한(target) 솔루션으로 Opensearch를 사용할 것임

프로젝트의 추구 방향 중 클라우스 솔루션 사용과, 최소 비용을 원칙이 있기에 가장 비용이 저렴한 인스턴스를 찾아 사용할 것임

 

 

들어가서 원하는 조건에 맞춰 구축을 하면 아래와 같이 생성됨을 알 수 있음

만들어지는데 생각보다 시간이 오래 걸렸음...(20분 넘게 걸림...)

 

생성을 할때 노드나 인스턴스를 쓸 만큼만 선택해서 비용을 최소화하는게 중요함(물론 목적에 맞겍 하면 될듯)

 

이제 DynamoDB의 Tables에 대해 DynamoDB Streams를 Turn On 해주자

이때 New and old images를 해줘야 CDC를 수행할 수 있음

이 설정을 통해서 람다에게 과거와 현재 image 두 개를 제공할 수 있음

 

이제 lambda가 DynamoDB를 보고 Opensearch에 쓸 수 있는 권한을 만들거임

 

lambda함수도 작성해주고 deploy까지 완료

코드를 살펴보면

멱등성 문제

async function sendToOpenSearch(index, data, docId) {
    // URL 끝에 /_doc/아이디 를 붙여서 이 아이디로 덮어써라(PUT)고 명령
    const url = `${OPENSEARCH_DOMAIN}/${index}/_doc/${docId}`;
    
    const response = await fetch(url, {
        method: 'PUT', 
        headers: {
            'Content-Type': 'application/json',
            'Authorization': AUTH
        },
        body: JSON.stringify(data)
    });

    if (!response.ok) {
        const errMsg = await response.text();
        console.error(`[전송 실패] Status: ${response.status}, Msg: ${errMsg}`);
        throw new Error('OpenSearch 전송 실패');
    }
    console.log(`[전송 성공] ID: ${docId}, Type: ${data._log_type}`);
}

 

처음 lambda 함수 코드를 POST 방식으로 작성하고 시험을 하던 중 오류가 발생하면,

retry가 수행되는데 이때 동일 이벤트가 두 번 발생되는 문제가 발생했음

왜냐하면 POST로 수행을 하면 매번 ID가 새로 생성되기에 똑같은 로그가 2개, 3개가 중복해서 쌓이게 된거임

그래서 DynamoDB의 PK를 OpenSearch의 문서 ID로 그대로 써서 PUT을 통해 해결했음

 

트리거도 추가해주고..

 

코드는 db의 변화를 감지할 수 있는 수준으로 작성하였음

 

Test

{
  "donation_id": { "S": "TEST-DONATION-001" },
  "item_name": { "S": "도시락" },
  "category": { "S": "한식" },
  "quantity": { "N": "50" },
  "status": { "S": "PENDING" },
  "donor_id": { "S": "TEST-DONOR-999" },
  "created_at": { "S": "2026-01-16T12:00:00Z" }
}

DynamoDB에 테스트를 위해 Insert를 줘보고 흐름을 살펴보자

 

 

opensearch 대시보드에 접속해서 로그인을 하고 결과를 살펴보면 됨

GET integrated-logs/_search
{
  "sort": [
    {
      "_timestamp": {
        "order": "desc"
      }
    }
  ]
}

코드를 작성하면 아래와 같은 로그 수집 된 것을 확인할 수 있음

 

 

 

최종적으로 이렇게 구성이 되었음

시각화를 위해서 Opensearch를 가져다가 어떻게 시각화 할지 디자인도 고민을 해보면서 마무리를 해봄...