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

SIENTIAPDE-1325

Browse Source 
Remove unused Couchbase configurations and related code. Refactor Activities class to eliminate Couchbase dependency, updating initialization and shutdown methods. Update README and tests to reflect these changes. Upgrade sientia-dataops-library dependency version in requirements.txt.
This commit is contained in:
Vitor Pimentel dos Santos
2025-11-06 10:45:23 -03:00
parent f099366bed
commit acde9475ec
20 changed files with 458 additions and 614 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
orchestrator/activities/activities.py
View File

@@ -5,6 +5,7 @@ with workflow.unsafe.imports_passed_through():
from typing import Any
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.temporal.activities.postgres import Postgres
from orchestrator.activities.email import Email
@@ -14,9 +15,7 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.activities.temporal_manager import TemporalManager
class Activities( # Couchbase,
TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres
):
class Activities(TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres):
"""
Central activities orchestrator for Temporal workflow operations.
@@ -47,11 +46,8 @@ class Activities( # Couchbase,
notification_handler: NotificationHandler,
):
# Initialize parent classes
# Couchbase.__init__(self, connection_string=couchbase_config['connection_string'],
# username=couchbase_config['username'],
# password=couchbase_config['password'],
# logger=logger,
# notification_handler=notification_handler)
metrics_controller = MetricsController(logger=logger)
TemporalManager.__init__(
self,
@@ -60,6 +56,7 @@ class Activities( # Couchbase,
laborious_namespace=temporal_config['temporal_laborious_namespace'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
SlotManager.__init__(
@@ -70,6 +67,7 @@ class Activities( # Couchbase,
password=redis_config['password'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
Formatters.__init__(
@@ -78,6 +76,7 @@ class Activities( # Couchbase,
laborious_namespace=temporal_config['temporal_laborious_namespace'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
MongoDB.__init__(
@@ -87,6 +86,7 @@ class Activities( # Couchbase,
ttl_index_seconds=mongodb_config['ttl_index_seconds'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
Email.__init__(
@@ -97,6 +97,7 @@ class Activities( # Couchbase,
smtp_port=email_config['smtp_port'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
Postgres.__init__(
@@ -110,6 +111,7 @@ class Activities( # Couchbase,
max_connections=postgres_config['max_connections'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def shutdown(self):
@@ -120,6 +122,9 @@ class Activities( # Couchbase,
services, and other resources to ensure proper cleanup when the
application terminates.
"""
MongoDB.shutdown(self)
MongoDB.close(self)
Postgres.close(self)
Email.shutdown(self)
Email.close(self)
Formatters.close(self)
SlotManager.close(self)
TemporalManager.close(self)

112
orchestrator/activities/couchbase.py
View File

