Apache CouchDB CDC: Changes feed setup with Gluesync

Source data from Apache CouchDB

The Apache CouchDB source agent (agent tag couchdb) copies the databases you choose with a snapshot and then follows their _changes feed. It creates nothing in your databases and installs nothing on the server.

Databases, entities and fields

One entity is one CouchDB database. CouchDB has no level above the database, so the only schema is _default;
Discovery lists the databases with GET /_all_dbs, without the system databases (every name that starts with _, such as _users, _replicator and _global_changes). In CouchDB 3.x that endpoint is reserved to server administrators by default. When the server answers 401 or 403, or when you fill the Databases field of the connection, the list is the one from the connection and each name is checked with GET /<database>;
The fields of a database are inferred from a sample of its documents (Schema discovery sample size, default 1000): half of the sample comes from the start of _all_docs and half from its end. Design documents are left out of the sample. Names that start with _ are CouchDB members (_rev, _attachments, …) and are never fields. The key of the entity is _id, a STRING that is not nullable;
The type of a field comes from the JSON kinds of its values, ignoring null: a field that only holds integers (no fraction, no exponent, within 64 bits) is integer, any other number is number, and a field with one other kind is string, boolean, object or array. A field with more than one kind, or with nothing but null, is json. A field is nullable when a sampled document lacks it or holds null.

A field that appears only in documents outside the sample is not discovered. Raise Schema discovery sample size for a database with documents of different shapes.

How it works

Snapshot

The snapshot never pages _all_docs: that endpoint is bound to one stream on the server and does not scale with key ranges. The agent takes the ids from the changes feed instead, and reads the documents with many parallel _bulk_get requests:

producer (pages of ids) -> bounded channel of id chunks -> N workers (_bulk_get + mapping) -> bounded channel -> collector (emits)

The producer reads the ids from _changes in pages of Snapshot page size (default 10000), asking for ids only, which spares the server the sequence it would otherwise compute for every row. Chunks of Snapshot fetch batch size ids (default 1000) go to Snapshot read concurrency workers (default 8), each of which calls _bulk_get and maps the documents;
Core Hub hands over the position CDC will start from: the one read before the first id was listed and, when an interrupted snapshot is continued, the one of its first run. Changes made while the snapshot runs are replayed by CDC from that position: a document read with its newer body is written again, a document not read yet is inserted, and a document that was deleted is deleted again. Keep the insert conflict strategy of the entity on its default, Upsert. See Insert conflict strategy;
Deleted documents and design documents are not read. A document that is gone when its chunk is fetched is skipped;
Both channels hold a handful of chunks, so a slow target slows the workers and then the producer down, and a database is never held in memory. A snapshot needs about 3 x Snapshot read concurrency x Snapshot fetch batch size documents in memory: lower the batch size for very large documents. A message to Core Hub is cut by maxTransactionMessageKbSize and carries whole chunks;
The row count of an entity is the doc_count of the database minus its design documents. The snapshot ends at the end of the feed: a database that is written faster than it is read keeps the snapshot chasing the tail;
Resume. The workers finish their chunks in any order, so the agent saves only the contiguous prefix of pages whose chunks all reached a message that was emitted, with the position of the feed at the end of that prefix. A resumed snapshot restarts the id listing at that position and counts on from those rows. The rows read beyond the prefix are read again, which the upsert absorbs. A position that the feed no longer knows (HTTP 400) fails the entity with "run a new snapshot";
A snapshot ignores the logical partitions of an entity, because it already reads in parallel. Asking Core Hub to compute or count the partitions fails with a message that says so.

Measured on CouchDB 3.5.1 (single node, Docker on Apple silicon), 2,000 ids per _bulk_get request:

Way to read the documents Documents per second

Way to read the documents	Documents per second
`_all_docs?include_docs=true`, one stream or 4 and 16 key-range streams	28,000 (does not scale)
`_bulk_get`, 1 request at a time	22,000
`_bulk_get`, 8 requests at a time	81,000
`_bulk_get`, 16 requests at a time	94,000

_all_docs?include_docs=true, one stream or 4 and 16 key-range streams

28,000 (does not scale)

_bulk_get, 1 request at a time

22,000

_bulk_get, 8 requests at a time

81,000

_bulk_get, 16 requests at a time

94,000

