InfluxDB dan Telegraf untuk IoT

1. Apa Itu InfluxDB & Telegraf?

InfluxDB adalah database time-series open-source yang dioptimalkan untuk menyimpan dan meng-query data yang berubah seiring waktu — seperti data sensor IoT, metrik server, dan log aplikasi. Dikembangkan oleh InfluxData, InfluxDB menjadi salah satu pilihan paling populer untuk time-series data berkat performa tinggi dan kemudahan penggunaan.

Telegraf adalah agent pengumpul data yang ringan (lightweight) yang berjalan di edge device atau server. Telegraf mengumpulkan metrik dari berbagai sumber (sensor, sistem, aplikasi), mengubah formatnya, dan mengirim ke InfluxDB atau destination lainnya. Dengan 300+ input plugins, Telegraf bisa mengumpulkan data dari hampir semua sumber.

Fitur Utama InfluxDB

Fitur	Deskripsi
Time-Series Optimized	Penyimpanan dan indexing khusus untuk data berbasis waktu
High Throughput	Mampu menulis jutaan data points per detik
InfluxQL & Flux	Dua bahasa query yang powerful untuk analisis data
Retention Policies	Auto-delete data lama berdasarkan durasi
Continuous Queries	Downsampling otomatis untuk data historis
Built-in HTTP API	REST API untuk read/write data
Clustering	High availability (InfluxDB Enterprise / Cloud)

InfluxDB 2.x vs 1.x

Aspek	InfluxDB 1.x	InfluxDB 2.x
Query Language	InfluxQL	Flux + InfluxQL
Dashboard	Chronograf terpisah	Built-in UI
Alerting	Kapacitor terpisah	Built-in alert rules
API	REST v1	API v2 + compatibility v1
Auth	User-based	Token-based
Storage Engine	TSM (InMemory + WAL)	TSM (sama)

2. Arsitektur & Komponen

Arsitektur InfluxDB + Telegraf untuk IoT terdiri dari tiga layer utama: Collection Layer (Telegraf), Storage Layer (InfluxDB), dan Visualization Layer (Grafana atau InfluxDB UI).

Data Flow IoT

Telegraf Architecture

# Arsitektur Telegraf Pipeline:
# Input Plugins → Processor Plugins → Aggregator Plugins → Output Plugins

# Input: Mengumpulkan data dari berbagai sumber
# - MQTT Consumer (dari sensor)
# - HTTP Listener (webhook)
# - SNMP (perangkat jaringan)
# - System metrics (CPU, RAM, disk)
# - GPIO (langsung dari Raspberry Pi)

# Processor: Transform data sebelum dikirim
# - rename: rename measurement/field
# - converter: tipe data conversion
# - regex: filter/transform berdasarkan regex
# - starlark: Python-like scripting

# Aggregator: Hitung statistik dari window data
# - min, max, mean, count
# - histogram, quantile
# - derivative (rate of change)

# Output: Kirim data ke destination
# - influxdb_v2: ke InfluxDB
# - mqtt: publish ke MQTT broker
# - file: tulis ke file
# - kafka: publish ke Kafka

InfluxDB Data Model

# InfluxDB data model:
# Measurement: kumpulan data points (tabel)
# Tag: indexed metadata (untuk filtering) — string
# Field: nilai data (tidak di-index) — float/int/string/boolean
# Timestamp: waktu data point

# Contoh data point:
# measurement,tag1=val1,tag2=val2 field1=1.0,field2=2i,field3="ok" timestamp

# Example sensor data:
# temperature,device=sensor-01,location=warehouse-A value=25.6,humidity=65 1719600000000000000
# power,device=meter-01,building=B3 current=15.2,voltage=220.5,power=3351.6 1719600000000000000

3. Instalasi InfluxDB & Telegraf

Docker Compose Setup

# docker-compose.yml
version: "3.8"
services:
  influxdb:
    image: influxdb:2.7
    ports:
      - "8086:8086"
    volumes:
      - influxdb-data:/var/lib/influxdb2
      - influxdb-config:/etc/influxdb2
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=admin
      - DOCKER_INFLUXDB_INIT_PASSWORD=admin12345
      - DOCKER_INFLUXDB_INIT_ORG=iot-org
      - DOCKER_INFLUXDB_INIT_BUCKET=iot-data
      - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-token
    restart: unless-stopped

  telegraf:
    image: telegraf:1.28
    volumes:
      - ./telegraf.conf:/etc/telegraf/telegraf.conf:ro
    depends_on:
      - influxdb
    restart: unless-stopped

  grafana:
    image: grafana/grafana:11.0.0
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin123
    depends_on:
      - influxdb
    restart: unless-stopped

