diff --git a/ai/docker-compose.yaml b/ai/docker-compose.yaml index ac63a90..3223d47 100644 --- a/ai/docker-compose.yaml +++ b/ai/docker-compose.yaml @@ -52,6 +52,10 @@ services: ENABLE_LOGIN_FORM: "true" VECTOR_DB: qdrant QDRANT_URI: http://qdrant:6333 + RAG_RERANKING_MODEL: "reranker" + RAG_RERANKING_MODEL_AUTO_UPDATE: "false" + RAG_OPENAI_API_BASE_URL: "http://litellm:4000" + RAG_OPENAI_API_KEY: ${LITELLM_MASTER_KEY} volumes: - /mnt/docker-ssd/docker/appdata/openwebui:/app/backend/data depends_on: @@ -101,19 +105,6 @@ services: extra_hosts: - "host.docker.internal:host-gateway" - reranker: - container_name: reranker - image: ghcr.io/huggingface/text-embeddings-inference:cpu-latest - restart: unless-stopped - networks: - - ai-net - - ai-services - ports: - - "8787:80" - command: ["--model-id", "BAAI/bge-reranker-v2-m3"] - volumes: - - /mnt/docker-ssd/docker/appdata/reranker:/data - whisper: container_name: whisper image: fedirz/faster-whisper-server:latest-cuda diff --git a/ai/litellm/config.yaml b/ai/litellm/config.yaml index c782fdc..7a90da6 100644 --- a/ai/litellm/config.yaml +++ b/ai/litellm/config.yaml @@ -1,4 +1,9 @@ model_list: + # ---- Reranker (Serenity, 10.5.1.5) ---- + - model_name: "reranker" + litellm_params: + model: "huggingface/BAAI/bge-reranker-v2-m3" + api_base: "http://10.5.1.5:9787" # ---- ROCINANTE (RTX 4090, 24GB) — heavy reasoning ---- - model_name: "heavy" litellm_params: diff --git a/automation/docker-compose.yaml b/automation/docker-compose.yaml index ef329d8..0ce71a0 100644 --- a/automation/docker-compose.yaml +++ b/automation/docker-compose.yaml @@ -40,6 +40,7 @@ services: EXECUTIONS_DATA_MAX_AGE: 336 volumes: - /mnt/tank/docker/appdata/n8n:/home/node/.n8n + - /mnt/data/class-recordings:/data/class-recordings networks: - pangolin - ix-databases_shared-databases diff --git a/documents/.env.example b/documents/.env.example index dfb6984..e7c4812 100644 --- a/documents/.env.example +++ b/documents/.env.example @@ -2,3 +2,20 @@ PAPERLESS_DBUSER=paperless PAPERLESS_DBPASS=changeme PAPERLESS_SECRET_KEY=generate_with_openssl_rand_hex_32 PAPERLESS_REDIS=redis://:URL_ENCODED_REDIS_PASSWORD@shared-redis:6379 + +# ----------------------------- +# Karakeep +# ----------------------------- +KARAKEEP_VERSION=release +KARAKEEP_NEXTAUTH_URL=https://karakeep.paccoco.com +KARAKEEP_NEXTAUTH_SECRET=generate_with_openssl_rand_base64_36 +KARAKEEP_MEILI_MASTER_KEY=generate_with_openssl_rand_base64_36_alnum +KARAKEEP_DISABLE_SIGNUPS=false +KARAKEEP_DB_WAL_MODE=true +KARAKEEP_LOG_LEVEL=notice + +# Optional local AI later: +# KARAKEEP_OLLAMA_BASE_URL=http://10.5.1.6:11434 +# KARAKEEP_INFERENCE_TEXT_MODEL=llama3.2 +# KARAKEEP_INFERENCE_ENABLE_AUTO_TAGGING=true +# KARAKEEP_INFERENCE_ENABLE_AUTO_SUMMARIZATION=false diff --git a/documents/docker-compose.yaml b/documents/docker-compose.yaml index da885cc..ab6f389 100644 --- a/documents/docker-compose.yaml +++ b/documents/docker-compose.yaml @@ -12,15 +12,18 @@ services: - /mnt/unraid/data/paperless/media:/usr/src/paperless/media - /mnt/unraid/data/paperless/export:/usr/src/paperless/export - /mnt/unraid/data/paperless/consume:/usr/src/paperless/consume - env_file: .env + env_file: + - .env environment: - - PAPERLESS_REDIS=${PAPERLESS_REDIS} - - PAPERLESS_DBHOST=shared-postgres - - PAPERLESS_DBNAME=paperless - - PAPERLESS_DBPORT=5432 - - PAPERLESS_TIME_ZONE=America/Chicago - - PAPERLESS_OCR_LANGUAGE=eng - - PAPERLESS_URL=https://paperless.paccoco.com + PAPERLESS_REDIS: ${PAPERLESS_REDIS} + PAPERLESS_DBHOST: shared-postgres + PAPERLESS_DBNAME: paperless + PAPERLESS_DBUSER: ${PAPERLESS_DBUSER} + PAPERLESS_DBPASS: ${PAPERLESS_DBPASS} + PAPERLESS_DBPORT: 5432 + PAPERLESS_TIME_ZONE: America/Chicago + PAPERLESS_OCR_LANGUAGE: eng + PAPERLESS_URL: https://paperless.paccoco.com healthcheck: test: ["CMD-SHELL", "curl -sf http://127.0.0.1:8000/ || exit 1"] interval: 30s @@ -31,8 +34,70 @@ services: - ix-databases_shared-databases - pangolin + karakeep: + image: ghcr.io/karakeep-app/karakeep:${KARAKEEP_VERSION:-release} + container_name: karakeep + restart: unless-stopped + ports: + - "3100:3000" + volumes: + - /mnt/docker-ssd/docker/appdata/karakeep/data:/data + env_file: + - .env + environment: + DATA_DIR: /data + NEXTAUTH_URL: ${KARAKEEP_NEXTAUTH_URL} + NEXTAUTH_SECRET: ${KARAKEEP_NEXTAUTH_SECRET} + MEILI_ADDR: http://karakeep-meilisearch:7700 + MEILI_MASTER_KEY: ${KARAKEEP_MEILI_MASTER_KEY} + BROWSER_WEB_URL: http://karakeep-chrome:9222 + DISABLE_SIGNUPS: ${KARAKEEP_DISABLE_SIGNUPS} + DB_WAL_MODE: ${KARAKEEP_DB_WAL_MODE} + LOG_LEVEL: ${KARAKEEP_LOG_LEVEL} + TZ: America/Chicago + # Optional later: + # OLLAMA_BASE_URL: ${KARAKEEP_OLLAMA_BASE_URL} + # INFERENCE_TEXT_MODEL: ${KARAKEEP_INFERENCE_TEXT_MODEL} + # INFERENCE_ENABLE_AUTO_TAGGING: ${KARAKEEP_INFERENCE_ENABLE_AUTO_TAGGING} + # INFERENCE_ENABLE_AUTO_SUMMARIZATION: ${KARAKEEP_INFERENCE_ENABLE_AUTO_SUMMARIZATION} + depends_on: + - karakeep-chrome + - karakeep-meilisearch + networks: + - karakeep + - pangolin + + karakeep-chrome: + image: gcr.io/zenika-hub/alpine-chrome:124 + container_name: karakeep-chrome + restart: unless-stopped + command: + - --no-sandbox + - --disable-gpu + - --disable-dev-shm-usage + - --remote-debugging-address=0.0.0.0 + - --remote-debugging-port=9222 + - --hide-scrollbars + networks: + - karakeep + + karakeep-meilisearch: + image: getmeili/meilisearch:v1.41.0 + container_name: karakeep-meilisearch + restart: unless-stopped + environment: + MEILI_MASTER_KEY: ${KARAKEEP_MEILI_MASTER_KEY} + MEILI_NO_ANALYTICS: "true" + TZ: America/Chicago + volumes: + - /mnt/docker-ssd/docker/appdata/karakeep/meilisearch:/meili_data + networks: + - karakeep + networks: ix-databases_shared-databases: external: true pangolin: external: true + karakeep: + name: documents_karakeep diff --git a/mesh-mqtt-observer/Dockerfile b/mesh-mqtt-observer/Dockerfile new file mode 100644 index 0000000..5877472 --- /dev/null +++ b/mesh-mqtt-observer/Dockerfile @@ -0,0 +1,17 @@ +FROM python:3.12-slim + +ENV PYTHONUNBUFFERED=1 +ENV PYTHONDONTWRITEBYTECODE=1 + +WORKDIR /app + +RUN apt-get update \ + && apt-get install -y --no-install-recommends gcc libpq-dev curl \ + && rm -rf /var/lib/apt/lists/* + +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r /app/requirements.txt + +COPY app /app/app + +CMD ["python", "-m", "app.main"] diff --git a/mesh-mqtt-observer/app/__init__.py b/mesh-mqtt-observer/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/mesh-mqtt-observer/app/healthcheck.py b/mesh-mqtt-observer/app/healthcheck.py new file mode 100644 index 0000000..d16ca52 --- /dev/null +++ b/mesh-mqtt-observer/app/healthcheck.py @@ -0,0 +1,25 @@ +import os +import sys + +import psycopg + + +def main() -> int: + dsn = os.getenv("DATABASE_URL") + if not dsn: + print("DATABASE_URL missing") + return 1 + + try: + with psycopg.connect(dsn, autocommit=True) as conn: + with conn.cursor() as cur: + cur.execute("SELECT 1") + cur.fetchone() + return 0 + except Exception as exc: + print(exc) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/mesh-mqtt-observer/app/main.py b/mesh-mqtt-observer/app/main.py new file mode 100644 index 0000000..f9db504 --- /dev/null +++ b/mesh-mqtt-observer/app/main.py @@ -0,0 +1,685 @@ +import json +import logging +import os +import signal +import ssl +import sys +import threading +import time +from dataclasses import dataclass +from typing import Any + +import orjson +import paho.mqtt.client as mqtt +import psycopg +from psycopg.rows import dict_row + + +STOP_EVENT = threading.Event() + + +def env_bool(name: str, default: bool = False) -> bool: + value = os.getenv(name) + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "y", "on"} + + +def env_int(name: str, default: int) -> int: + value = os.getenv(name) + if not value: + return default + return int(value) + + +def split_topics(value: str) -> list[str]: + return [x.strip() for x in value.split(",") if x.strip()] + + +def setup_logging() -> None: + level_name = os.getenv("LOG_LEVEL", "INFO").upper() + level = getattr(logging, level_name, logging.INFO) + + logging.basicConfig( + level=level, + format="%(asctime)s [%(levelname)s] %(message)s", + stream=sys.stdout, + ) + + +@dataclass +class BrokerConfig: + source_system: str + network_name: str | None + modem_preset: str | None + host: str + port: int + username: str | None + password: str | None + tls: bool + topics: list[str] + + +class Database: + def __init__(self, dsn: str): + self.dsn = dsn + + def connect(self): + return psycopg.connect(self.dsn, row_factory=dict_row, autocommit=True) + + def wait_until_ready(self, timeout_seconds: int = 120) -> None: + deadline = time.time() + timeout_seconds + last_error = None + + while time.time() < deadline: + try: + with self.connect() as conn: + with conn.cursor() as cur: + cur.execute("SELECT 1") + logging.info("Postgres connection ready") + return + except Exception as exc: + last_error = exc + logging.warning("Waiting for Postgres: %s", exc) + time.sleep(3) + + raise RuntimeError(f"Postgres not ready after {timeout_seconds}s: {last_error}") + + def init_schema(self) -> None: + schema_path = os.path.join(os.path.dirname(__file__), "schema.sql") + + with open(schema_path, "r", encoding="utf-8") as f: + sql = f.read() + + with self.connect() as conn: + with conn.cursor() as cur: + cur.execute(sql) + + logging.info("Schema initialized") + + def insert_raw( + self, + *, + source_system: str, + network_name: str | None, + modem_preset: str | None, + broker_host: str, + broker_port: int, + topic: str, + qos: int, + retain: bool, + payload_raw: bytes | None, + payload_text: str | None, + payload_json: Any | None, + parse_status: str, + parser_error: str | None, + ) -> int: + with self.connect() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO raw_mqtt_events ( + source_system, + network_name, + modem_preset, + broker_host, + broker_port, + topic, + qos, + retain, + payload_raw, + payload_text, + payload_json, + parse_status, + parser_error + ) + VALUES ( + %(source_system)s, + %(network_name)s, + %(modem_preset)s, + %(broker_host)s, + %(broker_port)s, + %(topic)s, + %(qos)s, + %(retain)s, + %(payload_raw)s, + %(payload_text)s, + %(payload_json)s, + %(parse_status)s, + %(parser_error)s + ) + RETURNING id + """, + { + "source_system": source_system, + "network_name": network_name, + "modem_preset": modem_preset, + "broker_host": broker_host, + "broker_port": broker_port, + "topic": topic, + "qos": qos, + "retain": retain, + "payload_raw": payload_raw, + "payload_text": payload_text, + "payload_json": json.dumps(payload_json) if payload_json is not None else None, + "parse_status": parse_status, + "parser_error": parser_error, + }, + ) + return int(cur.fetchone()["id"]) + + def insert_normalized( + self, + *, + raw_event_id: int, + source_system: str, + network_name: str | None, + modem_preset: str | None, + topic: str, + from_node_id: str | None, + to_node_id: str | None, + message_type: str | None, + text_body: str | None, + rssi: float | None, + snr: float | None, + payload: Any | None, + ) -> None: + with self.connect() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO normalized_events ( + raw_event_id, + source_system, + network_name, + modem_preset, + topic, + from_node_id, + to_node_id, + message_type, + text_body, + rssi, + snr, + payload + ) + VALUES ( + %(raw_event_id)s, + %(source_system)s, + %(network_name)s, + %(modem_preset)s, + %(topic)s, + %(from_node_id)s, + %(to_node_id)s, + %(message_type)s, + %(text_body)s, + %(rssi)s, + %(snr)s, + %(payload)s + ) + """, + { + "raw_event_id": raw_event_id, + "source_system": source_system, + "network_name": network_name, + "modem_preset": modem_preset, + "topic": topic, + "from_node_id": from_node_id, + "to_node_id": to_node_id, + "message_type": message_type, + "text_body": text_body, + "rssi": rssi, + "snr": snr, + "payload": json.dumps(payload) if payload is not None else None, + }, + ) + + def upsert_node(self, source_system: str, node_id: str, payload: Any | None) -> None: + with self.connect() as conn: + with conn.cursor() as cur: + cur.execute( + """ + INSERT INTO nodes_seen ( + source_system, + node_id, + first_seen, + last_seen, + metadata + ) + VALUES ( + %(source_system)s, + %(node_id)s, + now(), + now(), + %(metadata)s + ) + ON CONFLICT (source_system, node_id) + DO UPDATE SET + last_seen = now(), + metadata = COALESCE(EXCLUDED.metadata, nodes_seen.metadata) + """, + { + "source_system": source_system, + "node_id": node_id, + "metadata": json.dumps(payload) if payload is not None else None, + }, + ) + + +def try_decode_text(payload: bytes) -> str | None: + try: + text = payload.decode("utf-8") + except UnicodeDecodeError: + return None + + if not text: + return None + + if any(ord(ch) < 9 for ch in text): + return None + + return text + + +def try_decode_json(payload_text: str | None) -> tuple[Any | None, str | None]: + if payload_text is None: + return None, None + + stripped = payload_text.strip() + + if not stripped: + return None, None + + if not stripped.startswith("{") and not stripped.startswith("["): + return None, None + + try: + return orjson.loads(stripped), None + except Exception as exc: + return None, str(exc) + + +def deep_get(data: Any, keys: list[str]) -> Any: + if isinstance(data, dict): + for key in keys: + if key in data: + return data[key] + + for value in data.values(): + result = deep_get(value, keys) + if result is not None: + return result + + if isinstance(data, list): + for item in data: + result = deep_get(item, keys) + if result is not None: + return result + + return None + + +def as_str(value: Any) -> str | None: + if value is None: + return None + return str(value) + + +def as_float(value: Any) -> float | None: + if value is None: + return None + + try: + return float(value) + except Exception: + return None + + +def infer_type(payload_json: Any, topic: str) -> str: + explicit = deep_get( + payload_json, + [ + "message_type", + "messageType", + "type", + "portnum", + "portNum", + "port_num", + "event_type", + "eventType", + ], + ) + + if explicit is not None: + text = str(explicit).lower() + + if "text" in text: + return "text" + if "position" in text: + return "position" + if "telemetry" in text: + return "telemetry" + if "node" in text or "user" in text: + return "nodeinfo" + + return text + + body = deep_get( + payload_json, + [ + "text", + "message", + "body", + "decoded_text", + "decodedText", + "payload_text", + "payloadText", + "content", + ], + ) + + if body: + return "text" + + lowered_topic = topic.lower() + + if "position" in lowered_topic or "/pos" in lowered_topic: + return "position" + if "telemetry" in lowered_topic: + return "telemetry" + if "nodeinfo" in lowered_topic or "node" in lowered_topic or "user" in lowered_topic: + return "nodeinfo" + + return "unknown" + + +def normalize_json(payload_json: Any, topic: str) -> dict[str, Any]: + from_node_id = as_str( + deep_get( + payload_json, + [ + "from", + "from_id", + "fromId", + "from_node", + "fromNode", + "from_node_id", + "fromNodeId", + "sender", + "sender_id", + "senderId", + "node_id", + "nodeId", + "id", + ], + ) + ) + + to_node_id = as_str( + deep_get( + payload_json, + [ + "to", + "to_id", + "toId", + "to_node", + "toNode", + "to_node_id", + "toNodeId", + "recipient", + "recipient_id", + "recipientId", + ], + ) + ) + + text_body = as_str( + deep_get( + payload_json, + [ + "text", + "message", + "body", + "decoded_text", + "decodedText", + "payload_text", + "payloadText", + "content", + ], + ) + ) + + rssi = as_float(deep_get(payload_json, ["rssi", "rx_rssi", "rxRssi"])) + snr = as_float(deep_get(payload_json, ["snr", "rx_snr", "rxSnr"])) + + return { + "from_node_id": from_node_id, + "to_node_id": to_node_id, + "message_type": infer_type(payload_json, topic), + "text_body": text_body, + "rssi": rssi, + "snr": snr, + } + + +class MeshMqttClient: + def __init__(self, db: Database, config: BrokerConfig): + self.db = db + self.config = config + + client_id = f"mesh-mqtt-observer-{config.source_system}-{os.getpid()}" + + self.client = mqtt.Client( + mqtt.CallbackAPIVersion.VERSION2, + client_id=client_id, + clean_session=True, + ) + + if config.username: + self.client.username_pw_set(config.username, config.password or None) + + if config.tls: + self.client.tls_set(cert_reqs=ssl.CERT_REQUIRED) + self.client.tls_insecure_set(False) + + self.client.on_connect = self.on_connect + self.client.on_disconnect = self.on_disconnect + self.client.on_subscribe = self.on_subscribe + self.client.on_message = self.on_message + + def on_connect(self, client, userdata, flags, reason_code, properties): + logging.info( + "[%s] Connected to MQTT broker %s:%s: %s", + self.config.source_system, + self.config.host, + self.config.port, + reason_code, + ) + + for topic in self.config.topics: + logging.info("[%s] Subscribing to %s", self.config.source_system, topic) + client.subscribe(topic, qos=0) + + def on_disconnect(self, client, userdata, disconnect_flags, reason_code, properties): + logging.warning( + "[%s] Disconnected from MQTT broker: %s", + self.config.source_system, + reason_code, + ) + + def on_subscribe(self, client, userdata, mid, reason_codes, properties): + logging.info( + "[%s] Subscribe ACK mid=%s reason_codes=%s", + self.config.source_system, + mid, + reason_codes, + ) + + def on_message(self, client, userdata, msg): + payload = bytes(msg.payload or b"") + + payload_raw = payload if env_bool("STORE_RAW_BYTES", True) else None + payload_text = try_decode_text(payload) if env_bool("STORE_TEXT_PAYLOAD", True) else None + payload_json, json_error = try_decode_json(payload_text) + + parse_status = "json" if payload_json is not None else "raw" + + try: + raw_id = self.db.insert_raw( + source_system=self.config.source_system, + network_name=self.config.network_name, + modem_preset=self.config.modem_preset, + broker_host=self.config.host, + broker_port=self.config.port, + topic=msg.topic, + qos=msg.qos, + retain=msg.retain, + payload_raw=payload_raw, + payload_text=payload_text, + payload_json=payload_json, + parse_status=parse_status, + parser_error=json_error, + ) + + if payload_json is not None: + normalized = normalize_json(payload_json, msg.topic) + + self.db.insert_normalized( + raw_event_id=raw_id, + source_system=self.config.source_system, + network_name=self.config.network_name, + modem_preset=self.config.modem_preset, + topic=msg.topic, + from_node_id=normalized["from_node_id"], + to_node_id=normalized["to_node_id"], + message_type=normalized["message_type"], + text_body=normalized["text_body"], + rssi=normalized["rssi"], + snr=normalized["snr"], + payload=payload_json, + ) + + if normalized["from_node_id"]: + self.db.upsert_node( + self.config.source_system, + normalized["from_node_id"], + payload_json, + ) + + logging.debug( + "[%s] Stored topic=%s bytes=%s status=%s", + self.config.source_system, + msg.topic, + len(payload), + parse_status, + ) + + except Exception as exc: + logging.exception( + "[%s] Failed to store topic=%s: %s", + self.config.source_system, + msg.topic, + exc, + ) + + def start(self): + logging.info( + "[%s] Connecting to %s:%s", + self.config.source_system, + self.config.host, + self.config.port, + ) + + self.client.connect(self.config.host, self.config.port, keepalive=60) + self.client.loop_start() + + def stop(self): + logging.info("[%s] Stopping MQTT client", self.config.source_system) + self.client.loop_stop() + self.client.disconnect() + + +def load_configs() -> list[BrokerConfig]: + configs: list[BrokerConfig] = [] + + if env_bool("MESHTASTIC_ENABLED", True): + configs.append( + BrokerConfig( + source_system="meshtastic", + network_name=os.getenv("MESHTASTIC_NETWORK_NAME") or "Meshtastic", + modem_preset=os.getenv("MESHTASTIC_MODEM_PRESET") or None, + host=os.getenv("MESHTASTIC_MQTT_HOST", "mqtt.meshtastic.org"), + port=env_int("MESHTASTIC_MQTT_PORT", 1883), + username=os.getenv("MESHTASTIC_MQTT_USERNAME") or None, + password=os.getenv("MESHTASTIC_MQTT_PASSWORD") or None, + tls=env_bool("MESHTASTIC_MQTT_TLS", False), + topics=split_topics(os.getenv("MESHTASTIC_MQTT_TOPICS", "msh/#")), + ) + ) + + if env_bool("MESHCORE_ENABLED", False): + meshcore_host = os.getenv("MESHCORE_MQTT_HOST", "").strip() + + if meshcore_host and meshcore_host != "CHANGE_ME": + configs.append( + BrokerConfig( + source_system="meshcore", + network_name=os.getenv("MESHCORE_NETWORK_NAME") or "MeshCore", + modem_preset=os.getenv("MESHCORE_MODEM_PRESET") or None, + host=meshcore_host, + port=env_int("MESHCORE_MQTT_PORT", 1883), + username=os.getenv("MESHCORE_MQTT_USERNAME") or None, + password=os.getenv("MESHCORE_MQTT_PASSWORD") or None, + tls=env_bool("MESHCORE_MQTT_TLS", False), + topics=split_topics(os.getenv("MESHCORE_MQTT_TOPICS", "meshcore/#")), + ) + ) + else: + logging.warning("MeshCore is enabled but MESHCORE_MQTT_HOST is not set; skipping") + + return configs + + +def handle_signal(signum, frame): + logging.info("Received signal %s; stopping", signum) + STOP_EVENT.set() + + +def main() -> None: + setup_logging() + + signal.signal(signal.SIGTERM, handle_signal) + signal.signal(signal.SIGINT, handle_signal) + + dsn = os.getenv("DATABASE_URL") + + if not dsn: + raise RuntimeError("DATABASE_URL is required") + + db = Database(dsn) + db.wait_until_ready() + db.init_schema() + + configs = load_configs() + + if not configs: + raise RuntimeError("No MQTT brokers configured") + + clients = [MeshMqttClient(db, config) for config in configs] + + for client in clients: + client.start() + + logging.info("mesh-mqtt-observer started with %s broker(s)", len(clients)) + + try: + while not STOP_EVENT.is_set(): + time.sleep(1) + finally: + for client in clients: + client.stop() + + logging.info("mesh-mqtt-observer stopped") + + +if __name__ == "__main__": + main() diff --git a/mesh-mqtt-observer/app/schema.sql b/mesh-mqtt-observer/app/schema.sql new file mode 100644 index 0000000..dc896d4 --- /dev/null +++ b/mesh-mqtt-observer/app/schema.sql @@ -0,0 +1,146 @@ +CREATE TABLE IF NOT EXISTS raw_mqtt_events ( + id BIGSERIAL PRIMARY KEY, + received_at TIMESTAMPTZ NOT NULL DEFAULT now(), + + source_system TEXT NOT NULL, + network_name TEXT, + modem_preset TEXT, + + broker_host TEXT, + broker_port INTEGER, + + topic TEXT NOT NULL, + qos INTEGER, + retain BOOLEAN, + + payload_raw BYTEA, + payload_text TEXT, + payload_json JSONB, + + parse_status TEXT NOT NULL DEFAULT 'raw', + parser_error TEXT +); + +CREATE INDEX IF NOT EXISTS idx_raw_mqtt_events_received_at + ON raw_mqtt_events(received_at DESC); + +CREATE INDEX IF NOT EXISTS idx_raw_mqtt_events_source_received + ON raw_mqtt_events(source_system, received_at DESC); + +CREATE INDEX IF NOT EXISTS idx_raw_mqtt_events_topic + ON raw_mqtt_events(topic); + +CREATE TABLE IF NOT EXISTS normalized_events ( + id BIGSERIAL PRIMARY KEY, + raw_event_id BIGINT REFERENCES raw_mqtt_events(id) ON DELETE CASCADE, + + event_time TIMESTAMPTZ NOT NULL DEFAULT now(), + source_system TEXT NOT NULL, + network_name TEXT, + modem_preset TEXT, + topic TEXT NOT NULL, + + from_node_id TEXT, + to_node_id TEXT, + message_type TEXT, + text_body TEXT, + + rssi NUMERIC, + snr NUMERIC, + + payload JSONB +); + +CREATE INDEX IF NOT EXISTS idx_normalized_events_event_time + ON normalized_events(event_time DESC); + +CREATE INDEX IF NOT EXISTS idx_normalized_events_source_time + ON normalized_events(source_system, event_time DESC); + +CREATE INDEX IF NOT EXISTS idx_normalized_events_from_node + ON normalized_events(source_system, from_node_id); + +CREATE INDEX IF NOT EXISTS idx_normalized_events_message_type + ON normalized_events(message_type); + +CREATE TABLE IF NOT EXISTS nodes_seen ( + id BIGSERIAL PRIMARY KEY, + source_system TEXT NOT NULL, + node_id TEXT NOT NULL, + first_seen TIMESTAMPTZ NOT NULL DEFAULT now(), + last_seen TIMESTAMPTZ NOT NULL DEFAULT now(), + metadata JSONB, + UNIQUE(source_system, node_id) +); + +CREATE INDEX IF NOT EXISTS idx_nodes_seen_last_seen + ON nodes_seen(last_seen DESC); + +CREATE OR REPLACE VIEW v_packets_per_hour AS +SELECT + date_trunc('hour', event_time) AS hour, + source_system, + network_name, + modem_preset, + COUNT(*) AS packets +FROM normalized_events +GROUP BY 1, 2, 3, 4 +ORDER BY 1 DESC; + +CREATE OR REPLACE VIEW v_text_messages_per_day AS +SELECT + date_trunc('day', event_time) AS day, + source_system, + network_name, + modem_preset, + COUNT(*) AS text_messages +FROM normalized_events +WHERE message_type = 'text' +GROUP BY 1, 2, 3, 4 +ORDER BY 1 DESC; + +CREATE OR REPLACE VIEW v_unique_nodes_per_day AS +SELECT + date_trunc('day', event_time) AS day, + source_system, + network_name, + modem_preset, + COUNT(DISTINCT from_node_id) AS unique_nodes +FROM normalized_events +WHERE from_node_id IS NOT NULL +GROUP BY 1, 2, 3, 4 +ORDER BY 1 DESC; + +CREATE OR REPLACE VIEW v_utilization_hourly AS +SELECT + date_trunc('hour', event_time) AS hour, + CASE + WHEN source_system = 'meshtastic' AND modem_preset IS NOT NULL AND modem_preset <> '' + THEN 'Meshtastic ' || modem_preset + WHEN source_system = 'meshtastic' + THEN 'Meshtastic' + WHEN source_system = 'meshcore' + THEN 'MeshCore' + ELSE source_system + END AS network, + COUNT(*) AS packets, + COUNT(*) FILTER (WHERE message_type = 'text') AS text_messages, + COUNT(DISTINCT from_node_id) AS unique_nodes +FROM normalized_events +GROUP BY 1, 2 +ORDER BY 1 DESC; + +CREATE OR REPLACE VIEW v_top_talkers_24h AS +SELECT + source_system, + network_name, + modem_preset, + from_node_id, + COUNT(*) AS packets, + COUNT(*) FILTER (WHERE message_type = 'text') AS text_messages, + MAX(event_time) AS last_seen +FROM normalized_events +WHERE event_time >= now() - interval '24 hours' + AND from_node_id IS NOT NULL +GROUP BY 1, 2, 3, 4 +ORDER BY packets DESC; diff --git a/mesh-mqtt-observer/docker-compose.yml b/mesh-mqtt-observer/docker-compose.yml new file mode 100644 index 0000000..1524366 --- /dev/null +++ b/mesh-mqtt-observer/docker-compose.yml @@ -0,0 +1,24 @@ +services: + mesh-mqtt-observer: + build: + context: . + dockerfile: Dockerfile + container_name: mesh-mqtt-observer + restart: unless-stopped + env_file: + - .env + volumes: + - /mnt/tank/docker/appdata/mesh-mqtt-observer:/data + networks: + - shared_postgres + healthcheck: + test: ["CMD", "python", "-m", "app.healthcheck"] + interval: 30s + timeout: 10s + retries: 5 + start_period: 20s + +networks: + shared_postgres: + external: true + name: ${SHARED_POSTGRES_NETWORK} diff --git a/mesh-mqtt-observer/requirements.txt b/mesh-mqtt-observer/requirements.txt new file mode 100644 index 0000000..f6e7514 --- /dev/null +++ b/mesh-mqtt-observer/requirements.txt @@ -0,0 +1,4 @@ +paho-mqtt==2.1.0 +psycopg[binary,pool]==3.2.3 +python-dotenv==1.0.1 +orjson==3.10.12 diff --git a/meshtastic/docker-compose.yaml b/meshtastic/docker-compose.yaml index 7607a2b..2e4f1cb 100644 --- a/meshtastic/docker-compose.yaml +++ b/meshtastic/docker-compose.yaml @@ -1,6 +1,6 @@ services: meshmonitor: - image: ghcr.io/yeraze/meshmonitor:4.2.1 + image: ghcr.io/yeraze/meshmonitor:latest container_name: meshmonitor ports: - "8081:3001" @@ -13,7 +13,7 @@ services: environment: - NODE_ENV=production - TZ=America/Chicago - - MESHTASTIC_NODE_IP=10.5.1.172 + - MESHTASTIC_NODE_IP=10.5.1.120 - TRUST_PROXY=true - COOKIE_SECURE=true - ALLOWED_ORIGINS=https://meshmonitor.paccoco.com @@ -22,6 +22,10 @@ services: - PIRATE_WEATHER_API_KEY=${PIRATE_WEATHER_API_KEY} - LOCAL_LAT=${LOCAL_LAT} - LOCAL_LON=${LOCAL_LON} + # Rate limiting + - RATE_LIMIT_API=${RATE_LIMIT_API} + - RATE_LIMIT_AUTH=${RATE_LIMIT_AUTH} + networks: - default - ix-databases_shared-databases diff --git a/monitoring/docker-compose.yaml b/monitoring/docker-compose.yaml new file mode 100644 index 0000000..55e0b2d --- /dev/null +++ b/monitoring/docker-compose.yaml @@ -0,0 +1,59 @@ +name: monitoring + +services: + prometheus: + image: prom/prometheus:v3.4.0 + container_name: prometheus + restart: unless-stopped + ports: + - "9090:9090" + command: + - '--config.file=/etc/prometheus/prometheus.yml' + - '--storage.tsdb.path=/prometheus' + - '--storage.tsdb.retention.time=90d' + - '--web.enable-lifecycle' + volumes: + - /mnt/tank/docker/appdata/prometheus:/prometheus + - /mnt/docker-ssd/docker/compose/monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro + networks: + - default + + grafana: + image: grafana/grafana:13.0.1 + container_name: grafana + restart: unless-stopped + ports: + - "3000:3000" + environment: + TZ: ${TZ} + GF_SECURITY_ADMIN_USER: ${GF_ADMIN_USER} + GF_SECURITY_ADMIN_PASSWORD: ${GF_ADMIN_PASS} + GF_SERVER_ROOT_URL: https://${GF_HOST}/ + volumes: + - /mnt/tank/docker/appdata/grafana:/var/lib/grafana + - /mnt/docker-ssd/docker/compose/monitoring/provisioning:/etc/grafana/provisioning:ro + networks: + - pangolin + - default + + node-exporter: + image: prom/node-exporter:v1.9.0 + container_name: node-exporter + restart: unless-stopped + ports: + - "9100:9100" + command: + - '--path.procfs=/host/proc' + - '--path.sysfs=/host/sys' + - '--path.rootfs=/rootfs' + - '--collector.filesystem.mount-points-exclude=^/(sys|proc|dev|host|etc)($$|/)' + volumes: + - /proc:/host/proc:ro + - /sys:/host/sys:ro + - /:/rootfs:ro + networks: + - default + +networks: + pangolin: + external: true diff --git a/monitoring/prometheus.yml b/monitoring/prometheus.yml new file mode 100644 index 0000000..85a6141 --- /dev/null +++ b/monitoring/prometheus.yml @@ -0,0 +1,28 @@ +global: + scrape_interval: 15s + evaluation_interval: 15s + +scrape_configs: + # ------- PlausibleDeniability (local via node-exporter) ------- + - job_name: 'pd-node' + static_configs: + - targets: ['node-exporter:9100'] + labels: + host: 'plausible-deniability' + + # ------- Serenity (via Netdata Prometheus exporter) ------- + - job_name: 'serenity-netdata' + metrics_path: /api/v1/allmetrics + params: + format: [prometheus] + static_configs: + - targets: ['10.5.1.5:19999'] + labels: + host: 'serenity' + + # ------- N.O.M.A.D. (via node-exporter) ------- + - job_name: 'nomad-node' + static_configs: + - targets: ['10.5.1.16:9100'] + labels: + host: 'nomad' diff --git a/monitoring/provisioning/dashboards/dashboards.yml b/monitoring/provisioning/dashboards/dashboards.yml new file mode 100644 index 0000000..eadf916 --- /dev/null +++ b/monitoring/provisioning/dashboards/dashboards.yml @@ -0,0 +1,12 @@ +apiVersion: 1 + +providers: + - name: 'default' + orgId: 1 + folder: '' + type: file + disableDeletion: false + editable: true + options: + path: /var/lib/grafana/dashboards + foldersFromFilesStructure: false diff --git a/monitoring/provisioning/datasources/prometheus.yml b/monitoring/provisioning/datasources/prometheus.yml new file mode 100644 index 0000000..1a57b69 --- /dev/null +++ b/monitoring/provisioning/datasources/prometheus.yml @@ -0,0 +1,9 @@ +apiVersion: 1 + +datasources: + - name: Prometheus + type: prometheus + access: proxy + url: http://prometheus:9090 + isDefault: true + editable: true