Gunakan MongoShake untuk sinkronisasi satu arah antar instans MongoDB - ApsaraDB for MongoDB

MongoShake adalah alat open-source Platform as a Service (PaaS) yang dikembangkan oleh Alibaba Cloud menggunakan Go. Alat ini membaca oplog dari instans MongoDB sumber dan memainkannya ulang di tujuan, sehingga memungkinkan sinkronisasi satu arah berkelanjutan untuk analitik data, pemulihan bencana, dan skenario aktif-aktif.

Selain replikasi langsung, MongoShake mendukung langganan dan konsumsi log melalui SDK, Kafka, dan MetaQ — menjadikannya cocok untuk langganan log, sinkronisasi pusat data, dan penghapusan cache secara asinkron.

Untuk informasi lebih lanjut, lihat repositori GitHub MongoShake.

Topologi yang didukung

Source	Destination
Self-managed MongoDB pada Instance ECS	Self-managed MongoDB pada Instance ECS
Self-managed MongoDB on-premises	Self-managed MongoDB on-premises
ApsaraDB for MongoDB instance	ApsaraDB for MongoDB instance
Third-party cloud MongoDB	Third-party cloud MongoDB

Catatan penggunaan

Jangan lakukan operasi Data Definition Language (DDL) pada database sumber sebelum sinkronisasi penuh selesai. Ketidakkonsistenan data dapat terjadi.
Database local tidak dapat disinkronkan. Database admin dapat disinkronkan — lihat Migrasi data bisnis dari database admin.
Password dalam URI string koneksi tidak boleh mengandung @. Jika password Anda mengandung @, koneksi akan gagal.
Sinkronisasi DDL tidak didukung ketika sumbernya adalah instansi kluster sharded.

Izin yang diperlukan

Role	Required permissions
Instans MongoDB sumber	`readAnyDatabase`, `read` pada database local, dan `readWrite` pada database mongoshake
Instans MongoDB tujuan	`readWriteAnyDatabase`, atau `readWrite` pada database tujuan

MongoShake secara otomatis membuat database mongoshake pada instans sumber saat sinkronisasi inkremental dimulai.

Untuk mengelola pengguna database MongoDB, lihat Gunakan DMS untuk mengelola pengguna database MongoDB atau perintah db.createUser.

Prasyarat

Sebelum memulai, pastikan Anda telah:

Memiliki instans ApsaraDB for MongoDB set replika sumber yang menggunakan virtual private cloud (VPC). Jika menggunakan jaringan klasik, ubah ke VPC terlebih dahulu.
Memiliki instans ApsaraDB for MongoDB set replika tujuan dalam VPC yang sama dengan sumber, untuk meminimalkan latensi jaringan. Lihat Buat instans set replika.
Memiliki Instance ECS dalam VPC yang sama dengan instans MongoDB sumber, untuk menjalankan MongoShake. Lihat Buat Instance ECS.
Alamat IP pribadi Instance ECS telah ditambahkan ke daftar putih kedua instans MongoDB (sumber dan tujuan). Lihat Ubah daftar putih.

Jika jaringan Anda tidak memenuhi persyaratan ini, ajukan titik akhir publik pada kedua instans MongoDB, lalu tambahkan alamat IP publik Instance ECS ke daftar putih keduanya. Lihat Ajukan titik akhir publik dan Ubah daftar putih.

Siapkan sinkronisasi satu arah

Prosedur ini menginstal MongoShake di direktori /test/mongoshake.

Unduh MongoShake dan ubah nama arsipnya:

Tautan ini mengunduh MongoShake 2.8.7. Untuk versi terbaru, lihat halaman rilis.

wget "https://github.com/alibaba/MongoShake/releases/download/release-v2.8.7-20251218/mongo-shake-v2.8.7.tgz" -O mongoshake.tar.gz

Ekstrak arsip dan pindahkan ke direktori instalasi:

tar zxvf mongoshake.tar.gz && mv mongo-shake-v2.8.7 /test/mongoshake && cd /test/mongoshake

Konfigurasikan string koneksi dan mode sinkronisasi. Buka collector.conf:

Parameter	Description	Example
`mongo_urls`	URI string koneksi instans MongoDB sumber. Gunakan titik akhir VPC untuk meminimalkan latensi. Password tidak boleh mengandung `@`. Untuk format URI, lihat Deskripsi koneksi untuk instans set replika.	`mongo_urls = mongodb://test:@dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717,dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717`
`tunnel.address`	URI string koneksi instans MongoDB tujuan. Gunakan titik akhir VPC untuk meminimalkan latensi. Password tidak boleh mengandung `@`. Untuk format URI, lihat Deskripsi koneksi untuk instans set replika.	`tunnel.address = mongodb://test:@dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717,dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717`
`sync_mode`	Cakupan sinkronisasi. `all` menjalankan sinkronisasi penuh lalu inkremental; `full` hanya menjalankan sinkronisasi penuh; `incr` hanya menjalankan sinkronisasi inkremental. Default: `incr`.	`sync_mode = all`

vi collector.conf

Setidaknya atur parameter berikut. Untuk semua parameter yang tersedia, lihat Referensi konfigurasi.

Jalankan tugas sinkronisasi:

./collector.linux -conf=collector.conf -verbose

Pantau output log. Saat sinkronisasi penuh selesai dan sinkronisasi inkremental dimulai, Anda akan melihat baris seperti berikut:
```
[09:38:57 CST 2019/06/20] [INFO] (mongoshake/collector.(*ReplicationCoordinator).Run:80) finish full sync, start incr sync with timestamp: fullBeginTs[1560994443], fullFinishTs[1560994737]
```
Sinyal utama adalah finish full sync, start incr sync. Setelah baris ini muncul, jangan lakukan operasi DDL pada sumber — lihat Catatan penggunaan.

Monitor status sinkronisasi

Setelah sinkronisasi inkremental dimulai, buka terminal baru dan jalankan:

cd /test/mongoshake && ./mongoshake-stat --port=9100

mongoshake-stat adalah skrip Python. Instal Python 2.7 sebelum menjalankannya. Lihat situs resmi Python.

Contoh output:

Monitoring output

Parameter	Description
`logs_get/sec`	Jumlah oplog yang diambil per detik
`logs_repl/sec`	Jumlah operasi replay oplog per detik
`logs_success/sec`	Jumlah operasi replay oplog yang berhasil per detik
`lsn.time`	Timestamp oplog terakhir yang dikirim
`lsn_ack.time`	Timestamp saat tujuan mengonfirmasi penulisan
`lsn_ckpt.time`	Timestamp saat checkpoint dipersist
`now.time`	Waktu saat ini
`replset`	Nama set replika sumber

Migrasi data bisnis dari database admin

MongoDB tidak merekomendasikan menyimpan data bisnis di database admin, karena perilaku locking dan konflik dengan perintah internal dapat menurunkan performa.

MongoShake dapat menyinkronkan data dari database admin ke database non-admin. Ikuti prosedur di atas dan tambahkan konfigurasi berikut ke collector.conf pada langkah 4:

filter.pass.special.db = admin

# Migrasi semua koleksi dari database admin ke newDB
transform.namespace = admin:newDB

# Atau, migrasi koleksi tertentu
# (Anda dapat mengonfigurasi beberapa aturan)
transform.namespace = admin.abc:target.def

Referensi konfigurasi

Tabel 1. Parameter collector.conf

Category	Parameter	Description	Example
—	`conf.version`	Nomor versi file konfigurasi. Jangan ubah.	`conf.version = 4`
Global	`id`	ID tugas sinkronisasi. Digunakan untuk nama log, nama database checkpoint, dan nama database tujuan.	`id = mongoshake`
	`master_quorum`	Opsi high availability. Atur ke `true` pada node MongoShake primary ketika dua node melakukan sinkronisasi dari sumber yang sama. Default: `false`.	`master_quorum = false`
	`full_sync.http_port`	Port HTTP untuk melihat status sinkronisasi penuh. Default: `9101`.	`full_sync.http_port = 9101`
	`incr_sync.http_port`	Port HTTP untuk melihat status sinkronisasi inkremental. Default: `9100`.	`incr_sync.http_port = 9100`
	`system_profile_port`	Port profiling untuk melihat informasi stack internal.	`system_profile_port = 9200`
	`log.level`	Tingkat log. Nilai yang valid: `error`, `warning`, `info`, `debug`. Default: `info`.	`log.level = info`
	`log.dir`	Direktori untuk file log dan PID. Harus berupa jalur mutlak. Default ke direktori `logs` di path saat ini.	`log.dir = ./logs/`
	`log.file`	Nama file log. Default: `collector.log`.	`log.file = collector.log`
	`log.flush`	Laju refresh log. `true` mencetak setiap entri (mempengaruhi performa); `false` memastikan performa tetapi mungkin melewatkan beberapa entri. Default: `false`.	`log.flush = false`
	`sync_mode`	Cakupan sinkronisasi. `all`: penuh lalu inkremental; `full`: hanya penuh; `incr`: hanya inkremental. Default: `incr`.	`sync_mode = all`
	`mongo_urls`	URI string koneksi instans MongoDB sumber. Gunakan titik akhir VPC untuk meminimalkan latensi. Untuk format URI, lihat Deskripsi koneksi untuk instans set replika atau Deskripsi koneksi untuk instans kluster sharded.	`mongo_urls = mongodb://test:@dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717,dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717`
	`mongo_cs_url`	Titik akhir Configserver, diperlukan ketika sumbernya adalah instansi kluster sharded. Lihat Ajukan titik akhir untuk shard.	`mongo_cs_url = mongodb://test:@dds-bp19f409d7512-csxxx.mongodb.rds.aliyuncs.com:3717,dds-bp19f409d7512-csxxx.mongodb.rds.aliyuncs.com:3717/admin`
	`mongo_s_url`	Titik akhir Mongos, diperlukan ketika sumbernya adalah instansi kluster sharded. Pisahkan beberapa alamat dengan koma. Lihat Ajukan titik akhir untuk shard.	`mongos_s_url = mongodb://test:@s-bp19f409d7512.mongodb.rds.aliyuncs.com:3717,s-bp19f409d7512.mongodb.rds.aliyuncs.com:3717/admin`
	`tunnel`	Jenis saluran untuk sinkronisasi. `direct`: menulis langsung ke MongoDB tujuan; `rpc`: menggunakan NET/RPC; `tcp`: menggunakan TCP; `file`: menggunakan transfer file; `kafka`: menggunakan Kafka; `mock`: hanya untuk pengujian, tidak ada data yang ditulis. Default: `direct`.	`tunnel = direct`
	`tunnel.address`	Titik akhir tujuan. Format tergantung pada nilai `tunnel`: `direct` → URI string koneksi MongoDB; `rpc` → alamat penerima RPC; `tcp` → alamat penerima TCP; `file` → jalur file; `kafka` → `topic@brokers1,brokers2`; `mock` → biarkan kosong.	`tunnel.address = mongodb://test:@dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717,dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717`
	`tunnel.message`	Format data dalam saluran. Hanya berlaku ketika `tunnel` bernilai `kafka` atau `file`. `raw`: mode agregasi (default); `json`: format JSON agar mudah dibaca; `bson`: format biner BSON. Default: `raw`.	`tunnel.message = raw`
	`mongo_connect_mode`	Mode koneksi. Hanya berlaku ketika `tunnel` bernilai `direct`. `primary`: tarik dari node primary; `secondaryPreferred`: tarik dari node secondary; `standalone`: tarik dari satu node yang ditentukan. Default: `secondaryPreferred`.	`mongo_connect_mode = secondaryPreferred`
	`filter.namespace.black`	Namespace yang dikecualikan dari sinkronisasi. Pisahkan beberapa nilai dengan titik koma (`;`). Namespace adalah kombinasi nama database dan nama koleksi atau indeks, misalnya `mongodbtest.customer`.	`filter.namespace.black = mongodbtest.customer;testdata.test123`
	`filter.namespace.white`	Namespace yang disertakan dalam sinkronisasi. Hanya namespace yang tercantum yang akan disinkronkan. Pisahkan beberapa nilai dengan titik koma (`;`).	`filter.namespace.white = mongodbtest.customer;test123`
	`filter.pass.special.db`	Database yang disertakan meskipun biasanya difilter (seperti admin, local, mongoshake, config, dan system.views). Pisahkan beberapa nilai dengan titik koma (`;`).	`filter.pass.special.db = admin;mongoshake`
	`filter.ddl_enable`	Aktifkan sinkronisasi DDL. Tidak didukung ketika sumbernya adalah instansi kluster sharded. Default: `false`.	`filter.ddl_enable = false`
	`checkpoint.storage.url`	Alamat penyimpanan checkpoint untuk transmisi yang dapat dilanjutkan. Jika tidak diatur, MongoShake menulis ke database mongoshake (sumber set replika) atau database admin pada node Configserver (sumber kluster sharded).	`checkpoint.storage.url = mongodb://test:@dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717,dds-bp19f409d7512.mongodb.rds.aliyuncs.com:3717`
	`checkpoint.storage.db`	Nama database untuk menyimpan checkpoint. Default: `mongoshake`.	`checkpoint.storage.db = mongoshake`
	`checkpoint.storage.collection`	Nama koleksi untuk menyimpan checkpoint. Ubah ini ketika dua node MongoShake melakukan sinkronisasi dari sumber yang sama, untuk mencegah konflik nama. Default: `ckpt_default`.	`checkpoint.storage.collection = ckpt_default`
	`checkpoint.start_position`	Posisi awal untuk transmisi yang dapat dilanjutkan. Diabaikan jika checkpoint sudah ada. Format: `YYYY-MM-DDTHH:MM:SSZ`. Default: `1970-01-01T00:00:00Z`.	`checkpoint.start_position = 1970-01-01T00:00:00Z`
	`transform.namespace`	Mengganti nama database atau koleksi sumber sebelum ditulis ke tujuan. Misalnya, `fromA.fromB:toC.toD` mengganti nama koleksi `fromB` di database `fromA` menjadi koleksi `toD` di database `toC`.	`transform.namespace = fromA.fromB:toC.toD`
Full sync	`full_sync.reader.collection_parallel`	Jumlah maksimum koleksi yang ditarik secara konkuren.	`full_sync.reader.collection_parallel = 6`
	`full_sync.reader.write_document_parallel`	Jumlah thread penulisan konkuren per koleksi.	`full_sync.reader.write_document_parallel = 8`
	`full_sync.reader.document_batch_size`	Jumlah dokumen yang diagregasi per batch penulisan.	`full_sync.reader.document_batch_size = 128`
	`full_sync.collection_exist_drop`	Tindakan ketika koleksi tujuan memiliki nama yang sama dengan koleksi sumber. `true`: hapus koleksi tujuan dan sinkronkan (backup data terlebih dahulu); `false`: laporkan error dan keluar.	`full_sync.collection_exist_drop = true`
	`full_sync.create_index`	Pembuatan indeks setelah sinkronisasi. `foreground`: indeks foreground; `background`: indeks background; `none`: lewati pembuatan indeks.	`full_sync.create_index = none`
	`full_sync.executor.insert_on_dup_update`	Konversi `INSERT` menjadi `UPDATE` ketika `_id` duplikat ada di tujuan. Default: `false`.	`full_sync.executor.insert_on_dup_update = false`
	`full_sync.executor.filter.orphan_document`	Filter dokumen yatim ketika sumbernya adalah instansi kluster sharded. Default: `false`.	`full_sync.executor.filter.orphan_document = false`
	`full_sync.executor.majority_enable`	Aktifkan penulisan mayoritas di tujuan. Default: `false`.	`full_sync.executor.majority_enable = false`
Incremental sync	`incr_sync.mongo_fetch_method`	Metode untuk menarik data inkremental. `oplog`: tarik oplog dari sumber; `change_stream`: tarik event perubahan (memerlukan MongoDB 4.0 atau lebih baru). Default: `oplog`.	`incr_sync.mongo_fetch_method = oplog`
	`incr_sync.oplog.gids`	Digunakan untuk mengatur replikasi dua arah untuk kluster cloud.	`incr_sync.oplog.gids = xxxxxxxxxxxx`
	`incr_sync.shard_key`	Metode konkurensi internal. Jangan ubah.	`incr_sync.shard_key = collection`
	`incr_sync.worker`	Jumlah thread konkuren untuk mentransmisikan oplog. Jika sumbernya adalah instansi kluster sharded, nilai ini harus sama dengan jumlah shard.	`incr_sync.worker = 8`
	`incr_sync.worker.oplog_compressor`	Kompresi data untuk mengurangi konsumsi bandwidth. `none`: tanpa kompresi; `gzip`, `zlib`, `deflate`: kompres dalam format masing-masing. Hanya berlaku ketika `tunnel` bukan `direct`. Ketika `tunnel` adalah `direct`, atur ini ke `none`. Default: `none`.	`incr_sync.worker.oplog_compressor = none`
	`incr_sync.target_delay`	Penundaan dalam detik sebelum perubahan diterapkan ke tujuan. Atur ke nilai bukan nol untuk mencegah operasi tidak disengaja — misalnya, `1800` memberikan penundaan 30 menit. `0` menonaktifkan sinkronisasi tertunda.	`incr_sync.target_delay = 1800`
	`incr_sync.worker.batch_queue_size`	Konfigurasi antrian internal. Jangan ubah kecuali diperlukan.	`incr_sync.worker.batch_queue_size = 64`
	`incr_sync.adaptive.batching_max_size`		`incr_sync.adaptive.batching_max_size = 1024`
	`incr_sync.fetcher.buffer_capacity`		`incr_sync.fetcher.buffer_capacity = 256`
MongoDB (`direct` mode only)	`incr_sync.executor.upsert`	Konversi `UPDATE` menjadi `INSERT` ketika `_id` atau indeks unik tidak ada. Default: `false`.	`incr_sync.executor.upsert = false`
	`incr_sync.executor.insert_on_dup_update`	Konversi `INSERT` menjadi `UPDATE` ketika `_id` atau indeks unik tidak ada. Default: `false`.	`incr_sync.executor.insert_on_dup_update = false`
	`incr_sync.conflict_write_to`	Catat dokumen yang konflik ketika terjadi konflik penulisan. `none`: jangan catat; `db`: tulis log konflik ke mongoshake_conflict; `sdk`: tulis log konflik ke SDK. Default: `none`.	`incr_sync.conflict_write_to = none`
	`incr_sync.executor.majority_enable`	Aktifkan penulisan mayoritas di tujuan. Mengaktifkan ini memengaruhi performa. Default: `false`.	`incr_sync.executor.majority_enable = false`

FAQ

Periksa terlebih dahulu FAQ MongoShake di GitHub. Untuk masalah yang tidak tercakup di sana, laporkan di GitHub Issues.