InfluxDB ile IoT Sensör Verisi Toplama ve Yönetimi

Tarafındankomut
0 Comments

IoT projelerinde en çok can sıkan şey şu: sensörlerden veri gelmeye başlıyor, güzel güzel çalışıyor, ama birkaç hafta sonra veriler nerede birikmeli, nasıl sorgulanmalı diye kafa yormaya başlıyorsun. Klasik ilişkisel veritabanlarıyla bu işe girişenler genellikle pişman oluyor çünkü zaman serisi verisi için tasarlanmamış bir sistemi zorluyorsunuz. İşte tam bu noktada InfluxDB devreye giriyor. Bu yazıda gerçek bir IoT sensör veri toplama senaryosu üzerinden InfluxDB kurulumunu, yapılandırmasını ve veri yönetimini ele alacağız.

InfluxDB Neden IoT için Uygun?

Bir fabrikada 50 adet sıcaklık sensörü, 30 adet nem sensörü ve 20 adet basınç sensörü olduğunu düşünelim. Her biri 5 saniyede bir veri gönderiyor. Bu günde yaklaşık 86.400 ölçüm noktası demek, sadece tek bir sensör için. 100 sensörle çarpınca 8.640.000 kayıt oluyor. PostgreSQL ya da MySQL ile bunu yönetmeye çalışmak giderek zorlaşıyor.

InfluxDB’nin zaman serisi verisi için özelleşmiş olması şu avantajları getiriyor:

  • Yüksek yazma hızı: Saniyede on binlerce nokta yazabilir
  • Otomatik veri sıkıştırma: Zaman damgalı veriler için optimize edilmiş sıkıştırma algoritmaları kullanır
  • Retention policy: Eski verileri otomatik siler veya downsample eder
  • Flux sorgu dili: Zaman serisi analizi için güçlü ve esnek
  • Native dashboard: Chronograf ile hazır görselleştirme

Kurulum Ortamının Hazırlanması

Bu örnekte Ubuntu 22.04 üzerinde çalışacağız. Üretim ortamına benzer bir senaryo kurmak için tek sunucu üzerinde InfluxDB, Telegraf (veri toplayıcı) ve örnek bir Python sensör simülatörü çalıştıracağız.

Önce sistemi güncelleyelim ve gerekli araçları kuralım:

sudo apt update && sudo apt upgrade -y
sudo apt install -y curl wget gnupg2 python3 python3-pip

# Python kütüphaneleri
pip3 install influxdb-client paho-mqtt

InfluxDB 2.x Kurulumu

InfluxDB 2.x, 1.x sürümünden önemli ölçüde farklı. Organizasyon ve bucket kavramları geldi, Flux sorgu dili zorunlu hale geldi ve web arayüzü çok daha olgunlaştı. Bu yazıda 2.x ile ilerliyoruz.

# InfluxDB GPG anahtarını ekle
curl -s https://repos.influxdata.com/influxdata-archive_compat.key | 
  gpg --dearmor | 
  sudo tee /etc/apt/trusted.gpg.d/influxdata-archive_compat.gpg > /dev/null

# Repo ekle
echo 'deb [signed-by=/etc/apt/trusted.gpg.d/influxdata-archive_compat.gpg] https://repos.influxdata.com/ubuntu stable main' | 
  sudo tee /etc/apt/sources.list.d/influxdata.list

# Kurulum
sudo apt update
sudo apt install -y influxdb2

# Servisi başlat ve etkinleştir
sudo systemctl start influxdb
sudo systemctl enable influxdb

# Durum kontrolü
sudo systemctl status influxdb

Kurulum sonrası http://sunucu-ip:8086 adresinden web arayüzüne erişebilirsiniz. İlk kurulum sihirbazı kullanıcı adı, şifre, organizasyon adı ve ilk bucket’ı oluşturmanızı istiyor.

Eğer başlangıç yapılandırmasını komut satırından yapmak istiyorsanız:

influx setup 
  --username admin 
  --password Gizli_Sifre_123 
  --org fabrika-iot 
  --bucket sensor-data 
  --retention 30d 
  --force

