# mqtt2elasticsearch

[![mqtt-smarthome](https://img.shields.io/badge/mqtt-smarthome-blue.svg)](https://github.com/mqtt-smarthome/mqtt-smarthome)
[![NPM version](https://badge.fury.io/js/mqtt2elasticsearch.svg)](http://badge.fury.io/js/mqtt2elasticsearch)
[![CI](https://github.com/hobbyquaker/mqtt2elasticsearch/actions/workflows/ci.yml/badge.svg)](https://github.com/hobbyquaker/mqtt2elasticsearch/actions/workflows/ci.yml)
[![License][mit-badge]][mit-url]

Index MQTT messages in Elasticsearch — searchable history and Kibana dashboards for everything on
your broker.

Subscribes to the topics you name and writes one document per message, in bulk, into a daily index.
Unlike [influx4mqtt](https://github.com/hobbyquaker/influx4mqtt), which keeps numbers only, this
keeps **every** message: text payloads, JSON, the lot.

Built on [mqtt-interfaces-core](https://github.com/hobbyquaker/mqtt-interfaces-core), so it has the
same options, `--install`, `<name>/info` and maintenance topics as the rest of the `xyz2mqtt` fleet.

Requires **Elasticsearch 7 or newer** (or OpenSearch). Version 1.x of this package wrote ES 6
mapping types and does not work with anything released since 2019.

## Install

```
npm install -g mqtt2elasticsearch
mqtt2elasticsearch -u mqtt://broker -e http://elastic:9200
sudo mqtt2elasticsearch --install -n elastic -u mqtt://broker   # systemd service
```

### Docker

```
docker run -d --name mqtt2elasticsearch --restart unless-stopped \
  -e MQTT2ELASTICSEARCH_MQTT_URL=mqtt://broker \
  -e MQTT2ELASTICSEARCH_ES_URL=http://elastic:9200 \
  -e MQTT2ELASTICSEARCH_SUBSCRIBE='+/status/#' \
  ghcr.io/hobbyquaker/mqtt2elasticsearch
```

## What gets indexed

One document per message, in `<index>-YYYY.MM.DD` (UTC), created from an index template the adapter
writes on start:

```json
{
  "@timestamp": 1699999999000,
  "topic": "hm/status/Wohnzimmer/TEMPERATURE",
  "payload": "{\"val\":21.5,\"ts\":1699999999000}",
  "val_number": 21.5,
  "ts": 1699999999000
}
```

- **`{val, ts, lc}` payloads are unpacked** (`--mqtt-smarthome`, on by default). The value goes in a
  **type-suffixed field** — `val_number`, `val_boolean`, `val_string` — because Elasticsearch fixes
  a field's type on first use, and one shared `val` would take the type of whatever arrived first
  and then reject everything else.
- **The device's `ts` becomes `@timestamp`** when there is one, so a chart shows when a value was
  measured rather than when it was indexed.
- **Retained messages are skipped.** The broker replays them on every reconnect, which would file
  bursts of duplicates under the reconnect time. `--retained-messages` keeps them.
- **Daily indices** make retention a matter of deleting old indices, which is the only way
  Elasticsearch deletes cheaply. Use ILM or a cron job.

## Options

| Option                 | Default                 | Meaning                                              |
| ---------------------- | ----------------------- | ---------------------------------------------------- |
| `-s, --subscribe`      | `#`                     | topic to index, `+`/`#` wildcards; repeat for more   |
| `-u, --mqtt-url`       | `mqtt://localhost`      | broker url                                           |
| `-n, --name`           | `elasticsearch`         | instance name = topic prefix of its own topics       |
| `-e, --es-url`         | `http://localhost:9200` | Elasticsearch base url                               |
| `-i, --index`          | `mqtt`                  | index name prefix; one index per day                 |
| `--es-username`        |                         | http basic auth                                      |
| `--es-password`        |                         | http basic auth                                      |
| `--es-api-key`         |                         | api key, used instead of username/password           |
| `-m, --mqtt-smarthome` | `true`                  | unpack `{val, ts, lc}` into `val_<type>`, `ts`, `lc` |
| `--retained-messages`  | `false`                 | also index retained messages                         |
| `--buf-length`         | `500`                   | write once this many documents are buffered          |
| `--buf-interval`       | `10`                    | seconds between writes                               |
| `-v, --verbosity`      | `info`                  | `error`, `warn`, `info`, `debug`                     |

Every option is also an environment variable (`MQTT2ELASTICSEARCH_SUBSCRIBE`, …); several
subscriptions are comma separated there. Put credentials in the instance's env file rather than on
the command line, where a process list would show them — `--install` does that for you.

## Topics of its own

| Topic                             | Meaning                                                               |
| --------------------------------- | --------------------------------------------------------------------- |
| `<name>/connected`                | `2` Elasticsearch is accepting writes · `1` broker only · `0` stopped |
| `<name>/info`                     | endpoint, index, subscriptions, documents indexed                     |
| `<name>/maintenance/set/loglevel` | `error` \| `warn` \| `info` \| `debug` at runtime                     |
| `<name>/maintenance/set/restart`  | graceful restart                                                      |
| `<name>/maintenance/stats`        | memory, cpu, event loop lag, uptime                                   |

## License

MIT © [Sebastian Raff](https://github.com/hobbyquaker)

[mit-badge]: https://img.shields.io/badge/License-MIT-blue.svg?style=flat
[mit-url]: LICENSE