Orkestra.ES.Index (orkestra v0.2.0)

Copy Markdown View Source

Elasticsearch/OpenSearch index lifecycle utilities.

Provides two layers of functionality:

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

schema()

@type schema() :: module()

A schema module implementing the Orkestra.ES.Schema contract.

status_result()

@type status_result() :: %{
  alias: String.t(),
  exists: boolean(),
  physical_index: String.t() | nil,
  current_hash: String.t() | nil,
  schema_hash: String.t(),
  drift?: boolean()
}

Result map returned by status/3.

Functions

detect_engine(cluster)

@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 version without distribution:elasticsearch
  • Connection failure or error → defaults to :elasticsearch with 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).

ensure_index(cluster, index_name, mapping)

@spec ensure_index(module(), String.t(), map()) ::
  :ok | {:error, {:index_creation_failed, term()}}

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 — the Snap.Cluster module
  • index_name — the index name string (e.g., "orders")
  • mapping — the Elasticsearch mapping map with "mappings" and optional analysis settings (the dynamic: "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.

migrate(cluster, schema, culture \\ nil, opts \\ [])

@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.Scroll and re-indexed into a fresh versioned index through Snap.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 (default 500).
  • :page_size, :page_wait, :max_errors, :request_opts — forwarded to Snap.Indexes.hotswap/5.
  • :scroll, :params, :headers, :opts — forwarded to Snap.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}.

migrate_all(cluster, schema)

@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}}.

physical_mapping(schema, culture \\ nil)

@spec physical_mapping(schema(), atom() | nil) :: map()

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.

setup(cluster, schema, culture \\ nil)

@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" and mappings._meta.orkestra_schema_hash injected — and the alias is pointed at it via Snap.Indexes.alias/4.

Returns {:ok, :created | :already_exists} or {:error, reason}.

setup_all(cluster, schema)

@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}}.

status(cluster, schema, culture \\ nil)

@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, or nil when absent
  • :current_hash — the deployed mapping hash, or nil when the index was created outside Orkestra (no _meta)
  • :schema_hash — the current schema's mapping hash
  • :drift?true when the deployed mapping differs from the schema (including the missing-_meta case); false when the alias is absent or in sync

Returns {:error, reason} on an unexpected cluster error.