From 5132162eaaa69a58a1ad8cdf0df18fc2777d5630 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 16 Jul 2025 10:54:34 -0300 Subject: [PATCH] SIENTIAPDE-1163 Refactor Ingestor and Manager classes to use CoreNotificationHandler and improve metadata handling - Updated Ingestor class to initialize logger using get_logger and replaced notification handler initialization with metadata dictionary. - Refactored DataManager, IngestorManager, OpcManager, and ResourceManager classes to inherit from BaseActivity, allowing for consistent logger and notification handler usage. - Enhanced notification handling by integrating metadata into notification methods across various managers. - Added .vscode/ to .gitignore to exclude VSCode configuration files. --- .gitignore | 3 ++ ingestor/ingestor.py | 60 ++++++++----------------- ingestor/managers/data_manager.py | 19 ++++---- ingestor/managers/ingestor_manager.py | 65 ++++++++++++++++++--------- ingestor/managers/opc_manager.py | 26 ++++++----- ingestor/managers/resource_manager.py | 26 +++++++++-- 6 files changed, 116 insertions(+), 83 deletions(-) diff --git a/.gitignore b/.gitignore index 170a3c3..3255fbb 100644 --- a/.gitignore +++ b/.gitignore @@ -173,3 +173,6 @@ cython_debug/ # PyPI configuration file .pypirc + +# VSCode +.vscode/ \ No newline at end of file diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 268a520..28a8644 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -1,4 +1,3 @@ -from logging import Formatter, StreamHandler, getLogger from os import getenv from copy import deepcopy from typing import Dict, Any @@ -62,8 +61,7 @@ class Ingestor: self.mongo_connection_string = f"mongodb://{mongo_username}:{mongo_password}@{mongo_url}" self.kafka_servers = kafka_servers.split(",") - self.logger = None - self.init_logger() + self.logger = get_logger(__name__) self.notification_handler = NotificationHandler( connection_string=self.mongo_connection_string, database=self.mongo_database, @@ -71,11 +69,12 @@ class Ingestor: project_name="OPC_INGESTOR" ) - self.notification_handler.base_notification.pipeline = 'OPC_INGESTOR' - self.notification_handler.base_notification.trigger = 'INGESTOR' - self.notification_handler.base_notification.model_name = '-' - self.notification_handler.base_notification.model_id = '-' - + self.metadata = { + 'schedule_name': 'OPC_INGESTOR', + 'trigger': 'INGESTOR', + 'model_name': '-', + 'model_id': '-' + } self.ingestor_manager = None def shutdown(self): @@ -85,28 +84,6 @@ class Ingestor: def __del__(self): self.shutdown() - def init_logger(self): - """ - Initializes a logger instance for the class. - This method sets up a logger with a specified log level, a stream handler, - and a formatter. The log level is determined by the environment variable - "LOG_LEVEL", defaulting to "INFO" if not set. The logger is then attached - to the instance for use throughout the class. - Attributes: - self.logger (logging.Logger): The configured logger instance. - """ - - logger = getLogger(__name__) - logger.setLevel(getenv("LOG_LEVEL", "INFO")) - handler = StreamHandler() - formatter = Formatter( - "%(asctime)s - %(name)s - %(levelname)s - %(message)s") - handler.setFormatter(formatter) - - logger.addHandler(handler) - - self.logger = logger - def handle_acquired_tags(self, acquired): """ Handles the acquired tags by subscribing to them if available. @@ -150,22 +127,23 @@ class Ingestor: """ self.ingestor_manager = IngestorManager( - self.kafka_servers, - { + kafka_servers=self.kafka_servers, + redis_config={ 'host': self.redis_host, 'port': self.redis_port, 'username': self.redis_username, 'password': self.redis_password, }, - self.lease_ttl, - self.heartbeat_ttl, - self.pod_id, - self.poll_interval, - self.mongo_connection_string, - self.mongo_database, - self.logger, - self.notification_handler, - self.export_to_kafka, + lease_ttl=self.lease_ttl, + heartbeat_ttl=self.heartbeat_ttl, + pod_id=self.pod_id, + poll_interval=self.poll_interval, + mongo_connection_string=self.mongo_connection_string, + mongo_database=self.mongo_database, + metadata=self.metadata, + logger=self.logger, + notification_handler=self.notification_handler, + export_to_kafka=self.export_to_kafka, ) # Declare ingestor ative diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 5f1e634..9fd50f6 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -1,25 +1,26 @@ import json -from logging import Logger from time import sleep from datetime import datetime, timezone from pymongo import MongoClient from kafka import KafkaProducer from kafka.errors import NoBrokersAvailable -from sientia_do.notifications.handlers import NotificationHandler +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel -from sientia_do. +from sientia_do.temporal.activities.base import BaseActivity +from sientia_do.temporal.utils.logger import Logger import traceback import ingestor.metrics as metrics import os -class DataManager(): +class DataManager(BaseActivity): def __init__( self, kafka_servers: str, mongo_connection_string: str, mongo_database: str, export_to_kafka: bool, + metadata: dict, logger: Logger, notification_handler: NotificationHandler, ) -> None: @@ -92,8 +93,8 @@ class DataManager(): f"DataManager initialized with MongoDB servers: {self.connection_string}" ) - self.logger = logger - self.notification_handler = notification_handler + BaseActivity.__init__(self, logger=logger, + notification_handler=notification_handler) def shutdown(self): """Closes the Kafka producer connection.""" @@ -164,7 +165,8 @@ class DataManager(): metrics.KAFKA_MESSAGES_ERRORS.labels( pod_id=self.pod_id, topic=topic).inc() trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=self.metadata, notification_id=f"KAFKA_PRODUCER_ERROR_{topic}", message=f"Error publishing message to topic {topic}: {e}", block="kafka_producer", @@ -187,7 +189,8 @@ class DataManager(): except Exception as e: trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=self.metadata, notification_id=f"MONGO_PRODUCER_ERROR_{topic}", message=f"Error inserting message to MongoDB: {e}", block="mongo_producer", diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 8501e81..47499fd 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -1,20 +1,22 @@ -from logging import Logger import traceback from typing import Dict, List from copy import deepcopy -from sientia_do.notifications.handlers import NotificationHandler +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel +from sientia_do.temporal.activities.base import BaseActivity +from sientia_do.temporal.utils.logger import Logger from ingestor.managers.data_manager import DataManager from ingestor.managers.opc_manager import OpcManager from ingestor.managers.resource_manager import ResourceManager import ingestor.metrics as metrics -class IngestorManager(): +class IngestorManager(BaseActivity): def __init__(self, kafka_servers: str, redis_data: dict, lease_ttl: int, heartbeat_ttl: int, pod_id: str, poll_interval: int, mongo_connection_string: str, mongo_database: str, + metadata: dict, logger: Logger, notification_handler: NotificationHandler, export_to_kafka: bool = False): @@ -24,23 +26,39 @@ class IngestorManager(): redis_password = redis_data.get('password', None) self.data_manager = DataManager( - kafka_servers, mongo_connection_string, mongo_database, - export_to_kafka, logger, notification_handler) + kafka_servers=kafka_servers, + mongo_connection_string=mongo_connection_string, + mongo_database=mongo_database, + export_to_kafka=export_to_kafka, + metadata=metadata, + logger=logger, + notification_handler=notification_handler + ) self.opc_managers = {} self.resource_manager = ResourceManager( - redis_host, redis_port, lease_ttl, heartbeat_ttl, pod_id, redis_username, redis_password + host=redis_host, + port=redis_port, + lease_ttl=lease_ttl, + heartbeat_ttl=heartbeat_ttl, + pod_id=pod_id, + metadata=metadata, + logger=logger, + notification_handler=notification_handler, + username=redis_username, + password=redis_password, ) self.number_of_slots = 0 self.poll_interval = poll_interval - self.logger = logger self.managed_tags = {} self.opc_servers = {} - self.notification_handler = notification_handler self.pod_id = pod_id + self.metadata = metadata - def initialize_opc_from_config(self, server_config: dict, - data_manager: DataManager, logger: Logger) -> OpcManager | None: + BaseActivity.__init__(self, logger=logger, + notification_handler=notification_handler) + + def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None: """ Initializes an OPC Manager instance using the provided server configuration. Args: @@ -63,12 +81,17 @@ class IngestorManager(): self.logger.info( f"Initializing OpcManager at {server_config['url']}") manager = OpcManager( - server_config['name'], server_config['url'], - data_manager, logger, server_config['server_uri'], - self.notification_handler, self.pod_id, server_config.get( - 'cert_path'), - server_config.get('private_key_path'), - server_config.get('server_cert_path') + name=server_config['name'], + url=server_config['url'], + data_manager=self.data_manager, + logger=self.logger, + server_uri=server_config['server_uri'], + notification_handler=self.notification_handler, + pod_id=self.pod_id, + metadata=self.metadata, + cert_path=server_config.get('cert_path'), + private_key_path=server_config.get('private_key_path'), + server_cert_path=server_config.get('server_cert_path') ) manager.config = server_config @@ -76,7 +99,8 @@ class IngestorManager(): except Exception as e: trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=self.metadata, notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}', message=f'Error initializing OPC manager: {e}', block="opc_manager", @@ -135,7 +159,7 @@ class IngestorManager(): f"Initializing OPC manager for server {server}" ) server_instance = self.initialize_opc_from_config( - server_config, self.data_manager, self.logger + server_config ) elif server_instance.config != server_config: @@ -145,7 +169,7 @@ class IngestorManager(): server_instance.disconnect() del self.opc_managers[server] server_instance = self.initialize_opc_from_config( - server_config, self.data_manager, self.logger + server_config ) else: self.logger.debug( @@ -416,7 +440,8 @@ class IngestorManager(): metrics.OPC_SUBSCRIPTION_ERRORS.labels( pod_id=self.pod_id, server=server, slot=slot).inc() trace = traceback.format_exc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=self.metadata, notification_id=f'OPC_SUBSCRIPTION_ERROR_{slot}:{server}', message=f'Failed to subscribe to tags from {slot}:{server}\n{tags}: {e}', block="opc_manager", diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index cba0286..cdf37b6 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -1,24 +1,23 @@ import json -from logging import Logger from pathlib import Path -from typing import Callable from asyncua.crypto.security_policies import SecurityPolicyBasic256 from asyncua.sync import Client from sientia_do.notifications.models import NotificationLevel -from sientia_do.notifications.handlers import NotificationHandler +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.temporal.activities.base import BaseActivity +from sientia_do.temporal.utils.logger import Logger from ingestor.managers.data_manager import DataManager import ingestor.metrics as metrics -class OpcManager(): +class OpcManager(BaseActivity): def __init__(self, name: str, url: str, data_manager: DataManager, - logger: Logger, server_uri: str, notification_handler: NotificationHandler, pod_id: str, + logger: Logger, server_uri: str, notification_handler: NotificationHandler, pod_id: str, metadata: dict, cert_path: str = None, private_key_path: str = None, server_cert_path: str = None): self.url = url self.name = name self.server_uri = server_uri self.data_queue = {} - self.logger = logger self.non_receive_count = 0 self.client = None self.cert_path = cert_path @@ -27,13 +26,17 @@ class OpcManager(): self.nodes = {} self.subscriptions = {} self.data_manager = data_manager - self.notification_handler = notification_handler self.pod_id = pod_id + self.metadata = metadata + metrics.OPC_CONNECTION_STATUS.labels( pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(0) metrics.OPC_TAGS_SUBSCRIBED.labels( pod_id=self.pod_id, server_name=self.name).set(0) + BaseActivity.__init__(self, logger=logger, + notification_handler=notification_handler) + def __str__(self): return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \ f"nodes={self.nodes}, subscriptions={self.subscriptions}" @@ -296,7 +299,8 @@ class OpcManager(): if self.nodes[node]['cycle_rule']['cycle_count'] >= 5: name = config['tag_name'] cycles = self.nodes[node]['cycle_rule']['cycle_count'] - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=self.metadata, notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED', message=f'{cycles} cycles without receive from {node}:{name}', block="opc_manager", @@ -314,7 +318,8 @@ class OpcManager(): metrics.OPC_CYCLES_WITHOUT_DATA.labels( pod_id=self.pod_id, server_name=self.name).set(self.non_receive_count) if self.non_receive_count >= 5: - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=self.metadata, notification_id=f'OPC_LISTENNING_STOPPED__{self.name}', message=f'{self.non_receive_count} cycles without ' f'receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}', @@ -324,7 +329,8 @@ class OpcManager(): if self.non_receive_count >= 15: metrics.OPC_RECONNECTIONS_TOTAL.labels( pod_id=self.pod_id, server_name=self.name).inc() - self.notification_handler.build_and_send_notification( + self.send_notification( + metadata=self.metadata, notification_id=f'OPC_CONNECTION_RETRY__{self.name}', message=f'Retrying to connect to server {self.name}', block="opc_manager", diff --git a/ingestor/managers/resource_manager.py b/ingestor/managers/resource_manager.py index d958d85..9dc3ddc 100644 --- a/ingestor/managers/resource_manager.py +++ b/ingestor/managers/resource_manager.py @@ -3,9 +3,13 @@ from typing import List from redis import Redis from time import time import ingestor.metrics as metrics +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.notifications.models import NotificationLevel +from sientia_do.temporal.utils.logger import Logger +from sientia_do.temporal.activities.base import BaseActivity -class ResourceManager: +class ResourceManager(BaseActivity): def __init__( self, host: str, @@ -13,6 +17,9 @@ class ResourceManager: lease_ttl: int, heartbeat_ttl: int, pod_id: str, + metadata: dict, + logger: Logger, + notification_handler: NotificationHandler, username: str | None = None, password: str | None = None, ) -> None: @@ -28,12 +35,16 @@ class ResourceManager: self.redis.ping() metrics.REDIS_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(1) except Exception as e: - print(f"Failed to connect to Redis: {e}") + logger.error(f"Failed to connect to Redis: {e}") metrics.REDIS_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0) raise self.lease_ttl = lease_ttl self.heartbeat_ttl = heartbeat_ttl + self.metadata = metadata + + BaseActivity.__init__(self, logger=logger, + notification_handler=notification_handler) def _execute_redis_op(self, operation_name: str, func, *args, **kwargs): """Wrapper to execute Redis operations and record metrics.""" @@ -52,7 +63,13 @@ class ResourceManager: metrics.REDIS_OPERATIONS_ERRORS.labels( pod_id=self.pod_id, operation=operation_name ).inc() - print(f"Error in Redis operation '{operation_name}': {e}") + self.send_notification( + metadata=self.metadata, + notification_id=f"REDIS_OPERATION_ERROR_{operation_name}", + message=f"Error in Redis operation '{operation_name}': {e}", + block="redis_manager", + level=NotificationLevel.ERROR, + ) raise def get(self, key: str) -> dict: @@ -151,7 +168,8 @@ class ResourceManager: None """ - self._execute_redis_op("delete", self.redis.delete, f"lease:opc_tags:{tag_id}") + self._execute_redis_op("delete", self.redis.delete, + f"lease:opc_tags:{tag_id}") def get_all_ingestors(self) -> List[str]: """