Tabellarius CDC, Cursus 생태계의 시작

메시지 브로커로 시작한 오픈소스 프로젝트 cursus는 작업해온 양이 방대해서 문서화하는데도 오래 걸리고 있다.

 

플랫폼화해서 생태계를 이루는 것을 다음 목표로 생각하고 있던 터라, cursus connector에 해당하는 CDC 사이드 프로젝트를 함께 시작했다.

 

https://github.com/downfa11-org/tabellarius

GitHub - downfa11-org/tabellarius: Change data capture source

 

이번 글은 CDC 프로젝트의 시작과 함께 코어 시스템 설계에 대한 논의와 고민해온 과정을 소개하고자 한다.

 

CDC(Change Data Capture)란 무엇인가?

CDC(Change Data Capture)는 DB에 발생한 변경 사항을 감지하고 이를 이벤트로 외부에 전달하는 기술이다.

• 누가, 언제, 어떤 테이블의, 어떤 row를, 어떻게 변경했는지 감지

프로젝트에서는 가장 활발하게 사용되면서 자료도 풍부한 CDC, Debezium을 참고하되, 초기 단계에서는 최대한 가볍게 구현하고자 했다.

그럼 CDC 기술은 어디서 활용되는가?

데이터의 변경 정보는 DB 안에는 분명히 존재하지만, 애플리케이션 바깥에서는 쉽게 접근할 수 없다. 그리고 대부분의 시스템은 이 “변경 사실”을 다른 시스템과 공유해야 한다:

• 검색 인덱스 동기화
• 캐시 갱신
• 통계 집계
• 이벤트 기반 마이크로서비스 연동
• 감사 로그(Audit Log)

 

트랜잭션 아웃박스(Transaction Outbox) 패턴

이벤트 기반 아키텍처에서 다음 상황을 가정해보자:

• (Kafka 장애) DB 트랜잭션은 성공적으로 처리했지만, 그 결과 이벤트를 발행하는 과정에서 장애가 발생한 경우
• (DB 장애) 이벤트 발행에는 성공했지만 DB의 트랜잭션은 실패해서 롤백된 경우

다른 마이크로 서비스들에게 전파되지 않거나, 롤백된 실제값과 다른 내용을 전파받아서 전체 시스템이 불일치한다.

 

이 문제의 본질은 DB 작업과 이벤트 발행이 원자적으로 이뤄지지 않기 때문이다.

 

 

이걸 Spring에서 제공하는 TransactionalEventListener같은 걸로 처리하면 일부면 몰라도 본질적인 해결이 되진 않는다.

@Transactional
public void createOrder() {
    orderRepository.save(order);
    applicationEventPublisher.publishEvent(new OrderCreatedEvent(order));
}

@TransactionalEventListener(phase = AFTER_COMMIT)
public void handle(OrderCreatedEvent event) {
    kafkaTemplate.send(...);
}

 

phase = AFTER_COMMIT가 전송의 성공을 보장하는게 아니다.

 

메시지 발행의 성공 여부는 이미 스프링 트랜잭션의 제어 범위 밖이다.

이미 여기 DB Commit은 완료했지만 메시지 발행에 실패한 시점에서 스프링은 아무것도 해주지 않는다.

 

---

Outbox 패턴의 핵심 아이디어

이 불일치 문제에 대한 해결책은 의외로 단순하다.

• 비즈니스 데이터와 이벤트를 같은 DB 트랜잭션으로 저장
• 이벤트는 직접 발행하지 않고 outbox 테이블에 기록
• 별도의 프로세스가 outbox 테이블을 읽어 외부로 발행

이렇게 되면 DB 트랜잭션이 성공하면 비즈니스 데이터와 이벤트 기록이 항상 함께 존재하고, 롤백된 경우에는 이벤트도 존재하지 않는다.

 

여기서 이 outbox 테이블을 안정적으로 감지해줄 CDC가 필요해진다.

• CDC가 outbox 테이블의 INSERT만 감지하면서, 이벤트를 외부 메시지 브로커로 전달해준다.
• 이때 발행의 성공 여부에 따라서 상태를 관리하면 재처리 등의 후대응이 가능해진다.

애플리케이션이 직접 polling 하는 경우는 성능상의 문제나 중복 처리, 장애 복구에 대한 복잡성만 올라간다. 

 

설계1. Trigger를 통한 Polling 방식의 변경 감지

그리고 사실 이걸로 초기 MVP를 이미 작성했다. 너무 직관적이고 쉽기 때문에 누구나 생각해볼법 했다.

 

기본 골자는 감지하고자 하는 DB와 테이블 정보를 미리 config.yaml로 수집하고 CDC 시작시 초기화 과정을 거친다.

 

config.yaml

