MQTT Broker Sağlık Kontrolü: Uptime İzleme Betikleri

#!/usr/bin/env bash
# Round-trip health check. Exit: 0 OK, 2 CRITICAL, 3 UNKNOWN.
HOST="${MQTT_HOST:-broker.connectmq.com}"
PORT="${MQTT_PORT:-8883}"
TIMEOUT="${MQTT_TIMEOUT:-5}"
TOPIC="testmqtt/health/$(hostname)-$$"
TOKEN="ok-$(date +%s)-$RANDOM"

command -v mosquitto_pub >/dev/null || { echo "UNKNOWN - mosquitto clients not installed"; exit 3; }

OPTS=(-h "$HOST" -p "$PORT")
[ "$PORT" = "8883" ] && OPTS+=(--capath /etc/ssl/certs)
[ -n "$MQTT_USER" ] && OPTS+=(-u "$MQTT_USER" -P "$MQTT_PASS")

OUT=$(mktemp)
mosquitto_sub "${OPTS[@]}" -i "health-sub-$$" -t "$TOPIC" -q 1 -C 1 -W "$TIMEOUT" >"$OUT" 2>&1 &
SUB_PID=$!
sleep 1   # give the subscriber time to receive its SUBACK

if ! ERR=$(mosquitto_pub "${OPTS[@]}" -i "health-pub-$$" -t "$TOPIC" -q 1 -m "$TOKEN" 2>&1); then
  kill "$SUB_PID" 2>/dev/null
  echo "CRITICAL - publish failed: $ERR"; rm -f "$OUT"; exit 2
fi

wait "$SUB_PID"
RECEIVED=$(cat "$OUT"); rm -f "$OUT"

if [ "$RECEIVED" = "$TOKEN" ]; then
  echo "OK - round trip via $HOST:$PORT"; exit 0
fi
echo "CRITICAL - no message within ${TIMEOUT}s (${RECEIVED:-no output})"; exit 2

#!/usr/bin/env python3
"""MQTT round-trip health check. Exit: 0 OK, 1 WARNING, 2 CRITICAL."""
import os, sys, threading, time, uuid
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "8883"))
TIMEOUT = float(os.getenv("MQTT_TIMEOUT", "5"))
WARN_MS = float(os.getenv("MQTT_WARN_MS", "500"))

topic = f"testmqtt/health/{uuid.uuid4().hex[:12]}"
token = uuid.uuid4().hex
connected, subscribed, received = threading.Event(), threading.Event(), threading.Event()
state = {"connack": None, "suback": None, "t0": 0.0, "rtt": 0.0}


def on_connect(client, userdata, flags, reason_code, properties):
    state["connack"] = reason_code
    connected.set()
    if not reason_code.is_failure:
        client.subscribe(topic, qos=1)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    state["suback"] = reason_codes[0]
    subscribed.set()
    if not reason_codes[0].is_failure:
        state["t0"] = time.perf_counter()
        client.publish(topic, token, qos=1)


def on_message(client, userdata, msg):
    if msg.payload.decode() == token:
        state["rtt"] = (time.perf_counter() - state["t0"]) * 1000
        received.set()


def finish(code, text):
    print(text)
    sys.exit(code)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"health-{uuid.uuid4().hex[:8]}")
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
if os.getenv("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
if PORT == 8883:
    client.tls_set()  # system CA bundle

try:
    client.connect(HOST, PORT, keepalive=30)
except Exception as exc:  # DNS, refused, TLS errors, timeouts
    finish(2, f"CRITICAL - cannot connect to {HOST}:{PORT}: {exc}")

client.loop_start()
try:
    if not connected.wait(TIMEOUT):
        finish(2, "CRITICAL - no CONNACK")
    if state["connack"].is_failure:
        finish(2, f"CRITICAL - connection refused: {state['connack']}")
    if not subscribed.wait(TIMEOUT):
        finish(2, "CRITICAL - no SUBACK")
    if state["suback"].is_failure:
        finish(2, f"CRITICAL - subscribe refused: {state['suback']}")
    if not received.wait(TIMEOUT):
        finish(2, f"CRITICAL - message not delivered within {TIMEOUT:.0f}s")
    rtt = state["rtt"]
    level, code = ("WARNING", 1) if rtt > WARN_MS else ("OK", 0)
    finish(code, f"{level} - round trip {rtt:.0f} ms | rtt={rtt:.1f}ms;{WARN_MS:.0f}")
finally:
    client.disconnect()
    client.loop_stop()

import mqtt from 'mqtt';
import { randomUUID } from 'node:crypto';

const url = process.env.MQTT_URL ?? 'mqtts://broker.connectmq.com:8883';
const timeoutMs = Number(process.env.MQTT_TIMEOUT_MS ?? 5000);
const warnMs = Number(process.env.MQTT_WARN_MS ?? 500);
const topic = `testmqtt/health/${randomUUID().slice(0, 12)}`;
const token = randomUUID();

function finish(code, text) {
  console.log(text);
  process.exit(code);
}
setTimeout(() => finish(2, `CRITICAL - no round trip within ${timeoutMs} ms`), timeoutMs);

const client = mqtt.connect(url, {
  clientId: `health-${randomUUID().slice(0, 8)}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  connectTimeout: timeoutMs,
  reconnectPeriod: 0, // a health check must not retry silently
});

client.on('error', (err) => finish(2, `CRITICAL - ${err.message}`));

client.on('connect', async () => {
  try {
    const granted = await client.subscribeAsync(topic, { qos: 1 });
    if (!granted[0] || granted[0].qos > 2) finish(2, 'CRITICAL - subscribe refused');

    let t0 = 0;
    client.on('message', (_topic, payload) => {
      if (payload.toString() !== token) return;
      const ms = Math.round(performance.now() - t0);
      const [code, level] = ms > warnMs ? [1, 'WARNING'] : [0, 'OK'];
      client.end(true, () => finish(code, `${level} - round trip ${ms} ms | rtt=${ms}ms;${warnMs}`));
    });
    t0 = performance.now();
    await client.publishAsync(topic, token, { qos: 1 });
  } catch (err) {
    finish(2, `CRITICAL - ${err.message}`);
  }
});

# every minute; alert only on non-zero exit
* * * * * /opt/checks/mqtt-health.sh >/tmp/mqtt-health.log 2>&1 || /opt/checks/notify.sh "$(cat /tmp/mqtt-health.log)"

containers:
  - name: mosquitto
    image: eclipse-mosquitto:2
    ports:
      - containerPort: 1883
    readinessProbe:
      tcpSocket:
        port: 1883
      periodSeconds: 5
    livenessProbe:
      exec:
        command:
          - sh
          - -c
          - >
            mosquitto_sub -h 127.0.0.1 -i k8s-probe-sub -t testmqtt/health/k8s -C 1 -W 5 &
            sleep 1;
            mosquitto_pub -h 127.0.0.1 -i k8s-probe-pub -t testmqtt/health/k8s -m ok;
            wait $!
      initialDelaySeconds: 10
      periodSeconds: 30
      timeoutSeconds: 8
      failureThreshold: 3