diff --git a/.vscode/settings.json b/.vscode/settings.json index 3e99ede..78e50ab 100644 --- a/.vscode/settings.json +++ b/.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" + } } \ No newline at end of file diff --git a/ingestor/app.py b/ingestor/app.py index 7b3f05b..20d2a9f 100644 --- a/ingestor/app.py +++ b/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) diff --git a/ingestor/metrics.py b/ingestor/metrics.py new file mode 100644 index 0000000..c605497 --- /dev/null +++ b/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, +) diff --git a/requirements.txt b/requirements.txt index e564b8e..40a9a06 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ asyncua==1.1.5 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git \ No newline at end of file +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git +prometheus_client