From 3e6ab33ef40b3a0d22ca2fe1f30eab411a39b057 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 16 Jul 2025 08:59:24 -0300 Subject: [PATCH 01/10] SIENTIAPDE-1163 Update requirements and refactor Ingestor class for MongoDB integration - Updated sientia-dataops-library dependency version to 1.3.0 in requirements.txt. - Refactored Ingestor class to use CoreNotificationHandler and added MongoDB connection parameters to improve configuration management. --- ingestor/ingestor.py | 18 ++++++++++++------ requirements.txt | 2 +- 2 files changed, 13 insertions(+), 7 deletions(-) diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 528bf0f..268a520 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -3,7 +3,8 @@ from os import getenv from copy import deepcopy from typing import Dict, Any -from sientia_do.notifications.handlers import NotificationHandler +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.temporal.utils.logger import get_logger from ingestor.managers.ingestor_manager import IngestorManager @@ -23,6 +24,10 @@ class Ingestor: HEARTBEAT_TTL (int): Time-to-live for heartbeats in seconds. Defaults to 20. HOSTNAME (str): Identifier for the current pod or host. Defaults to "localhost". POLL_INTERVAL (int): Interval in seconds for polling operations. Defaults to 5. + MONGODB_URL (str): URL of the MongoDB server. Defaults to "localhost:27017". + MONGODB_USERNAME (str): Username for the MongoDB server. Defaults to "sientia". + MONGODB_PASSWORD (str): Password for the MongoDB server. Defaults to "sientia". + MONGODB_DATABASE (str): Name of the MongoDB database. Defaults to "sientia". Attributes: kafka_servers (list): List of Kafka server addresses. redis_host (str): Hostname of the Redis server. @@ -50,17 +55,18 @@ class Ingestor: self.heartbeat_ttl = int(getenv("HEARTBEAT_TTL", "20")) self.pod_id = getenv("HOSTNAME", "localhost") self.poll_interval = int(getenv("POLL_INTERVAL", "5")) - mongo_url = getenv("MONGODB_URL", 'localhost:27017') - mongo_username = getenv("MONGODB_USERNAME", 'sientia') - mongo_password = getenv("MONGODB_PASSWORD", 'sientia') - self.mongo_database = getenv("MONGODB_DATABASE", 'sientia') + mongo_url = getenv("MONGODB_URL", "localhost:27017") + mongo_username = getenv("MONGODB_USERNAME", "sientia") + mongo_password = getenv("MONGODB_PASSWORD", "sientia") + self.mongo_database = getenv("MONGODB_DATABASE", "sientia") 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.notification_handler = NotificationHandler( - servers=self.kafka_servers, + connection_string=self.mongo_connection_string, + database=self.mongo_database, logger=self.logger, project_name="OPC_INGESTOR" ) diff --git a/requirements.txt b/requirements.txt index 47e8725..f7e0a73 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ asyncua==1.1.5 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.2.0 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.0 prometheus_client pymongo \ No newline at end of file From 0885544bc7efcb468c1fb08c0ebee636e1b402b4 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 16 Jul 2025 10:16:02 -0300 Subject: [PATCH 02/10] SIENTIAPDE-1163 Update values.yaml for image tag and GITHUB_BRANCH - Updated image tag to "0.2.4" for the latest version. - Changed GITHUB_BRANCH value to "SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka". - Removed .vscode/settings.json as it is no longer needed. --- .vscode/settings.json | 13 ------------- ingestor/managers/data_manager.py | 3 ++- values.yaml | 5 +++-- 3 files changed, 5 insertions(+), 16 deletions(-) delete mode 100644 .vscode/settings.json diff --git a/.vscode/settings.json b/.vscode/settings.json deleted file mode 100644 index e7d55f8..0000000 --- a/.vscode/settings.json +++ /dev/null @@ -1,13 +0,0 @@ -{ - "python.testing.pytestArgs": ["."], - "python.testing.unittestEnabled": false, - "python.testing.pytestEnabled": true, - "sonarlint.connectedMode.project": { - "connectionId": "sonardev-sientia-ai", - "projectKey": "Aignosi_sientia-dataops-opc-ingestor_1642c8bf-a148-4911-8362-0903d7fef99a" - }, - "python.languageServer": "Jedi", - "python.analysis.typeCheckingMode": "standard", - "editor.suggestSelection": "first", - "windsurfPyright.disableLanguageServices": true -} diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 9c689ed..5f1e634 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -7,12 +7,13 @@ from kafka import KafkaProducer from kafka.errors import NoBrokersAvailable from sientia_do.notifications.handlers import NotificationHandler from sientia_do.notifications.models import NotificationLevel +from sientia_do. import traceback import ingestor.metrics as metrics import os -class DataManager: +class DataManager(): def __init__( self, kafka_servers: str, diff --git a/values.yaml b/values.yaml index cd916b5..1d45d45 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.2.2" + tag: "0.2.4" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: @@ -140,7 +140,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-opc-ingestor.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1110-criar-testes-e-2-e" + value: "SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka" - name: PYTHON_APP value: "ingestor.app" @@ -193,6 +193,7 @@ ssh: knownHostsPath: /mnt/known_hosts # kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp + # helm upgrade --install sientia-opc-ingestor sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.4.0-uat # kubectl create secret generic git-ssh-key-sientia-opc-ingestor \ From 5132162eaaa69a58a1ad8cdf0df18fc2777d5630 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 16 Jul 2025 10:54:34 -0300 Subject: [PATCH 03/10] 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]: """ From b87bfc8e681ed01e6369b16804d59196944b27c9 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 16 Jul 2025 11:17:43 -0300 Subject: [PATCH 04/10] SIENTIAPDE-1163 Refactor logging statements in Ingestor and app modules to use f-strings for improved readability and consistency. Updated error and info logs to enhance clarity in the Ingestor class. --- ingestor/app.py | 2 +- ingestor/ingestor.py | 38 +++++++++++++++----------------------- 2 files changed, 16 insertions(+), 24 deletions(-) diff --git a/ingestor/app.py b/ingestor/app.py index d742bf5..e2657f7 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -21,7 +21,7 @@ def main(): except Exception as e: metrics.APP_ERRORS_TOTAL.labels( pod_id=POD_ID).inc() # Increment errors - ingestor.logger.error("Failed to prepare ingestor: %s", e) + ingestor.logger.error(f"Failed to prepare ingestor: {e}") exit_signal.set() ingestor.logger.info("Ingestor prepared. Starting main loop.") diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 28a8644..18dc49c 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -128,7 +128,7 @@ class Ingestor: self.ingestor_manager = IngestorManager( kafka_servers=self.kafka_servers, - redis_config={ + redis_data={ 'host': self.redis_host, 'port': self.redis_port, 'username': self.redis_username, @@ -151,7 +151,7 @@ class Ingestor: # Get slot lease acquired = self.ingestor_manager.get_slot_leases() - self.logger.info("Acquired slots: %s", acquired) + self.logger.info(f"Acquired slots: {acquired}") self.handle_acquired_tags(acquired) @@ -198,14 +198,14 @@ class Ingestor: if available_slots > 0 and lacking_ingestors > 0: # Some ingestors are innactive, so theres "available_slots" slots available - self.logger.info("Slots available: %s", available_slots) + self.logger.info(f"Slots available: {available_slots}") # Get slot lease self.ingestor_manager.get_slot_leases(available_slots) elif lacking_ingestors <= 0 and slot_diff > 0: - self.logger.info("Extra slots available: %s", slot_diff) + self.logger.info(f"Extra slots available: {slot_diff}") # There's enough slots for all ingestors, but this ingestor has more than one slot # So we need to drop the extra leases @@ -230,16 +230,13 @@ class Ingestor: """ self.logger.debug( - "Current managed tags: %s", self.ingestor_manager.managed_tags - ) + f"Current managed tags: {self.ingestor_manager.managed_tags}") self.ingestor_manager.update_opc_servers() new_managed_tags = deepcopy(self.ingestor_manager.managed_tags) self.logger.debug( - "Comparing new managed tags %s with old managed tags %s", - new_managed_tags, - old_managed_tags, + f"Comparing new managed tags {new_managed_tags} with old managed tags {old_managed_tags}" ) keys = set(new_managed_tags) | set(old_managed_tags) @@ -247,22 +244,22 @@ class Ingestor: changes = {k: (new_managed_tags.get(k), old_managed_tags.get(k)) for k in keys if new_managed_tags.get(k) != old_managed_tags.get(k)} - self.logger.debug("Changes: %s", changes) + self.logger.debug(f"Changes: {changes}") for slot, config in new_managed_tags.items(): if slot not in old_managed_tags: - self.logger.debug("Subscribing to new slot %s", slot) + self.logger.debug(f"Subscribing to new slot {slot}") self.ingestor_manager.subscribe_to_tags({slot: config}) continue if config != old_managed_tags[slot]: - self.logger.debug("Resubscribing to slot %s", slot) + self.logger.debug(f"Resubscribing to slot {slot}") self.ingestor_manager.unsubscribe_slot(slot) self.ingestor_manager.subscribe_to_tags({slot: config}) for slot in old_managed_tags.keys(): if slot not in new_managed_tags: - self.logger.debug("Unsubscribing from slot %s", slot) + self.logger.debug(f"Unsubscribing from slot {slot}") self.ingestor_manager.unsubscribe_slot(slot) # Ensure the gauge is updated after any potential changes here @@ -321,16 +318,11 @@ class Ingestor: ) self.logger.debug( - "Active ingestors: %s, " - "Number of slots: %s, " - "Number of leases: %s, " - "Managed tags: %s, " - "Managed servers: %s", - ingestors, - number_of_slots, - number_of_leases, - self.ingestor_manager.managed_tags, - self.ingestor_manager.opc_managers, + f"Active ingestors: {ingestors}, " + f"Number of slots: {number_of_slots}, " + f"Number of leases: {number_of_leases}, " + f"Managed tags: {self.ingestor_manager.managed_tags}, " + f"Managed servers: {self.ingestor_manager.opc_managers}" ) if not self.ingestor_manager.managed_tags: # No slots acquired From 3088701be0f468498e20787d0dd1616681c25b6e Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 17 Jul 2025 10:50:54 -0300 Subject: [PATCH 05/10] SIENTIAPDE-1163 Update sientia-dataops-library dependency version to 1.3.3 in requirements.txt --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index f7e0a73..6c99d3f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ asyncua==1.1.5 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.0 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.3 prometheus_client pymongo \ No newline at end of file From c18d6e514270fac7522b4bd0430bd43e4d4e842f Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 16 Jul 2025 17:33:09 -0300 Subject: [PATCH 06/10] Add port configuration for serviceMonitor in values.yaml. --- values.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/values.yaml b/values.yaml index 1d45d45..be192ef 100644 --- a/values.yaml +++ b/values.yaml @@ -134,6 +134,7 @@ serviceMonitor: # Configurações de relabeling adicionais, se necessário. # ref: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config relabelings: [] + port: opc-server env: # Entrypoint variables From d1215344b86a5942d26aa82ab1c7d925a01f1996 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 17 Jul 2025 10:52:33 -0300 Subject: [PATCH 07/10] SIENTIAPDE-1163 Update values.yaml to restructure service definitions - Changed the service configuration to a nested structure under 'services'. - Introduced 'api' and 'opc' services with individual settings, both initially disabled. - Updated ports for the 'api' service to 4841 while keeping the 'opc' service at 4840. --- values.yaml | 20 +++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/values.yaml b/values.yaml index be192ef..db2b1f2 100644 --- a/values.yaml +++ b/values.yaml @@ -111,11 +111,21 @@ tolerations: [] affinity: {} -service: - enabled: true - type: ClusterIP - port: 4840 - targetPort: 4840 +services: + api: + enabled: false + type: ClusterIP + port: 4841 + targetPort: 4841 + name: api + + opc: + enabled: false + type: ClusterIP + port: 4840 + targetPort: 4840 + name: server + # Configuração do ServiceMonitor para o Prometheus Operator # ref: https://github.com/prometheus-operator/prometheus-operator From 6ad0c5d7681000bd0d97f1cf2b44e157ddfb764c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 17 Jul 2025 12:42:04 -0300 Subject: [PATCH 08/10] SIENTIAPDE-1163 Update .gitignore to include git_log and ensure .vscode/ is excluded --- .gitignore | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 3255fbb..0d8a542 100644 --- a/.gitignore +++ b/.gitignore @@ -175,4 +175,6 @@ cython_debug/ .pypirc # VSCode -.vscode/ \ No newline at end of file +.vscode/ + +git_log \ No newline at end of file From 9309d7591e0408f1a926f9af1b07fa5ba5712ac1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 17 Jul 2025 14:27:13 -0300 Subject: [PATCH 09/10] SIENTIAPDE-1163 SIENTIAPDE-1163 Refactor Ingestor and DataManager classes to enhance metadata handling - Updated Ingestor class to include additional metadata fields: 'workflow_name' and 'schema_name'. - Modified DataManager to accept metadata during initialization. - Adjusted unit tests to validate the new metadata structure and ensure proper functionality across various managers. - Removed unused logger initialization in tests for cleaner code. --- ingestor/ingestor.py | 6 +- ingestor/managers/data_manager.py | 2 + tests/unit/managers/test_data_manager.py | 36 ++++++-- tests/unit/managers/test_ingestor_manager.py | 77 ++++++++++++----- tests/unit/managers/test_opc_manager.py | 28 +++++-- tests/unit/managers/test_resource_manager.py | 88 +++++++++++++++----- tests/unit/test_ingestor.py | 60 +++++-------- 7 files changed, 203 insertions(+), 94 deletions(-) diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 18dc49c..9614302 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -70,10 +70,10 @@ class Ingestor: ) self.metadata = { - 'schedule_name': 'OPC_INGESTOR', - 'trigger': 'INGESTOR', + 'model_id': '-', 'model_name': '-', - 'model_id': '-' + 'workflow_name': 'OPC_INGESTOR', + 'schema_name': 'OPC_INGESTOR', } self.ingestor_manager = None diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 9fd50f6..68dbe06 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -87,6 +87,8 @@ class DataManager(BaseActivity): self.mongo_client = MongoClient(self.connection_string) self.mongo_client.server_info() + self.metadata = metadata + self.mongo_db = self.mongo_client[self.database] logger.info( diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index da78e26..ed157cc 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/tests/unit/managers/test_data_manager.py @@ -4,20 +4,35 @@ from kafka.errors import NoBrokersAvailable from sientia_do.notifications.models import NotificationLevel from ingestor.managers.data_manager import DataManager +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @fixture @patch("ingestor.managers.data_manager.KafkaProducer") @patch("ingestor.managers.data_manager.MongoClient") def data_manager(mongo, kafka): - return DataManager( + + data_manager = DataManager( kafka_servers="localhost:9092", mongo_connection_string="mongodb://localhost:27017", mongo_database="sientia", export_to_kafka=True, logger=MagicMock(), - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) + data_manager.send_notification = MagicMock() + + return data_manager + @patch("ingestor.managers.data_manager.KafkaProducer") @patch("ingestor.managers.data_manager.MongoClient") @@ -25,6 +40,7 @@ def test___init___success(mongo, kafka): logger_mock = MagicMock() data_manager = DataManager( + metadata=metadata["metadata"], kafka_servers="localhost:9092", mongo_connection_string="mongodb://localhost:27017", mongo_database="sientia", @@ -61,7 +77,8 @@ def test___init___second_attempt(mongo, kafka): mongo_database="sientia", export_to_kafka=True, logger=logger_mock, - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) kafka.assert_any_call( @@ -99,7 +116,8 @@ def test___init___failure_max_attempts(mongo, kafka): mongo_database="sientia", export_to_kafka=True, logger=logger_mock, - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) except NoBrokersAvailable as e: assert str( @@ -255,12 +273,13 @@ def test_publish_error(traceback, data_manager): ) # Check if the error was logged - data_manager.notification_handler.build_and_send_notification.assert_called_once_with( + data_manager.send_notification.assert_called_once_with( notification_id=f"KAFKA_PRODUCER_ERROR_{topic}", message=f"Error publishing message to topic {topic}: Test error", block="kafka_producer", level=NotificationLevel.ERROR, - attachment_content=traceback.format_exc.return_value + attachment_content=traceback.format_exc.return_value, + metadata=metadata["metadata"] ) @@ -271,10 +290,11 @@ def test_publish_error_mongo(data_manager): data_manager.publish("test_topic", {"key": "value"}) - data_manager.notification_handler.build_and_send_notification.assert_called_once_with( + data_manager.send_notification.assert_called_once_with( notification_id="MONGO_PRODUCER_ERROR_test_topic", message="Error inserting message to MongoDB: Test error", block="mongo_producer", level=NotificationLevel.ERROR, - attachment_content=ANY + attachment_content=ANY, + metadata=metadata["metadata"] ) diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 9d27c55..f6ace9f 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -3,12 +3,21 @@ from pytest import fixture from sientia_do.notifications.models import NotificationLevel from ingestor.managers.ingestor_manager import IngestorManager +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @fixture @patch('ingestor.managers.ingestor_manager.DataManager') @patch('ingestor.managers.ingestor_manager.ResourceManager') def ingestor_manager(data_manager_mock, resource_manager_mock): - return IngestorManager( + ingestor = IngestorManager( kafka_servers="localhost:9092", redis_data={ "host": "localhost", @@ -22,9 +31,14 @@ def ingestor_manager(data_manager_mock, resource_manager_mock): mongo_database="sientia", export_to_kafka=False, logger=MagicMock(), - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) + ingestor.send_notification = MagicMock() + + return ingestor + @patch('ingestor.managers.ingestor_manager.OpcManager') @patch('ingestor.managers.ingestor_manager.DataManager') @@ -46,15 +60,32 @@ def test___init__(notification_handler_mock, resource_manager_mock, data_manager mongo_database="sientia", export_to_kafka=False, logger=MagicMock(), - notification_handler=MagicMock() + notification_handler=MagicMock(), + metadata=metadata["metadata"], ) opc_manager_mock.assert_not_called() data_manager_mock.assert_called_once_with( - "localhost:9092", "mongodb://localhost:27017", "sientia", False, - ingestor.logger, ingestor.notification_handler) + kafka_servers="localhost:9092", + mongo_connection_string="mongodb://localhost:27017", + mongo_database="sientia", + export_to_kafka=False, + metadata=metadata["metadata"], + logger=ingestor.logger, + notification_handler=ingestor.notification_handler, + ) resource_manager_mock.assert_called_once_with( - "localhost", 6379, 60, 60, "test_pod", None, None) + host="localhost", + port=6379, + lease_ttl=60, + heartbeat_ttl=60, + pod_id="test_pod", + metadata=metadata["metadata"], + logger=ingestor.logger, + notification_handler=ingestor.notification_handler, + username=None, + password=None, + ) assert ingestor.poll_interval == 5 assert ingestor.managed_tags == {} assert ingestor.opc_servers == {} @@ -77,14 +108,20 @@ def test_initialize_opc_from_config(opc_manager, ingestor_manager): opc_manager.return_value = MagicMock() result = ingestor_manager.initialize_opc_from_config( - server_config, ingestor_manager.data_manager, ingestor_manager.logger) + server_config) opc_manager.assert_called_once_with( - server_config['name'], server_config['url'], ingestor_manager.data_manager, ingestor_manager.logger, - server_config['server_uri'], ingestor_manager.notification_handler, - server_config['pod_id'], - server_config['cert_path'], server_config['private_key_path'], - server_config['server_cert_path'] + name=server_config['name'], + url=server_config['url'], + data_manager=ingestor_manager.data_manager, + logger=ingestor_manager.logger, + server_uri=server_config['server_uri'], + notification_handler=ingestor_manager.notification_handler, + pod_id=server_config['pod_id'], + cert_path=server_config['cert_path'], + private_key_path=server_config['private_key_path'], + server_cert_path=server_config['server_cert_path'], + metadata=metadata["metadata"], ) assert result == opc_manager.return_value @@ -107,12 +144,13 @@ def test_initialize_opc_from_config_exception(traceback_mock, opc_manager, inges opc_manager.side_effect = Exception("Initialization error") result = ingestor_manager.initialize_opc_from_config( - server_config, ingestor_manager.data_manager, ingestor_manager.logger) + server_config) assert result is None traceback_mock.format_exc.assert_called_once() - ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with( + ingestor_manager.send_notification.assert_called_once_with( + metadata=metadata["metadata"], notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}', message='Error initializing OPC manager: Initialization error', block="opc_manager", @@ -131,7 +169,7 @@ def test_update_opc_servers(metrics, opc_manager, ingestor_manager): manager3 = MagicMock( config={"config": "config3"}) - def mock_initialize_from_config(config, data_manager, logger): + def mock_initialize_from_config(config): if config == {"config": "config1"}: return manager1 elif config == {"config": "config2"}: @@ -169,11 +207,11 @@ def test_update_opc_servers(metrics, opc_manager, ingestor_manager): assert len(ingestor_manager.opc_managers) == 3 ingestor_manager.initialize_opc_from_config.assert_any_call( - {"config": "config1"}, ingestor_manager.data_manager, ingestor_manager.logger) + {"config": "config1"}) ingestor_manager.initialize_opc_from_config.assert_any_call( - {"config": "config2"}, ingestor_manager.data_manager, ingestor_manager.logger) + {"config": "config2"}) ingestor_manager.initialize_opc_from_config.assert_any_call( - {"config": "config5"}, ingestor_manager.data_manager, ingestor_manager.logger) + {"config": "config5"}) assert ingestor_manager.initialize_opc_from_config.call_count == 3 assert ingestor_manager.opc_managers['server1'].config == { @@ -470,7 +508,8 @@ def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager): traceback_mock.format_exc.assert_called_once() - ingestor_manager.notification_handler.build_and_send_notification.assert_called_once_with( + ingestor_manager.send_notification.assert_called_once_with( + metadata=metadata["metadata"], notification_id='OPC_SUBSCRIPTION_ERROR_slot1:server1', message='Failed to subscribe to tags from slot1:server1\n{\'tags\': \'config1\'}: Subscription error', block="opc_manager", diff --git a/tests/unit/managers/test_opc_manager.py b/tests/unit/managers/test_opc_manager.py index f9be3dc..386f2ae 100644 --- a/tests/unit/managers/test_opc_manager.py +++ b/tests/unit/managers/test_opc_manager.py @@ -33,6 +33,15 @@ tags = { }, } +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @fixture def raw_opc_manager(): @@ -44,6 +53,7 @@ def raw_opc_manager(): "opc.tcp://localhost:4840", MagicMock(), "localhost", + metadata=metadata["metadata"], ) @@ -53,6 +63,7 @@ def opc_manager(raw_opc_manager): raw_opc_manager.cert_path = "cert.pem" raw_opc_manager.private_key_path = "private_key.pem" raw_opc_manager.server_cert_path = "server_cert.pem" + raw_opc_manager.send_notification = MagicMock() return raw_opc_manager @@ -414,7 +425,6 @@ def test_check_cycles_no_notification(opc_manager): "cycle_rule": {"cycle_increment": 1.0, "cycle_count": 3.0}, } } - opc_manager.notification_handler.build_and_send_notification = MagicMock() opc_manager.check_cycles() @@ -422,7 +432,7 @@ def test_check_cycles_no_notification(opc_manager): assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][ "cycle_count" ] == pytest.approx(4.0) - opc_manager.notification_handler.build_and_send_notification.assert_not_called() + opc_manager.send_notification.assert_not_called() def test_check_cycles_triggers_notification(opc_manager): @@ -441,11 +451,12 @@ def test_check_cycles_triggers_notification(opc_manager): assert opc_manager.nodes["ns=3;i=1001"]["cycle_rule"][ "cycle_count" ] == pytest.approx(5.5) - opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( + opc_manager.send_notification.assert_called_once_with( notification_id="TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED", message="5.5 cycles without receive from ns=3;i=1001:Counter", block="opc_manager", level=NotificationLevel.WARNING, + metadata=metadata["metadata"], ) @@ -473,16 +484,16 @@ def test_check_opc_listenning_no_notification(metrics, opc_manager): def test_check_opc_listenning_warning_notification(opc_manager): opc_manager.non_receive_count = 4 - opc_manager.notification_handler.build_and_send_notification = MagicMock() result = opc_manager.check_opc_listenning() assert opc_manager.non_receive_count == 5 - opc_manager.notification_handler.build_and_send_notification.assert_called_once_with( + opc_manager.send_notification.assert_called_once_with( notification_id=f"OPC_LISTENNING_STOPPED__{opc_manager.name}", message=f"5 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}", block="opc_manager", level=NotificationLevel.ERROR, + metadata=metadata["metadata"], ) assert result is False @@ -498,14 +509,15 @@ def test_check_opc_listenning_error_notification_and_retry(metrics, opc_manager) assert opc_manager.non_receive_count == 15 # Should be called twice: once for 5, once for 15 - assert opc_manager.notification_handler.build_and_send_notification.call_count == 2 - calls = opc_manager.notification_handler.build_and_send_notification.call_args_list + assert opc_manager.send_notification.call_count == 2 + calls = opc_manager.send_notification.call_args_list # First call: 5 cycles warning assert calls[0].kwargs == dict( notification_id=f"OPC_LISTENNING_STOPPED__{opc_manager.name}", message=f"15 cycles without receive from OPC {opc_manager.name}. Tags: {json.dumps(opc_manager.nodes)}", block="opc_manager", level=NotificationLevel.ERROR, + metadata=metadata["metadata"], ) # Second call: 15 cycles retry assert calls[1].kwargs == dict( @@ -513,6 +525,7 @@ def test_check_opc_listenning_error_notification_and_retry(metrics, opc_manager) message=f"Retrying to connect to server {opc_manager.name}", block="opc_manager", level=NotificationLevel.ERROR, + metadata=metadata["metadata"], ) assert result is True @@ -540,6 +553,7 @@ def test_init_metrics_calls_correct_metric_methods(mock_metrics): server_uri="opc.tcp://init.test:4840/uri", notification_handler=MagicMock(), pod_id="init_pod_localhost", + metadata=metadata["metadata"], ) mock_metrics.OPC_CONNECTION_STATUS.labels.assert_called_once_with( diff --git a/tests/unit/managers/test_resource_manager.py b/tests/unit/managers/test_resource_manager.py index 19ca732..0e528a2 100644 --- a/tests/unit/managers/test_resource_manager.py +++ b/tests/unit/managers/test_resource_manager.py @@ -1,13 +1,36 @@ from unittest.mock import MagicMock, patch from pytest import fixture, raises +from sientia_do.notifications.models import NotificationLevel from ingestor.managers.resource_manager import ResourceManager +metadata = { + "metadata": { + "model_id": "test_model", + "model_name": "test_model", + "workflow_name": "test_workflow", + "schema_name": "test_schedule", + }, +} + @fixture @patch("ingestor.managers.resource_manager.Redis") def resource_manager(redis): - return ResourceManager("localhost", 6379, 10, 10, "pod_id") + resource_manager = ResourceManager( + host="localhost", + port=6379, + lease_ttl=10, + heartbeat_ttl=10, + pod_id="pod_id", + metadata=metadata["metadata"], + logger=MagicMock(), + notification_handler=MagicMock(), + ) + + resource_manager.send_notification = MagicMock() + + return resource_manager def test_get_success(resource_manager): @@ -54,7 +77,8 @@ def test_renew_tag_lease_success(resource_manager): result = resource_manager.renew_tag_lease("tag_id") assert result is True resource_manager.redis.get.assert_called_once_with("lease:opc_tags:tag_id") - resource_manager.redis.expire.assert_called_once_with("lease:opc_tags:tag_id", 10) + resource_manager.redis.expire.assert_called_once_with( + "lease:opc_tags:tag_id", 10) def test_renew_tag_lease_failure(resource_manager): @@ -69,7 +93,8 @@ def test_renew_tag_lease_failure(resource_manager): def test_drop_tag_lease(resource_manager): resource_manager.redis.delete.return_value = True resource_manager.drop_tag_lease("tag_id") - resource_manager.redis.delete.assert_called_once_with("lease:opc_tags:tag_id") + resource_manager.redis.delete.assert_called_once_with( + "lease:opc_tags:tag_id") def test_get_all_ingestors(resource_manager): @@ -106,7 +131,16 @@ def test_init_connection_failure(monkeypatch): mock_metrics.labels.return_value = mock_status with raises(Exception, match="Connection failed"): - ResourceManager("localhost", 6379, 10, 10, "pod_id") + ResourceManager( + host="localhost", + port=6379, + lease_ttl=10, + heartbeat_ttl=10, + pod_id="pod_id", + metadata=metadata["metadata"], + logger=MagicMock(), + notification_handler=MagicMock(), + ) mock_metrics.labels.assert_called_once_with(pod_id="pod_id") mock_status.set.assert_called_once_with(0) @@ -118,7 +152,8 @@ def test_init_ping_failure(monkeypatch): mock_redis_instance.ping.side_effect = Exception("Ping failed") mock_redis_class = MagicMock(return_value=mock_redis_instance) - monkeypatch.setattr("ingestor.managers.resource_manager.Redis", mock_redis_class) + monkeypatch.setattr( + "ingestor.managers.resource_manager.Redis", mock_redis_class) # Test that the exception is raised and metrics are set properly with patch("ingestor.metrics.REDIS_CONNECTION_STATUS") as mock_metrics: @@ -126,7 +161,16 @@ def test_init_ping_failure(monkeypatch): mock_metrics.labels.return_value = mock_status with raises(Exception, match="Ping failed"): - ResourceManager("localhost", 6379, 10, 10, "pod_id") + ResourceManager( + host="localhost", + port=6379, + lease_ttl=10, + heartbeat_ttl=10, + pod_id="pod_id", + metadata=metadata["metadata"], + logger=MagicMock(), + notification_handler=MagicMock(), + ) mock_metrics.labels.assert_called_once_with(pod_id="pod_id") mock_status.set.assert_called_once_with(0) @@ -172,19 +216,23 @@ def test_execute_redis_op_exception(resource_manager): with patch("ingestor.managers.resource_manager.time", return_value=100): with patch("ingestor.metrics.REDIS_OPERATIONS_ERRORS") as mock_errors: - with patch("builtins.print") as mock_print: - mock_errors_labels = MagicMock() - mock_errors.labels.return_value = mock_errors_labels + mock_errors_labels = MagicMock() + mock_errors.labels.return_value = mock_errors_labels - # Execute the operation and expect an exception - with raises(Exception, match="Operation failed"): - resource_manager._execute_redis_op("test_op", mock_func, "arg1") + # Execute the operation and expect an exception + with raises(Exception, match="Operation failed"): + resource_manager._execute_redis_op( + "test_op", mock_func, "arg1") - # Verify metrics and error handling - mock_errors.labels.assert_called_once_with( - pod_id="pod_id", operation="test_op" - ) - mock_errors_labels.inc.assert_called_once() - mock_print.assert_called_once_with( - "Error in Redis operation 'test_op': Operation failed" - ) + # Verify metrics and error handling + mock_errors.labels.assert_called_once_with( + pod_id="pod_id", operation="test_op" + ) + mock_errors_labels.inc.assert_called_once() + resource_manager.send_notification.assert_called_once_with( + metadata=metadata["metadata"], + notification_id="REDIS_OPERATION_ERROR_test_op", + message="Error in Redis operation 'test_op': Operation failed", + block="redis_manager", + level=NotificationLevel.ERROR, + ) diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 189c0a6..2f0a42f 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -6,9 +6,8 @@ from ingestor.ingestor import Ingestor @patch("ingestor.ingestor.getenv") -@patch("ingestor.ingestor.Ingestor.init_logger") @patch("ingestor.ingestor.NotificationHandler") -def test___init__(notification_handler, init_logger, getenv): +def test___init__(notification_handler, getenv): getenv.side_effect = [ "localhost:9092,localhost:35", # KAFKA_SERVERS "true", # EXPORT_TO_KAFKA @@ -20,7 +19,7 @@ def test___init__(notification_handler, init_logger, getenv): '200', # HEARTBEAT_TTL "localhost1", # HOSTNAME '50', # POLL_INTERVAL - "mongodb://localhost:27017", # MONGODB_URL + "localhost:27017", # MONGODB_URL "sientia", # MONGODB_USERNAME "sientia", # MONGODB_PASSWORD "sientia" # MONGODB_DATABASE @@ -47,10 +46,16 @@ def test___init__(notification_handler, init_logger, getenv): assert ingestor.heartbeat_ttl == 200 assert ingestor.pod_id == "localhost1" assert ingestor.poll_interval == 50 + assert ingestor.metadata == { + "model_id": "-", + "model_name": "-", + "workflow_name": "OPC_INGESTOR", + "schema_name": "OPC_INGESTOR", + } - init_logger.assert_called_once() notification_handler.assert_called_once_with( - servers=["localhost:9092", "localhost:35"], + connection_string="mongodb://sientia:sientia@localhost:27017", + database="sientia", logger=ingestor.logger, project_name="OPC_INGESTOR" ) @@ -58,9 +63,8 @@ def test___init__(notification_handler, init_logger, getenv): @fixture @patch("ingestor.ingestor.getenv") -@patch("ingestor.ingestor.Ingestor.init_logger") @patch("ingestor.ingestor.NotificationHandler") -def ingestor(_notification_handler, _init_logger, _getenv): +def ingestor(_notification_handler, _getenv): ing = Ingestor() ing.logger = MagicMock() @@ -73,25 +77,6 @@ def ingestor_manager_started(ingestor): return ingestor -@patch("ingestor.ingestor.getLogger") -@patch("ingestor.ingestor.StreamHandler") -@patch("ingestor.ingestor.Formatter") -def test_init_logger(formatter, stream_handler, get_logger, ingestor): - ingestor.logger = None - ingestor.init_logger() - - get_logger.assert_called_once_with('ingestor.ingestor') - stream_handler.assert_called_once() - formatter.assert_called_once_with( - '%(asctime)s - %(name)s - %(levelname)s - %(message)s') - ingestor.logger.setLevel.assert_called_once_with( - getenv("LOG_LEVEL", "INFO")) - ingestor.logger.addHandler.assert_called_once_with( - stream_handler.return_value) - stream_handler.return_value.setFormatter.assert_called_once_with( - formatter.return_value) - - def test_handle_acquired_tags_not_acquired(ingestor_manager_started): ingestor_manager_started.handle_acquired_tags([]) @@ -118,22 +103,23 @@ def test_prepare_ingestor(ingestor_manager_mock, ingestor): ingestor.prepare_ingestor() ingestor_manager_mock.assert_called_once_with( - ingestor.kafka_servers, - { + kafka_servers=ingestor.kafka_servers, + redis_data={ "host": ingestor.redis_host, "port": ingestor.redis_port, "username": ingestor.redis_username, "password": ingestor.redis_password }, - ingestor.lease_ttl, - ingestor.heartbeat_ttl, - ingestor.pod_id, - ingestor.poll_interval, - ingestor.mongo_connection_string, - ingestor.mongo_database, - ingestor.logger, - ingestor.notification_handler, - ingestor.export_to_kafka + lease_ttl=ingestor.lease_ttl, + heartbeat_ttl=ingestor.heartbeat_ttl, + pod_id=ingestor.pod_id, + poll_interval=ingestor.poll_interval, + mongo_connection_string=ingestor.mongo_connection_string, + mongo_database=ingestor.mongo_database, + metadata=ingestor.metadata, + logger=ingestor.logger, + notification_handler=ingestor.notification_handler, + export_to_kafka=ingestor.export_to_kafka, ) ingestor_manager.declare_active.assert_called_once() ingestor_manager.get_slot_leases.assert_called_once() From d9c776907df195a4a177fe334083c17d371484b7 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 17 Jul 2025 14:58:46 -0300 Subject: [PATCH 10/10] SIENTIAPDE-1163 Update values.yaml to add metrics service configuration - Introduced a new 'metrics' service with ClusterIP type, port 9090, and targetPort 9090. - Updated the serviceMonitor port to reference the new 'metrics' service. - Changed HTTP_SERVER_PORT environment variable value to 9090 to align with the new metrics service. --- values.yaml | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/values.yaml b/values.yaml index db2b1f2..673fcdc 100644 --- a/values.yaml +++ b/values.yaml @@ -125,6 +125,13 @@ services: port: 4840 targetPort: 4840 name: server + + metrics: + enabled: true + type: ClusterIP + port: 9090 + targetPort: 9090 + name: metrics # Configuração do ServiceMonitor para o Prometheus Operator @@ -144,8 +151,8 @@ serviceMonitor: # Configurações de relabeling adicionais, se necessário. # ref: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config relabelings: [] - port: opc-server - + port: metrics + env: # Entrypoint variables - name: GITHUB_REPO_URL @@ -184,7 +191,7 @@ env: - name: LOG_LEVEL value: "DEBUG" - name: HTTP_SERVER_PORT - value: "4840" + value: "9090" - name: MONGODB_USERNAME