Elasticsearch receiver

Takes what Elasticsearch clients send to Elasticsearch's Bulk API at its port. It answers each request once every destination has its documents.

Filebeat and the other Beats, Fluent Bit and other clients of the Bulk API send to it. Every worker in the group listens on its port, and the sender, or a balancer in front of the workers, picks a worker for each request. Add workers to take more.

Example

Takes bulk requests on port 9200, over TLS with the identity receiver-tls. Senders present the token held in the worker-host variable ES_TOKEN.

{
  "id": "beats",
  "type": "elasticReceiver",
  "port": 9200,
  "tokens": {
    "beats": "ES_TOKEN"
  },
  "identity": "receiver-tls"
}

The Elasticsearch receiver form on the Sources page: id beats, port 9200, no tokens yet. — The same source in the Sources page's add dialog, before you add its token and identity.

Filebeat sends to it with an output.elasticsearch block like this, where ES_TOKEN holds an id:key such as beats:k3y, as the receiver's credential does:

output.elasticsearch:
  hosts: ["https://workers.example.com:9200"]
  api_key: "${ES_TOKEN}"
  ssl.certificate_authorities: ["/etc/filebeat/smeltr-ca.pem"]
setup.template.enabled: false
setup.ilm.enabled: false

Before you start

Pick a port no other receiver in the group uses, such as 9200.
Turn off the sender's setup of index templates and lifecycle policies. The receiver holds neither, and answers 404 to a request for one. On a Beat, set setup.template.enabled: false and setup.ilm.enabled: false.
Decide how senders prove who they are: a token, a client certificate, or both. See Tokens and certificates.
Put each token in a credential.
Open the port on every worker, and give senders one address that reaches them all. See Open its port.

Settings

It also takes the settings every source takes.

port integer required

The port it listens on. Every worker of the group binds it, so no two receivers share one.

Examples

920019200

tokens map default none

Each token's name to the credential holding it. A sender presents one as Authorization: Bearer <token>, as ApiKey <token>, the key as sent or its id:key, or as a Basic password, and any other request is refused 401. A receiver with no token needs an identity with authorities.

Examples

beats = ES_TOKEN{"shippers": {"secret": "es-shippers","version": 1}}

Advanced

identity identity default plain HTTP

The connection identity it serves TLS under: its certificate and key. With authorities, a sender must present a certificate that chains to one of them.

Example

receiver-tls

maxRequestBytes integer default 10485760

The largest request body it reads, between 1024 and 67108864. A larger one is refused 413, so set a sender's bulk size below it.

Examples

104857667108864

Paths it serves

POST or PUT /_bulk, /{index}/_bulk: The body is a bulk request: newline-delimited JSON, plain or with Content-Encoding: gzip. An action line names what to do, and the index, create and update actions are followed by their document. An index in the path, and a pipeline in the query, apply to each action that names none itself.
GET /health: Says whether the receiver takes requests. See Health check.

Records it gives

Each index or create action's document becomes one record: the document as it was sent. This request gives two records:

{"index":{"_index":"web","_id":"a1"}}
{"message":"GET /cart 200","host":"web-1"}
{"create":{"_index":"web"}}
{"message":"GET /cart 500","host":"web-2"}

The receiver takes no update or delete. Each is answered 400 in its own item, and the request's other actions are taken all the same. So is a document that is not a JSON object.

Fields it gives

The index, _id and pipeline an action names for its document, or the path and query name for it, ride with the record. A field the sender did not set is null. __source.remote holds the IP address the request came from: the balancer's, when one is in front of the workers.

Field	Type	Holds
`__source.name`	`str`	The name of the source that read the record.
`__source.type`	`str`	The kind of source that read it, its `type`, such as `kafka`, `s3List`, `httpApi` or `hecReceiver`.
`__source.remote`	`str`	The address of the sender that delivered the record to a receiver.
`__source.index`	`str`	The index a HEC sender filed the event under, or an Elasticsearch client's bulk action named for its document; null where it named none.
`__source.id`	`str`	The `_id` an Elasticsearch client's bulk action named for its document, or null where it named none.
`__source.pipeline`	`str`	The ingest pipeline an Elasticsearch client's bulk action, or its request's `pipeline` query, named for its document, or null where neither named one.
`__source.bytes`	`bytes`	The record's payload as the source read it, before any decoding. The source decodes `_raw` from it. A `bytes` destination writes it unless the destination names another field. Text functions refuse it; `decode(__source.bytes, encoding)` reads it as text.

Tokens and certificates

Anyone a receiver admits can put records into the group's pipelines, so a receiver needs tokens, an identity with authorities, or both. The leader refuses a receiver with neither.

