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

Update dependencies, improve CI workflow, and enhance code formatting

Browse Source 
- Updated the `sientia-dataops-library` dependency version from 1.4.3 to 1.4.6 in `requirements.txt`.
- Modified the GitHub Actions workflow to install development and runtime dependencies separately, improving clarity and organization.
- Added code formatting and linting checks using Ruff, along with type checking using mypy, to ensure code quality.
- Updated `.gitignore` to include additional cache directories and log files.
- Refactored code in various files for consistency in string formatting and improved logging messages.
This commit is contained in:
vitor-aignosi
2025-10-17 12:58:24 -03:00
parent 464c9aae9b
commit e2462af31c
21 changed files with 1536 additions and 1368 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
ingestor/app.py
View File

@@ -1,9 +1,9 @@
import os
import asyncio
import os
import signal
import traceback
from threading import Event
from time import sleep, time
from time import time
from prometheus_client import start_http_server
@@ -11,7 +11,7 @@ import ingestor.metrics as metrics
from ingestor.ingestor import Ingestor
exit_signal = Event()
POD_ID = os.getenv("HOSTNAME", "localhost")
POD_ID = os.getenv('HOSTNAME', 'localhost')
async def main():
@@ -43,30 +43,27 @@ async def main():
try:
await ingestor.prepare_ingestor()
except Exception as e:
metrics.APP_ERRORS_TOTAL.labels(
pod_id=POD_ID).inc() # Increment errors
ingestor.logger.error(f"Failed to prepare ingestor: {e}")
metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc() # Increment errors
ingestor.logger.error(f'Failed to prepare ingestor: {e}')
exit_signal.set()
ingestor.logger.info("Ingestor prepared. Starting main loop.")
ingestor.logger.info('Ingestor prepared. Starting main loop.')
while not exit_signal.is_set():
start_time = time() # Start loop timer
try:
await ingestor.loop()
metrics.APP_LOOP_COUNT.labels(
pod_id=POD_ID).inc() # Increment loop counter
metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc() # Increment loop counter
# Use asyncio.sleep instead of exit_signal.wait for better async compatibility
await asyncio.sleep(ingestor.poll_interval)
except KeyboardInterrupt: # Handle Ctrl+C gracefully
print("KeyboardInterrupt received. Setting exit_signal flag.")
print('KeyboardInterrupt received. Setting exit_signal flag.')
exit_signal.set()
except Exception:
print("Exception in main loop. Setting exit_signal flag.")
print('Exception in main loop. Setting exit_signal flag.')
traceback.print_exc()
metrics.APP_ERRORS_TOTAL.labels(
pod_id=POD_ID).inc() # Increment errors
metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc() # Increment errors
exit_signal.set()
finally:
# Record loop duration
@@ -76,7 +73,7 @@ async def main():
await ingestor.shutdown()
metrics.APP_UP.labels(pod_id=POD_ID).set(0) # Mark app as DOWN
ingestor.logger.info("Main loop exit_signaled.")
ingestor.logger.info('Main loop exit_signaled.')
# Give Prometheus a chance to scrape one last time before exiting (optional)
await asyncio.sleep(5)
@@ -96,7 +93,7 @@ def signal_handler(_signum, _frame):
_signum: The signal number received
_frame: The current stack frame (unused)
"""
print(f"Received signal {_signum}. Setting exit_signal flag.")
print(f'Received signal {_signum}. Setting exit_signal flag.')
exit_signal.set()
@@ -115,12 +112,12 @@ def start_prometheus_server():
Exception: If the server fails to start, the application will exit
"""
try:
port = int(os.getenv("HTTP_METRICS_PORT", 9090))
port = int(os.getenv('HTTP_METRICS_PORT', 9090))
start_http_server(port)
print(f"Prometheus server started on port {port}.")
print(f'Prometheus server started on port {port}.')
metrics.APP_UP.labels(pod_id=POD_ID).set(1) # Mark app as UP
except Exception as e:
print(f"Failed to start Prometheus server: {e}")
print(f'Failed to start Prometheus server: {e}')
os._exit(1)
@@ -139,13 +136,13 @@ def run_async_main():
asyncio.set_event_loop(loop)
loop.run_until_complete(main())
except KeyboardInterrupt:
print("KeyboardInterrupt received in main thread.")
print('KeyboardInterrupt received in main thread.')
exit_signal.set()
finally:
loop.close()
if __name__ == "__main__":
if __name__ == '__main__':
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGHUP, signal_handler)

103
ingestor/ingestor.py
View File