database:
  schema: mydb
  user: root
  password: root
  host: mysql
  port: 3306

cdc_log:
  table: cdc_log

tables:
  - name: orders
    pk: id
  - name: users
    pk: id

cdc_server:
  offset_file: offset.txt
  publisher_addr: localhost:9092

 

이 초기화 과정에서는 다음 내용을 확인한다.

• 실제 DB에 접속이 이뤄지는가(ping)?
• 감지할 테이블이 존재하는가?

 

감지할 테이블이 존재한다면 재시작한 경우를 감안해서 트리거가 미리 존재하는지 확인한다.

 

트리거의 생성같은 경우를 자동화하는건 DB 운영하는 입장에서 굉장히 불쾌한 경험이다. 나도 모르는 사이에 운영 DB의 깊숙하게 성능 저하나 장애의 원인이 될지도 모르는 소모가 발생하는 것이기 때문이다.

 

그래서 탐지만 초기화 과정에서 자동으로 처리하고, 실제 트리거 생성은 CLI를 통해서 사용자가 명시적으로 실행하도록 처리했다.

 

MySQL Trigger를 사용해 INSERT / UPDATE / DELETE 발생시 별도의 테이블이나 큐에 변경 내용을 기록한다.

CREATE TRIGGER after_insert_user
AFTER INSERTONuser
FOREACH ROW
INSERT INTO user_cdc_log (...)
VALUES (...);

 

이를 통해서 감지할 테이블에 새로 변경사항이 생기면 Trigger를 통해서 cdc-table에 기록되게 된다. 우리는 Polling 방식으로 이 cdc-table만 매번 조회하면 간단하게 CDC를 구현할 수 있다.

 

detect

func (c *CDCLogInspector) Detect(from uint64, limit int) ([]model.TriggerEvent, error) {
	rows, err := c.db.Query(`
		SELECT seq, table_name, op, row_id, payload
		FROM cdc_log
		WHERE seq > ?
		ORDER BY seq ASC
		LIMIT ?
	`, from, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var events []model.TriggerEvent

	for rows.Next() {
		var (
			seq   uint64
			table string
			op    string
			key   []byte
			value []byte
		)

		if err := rows.Scan(&seq, &table, &op, &key, &value); err != nil {
			return nil, err
		}

		events = append(events, model.TriggerEvent{
			Seq:   seq,
			Table: table,
			Op:    op,
			Key:   key,
			Value: value,
		})
	}

	return events, nil
}

 

polling  

func (e *Engine) Run(ctx context.Context, out chan<- model.TriggerEvent) error {
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			events, err := e.inspector.Detect(e.from, e.batchSize)
			if err != nil {
				return err
			}
			if len(events) == 0 {
				time.Sleep(200 * time.Millisecond)
				continue
			}
			log.Printf("fetched %d events (from=%d)", len(events), e.from)
			for _, evt := range events {
				out <- evt
				e.from = evt.Seq
			}
			if e.offsetFile != "" {
				offset.Save(e.offsetFile, int64(e.from))
			}
		}
	}
}

 

비어 있는 경우, 단순하게 200ms만큼 대기하도록 했지만 좀 더 구체화하면 지수 백오프로 long-polling 하도록 할 생각까지 미리 염두했었다.

---

하지만, 문제는 생각보다 많다

1. 트랜잭션 경계 문제

Trigger는 트랜잭션 내부에서 실행되기 때문에, origin 트랜잭션이 영향을 받게 된다.

• CDC 로직이 느리면?
• 로그 테이블 insert가 병목이면?
• 외부 시스템과 연동하면?

실제 서비스의 로직은 정상적으로 작동했지만, CDC의 로직이 느리거나 cdc-table에 insert하는 과정에서 병목이 생긴다고 상상해보자.

 

운영자 입장에서는 비즈니스 트랜잭션이 CDC 때문에 느려지고 있다.

자기들 서비스도 아닌 외부 시스템에 의해서 발생한 장애를 찾기도 어렵다.

 

이건 정말 최악이다.

 

2. DB schema와의 강한 결합

• 테이블 추가 → Trigger 추가
• 컬럼 변경 → Trigger 수정
• 모든 환경(dev/stage/prod)에 반영

앞서 그 불쾌한 경험으로 설명했지만, 외부 시스템(CDC)에서 DB schema와 강하게 결합된다.

감지할 테이블이 추가되면 트리거도 추가해야하고, 테이블이 변경되면 트리거의 스키마도 같이 변경해줘야 한다.

 

그렇다고 공짜인가? DB 부하가 더 심해진다.

운영 DB는 소중하고 귀중하게 다뤄야하는데 외부 시스템에 의해서 자꾸 쓰기/조회 비용이 추가적으로 든다.

