Push events to Logstash

By default, prismeai-events writes the platform events directly into Elasticsearch or OpenSearch. 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. 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.
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.

prismeai-core-values.yml

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 (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: otherwise, documents rejected by Elasticsearch/OpenSearch (e.g. on a mapping conflict) only show up in Logstash logs.

Elasticsearch
OpenSearch

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).

prismeai-events.conf

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

Logstash’s elasticsearch output refuses to connect to OpenSearch: this pipeline uses the opensearch output plugin, which Elastic’s Logstash images do not include. Install it in your image:

Dockerfile

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.

prismeai-events.conf

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

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), 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), 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 “forked path” pattern: the Prisme.ai pipeline sends a copy of each event to your own pipelines, which filter and transform their copy independently.

In prismeai-events.conf, add a pipeline output next to the elasticsearch or opensearch one, inside the if [@metadata][index] block:
pipeline { send_to => ["siem"] }

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

pipelines.yml

- 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

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

siem.conf

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

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): 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: 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:

EVENTS_STORAGE_WRITE_MODE=logstash docker compose --profile logstash up

Overview

Installation

Databases

Production Readiness

Entreprise Services

Operations

Push events to Logstash

How it works

Configure prismeai-events

Configure Logstash

Least privileges

Add your own outputs

Try it locally

​How it works

​Configure prismeai-events

​Configure Logstash

​Least privileges

​Add your own outputs

​Try it locally

How it works

Configure prismeai-events

Configure Logstash

Least privileges

Add your own outputs

Try it locally