Bu komut 30 günlük retention policy ile sensor-data adında bir bucket oluşturuyor. Retention parametresi çok önemli; IoT projelerinde disk dolması ciddi bir problem. Veriyi ne kadar tutacağınızı baştan belirlemeniz gerekiyor.

API Token Oluşturma

InfluxDB 2.x token tabanlı kimlik doğrulama kullanıyor. Sensörleriniz ve uygulamalarınız için ayrı tokenlar oluşturmak iyi bir pratik:

# Admin token'ı öğren
influx auth list

# Sadece sensor-data bucket'ına yazma yetkisi olan token oluştur
influx auth create 
  --org fabrika-iot 
  --write-bucket sensor-data 
  --description "Sensor yazma tokeni"

# Okuma yetkili token (dashboard uygulamaları için)
influx auth create 
  --org fabrika-iot 
  --read-bucket sensor-data 
  --description "Dashboard okuma tokeni"

Token’ları bir yere not edin çünkü bir daha göremiyorsunuz.

Telegraf ile Sensör Verisi Toplama

Telegraf, InfluxData’nın veri toplama ajanı. MQTT, HTTP, Modbus ve daha onlarca protokolü destekliyor. Fabrika senaryomuzda sensörler MQTT broker’a veri gönderiyor, Telegraf ise MQTT’yi dinleyip InfluxDB’ye yazıyor.

# Telegraf kurulumu
sudo apt install -y telegraf

# Mosquitto MQTT broker (test için)
sudo apt install -y mosquitto mosquitto-clients
sudo systemctl start mosquitto
sudo systemctl enable mosquitto

Telegraf yapılandırma dosyasını oluşturalım:

sudo nano /etc/telegraf/telegraf.conf
# /etc/telegraf/telegraf.conf

[global_tags]
  datacenter = "istanbul-fabrika-1"
  environment = "production"

[agent]
  interval = "10s"
  round_interval = true
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  flush_interval = "10s"
  flush_jitter = "0s"
  precision = ""
  hostname = ""
  omit_hostname = false

# InfluxDB v2 output
[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "BURAYA_YAZMA_TOKENINIZI_YAZIN"
  organization = "fabrika-iot"
  bucket = "sensor-data"

# MQTT Consumer input
[[inputs.mqtt_consumer]]
  servers = ["tcp://localhost:1883"]
  topics = [
    "fabrika/+/sicaklik",
    "fabrika/+/nem",
    "fabrika/+/basinc"
  ]
  data_format = "json"
  json_time_key = "timestamp"
  json_time_format = "unix"
  tag_keys = ["sensor_id", "konum", "tip"]
# Telegraf'ı yeniden başlat
sudo systemctl restart telegraf
sudo systemctl status telegraf

# Log takibi
sudo journalctl -u telegraf -f

Python ile Sensör Simülatörü

Gerçek sensörler olmadan test yapabilmek için küçük bir simülatör yazalım. Bu aynı zamanda kendi uygulamalarınızdan InfluxDB’ye doğrudan veri nasıl yazılır onu da gösteriyor:

#!/usr/bin/env python3
# sensor_simulator.py

import time
import random
import json
import paho.mqtt.client as mqtt
from datetime import datetime

# MQTT bağlantı ayarları
BROKER = "localhost"
PORT = 1883

# Sensör tanımları
sensors = [
    {"id": "SNS-001", "konum": "uretim-hatti-1", "tip": "sicaklik"},
    {"id": "SNS-002", "konum": "uretim-hatti-1", "tip": "nem"},
    {"id": "SNS-003", "konum": "uretim-hatti-2", "tip": "sicaklik"},
    {"id": "SNS-004", "konum": "depo-1", "tip": "basinc"},
    {"id": "SNS-005", "konum": "depo-1", "tip": "sicaklik"},
]

def generate_value(tip):
    if tip == "sicaklik":
        return round(random.uniform(18.0, 85.0), 2)
    elif tip == "nem":
        return round(random.uniform(30.0, 95.0), 2)
    elif tip == "basinc":
        return round(random.uniform(1.0, 10.0), 3)

client = mqtt.Client()
client.connect(BROKER, PORT, 60)
client.loop_start()

print("Sensör simülatörü başlatıldı...")

while True:
    for sensor in sensors:
        payload = {
            "sensor_id": sensor["id"],
            "konum": sensor["konum"],
            "tip": sensor["tip"],
            "deger": generate_value(sensor["tip"]),
            "timestamp": int(time.time())
        }
        
        topic = f"fabrika/{sensor['id']}/{sensor['tip']}"
        client.publish(topic, json.dumps(payload))
        print(f"Gönderildi: {topic} -> {payload['deger']}")
    
    time.sleep(5)

Bu simülatörü çalıştırdığınızda her 5 saniyede bir MQTT broker’a veri göndermeye başlıyor, Telegraf bunları yakalayıp InfluxDB’ye yazıyor.

Doğrudan Python ile InfluxDB’ye Yazma

Bazen MQTT kullanmak istemeyebilirsiniz ya da Telegraf olmadan doğrudan uygulamanızdan InfluxDB’ye yazmak gerekebilir:

#!/usr/bin/env python3
# direct_influx_writer.py

from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
import time
import random

# Bağlantı ayarları
INFLUX_URL = "http://localhost:8086"
INFLUX_TOKEN = "TOKENINIZ"
INFLUX_ORG = "fabrika-iot"
INFLUX_BUCKET = "sensor-data"

client = InfluxDBClient(
    url=INFLUX_URL,
    token=INFLUX_TOKEN,
    org=INFLUX_ORG
)

write_api = client.write_api(write_options=SYNCHRONOUS)

def write_sensor_data(sensor_id, konum, sicaklik, nem):
    point = (
        Point("ortam_olcumu")
        .tag("sensor_id", sensor_id)
        .tag("konum", konum)
        .field("sicaklik", float(sicaklik))
        .field("nem", float(nem))
        .time(time.time_ns(), WritePrecision.NANOSECONDS)
    )
    
    write_api.write(
        bucket=INFLUX_BUCKET,
        org=INFLUX_ORG,
        record=point
    )

# Test verisi yaz
for i in range(10):
    write_sensor_data(
        sensor_id=f"SNS-00{i % 3 + 1}",
        konum="uretim-hatti-1",
        sicaklik=round(random.uniform(20, 80), 2),
        nem=round(random.uniform(40, 90), 2)
    )
    print(f"Kayıt {i+1} yazıldı")
    time.sleep(1)

client.close()
print("Tamamlandı.")

Flux ile Veri Sorgulama

Flux dili başlangıçta biraz alışılmadık gelebilir ama mantığını kavrayınca çok güçlü. Temel sorgulama örnekleri:

// Son 1 saatin sicaklik verileri
from(bucket: "sensor-data")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "ortam_olcumu")
  |> filter(fn: (r) => r._field == "sicaklik")
  |> filter(fn: (r) => r.konum == "uretim-hatti-1")
  |> sort(columns: ["_time"], desc: true)

// 10 dakikalik ortalamalar (downsample)
from(bucket: "sensor-data")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "ortam_olcumu")
  |> filter(fn: (r) => r._field == "sicaklik")
  |> aggregateWindow(every: 10m, fn: mean, createEmpty: false)

// En yuksek sicaklik deger ve sensörü
from(bucket: "sensor-data")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "ortam_olcumu")
  |> filter(fn: (r) => r._field == "sicaklik")
  |> max()
  |> group()
  |> sort(columns: ["_value"], desc: true)
  |> limit(n: 1)

Bu sorguları influx CLI ile de çalıştırabilirsiniz:

influx query '
from(bucket: "sensor-data")
  |> range(start: -30m)
  |> filter(fn: (r) => r._measurement == "ortam_olcumu")
  |> filter(fn: (r) => r._field == "sicaklik")
  |> mean()
' --org fabrika-iot

Retention Policy ve Veri Yönetimi

