Elasticsearch Source

Nanosync replicates documents out of Elasticsearch using sliced Point-in-Time (PIT) snapshots for the initial backfill and watermark polling for continuous ingestion. Elasticsearch has no native CDC/binlog, so the connector emulates change capture by re-querying a monotonic field (default @timestamp) on a long-lived PIT, with a configurable safety lag to absorb out-of-order writes.

Compatible with Elasticsearch 7.10+, 8.x, and 9.x. OpenSearch is API-compatible but not formally CI-tested in v1.

Prerequisites

Elasticsearch 7.10 or later (PIT API)
A user (basic auth or API key) with cluster privilege monitor and index privileges read, view_index_metadata on every replicated index
_source enabled on every replicated index (the connector rejects indices with _source: false at preflight)
A monotonic field present on every replicated index — default @timestamp. Indices without this field will only run snapshot, not CDC.

Deletes are not captured in v1. Elasticsearch’s polling model cannot distinguish a deleted document from one that simply hasn’t been updated. The delete_detection property is reserved for a future periodic full-diff mode. Sinks must tolerate dangling rows or be paired with an external delete signal.

Setup

Create a read-only role

POST /_security/role/nanosync_reader
{
  "cluster": ["monitor"],
  "indices": [
    {
      "names": ["orders-*", "events-*"],
      "privileges": ["read", "view_index_metadata"]
    }
  ]
}

Create a user or API key

User:

POST /_security/user/nanosync
{
  "password": "secret",
  "roles": ["nanosync_reader"]
}

API key (preferred for service auth):

POST /_security/api_key
{
  "name": "nanosync",
  "role_descriptors": {
    "nanosync_reader": {
      "cluster": ["monitor"],
      "indices": [{ "names": ["orders-*"], "privileges": ["read", "view_index_metadata"] }]
    }
  }
}

Verify each replicated index has _source enabled
```
GET /orders-2026.06/_mapping
```
If _source.enabled: false is present, the connector will fail preflight. Re-index with _source enabled before adding the index to the pipeline.
Confirm the watermark field

The default CDC pagination field is @timestamp. If your documents use a different monotonic field (e.g. updated_at, event_time), set cdc_watermark_field in the pipeline properties. The field must be of type date or date_nanos and updated on every write.

_seq_no is not used as the default watermark. It is per-shard monotonic and gap-tolerant, which complicates multi-shard cursor logic. An opt-in _seq_no mode is planned post-v1.

Connection configuration

connections:
  - name: prod-elasticsearch
    type: elasticsearch
    dsn: "https://es-1.prod:9200"
    properties:
      api_key: "${env:ES_API_KEY}"
      ca_cert: "/etc/ssl/es-ca.pem"

Either dsn (a single URL) or the hosts / cloud_id properties must be provided.

Property	Example	Description
`hosts`	`https://es-1:9200,https://es-2:9200`	Comma-separated cluster URLs. Overrides `dsn` when set.
`cloud_id`	`my-cluster:dXMtZWFzdC0x...`	Elastic Cloud identifier. Mutually exclusive with `hosts`.
`username`	`nanosync`	Basic-auth username.
`password`	`${env:ES_PASSWORD}`	Basic-auth password.
`api_key`	`${env:ES_API_KEY}`	API key. Takes precedence over basic auth when set.
`ca_cert`	`/etc/ssl/es-ca.pem`	Path to a PEM-encoded CA certificate.
`tls_skip_verify`	`false`	Disable TLS certificate verification. Development only.
`request_timeout`	`30s`	Per-HTTP-request timeout.

Pipeline configuration

pipelines:
  - name: orders-pipeline
    source:
      connection: prod-elasticsearch
      properties:
        index_pattern:       "orders-*"
        exclude_indices:     "orders-archive-*"
        snapshot_slices:     "0"           # 0 = auto (primary shard count)
        page_size:           "1000"
        cdc_watermark_field: "@timestamp"
        cdc_poll_interval:   "2s"
        cdc_safety_lag:      "30s"

All pipeline-level properties below — including cdc_watermark_field, include_indices, pit_keep_alive, and request_timeout — are also editable from the visual pipeline canvas. Open the source node and switch to the Advanced tab to configure them without touching YAML.

Source properties

Index selection

Property	Default	Description
`index_pattern`	`*`	Glob over indices to capture (e.g. `orders-*`). Used only when `include_indices` is empty.
`include_indices`	—	Comma-separated list of concrete index names. Overrides `index_pattern` when set.
`exclude_indices`	—	Comma-separated list of indices to skip. Applied after pattern/include resolution.

Snapshot

Property	Default	Description
`snapshot_slices`	`0` (auto)	Number of parallel PIT slices per index. When `0`, defaults to the index’s primary shard count, capped by the pipeline’s `MaxPartitions` hint.
`page_size`	`1000`	Hits per `_search` page. Keep well below `http.max_content_length` (default 100 MB).
`pit_keep_alive`	`5m`	PIT `keep_alive` duration. Refreshed automatically on every page.

Elasticsearch warns that slice_max > number_of_primary_shards actively degrades performance — each slice opens its own shard-level cursor. The default snapshot_slices: 0 resolves to the primary shard count, which is almost always the right answer.

CDC / streaming

Property	Default	Description
`cdc_watermark_field`	`@timestamp`	Monotonic field used for CDC pagination. Must be a `date` or `date_nanos` field updated on every write.
`cdc_poll_interval`	`2s`	Sleep between CDC poll iterations.
`cdc_safety_lag`	`30s`	Trailing-edge offset: only documents with `watermark <= now() - safety_lag` are read. Absorbs index refresh delay and clock skew.
`delete_detection`	`none`	Reserved. Only `none` is accepted in v1. Setting any other value is a configuration error.

