gitea_admin/sientia-dataops-opc-ingestor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #12 from Aignosi/SIENTIAPDE-1163-alterar-dinamica-de-notificacoes-para-usar-o-mongodb-ao-inves-do-kafka

Browse Source 
Sientiapde 1163 alterar dinamica de notificacoes para usar o mongodb ao inves do kafka
This commit is contained in:
Bruno Domingues
2025-07-22 10:14:35 -03:00
committed by GitHub
parent d27127968a d9c776907d
commit a04bf1cadd
15 changed files with 374 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@@ -173,3 +173,8 @@ cython_debug/
# PyPI configuration file
.pypirc
# VSCode
.vscode/
git_log

13
.vscode/settings.json vendored
View File

@@ -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
}

2
ingestor/app.py
View File

@@ -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.")

114
ingestor/ingestor.py
View File

@@ -1,9 +1,9 @@
from logging import Formatter, StreamHandler, getLogger
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 +23,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,26 +54,27 @@ 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.logger = get_logger(__name__)
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"
)
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 = {
'model_id': '-',
'model_name': '-',
'workflow_name': 'OPC_INGESTOR',
'schema_name': 'OPC_INGESTOR',
}
self.ingestor_manager = None
def shutdown(self):
@@ -79,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.
@@ -144,22 +127,23 @@ class Ingestor:
"""
self.ingestor_manager = IngestorManager(
self.kafka_servers,
{
kafka_servers=self.kafka_servers,
redis_data={
'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
@@ -167,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)
@@ -214,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
@@ -246,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)
@@ -263,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
@@ -337,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

20
ingestor/managers/data_manager.py
View File

@@ -1,24 +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.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:
@@ -85,14 +87,16 @@ class DataManager:
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(
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."""
@@ -163,7 +167,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",
@@ -186,7 +191,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",

65
ingestor/managers/ingestor_manager.py
View File

@@ -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",

26
ingestor/managers/opc_manager.py
View File

@@ -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",

26
ingestor/managers/resource_manager.py
View File

@@ -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]:
"""

2
requirements.txt
View File

@@ -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.3
prometheus_client
pymongo

36
tests/unit/managers/test_data_manager.py
View File

@@ -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"]
)

77
tests/unit/managers/test_ingestor_manager.py
View File

@@ -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",

28
tests/unit/managers/test_opc_manager.py
View File

@@ -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(

88
tests/unit/managers/test_resource_manager.py
View File

@@ -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,
)

60
tests/unit/test_ingestor.py
View File

@@ -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()

38
values.yaml
View File

@@ -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:
@@ -111,11 +111,28 @@ 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
metrics:
enabled: true
type: ClusterIP
port: 9090
targetPort: 9090
name: metrics
# Configuração do ServiceMonitor para o Prometheus Operator
# ref: https://github.com/prometheus-operator/prometheus-operator
@@ -134,14 +151,14 @@ 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
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"
@@ -174,7 +191,7 @@ env:
- name: LOG_LEVEL
value: "DEBUG"
- name: HTTP_SERVER_PORT
value: "4840"
value: "9090"
- name: MONGODB_USERNAME
@@ -194,6 +211,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 \