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

# Elasticsearch and Elastic Cloud

> Client-mode REST API connector for Elasticsearch (self-managed, Elastic Cloud Hosted and Serverless): index MQTT telemetry as documents or into time-series data streams, bulk-load NDJSON batches, and query back with search, count and ES|QL

Technical Preview

## Overview

A client-mode connector for the [Elasticsearch REST API](https://www.elastic.co/docs/api/doc/elasticsearch) that writes MQTT data into Elasticsearch and reads it back:

| Catalog connector event | Endpoint | Notes |
| - | - | - |
| Index a document | `POST /{index}/_doc` | [Index API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-index). Auto-generated `_id`, returns `201` with `"result":"created"`. Works for data streams too. |
| Bulk-load a batch | `POST /_bulk` | [Bulk API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-bulk). NDJSON body (`application/x-ndjson`) of action + source line pairs. Returns `200` even when some items fail, so check `errors`. |
| Search | `POST /{index}/_search` | [Search API](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-search) with Query DSL, sort, aggregations. |
| Put / get / update by id | `PUT /{index}/_doc/{id}`, `GET /{index}/_doc/{id}`, `POST /{index}/_update/{id}` | Latest asset state keyed by a stable id (for example `pump-7`). Not for data streams. |
| Count | `POST /{index}/_count` | Query DSL count. |
| ES\|QL | `POST /_query` | [ES\|QL query](https://www.elastic.co/docs/api/doc/elasticsearch/operation/operation-esql-query), for example `FROM metrics-coreflux-* \| STATS AVG(temperature) BY site`. |
| Refresh | `POST /{index}/_refresh` | Make recent writes searchable now (normally about 1 s). Use sparingly. |
| Time-series setup | `PUT /_index_template/{name}`, `GET /_data_stream/{name}` | Create an index template with `"data_stream": {}` so writes to `metrics-…` create a [data stream](https://www.elastic.co/docs/manage-data/data-store/data-streams). |
| Smoke | `GET /`, `GET /_cluster/health` | Cluster info and health. |

Topics use `data/elasticsearch/<object>/<verb>`, and replies go to `<topic>/response`. `BASE_ADDRESS` has no trailing slash; one connector instance talks to one cluster or project:

| Deployment | `BASE_ADDRESS` |
| - | - |
| Elastic Cloud Hosted | `https://<deployment>.es.<region>.<provider>.elastic-cloud.com` (port 443) |
| Elastic Cloud Serverless | `https://<project>.es.<region>.<provider>.elastic.cloud` (port 443) |
| Self-managed | `https://<host>:9200` |

Copy the Elasticsearch endpoint from the Elastic Cloud console (**Manage** / **Connection details**), not the Kibana URL.

## Auth

Every event sends `Authorization: ApiKey …`. The template reads the key from the MQTT payload field `api_key`. Keep it in a Hub secret such as `GET SECRET "ELASTICSEARCH_API_KEY"` and never commit it.

* **API key (recommended, required on Serverless):** create one in Kibana (**Stack Management → API keys**) or with `POST /_security/api_key`, and use the `encoded` value, which is base64 of `id:api_key`. See [Elasticsearch API keys](https://www.elastic.co/docs/deploy-manage/api-keys/elasticsearch-api-keys).
* **Scope the key** to what the connector writes. A key like the one used in testing:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "name": "coreflux-writer",
  "role_descriptors": {
    "coreflux_writer": {
      "cluster": ["monitor", "manage_index_templates"],
      "indices": [{
        "names": ["factory-*", "metrics-coreflux-*"],
        "privileges": ["create_doc", "index", "read", "view_index_metadata", "create_index", "auto_configure", "maintenance"]
      }]
    }
  }
}
```

Drop `manage_index_templates` once the template exists, and drop `index` if the connector only appends (`create_doc`).

* **Basic auth (alternative, self-managed and Hosted):** remove the `Authorization` header lines and add `WITH USERNAME "coreflux_writer"` and `WITH PASSWORD GET SECRET "ELASTICSEARCH_PASSWORD"` to `REST_API_CONFIG`. Don't use the `elastic` superuser.

A `401` `security_exception` means the key is missing, wrong or expired. A `403` naming an action such as `indices:admin/auto_create` means the key is valid but lacks that privilege on that index.

## Basic Syntax

```lot wrap theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
DEFINE ROUTE ElasticsearchRestApi WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH BASE_ADDRESS "https://YOUR_DEPLOYMENT.es.YOUR_REGION.aws.elastic-cloud.com"
        WITH ENABLE_CLIENT "true"
        WITH USE_SSL "true"
        WITH REQUEST_TIMEOUT '60'
        WITH DEFAULT_CONTENT_TYPE "application/json"
        # Auth (prefer Hub secrets / GET SECRET "ELASTICSEARCH_API_KEY"): api_key on each MQTT payload → ApiKey header below.
        # Smoke: GET / → 200 cluster info; POST {index}/_doc → 201 "result":"created".
    ADD EVENT GetClusterInfo
        WITH SOURCE_TOPIC "data/elasticsearch/cluster/info"
        WITH DESTINATION_TOPIC "data/elasticsearch/cluster/info/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
    ADD EVENT GetClusterHealth
        WITH SOURCE_TOPIC "data/elasticsearch/cluster/health"
        WITH DESTINATION_TOPIC "data/elasticsearch/cluster/health/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/_cluster/health"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
    ADD EVENT IndexDocument
        WITH SOURCE_TOPIC "data/elasticsearch/document/index"
        WITH DESTINATION_TOPIC "data/elasticsearch/document/index/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/{value.json.index}/_doc"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
        WITH BODY "{value.json.body}"
    ADD EVENT PutDocument
        WITH SOURCE_TOPIC "data/elasticsearch/document/put"
        WITH DESTINATION_TOPIC "data/elasticsearch/document/put/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/{value.json.index}/_doc/{value.json.id}"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
        WITH BODY "{value.json.body}"
    ADD EVENT GetDocument
        WITH SOURCE_TOPIC "data/elasticsearch/document/get"
        WITH DESTINATION_TOPIC "data/elasticsearch/document/get/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/{value.json.index}/_doc/{value.json.id}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
    ADD EVENT UpdateDocument
        WITH SOURCE_TOPIC "data/elasticsearch/document/update"
        WITH DESTINATION_TOPIC "data/elasticsearch/document/update/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/{value.json.index}/_update/{value.json.id}"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
        WITH BODY "{value.json.body}"
    ADD EVENT BulkIndex
        WITH SOURCE_TOPIC "data/elasticsearch/bulk"
        WITH DESTINATION_TOPIC "data/elasticsearch/bulk/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/_bulk"
        WITH HEADERS
            ADD "Content-Type" "application/x-ndjson"
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
        WITH BODY "{value.json.content}"
    ADD EVENT Search
        WITH SOURCE_TOPIC "data/elasticsearch/search"
        WITH DESTINATION_TOPIC "data/elasticsearch/search/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/{value.json.index}/_search"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
        WITH BODY "{value.json.body}"
    ADD EVENT CountDocuments
        WITH SOURCE_TOPIC "data/elasticsearch/count"
        WITH DESTINATION_TOPIC "data/elasticsearch/count/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/{value.json.index}/_count"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
        WITH BODY "{value.json.body}"
    ADD EVENT EsqlQuery
        WITH SOURCE_TOPIC "data/elasticsearch/esql/query"
        WITH DESTINATION_TOPIC "data/elasticsearch/esql/query/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/_query"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
        WITH BODY "{value.json.body}"
    ADD EVENT RefreshIndex
        WITH SOURCE_TOPIC "data/elasticsearch/index/refresh"
        WITH DESTINATION_TOPIC "data/elasticsearch/index/refresh/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/{value.json.index}/_refresh"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
    ADD EVENT PutIndexTemplate
        WITH SOURCE_TOPIC "data/elasticsearch/index-template/put"
        WITH DESTINATION_TOPIC "data/elasticsearch/index-template/put/response"
        WITH METHOD "PUT"
        WITH ENDPOINT "/_index_template/{value.json.name}"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
        WITH BODY "{value.json.body}"
    ADD EVENT GetDataStream
        WITH SOURCE_TOPIC "data/elasticsearch/data-stream/get"
        WITH DESTINATION_TOPIC "data/elasticsearch/data-stream/get/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/_data_stream/{value.json.name}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "ApiKey {value.json.api_key}"
```

> [!NOTE]
>
> JSON bodies come from the payload field `body`, so control fields (`api_key`, `index`, `id`, `name`) are not sent to Elasticsearch. `BulkIndex` sends the payload field `content` as the raw NDJSON body.

## Example MQTT payloads

Index one reading into a data stream (`data/elasticsearch/document/index`):

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "api_key": "<from GET SECRET ELASTICSEARCH_API_KEY>",
  "index": "metrics-coreflux-factory",
  "body": { "@timestamp": "2026-10-04T22:50:00.000Z", "site": "porto", "line": "L1", "temperature": 22.0 }
}
```

Bulk-load a batch (`data/elasticsearch/bulk`). Every line ends in `\n`, including the last one. Data streams only accept `create`:

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "api_key": "<…>",
  "content": "{\"create\":{\"_index\":\"metrics-coreflux-factory\"}}\n{\"@timestamp\":\"2026-10-04T22:50:00.000Z\",\"site\":\"porto\",\"temperature\":21.5}\n{\"index\":{\"_index\":\"factory-assets\",\"_id\":\"pump-7\"}}\n{\"asset\":\"pump-7\",\"state\":\"running\",\"rpm\":1450}\n"
}
```

Upsert the latest state of an asset (`data/elasticsearch/document/put`, then `data/elasticsearch/document/update`):

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{ "api_key": "<…>", "index": "factory-assets", "id": "press-3", "body": { "asset": "press-3", "state": "idle", "rpm": 0 } }
```

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{ "api_key": "<…>", "index": "factory-assets", "id": "press-3", "body": { "doc": { "state": "running", "rpm": 900 }, "doc_as_upsert": true } }
```

Search the latest readings for a site (`data/elasticsearch/search`):

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "api_key": "<…>",
  "index": "metrics-coreflux-factory",
  "body": { "size": 5, "query": { "term": { "site": "porto" } }, "sort": [ { "@timestamp": "desc" } ] }
}
```

Average temperature by site with ES|QL (`data/elasticsearch/esql/query`):

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{ "api_key": "<…>", "body": { "query": "FROM metrics-coreflux-factory | STATS avg_t = AVG(temperature) BY site | SORT site" } }
```

Create the time-series index template once (`data/elasticsearch/index-template/put`):

```json theme={"theme":"css-variables","languages":{"custom":["/languages/lot.json"]}}
{
  "api_key": "<…>",
  "name": "coreflux-factory-metrics",
  "body": {
    "index_patterns": ["metrics-coreflux-*"],
    "data_stream": {},
    "priority": 500,
    "template": {
      "mappings": { "properties": {
        "@timestamp": { "type": "date" }, "site": { "type": "keyword" },
        "line": { "type": "keyword" }, "temperature": { "type": "float" }
      } }
    }
  }
}
```

Smoke (`data/elasticsearch/cluster/info`): `{ "api_key": "<…>" }`.

> [!TIP]
>
> Batch high-rate telemetry with `_bulk` (a few hundred to a few thousand docs, or about 5 to 15 MB per request) instead of one `_doc` call per sample. Elastic recommends this for indexing throughput.

## Gotchas

* **`_bulk` returns `200` even when items fail.** Check `"errors": true` and each `items[].<action>.status` in the response. A body that doesn't end in a newline is rejected with `400` `The bulk request must be terminated by a newline [\n]`.
* **Data streams are append-only.** Each document needs `@timestamp`. In `_bulk`, use `create`, not `index`. `PUT /{ds}/_doc/{id}` and `_update` are rejected on data streams; use a regular index (such as `factory-assets`) for latest-state documents.
* **Index names** must be lowercase and can't start with `_`, `-` or `+`. Several indices can be comma-separated in `Search`, `CountDocuments` and `RefreshIndex` (for example `factory-telemetry,metrics-coreflux-*`).
* **Near real-time:** new documents become searchable after the next refresh (about 1 s by default; on Serverless refreshes are managed). Call `RefreshIndex` only in tests or after a batch, never per document.
* **Self-managed TLS:** Elasticsearch 8.x and 9.x turn on TLS with a self-signed CA at first start. The broker host must trust that CA (or use a publicly trusted certificate) for `USE_SSL "true"` to work.
* **Serverless** doesn't expose every API. Per Elastic's Serverless OpenAPI, `GET /_cluster/health` is not available there (use `GET /` as the smoke instead), and some index settings such as `number_of_replicas` are managed for you. Indexing, `_bulk`, search, count, `_update`, `_refresh`, ES|QL, index templates and data streams are all available.
* Deletes (documents, indices, data streams, templates), `_delete_by_query`, `_update_by_query`, security and API key management, cluster settings, snapshots, ILM and Kibana APIs are left out on purpose.

## Rate limits

Elasticsearch has no fixed request quota. When the write thread pool queue is full, it returns `429` `es_rejected_execution_exception`, either for the whole request or for individual `_bulk` items. Back off and retry only the rejected items. Elastic Cloud Serverless also applies per-project ingest limits. See [Tune for indexing speed](https://www.elastic.co/docs/deploy-manage/production-guidance/optimize-performance/indexing-speed).

## Testing

**Status: ready (beta).** On 2026-10-04 every event was sent over HTTP, built exactly as the connector builds it (method, endpoint, headers and body substituted from sample MQTT payloads), against a local Elasticsearch 9.5.4 with security on and a scoped API key:

| Event | Result |
| - | - |
| `GetClusterInfo`, `GetClusterHealth` | `200`, cluster `green` |
| `PutIndexTemplate` (data stream template) | `200` `acknowledged` |
| `IndexDocument` into an index and into a data stream | `201` `created` (backing index `.ds-metrics-coreflux-factory-…`) |
| `PutDocument`, `UpdateDocument`, `GetDocument` | `201`, `200` `updated`, `200` `found` with the updated source |
| `BulkIndex` (2 data stream `create` + 1 `index`) | `200` `"errors": false` |
| `RefreshIndex`, `Search`, `CountDocuments` | `200`, 2 hits for `porto`, count 3 |
| `EsqlQuery` (`STATS AVG(temperature) BY site`) | `200`, lisbon 19.0, porto 21.75 |
| `GetDataStream` | `200` |
| Wrong API key | `401` `security_exception` |
| `_bulk` without trailing newline | `400` |
| Index outside the key's scope | `403` `indices:admin/auto_create` unauthorized |

Not tested yet: Elastic Cloud Hosted or Serverless, TLS to a self-managed node, and a broker MQTT round-trip (including how the broker substitutes newlines in `content` for `_bulk`).

## Related

* [REST API Connectors](/latest/lot-language/routes/rest-api-routes)
* [REST API Apps](/latest/lot-language/routes/rest-api-apps/index)
* [ClickHouse](/latest/lot-language/routes/rest-api-apps/clickhouse)
* [InfluxDB](/latest/lot-language/routes/rest-api-apps/influxdb)
* [Splunk Enterprise and Cloud](/latest/lot-language/routes/rest-api-apps/splunk-enterprise-and-cloud)


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