Every read of the changes feed sets seq_interval, which spares the server a sequence per row. Listing ids only reaches 108,000 rows per second with it; reading documents from the feed with include_docs=true reaches 41,000 rows per second with it and 14,000 without it.

Through the source of the agent, reading 200,000 documents of about 1 KB from a database with 8 shards, a snapshot read concurrency of 1 gives 16,300 documents per second and the default of 8 gives 61,800. The reads in flight are the lever (Snapshot read concurrency), and the number of shards of the source database sets the ceiling. In a full pipeline, copying 10 million documents from CouchDB to PostgreSQL took 274 seconds end to end.

Change data capture

A poll of an entity is GET /<database>/_changes in longpoll mode, from the position of the last change read, with include_docs=true, style=main_only, and a limit and a seq_interval of Changes batch size rows (default 5000). Core Hub asks for the last position of the database on every tick and polls only when it differs from the one it holds. A long poll that finds nothing answers empty after Changes long poll timeout (ms) (default 1000).

The position of an entity is the opaque seq of its changes feed. In a cluster, a sequence encodes one position per shard. The agent orders positions by the number before the first dash (the sum of the sequences of the shards), the only part that is the same in every spelling CouchDB uses;
Insert, update or delete. A row of the feed marked deleted is a delete. Any other document is an insert when its revision generation is 1 (it was created and not modified since) and an update otherwise. The agent sends the whole document: CouchDB keeps no before image;
Deletes carry only the _id. A delete has every column of the entity, with a value only for the key, which is _id. A delete cannot name any other key: on an entity keyed on more than _id the delete is skipped, and the agent logs it. Keep _id as the key of the entity;
Design documents are skipped. Ids that start with _design/ are not replicated, in the snapshot and in CDC;
Delivery is at least once. A page is emitted as one or more messages cut by size. Every message but the last carries the position the page started from, and only the last carries the position at the end of the page, so a crash in between replays the page. The target applies a change twice without harm: an upsert and a delete that tolerates a missing document;
A backlog is read in one go. A poll keeps reading while the pages come back full, up to ten pages: only its first page waits for a change, so catching up is not limited to one page per polling interval;
Failures. A network failure, a timeout and the answers 408, 429 and 5xx are asked again before the entity fails. 404 fails the entity with the name of the database. 400 on the position fails it with "run a new snapshot".

Prerequisites

To have Gluesync working on your Apache CouchDB instance as a source you will need:

Apache CouchDB 3.3, 3.4 or 3.5, a single node or a cluster;
A user that is a member of every database to replicate, with read access to its documents. In CouchDB 3.x the database _security object lists the members; a database without members is open to every authenticated user. Reading the changes feed and the documents is all the agent does;
To list the databases, a server administrator. GET /_all_dbs is reserved to administrators by default in CouchDB 3.x. With a non-administrator user, put the names of the databases in the Databases field of the connection, separated by commas;
Network access from Core Hub to the HTTP port of the node (default 5984), or to the TLS port you use.

To create a user that can read a database, you can follow this quick guide, with an administrator account:

# create the user
curl -u admin:password -X PUT http://couchdb-host:5984/_users/org.couchdb.user:gluesync \
  -H 'Content-Type: application/json' \
  -d '{"name": "gluesync", "password": "your_password", "roles": [], "type": "user"}'

# make it a member of the database to replicate
curl -u admin:password -X PUT http://couchdb-host:5984/db_name/_security \
  -H 'Content-Type: application/json' \
  -d '{"admins": {"names": [], "roles": []}, "members": {"names": ["gluesync"], "roles": []}}'

Run the second request for every database to replicate. The _security request replaces the whole object: add the user to the members the database already has.

Setup via Web UI

