SIENTIAPDE-1083: add prometheus and app metrics.

.vscode/settings.json

@@ -3,5 +3,9 @@
         "."
     ],
     "python.testing.unittestEnabled": false,
-    "python.testing.pytestEnabled": true
+    "python.testing.pytestEnabled": true,
+    "sonarlint.connectedMode.project": {
+        "connectionId": "sonardev-sientia-ai",
+        "projectKey": "Aignosi_sientia-dataops-opc-ingestor_1642c8bf-a148-4911-8362-0903d7fef99a"
+    }
 }

ingestor/app.py

@@ -1,32 +1,53 @@
-from threading import Event
-import signal
 import os
+import signal
 import traceback
+from threading import Event
+from time import sleep, time
+
+from prometheus_client import start_http_server
+
+import ingestor.metrics as metrics
 from ingestor.ingestor import Ingestor
 
 exit_signal = Event()
+POD_ID = os.getenv("HOSTNAME", "localhost")
 
 
 def main():
+    start_prometheus_server()
     ingestor = Ingestor()
     ingestor.prepare_ingestor()
     ingestor.logger.info("Ingestor prepared. Starting main loop.")
 
     while not exit_signal.is_set():
+        start_time = time()  # Start loop timer
         try:
             ingestor.loop()
+            metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc()  # Increment loop counter
 
             exit_signal.wait(ingestor.poll_interval)
 
+        except KeyboardInterrupt:  # Handle Ctrl+C gracefully
+            print("KeyboardInterrupt received. Setting exit_signal flag.")
+            exit_signal.set()
         except Exception:
             print("Exception in main loop. Setting exit_signal flag.")
             traceback.print_exc()
+            metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()  # Increment errors
             exit_signal.set()
+        finally:
+            # Record loop duration
+            duration = time() - start_time
+            metrics.APP_LOOP_DURATION.labels(pod_id=POD_ID).observe(duration)
 
     ingestor.shutdown()
+    metrics.APP_UP.labels(pod_id=POD_ID).set(0)  # Mark app as DOWN
 
     ingestor.logger.info("Main loop exit_signaled.")
 
+    # Give Prometheus a chance to scrape one last time before exiting (optional)
+    sleep(5)
+
     os._exit(0)
 
 
@@ -35,6 +56,16 @@ def signal_handler(_signum, _frame):
     exit_signal.set()
 
 
+def start_prometheus_server():
+    try:
+        start_http_server(8000)
+        print("Prometheus server started on port 8000.")
+        metrics.APP_UP.labels(pod_id=POD_ID).set(1)  # Mark app as UP
+    except Exception as e:
+        print(f"Failed to start Prometheus server: {e}")
+        os._exit(1)
+
+
 if __name__ == "__main__":
     signal.signal(signal.SIGINT, signal_handler)
     signal.signal(signal.SIGTERM, signal_handler)

ingestor/metrics.py

