MongoDB to ElasticSearch Realtime sync

MongoDB to ElasticSearch Realtime sync

이기종 DB간의 데이터 동기화는 여러가지 방법이 있습니다. CDC를 이용하는 방법, 카프카를 이용하는 방법 등 여러가지가 있고, MongoDB에서 ElasticSearch 로 데이터를 동기화 하는 방법 역시 Logstash를 이용한다거나 다른 방식을 이용하는 방법이 있습니다.

제가 MongoDB에서 ElasticSearch로 데이터를 동기화 하려고 결정한 이유는 크게 두가지가 있습니다.

첫번째, MongoDB의 ngram 검색의 부재

두번째, 누적데이터에 대한 처리

Percona MongoDB를 사용하면 ngram에 대한 문제도 해결되고, Percona fork에서 사용할 수 있는 다양한 기능들을 무료로 사용할 수 있습니다. 하지만 미묘하게 다른 설정, 문제가 생기면 Percona에 의지해야 한다는 단점 때문에 Percona MongoDB의 선택을 다소 망설이게 만들었습니다.

그래서 선택한것이 엘라스틱 서치에 실시간 동기화를 통해 검색에 이용되는 컬렉션들을 엘라스틱 서치로 보내고 전문 검색엔진의 기능을 활용하는 것을 선택했습니다.

두번째 누적 데이터에 대한 처리입니다. 로그성 수집데이터와 비슷한 느낌으로 쌓이는 컬렉션에서 너무 많은 데이터가 쌓인경우 운영에 사용하는 DB에 결코 좋을리가 없다는 것이죠. 물론 MongoDB가 샤딩을 통해 분산 처리가 잘 된다는 장점이 있지만, 샤드된 클러스터에서는 aggregation 쿼리들이 샤드의 특성을 타는 경우도 많고, 운영 DB에 로그성 누적데이터를 위해 샤드를 늘려가는것도 비효율적이라고 생각했습니다. 그리고 데이터 누적과 누적된 데이터를 통한 데이터 분석, 통계 작업을 운영에서 할 수도 없는 노릇이기에 아예 처음부터 분산해서 가자고 결정을 했습니다.

엘라스틱 서치를 선택한 이유는 똑같은 분산 처리에 강력하고, DB보다는 검색엔진으로 제품을 내세울 정도로 강력한 검색기능을 가지고 있으며, 키바나를 통한 시각화가 쉽고, MongoDB와 동일하게 JSON 타입의 데이터를 저장하는 것이었습니다.

MongoDB에서 업데이트가 일어나지 않는 로그성 누적데이터를 쌓는 컬렉션을 Capped로 생성한 다음, 해당 컬렉션에 쌓이는 데이터를 엘라스틱 서치로 보내는 것이죠. 애플리케이션 서버는 MongoDB에서 직접 검색을 하는 것이 아니라, 엘라스틱에서 검색을 한 다음 ObecjtID만 결과값으로 반환하여 해당 ObjectID를 가지고 MongoDB에서 필요한 데이터를 소환하는 방식입니다.

HashTag가 누구에 의해 언제 어디에 인용되어 있는지를 기록하는 컬렉션에서 기간별, 이용자 별, 태그가 인용된 포스트 검색등 다양한 검색 기능을 갖춰야 했기 때문에 조건에 맞는 검색은 엘라스틱으로 해서 그에 맞는 ObjectID만 골라서 데이터를 로딩할 수 있게 처리한 구조입니다.

Monstache

우선 동기화에 이용한 툴은 Monstache 입니다.

다음은 Monstache에서 제공하는 것들입니다.

• Supports up to and including the latest versions of Elasticsearch and MongoDB
• Single binary with a light footprint
• Support for MongoDB change streams and aggregation pipelines
• Pre built Docker containers
• Optionally filter the set of collections to sync
• Advanced support for sharded MongoDB clusters including auto-detection of new shards
• Direct read mode to do a full sync of collections in addition to tailing the oplog
• Transform and filter documents before indexing using Golang plugins or JavaScript
• Index the content of GridFS files
• Support for propogating hard/soft document deletes
• Support for propogating database and collection drops as index deletes
• Optional custom document routing in Elasticsearch
• Stateful resume feature
• Time machine feature to track document changes over time
• Worker and Clustering modes for High Availability
• Support for rfc7396 JSON merge patches
• Systemd support
• Optional http server to get access to liveness, stats, profiling, etc

뭐 홈페이지에도 맨 처음 화면에 나와있는 것들이긴 한데, 로그스태쉬보다 간단한 설정으로 DB간의 동기화를 구성할 수 있습니다.

설정 방법

우선 Monstache를 이용하기 위해서는 서버에 Go(Go Lang)를 설치 해야합니다.

최신 버전 Go 받기

$ wget https://golang.org/dl/go1.16.2.linux-amd64.tar.gz