Protocol: HTTP REST, the only one;
Node address: DNS record or IP address of one node of the server, without the scheme (couchdb.example.com, not http://couchdb.example.com);
Port: Optional, defaults to 5984;
Username: Username with the access listed in the prerequisites;
Password: Password belonging to the given username;
Disable authentication: Optional, defaults to false. Sends the requests with no authorization, for a server that allows anonymous access;
Enable TLS: Optional, defaults to false. When it is on, the node URLs use https:// and the server certificate is always validated. There is no option to trust it blindly;
CA certificate path and its password: Optional. A PKCS12 or JKS store, or a PEM file, that holds the certificate authority of the server. The certificate can also be uploaded through Core Hub. Blank means the default trust store of the JVM. A server with a certificate from a private CA needs this;
Additional nodes: Optional. host1:port1,host2:port2, the other nodes of a cluster, tried in order when a node is down. A node without a port takes the port above;
Databases (comma separated): Optional. The databases to offer when the user may not list them, for example orders,customers.

Custom properties

Path prefix: (optional, defaults to empty) The path the server answers under, when it sits behind a reverse proxy, for example /couchdb;
Connect timeout (ms): (optional, defaults to 10000) How long a node may take to accept a connection before the next one is tried;
Request timeout (ms): (optional, defaults to 300000) How long a request may run, long polls and big requests included.

Specific configuration

Setting Default Description

Setting	Default	Description
Source change retention in hours (`sourceChangeRetention`)	`24`	Common setting of the source agents, see Source change retention in hours. This agent reads the changes feed of each entity directly and keeps no cache of changes of its own.
Schema discovery sample size (`schemaDiscoverySampleSize`)	`1000`	Documents sampled to infer the fields of a database: half from the start of `_all_docs`, half from its end.
Snapshot read concurrency (`snapshotReadConcurrency`)	`8`	Parallel `_bulk_get` requests of a snapshot. The main lever of the snapshot speed.
Snapshot fetch batch size (`snapshotFetchBatchSize`)	`1000`	Document ids in one `_bulk_get` request. Lower it for very large documents.
Snapshot page size (`snapshotPageSize`)	`10000`	Rows of one `_changes` page that lists the ids of a snapshot.
Changes batch size (`changesBatchSize`)	`5000`	Rows of one `_changes` page of the change feed.
Changes long poll timeout (ms) (`changesLongPollTimeoutMilliseconds`)	`1000`	How long a poll of the change feed waits for a change before it answers empty.
Technical fields (`technicalFields`)	none	Adds `_TRANSACTION_ID` (the position the page ended at: the feed has no sequence of its own for most rows), `_TRANSACTION_TIMESTAMP` (when the change was read: the feed has no commit time) and `_TRANSACTION_OPERATION` (`I`, `U` or `D`) to the replicated rows. See Technical fields. A CouchDB target has to map them to names that do not start with `_`.

Source change retention in hours (sourceChangeRetention)

24

Common setting of the source agents, see Source change retention in hours. This agent reads the changes feed of each entity directly and keeps no cache of changes of its own.

Schema discovery sample size (schemaDiscoverySampleSize)

1000

Documents sampled to infer the fields of a database: half from the start of _all_docs, half from its end.

Snapshot read concurrency (snapshotReadConcurrency)

8

Parallel _bulk_get requests of a snapshot. The main lever of the snapshot speed.

Snapshot fetch batch size (snapshotFetchBatchSize)

1000

Document ids in one _bulk_get request. Lower it for very large documents.

Snapshot page size (snapshotPageSize)

10000

Rows of one _changes page that lists the ids of a snapshot.

Changes batch size (changesBatchSize)

5000

Rows of one _changes page of the change feed.

Changes long poll timeout (ms) (changesLongPollTimeoutMilliseconds)

1000

How long a poll of the change feed waits for a change before it answers empty.

Technical fields (technicalFields)

none

Adds _TRANSACTION_ID (the position the page ended at: the feed has no sequence of its own for most rows), _TRANSACTION_TIMESTAMP (when the change was read: the feed has no commit time) and _TRANSACTION_OPERATION (I, U or D) to the replicated rows. See Technical fields. A CouchDB target has to map them to names that do not start with _.

Setup via Rest APIs

Here following an example of calling the Core Hub’s Rest API via curl to setup the connection for this Agent.

Connect the agent

curl --location --request PUT 'http://core-hub-ip-address/pipelines/{pipelineId}/agents/{agentId}/config/credentials' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
    "agentType": "SOURCE",
    "agentTag": "couchdb",
    "hostCredentials": {
        "connectionName": "myAgentNickName",
        "host": "couchdb-host-address",
        "port": 5984,
        "username": "gluesync",
        "password": "your_password",
        "disableAuth": false,
        "enableTls": false,
        "additionalHosts": "couchdb-node2:5984,couchdb-node3:5984",
        "databaseName": "orders,customers"
    },
    "customHostCredentials": {
        "pathPrefix": "",
        "connectTimeoutMilliseconds": 10000,
        "requestTimeoutMilliseconds": 300000
    }
}'