@@ -0,0 +1,168 @@
+from prometheus_client import Counter, Gauge, Histogram
+
+# It's useful to have a common set of labels, like the pod_id.
+# We'll add 'pod_id' to many metrics to distinguish instances.
+POD_ID_LABEL = ["pod_id"]
+SERVER_LABELS = ["pod_id", "server_name", "server_url"]
+SLOT_LABELS = ["pod_id", "slot_id"]
+TAG_LABELS = ["pod_id", "server_name", "tag_id", "tag_name"]
+KAFKA_LABELS = ["pod_id", "topic"]
+REDIS_LABELS = ["pod_id", "operation"]
+NOTIFICATION_LABELS = ["pod_id", "level", "block"]
+
+
+# --- General Application Metrics ---
+APP_LOOP_COUNT = Counter(
+    "app_main_loop_total",
+    "Total number of times the application main loop has run",
+    POD_ID_LABEL,
+)
+APP_LOOP_DURATION = Histogram(
+    "app_main_loop_duration_seconds",
+    "Duration of the application main loop in seconds",
+    POD_ID_LABEL,
+)
+APP_ERRORS_TOTAL = Counter(
+    "app_errors_total",
+    "Total number of unhandled errors in the main loop",
+    POD_ID_LABEL,
+)
+APP_UP = Gauge(
+    "app_up",
+    "Indicates if the application is running (1) or shutting down (0)",
+    POD_ID_LABEL,
+)
+
+# --- Ingestor Manager Metrics ---
+INGESTORS_ATIVOS = Gauge(
+    "ingestor_active_total",
+    "Number of active ingestors reported by Redis",
+    # No pod_id here, as it's a global view from Redis
+)
+SLOTS_TOTAIS = Gauge(
+    "ingestor_slots_total",
+    "Total number of slots configured in Redis",
+    # No pod_id here, as it's a global view from Redis
+)
+LEASES_TOTAIS = Gauge(
+    "ingestor_leases_total",
+    "Total number of leases (allocated slots) in Redis",
+    # No pod_id here, as it's a global view from Redis
+)
+SLOTS_MANAGED = Gauge(
+    "ingestor_slots_managed_current",
+    "Number of slots currently managed by this ingestor instance",
+    POD_ID_LABEL,
+)
+SLOTS_ACQUIRED = Counter(
+    "ingestor_slots_acquired_total",
+    "Total number of slots acquired by this instance",
+    POD_ID_LABEL,
+)
+SLOTS_RELEASED = Counter(
+    "ingestor_slots_released_total",
+    "Total number of slots released by this instance",
+    POD_ID_LABEL,
+)
+OPC_MANAGERS_ACTIVE = Gauge(
+    "ingestor_opc_managers_active",
+    "Number of active OPC Managers in this instance",
+    POD_ID_LABEL,
+)
+OPC_SUBSCRIPTION_ERRORS = Counter(
+    "ingestor_opc_subscription_errors_total",
+    "Errors when trying to subscribe to OPC tags",
+    ["pod_id", "server", "slot"],
+)
+
+# --- OPC Manager Metrics ---
+OPC_CONNECTIONS_TOTAL = Counter(
+    "opc_connections_initiated_total",
+    "Total connection attempts to OPC servers",
+    ["pod_id", "server_name"],
+)
+OPC_CONNECTIONS_FAILED = Counter(
+    "opc_connections_failed_total",
+    "Total failed connection attempts to OPC servers",
+    ["pod_id", "server_name"],
+)
+OPC_CONNECTION_STATUS = Gauge(
+    "opc_connection_status",
+    "Connection status with the OPC server (1=connected, 0=disconnected)",
+    SERVER_LABELS,
+)
+OPC_SUBSCRIPTIONS_CREATED = Counter(
+    "opc_subscriptions_created_total",
+    "Total OPC subscriptions created",
+    ["pod_id", "server_name", "slot_name"],
+)
+OPC_TAGS_SUBSCRIBED = Gauge(
+    "opc_tags_subscribed_current",
+    "Current number of OPC tags subscribed on a server",
+    ["pod_id", "server_name"],
+)
+OPC_DATACHANGE_NOTIFICATIONS = Counter(
+    "opc_datachange_notifications_total",
+    "Total data change notifications received (tag readings)",
+    TAG_LABELS,
+)
+OPC_TAG_LAST_VALUE = Gauge(
+    "opc_tag_last_value", "The last value read from an OPC tag", TAG_LABELS
+)
+OPC_TAG_LAST_READ_TIMESTAMP = Gauge(
+    "opc_tag_last_read_timestamp_seconds",
+    "Timestamp of the last value read from an OPC tag",
+    TAG_LABELS,
+)
+OPC_CYCLES_WITHOUT_DATA = Gauge(
+    "opc_cycles_without_data",
+    "Current number of cycles without receiving data from a server",
+    ["pod_id", "server_name"],
+)
+OPC_RECONNECTIONS_TOTAL = Counter(
+    "opc_reconnections_tried_total",
+    "Reconnection attempts to an OPC server after a loss",
+    ["pod_id", "server_name"],
+)
+
+# --- Data Manager (Kafka) Metrics ---
+KAFKA_MESSAGES_SENT = Counter(
+    "kafka_messages_sent_total", "Total messages sent to Kafka", KAFKA_LABELS
+)
+KAFKA_MESSAGES_ERRORS = Counter(
+    "kafka_messages_errors_total",
+    "Total errors sending messages to Kafka",
+    KAFKA_LABELS,
+)
+KAFKA_CONNECTION_STATUS = Gauge(
+    "kafka_connection_status",
+    "Connection status with Kafka (1=connected, 0=disconnected)",
+    POD_ID_LABEL,
+)
+
+# --- Resource Manager (Redis) Metrics ---
+REDIS_OPERATIONS_TOTAL = Counter(
+    "redis_operations_total", "Total number of Redis operations performed", REDIS_LABELS
+)
+REDIS_OPERATIONS_ERRORS = Counter(
+    "redis_operations_errors_total",
+    "Total number of errors in Redis operations",
+    REDIS_LABELS,
+)
+REDIS_OPERATIONS_DURATION = Histogram(
+    "redis_operations_duration_seconds",
+    "Duration of Redis operations in seconds",
+    REDIS_LABELS,
+)
+REDIS_CONNECTION_STATUS = Gauge(
+    "redis_connection_status",
+    "Connection status with Redis (1=connected, 0=disconnected)",
+    POD_ID_LABEL,
+)
+
+# --- Notification Metrics ---
+NOTIFICATIONS_SENT = Counter(
+    "notifications_sent_total",
+    "Total number of notifications sent",
+    NOTIFICATION_LABELS,
+)

requirements.txt

@@ -1,3 +1,4 @@
 asyncua==1.1.5
 redis
 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git
+prometheus_client