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

Merge pull request #18 from Aignosi/SIENTIAPDE-1084-ajustar-documentacao

Browse Source 
SIENTIAPDE-1084: Enhance Data Management and Documentation
This commit is contained in:
Bruno Domingues
2025-09-03 13:32:48 +00:00
committed by GitHub
parent a384a077a9 0a3826b767
commit 28dd9eb5c7
14 changed files with 1206 additions and 418 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
.env.example Normal file
View File

@@ -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"

30
Dockerfile
View File

@@ -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"]

435
README.md
View File

@@ -1,81 +1,396 @@
# sientia-dataops-opc-ingestor # Sientia DataOps OPC Ingestor
OPC gateway to manage Scouter pipelines
## Local tests 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.
### Generate your ssh key to Docker
``` ## Features
ssh-keygen -t ed25519 -C "docker-access" -f ~/.ssh/id_ed25519_docker
``` ### Core Functionality
Add the public key to yout Git SSH keys - **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 historical data streaming
- **Persistent Storage**: MongoDB integration for historical data persistence
- **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 designed for scalability and fault tolerance:
### Enable Docker BuildKit
``` ```
export DOCKER_BUILDKIT=1 ┌────────────────────────────────────────────
``` │ Main App │ │ IngestorManager │ │ OPC Manager │
or make it permanent: │ │◄──►│ │◄──►│ │
``` │ - Signal Hand. │ │ - Slot Mgmt │ │ - Connections │
echo '{ "features": { "buildkit": true } }' | sudo tee /etc/docker/daemon.json │ - Metrics │ │ - Load Balancing │ │ - Subscriptions │
sudo systemctl restart docker │ - Lifecycle │ │ - Coordination │ │ - Data Handler │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│
▼
┌──────────────────┐ ┌─────────────────┐
│ ResourceManager │ │ DataManager │
│ │ │ │
│ - Redis Coord. │ │ - Kafka Export │
│ - Slot Leasing │ │ - MongoDB Store │
│ - Heartbeats │ │ - Data Pipeline │
└──────────────────┘ └─────────────────┘
``` ```
### Run docker compose ### Key Components
```
docker compose down -v - **Ingestor**: Main application orchestrator managing the overall lifecycle
docker compose build --ssh default=$HOME/.ssh/id_ed25519_docker - **IngestorManager**: Coordinates slot allocation, OPC server management, and load balancing
docker compose up -d - **OPC Manager**: Handles individual OPC UA server connections and tag subscriptions
- **Data Manager**: Manages data streaming (MongoDB | 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 <repository-url>
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 script will:
The ingestor can be run as a docker container or as a python script. - Activate the virtual environment
- Load environment variables from `.env`
- Start the ingestor application
As a docker container: ### Running Tests and Coverage
Uncomment the ingestor service in docker-compose.yaml
``` Use the provided script to run tests with coverage:
docker compose up -d ingestor
```bash
# Make script executable (first time only)
chmod +x run_coverage.sh
# Run tests with coverage
./run_coverage.sh
``` ```
As a python script: The script will:
``` - Activate the virtual environment
python -m ingestor.app - Run pytest with coverage reporting
``` - Generate HTML coverage report
- Open the coverage report in your browser
### Populate redis server ### Manual Test Execution
Create venv with python3.11
``` You can also run tests manually:
python3.11 -m venv venv
```bash
# Activate virtual environment
source ./venv/bin/activate source ./venv/bin/activate
```
Install requirements
```
pip install -r requirements.txt
```
Run feeder
```
python simulator/redis-feeder.py
```
## Unit tests # Run all tests
### Install pytest
```
pip install pytest
```
### Run pytest
```
pytest 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 pip install pytest-cov
pytest --cov=ingestor pytest --cov=ingestor
```
### Generate complete report # Generate HTML coverage report
```
pytest --cov=ingestor --cov-report=html 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",
"server_id": "1",
"url": "opc.tcp://localhost:4841",
"server_uri": "http://opcua-server.simulator",
"tags": {
"ns=2;i=2": {
"aggr_func": "avg",
"data_range": [
-100,
100
],
"frequency": "15000",
"server_id": "1",
"tag_address": "ns=2;i=2",
"tag_name": "Counter",
"topics": [
"raw_scouter-opcua-orchestrated-pipeline",
"raw_scouter-basic-sum-model"
]
}
}
}
}
``` ```
Prompt:
Write a summary of PR changes in markdown. Be objective and direct. Write to 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.

1
coverage.sh
View File

@@ -1 +0,0 @@
pytest --cov=ingestor --cov-report=html && xdg-open htmlcov/index.html

102
docker-compose.yaml
View File

@@ -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

57
ingestor/app.py
View File

@@ -15,6 +15,29 @@ POD_ID = os.getenv("HOSTNAME", "localhost")
async def main(): 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() start_prometheus_server()
ingestor = Ingestor() ingestor = Ingestor()
try: try:
@@ -62,11 +85,35 @@ async def main():
def signal_handler(_signum, _frame): 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.") print(f"Received signal {_signum}. Setting exit_signal flag.")
exit_signal.set() exit_signal.set()
def start_prometheus_server(): 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: try:
port = int(os.getenv("HTTP_METRICS_PORT", 9090)) port = int(os.getenv("HTTP_METRICS_PORT", 9090))
start_http_server(port) start_http_server(port)
@@ -78,7 +125,15 @@ def start_prometheus_server():
def run_async_main(): 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: try:
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)

208
ingestor/ingestor.py
View File

@@ -12,30 +12,65 @@ import ingestor.metrics as metrics
class Ingestor: 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): def __init__(self):
""" """
Initializes the ingestor with configuration values retrieved from environment variables. Initializes the ingestor with configuration values retrieved from environment variables.
Environment Variables:
KAFKA_SERVERS (str): Comma-separated list of Kafka server addresses. Sets up all necessary connections and configurations for:
Defaults to "localhost:9092". - Kafka connectivity (if enabled)
REDIS_HOST (str): Hostname of the Redis server. Defaults to "localhost". - Redis for slot management and coordination
REDIS_PORT (int): Port number of the Redis server. Defaults to 6379. - MongoDB for data persistence and notifications
LEASE_TTL (int): Time-to-live for leases in seconds. Defaults to 10. - OPC UA server management
HEARTBEAT_TTL (int): Time-to-live for heartbeats in seconds. Defaults to 20. - Metrics collection and monitoring
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.
""" """
kafka_servers = getenv("KAFKA_SERVERS", "localhost:9092") kafka_servers = getenv("KAFKA_SERVERS", "localhost:9092")
@@ -80,19 +115,36 @@ class Ingestor:
self.ingestor_manager = None self.ingestor_manager = None
async def shutdown(self): 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: if self.ingestor_manager:
await self.ingestor_manager.shutdown() await self.ingestor_manager.shutdown()
async def handle_acquired_tags(self, acquired): async def handle_acquired_tags(self, acquired):
""" """
Handles the acquired tags by subscribing to them if available. 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 This method processes the tags that have been allocated to this ingestor
updates the OPC servers and subscribes to the acquired tags. instance through the slot leasing system. It updates OPC server configurations
and establishes subscriptions to the allocated tags.
Args: Args:
acquired (list): A list of acquired tags to be processed. If the list acquired (list): A list of acquired tags to be processed. If the list
is empty or None, no action is taken other than logging is empty or None, no action is taken other than logging
a warning. 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: if not acquired:
@@ -105,24 +157,21 @@ class Ingestor:
async def prepare_ingestor(self): async def prepare_ingestor(self):
""" """
Prepares the ingestor by initializing the IngestorManager, declaring the ingestor as active, Prepares the ingestor by initializing all components and acquiring initial slot leases.
acquiring slot leases, and handling the acquired tags.
This method performs the following steps: This method performs the following steps:
1. Initializes the `IngestorManager` with the necessary configuration parameters. 1. Initializes the IngestorManager with all necessary configuration parameters
2. Declares the ingestor as active by calling `declare_active` on the `IngestorManager`. 2. Declares the ingestor as active in the coordination system
3. Acquires slot leases using the `get_slot_leases` method of the `IngestorManager`. 3. Acquires slot leases for tag management
4. Logs the acquired slots and processes them using the `handle_acquired_tags` method. 4. Processes the acquired tags and establishes OPC subscriptions
Attributes:
self.kafka_servers (list): List of Kafka server addresses. The preparation phase is critical for establishing the ingestor's role in
self.redis_host (str): Redis server hostname. the distributed system and ensuring it can begin processing OPC data.
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: Raises:
Exception: If any error occurs during the initialization or lease acquisition process. 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( self.ingestor_manager = IngestorManager(
@@ -144,7 +193,7 @@ class Ingestor:
export_to_kafka=self.export_to_kafka, export_to_kafka=self.export_to_kafka,
) )
# Declare ingestor ative # Declare ingestor active
self.ingestor_manager.declare_active() self.ingestor_manager.declare_active()
# Get slot lease # Get slot lease
@@ -160,12 +209,18 @@ class Ingestor:
def manage_no_slots(self, number_of_slots: int): def manage_no_slots(self, number_of_slots: int):
""" """
Manages the scenario where there are no slots assigned to the ingestor. 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 This method handles the case where an ingestor is active but has no
conditions are met, it attempts to acquire a slot lease and handles allocated slots. It attempts to acquire a slot lease if slots are
the acquired tags accordingly. available in the system.
Args: 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: 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 self, available_slots: int, lacking_ingestors: int, slot_diff: int
): ):
""" """
Manages the allocation and deallocation of slot leases for ingestors based on Manages the allocation and deallocation of slot leases for ingestors.
the number of available slots, lacking ingestors, and slot differences.
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: Args:
available_slots (int): The number of slots currently available for allocation. available_slots (int): The number of slots currently available for allocation.
lacking_ingestors (int): The number of ingestors that are active and without slots. 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. slot_diff (int): The difference between the total slots and the required slots.
Behavior: Behavior:
- If there are available slots and lacking ingestors, attempts to acquire slot leases - If there are available slots and lacking ingestors, attempts to acquire
for the available slots and processes the acquired tags. 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), - If there are no lacking ingestors but there are extra slots (slot_diff > 0),
releases the extra slot leases to ensure proper allocation. releases the extra slot leases to ensure proper allocation.
Logs: Logs:
- Logs the number of available slots when attempting to acquire leases. - Logs the number of available slots when attempting to acquire leases.
- Logs the number of extra slots when releasing leases. - Logs the number of extra slots when releasing leases.
""" """
if available_slots > 0 and lacking_ingestors > 0: 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}") self.logger.info(f"Slots available: {available_slots}")
# Get slot lease # Get slot lease
@@ -222,9 +283,23 @@ class Ingestor:
async def update_ingestor_manager(self, old_managed_tags: Dict[str, Any]): async def update_ingestor_manager(self, old_managed_tags: Dict[str, Any]):
""" """
Updates the ingestor manager with 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: 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( self.logger.debug(
@@ -267,18 +342,23 @@ class Ingestor:
async def loop(self): async def loop(self):
""" """
Executes the main loop for managing ingestors and slots. Executes the main processing loop for managing ingestors and slots.
This method performs the following tasks:
1. Declares the ingestor as active. This method is the core of the ingestor's operation, performing the following
2. Logs the start of the polling process for slot updates. tasks in each iteration:
3. Retrieves the list of active ingestors and the number of available slots. 1. Declares the ingestor as active to maintain its presence in the system
4. Handles scenarios where no slots are available. 2. Polls for slot updates and manages resource allocation
5. Calculates the difference between the number of slots and active ingestors, 3. Handles scenarios where no slots are available
as well as the difference in managed tags. 4. Manages slot leases based on system load and available resources
6. Manages leases based on the calculated differences. 5. Updates OPC server configurations and checks server integrity
7. Logs the current state of active ingestors, slots, managed tags, and servers. 6. Synchronizes managed tags with the current system state
8. Logs a message if no slots are acquired during the loop.
9. Updates the configuration of OPC servers. 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 This method is intended to be called repeatedly to ensure the ingestor
manager operates correctly and maintains synchronization with the slots manager operates correctly and maintains synchronization with the slots
and OPC servers. and OPC servers.

94
ingestor/managers/data_manager.py
View File

@@ -14,6 +14,38 @@ import os
class DataManager(BaseActivity): 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__( def __init__(
self, self,
kafka_servers: str, kafka_servers: str,
@@ -25,15 +57,32 @@ class DataManager(BaseActivity):
notification_handler: NotificationHandler, notification_handler: NotificationHandler,
) -> None: ) -> None:
""" """
Initializes the DataManager instance with a Kafka producer. Initializes the DataManager instance with Kafka and MongoDB connections.
This constructor attempts to establish a connection to the specified Kafka servers
and initializes a Kafka producer for sending messages. It retries the connection This constructor attempts to establish connections to the specified services:
up to 3 times if the Kafka servers are unavailable. 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: Args:
kafka_servers (str): A comma-separated string of Kafka server addresses. kafka_servers (str): Comma-separated string of Kafka server addresses
logger (Logger): A logger instance for logging messages. 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: Raises:
NoBrokersAvailable: If the connection to Kafka servers fails after 3 attempts. 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") self.pod_id = os.getenv("HOSTNAME", "localhost")
@@ -100,7 +149,17 @@ class DataManager(BaseActivity):
set_error_counter=True) set_error_counter=True)
def shutdown(self): def shutdown(self):
"""Closes the Kafka producer connection.""" """
Gracefully shuts down the DataManager and closes all connections.
This method ensures proper cleanup of:
- Kafka producer connection with message flushing
- MongoDB client connection
- Metrics recording for connection status
The method handles connection closure gracefully, logging any errors
that occur during shutdown while ensuring all resources are properly released.
"""
if self.kafka_producer: if self.kafka_producer:
try: try:
self.kafka_producer.flush(timeout=10) self.kafka_producer.flush(timeout=10)
@@ -127,13 +186,30 @@ class DataManager(BaseActivity):
self.shutdown() self.shutdown()
def delivery_report(self, msg: str): def delivery_report(self, msg: str):
"""Callback for delivery reports from Kafka.""" """
Callback for successful Kafka message delivery reports.
This method is called by the Kafka producer when a message is successfully
delivered to a topic. It logs the delivery details including topic, partition,
and offset information for debugging and monitoring purposes.
Args:
msg: Kafka message object containing delivery details
"""
self.logger.debug( self.logger.debug(
f"Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}" f"Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}"
) )
def delivery_error(self, err: str): def delivery_error(self, err: str):
"""Callback for delivery reports from Kafka.""" """
Callback for Kafka message delivery error reports.
This method is called by the Kafka producer when a message delivery fails.
It logs the error details for debugging and monitoring purposes.
Args:
err: Error information from the failed delivery attempt
"""
self.logger.error(f"Delivery failed for record : {err}") self.logger.error(f"Delivery failed for record : {err}")
def publish(self, topic: str, data: dict) -> None: def publish(self, topic: str, data: dict) -> None:

180
ingestor/managers/ingestor_manager.py
View File

@@ -13,6 +13,50 @@ import ingestor.metrics as metrics
class IngestorManager(BaseActivity): 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, def __init__(self,
kafka_servers: str, redis_data: dict, kafka_servers: str, redis_data: dict,
lease_ttl: int, heartbeat_ttl: int, 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: async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None:
""" """
Initializes an OPC Manager instance using the provided server configuration. 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: Args:
server_config (dict): A dictionary containing the OPC server configuration. server_config (dict): A dictionary containing the OPC server configuration.
Expected keys include: Expected keys include:
@@ -70,11 +118,15 @@ class IngestorManager(BaseActivity):
- 'cert_path' (str, optional): Path to the client certificate file. - 'cert_path' (str, optional): Path to the client certificate file.
- 'private_key_path' (str, optional): Path to the private key file. - 'private_key_path' (str, optional): Path to the private key file.
- 'server_cert_path' (str, optional): Path to the server certificate 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: Returns:
OpcManager | None: An initialized OpcManager instance if successful, OpcManager | None: An initialized OpcManager instance if successful,
otherwise None if an error occurs during initialization. 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: try:
@@ -112,6 +164,17 @@ class IngestorManager(BaseActivity):
return manager return manager
async def shutdown(self): async def shutdown(self):
"""
Gracefully shuts down the IngestorManager and all its components.
This method ensures proper cleanup of:
- All OPC manager instances and their connections
- Data manager connections and resources
- Active subscriptions and server connections
The shutdown process is performed asynchronously to allow proper cleanup
of all managed resources before termination.
"""
for _server_name, server in self.opc_managers.items(): for _server_name, server in self.opc_managers.items():
await server.shutdown() await server.shutdown()
self.data_manager.shutdown() self.data_manager.shutdown()
@@ -200,6 +263,17 @@ class IngestorManager(BaseActivity):
def check_opc_servers_integrity(self): def check_opc_servers_integrity(self):
""" """
Checks the integrity of the OPC servers and updates the OPC servers if necessary. Checks the integrity of the OPC servers and updates the OPC servers if necessary.
This method performs health checks on all managed OPC servers by:
- Checking cycle counts for data reception
- Monitoring connection health and data flow
- Triggering reconnection for lost servers
- Updating metrics for active OPC managers
Side Effects:
- Updates cycle monitoring for all nodes
- Removes lost servers from managed tags
- Updates OPC manager metrics
""" """
for server, opc_manager in self.opc_managers.items(): for server, opc_manager in self.opc_managers.items():
opc_manager.check_cycles() opc_manager.check_cycles()
@@ -220,8 +294,14 @@ class IngestorManager(BaseActivity):
def declare_active(self): def declare_active(self):
""" """
Declares the ingestor as active by sending a heartbeat signal to the resource manager. 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 This method ensures that the ingestor is marked as active by invoking the
`ingestor_heartbeat` method of the associated resource manager. `ingestor_heartbeat` method of the associated resource manager. The heartbeat
mechanism enables load balancers and monitoring systems to track active instances.
Side Effects:
- Updates Redis with current instance heartbeat
- Enables load balancing and health monitoring
""" """
self.resource_manager.ingestor_heartbeat() self.resource_manager.ingestor_heartbeat()
@@ -229,10 +309,15 @@ class IngestorManager(BaseActivity):
def get_active_ingestors(self) -> List[str]: def get_active_ingestors(self) -> List[str]:
""" """
Retrieve a list of active ingestors. Retrieve a list of active ingestors.
This method fetches all ingestors from the resource manager and returns them. This method fetches all ingestors from the resource manager and returns them.
If no ingestors are found, an empty list is returned. If no ingestors are found, an empty list is returned.
Returns: Returns:
List[str]: A list of active ingestor names, or an empty list if none are found. List[str]: A list of active ingestor names, or an empty list if none are found.
The method queries Redis for all active ingestor heartbeats and extracts
the pod identifiers for load balancing and coordination purposes.
""" """
ingestors = self.resource_manager.get_all_ingestors() ingestors = self.resource_manager.get_all_ingestors()
@@ -241,10 +326,16 @@ class IngestorManager(BaseActivity):
def get_number_of_leases(self) -> int: def get_number_of_leases(self) -> int:
""" """
Retrieves the number of leases managed by the resource manager. Retrieves the number of leases managed by the resource manager.
This method fetches all available leases from the resource manager, This method fetches all available leases from the resource manager,
calculates their count, and updates the `number_of_slots` attribute. calculates their count, and updates the `number_of_slots` attribute.
Returns: Returns:
int: The total number of leases. Returns 0 if no leases are available. int: The total number of leases. Returns 0 if no leases are available.
Side Effects:
- Updates internal slot count tracking
- Updates Prometheus metrics for total leases
""" """
leases = self.resource_manager.get_all_leases() leases = self.resource_manager.get_all_leases()
@@ -255,10 +346,16 @@ class IngestorManager(BaseActivity):
def get_number_of_slots(self) -> int: def get_number_of_slots(self) -> int:
""" """
Retrieves the number of slots managed by the resource manager. Retrieves the number of slots managed by the resource manager.
This method fetches all available slots from the resource manager, This method fetches all available slots from the resource manager,
calculates their count, and updates the `number_of_slots` attribute. calculates their count, and updates the `number_of_slots` attribute.
Returns: Returns:
int: The total number of slots. Returns 0 if no slots are available. int: The total number of slots. Returns 0 if no slots are available.
Side Effects:
- Updates internal slot count tracking
- Updates Prometheus metrics for total slots
""" """
slots = self.resource_manager.get_all_slots() slots = self.resource_manager.get_all_slots()
@@ -269,22 +366,29 @@ class IngestorManager(BaseActivity):
def get_slot_leases(self, max_slots: int = 1) -> Dict: def get_slot_leases(self, max_slots: int = 1) -> Dict:
""" """
Acquires a specified number of resource slots by leasing them from the resource manager. Acquires a specified number of resource slots by leasing them from the resource manager.
This method implements the slot acquisition logic for load balancing:
- Iterates through available slots and attempts to lease them
- Logs the leasing of each slot
- Updates the `managed_tags` attribute with the acquired slots
- Stops leasing once the specified `max_slots` are acquired
Args: Args:
max_slots (int): The maximum number of slots to lease. Defaults to 1. max_slots (int): The maximum number of slots to lease. Defaults to 1.
Returns: Returns:
Dict: A dictionary where the keys are the slot identifiers (as strings) Dict: A dictionary where the keys are the slot identifiers (as strings)
and the values are the leased slot details. and the values are the leased slot details.
Behavior: Behavior:
- Iterates through available slots and attempts to lease - Attempts to lease slots sequentially starting from slot 1
them using the resource manager. - Skips slots that cannot be retrieved after leasing
- Logs the leasing of each slot. - Logs warnings if unable to acquire the requested number of slots
- Updates the `managed_tags` attribute with the acquired slots. - Updates metrics for acquired slots and managed slots count
- 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: Notes:
- If a slot is leased but its details cannot be retrieved - If a slot is leased but its details cannot be retrieved
(i.e., `get_tag_slot` returns None), that slot is skipped. (i.e., `get_tag_slot` returns None), that slot is skipped.
""" """
acquired = {} acquired = {}
@@ -315,10 +419,20 @@ class IngestorManager(BaseActivity):
async def unsubscribe_slot(self, slot: str): async def unsubscribe_slot(self, slot: str):
""" """
Unsubscribes a specific slot from all associated OPC servers. Unsubscribes a specific slot from all associated OPC servers.
This method removes all subscriptions for a given slot across all
OPC servers that were managing it. It ensures clean cleanup of
resources when slots are released or reconfigured.
Args: Args:
slot (str): The name of the slot to unsubscribe. slot (str): The name of the slot to unsubscribe.
Raises: Raises:
KeyError: If the specified slot does not exist in the managed tags. KeyError: If the specified slot does not exist in the managed tags.
Side Effects:
- Removes subscriptions from all OPC servers for the specified slot
- Cleans up subscription resources on the OPC servers
""" """
for server in self.managed_tags[slot].keys(): for server in self.managed_tags[slot].keys():
@@ -329,6 +443,7 @@ class IngestorManager(BaseActivity):
""" """
Updates the configuration of managed slots by renewing their leases, Updates the configuration of managed slots by renewing their leases,
fetching the latest configurations, and handling any changes or removals. fetching the latest configurations, and handling any changes or removals.
This method performs the following steps: This method performs the following steps:
1. Renews the lease for each managed slot using the resource manager. 1. Renews the lease for each managed slot using the resource manager.
2. Fetches the latest configuration for each slot. 2. Fetches the latest configuration for each slot.
@@ -337,15 +452,18 @@ class IngestorManager(BaseActivity):
5. Unsubscribes and re-subscribes to slots with updated configurations. 5. Unsubscribes and re-subscribes to slots with updated configurations.
6. Removes slots from the managed tags if they are no longer valid. 6. Removes slots from the managed tags if they are no longer valid.
7. Updates the OPC servers after processing all slots. 7. Updates the OPC servers after processing all slots.
Side Effects: Side Effects:
- Modifies the `managed_tags` dictionary to reflect the latest slot configurations. - Modifies the `managed_tags` dictionary to reflect the latest slot configurations.
- Updates OPC server subscriptions based on the current state of managed slots. - Updates OPC server subscriptions based on the current state of managed slots.
Raises: Raises:
- None explicitly, but relies on the behavior of `resource_manager` and None explicitly, but relies on the behavior of `resource_manager` and
other dependencies for error handling. other dependencies for error handling.
Logging: Logging:
- Logs warnings for removed slots. - Logs warnings for removed slots.
- Logs informational messages for updated slot configurations. - Logs informational messages for updated slot configurations.
""" """
removed_slots = [] removed_slots = []
@@ -365,14 +483,22 @@ class IngestorManager(BaseActivity):
def drop_slot_leases(self, ids: List[str]) -> None: def drop_slot_leases(self, ids: List[str]) -> None:
""" """
Releases the leases associated with the specified slot IDs. Releases the leases associated with the specified slot IDs.
This method iterates through a list of slot IDs and calls the This method iterates through a list of slot IDs and calls the
`drop_tag_lease` method of the `resource_manager` to release `drop_tag_lease` method of the `resource_manager` to release
the lease for each ID. the lease for each ID. It's used during load balancing and
graceful shutdown scenarios.
Args: Args:
ids (List[str]): A list of slot IDs for which the leases ids (List[str]): A list of slot IDs for which the leases
should be released. should be released.
Returns: Returns:
None None
Side Effects:
- Releases Redis-based leases for specified slots
- Updates metrics for released slots count
""" """
for lease_id in ids: for lease_id in ids:
self.resource_manager.drop_tag_lease(lease_id) self.resource_manager.drop_tag_lease(lease_id)
@@ -381,28 +507,38 @@ class IngestorManager(BaseActivity):
async def manage_server(self, slot: str, server: str, server_config: dict, tags: dict) -> int: async 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. 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 This method ensures that the specified server and slot have an active subscription
for the provided tags. If the server or slot is not properly configured, or if for the provided tags. If the server or slot is not properly configured, or if
subscription fails, appropriate error handling is performed. subscription fails, appropriate error handling is performed.
Args: Args:
slot (str): The slot identifier for the subscription. slot (str): The slot identifier for the subscription.
server (str): The name of the OPC server. server (str): The name of the OPC server.
server_config (dict): Configuration dictionary for the server, which includes server_config (dict): Configuration dictionary for the server, which includes
the tags to be subscribed under the key 'tags'. the tags to be subscribed under the key 'tags'.
tags (dict): A dictionary of tags to be subscribed. tags (dict): A dictionary of tags to be subscribed.
Returns: Returns:
int: Status code indicating the result of the operation: int: Status code indicating the result of the operation:
- 0: Subscription was successful. - 0: Subscription was successful.
- 1: Server not found in `opc_managers`. - 1: Server not found in `opc_managers`.
- 2: Subscription creation or tag subscription failed. - 2: Subscription creation or tag subscription failed.
Logs: Logs:
- Logs informational messages about the subscription process. - Logs informational messages about the subscription process.
- Logs errors if the server is not found, subscription creation fails, or - Logs errors if the server is not found, subscription creation fails, or
tag subscription fails. tag subscription fails.
- Logs a warning if a subscription is removed due to failure. - Logs a warning if a subscription is removed due to failure.
Raises: Raises:
Exception: Any unexpected exceptions during subscription creation or tag Exception: Any unexpected exceptions during subscription creation or tag
subscription are logged but not propagated. subscription are logged but not propagated.
Side Effects:
- Creates or updates OPC subscriptions
- Manages tag subscriptions on OPC servers
- Updates error metrics and notifications
""" """
self.logger.info( self.logger.info(
@@ -459,19 +595,27 @@ class IngestorManager(BaseActivity):
async def subscribe_to_tags(self, tags: Dict) -> None: async def subscribe_to_tags(self, tags: Dict) -> None:
""" """
Subscribes to a set of tags and manages their configurations. Subscribes to a set of tags and manages their configurations.
This method processes a dictionary of tags, iterating through each slot and server This method processes a dictionary of tags, iterating through each slot and server
configuration. It attempts to manage the server configurations and removes any configuration. It attempts to manage the server configurations and removes any
servers that return a specific response code. servers that return a specific response code.
Args: Args:
tags (Dict): A dictionary containing tag configurations. The structure is tags (Dict): A dictionary containing tag configurations. The structure is
expected to be {slot: {server: server_config}}. expected to be {slot: {server: server_config}}.
Side Effects: Side Effects:
- Logs the provided tags for debugging purposes. - Logs the provided tags for debugging purposes.
- Updates the `managed_tags` attribute by removing servers that meet the - Updates the `managed_tags` attribute by removing servers that meet the
removal criteria. removal criteria.
- Establishes OPC subscriptions for all configured tags.
Removal Criteria: Removal Criteria:
- If the `manage_server` method returns a response code of 2 for a given - If the `manage_server` method returns a response code of 2 for a given
slot and server, that server is removed from the `managed_tags` attribute. slot and server, that server is removed from the `managed_tags` attribute.
The method ensures that only successfully configured servers remain in the
managed tags, maintaining system stability and preventing subscription errors.
""" """
to_remove = [] to_remove = []

150
ingestor/managers/opc_manager.py
View File

@@ -13,6 +13,47 @@ import ingestor.metrics as metrics
class OpcManager(BaseActivity): 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, def __init__(self, name: str, url: str, data_manager: DataManager, logger: Logger,
server_uri: str, notification_handler: NotificationHandler, metadata: dict, server_uri: str, notification_handler: NotificationHandler, metadata: dict,
cert_path: str = None, private_key_path: str = None, server_cert_path: str = None): 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) pod_id=self.pod_id, server_name=self.name).set(0)
def __str__(self): 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" \ return f"OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n" \
f"nodes={self.nodes}, subscriptions={self.subscriptions}" f"nodes={self.nodes}, subscriptions={self.subscriptions}"
async def shutdown(self): 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: try:
await self.disconnect() await self.disconnect()
@@ -54,21 +111,24 @@ class OpcManager(BaseActivity):
async def set_security(self): async def set_security(self):
""" """
Configures the security settings for the OPC UA client. Configures the security settings for the OPC UA client.
This method sets up the security policy, certificates, and timeouts This method sets up the security policy, certificates, and timeouts
required for establishing a secure connection with the OPC UA server. required for establishing a secure connection with the OPC UA server.
It implements Basic256 security policy with certificate-based authentication.
Raises: Raises:
ValueError: If either the certificate path or private key path is not provided. 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 Settings:
- Security Policy: Basic256 - Security Policy: Basic256
- Secure Channel Timeout: 10,000,000 ms - Secure Channel Timeout: 10,000,000 ms
- Session 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]): if not all([self.cert_path, self.private_key_path]):
@@ -93,11 +153,23 @@ class OpcManager(BaseActivity):
async def connect(self): async def connect(self):
""" """
Establishes a connection to the OPC server. Establishes a connection to the OPC server.
This method initializes the OPC client using the provided URL and This method initializes the OPC client using the provided URL and
sets up security if a certificate path is specified. It then sets up security if a certificate path is specified. It then
attempts to connect to the server and logs the connection status. 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: Raises:
Exception: If the connection to the OPC server fails. 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( metrics.OPC_CONNECTIONS_TOTAL.labels(
@@ -122,16 +194,22 @@ class OpcManager(BaseActivity):
async def create_subscription(self, name: str, period: int = 500): async def create_subscription(self, name: str, period: int = 500):
""" """
Creates a subscription with the specified monitoring period. Creates a subscription with the specified monitoring period.
This method establishes a subscription to monitor data changes or events 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. from the OPC UA server. If the client is not connected, an exception is raised.
Args: Args:
name (str): The name identifier for the subscription
period (int, optional): The monitoring period in milliseconds. Defaults to 500 ms. period (int, optional): The monitoring period in milliseconds. Defaults to 500 ms.
Raises: Raises:
ValueError: If the client is not connected. ValueError: If the client is not connected.
Side Effects: Side Effects:
- Sets the `self.period` attribute to the specified or default period. - Sets the `self.period` attribute to the specified or default period.
- Creates a subscription and assigns it to `self.subscription`. - Creates a subscription and assigns it to `self.subscriptions[name]`.
- Logs the creation of the subscription. - Logs the creation of the subscription.
- Increments subscription creation metrics.
""" """
if not self.client: if not self.client:
@@ -150,18 +228,26 @@ class OpcManager(BaseActivity):
async def subscribe(self, subscription: str, nodes: dict, collect_period: int): async def subscribe(self, subscription: str, nodes: dict, collect_period: int):
""" """
Subscribes to a set of OPC UA nodes for data change notifications. Subscribes to a set of OPC UA nodes for data change notifications.
This method adds the specified nodes to the subscription and configures This method adds the specified nodes to the subscription and configures
their data collection rules based on the provided collection period and their data collection rules based on the provided collection period and
node-specific frequency. node-specific frequency.
Args: Args:
nodes (dict): A dictionary where keys are node identifiers (e.g., node subscription (str): The name of the subscription to use
IDs or paths) and values are configurations for each node. Each nodes (dict): A dictionary where keys are node identifiers and values are
configuration must include a 'frequency' key indicating the configurations for each node. Each configuration must include a 'frequency'
frequency of data collection in Hz. key indicating the frequency of data collection in Hz.
collect_period (int): The data collection period in seconds. collect_period (int): The data collection period in seconds.
Raises: Raises:
ValueError: If the subscription has not been created by calling ValueError: If the subscription has not been created by calling
`create_subscription` prior to this method. `create_subscription` prior to this method.
Side Effects:
- Updates internal node tracking and cycle rules
- Establishes data change monitoring for specified nodes
- Updates metrics for subscribed tags count
""" """
if not self.subscriptions.get(subscription): if not self.subscriptions.get(subscription):
@@ -189,11 +275,17 @@ class OpcManager(BaseActivity):
async def unsubscribe(self, subscription: str): async def unsubscribe(self, subscription: str):
""" """
Unsubscribes from a given subscription. Unsubscribes from a given subscription.
This method removes the specified subscription and cleans up associated
resources. It handles cases where the subscription doesn't exist gracefully.
Args: Args:
subscription (str): The name of the subscription to unsubscribe from. subscription (str): The name of the subscription to unsubscribe from.
Logs: Logs:
- A warning if the specified subscription does not exist. - A warning if the specified subscription does not exist.
- An info message upon successful unsubscription. - An info message upon successful unsubscription.
Behavior: Behavior:
- If the subscription exists, it is deleted and removed from the - If the subscription exists, it is deleted and removed from the
subscriptions dictionary. subscriptions dictionary.
@@ -211,12 +303,20 @@ class OpcManager(BaseActivity):
async def disconnect(self): async def disconnect(self):
""" """
Disconnects from the OPC UA server. Disconnects from the OPC UA server.
This method handles the disconnection process by deleting the subscription
This method handles the disconnection process by deleting all subscriptions
and disconnecting the client from the OPC UA server. It logs the disconnection and disconnecting the client from the OPC UA server. It logs the disconnection
process and handles any exceptions that may occur during cleanup. process and handles any exceptions that may occur during cleanup.
Raises: Raises:
Exception: If an error occurs while deleting the subscription or disconnecting Exception: If an error occurs while deleting the subscription or disconnecting
from the OPC UA server, it logs the error details. from the OPC UA server, it logs the error details.
Side Effects:
- Deletes all active subscriptions
- Disconnects the OPC client
- Updates connection status metrics
- Clears internal client reference
""" """
self.logger.warning('Disconnecting from OPC server') self.logger.warning('Disconnecting from OPC server')
@@ -247,14 +347,17 @@ class OpcManager(BaseActivity):
async def datachange_notification(self, node, _val, data): async def datachange_notification(self, node, _val, data):
""" """
Handles data change notifications for monitored OPC UA nodes. Handles data change notifications for monitored OPC UA nodes.
This method is triggered when a monitored node's value changes. It processes This method is triggered when a monitored node's value changes. It processes
the notification, updates internal state, and publishes the data to the the notification, updates internal state, and publishes the data to the
appropriate topics. appropriate topics.
Args: Args:
node (NodeId): The OPC UA node that triggered the data change notification. node (NodeId): The OPC UA node that triggered the data change notification.
_val (Any): The new value of the node (unused in this implementation). _val (Any): The new value of the node (unused in this implementation).
data (DataChangeNotification): The data change notification object containing data (DataChangeNotification): The data change notification object containing
details about the change. details about the change.
Behavior: Behavior:
- Extracts the value and source timestamp from the monitored item. - Extracts the value and source timestamp from the monitored item.
- Resets the cycle count for the node's cycle rule. - Resets the cycle count for the node's cycle rule.
@@ -293,12 +396,15 @@ class OpcManager(BaseActivity):
def check_cycles(self): def check_cycles(self):
""" """
Checks the cycle counts for all monitored nodes and sends Checks the cycle counts for all monitored nodes and sends
notifications if thresholds are exceeded. notifications if thresholds are exceeded.
This method iterates through all monitored nodes and updates their cycle counts based on This method iterates through all monitored nodes and updates their cycle counts based on
configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers
a warning notification. a warning notification.
Side Effects:
- Updates cycle counts for all monitored nodes
- Sends warning notifications for nodes exceeding cycle thresholds
""" """
for node, config in self.nodes.items(): for node, config in self.nodes.items():
self.nodes[node]['cycle_rule']['cycle_count'] += config[ self.nodes[node]['cycle_rule']['cycle_count'] += config[
@@ -317,8 +423,20 @@ class OpcManager(BaseActivity):
def check_opc_listenning(self) -> bool: def check_opc_listenning(self) -> bool:
""" """
Checks the OPC connection and triggers notifications if the connection is lost. Checks the OPC connection and triggers notifications if the connection is lost.
This method monitors the data reception health by tracking cycles without
data. It sends notifications at different thresholds and can trigger
reconnection attempts.
Returns: Returns:
bool: True if the connection is lost, False otherwise. bool: True if the connection is lost and reconnection should be attempted,
False otherwise.
Side Effects:
- Increments non-receive count
- Updates metrics for cycles without data
- Sends warning notifications at 5 cycles
- Sends error notifications and triggers reconnection at 15 cycles
""" """
self.non_receive_count += 1 self.non_receive_count += 1

238
ingestor/managers/resource_manager.py
View File

@@ -10,6 +10,39 @@ from sientia_do.temporal.activities.base import BaseActivity
class ResourceManager(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__( def __init__(
self, self,
host: str, host: str,
@@ -22,6 +55,31 @@ class ResourceManager(BaseActivity):
username: str | None = None, username: str | None = None,
password: str | None = None, password: str | None = 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, BaseActivity.__init__(self, logger=logger,
notification_handler=notification_handler, notification_handler=notification_handler,
set_error_counter=True) set_error_counter=True)
@@ -45,7 +103,32 @@ class ResourceManager(BaseActivity):
self.metadata = metadata self.metadata = metadata
def _execute_redis_op(self, operation_name: str, func, *args, **kwargs): 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() start_time = time()
try: try:
result = func(*args, **kwargs) result = func(*args, **kwargs)
@@ -73,11 +156,20 @@ class ResourceManager(BaseActivity):
def get(self, key: str) -> dict: def get(self, key: str) -> dict:
""" """
Retrieve a value from Redis by its key and return it as a dictionary. 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: Args:
key (str): The key to look up in Redis. key (str): The key to look up in Redis.
Returns: Returns:
dict: The value associated with the key, parsed as a dictionary, dict: The value associated with the key, parsed as a dictionary,
or None if the key does not exist or the value is empty. 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) 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: def get_tag_slot(self, id: str) -> dict:
""" """
Retrieve the tag slot information for a given ID. 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: Args:
id (str): The unique identifier of the tag slot to retrieve. id (str): The unique identifier of the tag slot to retrieve.
Returns: 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}") return self.get(f"slot:opc_tags:{id}")
@@ -97,12 +198,20 @@ class ResourceManager(BaseActivity):
def ingestor_heartbeat(self) -> None: def ingestor_heartbeat(self) -> None:
""" """
Sends a heartbeat signal to Redis to indicate that the ingestor is active. 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 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 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 time defined by `self.heartbeat_ttl`. This allows monitoring systems to
track the activity and health of the ingestor. 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( self._execute_redis_op(
@@ -115,14 +224,28 @@ class ResourceManager(BaseActivity):
def lease_tag(self, tag_id: str) -> bool: 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). 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 This method uses the Redis `SET` command with the `NX` option to ensure that
defined by `lease_ttl`. 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: Args:
tag_id (str): The unique identifier of the tag to be leased. tag_id (str): The unique identifier of the tag to be leased.
Returns: 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( return self._execute_redis_op(
@@ -137,13 +260,26 @@ class ResourceManager(BaseActivity):
def renew_tag_lease(self, tag_id: str) -> bool: def renew_tag_lease(self, tag_id: str) -> bool:
""" """
Renews the lease for a specific OPC tag if the current pod holds the lease. 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 This method checks if the current pod (identified by `self.pod_id`) holds
in Redis to the configured lease TTL (`self.lease_ttl`). 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: Args:
tag_id (str): The identifier of the OPC tag whose lease is to be renewed. tag_id (str): The identifier of the OPC tag whose lease is to be renewed.
Returns: 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( current = self._execute_redis_op(
@@ -159,11 +295,22 @@ class ResourceManager(BaseActivity):
def drop_tag_lease(self, tag_id: str) -> None: def drop_tag_lease(self, tag_id: str) -> None:
""" """
Drops the lease for a specific OPC tag. 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: Args:
tag_id (str): The identifier of the OPC tag whose lease is to be dropped. 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, self._execute_redis_op("delete", self.redis.delete,
@@ -172,21 +319,47 @@ class ResourceManager(BaseActivity):
def get_all_ingestors(self) -> List[str]: def get_all_ingestors(self) -> List[str]:
""" """
Retrieves all active ingestors from Redis. 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: 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:*") return self._execute_redis_op("keys", self.redis.keys, "heartbeat:ingestor:*")
def get_all_slots(self) -> List[str]: def get_all_slots(self) -> List[str]:
""" """
Retrieves the number of slots available in Redis. Retrieves all available slots from Redis.
This method counts the number of keys in Redis that match the pattern for OPC tag leases
and returns the count. 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: 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:*") 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]: def get_all_leases(self) -> List[str]:
""" """
Retrieves all active leases from Redis. 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: 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:*") return self._execute_redis_op("keys", self.redis.keys, "lease:opc_tags:*")

18
ingestor/metrics.py
View File

@@ -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 from prometheus_client import Counter, Gauge, Histogram
# Metric label definitions for consistent labeling across all metrics
POD_ID_LABEL = ["pod_id"] POD_ID_LABEL = ["pod_id"]
SERVER_LABELS = ["pod_id", "server_name", "server_url"] SERVER_LABELS = ["pod_id", "server_name", "server_url"]
KAFKA_LABELS = ["pod_id", "topic"] KAFKA_LABELS = ["pod_id", "topic"]
@@ -15,7 +32,6 @@ TAG_WRITTEN_COUNT = Counter(
[*MAIN_LABELS, "tag_name", "collection_name"], [*MAIN_LABELS, "tag_name", "collection_name"],
) )
# --- General Application Metrics --- # --- General Application Metrics ---
APP_LOOP_COUNT = Counter( APP_LOOP_COUNT = Counter(
"app_main_loop_total", "app_main_loop_total",

30
simulator/Dockerfile
View File

@@ -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"]

60
simulator/redis-feeder.py
View File

@@ -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.")