OpenSearch Route

OpenSearch Overview

The OPENSEARCH route indexes MQTT messages in OpenSearch for full-text search, real-time analytics, and dashboard visualization. It uses the CLEAN query format for simplified document indexing.

OpenSearch is ideal for log aggregation, metrics visualization, and full-text search across IoT data. Connect it to OpenSearch Dashboards for powerful visualizations.

Basic Syntax

DEFINE ROUTE SensorIndex WITH TYPE OPENSEARCH
    ADD OPENSEARCH_CONFIG
        WITH BASE_URL "https://opensearch.example.com:9200"
        WITH USERNAME "admin"
        WITH PASSWORD "secure_password"
        WITH USE_SSL "true"
    ADD EVENT IndexSensorReading
        WITH SOURCE_TOPIC "sensors/+/data"
        WITH QUERY "CLEAN:{index: sensor-data, body: {@timestamp: {timestamp}, sensor: {sensor_id}, reading: {value.json}}}"

Connection Configuration

OPENSEARCH_CONFIG Parameters

Connection

BASE_URL

string

required

OpenSearch endpoint URL including port (e.g., https://opensearch.example.com:9200).

Authentication

USERNAME

string

required

OpenSearch username.

PASSWORD

string

required

OpenSearch password.

SSL/Security

USE_SSL

boolean

Enable HTTPS. Default: false.

IGNORE_CERT_ERRORS

boolean

Ignore certificate validation errors. Default: false.

CLEAN Query Format

OpenSearch routes use the CLEAN format for document indexing:

WITH QUERY "CLEAN:{index: <index_name>, body: {<field>: <value>, ...}}"

CLEAN Placeholders

Placeholder	Description
`{value.json}`	Full JSON payload as embedded object
`{value.json.field}`	Specific field from JSON payload
`{timestamp}`	Message timestamp (use `@timestamp` for Kibana)
`{source_topic}`	Original MQTT topic
`{field}`	Field extracted from topic path

Complete Examples

Basic Indexing
Log Aggregation
Time-Series Index
Self-Signed Certs

Index sensor data for search:

DEFINE ROUTE SensorIndex WITH TYPE OPENSEARCH
    ADD OPENSEARCH_CONFIG
        WITH BASE_URL "https://opensearch.example.com:9200"
        WITH USERNAME "admin"
        WITH PASSWORD "secure_password"
        WITH USE_SSL "true"
    ADD EVENT IndexReading
        WITH SOURCE_TOPIC "sensors/+/data"
        WITH DESTINATION_TOPIC "search/status"
        WITH QUERY "CLEAN:{index: sensor-data, body: {@timestamp: {timestamp}, sensor_id: {sensor_id}, data: {value.json}}}"

Aggregate application logs:

DEFINE ROUTE LogIndex WITH TYPE OPENSEARCH
    ADD OPENSEARCH_CONFIG
        WITH BASE_URL "https://opensearch.example.com:9200"
        WITH USERNAME "admin"
        WITH PASSWORD "admin_password"
        WITH USE_SSL "true"
    ADD EVENT IndexLogs
        WITH SOURCE_TOPIC "logs/+/+"
        WITH QUERY "CLEAN:{index: application-logs, body: {@timestamp: {timestamp}, source: {source_topic}, level: {value.json.level}, message: {value.json.message}}}"

Use date-based index names for time-series:

DEFINE ROUTE TimeSeriesIndex WITH TYPE OPENSEARCH
    ADD OPENSEARCH_CONFIG
        WITH BASE_URL "https://opensearch.example.com:9200"
        WITH USERNAME "admin"
        WITH PASSWORD "secure_password"
        WITH USE_SSL "true"
    ADD EVENT IndexMetrics
        WITH SOURCE_TOPIC "metrics/#"
        WITH QUERY "CLEAN:{index: metrics, body: {@timestamp: {timestamp}, topic: {source_topic}, metrics: {value.json}}}"

Configure Index Lifecycle Management (ILM) in OpenSearch for automatic rollover.

Connect with certificate validation disabled:

DEFINE ROUTE DevIndex WITH TYPE OPENSEARCH
    ADD OPENSEARCH_CONFIG
        WITH BASE_URL "https://localhost:9200"
        WITH USERNAME "admin"
        WITH PASSWORD "admin"
        WITH USE_SSL "true"
        WITH IGNORE_CERT_ERRORS "true"
    ADD EVENT IndexData
        WITH SOURCE_TOPIC "data/#"
        WITH QUERY "CLEAN:{index: dev-data, body: {@timestamp: {timestamp}, topic: {source_topic}, payload: {value.json}}}"

Only use IGNORE_CERT_ERRORS in development environments.

Index Mapping Example

Create an index with explicit mappings for better search and aggregation:

PUT /sensor-data
{
  "mappings": {
    "properties": {
      "@timestamp": { "type": "date" },
      "sensor_id": { "type": "keyword" },
      "data": {
        "properties": {
          "temperature": { "type": "float" },
          "humidity": { "type": "float" }
        }
      }
    }
  }
}

Troubleshooting

Connection Failed

Verify BASE_URL is correct including port
Check OpenSearch is running and accessible
Verify firewall allows connections on port 9200

Authentication Failed

Verify USERNAME and PASSWORD are correct
Check user has permissions to index documents

SSL/Certificate Errors

Ensure USE_SSL matches the endpoint (http vs https)
For self-signed certs, set IGNORE_CERT_ERRORS “true”
Verify certificate chain is complete

Index Errors

Verify index name is valid (lowercase, no special chars)
Check field mappings don’t conflict
Review OpenSearch logs for detailed errors

Next Steps

CrateDB Route

Configure CrateDB for time-series data.

Storage Overview

Compare all storage options.

Quick Start

MQTT Broker

LoT Language

LoT Notebooks

Actions

Routes

Models

Rules

OpenSearch Overview

Basic Syntax

Connection Configuration

OPENSEARCH_CONFIG Parameters

CLEAN Query Format

CLEAN Placeholders

Complete Examples

Index Mapping Example

Troubleshooting

Next Steps

CrateDB Route

Storage Overview

Quick Start

MQTT Broker

LoT Language

LoT Notebooks

Actions

Routes

Models

Rules

​OpenSearch Overview

​Basic Syntax

​Connection Configuration

​OPENSEARCH_CONFIG Parameters

​CLEAN Query Format

​CLEAN Placeholders

​Complete Examples

​Index Mapping Example

​Troubleshooting

​Next Steps

CrateDB Route

Storage Overview

OpenSearch Overview

Basic Syntax

Connection Configuration

OPENSEARCH_CONFIG Parameters

CLEAN Query Format

CLEAN Placeholders

Complete Examples

Index Mapping Example

Troubleshooting

Next Steps