@@ -1,14 +1,12 @@
import asyncio
from os import getenv
from copy import deepcopy
from typing import Dict, Any
from os import getenv
from typing import Any
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import get_logger
from ingestor.managers.ingestor_manager import IngestorManager
import ingestor.metrics as metrics
from ingestor.managers.ingestor_manager import IngestorManager
class Ingestor:
@@ -73,36 +71,36 @@ class Ingestor:
- Metrics collection and monitoring
"""
kafka_servers = getenv("KAFKA_SERVERS", "localhost:9092")
export_to_kafka = getenv("EXPORT_TO_KAFKA", "false")
kafka_servers = getenv('KAFKA_SERVERS', 'localhost:9092')
export_to_kafka = getenv('EXPORT_TO_KAFKA', 'false')
if export_to_kafka and export_to_kafka == "true":
if export_to_kafka and export_to_kafka == 'true':
export_to_kafka = True
else:
export_to_kafka = False
self.export_to_kafka = export_to_kafka
self.redis_host = getenv("REDIS_HOST", "localhost")
self.redis_port = int(getenv("REDIS_PORT", "6379"))
self.redis_username = getenv("REDIS_USERNAME", None)
self.redis_password = getenv("REDIS_PASSWORD", None)
self.lease_ttl = int(getenv("LEASE_TTL", "10"))
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")
self.mongo_connection_string = f"mongodb://{mongo_username}:{mongo_password}@{mongo_url}"
self.redis_host = getenv('REDIS_HOST', 'localhost')
self.redis_port = int(getenv('REDIS_PORT', '6379'))
self.redis_username = getenv('REDIS_USERNAME', None)
self.redis_password = getenv('REDIS_PASSWORD', None)
self.lease_ttl = int(getenv('LEASE_TTL', '10'))
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')
self.mongo_connection_string = f'mongodb://{mongo_username}:{mongo_password}@{mongo_url}'
self.kafka_servers = kafka_servers.split(",")
self.kafka_servers = kafka_servers.split(',')
self.logger = get_logger(__name__)
self.notification_handler = NotificationHandler(
connection_string=self.mongo_connection_string,
database=self.mongo_database,
logger=self.logger,
project_name="opc_ingestor"
project_name='opc_ingestor',
)
self.metadata = {
@@ -148,7 +146,7 @@ class Ingestor:
"""
if not acquired:
self.logger.warning("No slots available")
self.logger.warning('No slots available')
else:
# Subscribe to acquired slots
@@ -198,7 +196,7 @@ class Ingestor:
# Get slot lease
acquired = self.ingestor_manager.get_slot_leases()
self.logger.info(f"Acquired slots: {acquired}")
self.logger.info(f'Acquired slots: {acquired}')
await self.handle_acquired_tags(acquired)
@@ -229,9 +227,7 @@ class Ingestor:
# Get slot lease
self.ingestor_manager.get_slot_leases(1)
async def manage_leases(
self, available_slots: int, lacking_ingestors: int, slot_diff: int
):
async def manage_leases(self, available_slots: int, lacking_ingestors: int, slot_diff: int):
"""
Manages the allocation and deallocation of slot leases for ingestors.
@@ -257,14 +253,13 @@ class Ingestor:
if available_slots > 0 and lacking_ingestors > 0:
# Some ingestors are inactive, so there are "available_slots" slots available
self.logger.info(f"Slots available: {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(f"Extra slots available: {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
@@ -281,7 +276,7 @@ class Ingestor:
len(self.ingestor_manager.managed_tags)
)
async def update_ingestor_manager(self, old_managed_tags: Dict[str, Any]):
async def update_ingestor_manager(self, old_managed_tags: dict[str, Any]):
"""
Updates the ingestor manager with new managed tags and handles configuration changes.
@@ -302,37 +297,39 @@ class Ingestor:
- Updates metrics to reflect current state
"""
self.logger.debug(
f"Current managed tags: {self.ingestor_manager.managed_tags}")
self.logger.debug(f'Current managed tags: {self.ingestor_manager.managed_tags}')
await self.ingestor_manager.update_opc_servers()
new_managed_tags = deepcopy(self.ingestor_manager.managed_tags)
self.logger.debug(
f"Comparing new managed tags {new_managed_tags} with old 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)
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)}
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(f"Changes: {changes}")
self.logger.debug(f'Changes: {changes}')
for slot, config in new_managed_tags.items():
if slot not in old_managed_tags:
self.logger.info(f"Subscribing to new slot {slot}")
self.logger.info(f'Subscribing to new slot {slot}')
await self.ingestor_manager.subscribe_to_tags({slot: config})
continue
if config != old_managed_tags[slot]:
self.logger.info(f"Resubscribing to slot {slot}")
self.logger.info(f'Resubscribing to slot {slot}')
await self.ingestor_manager.unsubscribe_slot(slot)
await self.ingestor_manager.subscribe_to_tags({slot: config})
for slot in old_managed_tags.keys():
if slot not in new_managed_tags:
self.logger.info(f"Unsubscribing from slot {slot}")
self.logger.info(f'Unsubscribing from slot {slot}')
await self.ingestor_manager.unsubscribe_slot(slot)
# Ensure the gauge is updated after any potential changes here
@@ -366,7 +363,7 @@ class Ingestor:
self.ingestor_manager.declare_active()
self.logger.info("Polling for slot updates...")
self.logger.info('Polling for slot updates...')
# Get active ingestors
current_managed_tags = deepcopy(self.ingestor_manager.managed_tags)
@@ -380,14 +377,14 @@ class Ingestor:
metrics.ACTIVE_INGESTORS.set(number_of_ingestors)
# Handle no slots
self.logger.info("Managing no slots...")
self.logger.info('Managing no slots...')
self.manage_no_slots(number_of_slots)
available_slots = number_of_slots - number_of_leases
lacking_ingestors = number_of_slots - number_of_ingestors
slot_diff = len(self.ingestor_manager.managed_tags) - 1
self.logger.info("Managing leases...")
self.logger.info('Managing leases...')
await self.manage_leases(available_slots, lacking_ingestors, slot_diff)
# Update managed slots gauge
@@ -396,23 +393,23 @@ class Ingestor:
)
self.logger.debug(
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}"
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
self.logger.info("No slots acquired in this loop")
self.logger.info('No slots acquired in this loop')
# Update opc servers
self.logger.info("Updating slot config...")
self.logger.info('Updating slot config...')
self.ingestor_manager.update_slot_config()
# Check OPC cycles
self.logger.info("Checking OPC servers integrity...")
self.logger.info('Checking OPC servers integrity...')
self.ingestor_manager.check_opc_servers_integrity()
self.logger.info("Updating managed tags...")
self.logger.info('Updating managed tags...')
await self.update_ingestor_manager(current_managed_tags)

98
ingestor/managers/data_manager.py
View File

@@ -1,16 +1,18 @@
import json
import os
import traceback
from time import sleep
from pymongo import MongoClient
from kafka import KafkaProducer
from kafka.errors import NoBrokersAvailable
from pymongo import MongoClient
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.constants import now
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.logger import Logger
import traceback
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.temporal.constants import now
import ingestor.metrics as metrics
import os
class DataManager(BaseActivity):
@@ -85,49 +87,43 @@ class DataManager(BaseActivity):
- KAFKA_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure
"""
self.pod_id = os.getenv("HOSTNAME", "localhost")
self.pod_id = os.getenv('HOSTNAME', 'localhost')
self.kafka_producer = None
self.export_to_kafka = export_to_kafka
if self.export_to_kafka:
for i in range(0, 3):
logger.info(
f"Trying ({i}) to initializing DataManager with Kafka servers: {kafka_servers}"
f'Trying ({i}) to initializing DataManager with Kafka servers: {kafka_servers}'
)
try:
self.kafka_producer = KafkaProducer(
bootstrap_servers=kafka_servers,
value_serializer=lambda v: json.dumps(v).encode(
"utf-8"
'utf-8'
), # Serialize JSON messages
key_serializer=lambda k: str(
k).encode("utf-8") if k else None,
key_serializer=lambda k: str(k).encode('utf-8') if k else None,
)
# Kafka connected
metrics.KAFKA_CONNECTION_STATUS.labels(
pod_id=self.pod_id).set(1)
metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(1)
break
except NoBrokersAvailable:
logger.error(
f"Kafka servers {kafka_servers} are not available. Retrying..."
)
logger.error(f'Kafka servers {kafka_servers} are not available. Retrying...')
sleep(5)
else:
# Kafka not connected
metrics.KAFKA_CONNECTION_STATUS.labels(
pod_id=self.pod_id).set(0)
metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0)
logger.error(
f"Failed to connect to Kafka servers {kafka_servers} after 3 attempts."
f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.'
)
raise NoBrokersAvailable(
f"Failed to connect to Kafka servers {kafka_servers} after 3 attempts."
f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.'
)
logger.info(
f"DataManager initialized with Kafka servers: {kafka_servers}")
logger.info(f'DataManager initialized with Kafka servers: {kafka_servers}')
logger.info(
f"Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}"
f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}'
)
self.connection_string = mongo_connection_string
@@ -140,13 +136,11 @@ class DataManager(BaseActivity):
self.mongo_db = self.mongo_client[self.database]
logger.info(
f"DataManager initialized with MongoDB servers: {self.connection_string}"
)
logger.info(f'DataManager initialized with MongoDB servers: {self.connection_string}')
BaseActivity.__init__(self, logger=logger,
notification_handler=notification_handler,
set_error_counter=True)
BaseActivity.__init__(
self, logger=logger, notification_handler=notification_handler, set_error_counter=True
)
def shutdown(self):
"""
@@ -165,22 +159,19 @@ class DataManager(BaseActivity):
self.kafka_producer.flush(timeout=10)
self.kafka_producer.close()
# Mark as disconnected
metrics.KAFKA_CONNECTION_STATUS.labels(
pod_id=self.pod_id).set(0)
metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0)
except Exception as e:
self.logger.error(f"Error closing Kafka producer: {e}")
self.logger.error(f'Error closing Kafka producer: {e}')
else:
self.logger.warning(
"Kafka producer is already closed or not initialized.")
self.logger.warning('Kafka producer is already closed or not initialized.')
if self.mongo_client:
try:
self.mongo_client.close()
except Exception as e:
self.logger.error(f"Error closing MongoDB client: {e}")
self.logger.error(f'Error closing MongoDB client: {e}')
else:
self.logger.warning(
"MongoDB client is already closed or not initialized.")
self.logger.warning('MongoDB client is already closed or not initialized.')
def __del__(self):
self.shutdown()
@@ -197,7 +188,7 @@ class DataManager(BaseActivity):
msg: Kafka message object containing delivery details
"""
self.logger.debug(
f"Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}"
f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}'
)
def delivery_error(self, err: str):
@@ -210,7 +201,7 @@ class DataManager(BaseActivity):
Args:
err: Error information from the failed delivery attempt
"""
self.logger.error(f"Delivery failed for record : {err}")
self.logger.error(f'Delivery failed for record : {err}')
def publish(self, topic: str, data: dict) -> None:
"""
@@ -229,26 +220,22 @@ class DataManager(BaseActivity):
if self.export_to_kafka:
try:
self.logger.debug(
f"Publishing message to topic {topic}: {data}")
self.logger.debug(f'Publishing message to topic {topic}: {data}')
self.kafka_producer.send(topic=topic, value=data).add_callback(
self.delivery_report
).add_errback(self.delivery_error)
self.kafka_producer.flush(timeout=10)
metrics.KAFKA_MESSAGES_SENT.labels(
pod_id=self.pod_id, topic=topic).inc()
metrics.KAFKA_MESSAGES_SENT.labels(pod_id=self.pod_id, topic=topic).inc()
except Exception as e:
metrics.KAFKA_MESSAGES_ERRORS.labels(
pod_id=self.pod_id, topic=topic).inc()
metrics.KAFKA_MESSAGES_ERRORS.labels(pod_id=self.pod_id, topic=topic).inc()
trace = traceback.format_exc()
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",
notification_id=f'KAFKA_PRODUCER_ERROR_{topic}',
message=f'Error publishing message to topic {topic}: {e}',
block='kafka_producer',
level=NotificationLevel.ERROR,
attachment_content=trace,
)
@@ -260,25 +247,22 @@ class DataManager(BaseActivity):
collection.insert_one(
{
**data,
"inserted_at": now(),
'inserted_at': now(),
}
)
self.logger.debug(
f"Message inserted into MongoDB collection {topic}: {data}")
self.logger.debug(f'Message inserted into MongoDB collection {topic}: {data}')
metrics.TAG_WRITTEN_COUNT.labels(
pod_id=self.pod_id,
tag_name=data["name"],
collection_name=topic
pod_id=self.pod_id, tag_name=data['name'], collection_name=topic
).inc()
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=self.metadata,
notification_id=f"MONGO_PRODUCER_ERROR_{topic}",
message=f"Error inserting message to MongoDB: {e}",
block="mongo_producer",
notification_id=f'MONGO_PRODUCER_ERROR_{topic}',
message=f'Error inserting message to MongoDB: {e}',
block='mongo_producer',
level=NotificationLevel.ERROR,
attachment_content=trace,
)

