MongoDB to ClickHouse Migration Service

Migrates Countly drill_events* collections from MongoDB into a single ClickHouse table. Supports multi-pod horizontal scaling, range-parallel processing, async writes, pause/resume, crash recovery, backpressure monitoring, and a real-time dashboard.

Quick Start

cp .env.example .env   # edit with your connection details
docker compose up --build
curl http://localhost:8080/healthz

Open the dashboard at http://localhost:8080/viz.

Running

Docker Compose (recommended):

docker compose up --build

From source (Node 25+):

npm install
node --experimental-strip-types --expose-gc --max-old-space-size=2048 src/main.ts

Required env vars: SERVICE_NAME, MONGO_URI, CLICKHOUSE_URL, REDIS_URL.

Configuration

Copy .env.example and adjust. All values below show defaults where applicable.

Service

Variable	Default	Description
`SERVICE_NAME`	(required)	Service identifier
`SERVICE_PORT`	`8080`	HTTP server port
`SERVICE_HOST`	`0.0.0.0`	HTTP server bind address
`GRACEFUL_SHUTDOWN_TIMEOUT_MS`	`60000`	Max wait for graceful shutdown (ms)
`RERUN_MODE`	`resume`	`resume`, `new-run`, or `clone-run`
`LOG_LEVEL`	`info`	`fatal`, `error`, `warn`, `info`, `debug`, `trace`

MongoDB Source

Variable	Default	Description
`MONGO_URI`	(required)	MongoDB connection string
`MONGO_DB`	`countly_drill`	Source database
`MONGO_COUNTLY_DB`	`countly`	Countly database (for hash resolution)
`MONGO_COLLECTION_PREFIX`	`drill_events`	Prefix to discover collections
`MONGO_READ_PREFERENCE`	`primary`	Read preference
`MONGO_READ_CONCERN`	`majority`	Read concern level
`MONGO_RETRY_READS`	`true`	Enable retry reads
`MONGO_APP_NAME`	(optional)	Connection app name
`MONGO_BATCH_ROWS_TARGET`	`10000`	Target docs per ClickHouse write batch
`MONGO_PAGE_SIZE`	`10000`	MongoDB page size per cursor read
`MONGO_CURSOR_BATCH_SIZE`	`10000`	MongoDB cursor batch size
`MONGO_MAX_TIME_MS`	`600000`	Cursor timeout (ms)

Range-Parallel Processing

Variable	Default	Description
`RANGE_PARALLEL_THRESHOLD`	`500000`	Estimated doc count to trigger range splitting
`RANGE_COUNT`	`100`	Number of time-ranges to split a collection into
`RANGE_LEASE_TTL_SEC`	`300`	Range lease TTL for dead-pod reclaim (s)

Transform

Variable	Default	Description
`TRANSFORM_VERSION`	`v1`	Data transform version tag

ClickHouse Target

Variable	Default	Description
`CLICKHOUSE_URL`	(required)	ClickHouse HTTP endpoint
`CLICKHOUSE_DB`	`countly_drill`	Target database
`CLICKHOUSE_TABLE`	`drill_events`	Target table
`CLICKHOUSE_USERNAME`	`default`	Username
`CLICKHOUSE_PASSWORD`	(empty)	Password
`CLICKHOUSE_QUERY_TIMEOUT_MS`	`120000`	Query timeout (ms)
`CLICKHOUSE_MAX_RETRIES`	`8`	Max insert retry attempts
`CLICKHOUSE_RETRY_BASE_DELAY_MS`	`1000`	Backoff base delay (ms)
`CLICKHOUSE_RETRY_MAX_DELAY_MS`	`30000`	Backoff max delay (ms)
`CLICKHOUSE_USE_DEDUP_TOKEN`	`true`	Enable insert dedup tokens

Backpressure

Variable	Default	Description
`BACKPRESSURE_ENABLED`	`true`	Enable ClickHouse parts monitoring
`BACKPRESSURE_PARTS_TO_THROW_INSERT`	`300`	Parts threshold to pause inserts
`BACKPRESSURE_MAX_PARTS_IN_TOTAL`	`500`	Max total parts allowed
`BACKPRESSURE_PARTITION_PCT_HIGH`	`0.70`	Partition high watermark
`BACKPRESSURE_PARTITION_PCT_LOW`	`0.55`	Partition low watermark
`BACKPRESSURE_TOTAL_PCT_HIGH`	`0.70`	Total high watermark
`BACKPRESSURE_TOTAL_PCT_LOW`	`0.55`	Total low watermark
`BACKPRESSURE_POLL_INTERVAL_MS`	`5000`	Pressure polling interval (ms)
`BACKPRESSURE_MAX_PAUSE_EPISODE_MS`	`180000`	Max pause before force resume (ms)

