b9ef69a865
Refactor OpcManager initialization and unsubscribe method; enhance logging - Removed the `initialize_from_config` method from `OpcManager` class and adjusted the constructor to handle configuration directly. - Improved the `unsubscribe` method to include detailed logging for non-existent subscriptions. - Updated tests in `test_opc_manager.py` to reflect changes in the `OpcManager` class. - Commented out Docker-related fixtures in `conftest.py` for potential future use. - Enhanced test coverage in `test_single_node.py` and `test_data_manager.py` with additional scenarios and assertions. - Introduced new methods in `Ingestor` and `IngestorManager` classes to manage server subscriptions and leases more effectively. - Added new tests for `Ingestor` class to validate initialization and slot management logic. - Implemented logging improvements across various classes to ensure better traceability of actions and errors.
219 lines
9.7 KiB
Python
219 lines
9.7 KiB
Python
from logging import Formatter, StreamHandler, getLogger
|
|
from os import getenv
|
|
|
|
from ingestor.managers.ingestor_manager import IngestorManager
|
|
|
|
|
|
class Ingestor:
|
|
def __init__(self):
|
|
"""
|
|
Initializes the ingestor with configuration values retrieved from environment variables.
|
|
Environment Variables:
|
|
KAFKA_SERVERS (str): Comma-separated list of Kafka server addresses. Defaults to "localhost:9092".
|
|
REDIS_HOST (str): Hostname of the Redis server. Defaults to "localhost".
|
|
REDIS_PORT (int): Port number of the Redis server. Defaults to 6379.
|
|
LEASE_TTL (int): Time-to-live for leases in seconds. Defaults to 10.
|
|
HEARTBEAT_TTL (int): Time-to-live for heartbeats in seconds. Defaults to 20.
|
|
HOSTNAME (str): Identifier for the current pod or host. Defaults to "localhost".
|
|
POLL_INTERVAL (int): Interval in seconds for polling operations. Defaults to 5.
|
|
Attributes:
|
|
kafka_servers (list): List of Kafka server addresses.
|
|
redis_host (str): Hostname of the Redis server.
|
|
redis_port (int): Port number of the Redis server.
|
|
lease_ttl (int): Time-to-live for leases in seconds.
|
|
heartbeat_ttl (int): Time-to-live for heartbeats in seconds.
|
|
pod_id (str): Identifier for the current pod or host.
|
|
poll_interval (int): Interval in seconds for polling operations.
|
|
"""
|
|
|
|
kafka_servers = getenv("KAFKA_SERVERS", "localhost:9092")
|
|
self.redis_host = getenv("REDIS_HOST", "localhost")
|
|
self.redis_port = int(getenv("REDIS_PORT", 6379))
|
|
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))
|
|
|
|
self.kafka_servers = kafka_servers.split(",")
|
|
self.init_logger()
|
|
|
|
def init_logger(self):
|
|
"""
|
|
Initializes a logger instance for the class.
|
|
This method sets up a logger with a specified log level, a stream handler,
|
|
and a formatter. The log level is determined by the environment variable
|
|
"LOG_LEVEL", defaulting to "INFO" if not set. The logger is then attached
|
|
to the instance for use throughout the class.
|
|
Attributes:
|
|
self.logger (logging.Logger): The configured logger instance.
|
|
"""
|
|
|
|
logger = getLogger(__name__)
|
|
logger.setLevel(getenv("LOG_LEVEL", "INFO"))
|
|
handler = StreamHandler()
|
|
formatter = Formatter(
|
|
'%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
handler.setFormatter(formatter)
|
|
|
|
logger.addHandler(handler)
|
|
|
|
self.logger = logger
|
|
|
|
def handle_acquired_tags(self, acquired):
|
|
"""
|
|
Handles the acquired tags by subscribing to them if available.
|
|
This method checks if there are any acquired tags. If no tags are acquired,
|
|
it logs a warning indicating that no slots are available. Otherwise, it
|
|
updates the OPC servers and subscribes to the acquired tags.
|
|
Args:
|
|
acquired (list): A list of acquired tags to be processed. If the list
|
|
is empty or None, no action is taken other than logging
|
|
a warning.
|
|
"""
|
|
|
|
if not acquired:
|
|
self.logger.warning("No slots available")
|
|
|
|
else:
|
|
# Subscribe to acquired slots
|
|
self.ingestor_manager.update_opc_servers()
|
|
self.ingestor_manager.subscribe_to_tags(acquired)
|
|
|
|
def prepare_ingestor(self):
|
|
"""
|
|
Prepares the ingestor by initializing the IngestorManager, declaring the ingestor as active,
|
|
acquiring slot leases, and handling the acquired tags.
|
|
This method performs the following steps:
|
|
1. Initializes the `IngestorManager` with the necessary configuration parameters.
|
|
2. Declares the ingestor as active by calling `declare_active` on the `IngestorManager`.
|
|
3. Acquires slot leases using the `get_slot_leases` method of the `IngestorManager`.
|
|
4. Logs the acquired slots and processes them using the `handle_acquired_tags` method.
|
|
Attributes:
|
|
self.kafka_servers (list): List of Kafka server addresses.
|
|
self.redis_host (str): Redis server hostname.
|
|
self.redis_port (int): Redis server port.
|
|
self.lease_ttl (int): Time-to-live for slot leases.
|
|
self.heartbeat_ttl (int): Time-to-live for heartbeat signals.
|
|
self.pod_id (str): Identifier for the current pod.
|
|
self.poll_interval (int): Interval for polling operations.
|
|
self.logger (Logger): Logger instance for logging messages.
|
|
Raises:
|
|
Exception: If any error occurs during the initialization or lease acquisition process.
|
|
"""
|
|
|
|
self.ingestor_manager = IngestorManager(
|
|
self.kafka_servers, self.redis_host, self.redis_port, self.lease_ttl,
|
|
self.heartbeat_ttl, self.pod_id, self.poll_interval, self.logger
|
|
)
|
|
|
|
# Declare ingestor ative
|
|
self.ingestor_manager.declare_active()
|
|
|
|
# Get slot lease
|
|
acquired = self.ingestor_manager.get_slot_leases()
|
|
self.logger.info(f"Acquired slots: {acquired}")
|
|
|
|
self.handle_acquired_tags(acquired)
|
|
|
|
def manage_no_slots(self, number_of_slots: int):
|
|
"""
|
|
Manages the scenario where there are no slots assigned to the ingestor.
|
|
This method checks if the ingestor is active (i.e., has no managed tags)
|
|
and if the number of available slots is greater than zero. If both
|
|
conditions are met, it attempts to acquire a slot lease and handles
|
|
the acquired tags accordingly.
|
|
Args:
|
|
number_of_slots (int): The number of available slots.
|
|
"""
|
|
|
|
if not self.ingestor_manager.managed_tags and number_of_slots > 0:
|
|
# This ingestor is active and has no slots, so we need to try to
|
|
|
|
# Get slot lease
|
|
acquired = self.ingestor_manager.get_slot_leases(1)
|
|
|
|
self.handle_acquired_tags(acquired)
|
|
|
|
def manage_leases(self, ingestor_diff: int, slot_diff: int):
|
|
"""
|
|
Manages the allocation and deallocation of slot leases based on the
|
|
differences in the number of active ingestors and available slots.
|
|
Args:
|
|
ingestor_diff (int): The difference between the required and available
|
|
ingestors. A positive value indicates that there are inactive
|
|
ingestors and available slots.
|
|
slot_diff (int): The difference between the required and available
|
|
slots. A positive value indicates that there are active ingestors
|
|
without assigned slots.
|
|
Behavior:
|
|
- If `ingestor_diff` is greater than 0, it means there are available
|
|
slots due to inactive ingestors. The method will acquire slot leases
|
|
for the available slots and handle the acquired tags.
|
|
- If `slot_diff` is greater than 0, it means there are active ingestors
|
|
without slots. The method will drop slot leases for the excess
|
|
managed tags.
|
|
"""
|
|
|
|
if ingestor_diff > 0:
|
|
# Some ingestors are innactive, so theres "ingestor_diff" slots available
|
|
self.logger.info(f"Slots available: {ingestor_diff}")
|
|
|
|
# Get slot lease
|
|
acquired = self.ingestor_manager.get_slot_leases(ingestor_diff)
|
|
|
|
self.handle_acquired_tags(acquired)
|
|
|
|
elif slot_diff > 0:
|
|
# Some ingestors are active and without slots, so we need to drop
|
|
|
|
overleases = list(self.ingestor_manager.managed_tags.keys())[1:]
|
|
|
|
self.ingestor_manager.drop_slot_leases(overleases)
|
|
|
|
def loop(self):
|
|
"""
|
|
Executes the main loop for managing ingestors and slots.
|
|
This method performs the following tasks:
|
|
1. Declares the ingestor as active.
|
|
2. Logs the start of the polling process for slot updates.
|
|
3. Retrieves the list of active ingestors and the number of available slots.
|
|
4. Handles scenarios where no slots are available.
|
|
5. Calculates the difference between the number of slots and active ingestors,
|
|
as well as the difference in managed tags.
|
|
6. Manages leases based on the calculated differences.
|
|
7. Logs the current state of active ingestors, slots, managed tags, and servers.
|
|
8. Logs a message if no slots are acquired during the loop.
|
|
9. Updates the configuration of OPC servers.
|
|
This method is intended to be called repeatedly to ensure the ingestor
|
|
manager operates correctly and maintains synchronization with the slots
|
|
and OPC servers.
|
|
"""
|
|
|
|
self.ingestor_manager.declare_active()
|
|
|
|
self.logger.info("Polling for slot updates...")
|
|
# Get active ingestors
|
|
ingestors = self.ingestor_manager.get_active_ingestors()
|
|
number_of_slots = self.ingestor_manager.get_number_of_slots()
|
|
|
|
# Handle no slots
|
|
self.manage_no_slots(number_of_slots)
|
|
|
|
ingestor_diff = number_of_slots - len(ingestors)
|
|
slot_diff = len(self.ingestor_manager.managed_tags) - 1
|
|
|
|
self.manage_leases(ingestor_diff, slot_diff)
|
|
|
|
self.logger.info(
|
|
f"Active ingestors: {ingestors}, "
|
|
f"Number of slots: {number_of_slots}, "
|
|
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")
|
|
|
|
# Update opc servers
|
|
self.ingestor_manager.update_slot_config()
|