146 lines
5.0 KiB
Python
146 lines
5.0 KiB
Python
import os
|
|
import traceback
|
|
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
from sientia_do.observability.logger import Logger
|
|
from sientia_do.observability.metrics_controller import MetricsController
|
|
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
|
|
from sientia_do.repository.mongodb_repository import MongoDBRepository
|
|
from sientia_do.temporal.constants import now
|
|
|
|
import ingestor.metrics as metrics
|
|
|
|
|
|
class DataManager(SientiaMonitoring):
|
|
"""
|
|
Manages data persistence operations for the OPC Ingestor.
|
|
|
|
The DataManager is responsible for:
|
|
- Storing OPC data in MongoDB for historical analysis and persistence
|
|
- Managing database connections and ensuring data integrity
|
|
- Providing data access interfaces for other components
|
|
|
|
Args:
|
|
mongo_connection_string (str): MongoDB connection string
|
|
mongo_database (str): MongoDB database name
|
|
metadata (dict): Application metadata for notifications and tracking
|
|
logger (Logger): Logger instance for application logging
|
|
notification_handler (NotificationHandler): Handler for sending notifications
|
|
|
|
Attributes:
|
|
pod_id (str): Pod identifier for metrics labeling
|
|
connection_string (str): MongoDB connection string
|
|
database (str): MongoDB database name
|
|
mongo_client (MongoClient): MongoDB client instance
|
|
metadata (dict): Application metadata
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
mongo_connection_string: str,
|
|
mongo_database: str,
|
|
metadata: dict,
|
|
logger: Logger,
|
|
notification_handler: NotificationHandler,
|
|
metrics_controller: MetricsController,
|
|
) -> None:
|
|
"""
|
|
Initializes the DataManager instance with a MongoDB connection.
|
|
|
|
This constructor establishes a connection to MongoDB and verifies server
|
|
availability.
|
|
|
|
Args:
|
|
mongo_connection_string (str): MongoDB connection string
|
|
mongo_database (str): MongoDB database name
|
|
metadata (dict): Application metadata
|
|
logger (Logger): Logger instance
|
|
notification_handler (NotificationHandler): Notification handler
|
|
"""
|
|
|
|
self.pod_id = os.getenv('HOSTNAME', 'localhost')
|
|
|
|
SientiaMonitoring.__init__(
|
|
self,
|
|
logger=logger,
|
|
metrics_controller=metrics_controller,
|
|
notification_handler=notification_handler,
|
|
)
|
|
|
|
logger.info(
|
|
f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}'
|
|
)
|
|
|
|
self.connection_string = mongo_connection_string
|
|
self.database = mongo_database
|
|
|
|
self.mongo_repository = MongoDBRepository(
|
|
connection_string=self.connection_string,
|
|
database_name=self.database,
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
metrics_controller=metrics_controller,
|
|
)
|
|
|
|
self.metadata = metadata
|
|
|
|
logger.info(f'DataManager initialized with MongoDB servers: {self.connection_string}')
|
|
|
|
def shutdown(self):
|
|
"""
|
|
Gracefully shuts down the DataManager and closes all connections.
|
|
|
|
This method ensures proper cleanup of the MongoDB client connection,
|
|
logging any errors that occur during shutdown while ensuring all
|
|
resources are properly released.
|
|
"""
|
|
try:
|
|
self.mongo_repository.close()
|
|
except Exception as e:
|
|
self.logger.error(f'Error closing MongoDB client: {e}')
|
|
|
|
def __del__(self):
|
|
self.shutdown()
|
|
|
|
async def publish(self, topic: str, data: dict) -> None:
|
|
"""
|
|
Persists a message to the specified MongoDB collection.
|
|
|
|
Args:
|
|
topic (str): The name of the MongoDB collection to persist to.
|
|
data (dict): The message data to be inserted.
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
|
|
try:
|
|
await self.mongo_repository.insert(
|
|
collection_name=topic,
|
|
document={**data, 'inserted_at': now()},
|
|
metadata=self.metadata,
|
|
)
|
|
self.logger.debug(f'Message inserted into MongoDB collection {topic}: {data}')
|
|
|
|
await self.emit_metric(
|
|
metric_object=metrics.TAG_WRITTEN_COUNT,
|
|
tags={
|
|
'pod_id': self.pod_id,
|
|
'tag_name': data['name'],
|
|
'collection_name': topic,
|
|
},
|
|
)
|
|
|
|
except Exception as e:
|
|
trace = traceback.format_exc()
|
|
await self.send_notification_async(
|
|
metadata=self.metadata,
|
|
notification_id=f'MONGO_PRODUCER_ERROR_{topic}',
|
|
message=f'Error inserting message to MongoDB: {e}',
|
|
block='mongo_producer',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=trace,
|
|
)
|
|
self.logger.error(trace)
|