173
ingestor/managers/ingestor_manager.py
View File

@@ -1,15 +1,16 @@
import asyncio
import traceback
from typing import Dict, List
from copy import deepcopy
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.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
import ingestor.metrics as metrics
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(BaseActivity):
@@ -57,14 +58,20 @@ class IngestorManager(BaseActivity):
metadata (dict): Application metadata
"""
def __init__(self,
kafka_servers: str, redis_data: dict,
lease_ttl: int, heartbeat_ttl: int,
poll_interval: int, mongo_connection_string: str, mongo_database: str,
metadata: dict,
logger: Logger, notification_handler: NotificationHandler,
export_to_kafka: bool = False):
def __init__(
self,
kafka_servers: str,
redis_data: dict,
lease_ttl: int,
heartbeat_ttl: int,
poll_interval: int,
mongo_connection_string: str,
mongo_database: str,
metadata: dict,
logger: Logger,
notification_handler: NotificationHandler,
export_to_kafka: bool = False,
):
redis_host = redis_data.get('host')
redis_port = redis_data.get('port')
redis_username = redis_data.get('username', None)
@@ -77,7 +84,7 @@ class IngestorManager(BaseActivity):
export_to_kafka=export_to_kafka,
metadata=metadata,
logger=logger,
notification_handler=notification_handler
notification_handler=notification_handler,
)
self.opc_managers = {}
self.resource_manager = ResourceManager(
@@ -98,9 +105,9 @@ class IngestorManager(BaseActivity):
self.metadata = metadata
BaseActivity.__init__(self, logger=logger,
notification_handler=notification_handler,
set_error_counter=True)
BaseActivity.__init__(
self, logger=logger, notification_handler=notification_handler, set_error_counter=True
)
async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None:
"""
@@ -130,8 +137,7 @@ class IngestorManager(BaseActivity):
"""
try:
self.logger.info(
f"Initializing OpcManager at {server_config['url']}")
self.logger.info(f'Initializing OpcManager at {server_config["url"]}')
manager = OpcManager(
name=server_config['name'],
url=server_config['url'],
@@ -142,7 +148,7 @@ class IngestorManager(BaseActivity):
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')
server_cert_path=server_config.get('server_cert_path'),
)
manager.config = server_config
@@ -153,9 +159,9 @@ class IngestorManager(BaseActivity):
metadata=self.metadata,
notification_id=f'OPC_CONNECTION_ERROR_{server_config["name"]}',
message=f'Error initializing OPC manager: {e}',
block="opc_manager",
block='opc_manager',
level=NotificationLevel.ERROR,
attachment_content=trace
attachment_content=trace,
)
self.logger.error(trace)
@@ -216,49 +222,41 @@ class IngestorManager(BaseActivity):
server_config.pop('tags', None)
server_instance = self.opc_managers.get(server, None)
if server_instance is None:
self.logger.info(
f"Initializing OPC manager for server {server}"
)
server_instance = await self.initialize_opc_from_config(
server_config
)
self.logger.info(f'Initializing OPC manager for server {server}')
server_instance = await self.initialize_opc_from_config(server_config)
elif server_instance.config != server_config:
self.logger.warning(
f"Reinitializing OPC manager for server {server}"
)
self.logger.warning(f'Reinitializing OPC manager for server {server}')
server_instance.disconnect()
del self.opc_managers[server]
server_instance = await self.initialize_opc_from_config(
server_config
)
server_instance = await self.initialize_opc_from_config(server_config)
else:
self.logger.debug(
f"OPC manager for server {server} is already initialized and up to date"
f'OPC manager for server {server} is already initialized and up to date'
)
if server_instance is not None:
self.opc_managers[server] = server_instance
else:
self.logger.warning(
f"Failed to initialize OPC manager for server {server}, "
f"removing server from managed tags."
f'Failed to initialize OPC manager for server {server}, '
f'removing server from managed tags.'
)
_a = [self.managed_tags[slot].pop(server, None)
for slot, _value in current_managed_tags.items()]
_a = [
self.managed_tags[slot].pop(server, None)
for slot, _value in current_managed_tags.items()
]
servers = list(self.opc_managers.keys())
for server in servers:
if server not in registered_servers:
self.logger.warning(
f"Server {server} not found in managed tags. "
f"Desconnecting from server."
f'Server {server} not found in managed tags. Desconnecting from server.'
)
await self.opc_managers[server].disconnect()
self.opc_managers.pop(server, None)
metrics.OPC_MANAGERS_ACTIVE.labels(
pod_id=self.pod_id).set(len(self.opc_managers))
metrics.OPC_MANAGERS_ACTIVE.labels(pod_id=self.pod_id).set(len(self.opc_managers))
def check_opc_servers_integrity(self):
"""
@@ -280,16 +278,12 @@ class IngestorManager(BaseActivity):
is_lost = opc_manager.check_opc_listenning()
if is_lost:
self.logger.warning(
f"OPC server {server} is lost. "
f"Server will be disconnected."
)
self.logger.warning(f'OPC server {server} is lost. Server will be disconnected.')
for slot, _config in self.managed_tags.items():
self.managed_tags[slot].pop(server, None)
metrics.OPC_MANAGERS_ACTIVE.labels(
pod_id=self.pod_id).set(len(self.opc_managers))
metrics.OPC_MANAGERS_ACTIVE.labels(pod_id=self.pod_id).set(len(self.opc_managers))
def declare_active(self):
"""
@@ -306,7 +300,7 @@ class IngestorManager(BaseActivity):
self.resource_manager.ingestor_heartbeat()
def get_active_ingestors(self) -> List[str]:
def get_active_ingestors(self) -> list[str]:
"""
Retrieve a list of active ingestors.
@@ -363,7 +357,7 @@ class IngestorManager(BaseActivity):
metrics.SLOTS_TOTAL.set(self.number_of_slots)
return self.number_of_slots
def get_slot_leases(self, max_slots: int = 1) -> Dict:
def get_slot_leases(self, max_slots: int = 1) -> dict:
"""
Acquires a specified number of resource slots by leasing them from the resource manager.
@@ -377,7 +371,7 @@ class IngestorManager(BaseActivity):
max_slots (int): The maximum number of slots to lease. Defaults to 1.
Returns:
Dict: A dictionary where the keys are the slot identifiers (as strings)
Dict: A dictionary where the keys are the slot identifiers (as strings)
and the values are the leased slot details.
Behavior:
@@ -394,7 +388,7 @@ class IngestorManager(BaseActivity):
acquired = {}
for i in range(1, self.number_of_slots + 1):
if self.resource_manager.lease_tag(str(i)):
self.logger.info(f"Leased slot {i}")
self.logger.info(f'Leased slot {i}')
slots = self.resource_manager.get_tag_slot(str(i))
if slots is None:
continue
@@ -403,17 +397,14 @@ class IngestorManager(BaseActivity):
if len(acquired) >= max_slots:
self.managed_tags.update(acquired)
metrics.SLOTS_MANAGED.labels(
pod_id=self.pod_id).set(len(self.managed_tags))
metrics.SLOTS_MANAGED.labels(pod_id=self.pod_id).set(len(self.managed_tags))
return acquired
self.logger.warning(
f"Unable to acquire {max_slots} slots. "
f"Only {acquired} slots were leased."
f'Unable to acquire {max_slots} slots. Only {acquired} slots were leased.'
)
self.managed_tags.update(acquired)
metrics.SLOTS_MANAGED.labels(
pod_id=self.pod_id).set(len(self.managed_tags))
metrics.SLOTS_MANAGED.labels(pod_id=self.pod_id).set(len(self.managed_tags))
return acquired
async def unsubscribe_slot(self, slot: str):
@@ -441,7 +432,7 @@ class IngestorManager(BaseActivity):
def update_slot_config(self):
"""
Updates the configuration of managed slots by renewing their leases,
Updates the configuration of managed slots by renewing their leases,
fetching the latest configurations, and handling any changes or removals.
This method performs the following steps:
@@ -458,7 +449,7 @@ class IngestorManager(BaseActivity):
- Updates OPC server subscriptions based on the current state of managed slots.
Raises:
None explicitly, but relies on the behavior of `resource_manager` and
None explicitly, but relies on the behavior of `resource_manager` and
other dependencies for error handling.
Logging:
@@ -480,17 +471,17 @@ class IngestorManager(BaseActivity):
for slot in removed_slots:
self.managed_tags.pop(slot, None)
def drop_slot_leases(self, ids: List[str]) -> None:
def drop_slot_leases(self, ids: list[str]) -> None:
"""
Releases the leases associated with the specified slot IDs.
This method iterates through a list of slot IDs and calls the
`drop_tag_lease` method of the `resource_manager` to release
This method iterates through a list of slot IDs and calls the
`drop_tag_lease` method of the `resource_manager` to release
the lease for each ID. It's used during load balancing and
graceful shutdown scenarios.
Args:
ids (List[str]): A list of slot IDs for which the leases
ids (List[str]): A list of slot IDs for which the leases
should be released.
Returns:
@@ -509,7 +500,7 @@ class IngestorManager(BaseActivity):
Manages the subscription of tags to a specified OPC server and slot.
This method ensures that the specified server and slot have an active subscription
for the provided tags. If the server or slot is not properly configured, or if
for the provided tags. If the server or slot is not properly configured, or if
subscription fails, appropriate error handling is performed.
Args:
@@ -527,12 +518,12 @@ class IngestorManager(BaseActivity):
Logs:
- Logs informational messages about the subscription process.
- Logs errors if the server is not found, subscription creation fails, or
- Logs errors if the server is not found, subscription creation fails, or
tag subscription fails.
- Logs a warning if a subscription is removed due to failure.
Raises:
Exception: Any unexpected exceptions during subscription creation or tag
Exception: Any unexpected exceptions during subscription creation or tag
subscription are logged but not propagated.
Side Effects:
@@ -541,77 +532,63 @@ class IngestorManager(BaseActivity):
- Updates error metrics and notifications
"""
self.logger.info(
f"Subscribing to tags from {slot}:{server}"
)
self.logger.info(f'Subscribing to tags from {slot}:{server}')
tags_to_sub = server_config.get('tags')
if server not in self.opc_managers:
self.logger.error(
f"Server {server} not found in opc_managers."
)
self.logger.error(f'Server {server} not found in opc_managers.')
return 1
if slot not in self.opc_managers[server].subscriptions:
try:
await self.opc_managers[server].create_subscription(
slot
)
await self.opc_managers[server].create_subscription(slot)
except Exception as e:
self.logger.error(
f"Failed to create subscription for slot {slot}: {e}"
)
self.logger.error(f'Failed to create subscription for slot {slot}: {e}')
return 2
try:
self.logger.info(
tags_to_sub
)
self.logger.info(tags_to_sub)
await self.opc_managers[server].subscribe(
slot, deepcopy(tags_to_sub), self.poll_interval
)
self.logger.info(
tags_to_sub
)
self.logger.info(tags_to_sub)
except Exception as e:
metrics.OPC_SUBSCRIPTION_ERRORS.labels(
pod_id=self.pod_id, server=server, slot=slot).inc()
pod_id=self.pod_id, server=server, slot=slot
).inc()
trace = traceback.format_exc()
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",
block='opc_manager',
level=NotificationLevel.ERROR,
attachment_content=trace
attachment_content=trace,
)
self.logger.error(trace)
self.logger.warning(
"Removing subscription from server "
f"{server} for slot {slot}"
)
self.logger.warning(f'Removing subscription from server {server} for slot {slot}')
await self.opc_managers[server].unsubscribe(slot)
return 2
return 0
async def subscribe_to_tags(self, tags: Dict) -> None:
async def subscribe_to_tags(self, tags: dict) -> None:
"""
Subscribes to a set of tags and manages their configurations.
This method processes a dictionary of tags, iterating through each slot and server
configuration. It attempts to manage the server configurations and removes any
configuration. It attempts to manage the server configurations and removes any
servers that return a specific response code.
Args:
tags (Dict): A dictionary containing tag configurations. The structure is
tags (Dict): A dictionary containing tag configurations. The structure is
expected to be {slot: {server: server_config}}.
Side Effects:
- Logs the provided tags for debugging purposes.
- Updates the `managed_tags` attribute by removing servers that meet the
- Updates the `managed_tags` attribute by removing servers that meet the
removal criteria.
- Establishes OPC subscriptions for all configured tags.
Removal Criteria:
- If the `manage_server` method returns a response code of 2 for a given
- If the `manage_server` method returns a response code of 2 for a given
slot and server, that server is removed from the `managed_tags` attribute.
The method ensures that only successfully configured servers remain in the
@@ -622,9 +599,7 @@ class IngestorManager(BaseActivity):
self.logger.info(tags)
for slot, slot_config in tags.items():
for server, server_config in slot_config.items():
response = await self.manage_server(
slot, server, server_config, tags
)
response = await self.manage_server(slot, server, server_config, tags)
if response == 2:
to_remove.append([slot, server])

165
ingestor/managers/opc_manager.py
View File

@@ -1,15 +1,16 @@
import json
import asyncio
from pathlib import Path
from asyncua.crypto.security_policies import SecurityPolicyBasic256
from asyncua import Client
from sientia_do.notifications.models import NotificationLevel
from asyncua.crypto.security_policies import SecurityPolicyBasic256
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from sientia_do.temporal.constants import OPC_TIMEZONE, DATETIME_FORMAT_WITH_TZ
from ingestor.managers.data_manager import DataManager
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE
import ingestor.metrics as metrics
from ingestor.managers.data_manager import DataManager
class OpcManager(BaseActivity):
@@ -54,9 +55,19 @@ class OpcManager(BaseActivity):
metadata (dict): Application metadata
"""
def __init__(self, name: str, url: str, data_manager: DataManager, logger: Logger,
server_uri: str, notification_handler: NotificationHandler, metadata: dict,
cert_path: str = None, private_key_path: str = None, server_cert_path: str = None):
def __init__(
self,
name: str,
url: str,
data_manager: DataManager,
logger: Logger,
server_uri: str,
notification_handler: NotificationHandler,
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
@@ -71,14 +82,14 @@ class OpcManager(BaseActivity):
self.data_manager = data_manager
self.metadata = metadata
BaseActivity.__init__(self, logger=logger,
notification_handler=notification_handler,
set_error_counter=True)
BaseActivity.__init__(
self, logger=logger, notification_handler=notification_handler, set_error_counter=True
)
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)
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)
def __str__(self):
"""
@@ -87,8 +98,10 @@ class OpcManager(BaseActivity):
Returns:
str: Human-readable representation showing server details and current state.
"""
return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \
f"nodes={self.nodes}, subscriptions={self.subscriptions}"
return (
f'OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n'
f'nodes={self.nodes}, subscriptions={self.subscriptions}'
)
async def shutdown(self):
"""
@@ -103,10 +116,9 @@ class OpcManager(BaseActivity):
and ensure clean disconnection from OPC servers.
"""
try:
await self.disconnect()
except Exception as e:
self.logger.error(f"Error during cleanup: {e}")
self.logger.error(f'Error during cleanup: {e}')
async def set_security(self):
"""
@@ -133,11 +145,11 @@ class OpcManager(BaseActivity):
if not all([self.cert_path, self.private_key_path]):
raise ValueError(
"Certificate and private key paths must be provided for secure connection.")
'Certificate and private key paths must be provided for secure connection.'
)
cert = Path(self.cert_path)
private_key = Path(self.private_key_path)
server_cert = Path(
self.server_cert_path) if self.server_cert_path else None
server_cert = Path(self.server_cert_path) if self.server_cert_path else None
await self.client.set_application_uri(self.server_uri)
self.logger.info('Setting security...')
@@ -145,7 +157,7 @@ class OpcManager(BaseActivity):
SecurityPolicyBasic256,
certificate=str(cert),
private_key=str(private_key),
server_certificate=str(server_cert)
server_certificate=str(server_cert),
)
await self.client.set_secure_channel_timeout(10000000)
await self.client.set_session_timeout(10000000)
@@ -172,8 +184,7 @@ class OpcManager(BaseActivity):
- OPC_CONNECTION_STATUS: Set to 1 on successful connection
"""
metrics.OPC_CONNECTIONS_TOTAL.labels(
pod_id=self.pod_id, server_name=self.name).inc()
metrics.OPC_CONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc()
try:
self.client = Client(self.url, watchdog_intervall=3600000)
self.client.name = self.pod_id
@@ -182,14 +193,15 @@ class OpcManager(BaseActivity):
self.logger.info(f'Starting connection to {self.name}...')
await self.client.connect()
metrics.OPC_CONNECTION_STATUS.labels(
pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(1)
pod_id=self.pod_id, server_name=self.name, server_url=self.url
).set(1)
self.logger.info(f'Connection to {self.name} successful.')
except Exception as e:
metrics.OPC_CONNECTION_STATUS.labels(
pod_id=self.pod_id, server_name=self.name, server_url=self.url).set(0)
metrics.OPC_CONNECTIONS_FAILED.labels(
pod_id=self.pod_id, server_name=self.name).inc()
self.logger.error(f"Failed to connect to {self.name}: {e}")
pod_id=self.pod_id, server_name=self.name, server_url=self.url
).set(0)
metrics.OPC_CONNECTIONS_FAILED.labels(pod_id=self.pod_id, server_name=self.name).inc()
self.logger.error(f'Failed to connect to {self.name}: {e}')
raise
async def create_subscription(self, name: str, period: int = 500):
@@ -214,16 +226,16 @@ class OpcManager(BaseActivity):
"""
if not self.client:
raise ValueError("Client not connected. Call connect first.")
raise ValueError('Client not connected. Call connect first.')
try:
p = period if period is not None else 500
self.subscriptions[name] = await self.client.create_subscription(p, self)
self.logger.info(f'Subscription {name} created on {self.name}.')
metrics.OPC_SUBSCRIPTIONS_CREATED.labels(
pod_id=self.pod_id, server_name=self.name, slot_name=name).inc()
pod_id=self.pod_id, server_name=self.name, slot_name=name
).inc()
except Exception as e:
self.logger.error(
f"Failed to create subscription {name} on {self.name}: {e}")
self.logger.error(f'Failed to create subscription {name} on {self.name}: {e}')
raise
async def subscribe(self, subscription: str, nodes: dict, collect_period: int):
@@ -252,23 +264,23 @@ class OpcManager(BaseActivity):
"""
if not self.subscriptions.get(subscription):
raise ValueError(
"Subscription not created. Call create_subscription first.")
raise ValueError('Subscription not created. Call create_subscription first.')
self.logger.info(f"Subscribing to {subscription} on {self.name}...")
self.logger.info(f"Subscribing to nodes: {nodes}")
self.logger.info(f'Subscribing to {subscription} on {self.name}...')
self.logger.info(f'Subscribing to nodes: {nodes}')
addr_nodes = [self.client.get_node(n) for n in nodes]
self.logger.debug(f"Addr nodes: {addr_nodes}")
self.logger.debug(f'Addr nodes: {addr_nodes}')
self.nodes.update(nodes)
self.logger.debug(f"Nodes: {self.nodes}")
metrics.OPC_TAGS_SUBSCRIBED.labels(
pod_id=self.pod_id, server_name=self.name).set(len(self.nodes))
self.logger.debug(f'Nodes: {self.nodes}')
metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(
len(self.nodes)
)
self.collect_period = collect_period
for node, config in self.nodes.items():
self.nodes[node]['cycle_rule'] = {
'cycle_increment': collect_period*1000/float(config['frequency']),
'cycle_count': 0
'cycle_increment': collect_period * 1000 / float(config['frequency']),
'cycle_count': 0,
}
await self.subscriptions[subscription].subscribe_data_change(addr_nodes)
@@ -288,18 +300,17 @@ class OpcManager(BaseActivity):
- An info message upon successful unsubscription.
Behavior:
- If the subscription exists, it is deleted and removed from the
- If the subscription exists, it is deleted and removed from the
subscriptions dictionary.
- If the subscription does not exist, no action is taken.
"""
if not self.subscriptions.get(subscription):
self.logger.warning(
f"Subscription '{subscription}' not found. Cannot unsubscribe.")
self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.")
return
await self.subscriptions[subscription].delete()
del self.subscriptions[subscription]
self.logger.info(f"Unsubscribed from {subscription}.")
self.logger.info(f'Unsubscribed from {subscription}.')
async def disconnect(self):
"""
@@ -322,28 +333,27 @@ class OpcManager(BaseActivity):
self.logger.warning('Disconnecting from OPC server')
if self.client is None:
self.logger.warning("Client already disconnected.")
self.logger.warning('Client already disconnected.')
return
try:
for sub in self.subscriptions:
await self.subscriptions[sub].delete()
self.logger.warning("Deleted all subscriptions.")
self.logger.warning('Deleted all subscriptions.')
except Exception as sub_error:
self.logger.error(f"Failed to clean up subscription: {sub_error}")
self.logger.error(f'Failed to clean up subscription: {sub_error}')
try:
await self.client.disconnect()
except Exception as conn_error:
self.logger.error(
f"Failed to disconnect from OPC UA server: {conn_error}")
self.logger.error(f'Failed to disconnect from OPC UA server: {conn_error}')
finally:
del self.client
self.client = None
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)
self.logger.warning("Disconnected from OPC UA server.")
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)
self.logger.warning('Disconnected from OPC UA server.')
async def datachange_notification(self, node, _val, data):
"""
@@ -371,35 +381,33 @@ class OpcManager(BaseActivity):
monitored_item = data.monitored_item
value = monitored_item.Value.Value.Value
# source_timestamp
source_timestamp = monitored_item.Value.SourceTimestamp.replace(
tzinfo=OPC_TIMEZONE)
source_timestamp = monitored_item.Value.SourceTimestamp.replace(tzinfo=OPC_TIMEZONE)
tag = str(node)
self.logger.debug(
f"Data change notification received for tag:"
f"{tag} after {self.nodes[tag]['cycle_rule']['cycle_count']} cycles")
f'Data change notification received for tag:'
f'{tag} after {self.nodes[tag]["cycle_rule"]["cycle_count"]} cycles'
)
self.nodes[tag]['cycle_rule']['cycle_count'] = 0
self.non_receive_count = 0
metrics.OPC_CYCLES_WITHOUT_DATA.labels(
pod_id=self.pod_id, server_name=self.name).set(0)
metrics.OPC_CYCLES_WITHOUT_DATA.labels(pod_id=self.pod_id, server_name=self.name).set(0)
data = {
'tag': tag,
'name': self.nodes[str(node)]['tag_name'],
'timestamp': source_timestamp.strftime(DATETIME_FORMAT_WITH_TZ),
'value': value
'value': value,
}
_a = [self.data_manager.publish(e, data)
for e in self.nodes[tag]['topics']]
_a = [self.data_manager.publish(e, data) for e in self.nodes[tag]['topics']]
def check_cycles(self):
"""
Checks the cycle counts for all monitored nodes and sends
notifications if thresholds are exceeded.
This method iterates through all monitored nodes and updates their cycle counts based on
This method iterates through all monitored nodes and updates their cycle counts based on
configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers
a warning notification.
@@ -408,8 +416,7 @@ class OpcManager(BaseActivity):
- Sends warning notifications for nodes exceeding cycle thresholds
"""
for node, config in self.nodes.items():
self.nodes[node]['cycle_rule']['cycle_count'] += config[
'cycle_rule']['cycle_increment']
self.nodes[node]['cycle_rule']['cycle_count'] += config['cycle_rule']['cycle_increment']
if self.nodes[node]['cycle_rule']['cycle_count'] >= 5:
name = config['tag_name']
cycles = self.nodes[node]['cycle_rule']['cycle_count']
@@ -417,8 +424,8 @@ class OpcManager(BaseActivity):
metadata=self.metadata,
notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED',
message=f'{cycles} cycles without receive from {node}:{name}',
block="opc_manager",
level=NotificationLevel.WARNING
block='opc_manager',
level=NotificationLevel.WARNING,
)
def check_opc_listenning(self) -> bool:
@@ -441,26 +448,26 @@ class OpcManager(BaseActivity):
"""
self.non_receive_count += 1
metrics.OPC_CYCLES_WITHOUT_DATA.labels(
pod_id=self.pod_id, server_name=self.name).set(self.non_receive_count)
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.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)}',
block="opc_manager",
level=NotificationLevel.ERROR
block='opc_manager',
level=NotificationLevel.ERROR,
)
if self.non_receive_count >= 15:
metrics.OPC_RECONNECTIONS_TOTAL.labels(
pod_id=self.pod_id, server_name=self.name).inc()
metrics.OPC_RECONNECTIONS_TOTAL.labels(pod_id=self.pod_id, server_name=self.name).inc()
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",
level=NotificationLevel.ERROR
block='opc_manager',
level=NotificationLevel.ERROR,
)
return True
return False