certificatePath and certificatePassword go in hostCredentials when TLS needs a private certificate authority. additionalHosts and databaseName are optional.

Setup specific configuration

curl --location --request PUT 'http://core-hub-ip-address/pipelines/{pipelineId}/agents/{agentId}/config/specific' \
--header 'Content-Type: application/json' \
--header 'Authorization: ••••••' \
--data '{
  "configuration": {
      "schemaDiscoverySampleSize": 1000,
      "snapshotReadConcurrency": 8,
      "snapshotFetchBatchSize": 1000,
      "snapshotPageSize": 10000,
      "changesBatchSize": 5000,
      "changesLongPollTimeoutMilliseconds": 1000
    }
}'

Data types

The agent reads the type of each field from the sample of documents, as described in Databases, entities and fields. A field that is missing from a document is reported as absent, and the absent field policy of the entity (absentFieldPolicy) decides what the target receives.

CouchDB type Gluesync type Notes

CouchDB type	Gluesync type	Notes
`string`	`STRING`
`integer`	`LONG`	A value that is not an exact 64-bit integer fails the row, naming the field. Discovery picks `number` for a field with such values.
`number`	`BIG_DECIMAL`	Built from the literal text of the JSON number, so it is as exact as the stored JSON. CouchDB itself stores a decimal as an IEEE double, so `12345678901234567890.123456789` is read back as `12345678901234567000.0`. Integers keep every digit.
`boolean`	`BOOLEAN`
`object`	`MAP`	Converted recursively: strings, integers, decimals and booleans, `null`, lists and maps.
`array`	`ARRAY`	The same conversion, element by element.
`json`	`STRING`	The JSON text of the value. The type of a field that mixes more than one JSON kind, or that holds only `null`.

string

STRING

integer

LONG

A value that is not an exact 64-bit integer fails the row, naming the field. Discovery picks number for a field with such values.

number

BIG_DECIMAL

Built from the literal text of the JSON number, so it is as exact as the stored JSON. CouchDB itself stores a decimal as an IEEE double, so 12345678901234567890.123456789 is read back as 12345678901234567000.0. Integers keep every digit.

boolean

BOOLEAN

object

MAP

Converted recursively: strings, integers, decimals and booleans, null, lists and maps.

array

ARRAY

The same conversion, element by element.

json

STRING

The JSON text of the value. The type of a field that mixes more than one JSON kind, or that holds only null.

Notes

Attachments are neither read nor written. The _attachments member of a document is not a field;
A purged document (_purge) leaves no change in the feed, so the agent cannot delete it on the target;
Partitioned databases are read as normal ones: the agent uses no partition-specific API;
The agent has no recursion protection for a CouchDB to CouchDB pipeline;
In a cluster, the feed of a database is not in commit order across shards. Each document appears once, with its winning revision, so the target ends with the same documents.

Troubleshooting

The server answers 401 or 403

401 is shown as "Authentication Failed": check the username and the password, or turn Disable authentication on for a server that allows anonymous access. 403 is shown as "Permission Denied": the user is not a member of the database. Add the user to the members of the database in its _security object.

No database is listed

The user cannot call _all_dbs, which in CouchDB 3.x needs a server administrator. Put the names of the databases in the Databases (comma separated) field of the connection: the agent lists them and checks each one with GET /<database>.

The database is not found

404 is shown as "Database Not Found", with the name of the database. The database was deleted or renamed, or the Path prefix is wrong. A name in the Databases field that is not a database also ends here.

CDC says to run a new snapshot

The position the entity holds is no longer known to the feed (HTTP 400). Start a new snapshot of the entity.

Deletes do not arrive on the target

A delete carries only the _id. On an entity whose key is made of more columns than _id the agent skips the delete and logs it. Make _id the key of the entity.

The snapshot is slow

Raise Snapshot read concurrency: the reads in flight are the lever, and the number of shards of the source database sets the ceiling. A slow target slows the snapshot down by design, so look at the target too.

The server is overloaded

429 is shown as "Server Overloaded" and 5xx as "Server Error". The agent asks again before the entity fails. Lower Snapshot read concurrency or Changes batch size when the server is shared with other workloads.