Mosquitto 와 Telegraf

Mosquitto.conf

allow_anonymous false
listener 8883
protocol mqtt

max_connections -1
max_qos 2

password_file /mosquitto/pwfile

cafile /path
certfile /path
keyfile /path
require_certificate false

persistence false
persistence_location /path
autosave_interval (seconds)

connection name
address host:port
topic pattern
bridge_protocol_version x.x.x

max_inflight_messages 10
max_queued_messages 10
message_size_limit (bytes)

TIG 구조에서 mosquitto 는 중개자(broker) 역할 수행.

실제적으로 TOPIC관련 설정은 대부분 telefraf에서 담당하게 된다. 따라서 mosquitto.conf에서는 설정할 것이 그렇게 많지 않다.

주요 설정들을 정리하였고, 상세 사항은 공식문서를 참조하면 된다.

mosquitto.conf의 topic 접두어는 주로 브로커 간 토픽 라우팅을 할 때 사용된다.

Telegraf.conf

[agent]
  interval = "5s"
  flush_interval = "5s"

[[inputs.mqtt_consumer]]
  servers = ["ssl://mosquitto:8883"]
  topics = [
    "factory/sensor/temperature",
    "factory/sensor/humidity", 
    "factory/sensor/vibration",
    "factory/sensor/pressure",
    "factory/sensor/production"
  ]
  topic_tag = "topic"
  qos = 0
  data_format = "value"
  data_type = "float"
  username = "sensor"
  password = "sensorpass"
  tls_ca = "/etc/telegraf/certs/ca.crt"
  insecure_skip_verify = true
  
[[inputs.mqtt_consumer.topic_parsing]]
    topic = "factory/sensor/+"
    measurement = "factory_sensors"
    tags = "_/_/sensor_type"

[[outputs.influxdb_v2]]
  urls = ["<http://influxdb:8086>"]
  token = "my-super-secret-token-pearl-factory"
  organization = "pearl-factory"
  bucket = "sensors"

Telegraf는 publisher 가 보낸 메시지를 broker를 거쳐 수신하게 된다.

기본적으로 전역 에이전트 설정 → 데이터 수집, 가공, 집계, 전송 영역을 설정에 입력하면 된다.

# 설정 테스트
telegraf --config telegraf.conf --test

# 특정 입력만 테스트
telegraf --config telegraf.conf --test --input-filter mqtt_consumer

# 디버그 모드 실행
telegraf --config telegraf.conf --debug

# 설정 생성 (모든 플러그인 포함)
telegraf config > telegraf-full.conf

# 플러그인 목록
telegraf --input-list
telegraf --output-list
telegraf --processor-list

흔히 사용하는 패턴

# 패턴 1: 다중 소스 통합
[[inputs.mqtt_consumer]]
  servers = ["tcp://mqtt1:1883"]
  topics = ["sensors/#"]
  name_override = "mqtt_sensors"

[[inputs.modbus]]
  controller = "tcp://plc1:502"
  name_override = "plc_data"

# 패턴 2: 조건부 라우팅
[[outputs.influxdb_v2]]
  bucket = "production"
  namepass = ["prod_*"]

[[outputs.influxdb_v2]]  
  bucket = "development"
  namepass = ["dev_*"]

# 패턴 3: 실시간 + 집계 동시
[[outputs.influxdb_v2]]
  bucket = "realtime"

[[aggregators.basicstats]]
  period = "1m"
  drop_original = false

[[outputs.influxdb_v2]]
  bucket = "aggregated"
  namepass = ["*_basicstats"]