52
ingestor/managers/resource_manager.py
View File

@@ -1,13 +1,14 @@
import json
from typing import List
from redis import Redis
from time import time
import ingestor.metrics as metrics
from redis import Redis
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.temporal.activities.base import BaseActivity
import ingestor.metrics as metrics
class ResourceManager(BaseActivity):
"""
@@ -80,9 +81,9 @@ class ResourceManager(BaseActivity):
Metrics:
- REDIS_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure
"""
BaseActivity.__init__(self, logger=logger,
notification_handler=notification_handler,
set_error_counter=True)
BaseActivity.__init__(
self, logger=logger, notification_handler=notification_handler, set_error_counter=True
)
try:
self.redis = Redis(
host=host,
@@ -94,7 +95,7 @@ class ResourceManager(BaseActivity):
self.redis.ping()
metrics.REDIS_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(1)
except Exception as e:
self.logger.error(f"Failed to connect to Redis: {e}")
self.logger.error(f'Failed to connect to Redis: {e}')
metrics.REDIS_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0)
raise
@@ -146,9 +147,9 @@ class ResourceManager(BaseActivity):
).inc()
self.send_notification(
metadata=self.metadata,
notification_id=f"REDIS_OPERATION_ERROR_{operation_name}",
notification_id=f'REDIS_OPERATION_ERROR_{operation_name}',
message=f"Error in Redis operation '{operation_name}': {e}",
block="redis_manager",
block='redis_manager',
level=NotificationLevel.ERROR,
)
raise
@@ -172,7 +173,7 @@ class ResourceManager(BaseActivity):
- REDIS_OPERATIONS_DURATION: Records timing for get operations
"""
history = self._execute_redis_op("get", self.redis.get, key)
history = self._execute_redis_op('get', self.redis.get, key)
return json.loads(history) if history else None
def get_tag_slot(self, id: str) -> dict:
@@ -193,7 +194,7 @@ class ResourceManager(BaseActivity):
and delegates to the get() method for the actual Redis operation.
"""
return self.get(f"slot:opc_tags:{id}")
return self.get(f'slot:opc_tags:{id}')
def ingestor_heartbeat(self) -> None:
"""
@@ -215,9 +216,9 @@ class ResourceManager(BaseActivity):
"""
self._execute_redis_op(
"set",
'set',
self.redis.set,
f"heartbeat:ingestor:{self.pod_id}",
f'heartbeat:ingestor:{self.pod_id}',
1,
ex=self.heartbeat_ttl,
)
@@ -249,9 +250,9 @@ class ResourceManager(BaseActivity):
"""
return self._execute_redis_op(
"set_nx",
'set_nx',
self.redis.set,
f"lease:opc_tags:{tag_id}",
f'lease:opc_tags:{tag_id}',
self.pod_id,
nx=True,
ex=self.lease_ttl,
@@ -282,12 +283,10 @@ class ResourceManager(BaseActivity):
- REDIS_OPERATIONS_DURATION: Records timing for renewal operations
"""
current = self._execute_redis_op(
"get", self.redis.get, f"lease:opc_tags:{tag_id}"
)
current = self._execute_redis_op('get', self.redis.get, f'lease:opc_tags:{tag_id}')
if current == self.pod_id:
self._execute_redis_op(
"expire", self.redis.expire, f"lease:opc_tags:{tag_id}", self.lease_ttl
'expire', self.redis.expire, f'lease:opc_tags:{tag_id}', self.lease_ttl
)
return True
return False
@@ -313,10 +312,9 @@ class ResourceManager(BaseActivity):
- REDIS_OPERATIONS_DURATION: Records timing for lease dropping operations
"""
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]:
def get_all_ingestors(self) -> list[str]:
"""
Retrieves all active ingestors from Redis.
@@ -338,9 +336,9 @@ class ResourceManager(BaseActivity):
- REDIS_OPERATIONS_DURATION: Records timing for ingestor discovery
"""
return self._execute_redis_op("keys", self.redis.keys, "heartbeat:ingestor:*")
return self._execute_redis_op('keys', self.redis.keys, 'heartbeat:ingestor:*')
def get_all_slots(self) -> List[str]:
def get_all_slots(self) -> list[str]:
"""
Retrieves all available slots from Redis.
@@ -362,9 +360,9 @@ class ResourceManager(BaseActivity):
- REDIS_OPERATIONS_DURATION: Records timing for slot discovery
"""
return self._execute_redis_op("keys", self.redis.keys, "slot:opc_tags:*")
return self._execute_redis_op('keys', self.redis.keys, 'slot:opc_tags:*')
def get_all_leases(self) -> List[str]:
def get_all_leases(self) -> list[str]:
"""
Retrieves all active leases from Redis.
@@ -386,4 +384,4 @@ class ResourceManager(BaseActivity):
- REDIS_OPERATIONS_DURATION: Records timing for lease discovery
"""
return self._execute_redis_op("keys", self.redis.keys, "lease:opc_tags:*")
return self._execute_redis_op('keys', self.redis.keys, 'lease:opc_tags:*')

136
ingestor/metrics.py
View File

@@ -17,157 +17,157 @@ labels for multi-dimensional analysis and alerting.
from prometheus_client import Counter, Gauge, Histogram
# Metric label definitions for consistent labeling across all metrics
POD_ID_LABEL = ["pod_id"]
SERVER_LABELS = ["pod_id", "server_name", "server_url"]
KAFKA_LABELS = ["pod_id", "topic"]
REDIS_LABELS = ["pod_id", "operation"]
NOTIFICATION_LABELS = ["pod_id", "level", "block"]
POD_ID_LABEL = ['pod_id']
SERVER_LABELS = ['pod_id', 'server_name', 'server_url']
KAFKA_LABELS = ['pod_id', 'topic']
REDIS_LABELS = ['pod_id', 'operation']
NOTIFICATION_LABELS = ['pod_id', 'level', 'block']
MAIN_LABELS = ["pod_id"]
MAIN_LABELS = ['pod_id']
# --- Reliability Metrics ---
TAG_WRITTEN_COUNT = Counter(
"ingestor_tag_written_count",
"Number of writing process to the collection",
[*MAIN_LABELS, "tag_name", "collection_name"],
'ingestor_tag_written_count',
'Number of writing process to the collection',
[*MAIN_LABELS, 'tag_name', 'collection_name'],
)
# --- General Application Metrics ---
APP_LOOP_COUNT = Counter(
"app_main_loop_total",
"Total number of times the application main loop has run",
'app_main_loop_total',
'Total number of times the application main loop has run',
POD_ID_LABEL,
)
APP_LOOP_DURATION = Histogram(
"app_main_loop_duration_seconds",
"Duration of the application main loop in seconds",
'app_main_loop_duration_seconds',
'Duration of the application main loop in seconds',
POD_ID_LABEL,
)
APP_ERRORS_TOTAL = Counter(
"app_errors_total",
"Total number of unhandled errors in the main loop",
'app_errors_total',
'Total number of unhandled errors in the main loop',
POD_ID_LABEL,
)
APP_UP = Gauge(
"app_up",
"Indicates if the application is running (1) or shutting down (0)",
'app_up',
'Indicates if the application is running (1) or shutting down (0)',
POD_ID_LABEL,
)
# --- Ingestor Manager Metrics ---
ACTIVE_INGESTORS = Gauge(
"ingestor_active_total",
"Number of active ingestors reported by Redis",
'ingestor_active_total',
'Number of active ingestors reported by Redis',
)
SLOTS_TOTAL = Gauge(
"ingestor_slots_total",
"Total number of slots configured in Redis",
'ingestor_slots_total',
'Total number of slots configured in Redis',
)
LEASES_TOTAL = Gauge(
"ingestor_leases_total",
"Total number of leases (allocated slots) in Redis",
'ingestor_leases_total',
'Total number of leases (allocated slots) in Redis',
)
SLOTS_MANAGED = Gauge(
"ingestor_slots_managed_current",
"Number of slots currently managed by this ingestor instance",
'ingestor_slots_managed_current',
'Number of slots currently managed by this ingestor instance',
POD_ID_LABEL,
)
SLOTS_ACQUIRED = Counter(
"ingestor_slots_acquired_total",
"Total number of slots acquired by this instance",
'ingestor_slots_acquired_total',
'Total number of slots acquired by this instance',
POD_ID_LABEL,
)
SLOTS_RELEASED = Counter(
"ingestor_slots_released_total",
"Total number of slots released by this instance",
'ingestor_slots_released_total',
'Total number of slots released by this instance',
POD_ID_LABEL,
)
OPC_MANAGERS_ACTIVE = Gauge(
"ingestor_opc_managers_active",
"Number of active OPC Managers in this instance",
'ingestor_opc_managers_active',
'Number of active OPC Managers in this instance',
POD_ID_LABEL,
)
OPC_SUBSCRIPTION_ERRORS = Counter(
"ingestor_opc_subscription_errors_total",
"Errors when trying to subscribe to OPC tags",
["pod_id", "server", "slot"],
'ingestor_opc_subscription_errors_total',
'Errors when trying to subscribe to OPC tags',
['pod_id', 'server', 'slot'],
)
# --- OPC Manager Metrics ---
OPC_CONNECTIONS_TOTAL = Counter(
"opc_connections_initiated_total",
"Total connection attempts to OPC servers",
["pod_id", "server_name"],
'opc_connections_initiated_total',
'Total connection attempts to OPC servers',
['pod_id', 'server_name'],
)
OPC_CONNECTIONS_FAILED = Counter(
"opc_connections_failed_total",
"Total failed connection attempts to OPC servers",
["pod_id", "server_name"],
'opc_connections_failed_total',
'Total failed connection attempts to OPC servers',
['pod_id', 'server_name'],
)
OPC_CONNECTION_STATUS = Gauge(
"opc_connection_status",
"Connection status with the OPC server (1=connected, 0=disconnected)",
'opc_connection_status',
'Connection status with the OPC server (1=connected, 0=disconnected)',
SERVER_LABELS,
)
OPC_SUBSCRIPTIONS_CREATED = Counter(
"opc_subscriptions_created_total",
"Total OPC subscriptions created",
["pod_id", "server_name", "slot_name"],
'opc_subscriptions_created_total',
'Total OPC subscriptions created',
['pod_id', 'server_name', 'slot_name'],
)
OPC_TAGS_SUBSCRIBED = Gauge(
"opc_tags_subscribed_current",
"Current number of OPC tags subscribed on a server",
["pod_id", "server_name"],
'opc_tags_subscribed_current',
'Current number of OPC tags subscribed on a server',
['pod_id', 'server_name'],
)
OPC_CYCLES_WITHOUT_DATA = Gauge(
"opc_cycles_without_data",
"Current number of cycles without receiving data from a server",
["pod_id", "server_name"],
'opc_cycles_without_data',
'Current number of cycles without receiving data from a server',
['pod_id', 'server_name'],
)
OPC_RECONNECTIONS_TOTAL = Counter(
"opc_reconnections_tried_total",
"Reconnection attempts to an OPC server after a loss",
["pod_id", "server_name"],
'opc_reconnections_tried_total',
'Reconnection attempts to an OPC server after a loss',
['pod_id', 'server_name'],
)
# --- Data Manager (Kafka) Metrics ---
KAFKA_MESSAGES_SENT = Counter(
"kafka_messages_sent_total", "Total messages sent to Kafka", KAFKA_LABELS
'kafka_messages_sent_total', 'Total messages sent to Kafka', KAFKA_LABELS
)
KAFKA_MESSAGES_ERRORS = Counter(
"kafka_messages_errors_total",
"Total errors sending messages to Kafka",
'kafka_messages_errors_total',
'Total errors sending messages to Kafka',
KAFKA_LABELS,
)
KAFKA_CONNECTION_STATUS = Gauge(
"kafka_connection_status",
"Connection status with Kafka (1=connected, 0=disconnected)",
'kafka_connection_status',
'Connection status with Kafka (1=connected, 0=disconnected)',
POD_ID_LABEL,
)
# --- Resource Manager (Redis) Metrics ---
REDIS_OPERATIONS_TOTAL = Counter(
"redis_operations_total", "Total number of Redis operations performed", REDIS_LABELS
'redis_operations_total', 'Total number of Redis operations performed', REDIS_LABELS
)
REDIS_OPERATIONS_ERRORS = Counter(
"redis_operations_errors_total",
"Total number of errors in Redis operations",
'redis_operations_errors_total',
'Total number of errors in Redis operations',
REDIS_LABELS,
)
REDIS_OPERATIONS_DURATION = Histogram(
"redis_operations_duration_seconds",
"Duration of Redis operations in seconds",
'redis_operations_duration_seconds',
'Duration of Redis operations in seconds',
REDIS_LABELS,
)
REDIS_CONNECTION_STATUS = Gauge(
"redis_connection_status",
"Connection status with Redis (1=connected, 0=disconnected)",
'redis_connection_status',
'Connection status with Redis (1=connected, 0=disconnected)',
POD_ID_LABEL,
)
# --- Notification Metrics ---
NOTIFICATIONS_SENT = Counter(
"notifications_sent_total",
"Total number of notifications sent",
'notifications_sent_total',
'Total number of notifications sent',
NOTIFICATION_LABELS,
)