volumes:
  influxdb-data:
  influxdb-config:
  grafana-data:

Konfigurasi Telegraf

# telegraf.conf
[global_tags]
  environment = "production"
  site = "warehouse-A"

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

# Output ke InfluxDB v2
[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "my-super-token"
  organization = "iot-org"
  bucket = "iot-data"

# Input: MQTT Consumer — terima data dari sensor
[[inputs.mqtt_consumer]]
  servers = ["tcp://mqtt-broker:1883"]
  topics = ["sensors/+/data"]
  data_format = "json"
  json_name_key = "measurement"
  tag_keys = ["device_id", "location"]

# Input: System metrics
[[inputs.cpu]]
  percpu = true
  totalcpu = true
  collect_cpu_time = false

[[inputs.mem]]
[[inputs.disk]]
  ignore_fs = ["tmpfs", "devtmpfs", "devfs", "iso9660", "overlay", "aufs", "squashfs"]
[[inputs.net]]
  interfaces = ["eth*", "wlan*"]
[[inputs.system]]

# Input: HTTP Response monitoring
[[inputs.http_response]]
  urls = ["http://iot-api:8080/health"]
  response_timeout = "5s"
  method = "GET"

4. InfluxDB Line Protocol

Line Protocol adalah format text bawaan InfluxDB untuk menulis data. Setiap baris mewakili satu data point. Pemahaman line protocol sangat penting untuk menulis data langsung ke InfluxDB.

Syntax Line Protocol

# Format:
# <measurement>[,<tag_key>=<tag_value>] <field_key>=<field_value>[,<field_key2>=<field_value2>] <timestamp>

# Contoh dasar:
temperature,device=sensor-01,location=room-A value=25.6 1719600000000000000

# Multiple fields:
weather,station=jakarta temperature=32.5,humidity=78,pressure=1013.25 1719600000000000000

# Tanpa timestamp (akan menggunakan server timestamp):
energy,meter=bldg-01 voltage=220.5,current=15.2,power=3351.6

# Tipe data:
# 25.6   → float (default)
# 25i    → integer (akhiri dengan i)
# true   → boolean
# "text" → string (dalam kutip)

Menulis Data via API

# Write single data point via curl
curl -X POST "http://localhost:8086/api/v2/write?org=iot-org&bucket=iot-data&precision=ns"   -H "Authorization: Token my-super-token"   -H "Content-Type: text/plain"   -d "temperature,device=sensor-01 value=25.6"

# Write multiple data points (newline separated)
curl -X POST "http://localhost:8086/api/v2/write?org=iot-org&bucket=iot-data&precision=ns"   -H "Authorization: Token my-super-token"   -H "Content-Type: text/plain"   -d "temperature,device=sensor-01 value=25.6
temperature,device=sensor-02 value=24.8
humidity,device=sensor-01 value=65"

# Dari Python
# pip install influxdb-client
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS

client = InfluxDBClient(url="http://localhost:8086", token="my-super-token", org="iot-org")
write_api = client.write_api(write_options=SYNCHRONOUS)

point = Point("temperature") \
    .tag("device", "sensor-01") \
    .tag("location", "warehouse-A") \
    .field("value", 25.6) \
    .field("battery", 87)

write_api.write(bucket="iot-data", record=point)

Query Data dengan InfluxQL & Flux

# InfluxQL (v1 compatibility)
SELECT mean("value") FROM "temperature" WHERE time > now() - 1h GROUP BY time(5m), "device"

# Flux query (v2 native)
from(bucket: "iot-data")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "temperature")
  |> filter(fn: (r) => r.device == "sensor-01")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
  |> yield(name: "mean")

# Top 10 hottest sensors dalam 24 jam terakhir
from(bucket: "iot-data")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "temperature")
  |> filter(fn: (r) => r._field == "value")
  |> group(columns: ["device"])
  |> mean()
  |> sort(columns: ["_value"], desc: true)
  |> limit(n: 10)

5. Retention Policies & Downsampling