Go 설치

$ tar -C /usr/local -xzf go1.16.2.linux-amd64.tar.gz
$ export PATH=$PATH:/usr/local/go/bin
$ go version
go version go1.16.2 linux/amd64

Monstache 설치

$ wget https://github.com/rwynn/monstache/releases/download/v6.7.4/monstache-98f8bc6.zip
$ unzip monstache-98f8bc6.zip

파일을 받아 압축을 푼 후에 경로 안에 들어가면 여러 OS의 폴더가 나눠져있고, 그 안에 monstache 파일이 있습니다.

저는 /usr/local/bin 밑으로 복사했습니다.

$ cp monstache /usr/local/bin/

Monstache 설정파일 생성 및 설정

설정 파일은 아무데나 많들어도 되지만, 관리를 위해 Monstache 폴더를 만들겠습니다.

$ cd ~
$ mkdir monstache
$ cd monstache

설정 파일을 생성합니다.

vi mongo-elastic.toml

# connection settings

# connect to MongoDB using the following URL
mongo-url = "mongodb://uuusAdmin:dbthfahdrh12#$@10.100.170.45:27017"
# connect to the Elasticsearch REST API at the following node URLs
elasticsearch-urls = ["http://10.100.170.61:9200"]

# frequently required settings

# if you need to seed an index from a collection and not just listen and sync changes events
# you can copy entire collections or views from MongoDB to Elasticsearch
direct-read-namespaces = ["db.hashTags", "db.keywords"]

# if you want to use MongoDB change streams instead of legacy oplog tailing use change-stream-namespaces
# change streams require at least MongoDB API 3.6+
# if you have MongoDB 4+ you can listen for changes to an entire database or entire deployment
# in this case you usually don't need regexes in your config to filter collections unless you target the deployment.
# to listen to an entire db use only the database name.  For a deployment use an empty string.
change-stream-namespaces = ["db.hashTags", "db.keywords"]

# additional settings

# compress requests to Elasticsearch
gzip = true

# generate indexing statistics
stats = true

# index statistics into Elasticsearch
index-stats = true

# use 4 go routines concurrently pushing documents to Elasticsearch
elasticsearch-max-conns = 4

# propogate dropped collections in MongoDB as index deletes in Elasticsearch
dropped-collections = false

# propogate dropped databases in MongoDB as index deletes in Elasticsearch
dropped-databases = false

# in Elasticsearch with a newer version. Elasticsearch is preventing the old docs from overwriting new ones.
replay = false

# resume processing from a timestamp saved in a previous run
resume = false

# do not validate that progress timestamps have been saved
resume-write-unsafe = true

# override the name under which resume state is saved
resume-name = "default"

# use a custom resume strategy (tokens) instead of the default strategy (timestamps)
# tokens work with MongoDB API 3.6+ while timestamps work only with MongoDB API 4.0+
resume-strategy = 1

# print detailed information including request traces
verbose = true

index-as-update = false

index-oplog-time = false

index-files = false

file-highlighting = false

일단 제가 설정한 설정파일의 내용이며

direct-read-namespaces = [“db.hashTags”, “db.keywords”]

이부분이 가장 중요한데, namespace는 MongoDB의 네임스테이스 규칙, 즉 dbName.collectionName 의 규칙으로 생성을 해줘야합니다.

네임스페이스를 기준으로 동기화 대상을 정하는 것이기 때문에 컬렉션 명만 줘서는 동기화가 되질 않습니다.

Monstache의 설정 파일의 파라미터들은 아래 페이지에서 확인할 수 있습니다.

이 정도만 설정해도 동기화 하는데 문제는 없습니다.

index-as-update 파라미터는 기본값이 false 인데 true로 바꾸면 MongoDB에서 컬렉션에 업데이트가 일어난다면 엘라스틱 서치에서도 새로운 데이터를 쌓는 구조가 아닌 해당 데이터에 업데이트로 처리하게 하는 파라미터입니다.

엘라스틱 서치에서는 DB명을 index라고 부르니 차이점을 인지하시고 사용하시면 됩니다.

Monstache 구동

foreground에서 돌리면 터미널에서 빠져나오면 구동이 멈추기 때문에 Background에서 돌게 작업합니다.

$ nohup monstache -f mongo-elastic.toml 2>&1 &

프로세스는 몇개든 띄울수 있으며, 업데이트가 필요한 컬렉션은 index-as-update를 true, 필요없는 로그성 데이터는 false 로 설정한 설정파일을 따로 만들어서 각각 따로 동기화 하는 것도 가능합니다.

이렇게 MongoDB와 ElasticSearch를 이용한 실시간 동기화하는 법을 알아봤습니다.

Monstache는 Docker 이미지도 제공하고 있으며, 실제 데이터 동기화 작업이 매우 빠르다는 장점이 있습니다.