Tokens: a sender presents one in any of these three ways. The receiver refuses any other request before reading its body. Without an identity, the token crosses the network in the clear, so use a token alone only behind a balancer that terminates TLS.
- Authorization: Bearer <token>.
- Authorization: ApiKey <key>, a Beat's api_key. The credential holds either the key's base64 or its id:key.
- Basic, with the token as the password, Fluent Bit's HTTP_Passwd. Any user name will do.
Client certificates: with an identity, the receiver serves TLS with the identity's certificate and key. If the identity has authorities, every connection, health checks included, must present a certificate that chains to one of them.

With both, a sender presents a certificate and a token.

What it answers

The receiver answers success only once every destination has the request's documents, so a sender that retries an unanswered request loses nothing. Each refusal's body holds an error that says why in the reason field.

200

Every destination has the request's documents. The body holds an item per action, in the order sent: 201 for each document taken, with the _id its action named or one the receiver made up, and 400 for each action not taken.

{
  "took": 3,
  "errors": false,
  "items": [
    {
      "index": {
        "_index": "web",
        "_id": "a1",
        "_version": 1,
        "result": "created",
        "_shards": { "total": 1, "successful": 1, "failed": 0 },
        "_seq_no": 0,
        "_primary_term": 1,
        "status": 201
      }
    },
    {
      "create": {
        "_index": "web",
        "_id": "GcEG4pVuTOAAAAAAAAAA",
        "_version": 1,
        "result": "created",
        "_shards": { "total": 1, "successful": 1, "failed": 0 },
        "_seq_no": 0,
        "_primary_term": 1,
        "status": 201
      }
    }
  ]
}

429, with Retry-After

The receiver is full, blocked or stopping. A sender may send it again.

408

The sender took more than 30 seconds to send the whole body.

401

The request has no token, or gives one the receiver does not hold. See Tokens and certificates.

400

The body holds no action, an action line does not read, an action has no document after it, or an action names no index and neither does the path. Nothing of the request is taken.

413

The body is larger than maxRequestBytes: send smaller requests, or raise maxRequestBytes. Or the body holds more actions than a receiver answers at once: send fewer actions to a request.

415

The body uses a Content-Encoding other than gzip or identity.

Health check

GET /health on the receiver's port answers 200 while the receiver takes requests and 429 while it is full, blocked or stopping. It needs no token. Point a balancer's health check or a readiness probe at it, so a busy worker gets no requests until it can take them. Under an identity with authorities it needs a client certificate like any other request.

Senders it takes

These senders are tested against it:

Filebeat 8.17, through output.elasticsearch, as in the example.
Fluent Bit 3.2's es output, with Suppress_Type_Name On.

Fluent Bit sends to it with an es output like this, with ES_TOKEN set in its environment:

[OUTPUT]
    Name               es
    Match              *
    Host               workers.example.com
    Port               9200
    Index              apps
    HTTP_User          fluent
    HTTP_Passwd        ${ES_TOKEN}
    Suppress_Type_Name On
    tls                On
    tls.ca_file        /etc/fluent-bit/smeltr-ca.pem

Open its port

Every worker in the group listens on the receiver's port. Open that port to senders on each worker, and give senders one address that reaches them all, such as a balancer's. On a Unix host, a request sent while a worker deploys a new configuration waits, and the new configuration answers it.

On Kubernetes, add the port to the worker pods and put a Service in front of them: Expose a receiver.
On a host or in a container, open the host's firewall or publish the container's port: Enroll a worker on a host.

Verify

Send one document, and look for "errors":false in the answer:

curl -H "Authorization: Bearer $ES_TOKEN" \
  -H "Content-Type: application/x-ndjson" \
  --data-binary $'{"index":{"_index":"test"}}\n{"message":"hello"}\n' \
  https://workers.example.com:9200/_bulk

Then find the document at the destination.

Troubleshooting

The sender cannot connect: Check that the port is open on every worker and that the sender's address reaches them. An identity with authorities refuses a connection without a certificate that chains to one of them.
Every request gets 401: Check that the sender presents a token one of the receiver's credentials holds, as a bearer token, an API key or a password.
A Beat stops at startup with a 404: Set setup.template.enabled: false and setup.ilm.enabled: false.
Items come back 400 for update or delete: The receiver takes index and create actions only.
Every request gets 429: Check the source's status. A blocked receiver takes nothing until its destinations take records again. A full one takes more with more workers.
The same documents arrive twice: A sender that hears no answer may send the request again. Give it a timeout longer than its destinations take to deliver.

Elasticsearch receiver

Example

Before you start

Settings

Advanced

Paths it serves

Records it gives

Fields it gives

Tokens and certificates

What it answers

Health check

Senders it takes

Open its port

Verify

Troubleshooting

See also