da079c6320
Refactor ingestor lease management and update Docker configurations - Updated CMD in Dockerfile to use the correct application entry point. - Added docker compose command to remove volumes in README. - Removed unnecessary restart policy from ingestor service in docker-compose.yaml. - Refactored manage_leases method in Ingestor class to improve clarity and parameter naming. - Added get_number_of_leases method in IngestorManager to retrieve active leases. - Implemented get_all_leases method in ResourceManager to fetch active leases from Redis. - Enhanced unit tests for lease management in test_ingestor_manager and test_resource_manager.
383 lines
16 KiB
Python
383 lines
16 KiB
Python
from logging import Logger
|
|
import traceback
|
|
from typing import Dict, List
|
|
from copy import deepcopy
|
|
from ingestor.managers.data_manager import DataManager
|
|
from ingestor.managers.opc_manager import OpcManager
|
|
from ingestor.managers.resource_manager import ResourceManager
|
|
|
|
|
|
class IngestorManager():
|
|
def __init__(self,
|
|
kafka_servers: str, redis_host: str, redis_port: int,
|
|
lease_ttl: int, heartbeat_ttl: int, pod_id: str,
|
|
poll_interval: int, logger: Logger):
|
|
|
|
self.data_manager = DataManager(kafka_servers, logger)
|
|
self.opc_managers = {}
|
|
self.resource_manager = ResourceManager(
|
|
redis_host, redis_port, lease_ttl, heartbeat_ttl, pod_id
|
|
)
|
|
self.number_of_slots = 0
|
|
self.poll_interval = poll_interval
|
|
self.logger = logger
|
|
self.managed_tags = {}
|
|
self.opc_servers = {}
|
|
|
|
def initialize_opc_from_config(self, server_config: dict, data_manager: DataManager, logger: Logger) -> OpcManager | None:
|
|
"""
|
|
Initializes an OPC Manager instance using the provided server configuration.
|
|
Args:
|
|
server_config (dict): A dictionary containing the OPC server configuration.
|
|
Expected keys include:
|
|
- 'name' (str): The name of the OPC server.
|
|
- 'url' (str): The URL of the OPC server.
|
|
- 'server_uri' (str): The URI of the OPC server.
|
|
- 'cert_path' (str, optional): Path to the client certificate file.
|
|
- 'private_key_path' (str, optional): Path to the private key file.
|
|
- 'server_cert_path' (str, optional): Path to the server certificate file.
|
|
data_manager (DataManager): An instance of the DataManager to handle data operations.
|
|
logger (Logger): A logger instance for logging messages.
|
|
Returns:
|
|
OpcManager | None: An initialized OpcManager instance if successful,
|
|
otherwise None if an error occurs during initialization.
|
|
"""
|
|
|
|
try:
|
|
manager = OpcManager(
|
|
server_config['name'], server_config['url'], data_manager, logger, server_config['server_uri'],
|
|
server_config.get('cert_path'), server_config.get(
|
|
'private_key_path'), server_config.get('server_cert_path')
|
|
)
|
|
|
|
manager.config = server_config
|
|
manager.connect()
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize OpcManager: {e}")
|
|
return None
|
|
|
|
return manager
|
|
|
|
def update_opc_servers(self):
|
|
"""
|
|
Updates the OPC (OLE for Process Control) server connections managed by the ingestor.
|
|
This method ensures that the OPC servers defined in `self.managed_tags` are properly
|
|
initialized and updated. It performs the following tasks:
|
|
- Registers new OPC servers based on the configuration in `self.managed_tags`.
|
|
- Updates existing OPC server instances if their configuration has changed.
|
|
- Disconnects and removes OPC servers that are no longer present in `self.managed_tags`.
|
|
Steps:
|
|
1. Iterates through the `self.managed_tags` dictionary to identify and register servers.
|
|
2. Initializes new OPC server instances if they are not already managed.
|
|
3. Reinitializes OPC server instances if their configuration has changed.
|
|
4. Disconnects and removes OPC servers that are no longer registered.
|
|
Attributes:
|
|
self.managed_tags (dict): A nested dictionary containing slot and server configurations.
|
|
self.opc_managers (dict): A dictionary mapping server names to their OPC manager instances.
|
|
self.data_manager: An object responsible for managing data operations.
|
|
self.logger: A logging object for recording warnings and other messages.
|
|
Raises:
|
|
Any exceptions raised during OPC server initialization or disconnection.
|
|
Logs:
|
|
- Warnings for servers that are no longer found in `self.managed_tags`.
|
|
"""
|
|
|
|
registered_servers = []
|
|
for slot, slot_config in self.managed_tags.items():
|
|
for server, server_config in slot_config.items():
|
|
registered_servers.append(server)
|
|
server_config = server_config.copy()
|
|
server_config.pop('tags', None)
|
|
server_instance = None
|
|
if server not in self.opc_managers:
|
|
|
|
server_instance = self.initialize_opc_from_config(
|
|
server_config, self.data_manager, self.logger
|
|
)
|
|
|
|
elif self.opc_managers[server].config != server_config:
|
|
del self.opc_managers[server]
|
|
server_instance = self.initialize_opc_from_config(
|
|
server_config, self.data_manager, self.logger
|
|
)
|
|
|
|
if server_instance is not None:
|
|
self.opc_managers[server] = server_instance
|
|
|
|
for server in list(self.opc_managers.keys()):
|
|
if server not in registered_servers:
|
|
self.logger.warning(
|
|
f"Server {server} not found in managed tags. "
|
|
f"Desconnecting from server."
|
|
)
|
|
self.opc_managers[server].disconnect()
|
|
self.opc_managers.pop(server, None)
|
|
|
|
def declare_active(self):
|
|
"""
|
|
Declares the ingestor as active by sending a heartbeat signal to the resource manager.
|
|
This method ensures that the ingestor is marked as active by invoking the
|
|
`ingestor_heartbeat` method of the associated resource manager.
|
|
"""
|
|
|
|
self.resource_manager.ingestor_heartbeat()
|
|
|
|
def get_active_ingestors(self) -> List[str]:
|
|
"""
|
|
Retrieve a list of active ingestors.
|
|
This method fetches all ingestors from the resource manager and returns them.
|
|
If no ingestors are found, an empty list is returned.
|
|
Returns:
|
|
List[str]: A list of active ingestor names, or an empty list if none are found.
|
|
"""
|
|
|
|
ingestors = self.resource_manager.get_all_ingestors()
|
|
return ingestors if ingestors else []
|
|
|
|
def get_number_of_leases(self) -> int:
|
|
"""
|
|
Retrieves the number of leases managed by the resource manager.
|
|
This method fetches all available leases from the resource manager,
|
|
calculates their count, and updates the `number_of_slots` attribute.
|
|
Returns:
|
|
int: The total number of leases. Returns 0 if no leases are available.
|
|
"""
|
|
|
|
leases = self.resource_manager.get_all_leases()
|
|
self.number_of_slots = len(leases) if leases else 0
|
|
return self.number_of_slots
|
|
|
|
def get_number_of_slots(self) -> int:
|
|
"""
|
|
Retrieves the number of slots managed by the resource manager.
|
|
This method fetches all available slots from the resource manager,
|
|
calculates their count, and updates the `number_of_slots` attribute.
|
|
Returns:
|
|
int: The total number of slots. Returns 0 if no slots are available.
|
|
"""
|
|
|
|
slots = self.resource_manager.get_all_slots()
|
|
self.number_of_slots = len(slots) if slots else 0
|
|
return self.number_of_slots
|
|
|
|
def get_slot_leases(self, max_slots: int = 1) -> Dict:
|
|
"""
|
|
Acquires a specified number of resource slots by leasing them from the resource manager.
|
|
Args:
|
|
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)
|
|
and the values are the leased slot details.
|
|
Behavior:
|
|
- Iterates through available slots and attempts to lease them using the resource manager.
|
|
- Logs the leasing of each slot.
|
|
- Updates the `managed_tags` attribute with the acquired slots.
|
|
- Stops leasing once the specified `max_slots` are acquired.
|
|
- If unable to acquire the requested number of slots, logs a warning and returns the slots that were leased.
|
|
Notes:
|
|
- If a slot is leased but its details cannot be retrieved (i.e., `get_tag_slot` returns None),
|
|
that slot is skipped.
|
|
"""
|
|
|
|
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}")
|
|
slots = self.resource_manager.get_tag_slot(str(i))
|
|
if slots is None:
|
|
continue
|
|
acquired[str(i)] = slots
|
|
|
|
if len(acquired) >= max_slots:
|
|
self.managed_tags.update(acquired)
|
|
return acquired
|
|
|
|
self.logger.warning(
|
|
f"Unable to acquire {max_slots} slots. "
|
|
f"Only {acquired} slots were leased."
|
|
)
|
|
self.managed_tags.update(acquired)
|
|
return acquired
|
|
|
|
def unsubscribe_slot(self, slot: str):
|
|
"""
|
|
Unsubscribes a specific slot from all associated OPC servers.
|
|
Args:
|
|
slot (str): The name of the slot to unsubscribe.
|
|
Raises:
|
|
KeyError: If the specified slot does not exist in the managed tags.
|
|
"""
|
|
|
|
for server in self.managed_tags[slot].keys():
|
|
if server in self.opc_managers:
|
|
self.opc_managers[server].unsubscribe(slot)
|
|
|
|
def update_slot_config(self):
|
|
"""
|
|
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:
|
|
1. Renews the lease for each managed slot using the resource manager.
|
|
2. Fetches the latest configuration for each slot.
|
|
3. Logs and removes slots whose configurations are no longer available.
|
|
4. Updates the configuration of slots if changes are detected.
|
|
5. Unsubscribes and re-subscribes to slots with updated configurations.
|
|
6. Removes slots from the managed tags if they are no longer valid.
|
|
7. Updates the OPC servers after processing all slots.
|
|
Side Effects:
|
|
- Modifies the `managed_tags` dictionary to reflect the latest slot configurations.
|
|
- Updates OPC server subscriptions based on the current state of managed slots.
|
|
Raises:
|
|
- None explicitly, but relies on the behavior of `resource_manager` and
|
|
other dependencies for error handling.
|
|
Logging:
|
|
- Logs warnings for removed slots.
|
|
- Logs informational messages for updated slot configurations.
|
|
"""
|
|
|
|
removed_slots = []
|
|
update = {}
|
|
for slot, slot_config in self.managed_tags.items():
|
|
self.resource_manager.renew_tag_lease(slot)
|
|
update = self.resource_manager.get_tag_slot(slot)
|
|
if update is None:
|
|
self.logger.warning(
|
|
f"Slot {slot} configuration not found. "
|
|
f"Removing slot from managed tags."
|
|
)
|
|
self.unsubscribe_slot(slot)
|
|
removed_slots.append(slot)
|
|
|
|
continue
|
|
|
|
if update != slot_config:
|
|
self.logger.info(
|
|
f"Slot {slot} configuration updated. "
|
|
f"Old: {slot_config}, New: {update}"
|
|
)
|
|
self.managed_tags[slot] = update
|
|
|
|
self.unsubscribe_slot(slot)
|
|
self.update_opc_servers()
|
|
self.subscribe_to_tags({slot: update})
|
|
|
|
for slot in removed_slots:
|
|
del self.managed_tags[slot]
|
|
self.update_opc_servers()
|
|
|
|
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
|
|
the lease for each ID.
|
|
Args:
|
|
ids (List[str]): A list of slot IDs for which the leases
|
|
should be released.
|
|
Returns:
|
|
None
|
|
"""
|
|
|
|
for id in ids:
|
|
self.resource_manager.drop_tag_lease(id)
|
|
|
|
def manage_server(self, slot: str, server: str, server_config: dict, tags: dict) -> int:
|
|
"""
|
|
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
|
|
subscription fails, appropriate error handling is performed.
|
|
Args:
|
|
slot (str): The slot identifier for the subscription.
|
|
server (str): The name of the OPC server.
|
|
server_config (dict): Configuration dictionary for the server, which includes
|
|
the tags to be subscribed under the key 'tags'.
|
|
tags (dict): A dictionary of tags to be subscribed.
|
|
Returns:
|
|
int: Status code indicating the result of the operation:
|
|
- 0: Subscription was successful.
|
|
- 1: Server not found in `opc_managers`.
|
|
- 2: Subscription creation or tag subscription failed.
|
|
Logs:
|
|
- Logs informational messages about the subscription process.
|
|
- 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
|
|
subscription are logged but not propagated.
|
|
"""
|
|
|
|
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."
|
|
)
|
|
return 1
|
|
if slot not in self.opc_managers[server].subscriptions:
|
|
try:
|
|
self.opc_managers[server].create_subscription(
|
|
slot
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(
|
|
f"Failed to create subscription for slot {slot}: {e}"
|
|
)
|
|
return 2
|
|
try:
|
|
self.logger.info(
|
|
tags_to_sub
|
|
)
|
|
self.opc_managers[server].subscribe(
|
|
slot, deepcopy(tags_to_sub), self.poll_interval
|
|
)
|
|
self.logger.info(
|
|
tags_to_sub
|
|
)
|
|
except Exception as e:
|
|
self.logger.error(
|
|
f"Failed to subscribe to tags from {slot}:{server}\n{tags}: {e}"
|
|
)
|
|
self.logger.error(traceback.format_exc())
|
|
self.logger.warning(
|
|
"Removing subscription from server "
|
|
f"{server} for slot {slot}"
|
|
)
|
|
self.opc_managers[server].unsubscribe(slot)
|
|
return 2
|
|
return 0
|
|
|
|
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
|
|
servers that return a specific response code.
|
|
Args:
|
|
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
|
|
removal criteria.
|
|
Removal Criteria:
|
|
- 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.
|
|
"""
|
|
|
|
to_remove = []
|
|
self.logger.info(tags)
|
|
for slot, slot_config in tags.items():
|
|
for server, server_config in slot_config.items():
|
|
response = self.manage_server(
|
|
slot, server, server_config, tags
|
|
)
|
|
if response == 2:
|
|
to_remove.append([slot, server])
|
|
|
|
for slot, server in to_remove:
|
|
self.managed_tags[slot].pop(server, None)
|