Logstash

Logstash reaches OBSESC through its standard elasticsearch output. OBSESC speaks the Elasticsearch bulk protocol and answers the startup checks Logstash makes. Because Logstash supports many outputs in one pipeline, it’s also a convenient place to fan out to OBSESC alongside your existing destination.

Endpoint

Setting	Value
Protocol	Elasticsearch _bulk (NDJSON over HTTP)
Default port	9200 (ingest.es_bulk_port)
Compression	gzip accepted
Authentication	Authorization: ApiKey <token> or Bearer <token>, checked against security.ingest_tokens

When Logstash connects, it checks GET / for a compatible version, checks GET /_license, and looks for the X-elastic-product response header. The node answers all three as an Elasticsearch 8.x cluster with a basic licence.

Pipeline configuration

Add a second output next to your existing one. Each event goes to both.

input {
  beats {
    port => 5044
  }
}

filter {
  # Give every event a service. OBSESC groups everything by it.
  if ![service] {
    mutate { add_field => { "service" => "checkout" } }
  }
}

output {
  # Existing destination, unchanged
  # elasticsearch { hosts => ["https://es.your-domain.internal:9200"] ... }

  # OBSESC destination (added)
  elasticsearch {
    hosts            => ["http://obsesc.your-domain.internal:9200"]
    index            => "logs"
    data_stream      => false
    manage_template  => false
    ilm_enabled      => false
    http_compression => true
    # When the node enforces security.ingest_tokens (needs https:// hosts):
    # api_key        => "id:api_key"
  }
}

What these settings do:

• data_stream => false and a fixed index keep Logstash from choosing a data-stream name automatically. If an event has no service field, OBSESC falls back to the index name as the service.
• manage_template => false and ilm_enabled => false stop Logstash from installing index templates and lifecycle policies. The node doesn’t serve those APIs.
• http_compression => true gzips request bodies. The node decompresses them.

Set a service field

OBSESC reads the service from the event’s service field, then service.name, then the index name. Set service in a filter as above, or copy it from a field you already have:

filter {
  mutate { copy => { "[kubernetes][labels][app]" => "service" } }
}

Authentication

When security.ingest_tokens is set, use the output’s api_key option. Logstash only sends an API key over TLS, so the node needs TLS configured and the output’s hosts must use https:// (see Encryption). Logstash sends Authorization: ApiKey base64(id:api_key), and the node compares that base64 string with its allowlist. The token you configure on the node must therefore be the base64 of id:api_key:

printf '%s' 'id:api_key' | base64

Basic authentication (user / password) isn’t accepted.

How events are mapped

The node maps Logstash events exactly as it maps any bulk document:

Event field	OBSESC field
service, else service.name, else the index name	service
message, else log	body. If neither is present, the whole document is stored
@timestamp, else timestamp, else ts	timestamp
Everything else	Attributes, flattened to dotted keys

See Elasticsearch bulk for the full rules.

Delivery and retries

• The node answers with an Elasticsearch-compatible bulk response only after the batch is durable.
• Under backpressure it answers 503 with Retry-After, and the Logstash output retries.
• By default Logstash doesn’t set a document ID, so delivery is at-least-once.

Avoiding duplicates with document_id

If your events already carry an ID that’s unique per event (for example a request ID or a source record ID), pass it as the document ID:

elasticsearch {
  hosts       => ["http://obsesc.your-domain.internal:9200"]
  index       => "logs"
  document_id => "%{[event][id]}"
  # ...
}

The node then drops a retried event it has already stored, provided the retry arrives soon after the original. Don’t generate the ID by hashing the event’s content: two genuinely identical log lines would get the same ID, and the second one would be discarded.

Using Logstash as a fan-out point

Filebeat and Winlogbeat allow only one output each. To dual-write from them without running a second agent, point them at a Logstash beats input and let Logstash send to both destinations, as in the pipeline above.