elasticsearch subscriber

Indexes enriched events into Elasticsearch datastreams.

Add to the subscribers block in config.yaml:

subscribers:
  elasticsearch:
    enabled: true
    addresses:
      - "https://es-host:9200"
    username: ""             # prefer env var ES_USERNAME
    password: ""             # prefer env var ES_PASSWORD
    datastream_prefix: "ziti"
    flush_interval: 5s
    flush_size: 100
    workers: 2               # bulk indexer workers
    buffer_size: 1000
    # Optional field-based index routing (see below).
    # index_field: "namespace"
    # index_allowlist:
    #   - "zfw.*"
    #   - "zeek.*"
    # Per-event filter applied before each bulk-indexer Add (drops never
    # touch the bulk API). include is any-of, exclude is none-of. See
    # ../ "Per-Subscriber Filtering" for the comparator reference.
    include: []
    exclude: []

Events are indexed into datastreams named <datastream_prefix>.<namespace> (e.g., ziti.usage, ziti.circuit). Uses Elasticsearch's bulk indexer with _op_type: create for datastream compatibility. The @timestamp field is set from the event timestamp. workers maps directly to the bulk indexer's NumWorkers — raise it when bulk index latency is holding up the pipeline.

Available fields and defaults

Field	Default	Description
addresses	—	Elasticsearch node URLs
username	—	Basic auth username (prefer ES_USERNAME env var)
password	—	Basic auth password (prefer ES_PASSWORD env var)
datastream_prefix	ziti	Prefix for datastream names
flush_interval	5s	Max time before flushing a partial batch
flush_size	100	Events per bulk request
workers	2	Bulk indexer worker count
buffer_size	1000	Subscriber channel capacity
index_field	`` (off)	gjson path read from each document; when present, its value routes the document to that index (see below)
index_allowlist	[] (any)	Glob patterns gating index_field routing; a routed index matching none of these is silently dropped
include	[]	Per-event predicates against the enriched payload; any-of. Empty = pass everything. See Per-subscriber filtering.
exclude	[]	Per-event predicates; none-of — drop on match.

Field-based index routing

By default every event goes to <datastream_prefix>.<namespace>, where <namespace> is the event-envelope namespace (for Beats input this is the configured inputs.beats.namespace label, so all Beats data lands in one index).

Set index_field to route documents to different indices based on a field inside the document — useful when ingesting third-party data (e.g. filebeat shipping zfw and zeek logs) that you want in their own indices:

subscribers:
  elasticsearch:
    enabled: true
    addresses: ["https://es-host:9200"]
    index_field: "namespace"        # any gjson path, e.g. fields.log_index
    index_allowlist:                # optional safety gate
      - "zfw.*"
      - "zeek.*"

Resolution rules when index_field is set:

Document	Target index
Field present and value matches the allowlist (or allowlist empty)	the value verbatim, lowercased/sanitized (e.g. zfw.traffic → index zfw.traffic)
Field present but value matches no allowlist pattern	silently dropped (not indexed)
Field absent/empty	falls back to the default <datastream_prefix>.<namespace>

Notes:

• Routing applies only to non-controller inputs. Native Ziti controller telemetry always uses the default <datastream_prefix>.<namespace> scheme and is immune to index_field/index_allowlist, regardless of config — Ziti events carry their own top-level namespace field (metrics, circuit, …), so routing them would divert them out of ziti.* and an allowlist scoped to third-party data would silently drop every such event. Field routing consults only Beats (and future third-party) inputs, so enabling it for a Beats input can never break controller telemetry.
• The value becomes the full index name (not prefixed). Allowlist patterns are standard globs (path.Match) tested against the sanitized value.
• Values are lowercased and rejected (dropped) if they contain a space or any character Elasticsearch disallows in index names (\/*?"<>|,#) or start with -, _, + — this prevents arbitrary/junk index creation.
• index_allowlist is opt-in: with it empty, any present value routes verbatim. With it set, it is a hard gate — anything outside it is dropped.
• Immunity is by source, not by value. Only the input source (controller vs. Beats/third-party) decides whether routing runs — not the routing value. Controller telemetry is always safe. A Beats event is always subject to routing, even if it happens to carry a Ziti-style namespace value: with index_field: namespace + an allowlist that excludes it, that Beats event is dropped. If your Beats data sets a namespace you don't want to route on, point index_field at a dedicated field (e.g. fields.log_index) instead.

Automatic count-key normalization

Ziti entityCount events carry a counts map holding both a total and dotted breakdowns in the same object (e.g. {"services": 42, "services.edge": 40}). Elasticsearch expands dotted field names into nested objects, so counts.services would have to be both a scalar and an object — a mapping collision that rejects the whole bulk write. Before indexing, the subscriber de-dots the counts keys (services.edge → services_edge), matching the legacy Logstash pipeline, so the total and each breakdown stay distinct scalar fields. This is automatic and needs no configuration.

See Common tuning for workers and buffer_size semantics.