Elasticsearch and Elastic Cloud

Technical Preview

Overview

A client-mode connector for the Elasticsearch REST API that writes MQTT data into Elasticsearch and reads it back:

Catalog connector event	Endpoint	Notes
Index a document	`POST /{index}/_doc`	Index API. Auto-generated `_id`, returns `201` with `"result":"created"`. Works for data streams too.
Bulk-load a batch	`POST /_bulk`	Bulk API. 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 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, 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.
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.
Scope the key to what the connector writes. A key like the one used in testing:

{
  "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

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

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

{
  "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:

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

{ "api_key": "<…>", "index": "factory-assets", "id": "press-3", "body": { "asset": "press-3", "state": "idle", "rpm": 0 } }

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

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

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

{
  "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": "<…>" }.

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.

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

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

Overview

Auth

Basic Syntax

Example MQTT payloads

Gotchas

Rate limits

Testing

​Overview

​Auth

​Basic Syntax

​Example MQTT payloads

​Gotchas

​Rate limits

​Testing

​Related

Overview

Auth

Basic Syntax

Example MQTT payloads

Gotchas

Rate limits

Testing

Related