@@ -1,112 +0,0 @@
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
import json
import traceback
from datetime import timedelta
from logging import Logger
from typing import Any
from couchbase.auth import PasswordAuthenticator
from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity
class Couchbase(BaseActivity):
"""
Couchbase database operations activity (currently unused).
This class provides Couchbase database connectivity and query operations
for Temporal workflows. It handles connection management, query execution,
and error reporting with automatic connection lifecycle management.
Args:
connection_string (str): Couchbase cluster connection string
username (str): Couchbase authentication username
password (str): Couchbase authentication password
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
Note:
This class is currently commented out in the main Activities class
but maintained for potential future use.
"""
def __init__(
self,
connection_string: str,
username: str,
password: str,
logger: Logger,
notification_handler: NotificationHandler,
):
self.connection_string = connection_string
self.username = username
self.password = password
logger.info('Initializing Couchbase connection...')
self.cluster = Cluster(
connection_string,
ClusterOptions(
authenticator=PasswordAuthenticator(username=username, password=password)
),
)
logger.info('Awaiting Couchbase connection...')
self.cluster.wait_until_ready(timeout=timedelta(seconds=10))
logger.info('Couchbase connection ready')
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
def shutdown(self):
try:
self.cluster.close()
except Exception as e:
self.logger.error(f'Failed to close Couchbase connection: {e}')
def __del__(self):
self.shutdown()
@activity.defn(name='load_query_from_couchbase')
async def load_query_from_couchbase(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Load a query from couchbase
Args:
input_data (dict[str, Any]): The input data containing the query to execute
Returns:
list[dict[str, Any]]: The result of the query
"""
query = input_data['query']
self.logger.info(f'Executing couchbase query: {query}')
try:
result = self.cluster.query(query)
except Exception as e:
trace = traceback.format_exc()
self.notification_handler.build_and_send_notification(
notification_id='COUCHBASE_LOAD_QUERY_ERROR',
message=f'Failed to execute couchbase query: {e}',
block='load_query_from_couchbase',
level=NotificationLevel.ERROR,
attachment_content=trace,
)
self.logger.error(trace)
raise e
rows = []
for row in result.rows():
rows.append(row)
self.logger.info('Fetched %d rows from couchbase', len(rows))
self.logger.debug('Rows: \n %s', json.dumps(rows, indent=4, sort_keys=True))
return rows

23
orchestrator/activities/email.py
View File

@@ -12,13 +12,13 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.observability.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
from orchestrator import metrics
from orchestrator.utils.email_builder import EmailBuilder
class Email(BaseActivity):
class Email(SientiaMonitoring):
"""
Email service activity for sending workflow notifications.
@@ -43,6 +43,7 @@ class Email(BaseActivity):
smtp_port: int,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
self.email_builder = EmailBuilder(logger=logger)
@@ -60,13 +61,25 @@ class Email(BaseActivity):
self.server.starttls()
self.server.login(self.sender_email, self.sender_password)
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def shutdown(self):
def close(self):
"""
Shutdown the Email connection and clean up resources.
Close the Email connection and clean up resources.
"""
self.server.quit()
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the Email connection is closed when the object is garbage-collected.
"""
self.close()
@activity.defn(name='build_email_html')
async def build_email_html(self, input_data: dict[str, Any]) -> dict[str, Any]:

52
orchestrator/activities/formatters.py
View File

@@ -10,7 +10,7 @@ with workflow.unsafe.imports_passed_through():
from pandas import DataFrame
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.observability.sientia_monitoring import MetricsController, SientiaMonitoring
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
from orchestrator.utils.orchestrator_functions import (
@@ -24,7 +24,7 @@ with workflow.unsafe.imports_passed_through():
topic_separator = '\n ========== \n'
class Formatters(BaseActivity):
class Formatters(SientiaMonitoring):
"""
Schedule and slot configuration formatting and notification filtering activity.
@@ -53,10 +53,28 @@ class Formatters(BaseActivity):
laborious_namespace: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
self.scouter_namespace = scouter_namespace
self.laborious_namespace = laborious_namespace
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def close(self):
"""
Close the Formatters connection and clean up resources.
"""
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the Formatters connection is closed when the object is garbage-collected.
"""
self.close()
@activity.defn(name='process_schedules')
async def process_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
@@ -166,7 +184,7 @@ class Formatters(BaseActivity):
slot_config[f'{i}'], notifications = build_tag_config(slot_tags, opc_servers)
if notifications:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
message=f'Servers {", ".join(notifications)} not found in opc_servers',
@@ -179,7 +197,7 @@ class Formatters(BaseActivity):
slot_tags = tags[last_index:]
slot_config[f'{number_of_slots}'], notifications = build_tag_config(slot_tags, opc_servers)
if notifications:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
message=f'Servers {", ".join(notifications)} not found in opc_servers',
@@ -357,7 +375,7 @@ class Formatters(BaseActivity):
return output
def send_success_report(
async def send_success_report(
self,
metadata: dict[str, Any],
message: str,
@@ -373,7 +391,7 @@ class Formatters(BaseActivity):
notification_id (str): The ID of the notification to send.
attachment (Any | None, optional): Optional attachment content to include.
"""
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=notification_id,
message=message,
@@ -382,7 +400,7 @@ class Formatters(BaseActivity):
attachment_content=json.dumps(attachment, indent=4, sort_keys=True),
)
def send_error_report(
async def send_error_report(
self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str
) -> None:
"""
@@ -394,7 +412,7 @@ class Formatters(BaseActivity):
notification_id (str): The ID of the notification.
attachment (str): The attachment content for the notification.
"""
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=notification_id,
message=message,
@@ -455,7 +473,7 @@ class Formatters(BaseActivity):
return success_keys, error_keys
def manage_and_send_report(
async def manage_and_send_report(
self,
metadata: dict[str, Any],
success_keys: list[str],
@@ -474,7 +492,7 @@ class Formatters(BaseActivity):
schedule_data (dict[str, Any]): The schedule data containing items and notification ID.
"""
if len(success_keys) > 0:
self.send_success_report(
await self.send_success_report(
metadata=metadata,
message=f'Successfully {schedule_type}: \n {", ".join(success_keys)}',
notification_id=schedule_data['id'],
@@ -489,7 +507,7 @@ class Formatters(BaseActivity):
else:
attachment.append(f'{key}:\n{value["message"]}')
self.send_error_report(
await self.send_error_report(
metadata=metadata,
message=f'Fails on {schedule_type}: \n {", ".join(error_keys)}',
notification_id=f'{schedule_data["id"]}_ERROR',
@@ -535,7 +553,7 @@ class Formatters(BaseActivity):
if len(schedule_data['items']) > 0:
success_keys, error_keys = self.parse_report_schedule(schedule_data['items'])
self.manage_and_send_report(
await self.manage_and_send_report(
metadata=metadata,
success_keys=success_keys,
error_keys=error_keys,
@@ -566,14 +584,14 @@ class Formatters(BaseActivity):
success_keys, error_keys = self.parse_report(inserted_slots)
if len(success_keys) > 0:
self.send_success_report(
await self.send_success_report(
metadata=metadata,
message=f'Inserted slots: \n {", ".join(success_keys)}',
notification_id='REPORT_ORCHESTRATION_INSERTED_SLOTS',
)
if len(error_keys) > 0:
self.send_error_report(
await self.send_error_report(
metadata=metadata,
message=f'Failed to insert slots: \n {", ".join(error_keys)}',
notification_id='REPORT_ORCHESTRATION_INSERTED_SLOTS',
@@ -584,14 +602,14 @@ class Formatters(BaseActivity):
success_keys, error_keys = self.parse_report(deleted_slots)
if len(success_keys) > 0:
self.send_success_report(
await self.send_success_report(
metadata=metadata,
message=f'Deleted slots: \n {", ".join(success_keys)}',
notification_id='REPORT_ORCHESTRATION_DELETED_SLOTS',
)
if len(error_keys) > 0:
self.send_error_report(
await self.send_error_report(
metadata=metadata,
message=f'Failed to delete slots: \n {", ".join(error_keys)}',
notification_id='REPORT_ORCHESTRATION_DELETED_SLOTS',

148
orchestrator/activities/mongo_db.py
View File

@@ -6,41 +6,14 @@ with workflow.unsafe.imports_passed_through():
from logging import Logger
from typing import Any
from pymongo import MongoClient
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.observability.sientia_monitoring import MetricsController, SientiaMonitoring
from sientia_do.repository.mongodb_repository import MongoDBRepository
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
def clear_mongo_id(docs: list) -> list:
"""
Remove MongoDB internal `_id` fields from nested structures.
Args:
docs (list): The list of documents or nested structures to clean.
Returns:
list: The cleaned documents with `_id` fields removed wherever present.
"""
for doc in docs:
if isinstance(doc, list):
clear_mongo_id(doc)
elif isinstance(doc, dict):
if '_id' in doc:
del doc['_id']
for _key, value in doc.items():
if isinstance(value, list):
clear_mongo_id(value)
elif isinstance(value, dict):
clear_mongo_id([value])
return docs
class MongoDB(BaseActivity):
class MongoDB(SientiaMonitoring):
"""
MongoDB operations activity for Temporal workflows.
@@ -64,60 +37,43 @@ class MongoDB(BaseActivity):
ttl_index_seconds: int,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.connection_string = connection_string
self.database_name = database_name
self.client: MongoClient = MongoClient(
self.connection_string, serverSelectionTimeoutMS=5000
self.mongo_db_repository = MongoDBRepository(
connection_string=connection_string,
database_name=database_name,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.client.server_info() # Force early failure if connection is invalid
self.database = self.client[self.database_name]
self.ttl_index_seconds = ttl_index_seconds
# Initialize MongoDB client here (omitted for brevity)
logger.info('MongoDB connection initialized')
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
def shutdown(self):
def close(self):
"""
Shutdown the MongoDB client and clean up resources.
"""
try:
if self.client:
self.logger.info('Closing MongoDB connection...')
self.client.close()
self.logger.info('MongoDB connection closed successfully')
except Exception as e:
self.logger.error(f'Failed to close MongoDB connection: {e}')
self.mongo_db_repository.close()
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the MongoDB client is closed when the object is garbage-collected.
"""
self.shutdown()
def find(self, collection_name: str, filters: dict[str, Any]) -> list[dict[str, Any]]:
"""
Find documents in a MongoDB collection based on the provided filters.
Args:
collection_name (str): The name of the collection to search in.
filters (dict[str, Any]): The query filters to apply.
Returns:
list[dict[str, Any]]: Documents matching the filters (with `_id` removed).
"""
collection = self.database[collection_name]
documents = list(collection.find(filters, {'_id': 0}))
documents = clear_mongo_id(documents)
return documents
self.close()
@activity.defn(
name='find_documents_in_mongodb',
@@ -151,7 +107,7 @@ class MongoDB(BaseActivity):
)
try:
documents = self.find(collection_name, filters)
documents = await self.mongo_db_repository.find(collection_name, filters, metadata)
self.info(
f"Loaded {len(documents)} documents from collection '{collection_name}'",
@@ -173,7 +129,7 @@ class MongoDB(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_QUERY_ERROR',
message=f'Failed to execute MongoDB query: {e}',
@@ -219,11 +175,9 @@ class MongoDB(BaseActivity):
)
try:
collection = self.database[collection_name]
aggregated_documents = list(collection.aggregate(aggregation))
aggregated_documents = clear_mongo_id(aggregated_documents)
aggregated_documents = await self.mongo_db_repository.aggregate(
collection_name, aggregation, metadata
)
self.info(
f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'",
@@ -245,7 +199,7 @@ class MongoDB(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_AGGREGATION_ERROR',
message=f'Failed to execute MongoDB aggregation: {e}',
@@ -268,12 +222,9 @@ class MongoDB(BaseActivity):
updated_pipelines = input_data.get('updated_pipelines', [])
metadata = input_data.get('metadata', {})
date_now = now()
collection = self.database['orchestrated_schedules']
self.info('Updating pipelines timestamps...', metadata=metadata)
success_count = 0
argument = [
{'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace']}
for pipeline in updated_pipelines
@@ -282,11 +233,12 @@ class MongoDB(BaseActivity):
data_filter = {'$or': argument} if argument else {}
try:
collection.update_many(data_filter, {'$set': {'updated_at': date_now}})
success_count += 1
await self.mongo_db_repository.update_many(
'orchestrated_schedules', data_filter, {'$set': {'updated_at': date_now}}, metadata
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_UPDATE_PIPELINES_ERROR',
message=f'Failed to update pipelines timestamps: {e}',
@@ -298,7 +250,7 @@ class MongoDB(BaseActivity):
raise e
self.info(
f'Updated {success_count} of {len(updated_pipelines)} pipelines timestamps',
f'Updated {len(updated_pipelines)} pipelines timestamps',
metadata=metadata,
)
@@ -312,12 +264,9 @@ class MongoDB(BaseActivity):
"""
created_pipelines = input_data.get('created_pipelines', [])
metadata = input_data.get('metadata', {})
collection = self.database['orchestrated_schedules']
self.info('Creating pipelines timestamps...', metadata=metadata)
success_count = 0
date_now = now()
argument = [
@@ -333,11 +282,12 @@ class MongoDB(BaseActivity):
try:
if data_filter:
collection.insert_many(data_filter)
success_count += 1
await self.mongo_db_repository.insert_many(
'orchestrated_schedules', data_filter, metadata
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_CREATE_PIPELINES_ERROR',
message=f'Failed to create pipelines timestamps: {e}',
@@ -349,7 +299,7 @@ class MongoDB(BaseActivity):
raise e
self.info(
f'Created {success_count} of {len(created_pipelines)} pipelines timestamps',
f'Created {len(created_pipelines)} pipelines timestamps',
metadata=metadata,
)
@@ -363,12 +313,9 @@ class MongoDB(BaseActivity):
"""
deleted_pipelines = input_data.get('deleted_pipelines', [])
metadata = input_data.get('metadata', {})
collection = self.database['orchestrated_schedules']
self.info('Deleting pipelines timestamps...', metadata=metadata)
success_count = 0
argument = [
{'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace']}
for pipeline in deleted_pipelines
@@ -377,11 +324,12 @@ class MongoDB(BaseActivity):
data_filter = {'$or': argument} if argument else {}
try:
collection.delete_many(data_filter)
success_count += 1
await self.mongo_db_repository.delete_many(
'orchestrated_schedules', data_filter, metadata
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_DELETE_PIPELINES_ERROR',
message=f'Failed to delete pipelines timestamps: {e}',
@@ -393,7 +341,7 @@ class MongoDB(BaseActivity):
raise e
self.info(
f'Deleted {success_count} of {len(deleted_pipelines)} pipelines timestamps',
f'Deleted {len(deleted_pipelines)} pipelines timestamps',
metadata=metadata,
)
@@ -413,7 +361,7 @@ class MongoDB(BaseActivity):
metadata=metadata,
)
collection_names = self.database.list_collection_names()
collection_names = self.mongo_db_repository.database.list_collection_names()
created_collections = []
created_indexes = []
@@ -424,10 +372,10 @@ class MongoDB(BaseActivity):
try:
# Check if collection exists
if collection not in collection_names:
self.database.create_collection(collection)
self.mongo_db_repository.database.create_collection(collection)
created_collections.append(collection)
collection = self.database[collection]
collection = self.mongo_db_repository.database[collection]
# Check if TTL index exists
existing_indexes = collection.list_indexes()
ttl_index_exists = False
@@ -448,7 +396,7 @@ class MongoDB(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_CREATE_COLLECTION_ERROR',
message=f'Failed to create collection {collection} with TTL index: {e}',
@@ -510,7 +458,7 @@ class MongoDB(BaseActivity):
self.debug(f'Data filter: {data_filter}', metadata=metadata)
data = self.find(collection_name, data_filter)
data = await self.mongo_db_repository.find(collection_name, data_filter, metadata)
self.debug(f'Collected: {data}', metadata=metadata)
@@ -526,7 +474,7 @@ class MongoDB(BaseActivity):
return data
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGO_LOAD_ERROR',
message=f'Error loading data from MongoDB: {e}',

62
orchestrator/activities/slot_manager.py
View File

@@ -10,11 +10,12 @@ with workflow.unsafe.imports_passed_through():
from pandas import DataFrame
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.redis_base import Redis
from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
from sientia_do.repository.redis_repository import RedisRepository
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
class SlotManager(Redis):
class SlotManager(SientiaMonitoring):
"""
Redis-based OPC slot management and notification filtering activity.
@@ -48,8 +49,37 @@ class SlotManager(Redis):
password: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
Redis.__init__(self, host, port, username, password, logger, notification_handler)
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.redis_repository = RedisRepository(
host=host,
port=port,
username=username,
password=password,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def close(self):
"""
Close the SlotManager connection and clean up resources.
"""
self.redis_repository.close()
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the SlotManager connection is closed when the object is garbage-collected.
"""
self.close()
@activity.defn(name='load_opc_slots')
async def load_opc_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
@@ -80,7 +110,7 @@ class SlotManager(Redis):
opc_slots = {}
try:
slot_keys = self.redis_client.keys('slot:opc_tags:*')
slot_keys = await self.redis_repository.keys('slot:opc_tags:*')
self.debug(f'Slot keys: {slot_keys}', metadata=metadata)
@@ -91,10 +121,10 @@ class SlotManager(Redis):
decoded_keys = slot_keys
for key in decoded_keys:
opc_slots[key] = self.get(key)
opc_slots[key] = await self.redis_repository.get(key)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='REDIS_GET_ERROR',
message=f'Failed to load OPC slots: {e}',
@@ -123,7 +153,7 @@ class SlotManager(Redis):
self.info('Loading active ingestors...', metadata=metadata)
try:
active_ingestors = self.redis_client.keys('heartbeat:ingestor:*')
active_ingestors = await self.redis_repository.keys('heartbeat:ingestor:*')
self.info(f'Loaded {len(active_ingestors)} active ingestors', metadata=metadata)
@@ -140,7 +170,7 @@ class SlotManager(Redis):
return ingestors
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='REDIS_GET_ERROR',
message=f'Failed to load active ingestors: {e}',
@@ -175,7 +205,7 @@ class SlotManager(Redis):
for slot in to_insert:
try:
self.set(f'slot:opc_tags:{slot}', to_insert[slot], ttl=None)
await self.redis_repository.set(f'slot:opc_tags:{slot}', to_insert[slot], ttl=None)
report[slot] = {'success': True, 'message': 'Slot updated successfully'}
success_count += 1
except Exception as e:
@@ -213,7 +243,7 @@ class SlotManager(Redis):
for slot in to_delete:
try:
self.redis_client.delete(f'slot:opc_tags:{slot}')
await self.redis_repository.delete(f'slot:opc_tags:{slot}')
report[slot] = {'success': True, 'message': 'Slot deleted successfully'}
success_count += 1
except Exception as e:
@@ -243,9 +273,9 @@ class SlotManager(Redis):
key = f'notification_last_timestamp:{input_data["mail_type"]}'
try:
data_hold = self.get(key)
data_hold = await self.redis_repository.get(key)
except Exception as e:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='REDIS_GET_ERROR',
message=f'Error getting last data timestamp: {e}',
@@ -290,9 +320,9 @@ class SlotManager(Redis):
self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata)
try:
self.set(key, last_data_timestamp, ttl=60 * 60 * 5)
await self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5)
except Exception as e:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='REDIS_SET_ERROR',
message=f'Error setting last data timestamp: {e}',
@@ -361,7 +391,7 @@ class SlotManager(Redis):
# Check if notification was recently sent
key = f'{notification["trigger"]}:{notification_id}'
last_sent = self.get(key)
last_sent = await self.redis_repository.get(key)
if last_sent is None:
alert_type = 'core_alerts'
@@ -414,6 +444,6 @@ class SlotManager(Redis):
status = row['status']
if status == 'sent':
key = f'{row["schedule"]}:{row["notification_id"]}'
self.set(key, date_now, ttl=sent_ttl)
await self.redis_repository.set(key, date_now, ttl=sent_ttl)
self.info('Notification cache stored...', metadata=metadata)

25
orchestrator/activities/temporal_manager.py
View File

@@ -19,12 +19,12 @@ with workflow.unsafe.imports_passed_through():
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.observability.sientia_monitoring import MetricsController, SientiaMonitoring
from orchestrator.utils.converters import parse_frequency
class TemporalManager(BaseActivity):
class TemporalManager(SientiaMonitoring):
"""
Temporal workflow and schedule management activity.
@@ -48,7 +48,14 @@ class TemporalManager(BaseActivity):
laborious_namespace: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.temporal_host = host
self.scouter_namespace = scouter_namespace
self.laborious_namespace = laborious_namespace
@@ -58,7 +65,17 @@ class TemporalManager(BaseActivity):
self.model_name_id_key = SearchAttributeKey.for_keyword('model_name')
self.orchestrated_id_key = SearchAttributeKey.for_keyword('orchestrated')
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
def close(self):
"""
Close the TemporalManager connection and clean up resources.
"""
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the TemporalManager connection is closed when the object is garbage-collected.
"""
self.close()
async def connect_to_temporal(self):
"""
@@ -127,7 +144,7 @@ class TemporalManager(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='SCHEDULER_NORMALIZE_SCHEDULES_ERROR',
message=f'Failed to normalize schedules: {e}',