State

Variable	Default	Description
`MANIFEST_DB`	`countly_drill`	MongoDB database for run manifests
`REDIS_URL`	(required)	Redis connection URL
`REDIS_KEY_PREFIX`	`mig`	Redis key namespace
`TIMELINE_SNAPSHOT_INTERVAL`	`10`	Timeline snapshot every N batches

Memory / GC

Variable	Default	Description
`GC_ENABLED`	`true`	Enable manual GC
`GC_RSS_SOFT_LIMIT_MB`	`3072`	RSS soft limit to trigger GC
`GC_RSS_HARD_LIMIT_MB`	`6144`	RSS hard limit warning
`GC_HEAP_USED_RATIO`	`0.70`	Heap usage ratio trigger
`GC_EVERY_N_BATCHES`	`50`	Run GC every N batches

Worker / Multi-Pod

Variable	Default	Description
`POD_ID`	`hostname()`	Unique pod identifier
`MULTI_POD_ENABLED`	`true`	Enable distributed locking
`LOCK_TTL_SECONDS`	`300`	Collection lock TTL — crash safety net (s)
`LOCK_RENEW_MS`	`60000`	Lock renewal interval (ms)
`PROGRESS_UPDATE_MS`	`5000`	Progress reporting interval to Redis (ms)
`POD_HEARTBEAT_MS`	`30000`	Pod heartbeat interval (ms)
`POD_DEAD_AFTER_SEC`	`180`	Pod considered dead after this silence (s)

Async Write

Variable	Default	Description
`ASYNC_WRITE_FLUSH_INTERVAL_MS`	`5000`	Flush queued batch records to MongoDB every N ms
`ASYNC_WRITE_FLUSH_BATCH_SIZE`	`10`	Flush after N batch records queued

Multi-Collection Migration

The service automatically discovers all MongoDB collections matching MONGO_COLLECTION_PREFIX* (e.g. drill_events, drill_events5a2b3c4d...).

Collections are processed in priority order (base collection first, then alphabetical)
Each collection gets its own run ID and isolated Redis key prefix
Already-completed collections are skipped on restart
Missing {cd: 1, _id: 1} compound index is created automatically in background
If a collection fails, the service logs the error and continues to the next

Multi-Pod Mode

Multiple pods can process collections in parallel using Redis-based distributed locking. Each pod autonomously discovers collections, acquires locks, and processes work.

How It Works

Pod heartbeat: Each pod writes a TTL-based heartbeat key (mig:pod:{podId}) to Redis every 30s
Collection locking: Before processing a collection, a pod acquires a lock via atomic Lua script. If the lock is held by another live pod, it moves to the next collection
Dead pod detection: If a pod's heartbeat expires (180s default), other pods can steal its locks
Global progress: Each pod reports per-collection progress to Redis. The /stats endpoint and dashboard merge data from all pods

Range-Parallel Processing

Collections exceeding RANGE_PARALLEL_THRESHOLD (default 500K docs) are split into RANGE_COUNT equal time-ranges based on [min(cd), max(cd)]. Multiple pods can process different ranges of the same collection concurrently.

Range initialization: First pod to reach a large collection divides the time span into N ranges via Redis SETNX
Atomic claiming: Pods claim ranges via a Redis Lua script that atomically marks a pending range as processing
Exclusive boundaries: Ranges use [start, end) (exclusive upper bound) to prevent duplicate processing. The final range uses [start, max]
Stale reclaim: If a pod dies mid-range, other pods reclaim its ranges after the lease TTL expires
Batch sequence isolation: Each range gets 10,000 batch sequence slots to prevent collisions

