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

# Push events to Logstash

> Stream platform events to Logstash instead of writing them directly into Elasticsearch or OpenSearch, to filter, transform and forward them to a SIEM, an observability platform or any other tool.

By default, `prismeai-events` writes the platform events directly into [Elasticsearch or OpenSearch](https://docs.prisme.ai/self-hosting/databases/elasticsearch.md). It can instead **push them to Logstash**, which forwards them to the same data streams and can also send them, filtered or transformed, to your own tools: a SIEM that only accepts push ingestion, an OpenTelemetry conversion towards an observability platform, etc.

```mermaid theme={null}
flowchart LR
  E[prismeai-events] -- "JSON events, by batch (HTTP)" --> L[Logstash]
  L -- "untouched" --> ES[(Elasticsearch / OpenSearch)]
  L -. "filtered / transformed" .-> T[SIEM, OpenTelemetry, ...]
  E -- "search, usage, mappings, cleanup" --> ES
```

The two modes are exclusive: in Logstash mode, `prismeai-events` no longer writes events into Elasticsearch/OpenSearch itself. It still reads them there (activity, usage, search), and still manages the index templates, lifecycle policies and per-workspace mappings, so it **keeps its Elasticsearch/OpenSearch access** (`EVENTS_STORAGE_ES_*`).

## How it works

* Events are pushed by batch (`EVENTS_BUFFER_FLUSH_AT` / `EVENTS_BUFFER_FLUSH_EVERY`, as in direct mode) as a JSON array to the Logstash [http input](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-http.html).
* Each event is the exact Prisme.ai JSON document stored today, after the configured preprocessing rules (redaction), with one extra field, `prismeai_index`, giving the target data stream (one per workspace).
* The provided pipelines write each event to that data stream with the event id as document id, exactly as the direct mode does. An event pushed twice (e.g. retried after a timeout) is therefore written once, unless the data stream rolled over to a new backing index in between.
* When Logstash is unreachable, times out or answers `401`, `403`, `408`, `429` or a `5xx` status, the batch is retried with an exponential backoff (from 1.5s, ×1.5, capped at 30s) until Logstash accepts it, however long the outage lasts, or until wrong `EVENTS_STORAGE_LOGSTASH_USER` / `EVENTS_STORAGE_LOGSTASH_PASSWORD` are fixed.
* Meanwhile, the platform keeps working and new events wait in `prismeai-events` memory, up to `EVENTS_BUFFER_MAX_PENDING` events (50 000 by default): beyond, events are dropped, logged and counted by the `events_persistence_dropped_total` metric. Size this limit according to your pods' memory and average event size, keeping room for the events received while a push hangs (up to `EVENTS_STORAGE_LOGSTASH_TIMEOUT`). The `events_persistence_queue_total` metric gives the current number of waiting events. Being in memory, waiting events are lost if the `prismeai-events` pod restarts.
* Any other error status (`400`, `413` ...) is logged, and the whole batch (up to `EVENTS_BUFFER_HIGH_WATERMARK` events, 800 by default) is dropped without retry and counted by `events_persistence_dropped_total`. A proxy between `prismeai-events` and Logstash that limits the request size (1 MB by default with NGINX Ingress) and answers `413` therefore loses every event.
* The `prismeai-events` readiness probe checks Elasticsearch/OpenSearch, not Logstash: monitor Logstash itself, and alert on the `events_persistence_requests_failed_total` metric as soon as you enable this mode.

## Configure prismeai-events

| Setting | Default | Description |
| - | - | - |
| `EVENTS_STORAGE_WRITE_MODE` Helm: `prismeai-events.env` | `direct` | `direct` writes events into Elasticsearch/OpenSearch, `logstash` pushes them to Logstash. |
| `EVENTS_STORAGE_LOGSTASH_URL` Helm: `prismeai-events.env` | - | URL of the Logstash http input, e.g. `http://logstash:8080`. Required in `logstash` mode: `prismeai-events` refuses to start with an explicit configuration error if it is missing, invalid, or contains credentials (use the variables below). With `https` and a private CA, give it to `prismeai-events` with `NODE_EXTRA_CA_CERTS`. |
| `EVENTS_STORAGE_LOGSTASH_USER` `EVENTS_STORAGE_LOGSTASH_PASSWORD` | - | Basic auth credentials, if enabled on the http input. |
| `EVENTS_STORAGE_LOGSTASH_TIMEOUT` | `10000` ms | Timeout of each push. |
| `EVENTS_BUFFER_MAX_PENDING` | `50000` | Maximum number of events waiting to be persisted. Beyond, events are dropped. Also applies to the direct mode. |

```yaml prismeai-core-values.yml theme={null}
prismeai-events:
  env:
    EVENTS_STORAGE_WRITE_MODE: logstash
    EVENTS_STORAGE_LOGSTASH_URL: http://logstash.logging.svc.cluster.local:8080
```

## Configure Logstash

Prisme.ai provides one pipeline per backend, identical except for their output: they require **Logstash 8.x or later**. Enable the [persistent queue](https://www.elastic.co/guide/en/logstash/current/persistent-queues.html) (`queue.type: persisted`) so that events received by Logstash survive a restart: with several Logstash instances, each one has its own queue. Also enable the [dead letter queue](https://www.elastic.co/docs/reference/logstash/dead-letter-queues): otherwise, documents rejected by Elasticsearch/OpenSearch (e.g. on a mapping conflict) only show up in Logstash logs.

> [tab] Elasticsearch
>
> Set `ELASTICSEARCH_HOSTS` to the cluster `prismeai-events` reads from (`EVENTS_STORAGE_ES_HOST`), as a single URL (list several nodes directly in `hosts` otherwise), and uncomment the credentials if needed. Write the port, even `443`: Logstash uses `9200` when the URL has none. If Elasticsearch serves https with a private CA (e.g. deployed with ECK), mount that CA in Logstash and uncomment `ssl_certificate_authorities` (`cacert` before Logstash 8.8).
>
> ```ruby prismeai-events.conf theme={null}
> input {
>   http {
>     id => "prismeai_events_http"
>     port => "${LOGSTASH_HTTP_PORT:8080}"
>     # One event per element of the JSON array sent by the events service, whatever its content type.
>     # ECS compatibility disabled, else a request holding a single event gets the raw request copied to [event][original]
>     codec => json { ecs_compatibility => disabled }
>     additional_codecs => {}
>     # Keeps the added request metadata to 'headers' & 'host', removed below
>     ecs_compatibility => disabled
>     # Uncomment to require the credentials set in EVENTS_STORAGE_LOGSTASH_USER / EVENTS_STORAGE_LOGSTASH_PASSWORD
>     # user => "${LOGSTASH_HTTP_USER}"
>     # password => "${LOGSTASH_HTTP_PASSWORD}"
>   }
> }
>
> filter {
>   # Target data stream computed by the events service, kept out of the indexed document
>   mutate {
>     rename => { "prismeai_index" => "[@metadata][index]" }
>     # Added by Logstash & its http input, not part of Prisme.ai events
>     remove_field => ["@version", "headers", "host"]
>   }
> }
>
> output {
>   # Events without prismeai_index, i.e. not pushed by the events service, are written nowhere
>   if [@metadata][index] {
>     elasticsearch {
>       id => "prismeai_events_elasticsearch"
>       hosts => ["${ELASTICSEARCH_HOSTS:http://elastic:9200}"]
>       # Uncomment if Elasticsearch requires authentication, with a user only allowed to write events (see the platform documentation)
>       # user => "${ELASTICSEARCH_USER}"
>       # password => "${ELASTICSEARCH_PASSWORD}"
>       # Uncomment for https with a private CA (the EVENTS_STORAGE_ES_TLS_CA_FILE one); 'cacert' before Logstash 8.8
>       # ssl_certificate_authorities => ["${ELASTICSEARCH_CA_FILE}"]
>       # Same write as the events service direct mode: one data stream per workspace, event id as document id
>       index => "%{[@metadata][index]}"
>       action => "create"
>       document_id => "%{id}"
>       # An event already written (e.g. pushed again after a timeout) is rejected by Elasticsearch: nothing to report
>       silence_errors_in_log => ["version_conflict_engine_exception"]
>       # Templates, lifecycle policies & data streams are managed by the events service
>       manage_template => false
>       ilm_enabled => false
>       data_stream => "false"
>       ecs_compatibility => disabled
>     }
>   }
> }
> ```

> [tab] OpenSearch
>
> Logstash's `elasticsearch` output refuses to connect to OpenSearch: this pipeline uses the [`opensearch` output plugin](https://docs.opensearch.org/latest/tools/logstash/ship-to-opensearch/), which Elastic's Logstash images do not include. Install it in your image:
>
> ```dockerfile Dockerfile theme={null}
> FROM docker.elastic.co/logstash/logstash:8.15.3
> RUN bin/logstash-plugin install logstash-output-opensearch
> ```
>
> Set `OPENSEARCH_HOSTS` to the cluster `prismeai-events` reads from (`EVENTS_STORAGE_ES_HOST`), as a single URL (list several nodes directly in `hosts` otherwise), and uncomment the credentials if needed. Write the port, even `443` (e.g. Amazon OpenSearch Service): Logstash uses `9200` when the URL has none. Unlike `prismeai-events`, Logstash verifies the OpenSearch certificate: if it is self-signed or issued by a private CA, mount that CA in Logstash and uncomment `cacert`. With Amazon OpenSearch Service, uncomment `auth_type` instead of the credentials: requests are signed with the AWS credentials found in the environment, the AWS profile or the instance role.
>
> ```ruby prismeai-events.conf theme={null}
> input {
>   http {
>     id => "prismeai_events_http"
>     port => "${LOGSTASH_HTTP_PORT:8080}"
>     # One event per element of the JSON array sent by the events service, whatever its content type.
>     # ECS compatibility disabled, else a request holding a single event gets the raw request copied to [event][original]
>     codec => json { ecs_compatibility => disabled }
>     additional_codecs => {}
>     # Keeps the added request metadata to 'headers' & 'host', removed below
>     ecs_compatibility => disabled
>     # Uncomment to require the credentials set in EVENTS_STORAGE_LOGSTASH_USER / EVENTS_STORAGE_LOGSTASH_PASSWORD
>     # user => "${LOGSTASH_HTTP_USER}"
>     # password => "${LOGSTASH_HTTP_PASSWORD}"
>   }
> }
>
> filter {
>   # Target data stream computed by the events service, kept out of the indexed document
>   mutate {
>     rename => { "prismeai_index" => "[@metadata][index]" }
>     # Added by Logstash & its http input, not part of Prisme.ai events
>     remove_field => ["@version", "headers", "host"]
>   }
> }
>
> output {
>   # Events without prismeai_index, i.e. not pushed by the events service, are written nowhere
>   if [@metadata][index] {
>     opensearch {
>       id => "prismeai_events_opensearch"
>       hosts => ["${OPENSEARCH_HOSTS:http://opensearch:9200}"]
>       # Uncomment if OpenSearch requires authentication, with a user only allowed to write events (see the platform documentation)
>       # user => "${OPENSEARCH_USER}"
>       # password => "${OPENSEARCH_PASSWORD}"
>       # Uncomment for https with a self-signed or private CA certificate: unlike the events service, Logstash verifies it
>       # cacert => "${OPENSEARCH_CA_FILE}"
>       # Amazon OpenSearch Service, instead of user / password: uncomment.
>       # AWS credentials are read from the environment, the AWS profile or the instance role
>       # auth_type => { type => "aws_iam" region => "${AWS_REGION}" service_name => "es" }
>       # Same write as the events service direct mode: one data stream per workspace, event id as document id
>       index => "%{[@metadata][index]}"
>       action => "create"
>       document_id => "%{id}"
>       # An event already written (e.g. pushed again after a timeout) is rejected by OpenSearch: nothing to report
>       failure_type_logging_whitelist => ["version_conflict_engine_exception"]
>       # Templates, lifecycle policies & data streams are managed by the events service
>       manage_template => false
>       ecs_compatibility => disabled
>     }
>   }
> }
> ```

> [!WARNING]
>
> Do not filter or transform events in this pipeline, and keep the template, lifecycle and data stream management of its output disabled (`manage_template`, plus `ilm_enabled` and `data_stream` with Elasticsearch): the platform reads these documents back as is, and relies on the data streams, templates and lifecycle policies created by `prismeai-events`. An event that loses its `prismeai_index` field is written nowhere, without any error.

## Least privileges

The http input accepts any event sent to it, which Logstash then writes with its own credentials. Restrict both to what `prismeai-events` needs:

* Only let `prismeai-events` reach the http input (see [Network Policies](https://docs.prisme.ai/self-hosting/kubernetes/network-policies.md)), and enable its basic auth: uncomment `user` and `password` in the `http` input, and set the same credentials in `EVENTS_STORAGE_LOGSTASH_USER` / `EVENTS_STORAGE_LOGSTASH_PASSWORD`.
* Give Logstash its own user, only allowed to write events, rather than the `prismeai-events` one. With Elasticsearch:

  * cluster privilege `monitor`, without which Logstash does not start;
  * on `${eventsPrefix}-events-*` (see [Elasticsearch least privileges](https://docs.prisme.ai/self-hosting/databases/elasticsearch.md#least-privileges)), the `create_doc` index privilege, and `auto_configure` to create the data stream of a workspace on its first event.

  With OpenSearch, grant the equivalent permissions: reading the cluster information, and creating documents and data streams on the same pattern.

An event Logstash is not allowed to write is retried indefinitely and blocks a pipeline worker: check that the pattern matches the `EVENTS_STORAGE_NAMESPACE` of `prismeai-events`.

## Add your own outputs

To send events to other tools without altering what is written to Elasticsearch/OpenSearch, use Logstash's [pipeline-to-pipeline](https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html) "forked path" pattern: the Prisme.ai pipeline sends a copy of each event to your own pipelines, which filter and transform their copy independently.

1. In `prismeai-events.conf`, add a `pipeline` output next to the `elasticsearch` or `opensearch` one, inside the `if [@metadata][index]` block:

   ```ruby theme={null}
   pipeline { send_to => ["siem"] }
   ```

2. Declare your pipeline in `pipelines.yml`, with its own persistent queue:

   ```yaml pipelines.yml theme={null}
   - pipeline.id: prismeai-events
     path.config: "/usr/share/logstash/pipeline/prismeai-events.conf"
     queue.type: persisted
   - pipeline.id: siem
     path.config: "/usr/share/logstash/pipeline/siem.conf"
     queue.type: persisted
     # Holds the copies while the SIEM is down: size it for its longest outage
     queue.max_bytes: 4gb
   ```

3. Write it, for instance to forward only errors to a SIEM:

   ```ruby siem.conf theme={null}
   input { pipeline { address => "siem" } }
   filter {
     if [type] != "error" { drop {} }
   }
   output {
     http {
       url => "https://siem.example.com/ingest"
       http_method => "post"
       format => "json"
     }
   }
   ```

> [!WARNING]
>
> Your pipeline can stop the writes to Elasticsearch/OpenSearch. While your tool is down or slow, the copies pile up in your pipeline's queue. Once that queue is full, Logstash also blocks the Prisme.ai pipeline ([a blocked downstream pipeline blocks the sending pipeline](https://www.elastic.co/guide/en/logstash/current/pipeline-to-pipeline.html)): nothing is written to Elasticsearch/OpenSearch anymore, Logstash rejects or stops answering the pushes, and `prismeai-events` keeps up to `EVENTS_BUFFER_MAX_PENDING` events in memory before dropping events, which are then missing from the platform Activity and usage too. Size `queue.max_bytes` for the longest outage of your tool, and monitor that queue.

Each event has the fields shown in the workspace [Activity](https://docs.prisme.ai/products/ai-builder/activity.md#known-filter-fields): `id`, `type`, `source` (`workspaceId`, `userId`, `correlationId`, `appSlug` ...), `payload`, `createdAt`, plus `@timestamp` (equal to `createdAt`). The target data stream is available as `[@metadata][index]`. An event pushed twice reaches your outputs twice: deduplicate on `id` if your tool needs it.

## Try it locally

The platform `docker-compose.yml` ships an optional Logstash service using the Elasticsearch pipeline:

```bash theme={null}
EVENTS_STORAGE_WRITE_MODE=logstash docker compose --profile logstash up
```


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.