A Go-based connector that replicates data from MongoDB to RawTree using Change Streams for CDC (Change Data Capture).

GitHub Events Analytics via MongoDB CDC

How it works

The MongoDB CDC connector uses two phases: an initial snapshot and continuous change data capture (CDC). During the snapshot, the connector records the current MongoDB cluster time. It performs a full Find() scan of each collection. The connector sends existing documents to RawTree as synthetic insert events. It stores the captured cluster time as the starting point for CDC.

After the snapshot, the connector opens a MongoDB Change Stream from the saved cluster time. It receives insert, update, replace, and delete events in real time. The connector batches events and sends them to RawTree through the service HTTP API. It periodically saves resume tokens to disk for fault tolerance. After a crash or restart, the connector resumes from the last confirmed checkpoint. If the MongoDB oplog has passed the saved token, the connector automatically starts a new snapshot.

Each MongoDB collection maps directly to a corresponding RawTree table. For example, ecommerce.orders becomes mongo_orders, while ecommerce.users maps to mongo_users.

View in github

Features

Full snapshot: Initial replication of existing collections
CDC streaming: Real-time change capture via MongoDB Change Streams
Multi-collection: Watch multiple collections with regex filtering
At-least-once delivery: Checkpoint-based recovery with resume tokens
Auto table creation: RawTree tables are created automatically
Backpressure: Configurable batch sizes and flush intervals
Observability: Prometheus metrics, health/readiness endpoints

The connector requires MongoDB 4.0+ replica set or sharded cluster (Change Streams require a replica set) and Go 1.23+ for building from source.

Quick Start

Clone the repository and install dependencies (Go 1.23+, mongoimport, mongosh, and direnv).

git clone https://github.com/rawtreedb/rawtree-mongo-connector.git
cd your-project-name

Docker Compose

# Edit docker-compose.yaml with your RawTree API key
docker compose up

Binary

make build
./bin/rawtree-mongo-connector --config config.example.yaml

Docker

make docker
docker run -v ./config.yaml:/etc/connector/config.yaml rawtree-mongo-connector:dev

Helm (Kubernetes)

helm install mongo-connector deploy/helm/rawtree-mongo-connector \
  --set config.mongodb.database=mydb \
  --set config.rawtree.endpoint=https://api.rawtree.com \
  --set secrets.mongodbUri="mongodb+srv://..." \
  --set secrets.rawtreeApiKey="rt_..."

Configuration

See config.example.yaml for all options. Key settings:

Setting	Env Override	Description
`mongodb.uri`	`RAWTREE_MONGO_MONGODB_URI`	MongoDB connection string
`mongodb.database`	`RAWTREE_MONGO_MONGODB_DATABASE`	Database to watch
`rawtree.endpoint`	`RAWTREE_MONGO_RAWTREE_ENDPOINT`	Service API URL
`rawtree.api_key`	`RAWTREE_MONGO_RAWTREE_API_KEY`	Service API key
`rawtree.table_prefix`	`RAWTREE_MONGO_RAWTREE_TABLE_PREFIX`	Table name prefix (default: `mongo_`)

Collection Filtering

mongodb:
  collections:
    # Explicit list
    include: ["orders", "users", "products"]
    # OR regex-based
    include_regex: "^(orders|users)"
    exclude_regex: "^system\\."

Event Format

Each row sent to RawTree includes CDC metadata:

{
  "_rt_op": "insert",
  "_rt_ts": "2024-01-15T10:30:00Z",
  "_rt_ns": "ecommerce.orders",
  "_rt_id": "507f1f77bcf86cd799439011",
  "_rt_doc": { ... }
}

Metrics

Prometheus metrics on :9090/metrics:

Metric	Description
`connector_events_total`	Events processed (by collection, op)
`connector_events_sent_total`	Events sent to RawTree (by table)
`connector_batch_latency_seconds`	HTTP POST latency
`connector_lag_seconds`	CDC lag (by collection)
`connector_snapshot_progress`	Snapshot rows scanned
`connector_errors_total`	Errors (by type)

Health Endpoints

GET :9090/healthz — always 200 once started
GET :9090/readyz — 200 when streaming, 503 during init

MongoDB CDC Connector