Elasticsearch/OpenSearch index lifecycle utilities.
Provides two layers of functionality:
Low-level primitives —
detect_engine/1(distribution detection) andensure_index/3(idempotent index creation with automaticdynamic: "strict"injection).Schema lifecycle —
setup/3,status/3,migrate/3, and their*_all/2batch variants, which manage the alias + versioned-index topology for anOrkestra.ES.Schema(one alias per culture). These use the same physical index naming convention asSnap.Indexes.hotswap/5so thatSnap.Indexes.cleanup/4andlist_starting_with/3recognise the indexes they create.
Alias + versioning model
Every schema × culture maps to a stable alias (schema.alias_for/1)
that points to a versioned physical index named "#{alias}-#{unix_µs}".
The physical mapping carries the schema's mapping hash under
mappings._meta.orkestra_schema_hash, which lets status/3 detect drift
between the deployed mapping and the current schema definition.
Migrations reindex zero-downtime via Snap.Indexes.hotswap/5: a fresh
versioned index is created, every document currently behind the alias is
streamed (Snap.Scroll) into it, then the alias is atomically swapped and
old indexes are cleaned up.
Consistency window
migrate/4 does not capture writes that happen concurrently with the
reindex — documents indexed after the scroll snapshot but before the alias
swap are not carried over. Coordinating the write path during a migration is
the caller's responsibility, exactly as it is for a projection rebuild.
Observability
Each lifecycle operation opens an OpenTelemetry span (orkestra.es.setup,
orkestra.es.status, orkestra.es.migrate) carrying the es.index (alias)
and, when applicable, es.culture attributes. migrate/4 records a span
event with the outcome. Structured logs use the orkestra: :es tag and
never include cluster credentials or adapter options.
Summary
Types
A schema module implementing the Orkestra.ES.Schema contract.
Result map returned by status/3.
Functions
Detects the Elasticsearch or OpenSearch engine version.
Creates an Elasticsearch index with strict dynamic mapping enforcement.
Brings the alias for schema (and culture) in line with the schema.
Runs migrate/4 for every culture of schema.
Builds the physical index mapping for schema (and culture).
Ensures the alias + versioned index for schema (and culture) exists.
Runs setup/3 for every culture of schema.
Reports the deployed state of the alias for schema (and culture).
Types
Functions
@spec detect_engine(module()) :: {:ok, :elasticsearch | :opensearch}
Detects the Elasticsearch or OpenSearch engine version.
Calls GET / on the cluster to inspect the version response:
- Response contains
version.distribution: "opensearch"→:opensearch - Response contains
versionwithoutdistribution→:elasticsearch - Connection failure or error → defaults to
:elasticsearchwith a warning
Returns {:ok, :elasticsearch | :opensearch}.
Implementation Notes
Snap.Request.request/7 validates paths and rejects "/" because the URI
split produces an empty segment. We bypass path validation by calling
auth.sign/5 and Snap.HTTPClient.request/6 directly — this keeps full
authentication (API key or Basic Auth) while avoiding the path check.
Defaults to :elasticsearch on any connection or auth failure (defensive
fallback for distributed deployments).
Creates an Elasticsearch index with strict dynamic mapping enforcement.
Idempotent: returns :ok immediately if the index already exists.
The mapping parameter is injected with "dynamic" => "strict" in the
"mappings" block unconditionally, preventing mapping explosion attacks
(T-06-03 mitigation). Any user-supplied dynamic value in the input is
overridden.
Returns:
:ok— index created or already exists{:error, {:index_creation_failed, reason}}— creation failed
Parameters
cluster— theSnap.Clustermoduleindex_name— the index name string (e.g.,"orders")mapping— the Elasticsearch mapping map with"mappings"and optional analysis settings (thedynamic: "strict"injection happens here)
Observability
Emits an OpenTelemetry span orkestra.es.ensure_index with {"es.index"}
attribute. On creation failure, logs with orkestra: :es metadata and sets
span status to error.
@spec migrate(module(), schema(), atom() | nil, keyword()) :: {:ok, :noop | :created | :migrated} | {:error, term()}
Brings the alias for schema (and culture) in line with the schema.
Behaviour:
- Alias absent → delegates to
setup/3, returning{:ok, :created}. - No drift →
{:ok, :noop}(no changes). - Drift → zero-downtime reindex: every document behind the alias is
streamed via
Snap.Scrolland re-indexed into a fresh versioned index throughSnap.Indexes.hotswap/5, which then swaps the alias and cleans up old indexes. Returns{:ok, :migrated}.
Options
:batch_size— scroll page size for the reindex (default500).:page_size,:page_wait,:max_errors,:request_opts— forwarded toSnap.Indexes.hotswap/5.:scroll,:params,:headers,:opts— forwarded toSnap.Scroll.stream/4.
Consistency window
Writes issued during the reindex window are not migrated (see the module doc). Coordinate the write path externally, as with a projection rebuild.
Returns {:ok, :noop | :created | :migrated} or {:error, reason}.
@spec migrate_all(module(), schema()) :: {:ok, [{atom() | nil, :noop | :created | :migrated}]} | {:error, {atom() | nil, term()}}
Runs migrate/4 for every culture of schema.
A mono-culture schema is migrated once with culture nil. Iteration stops
at the first failure.
Returns {:ok, [{culture | nil, :noop | :created | :migrated}]} or
{:error, {culture, reason}}.
Builds the physical index mapping for schema (and culture).
This is the exact mapping used behind the alias by setup/3 and migrate/4:
the schema mapping with dynamic: "strict" and the
mappings._meta.orkestra_schema_hash drift marker injected. Pass culture
as nil (the default) for a mono-culture schema, or one of the declared
cultures for a multi-culture schema.
Exposed so callers that drive their own reindex — notably the
mix orkestra.projection.es.rebuild task via Snap.Indexes.hotswap/5 — use
the same physical mapping (hash included) as the lifecycle helpers, keeping
status/3 drift detection accurate after a rebuild.
@spec setup(module(), schema(), atom() | nil) :: {:ok, :created | :already_exists} | {:error, term()}
Ensures the alias + versioned index for schema (and culture) exists.
For a mono-culture schema pass culture as nil (the default). For a
multi-culture schema pass one of the declared cultures.
Behaviour:
- If the alias already exists →
{:ok, :already_exists}(no changes). - Otherwise a versioned physical index (
"#{alias}-#{unix_µs}") is created with the schema mapping —dynamic: "strict"andmappings._meta.orkestra_schema_hashinjected — and the alias is pointed at it viaSnap.Indexes.alias/4.
Returns {:ok, :created | :already_exists} or {:error, reason}.
@spec setup_all(module(), schema()) :: {:ok, [{atom() | nil, :created | :already_exists}]} | {:error, {atom() | nil, term()}}
Runs setup/3 for every culture of schema.
A mono-culture schema is set up once with culture nil. Iteration stops
at the first failure.
Returns {:ok, [{culture | nil, :created | :already_exists}]} or
{:error, {culture, reason}}.
@spec status(module(), schema(), atom() | nil) :: {:ok, status_result()} | {:error, term()}
Reports the deployed state of the alias for schema (and culture).
Reads the physical index behind the alias and compares its stored
mappings._meta.orkestra_schema_hash with the current schema hash.
Returns {:ok, status} where status is a map with:
:alias— the alias name:exists— whether the alias currently resolves to an index:physical_index— the physical index name, ornilwhen absent:current_hash— the deployed mapping hash, ornilwhen the index was created outside Orkestra (no_meta):schema_hash— the current schema's mapping hash:drift?—truewhen the deployed mapping differs from the schema (including the missing-_metacase);falsewhen the alias is absent or in sync
Returns {:error, reason} on an unexpected cluster error.