Telegraf MQTT Consumer: MQTT Verisini InfluxDB'ye Aktarın

Telegraf MQTT consumer eklentisi, kod yazmadan MQTT mesajlarını bir zaman serisi veritabanına taşımanın en basit yoludur. Bu eğitimde inputs.mqtt_consumer girdisini broker.connectmq.com üzerindeki topic'lere subscribe olacak, JSON payload'ları doğru zaman damgaları ve tag'lerle ayrıştıracak, sonucu InfluxDB 2'ye yazacak şekilde yapılandıracak ve her adımı doğrulayacaksınız.

Telegraf Bir MQTT Hattına Nasıl Oturur

Bir MQTT broker mesajları yönlendirir; geçmiş saklamaz. Geçen haftanın sıcaklıklarını grafiğe dökmek veya bir eğilim üzerine alarm üretmek için bir şeyin topic'lere subscribe olması ve her ölçümü bir veritabanına yazması gerekir. Telegraf, InfluxData'nın tam olarak bunu yapan küçük bir ajanıdır. Girdileri, isteğe bağlı işlemcileri ve çıktıları içeren tek bir TOML dosyasıyla yapılandırılır.

mqtt_consumer girdisi bir servis eklentisidir: broker ile kalıcı bir bağlantı tutar, mesajları geldikçe alır, metrik üretmek için bir ayrıştırıcıdan geçirir ve toplu halde yazan çıktılara teslim eder. Broker'ın bakış açısından Telegraf yalnızca başka bir MQTT istemcisidir. Bu, ConnectMQ dahil herhangi bir barındırılan broker için de dürüst desendir: broker mesajları taşır, depolamayla ise Telegraf gibi harici bir araç ilgilenir.

Ön Gereksinimler

Henüz InfluxDB'niz yoksa en hızlı yol Docker'dır:

docker run -d --name influxdb -p 8086:8086 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=iot \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Tüm yığın için tek bir komut mu tercih edersiniz? TIG stack Docker Compose rehberi Telegraf, InfluxDB ve Grafana'yı birlikte çalıştırır.

Topic ve Payload Tasarımı

Bu rehber, Python eğitimlerimizde kullanılan topic yapısını ve JSON payload'ı varsayar:

connectmq/demo/<run-id>/<line>/<device>/telemetry

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

Topic'lerinizin herkese açık broker'da başkalarınınkiyle çakışmaması için kendi rastgele run ID'nizi seçin, örneğin a1b2c3. Elle gönderilen test mesajları yerine sürekli bir akış istiyorsanız, tam olarak bu formatı publish eden Python MQTT cihaz simülatörünü çalıştırın.

Eksiksiz Bir Telegraf MQTT Consumer Yapılandırması

Aşağıdakini telegraf.conf olarak kaydedin:

[agent]
  interval = "10s"
  flush_interval = "10s"
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  omit_hostname = true

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/telemetry"]
  qos = 0
  client_id = "telegraf-a1b2c3"
  connection_timeout = "30s"

  # InfluxDB'deki measurement adı (varsayılan "mqtt_consumer" olurdu)
  name_override = "sensor"
  # Topic'in tamamını bir tag olarak tut; devre dışı bırakmak için "" yapın
  topic_tag = "topic"

  # JSON payload'ları ayrıştır
  data_format = "json"
  tag_keys = ["device"]
  json_time_key = "ts"
  json_time_format = "unix_ms"

  # Üretim hattını topic'ten ek bir tag olarak çıkar
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/telemetry"
    tags = "_/_/_/line/_/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "${INFLUX_TOKEN}"
  organization = "iot"
  bucket = "sensors"

# Geçici: test sırasında metrikleri stdout'a da yazdır
[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

Her ayar ne yapar

• servers, 1883 portunda düz MQTT için tcp:// kullanır. TLS için aşağıda gösterildiği gibi ssl:// ve 8883 portuna geçersiniz.
• topics MQTT wildcard'larını kabul eder. İki + seviyesi herhangi bir hat ve herhangi bir cihazla eşleşirken, sabit telemetry soneki diğer mesaj türlerini dışarıda bırakır.
• client_id açıkça ayarlanmalıdır. Ayarlanmazsa Telegraf her başlangıçta rastgele bir ID üretir; bu da kalıcı oturumları engeller ve istemciyi broker loglarında tanımlamayı zorlaştırır.
• name_override, InfluxDB measurement adını belirler. Bu olmadan her metrik mqtt_consumer adlı bir measurement'a düşer.
• tag_keys, JSON anahtarlarını tag'lere dönüştürür. Tag'ler index'lenir ve cihaz ID'leri gibi filtrelediğiniz veya grupladığınız değerler için uygundur. Sayısal ölçümler otomatik olarak field olur.
• json_time_key ve json_time_format, Telegraf'ın mesajı aldığı zaman yerine cihaz zaman damgasını kullanır. unix_ms milisaniye cinsinden zaman damgalarımızla eşleşir.
• topic_parsing, topic segmentlerini tag'lere eşler. Her _ bir seviyeyi atlar ve adlandırılmış segmentler tag olur; böylece line, topic'in dördüncü seviyesiyle doldurulur.

topic tag'i hakkında bir not: hata ayıklarken kullanışlıdır, ancak her farklı topic InfluxDB'de yeni bir seri oluşturur. Birkaç yüz cihazla bu zararsızdır. On binlerce topic'le veya mesaj ID'leri ya da zaman damgaları içeren topic'lerle seri kardinalitesi hızla büyür ve sorguları yavaşlatır. line ve device tag'leri aynı bilgiyi taşıdığında, topic_tag = "" ayarlamak veritabanını yalın tutar.

Token'ı bir ortam değişkeni olarak ayarlayın ve Telegraf'ı önce test modunda çalıştırın. mqtt_consumer bir servis girdisi olduğu için test modunun mesaj toplamak üzere bir bekleme süresine ihtiyacı vardır:

export INFLUX_TOKEN=my-super-secret-token
telegraf --config telegraf.conf --test --test-wait 20

Bu çalışırken başka bir terminalden bir test mesajı publish edin:

mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-a1b2c3 \
  -t connectmq/demo/a1b2c3/line1/sensor-0001/telemetry \
  -m "{\"device\":\"sensor-0001\",\"ts\":$(date +%s%3N),\"temperature\":21.7,\"humidity\":44.1}"

date +%s%3N çağrısı Linux'ta GNU date ile milisaniye cinsinden bir zaman damgası üretir; macOS'ta coreutils kurup gdate kullanın veya sabit, yakın tarihli bir değer yazın. Telegraf buna benzer bir InfluxDB line protocol satırı yazdırmalıdır:

sensor,device=sensor-0001,line=line1,topic=connectmq/demo/a1b2c3/line1/sensor-0001/telemetry humidity=44.1,temperature=21.7 1760600000123000000

Bu çalıştığında Telegraf'ı telegraf --config telegraf.conf ile normal şekilde başlatın, hatta güvendiğinizde outputs.file bloğunu kaldırın ve verilerin geldiğini doğrulamak için InfluxDB'yi sorgulayın:

influx query 'from(bucket: "sensors") |> range(start: -15m) |> filter(fn: (r) => r._measurement == "sensor") |> limit(n: 5)' \
  --org iot --token my-super-secret-token

Üretim için Bağlantıyı Güvenceye Almak

Herkese açık broker yalnızca denemeler içindir; üzerindeki her şey başkaları tarafından okunabilir. Gerçek veriler için kimlik bilgileri ve TLS içeren özel bir broker kullanın. ConnectMQ gibi tam yönetilen bir MQTT hizmetiyle girdi bloğu şu şekilde değişir:

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["connectmq/factory1/production/+/+/telemetry"]
  qos = 1
  client_id = "YOUR_CLIENT_ID"
  username = "${MQTT_USERNAME}"
  password = "${MQTT_PASSWORD}"
  persistent_session = true
  max_undelivered_messages = 1000

  name_override = "sensor"
  data_format = "json"
  tag_keys = ["device"]
  json_time_key = "ts"
  json_time_format = "unix_ms"

Broker, işletim sisteminin CA deposu tarafından güvenilen bir Let's Encrypt sertifikası sunar; bu yüzden tls_ca ayarına gerek yoktur. QoS 1 ve sabit bir client ID ile birleştirilen persistent_session = true, broker'dan Telegraf'ın oturumunu tutmasını ve kısa Telegraf yeniden başlatmaları sırasında mesajları kuyruğa almasını ister. Kimlik bilgilerini, özellikle dosya sürüm kontrolünde duruyorsa, yapılandırma dosyası yerine ortam değişkenlerinde tutun.

Telegraf MQTT Consumer Sorunlarını Giderme

Bağlantı olaylarını ve ayrıştırıcı hatalarını gerçekleştikleri anda görmek için Telegraf'ı --debug ile çalıştırın. İç içe nesneler veya diziler içeren payload'larda eski JSON ayrıştırıcısı hızla sınırlarına ulaşır; Telegraf json_v2 ayrıştırıcı rehberi iç içe yapıların ve topic segmentlerinin nasıl hassas şekilde eşleneceğini gösterir. Seçeneklerin tam listesi resmi mqtt_consumer eklenti dokümantasyonunda yer alır.

Sırada Ne Var

Veriler InfluxDB'ye akmaya başladığında paneller ve alarmlar için Grafana ekleyin. Yapılandırma yerine kodu tercih ediyorsanız, Python InfluxDB köprüsü aynı işi eşleme üzerinde tam kontrolle kısa bir betikte yapar. Cihazların kendisi için hâlâ istemci araçları seçiyorsanız MQTT istemci kütüphaneleri rehberine bakın.

ConnectMQ ile Başlayın

Bu yapılandırmanın tamamını birkaç dakika içinde herkese açık MQTT test broker'ına karşı test edebilirsiniz. Üretim verileri için ConnectMQ size 8883 portunda TLS, MQTT 5 ve Telegraf'ın ne alacağını kontrol etmek için canlı bir topic gezgini içeren bir portalla özel bir broker sunar. Kredi kartı olmadan app.connectmq.com adresinde ücretsiz bir Serverless broker oluşturun, ardından üretilen kimlik bilgilerini Telegraf yapılandırmanıza yapıştırın. Daha büyük bir dağıtım planlıyorsanız bulutta MQTT broker çalıştırma hakkında daha fazlasını okuyun.

Sık Sorulan Sorular

Telegraf MQTT consumer, MQTT 5'i destekliyor mu?

Güncel Telegraf sürümleri MQTT 5'i seçmek için bir protocol seçeneği ekler. Varsayılan MQTT 3.1.1'dir ve çoğu kurulumda subscribe olmak ve metrik yazmak için yeterlidir.

Tek bir Telegraf örneği birden fazla topic'e subscribe olabilir mi?

Evet. topics içinde birden fazla filtre listeleyin veya topic'ler farklı payload formatları taşıdığında farklı ayrıştırıcılarla birkaç mqtt_consumer bloğu ekleyin.

JSON string'lerim neden saklanmıyor?

Eski json ayrıştırıcısı, string değerleri json_string_fields veya tag_keys içinde listelemediğiniz sürece yok sayar. json_v2 ayrıştırıcısı string'leri açıkça işler.

Telegraf QoS 0 mı yoksa QoS 1 mi kullanmalı?

QoS 0, ara sıra kaybolan bir ölçümün önemli olmadığı yüksek hızlı telemetri için yeterlidir. Her ölçümün saklanması gerektiğinde kalıcı bir oturumla QoS 1 kullanın.