Monitoramento MQTT - Khomp IED302

Escrito por Hitfy

Preparação do Proxy e Serviço de coleta MQTT

Foi desenvolvido em python um serviço que executa na maquina pxHitfy-OCI para escuta e criação de threads para receber tópicos MQTT e enviar para o host correto. Lembrando que esse processo só é necessário caso necessite reconstruir o serviço MQTT.

Código do serviço em python (/etc/zabbix/mqtt_zabbix_service.py)

#!/usr/bin/env python3

import threading
import subprocess
from flask import Flask, request, jsonify
from paho.mqtt.client import Client as MQTTClient

app = Flask(__name__)

# Chave: (broker, topic, zabbix_host), Valor: Thread
active_listeners = {}
lock = threading.Lock()

ZABBIX_SERVER = "localhost"  # IP ou hostname fixo do servidor Zabbix

def send_to_zabbix(zabbix_host, key, value):
    try:
        subprocess.run([
            "zabbix_sender", "-z", ZABBIX_SERVER,
            "-s", zabbix_host, "-k", key, "-o", value
        ], check=True)
        print(f"[ZABBIX] {zabbix_host}: {key} = {value}")
    except subprocess.CalledProcessError as e:
        print(f"[ERRO] Falha ao enviar para Zabbix ({zabbix_host}): {e}")

def mqtt_listener(broker, topic, zabbix_host):
    client = MQTTClient()
    client.on_connect = lambda c, u, f, rc: c.subscribe(topic)
    client.on_message = lambda c, u, msg: send_to_zabbix(
        zabbix_host,
        f"mqtt.{topic.replace('/', '_')}",
        msg.payload.decode()
    )

    try:
        client.connect(broker, 1883, 60)
        client.loop_forever()
    except Exception as e:
        print(f"[ERRO] Conexão com {broker} falhou: {e}")

@app.route("/listen", methods=["POST"])
def listen():
    data = request.get_json()
    broker = data.get("broker")
    topic = data.get("topic")
    zabbix_host = data.get("zabbix_host")

    if not broker or not topic or not zabbix_host:
        return jsonify({"error": "Parâmetros 'broker', 'topic' e 'zabbix_host' são obrigatórios."}), 400

    key = (broker, topic, zabbix_host)

    with lock:
        if key in active_listeners:
            return jsonify({"status": "Escutando", "broker": broker, "topic": topic, "zabbix_host": zabbix_host})

        thread = threading.Thread(target=mqtt_listener, args=(broker, topic, zabbix_host), daemon=True)
        active_listeners[key] = thread
        thread.start()
        return jsonify({"status": "Iniciado", "broker": broker, "topic": topic, "zabbix_host": zabbix_host})

@app.route("/status", methods=["GET"])
def status():
    with lock:
        return jsonify({
            "ativos": [f"{b}::{t} -> {h}" for (b, t, h) in active_listeners.keys()]
        })

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Preparar libs para ativar o serviço do linux

Instalar as libs Flask, paho-mqtt com o comando abaixo

pip3 install flask paho-mqtt

Criando serviço no linux

Crie o arquivo vim /etc/systemd/system/mqtt_zabbix.service e coloque o código abaixo

[Unit]
Description=Serviço MQTT para envio ao Zabbix
After=network.target

[Service]
Type=simple
ExecStart=/usr/bin/python3 /etc/zabbix/mqtt_zabbix_service.py
WorkingDirectory=/etc/zabbix
Restart=always
RestartSec=5
User=nobody
Environment=PYTHONUNBUFFERED=1

[Install]
WantedBy=multi-user.target

Depois execute os comandos abaixo para criar o serviço, ativar e verificar o status

systemctl daemon-reexec
systemctl daemon-reload
systemctl enable --now mqtt_zabbix.service
systemctl status mqtt_zabbix.service

Verificar logs em tempo real do serviço

journalctl -u mqtt_zabbix.service -f

Comandos para verificar e ativar novos processo de escuta MQTT por tópico

Executar o comando abaixo para verificar o status e listar todos os tópicos que estão sendo ouvidos.

 curl <http://localhost:5000/status>

Executar na maquina que esta instalado o serviço e exibira o resultado abaixo:

{"ativos":["146.235.57.205::remota/211500/measures -> Broker_MQTT_DevIOT","{$BROKER}::{$TOPIC} -> Broker_MQTT_DevIOT"]}

Comando para ativar uma nova escuta de tópico

curl -X POST <http://localhost:5000/listen> -H "Content-Type: application/json"   -d '{"broker": "<IP do Broker MQTT>", "topic": "<Tópico do MQTT>", "zabbix_host": "<Nome do Host>"}'

Executar na maquina que esta instalado o serviço e exibira o resultado abaixo:

{"broker":"146.235.57.205","status":"Escutando","topic":"remota/211500/measures","zabbix_host":"Broker_MQTT_DevIOT"}

OBS.: O serviço MQTT já bloqueia a ativação do mesmo tópico.

Ativação do monitoramento IED302 para coleta de Corrente e Tensão

É necessário criar um host no zabbix, depois adicionar a template (Template_Khomp_IED302_Padrao) e ter as informações do Broker e Tópico:

Broker, é o IP que esta instalado o serviço que coleta e envia para o Zabbix. Hoje é o IP 146.235.57.205
Tópico, é o caminho que vamos pegar as informações e coletar no zabbix, o tópico é composto por remota/<Numero de serie>/measures. Exemplo: remota/211500/measures

Depois de adicionar o host, é necessário ajustar a macro desse host com as informações acima.

Implantação