Kubernetes Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: migration
spec:
  replicas: 3
  selector:
    matchLabels:
      app: migration
  template:
    metadata:
      labels:
        app: migration
    spec:
      terminationGracePeriodSeconds: 120
      containers:
        - name: migration
          image: countly-migration:latest
          ports:
            - containerPort: 8080
          env:
            - name: POD_ID
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name
          livenessProbe:
            httpGet:
              path: /healthz
              port: 8080
            initialDelaySeconds: 10
          readinessProbe:
            httpGet:
              path: /readyz
              port: 8080
            initialDelaySeconds: 15
          lifecycle:
            preStop:
              httpGet:
                path: /control/drain
                port: 8080
---
apiVersion: v1
kind: Service
metadata:
  name: migration
spec:
  selector:
    app: migration
  ports:
    - port: 8080
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: migration
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: migration
  minReplicas: 1
  maxReplicas: 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70

K8s lifecycle:

Scale up: New pods start, discover collections, acquire locks on unclaimed work
Scale down: preStop hook calls /control/drain, pod finishes current batch, releases locks, exits
Pod crash: SIGTERM triggers graceful shutdown. On kill -9, heartbeat expires after 180s, locks are stolen by live pods

Dashboard

The /viz endpoint serves a self-contained real-time HTML dashboard. No external dependencies.

Features:

Overall migration progress bar with ETA
Current collection progress and batch sequence
Per-pod progress bars with docs/rows/throughput stats
Live batches panel showing batch phase (READING, TRANSFORMING, WRITING, COMMITTING)
Range heatmap for range-parallel collections
Active locks table with release buttons
Stale pod detection with remove action
Global cluster controls (pause/resume/stop all pods)
Index build status
Skip reason breakdown
Memory and GC metrics
Infrastructure connection status

API Endpoints

Health

Method	Path	Description
`GET`	`/healthz`	Liveness probe (always 200)
`GET`	`/readyz`	Readiness check (mongo, clickhouse, redis, manifest, runner)

Stats

Method	Path	Description
`GET`	`/stats`	Full dashboard JSON: progress, throughput, integrity, cluster, live batches
`GET`	`/viz`	Real-time HTML dashboard

Control (Single Pod)

Method	Path	Description
`POST`	`/control/pause`	Pause after current batch
`POST`	`/control/resume`	Resume from pause
`POST`	`/control/stop-after-batch`	Stop cleanly after current batch
`POST`	`/control/gc`	Trigger manual GC (body: `{"mode":"now"}`)
`POST`	`/control/reindex/:collection`	Trigger index creation for a collection
`POST`	`/control/retry-collection/:collection`	Re-queue a failed/skipped collection
`POST`	`/control/retry-batch/:runId/:batchSeq`	Retry a specific failed/skipped batch
`POST`	`/control/retry-skipped-batches/:runId`	Retry all skipped_empty batches in a run

Control (Multi-Pod)

Method	Path	Description
`POST`	`/control/global/pause`	Pause all pods
`POST`	`/control/global/resume`	Resume all pods
`POST`	`/control/global/stop`	Stop all pods after current batch
`GET`	`/control/locks`	List all active collection locks
`POST`	`/control/locks/release/:collection`	Force-release a lock (admin)
`GET`	`/control/pods`	List all pods with status and locks held
`POST`	`/control/pods/remove/:podId`	Remove dead pod and release its locks
`POST`	`/control/drain`	Graceful drain for K8s scale-down

Runs

Method	Path	Description
`GET`	`/runs`	List all runs (`?status=`, `?limit=`, `?offset=`)
`GET`	`/runs/current`	Current active run with coverage
`GET`	`/runs/current/batches`	Batches for current run (`?status=`, `?limit=`)
`GET`	`/runs/:id`	Run details by ID
`GET`	`/runs/:id/batches`	Batch list (`?status=`, `?limit=`)
`GET`	`/runs/:id/failures`	Failed batches, retry errors, digest mismatches
`GET`	`/runs/:id/timeline`	Performance snapshots over time
`GET`	`/runs/:id/coverage`	Document range coverage and completion %
`DELETE`	`/runs/:id/cache`	Clean up Redis cache for a run

Operations

Pause and Resume

# Single pod
curl -X POST http://localhost:8080/control/pause
curl -X POST http://localhost:8080/control/resume

# All pods (multi-pod mode)
curl -X POST http://localhost:8080/control/global/pause
curl -X POST http://localhost:8080/control/global/resume

Graceful Stop

curl -X POST http://localhost:8080/control/stop-after-batch

Or send SIGTERM / SIGINT. The service will:

Finish the current batch
Flush async write queue to MongoDB
Release all collection locks
Close all connections
Exit cleanly

Lock Management

# List all active locks
curl http://localhost:8080/control/locks

# Force-release a stuck lock
curl -X POST http://localhost:8080/control/locks/release/drill_events5a2b...

# List pods and their status
curl http://localhost:8080/control/pods

# Remove a dead pod and release its locks
curl -X POST http://localhost:8080/control/pods/remove/pod-3

Retry Failed Work

# Re-queue a failed collection
curl -X POST http://localhost:8080/control/retry-collection/drill_events5a2b...

# Retry a specific failed batch
curl -X POST http://localhost:8080/control/retry-batch/{runId}/{batchSeq}

# Retry all skipped batches in a run
curl -X POST http://localhost:8080/control/retry-skipped-batches/{runId}

Crash Recovery

On restart after a crash, the service automatically:

Checks Redis for the last committed cursor (hot-path authority)
Falls back to MongoDB manifest if Redis is empty
Replays any inflight batches with digest verification
Resumes from the last committed position
ClickHouse dedup tokens (mig:{runId}:{batchSeq}) prevent duplicate inserts

Monitor Progress

# JSON stats
curl -s http://localhost:8080/stats | jq '.summary'

# Cluster overview
curl -s http://localhost:8080/stats | jq '.cluster.pods'

# Or use the dashboard
open http://localhost:8080/viz

Architecture

MongoDB (drill_events*)
    |
    |  cursor pagination on (cd, _id) compound index
    v
+------------------------------------------------------+
|  CollectionOrchestrator                              |
|  |-- discovers collections, index-aware scheduling   |
|  |-- multi-pod: Redis lock per collection            |
|  |                                                    |
|  +-- Small collections (< 500K docs):               |
|  |   BatchRunner --> ClickHouse                      |
|  |                                                    |
|  +-- Large collections (>= 500K docs):               |
|      RangeCoordinator                                |
|      |-- splits [min(cd), max(cd)] into N ranges     |
|      |-- atomic range claiming via Redis Lua script   |
|      +-- per-range BatchRunner --> ClickHouse         |
+------------------------------------------------------+
    |                    |                    |
    |  batch inserts     |  hot-path commit   |  async flush
    |  (dedup tokens)    |                    |
    v                    v                    v
 ClickHouse          Redis                MongoDB manifest
 (drill_events)      (cursor, bitmap,     (batch records,
                      live phases,         run metadata,
                      range stats,         audit trail)
                      pod heartbeats,
                      collection locks)

Redis (hot-path authority): Committed cursor, batch completion bitmap, live batch phases, range stats, pod heartbeats, collection locks, global commands
MongoDB manifest (async, authoritative for audit): Batch records flushed asynchronously every 5s or 10 batches. Run metadata, skip samples, error history
Async batch writer: Redis is the commit point for each batch. MongoDB writes are queued and bulk-flushed in the background. On graceful shutdown, the queue is drained
Backpressure: Monitors ClickHouse active parts count; pauses inserts when thresholds exceeded (hysteresis with high/low watermarks)
Dedup tokens: Each batch carries mig:{runId}:{batchSeq} for idempotent retries

Redis Key Schema

Key Pattern	Type	TTL	Description
`mig:active_run`	STRING	-	Current active run ID
`mig:run:{runId}:state`	STRING	-	Run state JSON blob
`mig:run:{runId}:cursor`	STRING	-	Last committed cursor (hot-path authority)
`mig:run:{runId}:done_bitmap`	STRING	-	Batch completion bitmap (SETBIT/GETBIT)
`mig:run:{runId}:stats:latest`	STRING	-	Latest run stats snapshot
`mig:run:{runId}:commands`	HASH	-	Operator command flags (pause, abort, etc.)
`mig:run:{runId}:recent_errors`	LIST	-	Last 100 batch errors
`mig:run:{runId}:timeline`	LIST	-	Last 1000 timeline snapshots
`mig:run:{runId}:batch:{seq}:errors`	LIST	-	Verbose errors per batch (last 20)
`mig:liveBatch:{collection}`	STRING	30s	Live batch phase data (heartbeat-refreshed)
`mig:rangeLive:{collection}:{idx}`	STRING	60s	Per-range live stats (heartbeat-refreshed)
`mig:lock:{collection}`	STRING	300s	Collection lock with pod owner
`mig:pod:{podId}`	STRING	180s	Pod heartbeat/liveness key
`mig:progress:{collection}`	STRING	300s	Per-collection progress for cluster aggregation
`mig:cmd:global`	HASH	-	Global commands (pause, stop)
`mig:ranges:{collection}`	HASH	-	Range entries (idx, status, podId, claimedAt)
`mig:ranges:{collection}:init`	STRING	60s	SETNX coordinator election for range init
`mig:ranges:{collection}:runId`	STRING	-	Shared run ID for range-parallel processing
`mig:ranges:{collection}:meta`	STRING	-	Range metadata (minCd, maxCd, count)
`mig:ranges:{collection}:finalized`	STRING	60s	SETNX for run finalization (one pod only)