IoT verisi hızla büyür. Fabrika senaryomuzda 100 sensörden günde milyonlarca veri noktası geliyor. Her şeyi sonsuza kadar tutmak ne mantıklı ne de ekonomik. InfluxDB’nin retention ve task özellikleri burada devreye giriyor.

Önce mevcut bucket’ımızın retention süresini güncelleyelim:

# Bucket ID'yi öğren
influx bucket list --org fabrika-iot

# Retention'ı 7 güne düşür (ham veri için)
influx bucket update 
  --id BUCKET_ID 
  --retention 168h

Uzun vadeli istatistikleri korumak için downsample task’ı oluşturalım. Ham veriyi 10 dakikalık ortalamalara indirgeyelim ve ayrı bir bucket’a yazalım:

# Önce uzun dönem bucket oluştur
influx bucket create 
  --name sensor-data-aggregated 
  --org fabrika-iot 
  --retention 365d
// InfluxDB Task - downsample_sensors
option task = {
  name: "Sensor Downsample - 10dk Ortalama",
  every: 10m,
  offset: 0m
}

from(bucket: "sensor-data")
  |> range(start: -20m, stop: -10m)
  |> filter(fn: (r) => r._measurement == "ortam_olcumu")
  |> aggregateWindow(every: 10m, fn: mean, createEmpty: false)
  |> set(key: "_measurement", value: "ortam_olcumu_10m_avg")
  |> to(bucket: "sensor-data-aggregated", org: "fabrika-iot")

Bu task’ı web arayüzünden Tasks menüsüne ekleyebilir ya da CLI ile oluşturabilirsiniz.

Alarm ve Uyarı Sistemi

InfluxDB Checks ve Notifications özellikleriyle alarm kurmak mümkün. Sıcaklık 75 dereceyi geçtiğinde uyarı alalım:

# Check oluştur (CLI ile)
influx check create threshold 
  --name "Yuksek Sicaklik Uyarisi" 
  --query 'from(bucket:"sensor-data") |> range(start: -5m) |> filter(fn: (r) => r._measurement == "ortam_olcumu" and r._field == "sicaklik") |> mean()' 
  --every 1m 
  --status-message-template "Sensor ${r.sensor_id} konumunda sicaklik kritik: ${r._value}" 
  --crit-min 75.0 
  --warn-min 65.0 
  --org fabrika-iot

Web arayüzünden Alerts menüsü bu işler için çok daha kullanıcı dostu. Notification endpoint olarak Slack, PagerDuty, email veya HTTP webhook tanımlayabiliyorsunuz.

Yedekleme Stratejisi

Üretim ortamında yedekleme kritik. InfluxDB 2.x’in built-in backup komutu var:

# Tam yedek al
influx backup /opt/influxdb-backups/$(date +%Y%m%d) 
  --org fabrika-iot

# Sadece belirli bir bucket'ı yedekle
influx backup /opt/influxdb-backups/sensor-$(date +%Y%m%d) 
  --bucket sensor-data 
  --org fabrika-iot

# Geri yükle
influx restore /opt/influxdb-backups/20240115 
  --org fabrika-iot

Cron ile günlük yedek:

sudo crontab -e
# Her gece 02:00'da yedek al, 30 günden eski yedekleri sil
0 2 * * * influx backup /opt/influxdb-backups/$(date +%Y%m%d) --org fabrika-iot
0 3 * * * find /opt/influxdb-backups -maxdepth 1 -type d -mtime +30 -exec rm -rf {} ;

Performans Ayarları

Yoğun IoT yazmaları için birkaç önemli ayar var. InfluxDB yapılandırma dosyasını düzenleyelim:

sudo nano /etc/influxdb/config.toml

