Kafka와 Strimzi를 이용하여 6개의 도메인을 하나의 도메인으로 합쳐보았습니다
안녕하세요. 무신사 커뮤니티 개발팀 백엔드 엔지니어 조유신입니다. 커뮤니티개발팀은 무신사에서 커뮤니티와 관련된 도메인을 맡고 있으며 스냅, 좋아요, 인플루언서 마케팅 등을 담당하는 팀입니다. 이번 포스트에서는 6개의 콘텐츠 도메인을 통합된 단일 도메인 모델로 마이그레이션한 사례를 공유 드리고자 합니다.
레거시 도메인 마이그레이션
레거시 도메인들을 통합된 단일 모델로 제공하기 위해서, API에서 추상화하거나 집계하여 제공하는 것도 가능하지만 분산되어 다른 형태로 저장된 데이터가 주는 구현의 한계점이 존재합니다. 도메인 모델들의 생명주기 관리와 필터, 추천 등의 구현을 쉽게 만들기 위해서는 데이터 통합이 필요합니다.
Strangler Fig Pattern
일반적으로 개선과 마이그레이션을 진행할 때에는 두 가지 결정으로 나누어집니다. 모든 것을 한번에(cut-over, big bang) 마이그레이션할 수도 있고 점진적으로(gradually) 영역을 넓혀가며 개선을 할 수도 있습니다. 저희또한 둘 중의 하나를 고민했고 마틴파울러의 Strangler Fig Pattern에서 해답을 찾을 수 있었습니다.
The most important reason to consider a strangler fig application over a cut-over rewrite is reduced risk. A strangler fig can give value steadily and the frequent releases allow you to monitor its progress more carefully. Many people still don’t consider a strangler fig since they think it will cost more — I’m not convinced about that. Since you can use shorter release cycles with a strangler fig you can avoid a lot of the unnecessary features that cut over rewrites often generate.
서비스의 방향성이 어떻게 바뀔지 모르고 마이그레이션 과정 중에 이전 도메인들의 화면들이 모두 동작해야 하기 때문에, DB DML이나 단순한 배치성 작업을 통한 마이그레이션 보다는 준 실시간에 가까운 레거시 도메인 데이터 마이그레이션 파이프라인을 구축하여 이전 도메인들의 하위호환성을 지키며 새로운 통합 도메인을 등장시키고, 통합 도메인 모델의 제공 영역을 넓혀가는 것으로 결정했습니다.
Data Migration Pipeline
일반적으로 데이터 마이그레이션 파이프라인 구성시에는 선택지가 크게 (1) Batch (2)Lambda Architecture (3)Kappa Architecture로 구성되는데 이 구성들에 대해 간략히 소개해 드리겠습니다.
(1) Batch
일반적으로 백엔드 개발자들이 모두 경험해보았을 배치는 일정 시간마다 동작하는 시스템을 의미합니다. 배치는 일반적으로 Batch Job을 구현하는 것과 Job을 일정한 주기로 동작시켜주는 Scheduler로 구성됩니다. 저희의 배치 인프라는 Spring Batch와 Argo Workflow를 사용하고 있었기 때문에 고가용성 스케줄러와 함께 배치잡에 인스턴스 리소스를 자유롭게 조절하여 사용할 수 있는 환경을 갖추고 있었습니다.
하지만 배치는 위의 그림에서와같이, 일정한 주기로 동작하기 때문에 이전 배치의 동작 이후 다음의 배치 동작 전까지의 신규 데이터는 실시간으로 마이그레이션되지 않는다는 문제점이 있었습니다. 즉 구현이 익숙하고 용이하지만 준 실시간성 보장이 어렵다는 단점이 있습니다.
(2) Lambda Architecture
Lambda Architecture는 이전의 배치 구성과 함께 Stream Layer(Speed Layer)라고 하는 것을 함께 유지하는 아키텍처입니다. 배치가 일정주기로 동작하며 시스템의 정합성을 맞추고 배치 동작 사이의 시간에서는 Stream Layer가 최신 데이터를 처리하여 준 실시간성을 보장해주는 아키텍처입니다.
일반적으로 Stream Layer에 전달되는 데이터는 코드에서 발행하는 도메인 이벤트 혹은 WAL(Write-Ahead Log, MySQL이라면 binlog, MongoDB라면 oplog) 데이터가 될 수 있습니다. Lambda Architecture는 배치와 스트림의 장점을 모두 사용하는 아키텍처로 배치와 스트림 레이어를 이용해 용이한 복원 배치, 준실시간성 보장이 가능하지만 두가지 레이어를 모두 구현 및 관리해야 한다는 단점이 있습니다.
(3) Kappa Architecture
Kappa Architecture는 Lambda Architecture에서 Batch Layer를 제외하고 Speed Layer(이하 Stream)만을 남겨 사용하는 아키텍처로, Lambda Architecture가 여러 레이어를 동시에 유지해야 한다는 단점을 제거한 아키텍처입니다. 현재까지의 구현 사례를 보면 대부분 Kafka와 함께 구현되며, 그만큼 Kafka의 이점을 잘 활용할 수 있는 아키텍처입니다. Kappa Architecture에서 Source Layer는 일반적으로 Kafka Cluster입니다. Kafka Topic에 저장된 데이터를 프로세싱하여 Serving Layer의 Datasource에 가공하고 적재하는 것으로, 이는 흔히 말하는 Kafka Consumer를 의미합니다. Kafka에서 제공하는 Topic Partitioning을 이용해 병렬처리를 쉽게 할 수 있으며 Consumer Offset을 이용해 At-Least Once 실행을 보장할 수 있습니다. 다만 아직은 다른 아키텍처에 비해 구현 사례가 부족하고 러닝 커브가 높다는 단점이 있습니다.
아키텍처 결정
저희 팀은 초기에 구현 경험이 많은 Lambda Architecture를 이용해 레거시 데이터 마이그레이션 파이프라인을 구축하고자 했습니다. 하지만 6개의 도메인 소스로부터 파생되는 테이블의 수는 매우 많았고 Polyglot MSA에서 다양한 언어와 서비스로 이루어진 소스코드에 도메인의 루트 테이블과 파생테이블들의 모든 데이터 변화를 추적하도록 도메인 이벤트 발행 코드를 심는 것은 불가능에 가까웠고, CDC 기술을 도입하기로 했습니다. 이전에 언급했던 것처럼 Kafka Connect Cluster를 구축하고 루트 테이블과 파생테이블을 모두 Kafka Topic에 저장하니 Lambda Architecture의 배치 레이어에서 RDBMS에 접근해 다시 리소스를 가져오는 것이 리소스 낭비라고 느껴졌습니다. 그래서 Kafka Streams를 이용해 테이블들의 CDC Topic의 이벤트 데이터들을 역정규화 하여 마이그레이션하는 로직을 작성했고, 결과적으로 Kappa Architecture와 비슷한 형태를 띠게 되었습니다. 그래서 이 도입 과정에서 Strimzi Operator와 함께 고민했던 것들을 소개합니다.
Strimzi Operator 소개
Strimzi Operator는 쿠버네티스 위에서 Kafka Cluster 및 Kafka와 관련된 자원들을 CRD를 이용해 쿠버네티스 오브젝트로 관리해줄 수 있게 해주는 오픈 소스로 현재 CNCF 프로젝트에도 등록되어있습니다. 이 글을 작성하는 기준 0.41버전까지 공개되어있습니다. Kafka Connect Cluster를 구축하는 당시에 기존에 사용하고 있던 AWS DMS 서비스도 있었지만, 토픽에 대한 설정, SMT 등 기능적 제약이 있었기 때문에 배제했습니다. 그 후 AWS MSK Connect 서비스와 Strimzi Operator 둘 중 고민했지만, AWS MSK Connect는 Terraform, Cloudformation을 통해 관리가 필요하고 커넥터의 설정을 변경할 때 기존 커넥터 태스크를 직접 제거해주어야 했습니다. 쿠버네티스의 YAML 형식이 개발자가 사용하기에 러닝커브가 낮고 편리하다고 판단했고 Strimzi Operator는 커넥터의 설정을 변경하더라도 재배포를 이용해 커넥터를 다시 재생성할 수 있었기 때문에 Strimzi Operator를 사용하고자 결정했습니다.
Strimzi Operator를 이용한 Kafka Connect Cluster 구축
Strimzi Operator를 현재 운영 중인 쿠버네티스 클러스터에 성공적으로 구축하고 RBAC를 적절히 할당했다면, 아래와 같은 CRD를 이용해 Kafka Connect Cluster를 구축할 수 있습니다.
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
# strimzi KafkaConnectSpec : https://strimzi.io/docs/operators/latest/configuring#type-KafkaConnectSpec-reference
# Debezium : https://debezium.io/documentation/reference/stable/operations/kubernetes.html
metadata:
name: # Kafka Connect Cluster 이름
annotations:
strimzi.io/use-connector-resources: "true"
spec:
image: # Connector Plugin 이 설치된 이미지
replicas: # Connect Cluster를 구성하는 파드 개수
bootstrapServers: # kafka Connect의 상태를 저장할 Kafka Cluster
resources:
limits:
memory: #
requests:
cpu: #
memory: #
config:
group.id: #
# Secret을 Kubernetes Secret을 사용한다면 설정
config.providers: secrets
config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
# Connect Cluster의 상태 토픽
offset.storage.topic: offset.storage.topic
status.storage.topic: status.storage.topic
config.storage.topic: config.storage.topic
offset.storage.replication.factor: 2
status.storage.replication.factor: 2
config.storage.replication.factor: 2
# 로깅 설정
logging:
template: # Kafka Connect Cluster Pod의 템플릿 설정
pod:즁요한 설정들만 남겨놓았지만, 생각보다 설정이 간결한 것을 확인하실 수 있습니다. Kafka Connect에서 중요한 설정은 실제로 Kafka Connect Cluster를 구성하는 파드들의 설정과, Kafka Connect에서 관리하는 상태 토픽들, 그리고 보안을 어떻게 관리할 것인지에 대한 것들입니다. 저희는 External Secret Operator를 이용해 AWS Secret을 Kubernetes Secret으로 Sync 하여 컨테이너에 주입하고 사용하고 있으며, 방화벽과 같은 설정들은 PodSelector를 이용해 주입하고 있습니다. 이러한 설정들은 실제로 CM과 Secret 등으로 프로비저닝 되며 배포했을 때 Strimzi Operator가 이러한 설정들을 이용해 파드와 태스크를 자동으로 생성해줍니다. 그리고 Kafka Connect Cluster를 생성할 때 파드의 이미지에는 Kafka Connector Plugin이 미리 설치되어있어야 하는데, buildTemplate이라는 설정을 이용해 이것도 자동화할 수 있습니다.
Connector 생성
저희는 Debezium을 이용해 Source Connector를 구성했습니다. 사전에 Kafka Connect Cluster 컨테이너 이미지에 필요한 Debezium Plugin을 모두 설치해놓았고 아래가 그 중 하나의 소스 커넥터에 대한 예시입니다.
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
name: # 커넥터 이름
labels:
strimzi.io/cluster: # Kafka Connect Cluster 이름
spec:
# MySQLConnector
class: io.debezium.connector.mysql.MySqlConnector
# WAL 처리시 작업 수는 1개를 권장함
tasksMax: 1
# Auto Restart
autoRestart:
enabled: true
maxRestarts: 10
config:
# DB 연결 정보
database.hostname:
database.port:
database.user: ${secrets:k8s시크릿}
database.password: ${secrets:k8s시크릿}
database.server.id:
# 데이터베이스 스키마 기록용 토픽. 내부용으로만 사용되며 Consumer가 사용해서는 안됨
schema.history.internal.kafka.bootstrap.servers:
schema.history.internal.kafka.topic:
schema.history.internal.producer.security.protocol:
schema.history.internal.consumer.security.protocol:
# Debezium MySQL 커넥터가 처음 시작되면 데이터베이스의 일관된 초기 스냅샷을 수행
# 스냅샷 중에 커넥터가 테이블 잠금을 획득하지 못하도록 설정
# 스냅샷이 실행되는 동안 스키마 변경이 발생하지 않는 경우에만 사용하는 것이 안전
snapshot.mode: initial
snapshot.locking.mode: none
# 스키마 변경에 대해 포함 여부 true 일 경우 토픽이 생성되며 스키마 변경이 기록됨
include.schema.changes: false
# 메세지 키에 스키마 정보 포함 여부
key.converter.schemas.enable:
key.converter:
# Signal 허용 채널 설정
signal.data.collection:
# 메세지 변경을 위한 SMT(single message transformations)
transforms: unwrap
transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones: false
transforms.unwrap.delete.tombstone.handling.mode: tombstone
# CDC 토픽 Prefix
topic.prefix:
# 대상 데이터베이스 목록
database.include.list:
table.include.list:
# 토픽 생성 설정
topic.creation.enable:
topic.creation.default.replication.factor:
topic.creation.default.partitions:
topic.creation.default.cleanup.policy: compact
topic.creation.default.compression.type: snappy
# notification
notification.enabled.channels: sink
notification.sink.topic.name: 기본적으로 Source DB에 접근할 수 있는 권한 및 접속 정보가 필요하고 토픽에 발행되는 메시지가 어떤 형태로 적재할지, 또한 SMT(Single Message Transformation)설정을 이용해 원하는 형태로 발행시킬 수 있습니다. 그 밖에도 snapshot.mode를 이용해 커넥터 초기 생성 시 데이터베이스의 데이터를 어떻게 적재할지 결정이 가능하며 커넥터가 자동으로 생성해주는 토픽들의 설정들도 미리 설정할 수 있습니다. 저희는 CDC 토픽들을 이용해 SSOT를 구현하고 이를 이용해 데이터 마이그레이션 파이프라인을 구축하고자 했기 때문에 토픽의 cleanup.ploicy를 compact로 설정하며 생성했습니다.
그밖의 설정들은 아래 링크들은 참고해주시면 감사합니다.
Kafka Event Denormalization 구현
Strimzi Operator를 이용해 성공적으로 Kafka Connect Cluster를 구축하여 마이그레이션 대상 도메인의 테이블들의 CDC 로그를 이용해 각 토픽들을 생성했습니다.
실제로 저희가 구축한 파이프라인을 간략히 소개해 드리면, Source Topic들을 모두 KTable로 만들고 이 KTable의 Changelog들을 Deduplication Processor에 의해 같은 키의 변화에 대해서 중복제거를 한 후 이 Stream과 기존에 생성했던 KTable들을 Join하여 통합된 데이터 모델을 만들고 이 모델을 Sink 하였습니다. 코파티셔닝을 위해 CDC Topic을 한번 더 다른 Topic으로 Sink하여 파티션 개수를 유동적으로 가져갈 수 있도록 구성했습니다.
Data Replay
개발 과정 중에 혹은 운영 중에 마이그레이션 요구사항이 변경된다거나, 추가적인 데이터 마이그레이션을 필요로 한다면 처음부터 다시 한번 더 마이그레이션이 필요합니다. 위와 같은 아키텍처에서 파이프라인을 전체 데이터 재마이그레이션 하기 위해서는 다음과 같은 방법들이 있습니다.
- Debezium Signal 기능을 이용한 Incremental Snapshot 이용
- Streams Application Consumer Offset 초기화 혹은 Application ID 변경
- Rich Data Stream의 Join Trigger로 이용하는 Deduplication Processor 이후 토픽에 대해 원하는 소스 스트림의 루트 스트림으로 재발행
저희는 위 세 가지 과정을 모두 구현했습니다. 저희의 서비스는 사용자 트래픽을 실시간으로 받으며 실시간 데이터에 대한 처리 보장도 필요하므로, 스트림즈 애플리케이션과 최종 Sink Consumer/Connector의 Consumer Lag이 발생하지 않는 속도로 조절하여 전체 데이터를 재 마이그레이션할 수 있도록 전체 데이터를 재실행하는 과정에서 적절한 설정과 속도를 부여했습니다. 또 다른 방법으로는, 이러한 파이프라인을 두 버전으로 유지하는 방법이 있는데, Sink 대상인 RDBMS의 한계와 코드를 이원화시키지 않고 해결할 방법이 떠오르지 않아 우선 보류해두었습니다. 현재 이러한 것을 더욱 원활하게 할 수 있도록 하는 티켓이 Kafka Streams에 발행되어있습니다.
이밖에도 상세 내용은 다음과 같습니다
- Kotlin의 Extension Function 문법을 이용하면 Kafka Streams 코드를 가독성 높게 작성할 수 있습니다
- Source Datasource의 실제 데이터 크기보다 파이프라인에서 사용되는 데이터의 크기가 더욱 큽니다
- Kafka Connect Cluster의 로그는 Vector를 이용해 수집합니다
- Kafka Streams 관련 설정은 무신사의 이전 포스트를 참고 부탁드립니다
결론
저희는 Polyglot MSA 환경에서 6개의 도메인을 하나의 통합 도메인 모델로 마이그레이션하기 위해 Strimzi Operator를 이용해 Kafka Connect Cluster를 구축하고 CDC 데이터와 Kafka Streams를 이용해 데이터 마이그레이션 파이프라인을 구축했습니다. 결론적으로 별도의 배치 레이어 없이 소스 데이터베이스의 부하를 줄이며 Kafka가 주는 병렬성과 고가용성, 높은 처리량을 기반으로 준 실시간성을 보장할 수 있었습니다. Strimzi Operator를 통해 개발자들이 CDC를 손쉽게 도입할 수 있게 되며 다양한 곳에서 데이터 통합도 손쉽게 이루어지고 있습니다. KSQL과 같은 도구들이 더욱 발전한다면 이러한 파이프라인을 더욱 손쉽게 개발할 수 있을 것으로 생각합니다.
개발팀과 SRE팀 동료들의 적극적인 지원과 조언 덕분에 파이프라인을 성공적으로 구축할 수 있었습니다. 무신사는 큰 규모의 트래픽을 다루며 빠르게 성장을 하고 있지만, 기술로 변화를 만들어내는 것에 열려있는 조직입니다. 기술을 통해 큰 변화를 만들어내고 싶으신 분들은 무신사에 많은 관심을 가져주시면 감사합니다.
Musinsa CAREER
함께할 동료를 찾습니다.
이처럼 무신사는 매년 빠르게 성장하며 새로운 문제를 마주하고, 문제를 해결하기 위해 새로운 기술을 적극적으로 도입하고 있습니다. 전국민이 사용하는 1위 패션 플랫폼 무신사에서 기술로 비즈니스를 성장시키는 경험을 함께하고 싶으시다면 아래 채용 페이지를 통해 지원해 주세요!
🚀 무신사 채용 페이지 : https://corp.musinsa.com/ko/career
