# 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](/docs/secrets#identities) `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.](/elastic-receiver-source-form-light.png)

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](#senders).
-   Put each token in a [credential](/docs/secrets).
-   Open the port on every worker, and give senders one address that reaches them all. See [Open its port](#port).

## Settings

It also takes the settings [every source](/docs/sources#settings) takes.

`port`  integer required

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

Examples:

- `9200`
- `19200`

`tokens`  map default none

Each token's name to the [credential](/docs/secrets) 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](/docs/secrets#identities) 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:

- `1048576`
- `67108864`

## 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](#health).

## 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](/docs/secrets#identities), 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](/docs/delivery#hold) 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](#senders).

**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](#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](/docs/enroll-worker/kubernetes#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](/docs/enroll-worker/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.

## See also

-   [Sources](/docs/sources)
-   [HEC receiver source](/docs/sources/hec-receiver)
-   [HTTP receiver source](/docs/sources/http-receiver)
-   [Elasticsearch destination](/docs/destinations/elasticsearch)
-   [Credentials](/docs/secrets)
-   [Delivery](/docs/delivery)