Grafana Loki

Technical Preview

Overview

A client-mode connector for the Grafana Loki HTTP API. Use it to ship broker events (alarms, state changes, machine messages) into Loki as log lines labelled with their MQTT topic, so they sit next to your other logs in Grafana, and to pull LogQL results (error counts per line, the last alarms on a machine) back onto MQTT.

Event	Request	Notes
Ready	`GET /ready`	No tenant or auth. `200` `ready`
BuildInfo	`GET /loki/api/v1/status/buildinfo`	Loki version and revision
Push	`POST /loki/api/v1/push`	JSON streams body. `204` with an empty body
Query	`POST /loki/api/v1/query`	Instant LogQL query (form body), usually a metric query
QueryRange	`POST /loki/api/v1/query_range`	LogQL over a time range (form body): log lines or a metric matrix
Labels	`GET /loki/api/v1/labels?since={since}`	Label names seen in the window
LabelValues	`GET /loki/api/v1/label/{label}/values?since={since}`	Values of one label, for example every `topic`
Series	`POST /loki/api/v1/series`	Streams matching a selector (form body)

Set BASE_ADDRESS to your Loki URL with no trailing slash, for example http://loki.plant.local:3100 (keep USE_SSL "false" for plain HTTP) or https://logs-prod-XXX.grafana.net for Grafana Cloud (set USE_SSL "true"). Topics use analytics/loki/<object>/<verb>; replies land on <topic>/response. See the base guide: REST API Connectors.

Tenant and auth: every /loki/api/v1 event sends X-Scope-OrgID: {value.json.tenant}. Multi-tenant Loki (auth_enabled: true) rejects calls without it (401 no org id) and keeps each tenant’s data separate. Single-tenant Loki ignores the header. Loki OSS has no built-in authentication, so put it behind a gateway or reverse proxy. Grafana Cloud Logs uses Basic auth with your numeric Loki user id and a Cloud access policy token (logs:write, logs:read); set them on the connector with WITH USERNAME and WITH PASSWORD GET SECRET "GRAFANA_CLOUD_LOKI_TOKEN" so the token stays in Hub secrets and out of MQTT payloads. Never commit tokens.

Basic Syntax

DEFINE ROUTE GrafanaLokiRestApi WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH BASE_ADDRESS "http://YOUR_LOKI_HOST:3100"
        WITH ENABLE_CLIENT "true"
        WITH USE_SSL "false"
        WITH REQUEST_TIMEOUT '60'
        WITH DEFAULT_CONTENT_TYPE "application/json"
        # Smoke: GET /ready -> 200 "ready" (no tenant, no auth). Push -> 204 empty body.
    ADD EVENT Ready
        WITH SOURCE_TOPIC "analytics/loki/ready/get"
        WITH DESTINATION_TOPIC "analytics/loki/ready/get/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/ready"
        WITH HEADERS
            ADD "Accept" "text/plain"
    ADD EVENT BuildInfo
        WITH SOURCE_TOPIC "analytics/loki/buildinfo/get"
        WITH DESTINATION_TOPIC "analytics/loki/buildinfo/get/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/loki/api/v1/status/buildinfo"
        WITH HEADERS
            ADD "Accept" "application/json"
    ADD EVENT Push
        WITH SOURCE_TOPIC "analytics/loki/push"
        WITH DESTINATION_TOPIC "analytics/loki/push/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/loki/api/v1/push"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "X-Scope-OrgID" "{value.json.tenant}"
        WITH BODY "{value.json.body}"
    ADD EVENT Query
        WITH SOURCE_TOPIC "analytics/loki/query"
        WITH DESTINATION_TOPIC "analytics/loki/query/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/loki/api/v1/query"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/x-www-form-urlencoded"
            ADD "X-Scope-OrgID" "{value.json.tenant}"
        WITH BODY "query={value.json.query}&limit={value.json.limit}"
    ADD EVENT QueryRange
        WITH SOURCE_TOPIC "analytics/loki/query_range"
        WITH DESTINATION_TOPIC "analytics/loki/query_range/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/loki/api/v1/query_range"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/x-www-form-urlencoded"
            ADD "X-Scope-OrgID" "{value.json.tenant}"
        WITH BODY "query={value.json.query}&since={value.json.since}&limit={value.json.limit}&direction={value.json.direction}"
    ADD EVENT Labels
        WITH SOURCE_TOPIC "analytics/loki/labels/list"
        WITH DESTINATION_TOPIC "analytics/loki/labels/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/loki/api/v1/labels?since={value.json.since}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "X-Scope-OrgID" "{value.json.tenant}"
    ADD EVENT LabelValues
        WITH SOURCE_TOPIC "analytics/loki/labels/values"
        WITH DESTINATION_TOPIC "analytics/loki/labels/values/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/loki/api/v1/label/{value.json.label}/values?since={value.json.since}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "X-Scope-OrgID" "{value.json.tenant}"
    ADD EVENT Series
        WITH SOURCE_TOPIC "analytics/loki/series/list"
        WITH DESTINATION_TOPIC "analytics/loki/series/list/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/loki/api/v1/series"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Content-Type" "application/x-www-form-urlencoded"
            ADD "X-Scope-OrgID" "{value.json.tenant}"
        WITH BODY "match[]={value.json.match}&since={value.json.since}"

MQTT examples

Push broker messages as log lines. Publish to analytics/loki/push:

{"tenant": "plant-porto", "body": {"streams": [{"stream": {"source": "coreflux", "topic": "plant/line1/press01/alarm", "level": "error"}, "values": [["1791158400000000000", "overtemp alarm press01", {"machine": "press01"}]]}]}}

Each stream is one label set; values holds [timestamp, line] pairs, where the timestamp is Unix time in nanoseconds as a string. An optional third element adds structured metadata (high-cardinality fields such as a machine or order id) without creating new streams. Loki answers 204 with an empty body. Keep labels few and low-cardinality (site, line, topic, level) and put values in the line or in structured metadata. Read recent lines. Publish to analytics/loki/query_range:

{"tenant": "plant-porto", "query": "{source=\"coreflux\", topic=~\"plant/line1/.*\"} |= \"overtemp\"", "since": "1h", "limit": "100", "direction": "backward"}

Reply (tested on Loki 3.7.8):

{"status":"success","data":{"resultType":"streams","result":[{"stream":{"detected_level":"error","level":"error","machine":"press01","service_name":"unknown_service","source":"coreflux","topic":"plant/line1/press01/alarm"},"values":[["1791158400000000000","overtemp alarm press01"]]}],"stats":{}}}

Structured metadata (machine) comes back as extra keys in stream, and stats (trimmed here) carries query statistics. Loki 3.x also adds service_name (set to unknown_service when the stream has no service label) and detected_level on ingest. Count alarms per level. Publish a metric query to analytics/loki/query (instant vector) or analytics/loki/query_range (matrix):

{"tenant": "plant-porto", "query": "sum by (level) (count_over_time({source=\"coreflux\"} [5m]))", "limit": "100"}

List the MQTT topics Loki has seen. Publish to analytics/loki/labels/values:

{"tenant": "plant-porto", "label": "topic", "since": "24h"}

Labels takes {"tenant","since"} and Series takes {"tenant","match","since"}, where match is a stream selector such as {source="coreflux"}.

Notes

Query events send application/x-www-form-urlencoded bodies, so LogQL braces, quotes and |= work as written. A query that contains &, + or % must be percent-encoded by whoever builds the payload.
since accepts Loki durations (5m, 1h, 24h). Instead of since you can edit the template to send start and end (Unix nanoseconds or RFC 3339).
Loki rejects lines older than reject_old_samples_max_age (one week by default) and, per stream, lines far out of order; send current timestamps.
Errors: bad LogQL returns 400 with the parse error; a malformed push returns 400; a missing tenant on multi-tenant Loki returns 401 no org id. Grafana Cloud returns 401 for a wrong user id or token.
Not covered: the protobuf/snappy push format, OTLP ingest (/otlp/v1/logs), tail over WebSocket, index stats and volume, log deletion, and ruler / alerting APIs.
Suggested branches: feat/rest-api-route-grafana-loki, feat/docs-rest-api-grafana-loki (Hugo pushes).

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

Overview

Basic Syntax

MQTT examples

Notes

​Overview

​Basic Syntax

​MQTT examples

​Notes

​Related

Overview

Basic Syntax

MQTT examples

Notes

Related