특히 쓰기 트래픽이 많아질수록 락 경합에 대한 문제는 무시할 수 없다.

 

3. 어차피 모든 변경을 모두 수집하지도 못한다.

CDC에서 가장 중요한 건 모든 변경 이벤트를 감지인데, Trigger 방식은 이를 보장하기 어렵다.

 

서비스적으로 soft delete를 구현한 경우는 감지할 수 있겠지만 아예 삭제(hard)해버리면 트리거로도 감지할 수 없다.

보통의 운영 서비스는 바로 삭제해버리지 않고 soft delete 방식을 이용하겠지만.. 그렇다고 해서 외부 시스템이 soft delete 방식을 강제하는 구조는 절대적으로 잘못된 설계라고 생각했다.

 

그리고 트랜잭션으로 처리된 이벤트들도 묶어서 처리하지 않고 개별 이벤트로 읽는다..

 

거기다가 Trigger가 실수로 테이블을 빠트릴 수도 있고, 장애가 발생하는 경우는 어떻게 처리할 것인가? 재처리는?

---

설계2. Binlog 구독 방식의 변경 감지 (like Debezium)

RDBMS Binlog란?

RDBMS의 binary log는 데이터의 변경 사항을 저장하며 복제, 복구 목적으로 사용한다.

대표적으로 MySQL replication 구성시 binlog가 사용된다.

• binlog, transaction log, redo log 등..

커밋되지 않은 트랜잭션은 binlog에 쌓이지 않는다. 변경사항이 없기 때문이다!!

 

binlog format

1. statement: 쿼리문을 평문 기록. now() 그대로 저장해서 데이터 일관성 유지X
2. row(default): 변경 사항이 발생한 row를 base64 인코딩해서 기록
3. mixed: 필요에 따라서 row, statement를 혼합해서 사용

binlog commands

binlog에 관한 설정 변경은 SET GLOBAL 변수를 변경하거나, cnf 파일을 수정하고 재시작하는 방법이 존재한다.

 

실제 binlog 내용 조회:

SHOW BINLOG_EVENTS IN '<binlog-file-name>';

 

하지만 쿼리로 바이너리 로그를 조회하면 로그의 시간이 조회되지 않는다. mysqlbinlog 유틸리티를 다운받아서 지정된 경로(/var/lib/mysql)에 기록된 raw를 직접 디코딩해야한다.

 

 

외부 CDC가 binlog를 읽어오는 방식 

Binlog는 복제(replication)를 위해 설계된 Binlog를 외부의 CDC가 읽기 전용으로 구독하는 방식이다.

 

일단 이 방식을 채택할려면 RDBMS Binlog format에 대한 이해나 RowEvent 처리에 대한 구현상 어려움이 발목을 잡는다.

 

하지만 이를 상쇄시킬 장점들을 소개하고자 한다:

• at-least-once 확보: DDL을 포함한 누락이 전혀 없으며, 트랜잭션 단위의 이벤트들도 모두 묶어서 한꺼번에 기록된다.
• 장애시 특정 시점에서 재처리 관리: binlog pos 기반으로 offset을 관리할 수 있다.
• DB 의존성 분리: 이미 쓰고 있는 binlog를 읽기만 해서 origin 트랜잭션에 아무 영향을 주지 않는다.
  • 당연히 DB Schema 변경에도 아무 영향을 받지 않는다.

 

Trigger 기반이 갖는 단점들이 너무 치명적인 반면, binlog 구독 방식이 시원하게 해결해주는 모습에 사실 이 쯤에서 도입이 결정되었다.

---

Tabellarius에서의 설계 방향

운영 안정성과 확장성을 고려하면 문제는 이걸 어떻게 DB에 부담 없이 운영하면서도, 안정적으로, 빠뜨리지 않고 가져오는지 여부를 중점적으로 비교해야 했다.

 

 

CDC는 편의 기능따위가 아니라 데이터 신뢰성에 기반한 인프라이기 때문에 선택지는 binlog로 좁혀질 수 밖에 없었다.

 

Binlog 방식을 채택하면서 다음 원칙을 세웠다:

• CDC는 DB 외부 프로세스로, Source(DB)와 Sink(Cursus) 분리
• Offset 기반 재처리 기능

 

여담으로, 감지할 DB는 stand-alone으로 제한했다.

운영 DB가 복제본(replication)을 갖는 구조라면 binlog의 형태나 권한에 대한 설정 소요가 발생할 수 있고, 초기 프로젝트에서 고려할 사항은 아니라고 판단했다.

 

Debezium과 유사하지만 더 단순한 구조의 독립 CDC Source를 목표로 했다.

• 각 서비스는 DB를 직접 바라보지 않는다.
• 변경은 이벤트로 전파된다.