Telegraf ile MQTT'den InfluxDB'ye: mqtt_consumer Kurulum Rehberi

Bir MQTT broker'dan zaman serisi veritabanına giden en kısa yol Telegraf'tır: mqtt_consumer input'u topic'lerinize abone olur, bir parser her payload'ı metriğe çevirir ve influxdb_v2 output'u bunu InfluxDB'ye yazar. Kod yok, tek bir yapılandırma dosyası var; veritabanının yanında ya da bir edge gateway üzerinde rahatça çalışır.

Bu rehberde Docker Compose ile çalışan bir pipeline kuracak, JSON payload'ları json_v2 ile ayrıştıracak, topic'in parçalarını tag'lere eşleyecek ve gerçek cihazları bağlamadan önce her aşamanın çalıştığını nasıl kanıtlayacağınızı göstereceğiz.

Pipeline'a genel bakış

Örnek cihaz testmqtt/telegraf/sensor-01/telemetry topic'ine şu tür bir JSON yayınlar:

{"ts": 1760700000000, "site": "lab", "temperature": 21.4, "humidity": 40.2, "status": "ok"}

InfluxDB ve Telegraf'ı başlatın

Bu Compose dosyası, başlangıç organizasyonu, bucket'ı ve token'ı ile InfluxDB 2.x'i ve yapılandırmanız salt okunur bağlanmış hâlde Telegraf'ı başlatır:

services:
  influxdb:
    image: influxdb:2.7
    ports: ["8086:8086"]
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_PASSWORD: change-me-please
      DOCKER_INFLUXDB_INIT_ORG: testmqtt
      DOCKER_INFLUXDB_INIT_BUCKET: mqtt
      DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: dev-token-change-me
    volumes:
      - influx-data:/var/lib/influxdb2

  telegraf:
    image: telegraf:1.32
    depends_on: [influxdb]
    environment:
      MQTT_USER: ${MQTT_USER}
      MQTT_PASS: ${MQTT_PASS}
      INFLUX_TOKEN: dev-token-change-me
    volumes:
      - ./telegraf.conf:/etc/telegraf/telegraf.conf:ro

volumes:
  influx-data:

Init değişkenleri yalnızca boş bir volume ile yapılan ilk açılışta uygulanır. Üretimde admin token'ını yeniden kullanmak yerine Telegraf için yalnızca yazma yetkisi olan ayrı bir token oluşturun. Broker ve dashboard'ları da içeren daha büyük bir yerel stack için MQTT Docker Compose stack yazısına bakın.

mqtt_consumer'ı yapılandırın

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

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["testmqtt/telegraf/+/telemetry"]
  qos = 1
  client_id = "telegraf-ingest-01"
  persistent_session = true
  username = "${MQTT_USER}"
  password = "${MQTT_PASS}"
  max_undelivered_messages = 1000
  topic_tag = ""
  data_format = "json_v2"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "testmqtt/telegraf/+/telemetry"
    tags = "_/_/device/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "telemetry"
    timestamp_path = "ts"
    timestamp_format = "unix_ms"

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "site"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "temperature"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "humidity"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "status"
      type = "string"