Dikkat edilmesi gereken parametreler:

  • storage-cache-max-memory-size: Yazma cache boyutu. 8GB RAM’li sunucuda 1gb yapabilirsiniz. Varsayılan 1gb.
  • storage-cache-snapshot-memory-size: Cache’i diske yazmaya başlayacağı eşik. 256mb makul bir değer.
  • storage-wal-fsync-delay: WAL senkron gecikmesi. 0s en güvenli ama performansı etkiler. Yüksek yazma ortamında 100ms denenebilir.
  • http-max-body-size: Batch yazma için maximum body boyutu. IoT batch işlemleri için artırılabilir.
  • query-concurrency: Eş zamanlı sorgu sayısı. CPU çekirdek sayısına göre ayarlayın.
# Değişiklikten sonra servisi yeniden başlat
sudo systemctl restart influxdb

Grafana Entegrasyonu

InfluxDB’nin kendi Chronograf arayüzü var ama çoğu ekip Grafana’yı tercih ediyor çünkü daha esnek ve yaygın kullanılıyor:

# Grafana kurulumu
sudo apt install -y apt-transport-https software-properties-common

wget -q -O - https://packages.grafana.com/gpg.key | sudo apt-key add -
echo "deb https://packages.grafana.com/oss/deb stable main" | 
  sudo tee /etc/apt/sources.list.d/grafana.list

sudo apt update
sudo apt install -y grafana
sudo systemctl start grafana-server
sudo systemctl enable grafana-server

Grafana kurulduktan sonra http://sunucu-ip:3000 adresine gidin. Varsayılan kullanıcı admin/admin. Configuration menüsünden Data Sources’a girin, InfluxDB seçin:

  • Query Language: Flux seçin
  • URL: http://localhost:8086
  • Organization: fabrika-iot
  • Token: Okuma tokenınızı girin
  • Default Bucket: sensor-data

Bağlantıyı test edip kaydettikten sonra dashboard oluşturabilirsiniz. IoT sensör verisi için Grafana’nın stat panel, time series ve heatmap görselleştirmeleri oldukça kullanışlı.

Güvenlik Önlemleri

Üretim ortamında mutlaka yapılması gerekenler:

  • Firewall kuralları: 8086 portu sadece uygulama sunucularından erişilebilir olmalı
  • TLS/SSL: InfluxDB’yi HTTPS ile çalıştırın
  • Token rotasyonu: Tokenları düzenli aralıklarla yenileyin
  • Minimum yetki: Her uygulama sadece ihtiyacı olan bucket’a yazma/okuma yetkisi olan token kullansın
# UFW ile basit firewall kuralı
sudo ufw allow from 192.168.1.0/24 to any port 8086
sudo ufw deny 8086
sudo ufw enable

TLS için InfluxDB config dosyasında:

[http]
  bind-address = ":8086"
  https-enabled = true
  https-certificate = "/etc/ssl/influxdb.crt"
  https-private-key = "/etc/ssl/influxdb.key"

Sonuç

InfluxDB ile IoT veri toplama altyapısı kurmak göründüğü kadar karmaşık değil. Telegraf ile MQTT entegrasyonu sayesinde sensörlerden veri toplamak birkaç satır konfigürasyona indiriniyor. Retention policy ve downsample task’larıyla disk kullanımını kontrol altında tutabiliyorsunuz. Grafana entegrasyonu ise operasyon ekibine gerçek zamanlı görünürlük sağlıyor.

Dikkat etmeniz gereken en kritik noktalar şunlar: baştan doğru bir veri modeli tasarlamak (measurement adları, tag ve field seçimi), retention policy’i projenin gereksinimlerine göre belirlemek ve yedeklemeyi ihmal etmemek. Özellikle tag ve field ayrımı önemli. Sorgularda filtrelemede kullandığınız şeyler tag olmalı, ölçüm değerleri field. Bunu karıştırırsanız hem sorgular yavaşlar hem de gereksiz kardinalite artışıyla karşılaşırsınız.

Küçük ölçekli projeler için tek sunucu bu mimari gayet yeterli. Yüzlerce sensör ve saniyede binlerce yazma söz konusu olduğunda InfluxDB Enterprise veya InfluxDB Cloud seçeneklerini değerlendirin. Açık kaynak sürüm ile başlayıp ihtiyaç büyüdükçe geçiş yapmak da makul bir yol.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir