Skip to main content
The MongoDB source reads change streams and streams documents into the sink. The current connector maps each MongoDB document to one sink document (raw 1:1), which fits collections that are already denormalized.
VentStream writes to a target. OpenSearch and Elasticsearch are the supported targets today — they share the _bulk API, so the same engine handles both. Set the endpoint with VS_OS_ENDPOINT.

​
Source requirements

Change streams require a replica set or sharded cluster — they are not supported on a standalone mongod. Works on self-hosted MongoDB 4.0+ and Atlas (mongodb+srv://). Connect via the replica-set URI or a mongos. The engine’s user needs read access plus the change-stream / find privileges on the watched database (read on the database is enough on most deployments).

​
How it works

  • Trigger, then re-read. Each change event is only a trigger. With fullDocument: updateLookup (the default) the source carries the current whole document on every insert/update/replace, so the sink always reflects the latest state — the change payload never needs to be complete.
  • Bootstrap. On cold start the source captures a change-stream resume token, scans every in-scope collection by _id (paged), emits each document as an insert, then tails the change stream from the captured token. A row mutated during the scan is re-emitted by the tail and de-duplicated by the deterministic doc id at the (idempotent) sink.
  • Resume. The resume token is persisted to a file in VS_MONGO_STATE_DIR (a PVC in Kubernetes), flushed every VS_MONGO_TOKEN_FLUSH_MS. A restart resumes from there. If the token has aged out of the oplog, the source detects it, wipes the cursor, and re-bootstraps.
  • Deletes become tombstones — the deterministic doc id targets the exact document to remove.

​
The _id mapping

The sink document id is the canonical, namespaced form {database}.{collection}:["<_id>"] (e.g. shop.orders:["64f0…"]), matching the Postgres source’s shop.orders:["…"] shape. An ObjectId is rendered as its plain hex string; DateTime as RFC 3339; Decimal128 as its decimal string.
MongoDB’s _id is relocated to id in the document body. OpenSearch (and Elasticsearch) reserve _id as a metadata field and reject it inside the body, so the source moves it to id (the original value is preserved as the sink’s _id via the doc id). If your document already has an id field, the _id is dropped rather than clobbering it.

​
Run the agent

VS_MONGO_COLLECTIONS is optional — omit it to watch every collection in the database. The index template renders one index per collection (orders, customers, …).

​
TLS

Atlas enables TLS through its connection string and uses publicly trusted certificates. To enforce the same policy independently of URI options, set:
An explicit TLS mode overrides tls settings in the URI. Strict mode validates the server certificate and hostname.

​
Key environment variables

Full list in the engine env reference.

​
Index mapping for mixed-type numeric arrays

If a collection has an array that mixes integers and floats — e.g. [4.5, 3.2, 5] where 5 is stored as an int32 — OpenSearch dynamic mapping infers the field type from the first value and can then reject later documents whose values don’t fit. This is source-agnostic OpenSearch behavior, not a MongoDB-specific issue. The engine handles it gracefully — the offending document is routed to the DLQ with a precise reason and the pipeline keeps running — but it won’t be indexed until the mapping accepts it. Fix it with an explicit index template that maps such fields to double (or define the mapping up front).

​
Current limitations

  • Raw 1:1 only. One document → one sink document; there are no cross-collection joins yet, so a change to a referenced collection does not recompose a parent document.
  • Replica set / sharded cluster required — change streams don’t run on a standalone mongod.
  • Oplog-bounded resume. A resume reaches back only as far as the oplog retains. Size the oplog to cover worst-case agent downtime, or the agent re-bootstraps after a long outage.