cdc_safety_lag must be larger than your index refresh_interval (default 1s). If it is not, the polling window can advance past documents that are written but not yet visible, causing data loss. Bump it to 60s or higher on busy clusters where indexing lag is variable.

How replication works

Resolve indices. On Connect, Nanosync calls _cat/indices filtered by index_pattern / include_indices, then strips exclude_indices. The list is cached for the connection lifetime and refreshed on ResolveTables.
Fetch mappings. For each index, _mapping is fetched once and converted to an Apache Arrow schema. Object and nested fields are flattened with dotted-path names. Two synthetic columns — _id (primary key) and _index — are always present.
Snapshot — sliced PIT. The planner queries _cat/shards for the primary shard count and emits one partition per slice. Each worker opens a PIT, then loops _search with slice, pit, and search_after until the page is short. The PIT is closed at the end of the partition.
CDC — watermark polling. After the snapshot completes, the connector opens a long-lived PIT and polls every cdc_poll_interval. Each poll reads [last_seen, now() - safety_lag) on the watermark field, sorted ascending, paginating with search_after. The latest watermark seen is persisted to the checkpoint store.
Checkpoint. The CDC watermark is saved to CheckpointStore as a small JSON resume token. PIT IDs are not persisted — they expire server-side, so a fresh PIT is opened on every restart. Snapshot partition completion is tracked in the existing pipeline_run_partitions table.

Schema mapping → Arrow

Leaf primitives map as follows. Object and nested fields are flattened into dotted-path columns.

Elasticsearch type	Arrow type
`keyword`, `text`, `ip`, `geo_*`	`String`
`integer`, `short`, `byte`	`Int32`
`long`, `unsigned_long`	`Int64`
`float`, `half_float`	`Float32`
`double`, `scaled_float`	`Float64`
`boolean`	`Boolean`
`date`, `date_nanos`	`Timestamp(µs, UTC)`
`binary`	`Binary`
`object`, `nested`	flatten with dotted-path names
Arrays of primitives	`List<T>`

Drift

When a CDC or snapshot hit contains a field that is not in the cached schema, the connector emits a SchemaDelta event with the added column(s), updates the cached *arrow.Schema, and continues. This matches the PostgreSQL pgoutput RELATION handling already in nanosync.

Supported operations

Operation	Supported	Notes
`INSERT`	Yes	Emitted as `Update` (polling cannot distinguish insert from update). Sinks upsert on `_id`.
`UPDATE`	Yes	Same as above. `PrimaryKeys: ["_id"]` is always set.
`DELETE`	No	Not captured in v1. Reserved knob: `delete_detection: full_diff` (future work).
Schema drift	Yes	New fields surface as `SchemaDelta`. Type changes pause the pipeline for review.

Monitoring

Key Prometheus metrics:

Metric	Description
`nanosync_source_events_total{source_type="elasticsearch"}`	Events emitted per index and event type
`nanosync_source_errors_total{source_type="elasticsearch"}`	Errors categorised by failure mode
`nanosync_source_elasticsearch_watermark_lag_seconds{index}`	`now() - last_seen_watermark`. Should hover around `cdc_safety_lag` in steady state.
`nanosync_source_elasticsearch_pit_open{index}`	`1` while a PIT is open for the index, `0` otherwise.

Per-pipeline status:

nanosync metrics pipeline orders-pipeline

Limitations

Deletes are not captured. Sinks accumulate dangling rows for any document removed from the source. Until delete_detection: full_diff ships, pair the pipeline with an external tombstone signal if delete fidelity matters.
Insert vs Update is not distinguishable. Both are emitted as Update. Sinks must perform upsert-on-_id.
_source must be enabled. Indices with _source: false are rejected at preflight — the connector cannot reconstruct rows from doc-values alone.
PIT IDs are not durable. A pipeline restart mid-snapshot re-runs only the partition that was running; completed slices are skipped via PartitionResumer. CDC always reopens a fresh PIT on restart.
Watermark must be monotonic. Out-of-order writes within cdc_safety_lag are absorbed; later out-of-order writes will be missed. Increase cdc_safety_lag if your application has variable indexing lag.
Multi-cluster federation is not supported. One connection points at one cluster (or one cloud_id).
OpenSearch is API-compatible and known to work, but is not formally CI-tested in v1.

Troubleshooting

preflight: index "..." has _source disabled Re-index with "_source": { "enabled": true } in the mapping. Without _source, Nanosync cannot reconstruct full documents.

Pipeline stays in snapshot, never advances to CDC The configured cdc_watermark_field is missing on at least one index. Check the mapping for that field and either add it to the source documents or override cdc_watermark_field to a field that exists.

CDC lag grows without bound cdc_poll_interval may be longer than the rate of writes can sustain, or page_size is too small for the document volume. Increase page_size (up to ~5000 for small docs) or decrease cdc_poll_interval. Watch nanosync_source_elasticsearch_watermark_lag_seconds.

Intermittent missed documents cdc_safety_lag is too tight relative to your index refresh_interval or clock skew across nodes. Increase cdc_safety_lag to 60s or more on busy clusters.

elasticsearch: either dsn/hosts or cloud_id must be provided The connection has no endpoint configured. Set dsn on the connection or hosts / cloud_id in the connection properties.