> ## Documentation Index
> Fetch the complete documentation index at: https://ventstream.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# OpenSearch / Elasticsearch sink

> Write denormalized documents to OpenSearch or Elasticsearch with idempotent bulk upserts.

The sink writes the engine's output documents to **OpenSearch** or
**Elasticsearch** — the same connector handles both, since they share the
`_bulk` API and document model.

## Configuration

| Variable                            | Default | Purpose                                                                                           |
| ----------------------------------- | ------- | ------------------------------------------------------------------------------------------------- |
| `VS_OS_ENDPOINT`                    | —       | base URL, e.g. `http://localhost:9200` (required)                                                 |
| `VS_OS_USERNAME` / `VS_OS_PASSWORD` | —       | HTTP basic auth                                                                                   |
| `VS_OS_API_KEY`                     | —       | `ApiKey`-header auth (alternative to basic)                                                       |
| `VS_OS_TLS_MODE`                    | unset   | `verify_full` requires HTTPS and validates the certificate and hostname; `disabled` requires HTTP |
| `VS_OS_TLS_CA_FILE`                 | —       | PEM CA bundle for a private CA                                                                    |
| `VS_INDEX_TEMPLATE`                 | —       | how each document's target index name is built                                                    |

Bulk batching, parallelism, and the dead-letter path are tuned with the
shared dispatcher knobs (`VS_DISPATCH_*`) — see the
[engine env reference](/docs/reference/engine-env).

For a private CA, mount its PEM bundle and set
`VS_OS_TLS_MODE=verify_full` with `VS_OS_TLS_CA_FILE`. Publicly issued
certificates need strict mode but no CA file.

## Index naming

`VS_INDEX_TEMPLATE` is expanded **per document** at write time. A literal
string is used as-is; tokens let you route by event/time:

| Token                         | Expands to                                                  |
| ----------------------------- | ----------------------------------------------------------- |
| `${header:NAME}`              | the value of header `NAME` (e.g. `ventstream.cdc.relation`) |
| `${subject}` / `${subject:N}` | the event subject, or its N-th dot-segment                  |
| `%Y` `%m` `%d` `%H`           | UTC time components of the write                            |

All substitutions are lowercased and sanitized to `[a-z0-9_-]` (index
names must be lowercase). Example: `events-${subject:1}-%Y-%m-%d`. A plain
value like `orders` just targets the `orders` index.

For raw table/collection sync, route by the output relation:

```bash theme={null}
VS_INDEX_TEMPLATE='${header:ventstream.cdc.relation}'
```

For PostgreSQL or MySQL projection-owned index names, add `target.index` to
every join/projection and route by the target header:

```bash theme={null}
VS_INDEX_TEMPLATE='${header:ventstream.target.index}'
```

That keeps the sink responsible for the OpenSearch connection while the
projection spec owns the document stream's index name. Canonical engine
configuration using `strategy: by_projection_target` fails startup if the
joins file is empty or any projection omits `target.index`.

## Idempotent writes

Every document's `_id` is derived from the source key (see
[deterministic document IDs](/docs/concepts/architecture#deterministic-document-ids)),
so each write is an **upsert** that targets exactly one document:

* Postgres: `shop.orders:["<order_id>"]`
* Neo4j: `products_denormalized:<elementId>`

Because the ID is a pure function of the key, bootstrap, live updates,
re-syncs, and reconciliation all converge on the same state no matter how
many times a record is re-emitted. A source delete becomes a document
**tombstone** (delete by `_id`).

## Reliability

* **Bulk upserts** are batched and flushed with bounded parallelism.
* Transport failures, HTTP `408`, `429`, and `5xx` responses retry with
  exponential backoff, jitter, and a 30-second delay ceiling. A
  `Retry-After` delta supplied by the server is honored up to that ceiling.
* Retryable sink outages apply backpressure for the lifetime of the process.
  They do not create DLQ records.
* A permanent bulk-item failure with an exact item offset, such as a document
  mapping conflict, goes to the dead-letter queue (`VS_DLQ_PATH`). Whole-request
  authentication, configuration, protocol, and unknown item failures stop
  delivery without advancing the source cursor.
* The source cursor only advances once the sink **confirms** the write —
  so a crash resumes from the last durably-written position. See
  [Architecture](/docs/concepts/architecture#crash-safety-the-cursor-watermark).
* `/readyz` returns `503` when a retryable sink outage lasts at least 30
  seconds, and immediately for an authentication or configuration blocker.
  `/healthz` remains a process-liveness check.

The Prometheus endpoint exposes `vs_sink_available` and
`vs_sink_outage_seconds`. Alert on sustained unavailability, sink retry growth,
source cursor age, and source-retention headroom.

<Warning>
  Fail-closed retry prevents the engine from discarding records, but it cannot
  extend the source system's retention. Size PostgreSQL WAL, MySQL binlogs,
  MongoDB's oplog, Neo4j CDC retention, or Kafka topic retention for the longest
  sink outage you intend to recover without a new snapshot.
</Warning>

## Production topology

Run the destination as a replicated service across failure domains. A
single-node OpenSearch or Elasticsearch cluster is suitable for local
development, not a production CDC sink.

For Amazon OpenSearch Service, use **Multi-AZ with Standby** across three
Availability Zones, three dedicated cluster-manager nodes, and three data nodes
or a multiple of three. Each index needs at least two replicas so all three
zones hold a copy. Enable Auto-Tune and software updates, then alarm on:

* `ClusterStatus.red` and sustained `ClusterStatus.yellow`
* `ClusterIndexWritesBlocked`
* low `FreeStorageSpace`
* `JVMMemoryPressure` and `OldGenJVMMemoryPressure`
* missing nodes, high CPU, and snapshot failures

The `infra/aws` example accepts `opensearch_topology = "ha"` to exercise this
topology. Size its node types, storage, shard count, and alarm actions from
measured indexing and query load before using it for customer traffic.

<Warning>
  The local demo disables the OpenSearch security plugin for convenience.
  In production, set `VS_OS_USERNAME`/`VS_OS_PASSWORD` (or `VS_OS_API_KEY`)
  and use TLS — never run an unauthenticated sink.
</Warning>