From 8f6ba4ddcfb2b3a173d7c040eac6e0bb8ad9c20f Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 29 Aug 2025 11:20:28 -0300 Subject: [PATCH] SIENTIAPDE-1084 Remove deprecated files and enhance documentation - Deleted `coverage.sh`, `docker-compose.yaml`, `Dockerfile`, and simulator-related files to streamline the project structure. - Updated `README.md` to provide a comprehensive overview of the OPC Ingestor, including features, architecture, installation, usage, and troubleshooting. - Enhanced docstrings across various classes and methods in the `ingestor` module for better clarity and maintainability. - Improved Prometheus metrics documentation in `metrics.py` to ensure proper monitoring and observability of the system. --- .env.example | 21 ++ Dockerfile | 30 -- README.md | 425 ++++++++++++++++++++++---- coverage.sh | 1 - docker-compose.yaml | 102 ------- ingestor/app.py | 57 +++- ingestor/ingestor.py | 208 +++++++++---- ingestor/managers/data_manager.py | 61 +++- ingestor/managers/ingestor_manager.py | 56 +++- ingestor/managers/opc_manager.py | 88 +++++- ingestor/managers/resource_manager.py | 238 +++++++++++++-- ingestor/metrics.py | 18 +- simulator/Dockerfile | 30 -- simulator/redis-feeder.py | 60 ---- 14 files changed, 1004 insertions(+), 391 deletions(-) create mode 100644 .env.example delete mode 100644 Dockerfile delete mode 100755 coverage.sh delete mode 100644 docker-compose.yaml delete mode 100644 simulator/Dockerfile delete mode 100644 simulator/redis-feeder.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..d6fcde0 --- /dev/null +++ b/.env.example @@ -0,0 +1,21 @@ +# Kafka +KAFKA_SERVERS="localhost:9092" +EXPORT_TO_KAFKA="false" + +# Redis +REDIS_HOST="localhost" +REDIS_PORT="6379" +REDIS_USERNAME="redis_user" +REDIS_PASSWORD="redis_password" + +# Pod +HOSTNAME="localhost" +POLL_INTERVAL="5" +LOG_LEVEL="DEBUG" + +# Mongo DB +MONGODB_URL="localhost:27018" +MONGODB_USERNAME="mongo_user" +MONGODB_PASSWORD="mongo_password" +MONGODB_DATABASE="sientia" + diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index 2e48159..0000000 --- a/Dockerfile +++ /dev/null @@ -1,30 +0,0 @@ -# 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"] \ No newline at end of file diff --git a/README.md b/README.md index 22facb3..2c53512 100644 --- a/README.md +++ b/README.md @@ -1,81 +1,386 @@ -# sientia-dataops-opc-ingestor -OPC gateway to manage Scouter pipelines +# Sientia DataOps OPC Ingestor -## 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 +A high-performance, scalable OPC UA data ingestion system designed for industrial data collection and real-time streaming. The OPC Ingestor provides enterprise-grade data acquisition from OPC UA servers with automatic load balancing, fault tolerance, and comprehensive monitoring. + +## Features + +### Core Functionality +- **OPC UA Integration**: Native support for OPC UA servers with secure and unsecured connections +- **Automatic Load Balancing**: Slot-based architecture for horizontal scaling across multiple instances +- **Real-time Data Streaming**: Kafka integration for high-throughput data distribution +- **Persistent Storage**: MongoDB integration for historical data retention and analysis +- **Health Monitoring**: Comprehensive Prometheus metrics and health checks +- **Fault Tolerance**: Automatic failover, reconnection, and error recovery + +### Advanced Capabilities +- **Certificate-based Security**: Support for X.509 certificates and private keys +- **Dynamic Tag Management**: Runtime configuration updates without service interruption +- **Resource Coordination**: Redis-based slot leasing and instance coordination +- **Notification System**: Integrated alerting and notification management +- **Performance Optimization**: Configurable polling intervals and data collection frequencies + +## ️ Architecture + +The OPC Ingestor uses a modular, manager-based architecture: -### 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 +┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐ +│ Main App │ │ IngestorManager │ │ OPC Manager │ +│ │◄──►│ │◄──►│ │ +│ - Signal Hand. │ │ - Slot Mgmt │ │ - Connections │ +│ - Metrics │ │ - Load Balancing │ │ - Subscriptions │ +│ - Lifecycle │ │ - Coordination │ │ - Data Handler │ +└─────────────────┘ └──────────────────┘ └─────────────────┘ + │ + ▼ + ┌──────────────────┐ ┌─────────────────┐ + │ ResourceManager │ │ DataManager │ + │ │ │ │ + │ - Redis Coord. │ │ - Kafka Export │ + │ - Slot Leasing │ │ - MongoDB Store │ + │ - Heartbeats │ │ - Data Pipeline │ + └──────────────────┘ └─────────────────┘ ``` -### Run docker compose -``` -docker compose down -v -docker compose build --ssh default=$HOME/.ssh/id_ed25519_docker -docker compose up -d +### Key Components + +- **Ingestor**: Main application orchestrator managing the overall lifecycle +- **IngestorManager**: Coordinates slot allocation, OPC server management, and load balancing +- **OPC Manager**: Handles individual OPC UA server connections and tag subscriptions +- **Data Manager**: Manages data persistence (MongoDB) and streaming (Kafka) +- **Resource Manager**: Coordinates resource allocation and instance coordination via Redis + +## 📋 Prerequisites + +- Python 3.11+ +- Redis server +- MongoDB server +- Kafka cluster (optional, for data streaming) +- OPC UA servers for data collection + +**Note**: External dependencies (Redis, MongoDB, Kafka) must be available either through: +- Port forwarding from a Kubernetes cluster +- External Docker Compose setup +- Cloud-managed services +- Local installations + +## ️ Installation + +### Local Development Setup + +1. **Clone the repository** + ```bash + git clone + cd sientia-dataops-opc-ingestor + ``` + +2. **Create virtual environment** + ```bash + python3.11 -m venv venv + source ./venv/bin/activate + ``` + +3. **Install dependencies** + ```bash + pip install -r requirements.txt + ``` + +4. **Create environment configuration file** + ```bash + cp .env.example .env + # Edit .env with your connection details + ``` + +5. **Configure external dependencies** + + You'll need to set up port forwarding or connections to external services. For example: + + ```bash + # Port forwarding from Kubernetes cluster + kubectl port-forward svc/redis-master 6379:6379 + kubectl port-forward svc/mongodb 27017:27017 + kubectl port-forward svc/kafka 9092:9092 + + # Or connect to external Docker Compose + # Ensure services are accessible on localhost with appropriate ports + ``` + +## Usage + +### Running the Ingestor + +Use the provided script to run the application locally: + +```bash +# Make script executable (first time only) +chmod +x run_local.sh + +# Run the application +./run_local.sh ``` -### Run Ingestor -The ingestor can be run as a docker container or as a python script. +The script will: +- Activate the virtual environment +- Load environment variables from `.env` +- Start the ingestor application -As a docker container: -Uncomment the ingestor service in docker-compose.yaml -``` -docker compose up -d ingestor +### Running Tests and Coverage + +Use the provided script to run tests with coverage: + +```bash +# Make script executable (first time only) +chmod +x run_coverage.sh + +# Run tests with coverage +./run_coverage.sh ``` -As a python script: -``` -python -m ingestor.app -``` +The script will: +- Activate the virtual environment +- Run pytest with coverage reporting +- Generate HTML coverage report +- Open the coverage report in your browser -### Populate redis server -Create venv with python3.11 -``` -python3.11 -m venv venv +### Manual Test Execution + +You can also run tests manually: + +```bash +# Activate virtual environment 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 -``` +# Run all tests pytest -``` -### Get current coverage -``` + +# Run with coverage +pytest --cov=ingestor --cov-report=html + + +### Populating Redis with OPC Configuration + +1. **Activate virtual environment** + ```bash + source ./venv/bin/activate + ``` + +2. **Run the Redis feeder** + ```bash + python simulator/redis-feeder.py + ``` + +The feeder creates sample OPC tag configurations in Redis that the ingestor can discover and manage. + +## 🧪 Testing + +### Unit Tests +```bash +# Install pytest +pip install pytest + +# Run tests +pytest + +# Run with coverage pip install pytest-cov pytest --cov=ingestor -``` -### Generate complete report -``` + +# Generate HTML coverage report pytest --cov=ingestor --cov-report=html ``` -#PR shortcut +### Functional Tests +```bash +# Run functional tests +pytest tests/functional/ ``` -git log origin/main..HEAD --no-merges > git_log + +## 📊 Monitoring and Metrics + +The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring: + +### Application Metrics +- `app_up`: Application health status +- `app_main_loop_total`: Main loop execution count +- `app_main_loop_duration_seconds`: Loop execution time +- `app_errors_total`: Error count + +### OPC Server Metrics +- `opc_connection_status`: Server connection status +- `opc_tags_subscribed_current`: Number of subscribed tags +- `opc_cycles_without_data`: Data reception health +- `opc_subscriptions_created_total`: Subscription count + +### Resource Management Metrics +- `ingestor_slots_managed_current`: Slots managed by this instance +- `ingestor_active_total`: Total active ingestor instances +- `redis_connection_status`: Redis connection health +- `kafka_connection_status`: Kafka connection health + +### Data Processing Metrics +- `ingestor_tag_written_count`: Data write operations +- `kafka_messages_sent_total`: Kafka message count +- `redis_operations_total`: Redis operation count + +## ⚙️ Configuration + +### Environment Variables + +| Variable | Description | Default | Required | +|----------|-------------|---------|----------| +| `KAFKA_SERVERS` | Comma-separated Kafka server addresses | `localhost:9092` | No | +| `EXPORT_TO_KAFKA` | Enable Kafka data export | `false` | No | +| `REDIS_HOST` | Redis server hostname | `localhost` | Yes | +| `REDIS_PORT` | Redis server port | `6379` | Yes | +| `REDIS_USERNAME` | Redis username | `None` | No | +| `REDIS_PASSWORD` | Redis password | `None` | No | +| `LEASE_TTL` | Slot lease time-to-live (seconds) | `10` | No | +| `HEARTBEAT_TTL` | Heartbeat time-to-live (seconds) | `20` | No | +| `HOSTNAME` | Pod identifier | `localhost` | No | +| `POLL_INTERVAL` | Main loop polling interval (seconds) | `5` | No | +| `MONGODB_URL` | MongoDB server address | `localhost:27017` | Yes | +| `MONGODB_DATABASE` | MongoDB database name | `sientia` | No | +| `MONGODB_USERNAME` | MongoDB username | `sientia` | No | +| `MONGODB_PASSWORD` | MongoDB password | `sientia` | No | +| `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | + +### OPC Server Configuration + +OPC servers are configured through Redis with the following structure: + +```json +{ + "slot:opc_tags:1": { + "server1": { + "name": "server1", + "url": "opc.tcp://localhost:4841", + "server_uri": "http://opcua-server.simulator", + "tags": { + "ns=2;i=2": { + "tag_name": "Counter", + "frequency": 1000, + "topics": ["opcua", "counter"] + } + } + } + } +} ``` -Prompt: -Write a summary of PR changes in markdown. Be objective and direct. Write to file \ No newline at end of file + +## 🔧 Development + +### Project Structure +``` +sientia-dataops-opc-ingestor/ +├── ingestor/ # Main application code +│ ├── managers/ # Component managers +│ │ ├── data_manager.py # Data persistence and export +│ │ ├── ingestor_manager.py # Main coordination +│ │ ├── opc_manager.py # OPC UA server management +│ │ └── resource_manager.py # Resource coordination +│ ├── app.py # Main application entry point +│ ├── ingestor.py # Core ingestor logic +│ └── metrics.py # Prometheus metrics definitions +├── simulator/ # OPC simulation and testing tools +├── tests/ # Test suite +├── docker-compose.yaml # Infrastructure services +└── requirements.txt # Python dependencies +``` + +### Adding New Features + +1. **Follow the manager pattern** for new components +2. **Add comprehensive docstrings** for all public methods +3. **Include Prometheus metrics** for monitoring +4. **Add unit tests** for new functionality +5. **Update this README** with new features and configuration + +## 🐛 Troubleshooting + +### Common Issues + +1. **OPC Connection Failures** + - Verify server URLs and network connectivity + - Check certificate paths and security settings + - Review server logs for authentication issues + +2. **Redis Connection Issues** + - Verify Redis server is running and accessible + - Check authentication credentials + - Ensure proper network configuration + +3. **Kafka Export Failures** + - Verify Kafka cluster is running + - Check broker addresses and network connectivity + - Review topic configuration and permissions + +4. **Performance Issues** + - Monitor Prometheus metrics for bottlenecks + - Adjust polling intervals and lease TTLs + - Review OPC server performance and network latency + +### Debug Mode + +Enable debug logging by setting the log level in your environment: +```bash +export LOG_LEVEL=DEBUG +``` + +## Performance Tuning + +### Key Parameters + +- **`POLL_INTERVAL`**: Main loop frequency (lower = more responsive, higher = less CPU) +- **`LEASE_TTL`**: Slot lease duration (lower = faster failover, higher = more stable) +- **`HEARTBEAT_TTL`**: Instance health check frequency +- **Tag frequency**: OPC tag collection rate (Hz) + +### Scaling Considerations + +- **Horizontal Scaling**: Deploy multiple ingestor instances for high availability +- **Load Distribution**: Use Redis-based slot allocation for automatic load balancing +- **Resource Limits**: Monitor CPU, memory, and network usage +- **Database Performance**: Optimize MongoDB indexes and Kafka partitioning + +## 🤝 Contributing + +1. Fork the repository +2. Create a feature branch +3. Make your changes with comprehensive testing +4. Update documentation and docstrings +5. Submit a pull request + +### Code Quality Standards + +- Follow PEP 8 style guidelines +- Include comprehensive docstrings for all public methods +- Maintain test coverage above 80% +- Use type hints where appropriate +- Follow the established architectural patterns + +## 📄 License + +This project is licensed under the terms specified in the LICENSE file. + +## 🆘 Support + +For support and questions: +- Check the troubleshooting section above +- Review the metrics and logs for error patterns +- Open an issue in the project repository +- Contact the development team + +--- + +**Note**: This OPC Ingestor is designed for production use in industrial environments. Ensure proper security configuration and network isolation for production deployments. + +This comprehensive documentation provides: + +1. **Complete feature overview** with architectural details +2. **Detailed installation and setup instructions** +3. **Comprehensive configuration documentation** +4. **Performance tuning and troubleshooting guides** +5. **Development guidelines and contribution standards** +6. **Updated docstrings** for all major classes and methods + +The documentation now serves as a complete reference for users, developers, and operators of the OPC Ingestor system. \ No newline at end of file diff --git a/coverage.sh b/coverage.sh deleted file mode 100755 index 276b6b0..0000000 --- a/coverage.sh +++ /dev/null @@ -1 +0,0 @@ -pytest --cov=ingestor --cov-report=html && xdg-open htmlcov/index.html \ No newline at end of file diff --git a/docker-compose.yaml b/docker-compose.yaml deleted file mode 100644 index 1dbc796..0000000 --- a/docker-compose.yaml +++ /dev/null @@ -1,102 +0,0 @@ -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: - - "4841:4840" - depends_on: - - kafka - - redis - networks: - - kafka-net - env_file: - - .env - -networks: - kafka-net: - driver: bridge diff --git a/ingestor/app.py b/ingestor/app.py index 825b35e..4505657 100644 --- a/ingestor/app.py +++ b/ingestor/app.py @@ -15,6 +15,29 @@ POD_ID = os.getenv("HOSTNAME", "localhost") async def main(): + """ + Main asynchronous function that orchestrates the OPC Ingestor application. + + This function performs the following operations: + 1. Starts the Prometheus metrics server for monitoring + 2. Initializes the Ingestor instance + 3. Prepares the ingestor (connects to services, acquires slot leases) + 4. Runs the main processing loop until shutdown is requested + 5. Handles graceful shutdown and cleanup + + The main loop continuously: + - Processes OPC data from subscribed tags + - Manages slot leases and resource allocation + - Monitors OPC server connections + - Records metrics for monitoring and observability + + Environment Variables: + HOSTNAME: Pod identifier for metrics labeling (default: "localhost") + HTTP_METRICS_PORT: Port for Prometheus metrics server (default: 9090) + + Raises: + Exception: If ingestor preparation fails, the application will exit + """ start_prometheus_server() ingestor = Ingestor() try: @@ -62,11 +85,35 @@ async def main(): def signal_handler(_signum, _frame): + """ + Signal handler for graceful application shutdown. + + This function handles system signals (SIGINT, SIGTERM, SIGHUP) by setting + the exit_signal flag, which triggers the main loop to complete its current + iteration and then shut down gracefully. + + Args: + _signum: The signal number received + _frame: The current stack frame (unused) + """ print(f"Received signal {_signum}. Setting exit_signal flag.") exit_signal.set() def start_prometheus_server(): + """ + Starts the Prometheus metrics HTTP server. + + This function initializes a Prometheus metrics server on the configured port + to expose application metrics for monitoring and alerting. The server provides + metrics about application health, performance, and operational status. + + Environment Variables: + HTTP_METRICS_PORT: Port number for the metrics server (default: 9090) + + Raises: + Exception: If the server fails to start, the application will exit + """ try: port = int(os.getenv("HTTP_METRICS_PORT", 9090)) start_http_server(port) @@ -78,7 +125,15 @@ def start_prometheus_server(): def run_async_main(): - """Run the async main function with proper event loop setup""" + """ + Run the async main function with proper event loop setup. + + This function sets up the asyncio event loop and runs the main async function. + It handles KeyboardInterrupt gracefully and ensures proper cleanup of the event loop. + + The function is designed to work with both direct execution and containerized + environments, providing consistent behavior across different deployment scenarios. + """ try: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 589a3b6..92a7c72 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -12,30 +12,65 @@ import ingestor.metrics as metrics class Ingestor: + """ + Main OPC Ingestor class that orchestrates data collection from OPC UA servers. + + The Ingestor is responsible for: + - Managing slot leases for load balancing across multiple instances + - Connecting to and monitoring OPC UA servers + - Subscribing to OPC tags and collecting real-time data + - Distributing data to Kafka and MongoDB + - Providing health monitoring and metrics collection + + The ingestor uses a slot-based architecture where each slot represents + a collection of OPC tags that can be managed by a single ingestor instance. + This allows for horizontal scaling and load distribution. + + Environment Variables: + KAFKA_SERVERS: Comma-separated list of Kafka server addresses (default: "localhost:9092") + EXPORT_TO_KAFKA: Enable/disable Kafka export (default: "false") + REDIS_HOST: Redis server hostname (default: "localhost") + REDIS_PORT: Redis server port (default: 6379) + REDIS_USERNAME: Redis username (optional) + REDIS_PASSWORD: Redis password (optional) + LEASE_TTL: Time-to-live for slot leases in seconds (default: 10) + HEARTBEAT_TTL: Time-to-live for heartbeats in seconds (default: 20) + HOSTNAME: Pod identifier (default: "localhost") + POLL_INTERVAL: Main loop polling interval in seconds (default: 5) + MONGODB_URL: MongoDB server address (default: "localhost:27017") + MONGODB_USERNAME: MongoDB username (default: "sientia") + MONGODB_PASSWORD: MongoDB password (default: "sientia") + MONGODB_DATABASE: MongoDB database name (default: "sientia") + + Attributes: + export_to_kafka (bool): Whether to export data to Kafka + redis_host (str): Redis server hostname + redis_port (int): Redis server port + redis_username (str): Redis username (optional) + redis_password (str): Redis password (optional) + lease_ttl (int): Time-to-live for slot leases in seconds + heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds + pod_id (str): Identifier for the current pod or host + poll_interval (int): Interval in seconds for polling operations + mongo_database (str): MongoDB database name + mongo_connection_string (str): Complete MongoDB connection string + kafka_servers (list): List of Kafka server addresses + logger: Logger instance for application logging + notification_handler: Handler for sending notifications + metadata (dict): Application metadata for notifications and tracking + ingestor_manager: Manager instance for coordinating operations + """ + 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. - MONGODB_URL (str): URL of the MongoDB server. Defaults to "localhost:27017". - MONGODB_USERNAME (str): Username for the MongoDB server. Defaults to "sientia". - MONGODB_PASSWORD (str): Password for the MongoDB server. Defaults to "sientia". - MONGODB_DATABASE (str): Name of the MongoDB database. Defaults to "sientia". - 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. + + Sets up all necessary connections and configurations for: + - Kafka connectivity (if enabled) + - Redis for slot management and coordination + - MongoDB for data persistence and notifications + - OPC UA server management + - Metrics collection and monitoring """ kafka_servers = getenv("KAFKA_SERVERS", "localhost:9092") @@ -80,19 +115,36 @@ class Ingestor: self.ingestor_manager = None async def shutdown(self): + """ + Gracefully shuts down the ingestor and all its components. + + This method ensures proper cleanup of: + - OPC UA connections and subscriptions + - Resource managers and data connections + - Active slot leases and heartbeats + + Should be called before application termination to prevent resource leaks. + """ if self.ingestor_manager: await self.ingestor_manager.shutdown() async 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. + + This method processes the tags that have been allocated to this ingestor + instance through the slot leasing system. It updates OPC server configurations + and establishes subscriptions to the allocated 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. + + Behavior: + - If no tags are acquired, logs a warning about no slots being available + - If tags are acquired, updates OPC server configurations and subscribes + to the allocated tags for data collection """ if not acquired: @@ -105,24 +157,21 @@ class Ingestor: async def prepare_ingestor(self): """ - Prepares the ingestor by initializing the IngestorManager, declaring the ingestor as active, - acquiring slot leases, and handling the acquired tags. + Prepares the ingestor by initializing all components and acquiring initial slot leases. + 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. + 1. Initializes the IngestorManager with all necessary configuration parameters + 2. Declares the ingestor as active in the coordination system + 3. Acquires slot leases for tag management + 4. Processes the acquired tags and establishes OPC subscriptions + + The preparation phase is critical for establishing the ingestor's role in + the distributed system and ensuring it can begin processing OPC data. + Raises: Exception: If any error occurs during the initialization or lease acquisition process. + This will cause the application to exit as the ingestor cannot function + without proper initialization. """ self.ingestor_manager = IngestorManager( @@ -144,7 +193,7 @@ class Ingestor: export_to_kafka=self.export_to_kafka, ) - # Declare ingestor ative + # Declare ingestor active self.ingestor_manager.declare_active() # Get slot lease @@ -160,12 +209,18 @@ class Ingestor: 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. + + This method handles the case where an ingestor is active but has no + allocated slots. It attempts to acquire a slot lease if slots are + available in the system. + Args: - number_of_slots (int): The number of available slots. + number_of_slots (int): The number of available slots in the system. + + Behavior: + - Only attempts to acquire slots if the ingestor is currently active + (has no managed tags) and there are slots available + - Requests a single slot lease to begin processing """ if not self.ingestor_manager.managed_tags and number_of_slots > 0: @@ -178,24 +233,30 @@ class Ingestor: 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. + Manages the allocation and deallocation of slot leases for ingestors. + + This method implements the load balancing logic for distributing OPC tag + processing across multiple ingestor instances. It ensures optimal resource + utilization and fair distribution of work. + 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 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 + # Some ingestors are inactive, so there are "available_slots" slots available self.logger.info(f"Slots available: {available_slots}") # Get slot lease @@ -222,9 +283,23 @@ class Ingestor: async def update_ingestor_manager(self, old_managed_tags: Dict[str, Any]): """ - Updates the ingestor manager with the new managed tags. + Updates the ingestor manager with new managed tags and handles configuration changes. + + This method compares the current managed tags with the previous state and + performs necessary operations to maintain synchronization: + - Subscribes to newly allocated tags + - Resubscribes to tags with changed configurations + - Unsubscribes from deallocated tags + Args: - old_managed_tags (Dict[str, Any]): The old managed tags. + old_managed_tags (Dict[str, Any]): The previous state of managed tags. + + Behavior: + - Compares current and previous tag configurations + - Establishes subscriptions for new tags + - Updates subscriptions for modified tags + - Removes subscriptions for deallocated tags + - Updates metrics to reflect current state """ self.logger.debug( @@ -267,18 +342,23 @@ class Ingestor: async 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. + Executes the main processing loop for managing ingestors and slots. + + This method is the core of the ingestor's operation, performing the following + tasks in each iteration: + 1. Declares the ingestor as active to maintain its presence in the system + 2. Polls for slot updates and manages resource allocation + 3. Handles scenarios where no slots are available + 4. Manages slot leases based on system load and available resources + 5. Updates OPC server configurations and checks server integrity + 6. Synchronizes managed tags with the current system state + + The loop implements a sophisticated load balancing algorithm that: + - Distributes OPC tag processing across multiple ingestor instances + - Ensures optimal resource utilization + - Maintains system stability during scaling operations + - Provides real-time monitoring and metrics collection + This method is intended to be called repeatedly to ensure the ingestor manager operates correctly and maintains synchronization with the slots and OPC servers. diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 28fe065..f9d7ca1 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -14,6 +14,38 @@ import os class DataManager(BaseActivity): + """ + Manages data persistence and export operations for the OPC Ingestor. + + The DataManager is responsible for: + - Storing OPC data in MongoDB for historical analysis and persistence + - Exporting data to Kafka for real-time streaming and downstream processing + - Managing database connections and ensuring data integrity + - Providing data access interfaces for other components + + The manager supports both MongoDB and Kafka operations, with Kafka export + being optional and configurable. It implements retry logic for connection + failures and provides comprehensive error handling and notification. + + Args: + kafka_servers (str): Comma-separated string of Kafka server addresses + mongo_connection_string (str): MongoDB connection string + mongo_database (str): MongoDB database name + export_to_kafka (bool): Whether to enable Kafka export functionality + metadata (dict): Application metadata for notifications and tracking + logger (Logger): Logger instance for application logging + notification_handler (NotificationHandler): Handler for sending notifications + + Attributes: + pod_id (str): Pod identifier for metrics labeling + kafka_producer (KafkaProducer): Kafka producer instance for data export + export_to_kafka (bool): Whether Kafka export is enabled + connection_string (str): MongoDB connection string + database (str): MongoDB database name + mongo_client (MongoClient): MongoDB client instance + metadata (dict): Application metadata + """ + def __init__( self, kafka_servers: str, @@ -25,15 +57,32 @@ class DataManager(BaseActivity): notification_handler: NotificationHandler, ) -> 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. + Initializes the DataManager instance with Kafka and MongoDB connections. + + This constructor attempts to establish connections to the specified services: + 1. Kafka: Initializes producer with retry logic (up to 3 attempts) + 2. MongoDB: Establishes connection and verifies server availability + + The initialization process includes: + - Kafka producer setup with JSON serialization + - MongoDB client initialization and connection testing + - Metrics recording for connection status + - Error handling with notifications + Args: - kafka_servers (str): A comma-separated string of Kafka server addresses. - logger (Logger): A logger instance for logging messages. + kafka_servers (str): Comma-separated string of Kafka server addresses + mongo_connection_string (str): MongoDB connection string + mongo_database (str): MongoDB database name + export_to_kafka (bool): Whether to enable Kafka export + metadata (dict): Application metadata + logger (Logger): Logger instance + notification_handler (NotificationHandler): Notification handler + Raises: NoBrokersAvailable: If the connection to Kafka servers fails after 3 attempts. + + Metrics: + - KAFKA_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure """ self.pod_id = os.getenv("HOSTNAME", "localhost") diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 9d42ee9..edaaf2e 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -13,6 +13,50 @@ import ingestor.metrics as metrics class IngestorManager(BaseActivity): + """ + Central coordinator for managing OPC data ingestion operations. + + The IngestorManager orchestrates the interaction between different components: + - DataManager: Handles data persistence and Kafka export + - OPC Managers: Manage individual OPC UA server connections + - ResourceManager: Coordinates slot leasing and load balancing + + This class implements a slot-based architecture where: + - Each slot represents a collection of OPC tags from one or more servers + - Slots are distributed across multiple ingestor instances for load balancing + - Dynamic slot allocation ensures optimal resource utilization + + Key Responsibilities: + - Slot lease management and distribution + - OPC server connection lifecycle management + - Tag subscription coordination + - System health monitoring and integrity checks + - Load balancing across multiple ingestor instances + + Args: + kafka_servers (str): Comma-separated list of Kafka server addresses + redis_data (dict): Redis connection parameters (host, port, username, password) + lease_ttl (int): Time-to-live for slot leases in seconds + heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds + poll_interval (int): Main loop polling interval in seconds + mongo_connection_string (str): MongoDB connection string + mongo_database (str): MongoDB database name + metadata (dict): Application metadata for notifications and tracking + logger (Logger): Logger instance for application logging + notification_handler (NotificationHandler): Handler for sending notifications + export_to_kafka (bool): Whether to export data to Kafka + + Attributes: + data_manager (DataManager): Manages data persistence and Kafka export + opc_managers (dict): Dictionary of OPC managers keyed by server name + resource_manager (ResourceManager): Manages Redis-based resource coordination + number_of_slots (int): Total number of slots configured in the system + poll_interval (int): Main loop polling interval + managed_tags (dict): Currently managed tags organized by slot + opc_servers (dict): OPC server configurations + metadata (dict): Application metadata + """ + def __init__(self, kafka_servers: str, redis_data: dict, lease_ttl: int, heartbeat_ttl: int, @@ -61,6 +105,10 @@ class IngestorManager(BaseActivity): async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None: """ Initializes an OPC Manager instance using the provided server configuration. + + This method creates and configures an OPC Manager for a specific OPC UA server, + establishing the connection and preparing it for tag subscriptions. + Args: server_config (dict): A dictionary containing the OPC server configuration. Expected keys include: @@ -70,11 +118,15 @@ class IngestorManager(BaseActivity): - '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. + + Raises: + Exception: If OPC manager initialization fails, the error is logged and + a notification is sent, but the method returns None to allow + the system to continue operating with other servers. """ try: diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 0e307c5..2685155 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -13,6 +13,47 @@ import ingestor.metrics as metrics class OpcManager(BaseActivity): + """ + Manages OPC UA server connections and tag subscriptions. + + The OpcManager is responsible for: + - Establishing and maintaining secure connections to OPC UA servers + - Managing tag subscriptions and data collection + - Handling server reconnection and error recovery + - Processing OPC data and forwarding it to the data manager + - Monitoring connection health and performance metrics + + The manager supports both secure and unsecured connections, with optional + certificate-based authentication for enhanced security. + + Args: + name (str): Unique identifier for the OPC server + url (str): OPC UA server endpoint URL + data_manager (DataManager): Manager for data persistence and export + logger (Logger): Logger instance for application logging + server_uri (str): OPC UA server application URI + notification_handler (NotificationHandler): Handler for sending notifications + metadata (dict): Application metadata for notifications and tracking + cert_path (str, optional): Path to client certificate file for secure connections + private_key_path (str, optional): Path to client private key file + server_cert_path (str, optional): Path to server certificate file for validation + + Attributes: + url (str): OPC UA server endpoint URL + name (str): Unique identifier for the OPC server + server_uri (str): OPC UA server application URI + data_queue (dict): Queue for buffering OPC data before processing + non_receive_count (int): Counter for cycles without data reception + client (Client): OPC UA client instance + cert_path (str): Path to client certificate file + private_key_path (str): Path to client private key file + server_cert_path (str): Path to server certificate file + nodes (dict): Dictionary of OPC node references + subscriptions (dict): Active OPC subscriptions + data_manager (DataManager): Manager for data persistence and export + metadata (dict): Application metadata + """ + def __init__(self, name: str, url: str, data_manager: DataManager, logger: Logger, server_uri: str, notification_handler: NotificationHandler, metadata: dict, cert_path: str = None, private_key_path: str = None, server_cert_path: str = None): @@ -40,11 +81,27 @@ class OpcManager(BaseActivity): pod_id=self.pod_id, server_name=self.name).set(0) def __str__(self): + """ + String representation of the OPC Manager. + + Returns: + str: Human-readable representation showing server details and current state. + """ return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \ f"nodes={self.nodes}, subscriptions={self.subscriptions}" async def shutdown(self): - """Comprehensive cleanup method""" + """ + Comprehensive cleanup method for graceful shutdown. + + This method ensures proper cleanup of all OPC UA resources: + - Closes active subscriptions + - Disconnects from the OPC server + - Releases allocated resources + + Should be called before the application terminates to prevent resource leaks + and ensure clean disconnection from OPC servers. + """ try: await self.disconnect() @@ -54,21 +111,24 @@ class OpcManager(BaseActivity): async 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. + It implements Basic256 security policy with certificate-based authentication. + 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 + + The method configures: + - Client application URI + - Certificate-based authentication + - Server certificate validation (if provided) + - Connection timeouts for stability """ if not all([self.cert_path, self.private_key_path]): @@ -93,11 +153,23 @@ class OpcManager(BaseActivity): async 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. + + The connection process includes: + 1. Client initialization with server URL + 2. Security configuration (if certificates are provided) + 3. Connection establishment + 4. Metrics recording for monitoring + Raises: Exception: If the connection to the OPC server fails. + + Metrics: + - OPC_CONNECTIONS_TOTAL: Incremented on connection attempt + - OPC_CONNECTION_STATUS: Set to 1 on successful connection """ metrics.OPC_CONNECTIONS_TOTAL.labels( diff --git a/ingestor/managers/resource_manager.py b/ingestor/managers/resource_manager.py index 5288d69..b77e3c3 100644 --- a/ingestor/managers/resource_manager.py +++ b/ingestor/managers/resource_manager.py @@ -10,6 +10,39 @@ from sientia_do.temporal.activities.base import BaseActivity class ResourceManager(BaseActivity): + """ + Manages Redis-based resource coordination and slot leasing for the OPC Ingestor. + + The ResourceManager is responsible for: + - Coordinating slot allocation across multiple ingestor instances + - Managing lease lifecycles and heartbeats for load balancing + - Providing distributed locking and resource management + - Monitoring Redis operations and connection health + + The manager implements a sophisticated slot leasing system that enables: + - Dynamic load distribution across multiple ingestor instances + - Automatic failover and recovery from instance failures + - Fair resource allocation based on system capacity + - Real-time monitoring of system health and performance + + Args: + host (str): Redis server hostname + port (int): Redis server port + lease_ttl (int): Time-to-live for slot leases in seconds + heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds + metadata (dict): Application metadata for notifications and tracking + logger (Logger): Logger instance for application logging + notification_handler (NotificationHandler): Handler for sending notifications + username (str, optional): Redis username for authentication + password (str, optional): Redis password for authentication + + Attributes: + redis (Redis): Redis client instance + lease_ttl (int): Time-to-live for slot leases + heartbeat_ttl (int): Time-to-live for heartbeat signals + metadata (dict): Application metadata + """ + def __init__( self, host: str, @@ -22,6 +55,31 @@ class ResourceManager(BaseActivity): username: str | None = None, password: str | None = None, ) -> None: + """ + Initializes the ResourceManager with Redis connection and configuration. + + This constructor establishes a connection to Redis and verifies connectivity + by performing a ping operation. It sets up the connection with optional + authentication and records the connection status in metrics. + + Args: + host (str): Redis server hostname + port (int): Redis server port + lease_ttl (int): Time-to-live for slot leases in seconds + heartbeat_ttl (int): Time-to-live for heartbeat signals in seconds + metadata (dict): Application metadata + logger (Logger): Logger instance + notification_handler (NotificationHandler): Notification handler + username (str, optional): Redis username for authentication + password (str, optional): Redis password for authentication + + Raises: + Exception: If Redis connection fails, the error is logged and metrics + are updated before re-raising the exception. + + Metrics: + - REDIS_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure + """ BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler, set_error_counter=True) @@ -45,7 +103,32 @@ class ResourceManager(BaseActivity): self.metadata = metadata def _execute_redis_op(self, operation_name: str, func, *args, **kwargs): - """Wrapper to execute Redis operations and record metrics.""" + """ + Wrapper to execute Redis operations and record metrics. + + This method provides a unified interface for Redis operations that: + - Records operation timing and success/failure metrics + - Handles error notifications consistently + - Ensures all Redis operations are properly monitored + + Args: + operation_name (str): Name of the Redis operation for metrics labeling + func: The Redis function to execute + *args: Positional arguments for the Redis function + **kwargs: Keyword arguments for the Redis function + + Returns: + The result of the Redis operation + + Raises: + Exception: Re-raises any exception from the Redis operation after + recording error metrics and sending notifications. + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented on successful operations + - REDIS_OPERATIONS_DURATION: Records operation timing + - REDIS_OPERATIONS_ERRORS: Incremented on operation failures + """ start_time = time() try: result = func(*args, **kwargs) @@ -73,11 +156,20 @@ class ResourceManager(BaseActivity): def get(self, key: str) -> dict: """ Retrieve a value from Redis by its key and return it as a dictionary. + + This method fetches a value from Redis and attempts to parse it as JSON. + If the key doesn't exist or the value is empty, it returns None. + 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. + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented with operation="get" + - REDIS_OPERATIONS_DURATION: Records timing for get operations """ history = self._execute_redis_op("get", self.redis.get, key) @@ -86,10 +178,19 @@ class ResourceManager(BaseActivity): def get_tag_slot(self, id: str) -> dict: """ Retrieve the tag slot information for a given ID. + + This method constructs the Redis key for a tag slot and retrieves + the associated configuration information. + 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. + dict: A dictionary containing the tag slot information associated + with the given ID, or None if not found. + + The method constructs the key using the pattern "slot:opc_tags:{id}" + and delegates to the get() method for the actual Redis operation. """ return self.get(f"slot:opc_tags:{id}") @@ -97,12 +198,20 @@ class ResourceManager(BaseActivity): 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 + + The heartbeat mechanism enables: + - Load balancers to identify active ingestor instances + - Health monitoring systems to detect failed instances + - Automatic failover and recovery mechanisms + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented with operation="set" + - REDIS_OPERATIONS_DURATION: Records timing for heartbeat operations """ self._execute_redis_op( @@ -115,14 +224,28 @@ class ResourceManager(BaseActivity): 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`. + Attempts to lease a tag by setting a key in Redis with a specified TTL. + + 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`. This implements a distributed + locking mechanism for tag allocation. + Args: tag_id (str): The unique identifier of the tag to be leased. + Returns: - bool: True if the lease was successfully acquired, False otherwise. + bool: True if the lease was successfully acquired, False if the tag + is already leased by another ingestor. + + The leasing mechanism ensures: + - Only one ingestor can process a specific tag at a time + - Automatic lease expiration prevents deadlocks + - Fair distribution of tags across available ingestor instances + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented with operation="set_nx" + - REDIS_OPERATIONS_DURATION: Records timing for lease operations """ return self._execute_redis_op( @@ -137,13 +260,26 @@ class ResourceManager(BaseActivity): 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`). + + 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. + 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. + bool: True if the lease was successfully renewed, False if the current + pod doesn't hold the lease or renewal failed. + + Lease renewal is essential for: + - Maintaining continuous tag processing without interruptions + - Preventing lease expiration during long-running operations + - Ensuring system stability and reliability + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented with operation="get" and "expire" + - REDIS_OPERATIONS_DURATION: Records timing for renewal operations """ current = self._execute_redis_op( @@ -159,11 +295,22 @@ class ResourceManager(BaseActivity): 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. + + This method removes the lease for the given OPC tag by deleting the + corresponding key in Redis. This is typically called when an ingestor + is shutting down or when it needs to release a tag for reallocation. + Args: tag_id (str): The identifier of the OPC tag whose lease is to be dropped. - Returns: - None + + Lease dropping enables: + - Graceful shutdown of ingestor instances + - Dynamic reallocation of tags for load balancing + - Recovery from failed or unresponsive ingestor instances + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented with operation="delete" + - REDIS_OPERATIONS_DURATION: Records timing for lease dropping operations """ self._execute_redis_op("delete", self.redis.delete, @@ -172,21 +319,47 @@ class ResourceManager(BaseActivity): 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. + + This method fetches all keys in Redis that match the pattern for ingestor + heartbeats and returns a list of active ingestor identifiers. The method + uses the pattern "heartbeat:ingestor:*" to find all active instances. + Returns: - list: A list of active ingestors. + List[str]: A list of active ingestor identifiers, extracted from + the Redis keys by removing the "heartbeat:ingestor:" prefix. + + This information is used for: + - Load balancing calculations + - System health monitoring + - Resource allocation decisions + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented with operation="keys" + - REDIS_OPERATIONS_DURATION: Records timing for ingestor discovery """ return self._execute_redis_op("keys", 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. + Retrieves all available slots from Redis. + + This method fetches all keys in Redis that match the pattern for OPC tag + slots and returns a list of slot identifiers. The method uses the pattern + "slot:opc_tags:*" to find all configured slots. + Returns: - int: The number of slots available. + List[str]: A list of slot identifiers, extracted from the Redis keys + by removing the "slot:opc_tags:" prefix. + + Slot information is used for: + - Resource allocation planning + - Load balancing across ingestor instances + - System capacity monitoring + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented with operation="keys" + - REDIS_OPERATIONS_DURATION: Records timing for slot discovery """ return self._execute_redis_op("keys", self.redis.keys, "slot:opc_tags:*") @@ -194,10 +367,23 @@ class ResourceManager(BaseActivity): 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. + + This method fetches all keys in Redis that match the pattern for OPC tag + leases and returns a list of lease identifiers. The method uses the pattern + "lease:opc_tags:*" to find all active leases. + Returns: - list: A list of active leases. + List[str]: A list of lease identifiers, extracted from the Redis keys + by removing the "lease:opc_tags:" prefix. + + Lease information is used for: + - Current resource utilization monitoring + - Load balancing calculations + - System health and performance analysis + + Metrics: + - REDIS_OPERATIONS_TOTAL: Incremented with operation="keys" + - REDIS_OPERATIONS_DURATION: Records timing for lease discovery """ return self._execute_redis_op("keys", self.redis.keys, "lease:opc_tags:*") diff --git a/ingestor/metrics.py b/ingestor/metrics.py index 8cb2110..e1ddb4e 100644 --- a/ingestor/metrics.py +++ b/ingestor/metrics.py @@ -1,5 +1,22 @@ +""" +Prometheus metrics configuration for the OPC Ingestor application. + +This module defines all the metrics used for monitoring and observability +of the OPC Ingestor system. It includes metrics for: + +- Application health and performance +- OPC server connections and subscriptions +- Data processing and storage operations +- Resource management and load balancing +- Error tracking and notification systems + +All metrics follow Prometheus naming conventions and include appropriate +labels for multi-dimensional analysis and alerting. +""" + from prometheus_client import Counter, Gauge, Histogram +# Metric label definitions for consistent labeling across all metrics POD_ID_LABEL = ["pod_id"] SERVER_LABELS = ["pod_id", "server_name", "server_url"] KAFKA_LABELS = ["pod_id", "topic"] @@ -15,7 +32,6 @@ TAG_WRITTEN_COUNT = Counter( [*MAIN_LABELS, "tag_name", "collection_name"], ) - # --- General Application Metrics --- APP_LOOP_COUNT = Counter( "app_main_loop_total", diff --git a/simulator/Dockerfile b/simulator/Dockerfile deleted file mode 100644 index d467676..0000000 --- a/simulator/Dockerfile +++ /dev/null @@ -1,30 +0,0 @@ -# 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"] diff --git a/simulator/redis-feeder.py b/simulator/redis-feeder.py deleted file mode 100644 index d8e42c0..0000000 --- a/simulator/redis-feeder.py +++ /dev/null @@ -1,60 +0,0 @@ -import json -import redis - - -# Redis connection settings -REDIS_HOST = "localhost" -REDIS_PORT = 6379 -REDIS_USERNAME = None # "default" -REDIS_PASSWORD = None # "bdnZOpcyiL" - -OPC_URL = "opc.tcp://sientia-opc-simulator-service.sientia-opc.svc.cluster.local:4841" -OPC_URL = "opc.tcp://localhost:4841" - -# Connect to Redis -r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, - decode_responses=True, username=REDIS_USERNAME, password=REDIS_PASSWORD) - -# 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_URL, - "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'], - }, - } - } - } -} - -for key, val in new_data.items(): - r.set(key, json.dumps(val)) - print(f"✅ Set: {key} -> {val}") - -print("🚀 OPC tag keys replaced successfully.")