Project Structure

src/
  main.ts                              # Entry point
  config/
    schema.ts                          # Zod config schema with defaults
    loader.ts                          # Env var loader and validator
  http/
    health-route.ts                    # /healthz, /readyz
    stats-route.ts                     # /stats (JSON dashboard)
    control-route.ts                   # /control/* (pause, resume, locks, pods)
    run-route.ts                       # /runs/* (history, batches, coverage)
    viz-route.ts                       # /viz (HTML dashboard)
  source/
    mongo-reader.ts                    # Cursor pagination on (cd, _id)
    discover-collections.ts            # Collection discovery
  target/
    clickhouse-writer.ts               # Batch inserts with dedup
    clickhouse-pressure.ts             # Backpressure monitoring
  transform/
    normalize.ts                       # Document normalization
    skip-reasons.ts                    # Skip tracking
    validators.ts                      # Data validation
    hash-resolver.ts                   # Collection hash defaults
  runtime/
    collection-orchestrator.ts         # Multi-collection + multi-pod orchestration
    batch-runner.ts                    # Core batch lifecycle (read, transform, write)
    range-coordinator.ts               # Range-parallel processing
    retry-policy.ts                    # Exponential backoff
    gc-controller.ts                   # Manual GC management
    process-metrics.ts                 # Memory/CPU metrics
    resolve-run.ts                     # Run resolution on startup
  state/
    manifest-store.ts                  # MongoDB manifest (authoritative audit trail)
    redis-hot-state.ts                 # Redis hot state (bitmap, timeline, phases)
    async-batch-writer.ts              # Async MongoDB write queue
    global-progress.ts                 # Multi-pod progress aggregation
    collection-lock.ts                 # Distributed collection locking
    coverage.ts                        # Coverage interval tracking
  types/
    cursor.ts                          # Cursor serialization

CI/CD

A GitHub Actions workflow automatically builds and pushes the Docker image to Docker Hub when a GitHub Release is published.

Tag Behavior

Release Tag	Docker Tags	`latest`?
`v1.0.0`	`1.0.0`, `1.0`, `1`, `latest`	Yes
`v2.3.1`	`2.3.1`, `2.3`, `2`, `latest`	Yes
`v1.0.0-rc.1`	`1.0.0-rc.1`	No
`v2.0.0-beta`	`2.0.0-beta`	No

Setting Up Docker Hub Credentials

Go to Docker Hub > Account Settings > Security > New Access Token
Add DOCKERHUB_USERNAME and DOCKERHUB_TOKEN as GitHub repo secrets
Create a release (e.g. tag v1.0.0) and the workflow builds and pushes

Using a Pre-Built Image

services:
  migration:
    image: <your-username>/countly-migration:1.0.0
    # ... rest of config unchanged

FilesExpand file tree

README.md

Latest commit

History

README.md

File metadata and controls

MongoDB to ClickHouse Migration Service

Quick Start

Running

Configuration

Service

MongoDB Source

Range-Parallel Processing

Transform

ClickHouse Target

Backpressure

State

Memory / GC

Worker / Multi-Pod

Async Write

Multi-Collection Migration

Multi-Pod Mode

How It Works

Range-Parallel Processing

Kubernetes Deployment

Dashboard

API Endpoints

Health

Stats

Control (Single Pod)

Control (Multi-Pod)

Runs

Operations

Pause and Resume

Graceful Stop

Lock Management

Retry Failed Work

Crash Recovery

Monitor Progress

Architecture

Redis Key Schema

Project Structure

CI/CD

Tag Behavior

Setting Up Docker Hub Credentials

Using a Pre-Built Image