Retention policies menentukan berapa lama data disimpan. Untuk data IoT, data detail (1 detik interval) mungkin hanya perlu disimpan 7 hari, sementara data yang sudah di-downsample (5 menit interval) disimpan lebih lama.

Membuat Bucket dengan Retention

# Via InfluxDB CLI
influx bucket create \
  --name iot-raw \
  --org iot-org \
  --retention 7d \
  --description "Raw sensor data, 7-day retention"

influx bucket create \
  --name iot-hourly \
  --org iot-org \
  --retention 90d \
  --description "Hourly aggregates, 90-day retention"

influx bucket create \
  --name iot-yearly \
  --org iot-org \
  --retention 365d \
  --description "Daily aggregates, 1-year retention"

# Via InfluxDB UI:
# 1. Buka InfluxDB UI → Load Data → Buckets
# 2. Klik "Create Bucket"
# 3. Set nama dan retention period
# 4. Simpan

Strategi Downsampling

Bucket	Interval	Retensi	Kegunaan
iot-raw	10 detik	7 hari	Debugging, analisis detail
iot-5min	5 menit	30 hari	Monitoring harian
iot-hourly	1 jam	90 hari	Trend analysis
iot-daily	1 hari	365 hari	Reporting bulanan

6. Continuous Queries & Tasks

Di InfluxDB 1.x, fitur ini disebut Continuous Queries (CQ). Di InfluxDB 2.x, evolusinya menjadi Tasks yang menggunakan Flux script. Tasks memungkinkan kamu menjalankan query secara periodik untuk downsampling, aggregasi, dan transformasi data.

Membuat Task untuk Downsampling

// Task: Downsample temperature data setiap 5 menit
// Jalankan setiap 5 menit

option task = {name: "downsample-5min", every: 5m}

from(bucket: "iot-raw")
  |> range(start: -task.every)
  |> filter(fn: (r) => r._measurement == "temperature")
  |> filter(fn: (r) => r._field == "value")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
  |> set(key: "_measurement", value: "temperature_5m")
  |> to(bucket: "iot-5min", org: "iot-org")

// Task: Hitung statistik harian
option task = {name: "daily-stats", every: 1d, offset: 1h}

from(bucket: "iot-5min")
  |> range(start: -1d)
  |> filter(fn: (r) => r._measurement == "temperature_5m")
  |> group(columns: ["device"])
  |> reduce(
      identity: {min: 999.0, max: -999.0, sum: 0.0, count: 0},
      fn: (r, accumulator) => ({
        min: if r._value < accumulator.min then r._value else accumulator.min,
        max: if r._value > accumulator.max then r._value else accumulator.max,
        sum: accumulator.sum + r._value,
        count: accumulator.count + 1
      })
    )
  |> map(fn: (r) => ({
    r with
    _measurement: "daily_stats",
    mean: r.sum / float(v: r.count)
  }))
  |> to(bucket: "iot-yearly", org: "iot-org")

Membuat Task dari UI

1. Buka InfluxDB UI → Tasks
2. Klik Create Task
3. Berikan nama dan atur interval
4. Tulis Flux script di editor
5. Klik Save
6. Monitor eksekusi di tab Task Runs

7. Telegraf Input Plugins untuk IoT

Telegraf memiliki ratusan input plugin yang bisa mengumpulkan data dari berbagai sumber. Berikut plugin yang paling relevan untuk IoT:

MQTT Consumer Plugin

# Menerima data dari MQTT broker
[[inputs.mqtt_consumer]]
  servers = ["tcp://mqtt-broker:1883"]
  topics = [
    "sensors/temperature/+",
    "sensors/humidity/+",
    "devices/+/telemetry"
  ]
  topic_tag = "topic"
  data_format = "json_v2"
  
  [[inputs.mqtt_consumer.json_v2]]
    measurement_name_path = "measurement"
    [[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.tag]]
      path = "device_id"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "location"

GPIO Plugin (Raspberry Pi)

# Baca digital input dari GPIO Raspberry Pi
[[inputs.gpio]]
  files = ["/sys/class/gpio/gpio17/value", "/sys/class/gpio/gpio27/value"]
  name_override = "gpio_input"
  tag_keys = ["file"]

# Baca dari I2C sensor
[[inputs.exec]]
  commands = ["python3 /opt/scripts/read_i2c_sensor.py"]
  timeout = "10s"
  data_format = "json"
  name_override = "i2c_sensor"