[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "${INFLUX_TOKEN}"
  organization = "testmqtt"
  bucket = "mqtt"

En önemli ayarlar:

• servers: şifresiz için tcp://host:1883, TLS için ssl://host:8883. Telegraf sunucu sertifikasını sistemin CA deposuna göre doğrular; özel bir CA için tls_ca = "/path/ca.pem" ekleyin.
• topics: standart MQTT filtreleri; yani + ve # çalışır. Bir filtreyi gerçek topic adlarıyla topic matcher üzerinde test edin. Filtreler hakkında arka plan: topic'ler ve wildcard'lar.
• qos = 1 + persistent_session = true + sabit bir client_id: broker Telegraf için bir oturum tutar ve Telegraf yeniden başlarken QoS 1 mesajlarını kuyruğa alır. Sabit bir client ID olmadan oturum devam ettirilemez. Bkz. kalıcı oturumlar.
• max_undelivered_messages: Telegraf'ın, output'a yazılmadan önce okuduğu QoS 1/2 mesaj sayısı. Bu, Telegraf'ın backpressure mekanizmasıdır: InfluxDB yavaşlarsa Telegraf onay göndermeyi bırakır ve kalanını broker tutar.
• topic_tag = "": varsayılan topic tag'ini kapatır. Topic'in tamamını tag olarak saklamak seri kardinalitesini katlar; topic_parsing yalnızca ihtiyacınız olan kısmı çıkarır.

json_v2 payload'ı nasıl eşler

Yukarıdaki yapılandırmayla örnek payload şu line protocol'e dönüşür:

telemetry,device=sensor-01,site=lab temperature=21.4,humidity=40.2,status="ok" 1760700000000000000

InfluxDB'yi sağlıklı tutan iki kural var: filtrelediğiniz ve az sayıda farklı değeri olan şeyler için tag, ölçümler için field. Sayıları da her zaman JSON sayısı olarak gönderin: tırnak içindeki "21.4" bir metindir; ya atılır ya da field üzerinde tür çakışması yaratır. Payload'lar bazen bozuk geliyorsa MQTT JSON ayrıştırma hataları yazısını okuyun.

Cihazlar zaman damgası göndermiyorsa timestamp_path satırını kaldırın; Telegraf varış zamanını kullanır. Canlı veri için bu yeterlidir, ancak çevrimdışı kalmış cihazlardan sonradan gelen veriler geç bir zamanla damgalanır.

Her aşamayı doğrulayın

Pipeline'ı verinin aktığı sırayla test edin; böylece hangi aşamanın sorunlu olduğunu her zaman bilirsiniz.

1. Veri broker'da mı? Online MQTT istemcisi ile testmqtt/telegraf/# topic'ine abone olun ve bir test ölçümü yayınlayın:

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/telegraf/sensor-01/telemetry \
  -m "{\"ts\":$(date +%s000),\"site\":\"lab\",\"temperature\":21.4,\"humidity\":40.2,\"status\":\"ok\"}"

2. Telegraf ayrıştırıyor mu? Telegraf'ı test modunda çalıştırın. Metrikleri yazmak yerine stdout'a basar; --test-wait ise MQTT input'unu mesaj alabilecek kadar uzun süre dinlemede tutar:

docker compose run --rm telegraf telegraf --config /etc/telegraf/telegraf.conf --test --test-wait 15

Beklerken yeniden yayın yapın. telemetry,device=sensor-01,... satırını görmelisiniz. Hiç satır yoksa abonelik veya parser hatalıdır; eksik alanlı bir satır varsa json_v2 path'leri veya türleri uyuşmuyordur.

3. InfluxDB'de mi? Stack'i docker compose up -d ile başlatın ve bucket'ı sorgulayın:

docker compose exec influxdb influx query --org testmqtt \
  'from(bucket: "mqtt") |> range(start: -10m) |> filter(fn: (r) => r._measurement == "telemetry") |> last()'

Telemetri topic'ini tarayıcıda izleyin

Sık görülen hatalar

Broker tarafındaki reason code'ları reason code aracıyla sorgulayın. Parser hataları E! [inputs.mqtt_consumer] önekiyle loglanır ve genellikle payload'ın geçerli JSON olmadığı ya da tanımlı bir alanın farklı türde geldiği anlamına gelir. Bağlantı ve flush ayrıntılarını görmek için Telegraf'ı --debug ile çalıştırın.

Ölçekleme ve saklama süresi

• Daha fazla throughput: Telegraf instance başına çok sayıda mesaj işler; birden fazla instance gerekiyorsa broker'ın akışı çoğaltmak yerine bölmesi için $share/telegraf/testmqtt/telegraf/+/telemetry gibi bir paylaşımlı abonelik kullanın. Her instance kendi client_id değerine ihtiyaç duyar.
• Saklama süresi: bucket üzerinde bir retention süresi tanımlayın (arayüzden veya influx bucket update --id <bucket-id> --retention 30d ile) ve eski verileri bir InfluxDB task'ı ile seyreltin. Ham telemetrinin ne kadar süre saklanacağını planlamak MQTT veri saklama ve depolama yazısında anlatılıyor.
• Görselleştirme: geçmiş için Grafana'yı InfluxDB'ye bağlayın; doğrudan broker'dan canlı bir panel de istiyorsanız Grafana MQTT data source ekleyin.

Telegraf, "MQTT'den InfluxDB'ye" işini bir yazılım projesi olmaktan çıkarıp bir yapılandırma incelemesine dönüştürür. Tag'lerinizi dikkatle seçin, kendi broker'ınızda kalıcı oturumla QoS 1 kullanın, canlıya almadan önce --test ile doğrulayın ve yapılandırmayı dashboard'larınızla birlikte sürüm kontrolünde tutun.

Sıkça sorulan sorular