Merge pull request #1 from Aignosi/SIENTIAPDE-988-criar-ingestor-opc

Sientiapde 988 criar ingestor opc

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.pytestArgs": [
+        "."
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

Dockerfile

@@ -0,0 +1,30 @@
+# syntax=docker/dockerfile:1.4
+
+from python:3.11-slim
+
+RUN apt-get update && apt-get install -y git openssh-client && rm -rf /var/lib/apt/lists/*
+
+# Set the working directory
+WORKDIR /app
+
+# Copy the requirements file into the container
+COPY requirements.txt .
+COPY __init__.py .
+
+# Copy code into the container
+COPY ./ingestor ./ingestor
+
+# Install the required packages
+# Add github to known hosts
+# This is needed for SSH to work
+# The SSH key will NOT remain in the image
+# IMPORTANT: this block requires BuildKit
+# and the --ssh flag during docker build
+RUN --mount=type=ssh \
+    mkdir -p ~/.ssh && \
+    ssh-keyscan github.com >> ~/.ssh/known_hosts && \
+    pip install --no-cache-dir -r requirements.txt
+
+
+# Run the application
+CMD ["python", "-m", "ingestor.app"]

README.md

@@ -1,2 +1,60 @@
 # sientia-dataops-opc-gateway
 OPC gateway to manage Scouter pipelines
+
+## Local tests
+### Generate your ssh key to Docker
+'''
+ssh-keygen -t ed25519 -C "docker-access" -f ~/.ssh/id_ed25519_docker
+'''
+Add the public key to yout Git SSH keys
+
+### Enable Docker BuildKit
+'''
+export DOCKER_BUILDKIT=1
+''' 
+or make it permanent:
+'''
+echo '{ "features": { "buildkit": true } }' | sudo tee /etc/docker/daemon.json
+sudo systemctl restart docker
+'''
+
+### Run docker compose
+'''
+docker compose down -v
+docker compose build --ssh default=$HOME/.ssh/id_ed25519_docker
+docker compose up -d
+'''
+
+### Populate redis server
+Create venv with python3.11
+'''
+python3.11 -m venv venv
+source ./venv/bin/activate
+'''
+Install requirements
+'''
+pip install -r requirements.txt
+'''
+Run feeder
+'''
+python simulator/redis-feeder.py
+'''
+
+## Unit tests
+### Install pytest
+'''
+pip install pytest
+'''
+### Run pytest
+'''
+pytest
+'''
+### Get current coverage
+'''
+pip install pytest-cov
+pytest --cov=ingestor
+'''
+### Generate complete report
+'''
+pytest --cov=ingestor --cov-report=html
+'''

docker-compose.yaml

@@ -0,0 +1,102 @@
+version: '3.8'
+
+services:
+  ingestor:
+    build:
+      context: .
+    environment:
+      HOSTNAME: ingestor
+    container_name: ingestor
+    depends_on:
+      - kafka
+      - redis
+    networks:
+      - kafka-net
+    env_file:
+      - .env
+
+  zookeeper:
+    image: confluentinc/cp-zookeeper:latest
+    container_name: zookeeper
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    networks:
+      - kafka-net
+    env_file:
+      - .env
+
+  kafka:
+    image: confluentinc/cp-kafka:latest
+    container_name: kafka
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092,
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+    depends_on:
+      - zookeeper
+    networks:
+      - kafka-net
+    env_file:
+      - .env
+
+  redis:
+    image: redis:latest
+    container_name: redis
+    ports:
+      - "6379:6379"
+    networks:
+      - kafka-net
+    env_file:
+      - .env
+
+  redis-commander:
+    image: rediscommander/redis-commander:latest
+    container_name: redis-commander
+    environment:
+      REDIS_HOSTS: local:redis:6379
+    ports:
+      - "8081:8081"
+    depends_on:
+      - redis
+    networks:
+      - kafka-net
+
+
+  kafdrop:
+    image: obsidiandynamics/kafdrop:latest
+    networks: 
+      - kafka-net
+    depends_on:
+      - kafka
+    ports:
+      - 19000:9000
+    environment:
+      KAFKA_BROKERCONNECT: kafka:29092
+
+  simulator:
+    build:
+      context: .
+      dockerfile: simulator/Dockerfile
+      args:
+        GIT_REPO: ${SIMULATOR_GIT_REPO}
+        GIT_BRANCH: ${SIMULATOR_GIT_BRANCH}
+    container_name: simulator
+    ports:
+      - "4840:4840"
+    depends_on:
+      - kafka
+      - redis
+    networks:
+      - kafka-net
+    env_file:
+      - .env
+
+networks:
+  kafka-net:
+    driver: bridge

ingestor/app.py

@@ -0,0 +1,19 @@
+from time import sleep
+from ingestor.ingestor import Ingestor
+
+
+def main():
+    ingestor = Ingestor()
+
+    ingestor.prepare_ingestor()
+
+    while True:
+
+        ingestor.loop()
+
+        # Sleep for poll interval
+        sleep(ingestor.poll_interval)
+
+
+if __name__ == "__main__":
+    main()

ingestor/ingestor.py

@@ -0,0 +1,223 @@
+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, available_slots: int, lacking_ingestors: int, slot_diff: int):
+        """
+        Manages the allocation and deallocation of slot leases for ingestors based on 
+        the number of available slots, lacking ingestors, and slot differences.
+        Args:
+            available_slots (int): The number of slots currently available for allocation.
+            lacking_ingestors (int): The number of ingestors that are active and without slots.
+            slot_diff (int): The difference between the total slots and the required slots.
+        Behavior:
+            - If there are available slots and lacking ingestors, attempts to acquire slot leases 
+              for the available slots and processes the acquired tags.
+            - If there are no lacking ingestors but there are extra slots (slot_diff > 0), 
+              releases the extra slot leases to ensure proper allocation.
+        Logs:
+            - Logs the number of available slots when attempting to acquire leases.
+            - Logs the number of extra slots when releasing leases.
+        """
+
+        if available_slots > 0 and lacking_ingestors > 0:
+            # Some ingestors are innactive, so theres "available_slots" slots available
+            self.logger.info(f"Slots available: {available_slots}")
+
+            # Get slot lease
+            acquired = self.ingestor_manager.get_slot_leases(available_slots)
+
+            self.handle_acquired_tags(acquired)
+
+        elif lacking_ingestors == 0 and slot_diff > 0:
+
+            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
+
+            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_ingestors = len(ingestors)
+        number_of_leases = self.ingestor_manager.get_number_of_leases()
+        number_of_slots = self.ingestor_manager.get_number_of_slots()
+
+        # Handle 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.manage_leases(available_slots, lacking_ingestors, slot_diff)
+
+        self.logger.info(
+            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")
+
+        # Update opc servers
+        self.ingestor_manager.update_slot_config()

ingestor/managers/data_manager.py

@@ -0,0 +1,94 @@
+import json
+from logging import Logger
+from time import sleep
+from kafka import KafkaProducer
+from kafka.errors import NoBrokersAvailable
+
+
+class DataManager():
+    def __init__(self, kafka_servers: str, logger: Logger) -> None:
+        """
+        Initializes the DataManager instance with a Kafka producer.
+        This constructor attempts to establish a connection to the specified Kafka servers
+        and initializes a Kafka producer for sending messages. It retries the connection
+        up to 3 times if the Kafka servers are unavailable.
+        Args:
+            kafka_servers (str): A comma-separated string of Kafka server addresses.
+            logger (Logger): A logger instance for logging messages.
+        Raises:
+            NoBrokersAvailable: If the connection to Kafka servers fails after 3 attempts.
+        """
+
+        self.kafka_producer = None
+        for i in range(0, 3):
+            logger.info(
+                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'),  # Serialize JSON messages
+                    key_serializer=lambda k: str(
+                        k).encode('utf-8') if k else None,
+                )
+                break
+            except NoBrokersAvailable:
+                logger.error(
+                    f"Kafka servers {kafka_servers} are not available. Retrying...")
+                sleep(5)
+        else:
+            logger.error(
+                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.")
+
+        logger.info(
+            f"DataManager initialized with Kafka servers: {kafka_servers}")
+        self.logger = logger
+
+    def __del__(self):
+        """Destructor to close the producer connection."""
+        print("Closing Kafka producer...")
+        if self.kafka_producer:
+            self.kafka_producer.flush(timeout=10)
+            self.kafka_producer.close()
+        else:
+            print("Kafka producer is already closed or not initialized.")
+
+    def delivery_report(self, msg: str):
+        """Callback for delivery reports from Kafka."""
+        self.logger.debug(
+            f"Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}")
+
+    def delivery_error(self, err: str):
+        """Callback for delivery reports from Kafka."""
+        self.logger.error(f"Delivery failed for record : {err}")
+
+    def publish(self, topic: str, data: dict) -> None:
+        """
+        Publishes a message to a specified Kafka topic.
+
+        Args:
+            topic (str): The name of the Kafka topic to which the message will be published.
+            data (dict): The message data to be sent to the Kafka topic.
+
+        Returns:
+            None
+
+        Raises:
+            Exception: If there is an error during message delivery, it will be handled by the `delivery_error` callback.
+        """
+
+        try:
+
+            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)
+
+        except Exception as e:
+            self.logger.error(f"Failed to publish message: {e}")

ingestor/managers/ingestor_manager.py

@@ -0,0 +1,382 @@
+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)

ingestor/managers/opc_manager.py

@@ -0,0 +1,302 @@
+import json
+from logging import Logger
+from pathlib import Path
+from typing import Callable
+from asyncua.crypto.security_policies import SecurityPolicyBasic256
+from asyncua.sync import Client
+from sientia_do.notifications.models import NotificationLevel
+
+from ingestor.managers.data_manager import DataManager
+
+
+class OpcManager():
+    def __init__(self, name: str, url: str, data_manager: DataManager,
+                 logger: Logger, server_uri: str, 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
+        self.data_queue = {}
+        self.logger = logger
+        self.non_receive_count = 0
+        self.client = None
+        self.cert_path = cert_path
+        self.private_key_path = private_key_path
+        self.server_cert_path = server_cert_path
+        self.nodes = {}
+        self.subscriptions = {}
+        self.data_manager = data_manager
+
+    def __str__(self):
+        return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \
+            f"nodes={self.nodes}, subscriptions={self.subscriptions}"
+
+    def set_security(self):
+        """
+        Configures the security settings for the OPC UA client.
+        This method sets up the security policy, certificates, and timeouts
+        required for establishing a secure connection with the OPC UA server.
+        Raises:
+            ValueError: If either the certificate path or private key path is not provided.
+        Attributes:
+            cert_path (str): Path to the client's certificate file.
+            private_key_path (str): Path to the client's private key file.
+            server_cert_path (str, optional): Path to the server's certificate file.
+            server_uri (str): The URI of the server to be used as the application URI.
+            client (opcua.Client): The OPC UA client instance.
+            logger (logging.Logger): Logger instance for logging information.
+        Security Settings:
+            - Security Policy: Basic256
+            - Secure Channel Timeout: 10,000,000 ms
+            - Session Timeout: 10,000,000 ms
+        """
+
+        if not all([self.cert_path, self.private_key_path]):
+            raise ValueError(
+                "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
+
+        self.client.application_uri = self.server_uri
+        self.logger.info('Setting security...')
+        self.client.set_security(
+            SecurityPolicyBasic256,
+            certificate=str(cert),
+            private_key=str(private_key),
+            server_certificate=str(server_cert)
+        )
+        self.client.secure_channel_timeout = 10000000
+        self.client.session_timeout = 10000000
+
+    def connect(self):
+        """
+        Establishes a connection to the OPC server.
+        This method initializes the OPC client using the provided URL and
+        sets up security if a certificate path is specified. It then
+        attempts to connect to the server and logs the connection status.
+        Raises:
+            Exception: If the connection to the OPC server fails.
+        """
+
+        self.client = Client(self.url)
+        if self.cert_path:
+            self.set_security()
+        self.logger.info('Starting connection...')
+        self.client.connect()
+
+    def create_subscription(self, name: str, period: int = 500):
+        """
+        Creates a subscription with the specified monitoring period.
+        This method establishes a subscription to monitor data changes or events
+        from the OPC UA server. If the client is not connected, an exception is raised.
+        Args:
+            period (int, optional): The monitoring period in milliseconds. Defaults to 500 ms.
+        Raises:
+            ValueError: If the client is not connected.
+        Side Effects:
+            - Sets the `self.period` attribute to the specified or default period.
+            - Creates a subscription and assigns it to `self.subscription`.
+            - Logs the creation of the subscription.
+        """
+
+        if not self.client:
+            raise ValueError("Client not connected. Call connect first.")
+
+        p = period if period != None else 500
+        self.subscriptions[name] = self.client.create_subscription(
+            p, self)
+        self.logger.info('Subscription created.')
+
+    def subscribe(self, subscription: str, nodes: dict, collect_period: int):
+        """
+        Subscribes to a set of OPC UA nodes for data change notifications.
+        This method adds the specified nodes to the subscription and configures
+        their data collection rules based on the provided collection period and
+        node-specific frequency.
+        Args:
+            nodes (dict): A dictionary where keys are node identifiers (e.g., node
+                IDs or paths) and values are configurations for each node. Each
+                configuration must include a 'frequency' key indicating the
+                frequency of data collection in Hz.
+            collect_period (int): The data collection period in seconds.
+        Raises:
+            ValueError: If the subscription has not been created by calling
+                `create_subscription` prior to this method.
+        """
+
+        if not self.subscriptions.get(subscription):
+            raise ValueError(
+                "Subscription not created. Call create_subscription first.")
+
+        self.logger.info(f"Subscribing to {subscription}...")
+        self.logger.info(f"Subscribing to nodes: {nodes}")
+        self.addr_nodes = [self.client.get_node(
+            n) for n in nodes if n not in self.nodes]
+        self.nodes.update(nodes)
+        self.collect_period = collect_period
+
+        for node, config in self.nodes.items():
+            self.nodes[node]['cycle_rule'] = {
+                'cycle_increment': collect_period*1000/config['frequency'],
+                'cycle_count': 0
+            }
+
+        self.subscriptions[subscription].subscribe_data_change(self.addr_nodes)
+
+    def unsubscribe(self, subscription: str):
+        """
+        Unsubscribes from a given subscription.
+        Args:
+            subscription (str): The name of the subscription to unsubscribe from.
+        Logs:
+            - A warning if the specified subscription does not exist.
+            - An info message upon successful unsubscription.
+        Behavior:
+            - 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.")
+            return
+        self.subscriptions[subscription].delete()
+        del self.subscriptions[subscription]
+        self.logger.info(f"Unsubscribed from {subscription}.")
+
+    def __del__(self):
+        self.disconnect()
+
+    def disconnect(self):
+        """
+        Disconnects from the OPC UA server.
+        This method handles the disconnection process by deleting the subscription
+        and disconnecting the client from the OPC UA server. It logs the disconnection
+        process and handles any exceptions that may occur during cleanup.
+        Raises:
+            Exception: If an error occurs while deleting the subscription or disconnecting
+                       from the OPC UA server, it logs the error details.
+        """
+
+        self.logger.warning('Disconnecting from OPC server')
+        if self.client is None:
+            self.logger.warning("Client already disconnected.")
+            return
+        try:
+            [self.subscriptions[sub].delete() for sub in self.subscriptions]
+            self.logger.warning("Deleted all subscriptions.")
+            del self.client
+            self.client = None
+            self.logger.warning("Disconnected from OPC UA server.")
+        except Exception as sub_error:
+            self.logger.error(f"Failed to clean up subscription: {sub_error}")
+
+    def datachange_notification(self, node, _val, data):
+        """
+        Handles data change notifications for monitored OPC UA nodes.
+        This method is triggered when a monitored node's value changes. It processes
+        the notification, updates internal state, and publishes the data to the
+        appropriate topics.
+        Args:
+            node (NodeId): The OPC UA node that triggered the data change notification.
+            _val (Any): The new value of the node (unused in this implementation).
+            data (DataChangeNotification): The data change notification object containing
+                details about the change.
+        Behavior:
+            - Extracts the value and source timestamp from the monitored item.
+            - Resets the cycle count for the node's cycle rule.
+            - Resets the non-receive count.
+            - Constructs a data dictionary containing the tag, tag name, timestamp, and value.
+            - Publishes the data to all topics associated with the node.
+        """
+
+        # get data value
+        monitored_item = data.monitored_item
+        value = monitored_item.Value.Value.Value
+        # source_timestamp
+        source_timestamp = monitored_item.Value.SourceTimestamp
+        tag = str(node)
+
+        self.nodes[tag]['cycle_rule']['cycle_count'] = 0
+        self.non_receive_count = 0
+
+        data = {
+            'tag': tag,
+            'name': self.nodes[str(node)]['tag_name'],
+            'timestamp': source_timestamp.strftime('%Y-%m-%d %H:%M:%S'),
+            'value': value
+        }
+
+        [self.data_manager.publish(e, data)
+         for e in self.nodes[tag]['topics']]
+
+    def check_cycles(self, removed_data: dict, node: str, config: dict,
+                     handle_listen_events: Callable[[str, str, NotificationLevel], None]):
+        """
+        Checks the cycle count for a specific node and triggers a notification if the cycle count exceeds a threshold.
+        Args:
+            removed_data (dict): A dictionary containing data that has been removed.
+                                    Used to check if the node is present.
+            node (str): The identifier of the node being checked.
+            config (dict): Configuration dictionary containing metadata such as the tag name.
+            handle_listen_events (Callable[[str, str, NotificationLevel], None]):
+                A callback function to handle notification events. It takes three arguments:
+                - A message string describing the event.
+                - A tag string identifying the event.
+                - A NotificationLevel enum indicating the severity of the event.
+        Behavior:
+            - If the node is not in `removed_data`, the cycle count for the node is incremented.
+            - If the cycle count reaches or exceeds 5, the `handle_listen_events` callback is invoked
+                with a warning message, a tag, and a notification level.
+        Notification Example:
+            If the cycle count exceeds the threshold, a warning message is generated in the format:
+            "{cycles} cycles without receive from {node}:{name}"
+            where `cycles` is the current cycle count, `node` is the node identifier, and `name` is the tag name
+            from the `config` dictionary.
+        """
+
+        if node not in removed_data.keys():
+            self.nodes[node]['cycle_rule']['cycle_count'] += self.nodes[node]['cycle_rule']['cycle_increment']
+        if self.nodes[node]['cycle_rule']['cycle_count'] >= 5 and handle_listen_events:
+            name = config['tag_name']
+            cycles = self.nodes[node]['cycle_rule']['cycle_count']
+            handle_listen_events(
+                f'{cycles} cycles without receive from {node}:{name}',
+                f'TAG_{node}:{name}_LISTENNING_STOPPED',
+                NotificationLevel.WARNING
+            )
+
+    def check_opc_listenning(self, handle_listen_events: Callable[[str, str, NotificationLevel], None]) -> None:
+        """
+        Monitors the OPC connection and triggers events based on the number of cycles
+        without receiving data from the OPC server.
+        Args:
+            handle_listen_events (Callable[[str, str, NotificationLevel], None]):
+                A callback function to handle notification events. It takes three arguments:
+                - A message string describing the event.
+                - An event code string.
+                - A NotificationLevel indicating the severity of the event.
+        Behavior:
+            - Increments the non-receive count each time the method is called.
+            - If the non-receive count reaches 5, triggers a notification event indicating
+              that the OPC server has stopped sending data.
+            - If the non-receive count reaches 15, triggers a notification event indicating
+              a retry to connect to the OPC server, disconnects the current session, and
+              reinitializes the collector with the existing configuration.
+        """
+
+        self.non_receive_count += 1
+        if self.non_receive_count >= 5 and handle_listen_events:
+            handle_listen_events(
+                f'{self.non_receive_count} cycles without receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}',
+                f'OPC_LISTENNING_STOPPED__{self.name}', NotificationLevel.ERROR)
+        if self.non_receive_count >= 15 and handle_listen_events:
+            handle_listen_events(
+                f'Retrying to connect to server {self.name}',
+                f'OPC_CONNECTION_RETRY__{self.name}', NotificationLevel.ERROR)
+            self.disconnect()
+            self.init_collector(
+                self.nodes, self.collect_period, self.period)

ingestor/managers/resource_manager.py

@@ -0,0 +1,130 @@
+import json
+from typing import List
+from redis import Redis
+
+
+class ResourceManager:
+    def __init__(self, host: str, port: int,
+                 lease_ttl: int, heartbeat_ttl: int, pod_id: str) -> None:
+        self.redis = Redis(host=host, port=port, decode_responses=True)
+        self.lease_ttl = lease_ttl
+        self.heartbeat_ttl = heartbeat_ttl
+        self.pod_id = pod_id
+
+    def get(self, key: str) -> dict:
+        """
+        Retrieve a value from Redis by its key and return it as a dictionary.
+        Args:
+            key (str): The key to look up in Redis.
+        Returns:
+            dict: The value associated with the key, parsed as a dictionary, 
+                  or None if the key does not exist or the value is empty.
+        """
+
+        history = self.redis.get(key)
+        return json.loads(history) if history else None
+
+    def get_tag_slot(self, id: str) -> dict:
+        """
+        Retrieve the tag slot information for a given ID.
+        Args:
+            id (str): The unique identifier of the tag slot to retrieve.
+        Returns:
+            dict: A dictionary containing the tag slot information associated with the given ID.
+        """
+
+        return self.get(f"slot:opc_tags:{id}")
+
+    def ingestor_heartbeat(self) -> None:
+        """
+        Sends a heartbeat signal to Redis to indicate that the ingestor is active.
+        This method sets a key in Redis with a specific format that includes the
+        ingestor's pod ID. The key is set with a value of 1 and an expiration
+        time defined by `self.heartbeat_ttl`. This allows monitoring systems to
+        track the activity and health of the ingestor.
+        Returns:
+            None
+        """
+
+        self.redis.set(
+            f"heartbeat:ingestor:{self.pod_id}", 1, ex=self.heartbeat_ttl)
+
+    def lease_tag(self, tag_id: str) -> bool:
+        """
+        Attempts to lease a tag by setting a key in Redis with a specified TTL (time-to-live).
+        This method uses the Redis `SET` command with the `NX` option to ensure that the key 
+        is only set if it does not already exist. The key is set with an expiration time 
+        defined by `lease_ttl`.
+        Args:
+            tag_id (str): The unique identifier of the tag to be leased.
+        Returns:
+            bool: True if the lease was successfully acquired, False otherwise.
+        """
+
+        return self.redis.set(
+            f"lease:opc_tags:{tag_id}", self.pod_id, nx=True, ex=self.lease_ttl)
+
+    def renew_tag_lease(self, tag_id: str) -> bool:
+        """
+        Renews the lease for a specific OPC tag if the current pod holds the lease.
+        This method checks if the current pod (identified by `self.pod_id`) holds the lease
+        for the given OPC tag. If so, it extends the lease by resetting its expiration time
+        in Redis to the configured lease TTL (`self.lease_ttl`).
+        Args:
+            tag_id (str): The identifier of the OPC tag whose lease is to be renewed.
+        Returns:
+            bool: True if the lease was successfully renewed, False otherwise.
+        """
+
+        current = self.redis.get(
+            f"lease:opc_tags:{tag_id}")
+        if current == self.pod_id:
+            self.redis.expire(f"lease:opc_tags:{tag_id}", self.lease_ttl)
+            return True
+
+        return False
+
+    def drop_tag_lease(self, tag_id: str) -> None:
+        """
+        Drops the lease for a specific OPC tag.
+        This method removes the lease for the given OPC tag by deleting the corresponding key in Redis.
+        Args:
+            tag_id (str): The identifier of the OPC tag whose lease is to be dropped.
+        Returns:
+            None
+        """
+
+        self.redis.delete(f"lease:opc_tags:{tag_id}")
+
+    def get_all_ingestors(self) -> List[str]:
+        """
+        Retrieves all active ingestors from Redis.
+        This method fetches all keys in Redis that match the pattern for ingestor leases
+        and returns a list of active ingestors.
+        Returns:
+            list: A list of active ingestors.
+        """
+
+        return self.redis.keys("heartbeat:ingestor:*")
+
+    def get_all_slots(self) -> List[str]:
+        """
+        Retrieves the number of slots available in Redis.
+        This method counts the number of keys in Redis that match the pattern for OPC tag leases
+        and returns the count.
+        Returns:
+            int: The number of slots available.
+        """
+
+        return self.redis.keys("slot:opc_tags:*")
+
+    def get_all_leases(self) -> List[str]:
+        """
+        Retrieves all active leases from Redis.
+        This method fetches all keys in Redis that match the pattern for OPC tag leases
+        and returns a list of active leases.
+        Returns:
+            list: A list of active leases.
+        """
+
+        return self.redis.keys("lease:opc_tags:*")

requirements.txt

@@ -0,0 +1,3 @@
+asyncua==1.1.5
+redis
+git+ssh://git@github.com/Aignosi/sientia-dataops-library.git

simulator/Dockerfile

@@ -0,0 +1,30 @@
+# syntax=docker/dockerfile:1.4
+
+FROM python:3.11-slim
+
+# Enable use of SSH agent/socket
+# This line enables SSH during build
+# (don't forget the syntax header above)
+RUN apt-get update && apt-get install -y git openssh-client && rm -rf /var/lib/apt/lists/*
+
+# Use build-time SSH mount for Git clone
+# The SSH key will NOT remain in the image
+# IMPORTANT: this block requires BuildKit
+# and the --ssh flag during docker build
+
+# SSH config to skip host key check (safe in CI/local dev)
+RUN mkdir -p /root/.ssh && echo "StrictHostKeyChecking no" > /root/.ssh/config
+
+WORKDIR /app
+
+# Clone using SSH
+ARG GIT_REPO
+ARG GIT_BRANCH=main
+
+# Mount SSH key just for this RUN
+RUN --mount=type=ssh git clone --branch ${GIT_BRANCH} ${GIT_REPO} .
+
+# Install requirements if exists
+RUN if [ -f requirements.txt ]; then pip install --no-cache-dir -r requirements.txt; fi
+
+CMD ["python", "server.py"]

simulator/redis-feeder.py

@@ -0,0 +1,78 @@
+import redis
+import json
+import os
+
+# Redis connection settings
+redis_host = "localhost"
+redis_port = 6379
+
+# Connect to Redis
+r = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
+
+# Define the key pattern to target
+pattern = "slot:opc_tags:*"
+
+# Step 1: Find and delete matching keys
+print("🔍 Searching for keys matching:", pattern)
+for key in r.scan_iter(match=pattern):
+    r.delete(key)
+    print(f"❌ Deleted: {key}")
+
+# Step 2: Insert new data
+# Example new OPC tag data
+new_data = {
+    "slot:opc_tags:1": {
+        "server1": {
+            "name": "server1",
+            "url": "opc.tcp://simulator:4840",
+            "server_uri": "http://opcua-server.simulator",
+            "tags": {
+                'ns=2;i=2': {
+                    'tag_name': 'Counter',
+                    'frequency': 1000,
+                    'topics': ['opcua', 'counter'],
+                },
+                'ns=2;i=3': {
+                    'tag_name': 'Rollout',
+                    'frequency': 1000,
+                    "topics": ['opcua', 'rollout'],
+                },
+                'ns=2;i=4': {
+                    'tag_name': 'Square',
+                    'frequency': 1000,
+                    "topics": ['opcua'],
+                },
+            }
+        }
+    },
+    "slot:opc_tags:2": {
+        "server2": {
+            "name": "server2",
+            "url": "opc.tcp://simulator:4840",
+            "server_uri": "http://opcua-server.simulator",
+            "tags": {
+                'ns=2;i=2': {
+                    'tag_name': 'Counter',
+                    'frequency': 1000,
+                    'topics': ['opcua2', 'counter'],
+                },
+                'ns=2;i=3': {
+                    'tag_name': 'Rollout',
+                    'frequency': 1000,
+                    "topics": ['opcua2', 'rollout'],
+                },
+                'ns=2;i=4': {
+                    'tag_name': 'Square',
+                    'frequency': 1000,
+                    "topics": ['opcua2'],
+                },
+            }
+        }
+    }
+}
+
+for key, val in new_data.items():
+    r.set(key, json.dumps(val))
+    print(f"✅ Set: {key} -> {val}")
+
+print("🚀 OPC tag keys replaced successfully.")

tests/functional/conftest.py

@@ -0,0 +1,47 @@
+# import subprocess
+# from time import sleep
+# from typing import Generator
+# import uuid
+# from kafka import KafkaConsumer
+# import pytest
+# from redis import Redis
+
+
+# @pytest.fixture(scope="session", autouse=True)
+# def docker_compose():
+#     """Sobe os containers antes dos testes e derruba depois."""
+#     print("\n🚀 Subindo Docker Compose...")
+#     subprocess.run(["docker", "compose", "up", "-d"], check=True)
+
+#     print("⏳ Aguardando containers ficarem prontos...")
+#     sleep(15)  # ajuste conforme necessário
+
+#     yield  # os testes rodam aqui
+
+#     print("\n🧹 Derrubando Docker Compose...")
+#     subprocess.run(["docker", "compose", "down"], check=True)
+
+
+# @pytest.fixture()
+# def redis_client():
+#     redis = Redis(host="localhost", port=6379, decode_responses=True)
+#     redis.flushdb()
+
+#     yield redis
+
+#     # Limpa o banco de dados após os testes
+#     redis.flushdb()
+#     redis.close()
+
+
+# def kafka_searcher(topic) -> Generator[KafkaConsumer, None, None]:
+#     consumer = KafkaConsumer(
+#         topic,
+#         bootstrap_servers="localhost:9092",
+#         group_id=f"test-group-{uuid.uuid4()}",
+#         auto_offset_reset="earliest",   # Começa a consumir apenas mensagens novas
+#         enable_auto_commit=True,
+#     )
+
+#     yield consumer
+#     consumer.close()

tests/functional/test_single_node.py

@@ -0,0 +1,118 @@
+# import json
+# import subprocess
+# from time import sleep
+
+# from tests.functional.conftest import kafka_searcher
+
+
+# new_data = {
+#     "slot:opc_tags:1": {
+#         "server1": {
+#             "name": "server1",
+#             "url": "opc.tcp://simulator:4840",
+#             "server_uri": "http://opcua-server.simulator",
+#             "tags": {
+#                 'ns=2;i=2': {
+#                     'tag_name': 'Counter',
+#                     'frequency': 1000,
+#                     'topics': [],
+#                 },
+#                 'ns=2;i=3': {
+#                     'tag_name': 'Rollout',
+#                     'frequency': 1000,
+#                     "topics": [],
+#                 },
+#                 'ns=2;i=4': {
+#                     'tag_name': 'Square',
+#                     'frequency': 1000,
+#                     "topics": [],
+#                 },
+#             }
+#         }
+#     },
+#     "slot:opc_tags:2": {
+#         "server2": {
+#             "name": "server2",
+#             "url": "opc.tcp://simulator:4840",
+#             "server_uri": "http://opcua-server.simulator",
+#             "tags": {
+#                 'ns=2;i=2': {
+#                     'tag_name': 'Counter',
+#                     'frequency': 1000,
+#                     'topics': [],
+#                 },
+#                 'ns=2;i=3': {
+#                     'tag_name': 'Rollout',
+#                     'frequency': 1000,
+#                     "topics": [],
+#                 },
+#                 'ns=2;i=4': {
+#                     'tag_name': 'Square',
+#                     'frequency': 1000,
+#                     "topics": [],
+#                 },
+#             }
+#         }
+#     }
+# }
+
+
+# def test_simple(redis_client):
+#     new_data['slot:opc_tags:1']['server1']['tags']['ns=2;i=2']['topics'] = [
+#         'test_topic_1']
+#     redis_client.set("slot:opc_tags:1",
+#                      json.dumps(new_data['slot:opc_tags:1']))
+
+#     sleep(20)  # Espera o Ingestor processar os dados
+
+#     # Check if lease is in Redis
+#     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
+#     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
+
+#     # Check if data is in Kafka
+
+#     kafka = next(kafka_searcher('test_topic_1'))
+#     sleep(1)
+#     messages = kafka.poll(timeout_ms=10000)
+
+#     assert messages, "Expected messages in Kafka, but got none."
+
+
+# def test_simple_double_slot(redis_client):
+
+#     new_data['slot:opc_tags:1']['server1']['tags']['ns=2;i=2']['topics'] = [
+#         'test_topic_double_slot1']
+#     redis_client.set("slot:opc_tags:1",
+#                      json.dumps(new_data['slot:opc_tags:1']))
+
+#     sleep(20)  # Espera o Ingestor processar os dados
+
+#     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
+#     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
+
+#     # Check if data is in Kafka
+#     kafka1 = next(kafka_searcher('test_topic_double_slot1'))
+#     messages = kafka1.poll(timeout_ms=10000)
+
+#     assert messages, "Expected messages in test_topic_double_slot1, but got none."
+
+#     new_data['slot:opc_tags:2']['server2']['tags']['ns=2;i=2']['topics'] = [
+#         'test_topic_double_slot2']
+#     redis_client.set("slot:opc_tags:2",
+#                      json.dumps(new_data['slot:opc_tags:2']))
+
+#     sleep(20)  # Espera o Ingestor processar os dados
+
+#     # Check if lease is in Redis
+#     assert redis_client.get("lease:opc_tags:2") == 'ingestor'
+#     assert redis_client.get("lease:opc_tags:1") == 'ingestor'
+#     assert redis_client.get("heartbeat:ingestor:ingestor") == '1'
+
+#     # Check if data is in Kafka
+#     kafka2 = next(kafka_searcher('test_topic_double_slot2'))
+#     messages = kafka2.poll(timeout_ms=10000)
+
+#     assert messages, "Expected messages in test_topic_double_slot2, but got none."
+
+#     messages = kafka1.poll(timeout_ms=10000)
+#     assert messages, "Expected messages in test_topic_double_slot1, but got none."

tests/unit/managers/test_data_manager.py

@@ -0,0 +1,194 @@
+from unittest.mock import ANY, MagicMock, patch
+from pytest import fixture
+from kafka.errors import NoBrokersAvailable
+
+from ingestor.managers.data_manager import DataManager
+
+
+@fixture
+@patch("ingestor.managers.data_manager.KafkaProducer")
+def data_manager(kafka):
+    return DataManager(
+        kafka_servers="localhost:9092",
+        logger=MagicMock()
+    )
+
+
+@patch("ingestor.managers.data_manager.KafkaProducer")
+def test___init___success(kafka):
+    logger_mock = MagicMock()
+
+    data_manager = DataManager(
+        kafka_servers="localhost:9092",
+        logger=logger_mock
+    )
+
+    kafka.assert_called_once_with(
+        bootstrap_servers="localhost:9092",
+        value_serializer=ANY,
+        key_serializer=ANY
+    )
+    assert data_manager.kafka_producer is not None
+    logger_mock.info.assert_any_call(
+        "Trying (0) to initializing DataManager with Kafka servers: localhost:9092"
+    )
+    logger_mock.info.assert_any_call(
+        "DataManager initialized with Kafka servers: localhost:9092"
+    )
+    logger_mock.error.assert_not_called()
+    assert logger_mock.info.call_count == 2
+
+
+@patch("ingestor.managers.data_manager.KafkaProducer")
+def test___init___second_attempt(kafka):
+    kafka.side_effect = [NoBrokersAvailable, MagicMock()]
+    logger_mock = MagicMock()
+
+    data_manager = DataManager(
+        kafka_servers="localhost:9092",
+        logger=logger_mock
+    )
+
+    kafka.assert_any_call(
+        bootstrap_servers="localhost:9092",
+        value_serializer=ANY,
+        key_serializer=ANY
+    )
+    assert kafka.call_count == 2
+    assert data_manager.kafka_producer is not None
+    logger_mock.info.assert_any_call(
+        "Trying (0) to initializing DataManager with Kafka servers: localhost:9092"
+    )
+    logger_mock.info.assert_any_call(
+        "Trying (1) to initializing DataManager with Kafka servers: localhost:9092"
+    )
+    logger_mock.info.assert_any_call(
+        "DataManager initialized with Kafka servers: localhost:9092"
+    )
+    logger_mock.error.assert_called_once_with(
+        "Kafka servers localhost:9092 are not available. Retrying..."
+    )
+    assert logger_mock.info.call_count == 3
+
+
+@patch("ingestor.managers.data_manager.KafkaProducer")
+def test___init___failure_max_attempts(kafka):
+    kafka.side_effect = NoBrokersAvailable
+    logger_mock = MagicMock()
+
+    try:
+        DataManager(
+            kafka_servers="localhost:9092",
+            logger=logger_mock
+        )
+    except NoBrokersAvailable as e:
+        assert str(
+            e) == "NoBrokersAvailable: Failed to connect to Kafka servers localhost:9092 after 3 attempts."
+
+        assert kafka.call_count == 3
+        logger_mock.info.assert_any_call(
+            "Trying (0) to initializing DataManager with Kafka servers: localhost:9092"
+        )
+        logger_mock.info.assert_any_call(
+            "Trying (1) to initializing DataManager with Kafka servers: localhost:9092"
+        )
+        logger_mock.info.assert_any_call(
+            "Trying (2) to initializing DataManager with Kafka servers: localhost:9092"
+        )
+        logger_mock.error.assert_called_with(
+            "Failed to connect to Kafka servers localhost:9092 after 3 attempts."
+        )
+        assert logger_mock.info.call_count == 3
+
+    else:
+        assert False, "Expected NoBrokersAvailable exception was not raised."
+
+
+def test___del___has_producer(data_manager):
+    flush_mock = MagicMock()
+    close_mock = MagicMock()
+
+    data_manager.kafka_producer.flush = flush_mock
+    data_manager.kafka_producer.close = close_mock
+
+    data_manager.__del__()
+    flush_mock.assert_called_once()
+    close_mock.assert_called_once()
+
+
+@patch("ingestor.managers.data_manager.print")
+def test___del___no_producer(print, data_manager):
+    data_manager.kafka_producer = None
+
+    # Call the __del__ method
+    data_manager.__del__()
+
+    # Check if the print statement was called
+    print.assert_any_call(
+        "Kafka producer is already closed or not initialized."
+    )
+
+
+def test_delivery_report(data_manager):
+    msg = MagicMock()
+    msg.topic = "test_topic"
+    msg.partition = 0
+    msg.offset = 1
+
+    data_manager.delivery_report(msg)
+
+    data_manager.logger.debug.assert_called_once_with(
+        f"Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}"
+    )
+
+
+def test_delivery_error(data_manager):
+    err = "Test error"
+    data_manager.delivery_error(err)
+
+    data_manager.logger.error.assert_called_once_with(
+        f"Delivery failed for record : {err}"
+    )
+
+
+def test_publish(data_manager):
+    topic = "test_topic"
+    data = {"key": "value"}
+
+    # Mock the send method of the Kafka producer
+    send_mock = MagicMock()
+    data_manager.kafka_producer.send = send_mock
+
+    # Call the publish method
+    data_manager.publish(topic, data)
+
+    # Check if the send method was called with the correct arguments
+    send_mock.assert_called_once_with(
+        topic=topic, value=data
+    )
+
+    send_mock.return_value.add_callback.assert_called_once()
+
+    data_manager.kafka_producer.flush.assert_called_once()
+
+
+def test_publish_error(data_manager):
+    topic = "test_topic"
+    data = {"key": "value"}
+
+    # Mock the send method of the Kafka producer to raise an exception
+    send_mock = MagicMock(side_effect=Exception("Test error"))
+    data_manager.kafka_producer.send = send_mock
+
+    # Call the publish method
+    data_manager.publish(topic, data)
+
+    # Check if the send method was called with the correct arguments
+    send_mock.assert_called_once_with(
+        topic=topic, value=data
+    )
+
+    # Check if the error was logged
+    data_manager.logger.error.assert_called_once_with(
+        "Failed to publish message: Test error"
+    )

tests/unit/managers/test_ingestor_manager.py

@@ -0,0 +1,477 @@
+from unittest.mock import MagicMock, patch
+from pytest import fixture
+
+from ingestor.managers.ingestor_manager import IngestorManager
+
+
+@fixture
+@patch('ingestor.managers.ingestor_manager.DataManager')
+@patch('ingestor.managers.ingestor_manager.ResourceManager')
+def ingestor_manager(data_manager_mock, resource_manager_mock):
+    return IngestorManager(
+        kafka_servers="localhost:9092",
+        redis_host="localhost",
+        redis_port=6379,
+        lease_ttl=60,
+        heartbeat_ttl=60,
+        pod_id="test_pod",
+        poll_interval=5,
+        logger=MagicMock()
+    )
+
+
+@patch('ingestor.managers.ingestor_manager.OpcManager')
+@patch('ingestor.managers.ingestor_manager.DataManager')
+@patch('ingestor.managers.ingestor_manager.ResourceManager')
+def test___init__(resource_manager_mock, data_manager_mock, opc_manager_mock):
+
+    ingestor = IngestorManager(
+        kafka_servers="localhost:9092",
+        redis_host="localhost",
+        redis_port=6379,
+        lease_ttl=60,
+        heartbeat_ttl=60,
+        pod_id="test_pod",
+        poll_interval=5,
+        logger=MagicMock()
+    )
+
+    opc_manager_mock.assert_not_called()
+    data_manager_mock.assert_called_once_with(
+        "localhost:9092", ingestor.logger)
+    resource_manager_mock.assert_called_once_with(
+        "localhost", 6379, 60, 60, "test_pod")
+
+    assert ingestor.poll_interval == 5
+    assert ingestor.managed_tags == {}
+    assert ingestor.opc_servers == {}
+    assert ingestor.opc_managers == {}
+    assert ingestor.data_manager == data_manager_mock.return_value
+    assert ingestor.resource_manager == resource_manager_mock.return_value
+
+
+@patch('ingestor.managers.ingestor_manager.OpcManager')
+def test_initialize_opc_from_config(opc_manager, ingestor_manager):
+    server_config = {
+        'name': 'server1',
+        'url': 'opc.tcp://localhost:4840',
+        'server_uri': 'http://opcua-server.simulator',
+        'cert_path': '/path/to/cert',
+        'private_key_path': '/path/to/private_key',
+        'server_cert_path': '/path/to/server_cert'
+    }
+
+    opc_manager.return_value = MagicMock()
+    result = ingestor_manager.initialize_opc_from_config(
+        server_config, ingestor_manager.data_manager, ingestor_manager.logger)
+
+    opc_manager.assert_called_once_with(
+        server_config['name'], server_config['url'], ingestor_manager.data_manager, ingestor_manager.logger,
+        server_config['server_uri'], server_config['cert_path'], server_config['private_key_path'],
+        server_config['server_cert_path']
+    )
+
+    assert result == opc_manager.return_value
+    result.connect.assert_called_once()
+
+
+@patch('ingestor.managers.ingestor_manager.OpcManager')
+def test_initialize_opc_from_config_exception(opc_manager, ingestor_manager):
+    server_config = {
+        'name': 'server1',
+        'url': 'opc.tcp://localhost:4840',
+        'server_uri': 'http://opcua-server.simulator',
+        'cert_path': '/path/to/cert',
+        'private_key_path': '/path/to/private_key',
+        'server_cert_path': '/path/to/server_cert'
+    }
+
+    ingestor_manager.logger.error = MagicMock()
+    opc_manager.side_effect = Exception("Initialization error")
+
+    result = ingestor_manager.initialize_opc_from_config(
+        server_config, ingestor_manager.data_manager, ingestor_manager.logger)
+
+    assert result is None
+    ingestor_manager.logger.error.assert_called_once_with(
+        "Failed to initialize OpcManager: Initialization error")
+
+
+@patch('ingestor.managers.ingestor_manager.OpcManager')
+def test_update_opc_servers(opc_manager, ingestor_manager):
+    manager1 = MagicMock(
+        config={"config": "config1"})
+    manager2 = MagicMock(
+        config={"config": "config2"})
+    manager3 = MagicMock(
+        config={"config": "config3"})
+
+    def mock_initialize_from_config(config, data_manager, logger):
+        if config == {"config": "config1"}:
+            return manager1
+        elif config == {"config": "config2"}:
+            return manager2
+        elif config == {"config": "config3"}:
+            return manager3
+        else:
+            return None
+
+    ingestor_manager.initialize_opc_from_config = MagicMock(
+        side_effect=mock_initialize_from_config
+    )
+
+    ingestor_manager.managed_tags = {
+        "slot1": {
+            "server1": {"config": "config1"},
+            "server2": {"config": "config2"},
+            'server5': {"config": "config5"}
+        },
+        "slot2": {
+            "server3": {"config": "config3"},
+            "server1": {"config": "config1"}
+        }
+    }
+
+    mock = MagicMock(
+        config={"config": "old_config2"})
+    ingestor_manager.opc_managers['server3'] = MagicMock(
+        config={"config": "config3"})
+    ingestor_manager.opc_managers['server2'] = mock
+    ingestor_manager.opc_managers['server4'] = MagicMock()
+
+    ingestor_manager.update_opc_servers()
+
+    assert len(ingestor_manager.opc_managers) == 3
+
+    ingestor_manager.initialize_opc_from_config.assert_any_call(
+        {"config": "config1"}, ingestor_manager.data_manager, ingestor_manager.logger)
+    ingestor_manager.initialize_opc_from_config.assert_any_call(
+        {"config": "config2"}, ingestor_manager.data_manager, ingestor_manager.logger)
+    ingestor_manager.initialize_opc_from_config.assert_any_call(
+        {"config": "config5"}, ingestor_manager.data_manager, ingestor_manager.logger)
+    assert ingestor_manager.initialize_opc_from_config.call_count == 3
+
+    assert ingestor_manager.opc_managers['server1'].config == {
+        "config": "config1"}
+    assert ingestor_manager.opc_managers['server2'].config == {
+        "config": "config2"}
+    assert ingestor_manager.opc_managers['server3'].config == {
+        "config": "config3"}
+    assert 'server4' not in ingestor_manager.opc_managers
+    assert 'server5' not in ingestor_manager.opc_managers
+
+    assert ingestor_manager.opc_managers['server2'] != mock
+
+
+def test_declare_active(ingestor_manager):
+    ingestor_manager.resource_manager.ingestor_heartbeat = MagicMock()
+    ingestor_manager.declare_active()
+    ingestor_manager.resource_manager.ingestor_heartbeat.assert_called_once()
+
+
+def test_get_active_ingestors(ingestor_manager):
+    ingestor_manager.resource_manager.get_all_ingestors = MagicMock()
+    ingestor_manager.get_active_ingestors()
+    ingestor_manager.resource_manager.get_all_ingestors.assert_called_once()
+
+
+def test_get_active_ingestors_empty(ingestor_manager):
+    ingestor_manager.resource_manager.get_all_ingestors = MagicMock(
+        return_value=None)
+    result = ingestor_manager.get_active_ingestors()
+    assert result == []
+    ingestor_manager.resource_manager.get_all_ingestors.assert_called_once()
+
+
+def test_get_number_of_leases_success(ingestor_manager):
+    ingestor_manager.resource_manager.get_all_leases = MagicMock(
+        return_value=["lease1", "lease2"])
+    result = ingestor_manager.get_number_of_leases()
+    assert result == 2
+    ingestor_manager.resource_manager.get_all_leases.assert_called_once()
+
+
+def test_get_number_of_leases_empty(ingestor_manager):
+    ingestor_manager.resource_manager.get_all_leases = MagicMock(
+        return_value=None)
+    result = ingestor_manager.get_number_of_leases()
+    assert result == 0
+    ingestor_manager.resource_manager.get_all_leases.assert_called_once()
+
+
+def test_get_number_of_slots_success(ingestor_manager):
+    ingestor_manager.resource_manager.get_all_slots = MagicMock(
+        return_value=["slot1", "slot2"])
+    result = ingestor_manager.get_number_of_slots()
+    assert result == 2
+    ingestor_manager.resource_manager.get_all_slots.assert_called_once()
+
+
+def test_get_number_of_slots_empty(ingestor_manager):
+    ingestor_manager.resource_manager.get_all_slots = MagicMock(
+        return_value=None)
+    result = ingestor_manager.get_number_of_slots()
+    assert result == 0
+    ingestor_manager.resource_manager.get_all_slots.assert_called_once()
+
+
+def test_get_slot_leases_1_success(ingestor_manager):
+    ingestor_manager.resource_manager.lease_tag = MagicMock(
+        return_value=True)
+    ingestor_manager.resource_manager.get_tag_slot = MagicMock(
+        return_value={"tags": ["tag1"]})
+
+    ingestor_manager.number_of_slots = 1
+    result = ingestor_manager.get_slot_leases()
+
+    assert result == {
+        "1": {"tags": ["tag1"]}
+    }
+
+
+def test_get_slot_leases_2_success(ingestor_manager):
+    ingestor_manager.resource_manager.lease_tag = MagicMock(
+        side_effect=[True, True])
+    ingestor_manager.resource_manager.get_tag_slot = MagicMock(
+        side_effect=[{"tags": ["tag1"]}, {"tags": ["tag2"]}])
+
+    ingestor_manager.number_of_slots = 2
+    result = ingestor_manager.get_slot_leases(max_slots=2)
+
+    assert result == {
+        "1": {"tags": ["tag1"]},
+        "2": {"tags": ["tag2"]}
+    }
+
+
+def test_get_slot_leases_2_1_none(ingestor_manager):
+    ingestor_manager.resource_manager.lease_tag = MagicMock(
+        side_effect=[True, True])
+    ingestor_manager.resource_manager.get_tag_slot = MagicMock(
+        side_effect=[None, {"tags": ["tag1"]}])
+
+    ingestor_manager.number_of_slots = 1
+    result = ingestor_manager.get_slot_leases(max_slots=1)
+
+    assert result == {}
+
+
+def test_get_slot_leases_1_failure(ingestor_manager):
+    ingestor_manager.resource_manager.lease_tag = MagicMock(
+        return_value=False)
+    ingestor_manager.resource_manager.get_tag_slot = MagicMock(
+        return_value={"tags": ["tag1"]})
+
+    result = ingestor_manager.get_slot_leases()
+    ingestor_manager.resource_manager.get_tag_slot.assert_not_called()
+
+    assert result == {}
+
+
+def test_unsubscribe_slot(ingestor_manager):
+    ingestor_manager.managed_tags = {
+        "slot1": {
+            "server1": {"tags": "config1"},
+            "server2": {"tags": "config2"}
+        },
+        "slot2": {
+            "server3": {"tags": "config3"},
+            "server1": {"tags": "config1"}
+        }
+    }
+    ingestor_manager.opc_managers = {
+        "server1": MagicMock(),
+        "server2": MagicMock(),
+        "server3": MagicMock()
+    }
+    ingestor_manager.unsubscribe_slot("slot1")
+
+    ingestor_manager.opc_managers["server1"].unsubscribe.assert_called_once_with(
+        "slot1")
+    ingestor_manager.opc_managers["server2"].unsubscribe.assert_called_once_with(
+        "slot1")
+    ingestor_manager.opc_managers["server3"].unsubscribe.assert_not_called()
+
+
+def test_update_slot_config(ingestor_manager):
+    ingestor_manager.managed_tags = {
+        "slot1": {"config": "old_config"},
+        "slot2": {"config": "new_config"},
+        "slot3": {"config": "old_config"}
+    }
+
+    ingestor_manager.resource_manager.get_tag_slot = MagicMock(
+        side_effect=[
+            {"config": "updated_config"},
+            {"config": "new_config"},
+            None
+        ]
+    )
+
+    ingestor_manager.update_opc_servers = MagicMock()
+    ingestor_manager.subscribe_to_tags = MagicMock()
+    ingestor_manager.unsubscribe_slot = MagicMock()
+
+    ingestor_manager.update_slot_config()
+
+    assert ingestor_manager.managed_tags["slot1"] == {
+        "config": "updated_config"}
+    assert ingestor_manager.managed_tags["slot2"] == {
+        "config": "new_config"}
+    assert "slot3" not in ingestor_manager.managed_tags
+
+    assert ingestor_manager.update_opc_servers.call_count == 2
+    ingestor_manager.subscribe_to_tags.assert_called_once_with(
+        {'slot1': {"config": "updated_config"}}
+    )
+    ingestor_manager.unsubscribe_slot.assert_any_call("slot3")
+    ingestor_manager.unsubscribe_slot.assert_any_call("slot1")
+    assert ingestor_manager.unsubscribe_slot.call_count == 2
+
+    ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot1")
+    ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot3")
+    ingestor_manager.resource_manager.renew_tag_lease.assert_any_call("slot2")
+    assert ingestor_manager.resource_manager.renew_tag_lease.call_count == 3
+
+
+def test_drop_slot_leases(ingestor_manager):
+    ingestor_manager.resource_manager.drop_tag_lease = MagicMock()
+    ingestor_manager.drop_slot_leases(["1", "2"])
+
+    ingestor_manager.resource_manager.drop_tag_lease.assert_any_call("1")
+    ingestor_manager.resource_manager.drop_tag_lease.assert_any_call("2")
+
+
+def test_manage_server_no_server(ingestor_manager):
+    ingestor_manager.opc_managers = {
+        "server1": MagicMock(),
+        "server2": MagicMock()
+    }
+    server_config = {
+        'tags': 'config1'
+    }
+
+    result = ingestor_manager.manage_server(
+        'slot1', 'server3', server_config, server_config)
+
+    assert result == 1
+    ingestor_manager.opc_managers["server1"].create_subscription.assert_not_called(
+    )
+    ingestor_manager.opc_managers["server1"].subscribe.assert_not_called()
+
+
+def test_manage_server_create_subscription_failure(ingestor_manager):
+    ingestor_manager.opc_managers = {
+        "server1": MagicMock(),
+        "server2": MagicMock()
+    }
+    ingestor_manager.subscriptions = {
+        "server1": MagicMock()
+    }
+    server_config = {
+        'tags': 'config1'
+    }
+
+    ingestor_manager.opc_managers["server1"].create_subscription.side_effect = Exception(
+        "Subscription error")
+
+    result = ingestor_manager.manage_server(
+        'slot1', 'server1', server_config, server_config)
+
+    assert result == 2
+    ingestor_manager.opc_managers["server1"].create_subscription.assert_called_once_with(
+        'slot1')
+    ingestor_manager.opc_managers["server1"].subscribe.assert_not_called()
+
+
+def test_manage_server(ingestor_manager):
+    ingestor_manager.opc_managers = {
+        "server1": MagicMock(),
+        "server2": MagicMock()
+    }
+    ingestor_manager.subscriptions = {
+        "server1": MagicMock()
+    }
+    server_config = {
+        'tags': 'config1'
+    }
+
+    result = ingestor_manager.manage_server(
+        'slot1', 'server1', server_config, server_config)
+
+    assert result == 0
+    ingestor_manager.opc_managers["server1"].create_subscription.assert_called_once_with(
+        'slot1')
+    ingestor_manager.opc_managers["server1"].subscribe.assert_called_once_with(
+        'slot1', 'config1', ingestor_manager.poll_interval)
+
+
+def test_manage_server_subscribe_failure(ingestor_manager):
+    ingestor_manager.opc_managers = {
+        "server1": MagicMock(),
+        "server2": MagicMock()
+    }
+    ingestor_manager.subscriptions = {
+        "server1": MagicMock()
+    }
+    server_config = {
+        'tags': 'config1'
+    }
+
+    ingestor_manager.opc_managers["server1"].subscribe.side_effect = Exception(
+        "Subscription error")
+
+    result = ingestor_manager.manage_server(
+        'slot1', 'server1', server_config, server_config)
+
+    assert result == 2
+    ingestor_manager.opc_managers["server1"].create_subscription.assert_called_once_with(
+        'slot1')
+    ingestor_manager.opc_managers["server1"].subscribe.assert_called_once_with(
+        'slot1', 'config1', ingestor_manager.poll_interval)
+    ingestor_manager.opc_managers["server1"].unsubscribe.assert_called_once_with(
+        'slot1')
+    ingestor_manager.logger.error.assert_any_call(
+        "Failed to subscribe to tags from slot1:server1\n{'tags': 'config1'}: Subscription error"
+    )
+    ingestor_manager.logger.warning.assert_any_call(
+        "Removing subscription from server server1 for slot slot1"
+    )
+
+
+def test_subscribe_to_tags(ingestor_manager):
+    ingestor_manager.manage_server = MagicMock(
+        side_effect=[0, 1, 2])
+    ingestor_manager.managed_tags = {
+        "slot1": MagicMock(),
+        "slot2": MagicMock()
+    }
+
+    ingestor_manager.opc_managers = {
+        "server1": MagicMock(),
+        "server2": MagicMock()
+    }
+    ingestor_manager.subscriptions = {
+        "server1": MagicMock()
+    }
+    tags = {
+        'slot1': {
+            "server1": {"tags": "config1"},
+            "server2": {"tags": "config2"},
+            'server3': {"tags": "config3"},
+        }
+    }
+
+    ingestor_manager.subscribe_to_tags(tags)
+
+    ingestor_manager.manage_server.assert_any_call(
+        'slot1', 'server1', {"tags": "config1"}, tags)
+    ingestor_manager.manage_server.assert_any_call(
+        'slot1', 'server2', {"tags": "config2"}, tags)
+    ingestor_manager.manage_server.assert_any_call(
+        'slot1', 'server3', {"tags": "config3"}, tags)
+
+    assert ingestor_manager.manage_server.call_count == 3
+
+    ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with(
+        'server3', None)

tests/unit/managers/test_opc_manager.py

@@ -0,0 +1,321 @@
+import json
+from datetime import datetime
+from unittest.mock import MagicMock, patch
+from pytest import fixture
+
+from asyncua.crypto.security_policies import SecurityPolicyBasic256
+from ingestor.managers.opc_manager import OpcManager
+from sientia_do.notifications.models import NotificationLevel
+
+tags = {
+    'ns=3;i=1001': {
+        'aggregation_function': 'LTS',
+        'frequency': 1000,
+        'max_value': 100,
+        'min_value': 0,
+        'tag_name': 'Counter'
+    },
+    'ns=3;i=1003': {
+        'aggregation_function': 'AVG',
+        'frequency': 1000,
+        'max_value': 100,
+        'min_value': 0,
+        'tag_name': 'Random'
+    },
+    'ns=3;i=1004': {
+        'aggregation_function': 'MDN',
+        'frequency': 1000,
+        'max_value': 100,
+        'min_value': 0,
+        'tag_name': 'Sawtooth'
+    }
+}
+
+
+@fixture
+def raw_opc_manager():
+    return OpcManager(
+        'TestConnector', 'opc.tcp://localhost:4840', MagicMock(),
+        MagicMock(), 'opc.tcp://localhost:4840'
+    )
+
+
+@fixture
+def opc_manager(raw_opc_manager):
+    raw_opc_manager.client = MagicMock()
+    raw_opc_manager.cert_path = 'cert.pem'
+    raw_opc_manager.private_key_path = 'private_key.pem'
+    raw_opc_manager.server_cert_path = 'server_cert.pem'
+
+    return raw_opc_manager
+
+
+@fixture
+def opc_manager_subscribed(opc_manager):
+    opc_manager.subscriptions['sub1'] = MagicMock()
+
+    return opc_manager
+
+
+def test___str__(opc_manager):
+    assert str(opc_manager) == 'OpcManager(name=TestConnector, url=opc.tcp://localhost:4840, server_uri=opc.tcp://localhost:4840)\nnodes={}, subscriptions={}'
+
+
+def test_set_security_success(opc_manager):
+    opc_manager.set_security()
+
+    assert opc_manager.client.application_uri == opc_manager.server_uri
+
+    opc_manager.client.set_security.assert_called_once_with(
+        SecurityPolicyBasic256,
+        certificate=opc_manager.cert_path,
+        private_key=opc_manager.private_key_path,
+        server_certificate=opc_manager.server_cert_path
+    )
+
+    assert opc_manager.client.secure_channel_timeout == 10000000
+    assert opc_manager.client.session_timeout == 10000000
+
+
+def test_set_security_no_cert(opc_manager):
+    opc_manager.cert_path = None
+    opc_manager.private_key_path = None
+
+    try:
+        opc_manager.set_security()
+    except ValueError as e:
+        assert str(
+            e) == "Certificate and private key paths must be provided for secure connection."
+    else:
+        assert False, "ValueError not raised"
+
+    assert opc_manager.client.set_security.call_count == 0
+
+
+@patch('ingestor.managers.opc_manager.Client')
+def test_connect_no_security(client, raw_opc_manager):
+    raw_opc_manager.set_security = MagicMock()
+
+    raw_opc_manager.connect()
+
+    client.assert_called_once_with(raw_opc_manager.url)
+    raw_opc_manager.client.connect.assert_called_once()
+    raw_opc_manager.set_security.assert_not_called()
+
+
+@patch('ingestor.managers.opc_manager.Client')
+def test_connect_with_security(client, raw_opc_manager):
+    raw_opc_manager.cert_path = 'cert.pem'
+    raw_opc_manager.private_key_path = 'private_key.pem'
+    raw_opc_manager.server_cert_path = 'server_cert.pem'
+    raw_opc_manager.set_security = MagicMock()
+
+    raw_opc_manager.connect()
+
+    client.assert_called_once_with(raw_opc_manager.url)
+    raw_opc_manager.client.connect.assert_called_once()
+    raw_opc_manager.set_security.assert_called_once()
+
+
+def test_create_subscription_no_client(raw_opc_manager):
+    try:
+        raw_opc_manager.create_subscription('sub1')
+    except ValueError as e:
+        assert str(e) == "Client not connected. Call connect first."
+    else:
+        assert False, "ValueError not raised"
+
+
+def test_create_subscription_success_has_period(opc_manager):
+    opc_manager.create_subscription('sub1', 1000)
+
+    opc_manager.client.create_subscription.assert_called_once_with(
+        1000, opc_manager)
+    assert opc_manager.subscriptions['sub1'] is not None
+
+
+def test_create_subscription_success_no_period(opc_manager):
+    opc_manager.create_subscription('sub1', None)
+
+    opc_manager.client.create_subscription.assert_called_once_with(
+        500, opc_manager)
+    assert opc_manager.subscriptions['sub1'] is not None
+
+
+def test_subscribe_no_subscription(opc_manager):
+    try:
+        opc_manager.subscribe('sub1', tags, 1000)
+    except ValueError as e:
+        assert str(
+            e) == "Subscription not created. Call create_subscription first."
+    else:
+        assert False, "ValueError not raised"
+
+
+def test_subscribe_success(opc_manager_subscribed):
+    opc_manager_subscribed.nodes = {
+        'ns=3;i=1001': 'data'
+    }
+
+    opc_manager_subscribed.subscribe('sub1', tags, 1000)
+
+    assert opc_manager_subscribed.nodes == tags
+    assert opc_manager_subscribed.addr_nodes == [
+        opc_manager_subscribed.client.get_node(n) for n in tags if n != 'ns=3;i=1001']
+
+
+def test_unsubscribe_no_subscription(opc_manager):
+    opc_manager.unsubscribe('sub1')
+
+    opc_manager.logger.warning.assert_called_once_with(
+        "Subscription 'sub1' not found. Cannot unsubscribe.")
+    assert opc_manager.subscriptions.get('sub1') is None
+
+
+def test_unsubscribe_success(opc_manager_subscribed):
+    opc_manager_subscribed.unsubscribe('sub1')
+
+    opc_manager_subscribed.subscriptions.get('sub1') is None
+
+
+def test_disconnect_success(opc_manager_subscribed):
+    opc_manager_subscribed.client = MagicMock()
+
+    opc_manager_subscribed.disconnect()
+
+    opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once()
+    assert opc_manager_subscribed.client is None
+
+
+def test_disconnect_error(opc_manager_subscribed):
+    opc_manager_subscribed.client = MagicMock()
+    opc_manager_subscribed.subscriptions['sub1'] = MagicMock(
+        delete=MagicMock(side_effect=Exception("Test error"))
+    )
+
+    opc_manager_subscribed.disconnect()
+
+    opc_manager_subscribed.subscriptions['sub1'].delete.assert_called_once()
+    opc_manager_subscribed.client.disconnect.assert_not_called()
+    opc_manager_subscribed.logger.error.assert_called_once_with(
+        "Failed to clean up subscription: Test error")
+
+
+def test_datachange_notification(opc_manager_subscribed):
+    data = MagicMock(
+        monitored_item=MagicMock(
+            Value=MagicMock(
+                Value=MagicMock(Value=42),
+                SourceTimestamp=datetime.strptime(
+                    '2021-01-01T00:00:00', '%Y-%m-%dT%H:%M:%S')
+            )))
+    opc_manager_subscribed.nodes = {
+        'ns=3;i=1001': {
+            'tag_name': 'Counter',
+            'cycle_rule': {
+                'cycle_increment': 1.0,
+                'cycle_count': 2
+            },
+            'topics': ['topic1', 'topic2']
+        }
+    }
+
+    opc_manager_subscribed.datachange_notification(
+        'ns=3;i=1001', None, data)
+
+    opc_manager_subscribed.data_manager.publish.assert_any_call(
+        'topic1', {
+            'tag': 'ns=3;i=1001',
+            'name': 'Counter',
+            'timestamp': '2021-01-01 00:00:00',
+            'value': 42
+        })
+    opc_manager_subscribed.data_manager.publish.assert_any_call(
+        'topic2', {
+            'tag': 'ns=3;i=1001',
+            'name': 'Counter',
+            'timestamp': '2021-01-01 00:00:00',
+            'value': 42
+        })
+    assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0
+
+
+def test_check_cycles(opc_manager_subscribed):
+    handler = MagicMock()
+    opc_manager_subscribed.nodes = tags
+    opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule'] = {
+        'cycle_increment': 1.0,
+        'cycle_count': 0
+    }
+    opc_manager_subscribed.check_cycles(
+        {}, 'ns=3;i=1001', tags['ns=3;i=1001'], handler)
+
+    assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 1
+
+    opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] = 0
+    opc_manager_subscribed.check_cycles({'ns=3;i=1001': {}},
+                                        'ns=3;i=1001', tags['ns=3;i=1001'], handler)
+
+    assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 0
+
+    opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] = 4
+    opc_manager_subscribed.check_cycles(
+        {}, 'ns=3;i=1001', tags['ns=3;i=1001'], handler)
+
+    assert opc_manager_subscribed.nodes['ns=3;i=1001']['cycle_rule']['cycle_count'] == 5
+    handler.assert_called_once_with('5.0 cycles without receive from ns=3;i=1001:Counter',
+                                    'TAG_ns=3;i=1001:Counter_LISTENNING_STOPPED', NotificationLevel.WARNING)
+
+
+def test_check_opc_listenning_5_cycles(opc_manager_subscribed):
+
+    handler = MagicMock()
+    opc_manager_subscribed.nodes = tags
+    opc_manager_subscribed.non_receive_count = 4
+
+    opc_manager_subscribed.check_opc_listenning(handler)
+
+    handler.assert_called_once_with(
+        f'5 cycles without receive from OPC TestConnector. Tags: {json.dumps(tags)}',
+        'OPC_LISTENNING_STOPPED__TestConnector',
+        NotificationLevel.ERROR
+    )
+
+
+def test_check_opc_listenning_no_cycles(opc_manager_subscribed):
+    handler = MagicMock()
+    opc_manager_subscribed.non_receive_count = 0
+
+    opc_manager_subscribed.check_opc_listenning(handler)
+
+    handler.assert_not_called()
+
+
+def test_check_opc_listenning_no_handler(opc_manager_subscribed):
+    opc_manager_subscribed.non_receive_count = 5
+    opc_manager_subscribed.check_opc_listenning(None)
+
+    assert opc_manager_subscribed.non_receive_count == 6
+
+
+def test_check_opc_listenning_15_cycles(opc_manager_subscribed):
+    handler = MagicMock()
+    opc_manager_subscribed.init_collector = MagicMock()
+    opc_manager_subscribed.non_receive_count = 14
+    opc_manager_subscribed.collect_period = 1000
+    opc_manager_subscribed.period = 500
+    opc_manager_subscribed.nodes = tags
+
+    opc_manager_subscribed.check_opc_listenning(handler)
+    handler.assert_any_call(
+        f'15 cycles without receive from OPC TestConnector. Tags: {json.dumps(tags)}',
+        'OPC_LISTENNING_STOPPED__TestConnector',
+        NotificationLevel.ERROR
+    )
+    handler.assert_any_call(
+        'Retrying to connect to server TestConnector',
+        'OPC_CONNECTION_RETRY__TestConnector',
+        NotificationLevel.ERROR
+    )
+    opc_manager_subscribed.init_collector.assert_called_once_with(
+        tags, 1000, 500)

tests/unit/managers/test_resource_manager.py

@@ -0,0 +1,109 @@
+from unittest.mock import MagicMock, patch
+from pytest import fixture
+from ingestor.managers.resource_manager import ResourceManager
+
+
+@fixture
+@patch('ingestor.managers.resource_manager.Redis')
+def resource_manager(redis):
+
+    return ResourceManager(
+        'localhost', 6379, 10, 10, 'pod_id'
+    )
+
+
+def test_get_success(resource_manager):
+    resource_manager.redis.get.return_value = '{"key": "value"}'
+    result = resource_manager.get('key')
+    assert result == {"key": "value"}
+    resource_manager.redis.get.assert_called_once_with('key')
+
+
+def test_get_failure(resource_manager):
+    resource_manager.redis.get.return_value = None
+    result = resource_manager.get('key')
+    assert result is None
+    resource_manager.redis.get.assert_called_once_with('key')
+
+
+def test_get_tag_slot(resource_manager):
+    resource_manager.get = MagicMock(return_value={"tag": "slot"})
+    result = resource_manager.get_tag_slot('id')
+    assert result == {"tag": "slot"}
+    resource_manager.get.assert_called_once_with('slot:opc_tags:id')
+
+
+def test_ingestor_heartbeat(resource_manager):
+    resource_manager.redis.set.return_value = True
+    resource_manager.ingestor_heartbeat()
+    resource_manager.redis.set.assert_called_once_with(
+        'heartbeat:ingestor:pod_id', 1, ex=10
+    )
+
+
+def test_lease_tag(resource_manager):
+    resource_manager.redis.set.return_value = True
+    output = resource_manager.lease_tag('tag_id')
+    assert output is True
+    resource_manager.redis.set.assert_called_once_with(
+        'lease:opc_tags:tag_id', 'pod_id', nx=True, ex=10
+    )
+
+
+def test_renew_tag_lease_success(resource_manager):
+    resource_manager.redis.get.return_value = 'pod_id'
+    resource_manager.redis.expire.return_value = True
+    result = resource_manager.renew_tag_lease('tag_id')
+    assert result is True
+    resource_manager.redis.get.assert_called_once_with(
+        'lease:opc_tags:tag_id'
+    )
+    resource_manager.redis.expire.assert_called_once_with(
+        'lease:opc_tags:tag_id', 10
+    )
+
+
+def test_renew_tag_lease_failure(resource_manager):
+    resource_manager.redis.get.return_value = 'other_pod_id'
+    resource_manager.redis.expire.return_value = False
+    result = resource_manager.renew_tag_lease('tag_id')
+    assert result is False
+    resource_manager.redis.get.assert_called_once_with(
+        'lease:opc_tags:tag_id'
+    )
+    resource_manager.redis.expire.assert_not_called()
+
+
+def test_drop_tag_lease(resource_manager):
+    resource_manager.redis.delete.return_value = True
+    resource_manager.drop_tag_lease('tag_id')
+    resource_manager.redis.delete.assert_called_once_with(
+        'lease:opc_tags:tag_id'
+    )
+
+
+def test_get_all_ingestors(resource_manager):
+    resource_manager.redis.keys.return_value = ['ingestor1', 'ingestor2']
+    result = resource_manager.get_all_ingestors()
+    assert result == ['ingestor1', 'ingestor2']
+    resource_manager.redis.keys.assert_called_once_with(
+        'heartbeat:ingestor:*'
+    )
+
+
+def test_get_all_slots(resource_manager):
+    resource_manager.redis.keys.return_value = ['slot1', 'slot2']
+    result = resource_manager.get_all_slots()
+    assert result == ['slot1', 'slot2']
+    resource_manager.redis.keys.assert_called_once_with(
+        'slot:opc_tags:*'
+    )
+
+
+def test_get_all_leases(resource_manager):
+    resource_manager.redis.keys.return_value = ['lease1', 'lease2']
+    result = resource_manager.get_all_leases()
+    assert result == ['lease1', 'lease2']
+    resource_manager.redis.keys.assert_called_once_with(
+        'lease:opc_tags:*'
+    )

tests/unit/test_ingestor.py

@@ -0,0 +1,246 @@
+from unittest.mock import MagicMock, patch
+
+from pytest import fixture
+from ingestor.ingestor import Ingestor
+
+
+@patch("ingestor.ingestor.getenv")
+@patch("ingestor.ingestor.Ingestor.init_logger")
+def test___init__(init_logger, getenv):
+    getenv.side_effect = [
+        "localhost:9092,localhost:35",  # KAFKA_SERVERS
+        "localhost1",       # REDIS_HOST
+        '63790',             # REDIS_PORT
+        '100',               # LEASE_TTL
+        '200',               # HEARTBEAT_TTL
+        "localhost1",      # HOSTNAME
+        '50'                 # POLL_INTERVAL
+    ]
+    ingestor = Ingestor()
+
+    getenv.assert_any_call("KAFKA_SERVERS", "localhost:9092")
+    getenv.assert_any_call("REDIS_HOST", "localhost")
+    getenv.assert_any_call("REDIS_PORT", 6379)
+    getenv.assert_any_call("LEASE_TTL", 10)
+    getenv.assert_any_call("HEARTBEAT_TTL", 20)
+    getenv.assert_any_call("HOSTNAME", "localhost")
+    getenv.assert_any_call("POLL_INTERVAL", 5)
+
+    assert ingestor.kafka_servers == ["localhost:9092", "localhost:35"]
+    assert ingestor.redis_host == "localhost1"
+    assert ingestor.redis_port == 63790
+    assert ingestor.lease_ttl == 100
+    assert ingestor.heartbeat_ttl == 200
+    assert ingestor.pod_id == "localhost1"
+    assert ingestor.poll_interval == 50
+
+    init_logger.assert_called_once()
+
+
+@fixture
+@patch("ingestor.ingestor.getenv")
+@patch("ingestor.ingestor.Ingestor.init_logger")
+def ingestor(init_logger, getenv):
+    ing = Ingestor()
+    ing.logger = MagicMock()
+
+    return ing
+
+
+@fixture
+def ingestor_manager_started(ingestor):
+    ingestor.ingestor_manager = MagicMock()
+    return ingestor
+
+
+@patch("ingestor.ingestor.getLogger")
+@patch("ingestor.ingestor.StreamHandler")
+@patch("ingestor.ingestor.Formatter")
+def test_init_logger(formatter, stream_handler, get_logger, ingestor):
+    ingestor.logger = None
+    ingestor.init_logger()
+
+    get_logger.assert_called_once_with('ingestor.ingestor')
+    stream_handler.assert_called_once()
+    formatter.assert_called_once_with(
+        '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    ingestor.logger.setLevel.assert_called_once_with("INFO")
+    ingestor.logger.addHandler.assert_called_once_with(
+        stream_handler.return_value)
+    stream_handler.return_value.setFormatter.assert_called_once_with(
+        formatter.return_value)
+
+
+def test_handle_acquired_tags_not_acquired(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags([])
+
+    ingestor_manager_started.logger.warning.assert_called_once_with(
+        "No slots available")
+    ingestor_manager_started.ingestor_manager.update_opc_servers.assert_not_called()
+    ingestor_manager_started.ingestor_manager.subscribe_to_tags.assert_not_called()
+
+
+def test_handle_acquired_tags_success(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags(["tag1", "tag2"])
+
+    ingestor_manager_started.logger.warning.assert_not_called()
+    ingestor_manager_started.ingestor_manager.update_opc_servers.assert_called_once()
+    ingestor_manager_started.ingestor_manager.subscribe_to_tags.assert_called_once_with(
+        ["tag1", "tag2"])
+
+
+@patch("ingestor.ingestor.IngestorManager")
+def test_prepare_ingestor(ingestor_manager_mock, ingestor):
+    ingestor_manager = ingestor_manager_mock.return_value
+    ingestor_manager.get_slot_leases.return_value = True
+
+    ingestor.prepare_ingestor()
+
+    ingestor_manager_mock.assert_called_once_with(
+        ingestor.kafka_servers,
+        ingestor.redis_host,
+        ingestor.redis_port,
+        ingestor.lease_ttl,
+        ingestor.heartbeat_ttl,
+        ingestor.pod_id,
+        ingestor.poll_interval,
+        ingestor.logger
+    )
+    ingestor_manager.declare_active.assert_called_once()
+    ingestor_manager.get_slot_leases.assert_called_once()
+
+    ingestor.handle_acquired_tags(
+        ingestor_manager.get_slot_leases.return_value)
+
+
+def test_manage_slots_has_slots(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags = MagicMock()
+    ingestor_manager_started.ingestor_manager.managed_tags = True
+
+    ingestor_manager_started.manage_no_slots(5)
+
+    ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called()
+    ingestor_manager_started.ingestor_manager.handle_acquired_tags.assert_not_called()
+
+
+def test_manage_no_slots_has_slots_none_available(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags = MagicMock()
+    ingestor_manager_started.ingestor_manager.managed_tags = True
+
+    ingestor_manager_started.manage_no_slots(0)
+
+    ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called()
+    ingestor_manager_started.ingestor_manager.handle_acquired_tags.assert_not_called()
+
+
+def test_manage_slots_none_available(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags = MagicMock()
+    ingestor_manager_started.ingestor_manager.managed_tags = False
+
+    ingestor_manager_started.manage_no_slots(0)
+
+    ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called()
+    ingestor_manager_started.ingestor_manager.handle_acquired_tags.assert_not_called()
+
+
+def test_manage_slots_none_available_none_available(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags = MagicMock()
+    ingestor_manager_started.ingestor_manager.managed_tags = False
+
+    ingestor_manager_started.manage_no_slots(2)
+
+    ingestor_manager_started.ingestor_manager.get_slot_leases.assert_called_once_with(
+        1)
+    ingestor_manager_started.handle_acquired_tags.assert_called_once_with(
+        ingestor_manager_started.ingestor_manager.get_slot_leases.return_value)
+
+
+def test_manage_leases_no_available_slots_no_extra_slots(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags = MagicMock()
+
+    ingestor_manager_started.manage_leases(0, 0, 0)
+
+    ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called()
+    ingestor_manager_started.ingestor_manager.handle_acquired_tags.assert_not_called()
+    ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_not_called()
+
+
+def test_manage_leases_available_slots_innactive_ingestors(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags = MagicMock()
+
+    ingestor_manager_started.manage_leases(2, 2, 5)
+
+    ingestor_manager_started.ingestor_manager.get_slot_leases.assert_called_once_with(
+        2)
+    ingestor_manager_started.handle_acquired_tags.assert_called_once_with(
+        ingestor_manager_started.ingestor_manager.get_slot_leases.return_value)
+
+    ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_not_called()
+
+
+def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_started):
+    ingestor_manager_started.handle_acquired_tags = MagicMock()
+    ingestor_manager_started.ingestor_manager.managed_tags = {
+        "tag1": "server1",
+        "tag2": "server2",
+        "tag3": "server3"
+    }
+
+    ingestor_manager_started.manage_leases(0, 0, 2)
+
+    ingestor_manager_started.ingestor_manager.get_slot_leases.assert_not_called()
+    ingestor_manager_started.handle_acquired_tags.assert_not_called()
+    ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_called_once_with(
+        ["tag2", "tag3"])
+
+
+def test_loop(ingestor_manager_started):
+    ingestor_manager_started.manage_no_slots = MagicMock()
+    ingestor_manager_started.manage_leases = MagicMock()
+    ingestor_manager_started.ingestor_manager.managed_tags = {
+        "slot1": "server1",
+        "slot2": "server2",
+        "slot3": "server3"
+    }
+    ingestor_manager_started.ingestor_manager.get_active_ingestors = MagicMock(
+        return_value=["ingestor1", "ingestor2"])
+    ingestor_manager_started.ingestor_manager.get_number_of_slots = MagicMock(
+        return_value=5)
+
+    ingestor_manager_started.loop()
+
+    ingestor_manager_started.ingestor_manager.declare_active.assert_called_once()
+    ingestor_manager_started.ingestor_manager.get_active_ingestors.assert_called_once()
+    ingestor_manager_started.ingestor_manager.get_number_of_slots.assert_called_once()
+
+    ingestor_manager_started.manage_no_slots.assert_called_once_with(
+        ingestor_manager_started.ingestor_manager.get_number_of_slots.return_value)
+    # Explanation: 5 - 2 = 3, 3 - 1 = 2
+    ingestor_manager_started.manage_leases.assert_called_once_with(
+        3, 2)
+    ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once()
+
+
+def test_loop_no_managed(ingestor_manager_started):
+    ingestor_manager_started.manage_no_slots = MagicMock()
+    ingestor_manager_started.manage_leases = MagicMock()
+    ingestor_manager_started.ingestor_manager.managed_tags = {}
+    ingestor_manager_started.ingestor_manager.get_active_ingestors = MagicMock(
+        return_value=["ingestor1", "ingestor2"])
+    ingestor_manager_started.ingestor_manager.get_number_of_slots = MagicMock(
+        return_value=5)
+
+    ingestor_manager_started.loop()
+
+    ingestor_manager_started.ingestor_manager.declare_active.assert_called_once()
+    ingestor_manager_started.ingestor_manager.get_active_ingestors.assert_called_once()
+    ingestor_manager_started.ingestor_manager.get_number_of_slots.assert_called_once()
+
+    ingestor_manager_started.manage_no_slots.assert_called_once_with(
+        ingestor_manager_started.ingestor_manager.get_number_of_slots.return_value)
+    # Explanation: 5 - 2 = 3, 3 - 1 = 2
+    ingestor_manager_started.manage_leases.assert_called_once_with(
+        3, -1)
+    ingestor_manager_started.ingestor_manager.update_slot_config.assert_called_once()
+    ingestor_manager_started.logger.info.assert_any_call(
+        "No slots acquired in this loop")