SNMP Plugin

# Monitoring perangkat jaringan via SNMP
[[inputs.snmp]]
  agents = ["192.168.1.1", "192.168.1.2"]
  version = 2
  community = "public"
  interval = "30s"

  [[inputs.snmp.field]]
    name = "hostname"
    oid = "RFC1213-MIB::sysName.0"
    is_tag = true

  [[inputs.snmp.field]]
    name = "uptime"
    oid = "RFC1213-MIB::sysUpTime.0"

  [[inputs.snmp.table]]
    name = "interface"
    inherit_tags = ["hostname"]
    [[inputs.snmp.table.field]]
      name = "ifIndex"
      oid = "IF-MIB::ifIndex"
    [[inputs.snmp.table.field]]
      name = "ifDescr"
      oid = "IF-MIB::ifDescr"
    [[inputs.snmp.table.field]]
      name = "ifHCInOctets"
      oid = "IF-MIB::ifHCInOctets"
    [[inputs.snmp.table.field]]
      name = "ifHCOutOctets"
      oid = "IF-MIB::ifHCOutOctets"

HTTP Listener Plugin

# Terima data via HTTP webhook
[[inputs.http_listener_v2]]
  service_address = ":8080"
  path = "/telegraf"
  methods = ["POST"]
  data_format = "json_v2"
  
  [[inputs.http_listener_v2.json_v2]]
    measurement_name_path = "sensor_type"
    [[inputs.http_listener_v2.json_v2.field]]
      path = "value"
      type = "float"
    [[inputs.http_listener_v2.json_v2.tag]]
      path = "device"

Processor & Aggregator Plugins

# Processor: Rename tags
[[processors.rename]]
  [[processors.rename.tag]]
    dest = "sensor_id"
    tag = "device_id"

# Processor: Convert data type
[[processors.converter]]
  [processors.converter.fields]
    tag = ["device_id"]
    float = ["temperature", "humidity"]

# Aggregator: Hitung min/max/mean setiap 1 menit
[[aggregators.minmax]]
  period = "1m"
  drop_original = false
  namepass = ["temperature"]

# Aggregator: Quantile
[[aggregators.basicstats]]
  period = "5m"
  stats = ["count", "min", "max", "mean", "stdev", "s2"]
  drop_original = false

8. Integrasi dengan Grafana Dashboard

Grafana adalah tool visualisasi terbaik untuk data InfluxDB. Kamu bisa membuat dashboard real-time yang menampilkan data sensor IoT dengan berbagai jenis visualisasi.

Menambahkan InfluxDB Data Source

1. Buka Grafana → Configuration → Data Sources
2. Klik Add data source → pilih InfluxDB
3. Atur konfigurasi:
   • Query Language: Flux
   • URL: http://influxdb:8086
   • Organization: iot-org
   • Token: my-super-token
   • Default Bucket: iot-data
4. Klik Save & Test

Contoh Panel Queries

# Panel: Temperature Gauge
from(bucket: "iot-data")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "temperature")
  |> filter(fn: (r) => r._field == "value")
  |> last()

# Panel: Temperature Time Series (multi-device)
from(bucket: "iot-data")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "temperature")
  |> filter(fn: (r) => r._field == "value")
  |> aggregateWindow(every: v.windowPeriod, fn: mean)
  |> group(columns: ["device"])
  |> yield(name: "mean")

# Panel: Error Count Table
from(bucket: "iot-data")
  |> range(start: -24h)
  |> filter(fn: (r) => r._measurement == "errors")
  |> group(columns: ["device", "error_type"])
  |> count()
  |> sort(columns: ["_value"], desc: true)
  |> limit(n: 20)

Template Variables

# Variable: Pilih device
# Query: 
import "influxdata/influxdb/schema"
schema.tagValues(
  bucket: "iot-data",
  tag: "device",
  predicate: (r) => r._measurement == "temperature",
  start: -1h
)

# Variable: Pilih measurement
schema.measurements(bucket: "iot-data")

# Gunakan di panel query:
# filter(fn: (r) => r.device == v.device)

9. Quiz: Uji Pemahamanmu!

Setelah membaca tutorial di atas, jawablah 5 pertanyaan berikut untuk menguji pemahamanmu: