\nIf you read our previous article on storing OpenTelemetry signals in GridDB Cloud with Kafka<\/a>, this architecture will feel familiar: a producer at the edge, a routing\/buffering layer in the middle, specialized processors, and GridDB Cloud as the vault at the end. The difference is that this time there is zero custom code. Where the Kafka pipeline needed a hand-written Go bridge to flatten OTLP payloads, every stage here is an off-the-shelf component wired together with configuration.\n<\/p>\n \nThough this blog uses the GridDB Web API, let’s not forget about this recent release: GridDB Cloud v3.2’s non-WebAPI connection support. If you have not read about that release yet, start here: Connecting to GridDB Cloud v3.2 from Your Local Dev Environment<\/a>.\n<\/p>\n \nThe architecture for this is a bit strange; it’s been separated into four parts:\n<\/p>\n Layer 1 \u2014 Full data stream of mixed data (edge).<\/strong> Devices emit both signal types into one stream: structured JSON metrics and unstructured text logs, interleaved. Each message carries a small label ( Layer 2 \u2014 The router (Azure IoT Hub).<\/strong> IoT Hub’s message routing reads the label and splits the stream: metrics go to one Azure Event Hub, logs go to another. After this point the data is perfectly separated into two independent queues.<\/p>\n Layer 3 \u2014 The specialized processors (Telegraf and Logstash).<\/strong> Because the split already happened, neither tool has to do a job it is bad at. Telegraf pulls from the metrics queue and parses JSON. Logstash pulls from the logs queue (text) and runs grok.<\/p>\n Layer 4 \u2014 Datastore (GridDB Cloud).<\/strong> Both tools have an official GridDB output plugin. Telegraf writes its numbers into an optimized \nThe two Event Hubs in the middle are queues. The Azure IoT Hub pushes into them; Telegraf and Logstash pull from them at their own pace. That decoupling is what makes the pipeline resilient: if Logstash goes down for an hour, log messages simply pile up in \nWe will keep this section brief since the Azure resources are standard. You need:\n<\/p>\n \nLet’s discuss a couple of issues up front: first, IoT Hub routing cannot inspect an arbitrary message body to decide “is this JSON or plain text” \u2014 body-based routing queries only work when the message declares \nFor a data source we will use a small Python script that plays the role of an IoT device. It alternates between two data streams: most of the time it sends a clean JSON metrics payload, and roughly 30% of the time it also emits a raw, syslog-style firmware error line. Crucially, it labels every message with the \nInstall the SDK with \nGet the device connection string with:\n<\/p>\n \nBefore wiring up either consumer, it is worth verifying the split with a throwaway script that reads each event hub directly (the Telegraf<\/a> is InfluxData’s metrics-collection agent. It offers hundreds of input plugins and dozens of output plugins for shipping collected metrics to a time-series store. The GridDB Telegraf plugin is one of those outputs, and conveniently, Telegraf also ships an \nThe GridDB output plugin is not distributed as a prebuilt binary, it must be compiled into Telegraf from source. The v3.2 bundled pack ships the plugin source code, which you place into a Telegraf source checkout before building.\n<\/p>\n \nThis places \nCreate \nThen build and verify:\n<\/p>\n \nIf \nIn our previous walkthrough Telegraf played its traditional role of collecting local<\/em> system metrics. Here it plays a different one: a stream worker that pulls device metrics off an Azure queue. The input section changes; the GridDB output section is identical to before.\n<\/p>\n \nCreate a \nAdjust the \nTwo details worth calling out. \nWith the simulator running, the debug log shows Telegraf writing batches within seconds, and an \nWhere Telegraf handles metrics, Logstash<\/a> handles logs. It ingests unstructured text events, parses them into structured fields with its grok filter, and ships them to a destination of your choice. Our destination is GridDB Cloud, and our source is the \nThe bundle’s README points to a yum-based CentOS install path. On macOS we will download the tarball directly from Elastic \u2014 Homebrew’s \nThe plugin ships as a prebuilt \nThe input side needs no installation at all: the \nSave the following as \nThe grok filter does the core work: it takes a raw line like\n<\/p>\n \nand breaks it into \nThe connection string deserves special attention, because it bit us twice while building this. It must be the complete string \u2014 starting with \nYou will also see a startup warning that no \nStartup takes 20\u201330 seconds. Because of \nThe resulting schema is clean: one self-documenting column per parsed field, and timestamps that reflect when each error actually occurred on the device.\n<\/p>\n \nAt this point the full pipeline is live. Two containers, each shaped by the tool that filled it:\n<\/p>\n \nThe same device produced both, the same IoT Hub received both, and the same database stores both \u2014 but each signal traveled its own lane and was processed by the tool built for it.\n<\/p>\nThe Architecture: Four Layers<\/h2>\n
msgType<\/code>) declaring which kind it is.<\/p>\nTIME_SERIES<\/code> container; Logstash writes its parsed log rows into its own container. One database, two purpose-built containers.<\/p>\n \u250c\u2500(msgType='metric')\u2500\u25ba Event Hub eh-metrics \u2500\u25ba Telegraf \u2500\u25ba iot_sensor_metrics \u2510\r\ndevice \u2500\u2500\u25ba Azure IoT Hub \u2500\u2524 \u251c\u2500 GridDB Cloud\r\n \u2514\u2500(msgType='log')\u2500\u2500\u2500\u2500\u25ba Event Hub eh-logs \u2500\u2500\u2500\u2500\u25ba Logstash \u2500\u25ba iot_firmware_logs \u2518<\/code><\/pre>\n<\/div>\neh-logs<\/code> and get drained when it comes back. Nothing is lost and nothing blocks the devices.\n<\/p>\nSetting Up the Azure Side<\/h2>\n
\n
eh-metrics<\/code> and eh-logs<\/code>.<\/li>\nSend<\/code> rights (used by IoT Hub’s routing endpoints) and one with Listen<\/code> rights (used by Telegraf and Logstash). Keeping them separate is necessary because mixing them up produces a very specific error.<\/li>\nSend<\/code> connection string.<\/li>\n$ az iot hub message-route create -g <resource-group> -n <iot-hub-name> \\\r\n$ --route-name metrics-route --endpoint-name ep-metrics --source devicemessages \\\r\n$ --condition "msgType = 'metric'"\r\n\r\n$ az iot hub message-route create -g <resource-group> -n <iot-hub-name> \\\r\n$ --route-name logs-route --endpoint-name ep-logs --source devicemessages \\\r\n$ --condition "msgType = 'log'"<\/code><\/pre>\n<\/div>\n\n
contentType = application\/json<\/code> and contentEncoding = utf-8<\/code>, and plain-text bodies are opaque to the routing engine entirely. The robust pattern is what we use here: the device stamps an application property (msgType<\/code>) on every message and the routes filter on that property. Second, once you add custom routes, IoT Hub’s fallback route to the built-in endpoint is disabled by default \u2014 any message that matches no route is dropped, so a typo in the property name fails silently. Third, on recent Azure CLI versions, creating an event hub on Basic tier requires passing --cleanup-policy Delete --retention-time 24<\/code> together (the CLI’s defaults request 7-day retention, which Basic rejects, while the retention flag alone trips a serialization error).\n<\/p>\nLayer 1: The Device Simulator<\/h2>\n
msgType<\/code> application property that the routes key on.\n<\/p>\npip install azure-iot-device<\/code>, then:\n<\/p>\nimport json, random, time\r\nfrom datetime import datetime, timezone\r\nfrom azure.iot.device import IoTHubDeviceClient, Message\r\n\r\nCONN = "<your device connection string>"\r\nclient = IoTHubDeviceClient.create_from_connection_string(CONN)\r\n\r\nERRORS = ["E042 sensor read timeout", "E107 fw checksum mismatch", "E019 wifi rssi below threshold"]\r\n\r\nwhile True:\r\n m = Message(json.dumps({\r\n "deviceId": "sensor-001",\r\n "temperature": round(random.uniform(20, 35), 2),\r\n "humidity": round(random.uniform(30, 70), 2),\r\n }))\r\n m.content_type = "application\/json"\r\n m.content_encoding = "utf-8"\r\n m.custom_properties["msgType"] = "metric"\r\n client.send_message(m)\r\n\r\n # occasionally: an ugly text log, labeled for the logs route\r\n if random.random() < 0.3:\r\n line = f'{datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")} sensor-001 ERROR {random.choice(ERRORS)} ip=192.168.1.{random.randint(2,254)}'\r\n l = Message(line)\r\n l.content_type = "text\/plain"\r\n l.content_encoding = "utf-8"\r\n l.custom_properties["msgType"] = "log"\r\n client.send_message(l)\r\n\r\n time.sleep(3)<\/code><\/pre>\n<\/div>\n$ az iot hub device-identity connection-string show -g <resource-group> -n <iot-hub-name> -d sensor-001 -o tsv<\/code><\/pre>\n<\/div>\nazure-eventhub<\/code> pip package makes this a ten-liner). You want to see only<\/em> JSON blobs landing in eh-metrics<\/code> and only<\/em> text lines in eh-logs<\/code>. Debugging the routing and the consumers at the same time is a miserable experience \u2014 confirm the iot hub router is doing its job first.\n<\/p>\nLayer 3a: Telegraf \u2014 The Metrics Lane<\/h2>\n
eventhub_consumer<\/code> input<\/em> plugin whose primary use case is exactly this: consuming from Azure Event Hubs and IoT Hub.<\/p>\nBuilding Telegraf With the GridDB Plugin<\/h3>\n
$ mkdir -p ~\/go\/src\/github.com\/influxdata\r\n$ cd ~\/go\/src\/github.com\/influxdata\r\n$ git clone https:\/\/github.com\/influxdata\/telegraf.git\r\n$ cd telegraf\r\n\r\n# Copy in the plugin source from the v3.2 bundle\r\n$ cp -r \/path\/to\/telegraf-output-plugin\/plugins .\/<\/code><\/pre>\n<\/div>\nplugins\/outputs\/griddb\/griddb.go<\/code> into the Telegraf source tree. An important caveat:<\/strong> having the plugin source alone is not sufficient for Telegraf to recognize it. Modern Telegraf (v1.20+) uses a build-tag registration pattern in which each plugin requires a one-line import file under plugins\/outputs\/all\/<\/code>. Without this file, the plugin compiles into the binary as dead code and Telegraf will reject your configuration with undefined but requested output: griddb<\/code>.\n<\/p>\nplugins\/outputs\/all\/griddb.go<\/code>:\n<\/p>\n\/\/go:build !custom || outputs || outputs.griddb\r\n\r\npackage all\r\n\r\nimport _ "github.com\/influxdata\/telegraf\/plugins\/outputs\/griddb" \/\/ register plugin<\/code><\/pre>\n<\/div>\n$ make telegraf\r\n$ .\/telegraf --output-list | grep griddb<\/code><\/pre>\n<\/div>\ngriddb<\/code> appears in the output list, the plugin has been registered correctly.\n<\/p>\nConfiguring Telegraf as a Stream Consumer<\/h3>\n
griddb-iot.conf<\/code>:\n<\/p>\n$ [[inputs.eventhub_consumer]]\r\n$ ## The Listen connection string for eh-metrics, including EntityPath\r\n$ connection_string = "Endpoint=sb:\/\/<namespace>.servicebus.windows.net\/;SharedAccessKeyName=consumer-listen;SharedAccessKey=<key>;EntityPath=eh-metrics"\r\n$ data_format = "json"\r\n$ json_string_fields = ["deviceId"]\r\n$ ## Use the time IoT Hub received the message as the row timestamp\r\n$ iot_hub_enqueued_time_as_ts = true\r\n$ ## This becomes the GridDB container name\r\n$ name_override = "iot_sensor_metrics"\r\n\r\n$ [[outputs.griddb]]\r\n$ api_url = "https:\/\/cloud8737.griddb.com:443\/griddb\/v2\/gs_clustermfcloud8737\/dbs\/nl7QftSt"\r\n$ database = "${GRIDDB_DATABASE}"\r\n$ cluster_name = "gs_clustermfcloud8737"\r\n$ username = "${GRIDDB_USERNAME}"\r\n$ password = "${GRIDDB_PASSWORD}"\r\n$ update_mode = "append"\r\n$ containers = []\r\n$ is_timeseries = true\r\n$ timestamp_column = "timestamp"\r\n\r\n$ [agent]\r\n$ interval = "10s"\r\n$ flush_interval = "10s"\r\n$ metric_batch_size = 1000\r\n$ metric_buffer_limit = 10000\r\n$ omit_hostname = true\r\n$ debug = true<\/code><\/pre>\n<\/div>\napi_url<\/code>, database<\/code>, cluster_name<\/code>, and username<\/code> values to match your own GridDB Cloud instance, and set is_timeseries = true<\/code> so the plugin creates a TIME_SERIES<\/code> container \u2014 the appropriate choice for metrics data.\n<\/p>\nomit_hostname = true<\/code> matters more here than it did before: without it, Telegraf stamps every row with the hostname of the machine running Telegraf<\/em>, which in an IoT pipeline is misleading \u2014 the data came from sensor-001<\/code>, not from your consumer box, and the deviceId<\/code> field already carries the real source. And note that eventhub_consumer<\/code> is a service input<\/strong>: unlike ordinary polling inputs, it listens for events rather than gathering on an interval, which means the --test<\/code> and --once<\/code> dry-run flags from our previous walkthrough may produce no output for it. Just run it for real:\n<\/p>\n$ export GRIDDB_PASSWORD='your-password'\r\n$ .\/telegraf --config griddb-iot.conf<\/code><\/pre>\n<\/div>\niot_sensor_metrics<\/code> container appears in the GridDB Cloud portal with temperature<\/code>, humidity<\/code>, and deviceId<\/code> columns:\n<\/p>\n![\"\"](\"\/wp-content\/uploads\/2026\/08\/image.png\")
<\/a><\/p>\nLayer 3b: Logstash \u2014 The Logs Lane<\/h2>\n
eh-logs<\/code> queue full of raw firmware error lines.\n<\/p>\nInstalling Logstash<\/h3>\n
elastic\/tap<\/code> formula is currently broken on recent Homebrew versions.\n<\/p>\n$ mkdir -p ~\/logstash-demo && cd ~\/logstash-demo\r\n\r\n$ curl -O https:\/\/artifacts.elastic.co\/downloads\/logstash\/logstash-9.4.1-darwin-aarch64.tar.gz\r\n$ tar -xzf logstash-9.4.1-darwin-aarch64.tar.gz\r\n$ mv logstash-9.4.1 logstash\r\n\r\n$ .\/logstash\/bin\/logstash --version<\/code><\/pre>\n<\/div>\nInstalling the GridDB Output Plugin<\/h3>\n
.gem<\/code>, so no Ruby build step is required. Copy it alongside the Logstash install. Note that the leading .\/<\/code> matters \u2014 without it, logstash-plugin<\/code> treats the argument as a remote plugin name and constructs an invalid URL:\n<\/p>\n$ cp \/path\/to\/logstash-output-plugin\/logstash-output-griddb-1.0.0.gem ~\/logstash-demo\/\r\n\r\n$ cd ~\/logstash-demo\r\n$ .\/logstash\/bin\/logstash-plugin install .\/logstash-output-griddb-1.0.0.gem\r\n$ .\/logstash\/bin\/logstash-plugin list | grep griddb<\/code><\/pre>\n<\/div>\nazure_event_hubs<\/code> input plugin ships bundled with Logstash.\n<\/p>\nThe Logstash Config<\/h3>\n
~\/logstash-demo\/iot-logs-to-griddb.conf<\/code>:\n<\/p>\ninput {\r\n azure_event_hubs {\r\n event_hub_connections => ["Endpoint=sb:\/\/<namespace>.servicebus.windows.net\/;SharedAccessKeyName=consumer-listen;SharedAccessKey=<key>;EntityPath=eh-logs"]\r\n initial_position => "beginning"\r\n }\r\n}\r\n\r\nfilter {\r\n grok {\r\n match => {\r\n "message" => '%{TIMESTAMP_ISO8601:log_ts} %{NOTSPACE:device_id} %{LOGLEVEL:level} %{NOTSPACE:error_code} %{DATA:error_msg} ip=%{IP:device_ip}'\r\n }\r\n }\r\n\r\n date {\r\n match => [ "log_ts", "ISO8601" ]\r\n target => "@timestamp"\r\n }\r\n\r\n mutate {\r\n remove_field => [ "message", "log_ts", "event", "log", "@version", "host" ]\r\n }\r\n}\r\n\r\noutput {\r\n stdout { codec => rubydebug }\r\n\r\n griddb {\r\n host => "https:\/\/cloud8737.griddb.com:443"\r\n cluster => "your-cluster"\r\n database => "your-database"\r\n container => "iot_firmware_logs"\r\n username => "your-username"\r\n password => "${GRIDDB_PASSWORD}"\r\n insert_mode => "append"\r\n }\r\n}<\/code><\/pre>\n<\/div>\n2026-07-16T23:28:33Z sensor-001 ERROR E107 fw checksum mismatch ip=192.168.1.99<\/code><\/pre>\n<\/div>\ndevice_id<\/code>, level<\/code>, error_code<\/code>, error_msg<\/code>, and device_ip<\/code> fields. The date<\/code> filter takes the timestamp embedded inside<\/em> the log line \u2014 the moment the device actually recorded the error \u2014 and uses it as the event’s real timestamp, rather than the moment Logstash happened to process it.\n<\/p>\nEndpoint=sb:\/\/<\/code> and ending with EntityPath=eh-logs<\/code> \u2014 or the plugin fails at startup with Error parsing event hub string name for connection<\/code>. And it must use the Listen<\/strong> authorization rule, not the Send rule you gave to IoT Hub’s routing endpoints. If you paste the Send string, Logstash connects successfully, discovers the partitions, and then fails on every receive with Unauthorized access. 'Listen' claim(s) are required<\/code> \u2014 an error that is easy to miss in the AMQP log spam.\n<\/p>\nstorage_connection_string<\/code> is configured. Logstash uses an Azure Storage account to checkpoint its position across restarts and to coordinate multiple Logstash instances; for a single-instance demo it is safe to ignore, at the cost of re-reading the queue from the beginning on each restart.\n<\/p>\nRunning It<\/h3>\n
$ cd ~\/logstash-demo\r\n$ export GRIDDB_PASSWORD='your-password'\r\n$ .\/logstash\/bin\/logstash -f iot-logs-to-griddb.conf<\/code><\/pre>\n<\/div>\ninitial_position => \"beginning\"<\/code>, Logstash immediately drains everything sitting in eh-logs<\/code>, so you should see a burst of parsed events scroll past in the rubydebug<\/code> output, then a steady trickle as the simulator keeps emitting. Each one becomes a row in the iot_firmware_logs<\/code> container:\n<\/p>\n![\"\"](\"\/wp-content\/uploads\/2026\/08\/image-1.png\")
<\/a><\/p>\nWhat Lands in GridDB Cloud<\/h2>\n
\n
iot_sensor_metrics<\/code><\/strong> \u2014 a TIME_SERIES<\/code> container with one row per metrics message: timestamp<\/code>, deviceId<\/code>, temperature<\/code>, humidity<\/code>. Created and populated by Telegraf.<\/li>\niot_firmware_logs<\/code><\/strong> \u2014 one row per parsed firmware error: timestamp<\/code>, device_id<\/code>, level<\/code>, error_code<\/code>, error_msg<\/code>, device_ip<\/code>. Created and populated by Logstash.<\/li>\n<\/ul>\n
![\"\"](\"\/wp-content\/uploads\/2026\/08\/architecture-diagram.png\")
<\/a><\/p>\n