Merge branch 'feature/SIENTIAPDE-1972-gateway-foundations' into feature/SIENTIAPDE-1972-gateway-write-read

.github/workflows/deploy.yml

@@ -0,0 +1,29 @@
+name: Deploy DataOps Worker
+
+on:
+  pull_request:
+    types: [closed]
+    branches: [main, 'release/**', 'feature/**']
+  workflow_dispatch:
+
+permissions:
+  contents: write
+  id-token: write
+  pull-requests: read
+
+jobs:
+  deploy:
+    name: Deploy
+    if: github.event_name == 'workflow_dispatch' || github.event.pull_request.merged == true
+    uses: Aignosi/github_workflow_templates/.github/workflows/reusable-deploy-dataops.yml@main
+    with:
+      branch_name: ${{ github.event.pull_request.head.ref }}
+      project_name: 'ingestor'
+      chart_name: 'opc-ingestor'
+      env_list_path: '[".env"]'
+      update_version_in: '["pyproject"]'
+      create_tarball: ${{ github.event_name == 'pull_request' }}
+      mirror: true
+      bump_chart: true
+      deploy: true
+    secrets: inherit

.github/workflows/quality-gate.yml

@@ -1,16 +1,31 @@
 name: Quality gate
 
 on:
+  push:
+    branches:
+      - main
+      - 'release/**'
+      - 'feature/**'
   pull_request:
     branches:
       - main
+      - 'release/**'
+      - 'feature/**'
     types: [ opened, synchronize, reopened ]
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
   quality-gate:
-    uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-quality-gate.yml@main
+    uses: Aignosi/github_workflow_templates/.github/workflows/python-quality-gate.yml@main
-    permissions: write-all
+    permissions:
+      contents: read
+      pull-requests: write
+      issues: write
     with:
       project_name: 'ingestor'
       repositories: 'sientia-dataops-library'
+      requirements_file: 'requirements.txt'
     secrets: inherit

.github/workflows/release.yml

@@ -1,16 +0,0 @@
-name: Create Release on Merge to Main
-
-on:
-  pull_request:
-    types: [closed]
-    branches:
-      - main
-
-jobs:
-  release:
-    if: github.event.pull_request.merged == true
-    uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-release.yml@main
-    permissions: write-all
-    with:
-      project_name: 'ingestor'
-    secrets: inherit

.github/workflows/require-chart-ref-label.yml

@@ -0,0 +1,16 @@
+name: Require chart-ref label
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, labeled, unlabeled]
+    branches:
+      - main
+      - 'release/**'
+      - 'feature/**'
+
+permissions:
+  pull-requests: write
+
+jobs:
+  require-chart-ref-label:
+    uses: Aignosi/github_workflow_templates/.github/workflows/require-chart-ref-label.yml@main

CLAUDE.md

@@ -0,0 +1,74 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project
+
+OPC UA data ingestion service. Pulls tags from OPC UA servers, streams to Kafka (optional) and persists to MongoDB. Horizontally scaled via Redis-coordinated slot leasing — multiple pods share tag load without central orchestration.
+
+Depends on internal shared lib `sientia-dataops-library` (imported as `sientia_do`, pinned in `requirements.txt` via git+ssh) for logging, notifications, metrics controller, and Mongo/Redis repositories. Base classes/utilities from `sientia_do` won't be visible in this repo — check that lib's source if behavior there is unclear.
+
+## Commands
+
+```bash
+python3.11 -m venv venv && source ./venv/bin/activate
+pip install -r requirements.txt -r requirements-dev.txt
+
+./run_local.sh                          # loads .env, runs python -m ingestor.app
+./run_coverage.sh                       # pytest --cov=ingestor --cov-report=html, opens report
+
+pytest                                  # all tests
+pytest tests/unit                       # unit only
+pytest tests/functional                 # functional only (spins up real-ish clients, see conftest.py)
+pytest tests/unit/test_ingestor.py::TestClass::test_name   # single test
+pytest --cov=ingestor --cov-report=html # coverage report -> htmlcov/
+
+ruff check .                            # lint
+ruff format .                           # format
+mypy ingestor                           # type check
+bandit -r ingestor                      # security scan
+```
+
+External deps (Redis, MongoDB, Kafka) must be reachable — via `kubectl port-forward`, local docker-compose, or cloud services. Config lives in `.env` (copy from `.env.example`).
+
+## Architecture
+
+Manager-based pipeline, one manager per concern, composed by `IngestorManager`:
+
+```
+app.py (asyncio loop, signals, Prometheus server)
+  -> Ingestor (ingestor.py) — lifecycle: prepare_ingestor() -> loop() -> shutdown()
+       -> IngestorManager (managers/ingestor_manager.py) — central coordinator
+            -> ResourceManager (managers/resource_manager.py) — Redis slot leases, heartbeats, active-ingestor registry
+            -> OpcManager (managers/opc_manager.py), one instance per OPC server — connection, subscriptions, datachange callbacks
+            -> DataManager (managers/data_manager.py) — MongoDB persistence + optional Kafka publish
+```
+
+**Slot model**: OPC tag configs live in Redis under keys like `slot:opc_tags:<n>`, each holding one or more OPC servers and their tags (see README "OPC Server Configuration" for the JSON shape). A "slot" is the unit of lease/assignment; each ingestor pod leases some number of slots via `ResourceManager` and only subscribes to the tags in slots it holds.
+
+**Main loop** (`Ingestor.loop()` in `ingestor.py`) each cycle:
+1. `declare_active()` + heartbeat (keeps this pod visible to peers via Redis TTL keys — `LEASE_TTL`/`HEARTBEAT_TTL`)
+2. Reads current active-ingestor count, lease count, slot count
+3. `manage_no_slots()` — grabs one slot if this pod is idle and slots exist
+4. `manage_leases()` — rebalances: acquires slots if other pods are lacking, drops extras (keeps at most 1 slot per pod when supply is sufficient) — this is the load-balancing algorithm
+5. `update_slot_config()` + `check_opc_servers_integrity()` — refresh OPC configs, detect stalled connections
+6. `update_ingestor_manager()` — diffs current vs previous managed tags, subscribes new, resubscribes changed, unsubscribes removed
+
+Any change to slot/lease counts should be traced through `manage_leases`/`manage_no_slots` — that's where the rebalancing math lives, not in `ResourceManager` itself (which just does Redis primitives).
+
+**OpcManager** owns one OPC UA `Client` connection (asyncua), handles security policy setup (cert/key-based, `SecurityPolicyBasic256`), subscriptions per tag group, and `datachange_notification` callbacks that push into `data_queue` for `DataManager` to consume. Tracks `non_receive_count` to detect silently-dead subscriptions (`check_cycles`).
+
+**Metrics** (`ingestor/metrics.py`): all Prometheus metrics defined here, always labeled by `pod_id`. Emitted via `SientiaMonitoring.emit_metric(...)` (from `sientia_do`), not the prometheus client directly — new metrics should follow that pattern for consistency with logging/notifications.
+
+**Error handling convention**: broad `except Exception` + `traceback.print_exc()` + notification-handler alert is intentional throughout (see `ruff.lint.ignore` for `BLE001` in `pyproject.toml`) — this is an always-on service where a single tag/server failure must not kill the pod; failures are surfaced via notifications/metrics instead of raised.
+
+## Config
+
+Env vars documented in README ("Configuration" section) and `.env.example`. Notable ones affecting behavior: `LEASE_TTL`/`HEARTBEAT_TTL` (failover speed vs stability), `POLL_INTERVAL` (loop cadence), `EXPORT_TO_KAFKA`.
+
+## Testing notes
+
+- `tests/unit/managers/` mirrors `ingestor/managers/` 1:1.
+- `tests/functional/` exercises more end-to-end paths against `conftest.py` fixtures.
+- Coverage config (`pyproject.toml` / `.coveragerc`) excludes `ingestor/app.py` (thin entrypoint, signal wiring) — don't chase coverage there.
+- `ruff` complexity cap is `max-complexity = 15` (mccabe) — factor out branches in manager methods before that.

README.md

@@ -7,7 +7,6 @@ A high-performance, scalable OPC UA data ingestion system designed for industria
 ### Core Functionality
 - **OPC UA Integration**: Native support for OPC UA servers with secure and unsecured connections
 - **Automatic Load Balancing**: Slot-based architecture for horizontal scaling across multiple instances
-- **Real-time Data Streaming**: Kafka integration for 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
@@ -36,9 +35,9 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
                        ┌──────────────────┐    ┌─────────────────┐
                        │  ResourceManager │    │  DataManager    │
                        │                  │    │                 │
-                       │ - Redis Coord.   │    │ - Kafka Export  │
+                       │ - Redis Coord.   │    │ - MongoDB Store │
-                       │ - Slot Leasing   │    │ - MongoDB Store │
+                       │ - Slot Leasing   │    │ - Data Pipeline │
-                       │ - Heartbeats     │    │ - Data Pipeline │
+                       │ - Heartbeats     │    │                 │
                        └──────────────────┘    └─────────────────┘
 ```
 
@@ -47,7 +46,7 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
 - **Ingestor**: Main application orchestrator managing the overall lifecycle
 - **IngestorManager**: Coordinates slot allocation, OPC server management, and load balancing
 - **OPC Manager**: Handles individual OPC UA server connections and tag subscriptions
-- **Data Manager**: Manages data streaming (MongoDB | Kafka)
+- **Data Manager**: Manages data persistence (MongoDB)
 - **Resource Manager**: Coordinates resource allocation and instance coordination via Redis
 
 ## 📋 Prerequisites
@@ -55,10 +54,9 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
 - 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:
+**Note**: External dependencies (Redis, MongoDB) must be available either through:
 - Port forwarding from a Kubernetes cluster
 - External Docker Compose setup
 - Cloud-managed services
@@ -99,12 +97,15 @@ The OPC Ingestor uses a modular, manager-based architecture designed for scalabi
    # 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
    ```
 
+6. **Generate an OPC UA client certificate** (only if connecting to a secured OPC UA server)
+
+   See [Certificate Generation](#-certificate-generation) below.
+
 ##  Usage
 
 ### Running the Ingestor
@@ -155,7 +156,7 @@ pytest
 
 # Run with coverage
 pytest --cov=ingestor --cov-report=html
-
+```
 
 ### Populating Redis with OPC Configuration
 
@@ -171,6 +172,38 @@ pytest --cov=ingestor --cov-report=html
 
 The feeder creates sample OPC tag configurations in Redis that the ingestor can discover and manage.
 
+## 🔐 Certificate Generation
+
+OPC UA servers that require secure connections (e.g. KEPServer with `Basic256` policy) validate the ingestor via an X.509 client certificate. Use `scripts/generate-and-push-opc-cert.sh` to generate that certificate and push it to the Gitea repo the pod clones on boot.
+
+```bash
+GITEA_PASSWORD='<senha-do-gitea>' ./scripts/generate-and-push-opc-cert.sh
+```
+
+Environment variables (all optional except `GITEA_PASSWORD`):
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `GITEA_URL` | Gitea in-cluster URL | `https://git.sientia.ai` |
+| `GITEA_USER` | Gitea user | `gitea_admin` |
+| `GITEA_PASSWORD` | Gitea password (**required**) | — |
+| `GITEA_REPO` | Repo the ingestor pod clones | `gitea_admin/sientia-dataops-opc-ingestor` |
+| `BRANCH` | Branch to push the cert to | `main` |
+| `APP_URI` | SAN URI of the cert — **must match** the `uri` field of the corresponding document in the `OPC_servers` Mongo collection, since the ingestor uses it as `application_uri` (KEPServer rejects with `BadCertificateUriInvalid` on mismatch) | `urn:sientia:opc-ingestor` |
+| `CN` | Certificate common name | `sientia-opc-ingestor` |
+| `DAYS` | Certificate validity (days) | `3650` |
+| `FORCE` | Set to `1` to overwrite an existing cert already pushed to the repo | unset |
+
+What the script does:
+1. Generates a self-signed RSA-2048 X.509 cert + key with `openssl`, with the SAN URI, `keyUsage`, and `extendedKeyUsage` extensions OPC UA / KEPServer require.
+2. Clones the ingestor's Gitea repo, copies the cert/key (`.pem`) and a `.der` copy (for manual import) into `certs/`, commits, and pushes.
+3. Prints the paths the pod will see after cloning (`/app/certs/opc-ingestor-cert.pem`, `/app/certs/opc-ingestor-key.pem`) and the next manual steps.
+
+After pushing, you still need to:
+- Set `cert_path` / `private_key_path` (and `uri`) on the server's document in the `OPC_servers` collection so `OpcManager` picks them up (see [OPC Server Configuration](#opc-server-configuration)).
+- Restart the ingestor deployment so it re-clones the repo: `kubectl -n sientia rollout restart deployment/opc-ingestor`.
+- On first connection, the server rejects the new cert — trust it in the OPC UA server's console (e.g. KEPServer: *OPC UA Configuration Manager → Trusted Clients → Trust*) and restart its runtime.
+
 ## 🧪 Testing
 
 ### Unit Tests
@@ -215,11 +248,9 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 - `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
@@ -228,8 +259,6 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 
 | 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 |
@@ -246,7 +275,7 @@ The OPC Ingestor exposes comprehensive Prometheus metrics for monitoring:
 
 ### OPC Server Configuration
 
-OPC servers are configured through Redis with the following structure:
+OPC servers are configured through Redis with the following structure. `cert_path` / `private_key_path` are only required for secured connections (see [Certificate Generation](#-certificate-generation)):
 
 ```json
 {
@@ -256,6 +285,8 @@ OPC servers are configured through Redis with the following structure:
       "server_id": "1",
       "url": "opc.tcp://localhost:4841",
       "server_uri": "http://opcua-server.simulator",
+      "cert_path": "/app/certs/opc-ingestor-cert.pem",
+      "private_key_path": "/app/certs/opc-ingestor-key.pem",
       "tags": {
         "ns=2;i=2": {
             "aggr_func": "avg",
@@ -292,8 +323,8 @@ sientia-dataops-opc-ingestor/
 │   ├── ingestor.py         # Core ingestor logic
 │   └── metrics.py          # Prometheus metrics definitions
 ├── simulator/               # OPC simulation and testing tools
+├── scripts/                 # Ops scripts (e.g. OPC UA cert generation)
 ├── tests/                   # Test suite
-├── docker-compose.yaml      # Infrastructure services
 └── requirements.txt         # Python dependencies
 ```
 
@@ -319,12 +350,7 @@ sientia-dataops-opc-ingestor/
    - Check authentication credentials
    - Ensure proper network configuration
 
-3. **Kafka Export Failures**
+3. **Performance Issues**
-   - 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
@@ -350,7 +376,7 @@ export LOG_LEVEL=DEBUG
 - **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
+- **Database Performance**: Optimize MongoDB indexes
 
 ## 🤝 Contributing
 
@@ -368,10 +394,6 @@ export LOG_LEVEL=DEBUG
 - 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:
@@ -383,14 +405,3 @@ For support and questions:
 ---
 
 **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.

ingestor/ingestor.py

@@ -19,7 +19,7 @@ class Ingestor(SientiaMonitoring):
     - 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
-    - Persisting data to MongoDB
+    - Distributing data to MongoDB
     - Providing health monitoring and metrics collection
 
     The ingestor uses a slot-based architecture where each slot represents
@@ -146,7 +146,7 @@ class Ingestor(SientiaMonitoring):
         else:
             # Subscribe to acquired slots
             if self.ingestor_manager:
-                self.ingestor_manager.update_opc_servers()
+                await self.ingestor_manager.update_opc_servers()
                 await self.ingestor_manager.subscribe_to_tags(acquired)
 
     async def prepare_ingestor(self):

ingestor/managers/data_manager.py

@@ -48,9 +48,6 @@ class DataManager(SientiaMonitoring):
         """
         Initializes the DataManager instance with a MongoDB connection.
 
-        This constructor establishes a connection to MongoDB and verifies server
-        availability.
-
         Args:
             mongo_connection_string (str): MongoDB connection string
             mongo_database (str): MongoDB database name
@@ -90,10 +87,6 @@ class DataManager(SientiaMonitoring):
     def shutdown(self):
         """
         Gracefully shuts down the DataManager and closes all connections.
-
-        This method ensures proper cleanup of the MongoDB client connection,
-        logging any errors that occur during shutdown while ensuring all
-        resources are properly released.
         """
         try:
             self.mongo_repository.close()
@@ -105,11 +98,11 @@ class DataManager(SientiaMonitoring):
 
     async def publish(self, topic: str, data: dict) -> None:
         """
-        Persists a message to the specified MongoDB collection.
+        Persists a message to MongoDB.
 
         Args:
-            topic (str): The name of the MongoDB collection to persist to.
+            topic (str): The name of the MongoDB collection to which the message will be written.
-            data (dict): The message data to be inserted.
+            data (dict): The message data to be stored.
 
         Returns:
             None

ingestor/managers/ingestor_manager.py

@@ -1,4 +1,3 @@
-import asyncio
 import traceback
 from copy import deepcopy
 
@@ -188,9 +187,6 @@ class IngestorManager(SientiaMonitoring):
             await server.shutdown()
         self.data_manager.shutdown()
 
-    def __del__(self):
-        asyncio.run(self.shutdown())
-
     async def remove_server(self, server: str):
         """
         Removes an OPC server from the ingestor.

ingestor/managers/opc_manager.py

@@ -112,9 +112,6 @@ class OpcManager(SientiaMonitoring):
             f'nodes={self.nodes}, subscriptions={self.subscriptions}'
         )
 
-    def __del__(self):
-        asyncio.run(self.shutdown())
-
     async def shutdown(self):
         """
         Comprehensive cleanup method for graceful shutdown.
@@ -162,6 +159,7 @@ class OpcManager(SientiaMonitoring):
         cert = str(Path(self.cert_path)) if self.cert_path else None
         private_key = str(Path(self.private_key_path)) if self.private_key_path else None
         server_cert = str(Path(self.server_cert_path)) if self.server_cert_path else None
+        assert cert is not None and private_key is not None
 
         if self.client:
             self.client.application_uri = self.server_uri
@@ -209,7 +207,7 @@ class OpcManager(SientiaMonitoring):
             assert self.client is not None  # Informa ao mypy que client não é None
 
             self.client.name = self.pod_id
-            self.client.application_name = self.pod_id
+            self.client.description = self.pod_id
             pod_uri = self.pod_id.replace('-', ':')
             self.client.application_uri = pod_uri
             self.client.product_uri = pod_uri

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ingestor"
-version = "0.0.0"
+version = "1.4.4"
 description = "Sientia DataOps Ingestor - OPC Tag Ingestor"
 readme = "README.md"
 requires-python = ">=3.11"

requirements-local.txt

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

requirements.txt

@@ -1,9 +1,9 @@
-asyncua==1.1.5
+asyncua
 redis
-# TODO(SIENTIAPDE-1972): bump to the sientia-dataops-library release that includes
+# TODO(SIENTIAPDE-1972): bump once sientia-dataops-library releases SIENTIAPDE-1590
-# SIENTIAPDE-1590 (generic Mongo find_one/distinct/aggregate methods) once it is tagged.
+# (generic Mongo find_one/distinct/aggregate methods) — not in 1.12.2 yet.
 # Local dev uses an editable install of the local checkout (feature/SIENTIAPDE-1590).
-git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3
+sientia_do==1.12.2
 prometheus_client
 pymongo
 nats-py>=2.6

scripts/generate-and-push-opc-cert.sh

@@ -0,0 +1,95 @@
+#!/usr/bin/env bash
+# Gera o certificado X.509 de cliente OPC UA do ingestor e sobe pro Gitea
+# in-cluster (repo que o pod clona no boot — nada muda no values/deployment,
+# só rollout restart depois do push).
+#
+# Uso:
+#   GITEA_PASSWORD='<senha>' ./scripts/generate-and-push-opc-cert.sh
+#
+# Vars (defaults pro cluster dev/gcp):
+#   GITEA_URL      default https://git.sientia.ai
+#   GITEA_USER     default gitea_admin
+#   GITEA_PASSWORD obrigatória
+#   GITEA_REPO     default gitea_admin/sientia-dataops-opc-ingestor
+#   BRANCH         default main
+#   APP_URI        default urn:sientia:opc-ingestor  (SAN URI do cert — TEM que
+#                  ser igual ao campo `uri` do documento na collection
+#                  OPC_servers: o ingestor usa esse valor como application_uri
+#                  e o KEPServer rejeita com BadCertificateUriInvalid se divergir)
+#   DAYS           default 3650
+#   FORCE=1        sobrescreve cert já existente no repo
+set -euo pipefail
+
+GITEA_URL=${GITEA_URL:-https://git.sientia.ai}
+GITEA_USER=${GITEA_USER:-gitea_admin}
+GITEA_PASSWORD=${GITEA_PASSWORD:?export GITEA_PASSWORD com a senha do Gitea}
+GITEA_REPO=${GITEA_REPO:-gitea_admin/sientia-dataops-opc-ingestor}
+BRANCH=${BRANCH:-main}
+APP_URI=${APP_URI:-urn:sientia:opc-ingestor}
+CN=${CN:-sientia-opc-ingestor}
+DAYS=${DAYS:-3650}
+
+CERT_DIR_IN_REPO="certs"
+CERT_NAME="opc-ingestor-cert.pem"
+KEY_NAME="opc-ingestor-key.pem"
+DER_NAME="opc-ingestor-cert.der"
+
+workdir=$(mktemp -d)
+trap 'rm -rf "$workdir"' EXIT
+
+echo ">> Gerando certificado (CN=${CN}, URI=${APP_URI}, ${DAYS} dias)..."
+# Extensões exigidas por OPC UA (KEPServer valida keyUsage/EKU e o SAN URI).
+openssl req -x509 -newkey rsa:2048 -sha256 -nodes -days "$DAYS" \
+  -keyout "$workdir/$KEY_NAME" -out "$workdir/$CERT_NAME" \
+  -subj "/CN=${CN}/O=Aignosi/OU=Sientia" \
+  -addext "subjectAltName=URI:${APP_URI},DNS:${CN}" \
+  -addext "keyUsage=critical,digitalSignature,nonRepudiation,keyEncipherment,dataEncipherment" \
+  -addext "extendedKeyUsage=critical,clientAuth,serverAuth" \
+  -addext "basicConstraints=critical,CA:FALSE"
+
+# Cópia DER — só referência p/ import manual no KEPServer (o trust normal é
+# aceitar o cert rejeitado na 1ª conexão).
+openssl x509 -in "$workdir/$CERT_NAME" -outform der -out "$workdir/$DER_NAME"
+
+echo ">> Clonando ${GITEA_REPO}@${BRANCH} do Gitea..."
+encoded_pass=$(python3 - <<EOF
+import urllib.parse; print(urllib.parse.quote('''${GITEA_PASSWORD}''', safe=''))
+EOF
+)
+clone_url="${GITEA_URL%%/}"
+clone_url="${clone_url/:\/\//:\/\/${GITEA_USER}:${encoded_pass}@}/${GITEA_REPO}.git"
+git clone --depth 1 --branch "$BRANCH" "$clone_url" "$workdir/repo" 2>&1 | sed "s#${encoded_pass}#***#g"
+
+if [[ -f "$workdir/repo/$CERT_DIR_IN_REPO/$CERT_NAME" && "${FORCE:-0}" != "1" ]]; then
+  echo "ERRO: $CERT_DIR_IN_REPO/$CERT_NAME já existe no repo. Re-rodar com FORCE=1 sobrescreve" >&2
+  echo "      (o cert antigo deixa de valer — retrust no KEPServer necessário)." >&2
+  exit 1
+fi
+
+mkdir -p "$workdir/repo/$CERT_DIR_IN_REPO"
+cp "$workdir/$CERT_NAME" "$workdir/$KEY_NAME" "$workdir/$DER_NAME" "$workdir/repo/$CERT_DIR_IN_REPO/"
+
+cd "$workdir/repo"
+git add "$CERT_DIR_IN_REPO"
+git -c user.name="opc-cert-script" -c user.email="platform@aignosi.com.br" \
+  commit -m "Add OPC UA client certificate for KEPServer connection (URI ${APP_URI})"
+git push origin "$BRANCH" 2>&1 | sed "s#${encoded_pass}#***#g"
+
+fingerprint=$(openssl x509 -in "$CERT_DIR_IN_REPO/$CERT_NAME" -noout -fingerprint -sha1)
+
+cat <<EOF
+
+== PRONTO ==
+Cert no Gitea: ${GITEA_REPO}@${BRANCH} -> ${CERT_DIR_IN_REPO}/{${CERT_NAME},${KEY_NAME},${DER_NAME}}
+${fingerprint}
+
+Paths dentro do pod (o entrypoint clona o repo em /app):
+  cert_path:        /app/${CERT_DIR_IN_REPO}/${CERT_NAME}
+  private_key_path: /app/${CERT_DIR_IN_REPO}/${KEY_NAME}
+
+Próximos passos:
+  1. Documento na collection OPC_servers com "uri": "${APP_URI}" (ver README/PR).
+  2. kubectl -n sientia rollout restart deployment/opc-ingestor   # re-clona o repo
+  3. Pod tenta conectar -> no KEPServer: OPC UA Configuration Manager ->
+     Trusted Clients -> cert '${CN}' rejeitado -> Trust -> restart runtime.
+EOF

sonar-project.properties

@@ -7,5 +7,4 @@ sonar.qualitygate.timeout=300
 sonar.python.coverage.reportPaths=coverage.xml
 sonar.python.xunit.reportPath=pytest.xml
 sonar.python.version=3.11
-sonar.projectVersion=1.2.0
 sonar.coverage.exclusions=ingestor/app.py

tests/unit/managers/test_data_manager.py

@@ -3,7 +3,6 @@ from unittest.mock import ANY, AsyncMock, MagicMock, patch
 from pytest import fixture, mark
 from sientia_do.notifications.models import NotificationLevel
 
-import ingestor.metrics as metrics
 from ingestor.managers.data_manager import DataManager
 
 metadata = {
@@ -48,21 +47,11 @@ def test___init___success(mongodb_repository):
         metrics_controller=MagicMock(),
     )
 
-    mongodb_repository.assert_called_once_with(
-        connection_string='mongodb://localhost:27017',
-        database_name='sientia',
-        logger=logger_mock,
-        notification_handler=ANY,
-        metrics_controller=ANY,
-    )
-    logger_mock.info.assert_any_call(
-        'Trying to initializing DataManager with MongoDB servers: mongodb://localhost:27017'
-    )
     logger_mock.info.assert_any_call(
         'DataManager initialized with MongoDB servers: mongodb://localhost:27017'
     )
     logger_mock.error.assert_not_called()
-    assert data_manager.mongo_repository is not None
+    assert data_manager.connection_string == 'mongodb://localhost:27017'
 
 
 def test_shutdown(data_manager):
@@ -70,7 +59,6 @@ def test_shutdown(data_manager):
     data_manager.mongo_repository.close = close_mock
 
     data_manager.shutdown()
-
     close_mock.assert_called_once()
 
 
@@ -102,14 +90,6 @@ async def test_publish(data_manager):
         document={**data, 'inserted_at': ANY},
         metadata=metadata['metadata'],
     )
-    data_manager.emit_metric.assert_called_once_with(
-        metric_object=metrics.TAG_WRITTEN_COUNT,
-        tags={
-            'pod_id': data_manager.pod_id,
-            'tag_name': data['name'],
-            'collection_name': topic,
-        },
-    )
 
 
 @mark.asyncio

tests/unit/managers/test_ingestor_manager.py

@@ -183,13 +183,6 @@ async def test_shutdown(ingestor_manager):
     ingestor_manager.data_manager.shutdown.assert_called_once()
 
 
-@patch('ingestor.managers.ingestor_manager.asyncio')
-def test___del__(asyncio_mock, ingestor_manager):
-    ingestor_manager.shutdown = MagicMock()
-    ingestor_manager.__del__()
-    asyncio_mock.run.assert_called_once_with(ingestor_manager.shutdown.return_value)
-
-
 @mark.asyncio
 async def test_remove_server(ingestor_manager):
     server1 = AsyncMock()

validate.sh

@@ -1,124 +0,0 @@
-#!/bin/bash
-# Model Manager Code Validation Script
-# This script runs all code quality checks before committing or deploying
-
-set -e  # Exit on any error
-
-# Colors for output
-RED='\033[0;31m'
-GREEN='\033[0;32m'
-YELLOW='\033[1;33m'
-BLUE='\033[0;34m'
-NC='\033[0m' # No Color
-
-# Args
-FIX_MODE=false
-while [[ $# -gt 0 ]]; do
-    case "$1" in
-        --fix)
-            FIX_MODE=true
-            shift
-            ;;
-        -h|--help)
-            echo "Usage: $0 [--fix]"
-            echo "  --fix   Apply Ruff auto-fixes (format and lint fixes)."
-            exit 0
-            ;;
-        *)
-            echo -e "${RED}Unknown option: $1${NC}"
-            echo "Usage: $0 [--fix]"
-            exit 2
-            ;;
-    esac
-done
-
-echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}"
-echo -e "${BLUE}║  Model Manager - Code Validation Suite                ║${NC}"
-echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}"
-echo ""
-
-# Check if virtual environment is activated
-if [[ -z "${VIRTUAL_ENV}" ]] && [[ -z "${CONDA_DEFAULT_ENV}" ]]; then
-    echo -e "${YELLOW}⚠️  Warning: No virtual environment detected${NC}"
-    echo -e "${YELLOW}   Consider activating your venv/conda environment${NC}"
-    echo ""
-fi
-
-# Function to run a validation step
-run_step() {
-    local step_name=$1
-    local step_command=$2
-    
-    echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
-    echo -e "${BLUE}▶ ${step_name}${NC}"
-    echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
-    
-    if eval "$step_command"; then
-        echo -e "${GREEN}✅ ${step_name} - PASSED${NC}"
-        echo ""
-        return 0
-    else
-        echo -e "${RED}❌ ${step_name} - FAILED${NC}"
-        echo ""
-        return 1
-    fi
-}
-
-# Track failures
-FAILED_STEPS=()
-
-# Step 1: Code Formatting Check (Ruff)
-# - default: check only
-# - --fix: write changes
-if ! run_step "1. Code Formatting (Ruff)" "if \$FIX_MODE; then ruff format ingestor/ tests/; else ruff format --check ingestor/ tests/; fi"; then
-    FAILED_STEPS+=("Code Formatting")
-fi
-
-# Step 2: Linting (Ruff)
-# - default: check only
-# - --fix: apply autofixes
-if ! run_step "2. Code Linting (Ruff)" "if \$FIX_MODE; then ruff check --fix ingestor/ tests/; else ruff check ingestor/ tests/; fi"; then
-    FAILED_STEPS+=("Linting")
-fi
-
-# Step 3: Type Checking (mypy)
-if ! run_step "3. Type Checking (mypy)" "mypy ingestor/"; then
-    FAILED_STEPS+=("Type Checking")
-fi
-
-# Step 4: Security Analysis (Bandit)
-if ! run_step "4. Security Analysis (Bandit)" "bandit -r ingestor/ -ll -q"; then
-    FAILED_STEPS+=("Security Analysis")
-fi
-
-# Step 5: Unit Tests (pytest)
-if ! run_step "5. Unit Tests (pytest)" "pytest tests/ --cov=ingestor --cov-report=term-missing --cov-report=xml --cov-report=html --cov-fail-under=80 -q"; then
-    FAILED_STEPS+=("Unit Tests")
-fi
-
-# Summary
-echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}"
-echo -e "${BLUE}║  Validation Summary                                    ║${NC}"
-echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}"
-echo ""
-
-if [ ${#FAILED_STEPS[@]} -eq 0 ]; then
-    echo -e "${GREEN}✅ All validation checks passed!${NC}"
-    echo -e "${GREEN}   Your code is ready for commit/deployment.${NC}"
-    echo ""
-    exit 0
-else
-    echo -e "${RED}❌ Validation failed for the following steps:${NC}"
-    for step in "${FAILED_STEPS[@]}"; do
-        echo -e "${RED}   • ${step}${NC}"
-    done
-    echo ""
-    echo -e "${YELLOW}💡 Tips:${NC}"
-    echo -e "${YELLOW}   • Run 'ruff format ingestor/ tests/' to auto-fix formatting${NC}"
-    echo -e "${YELLOW}   • Run 'ruff check --fix ingestor/ tests/' to auto-fix linting issues${NC}"
-    echo -e "${YELLOW}   • Review mypy errors and add type hints where needed${NC}"
-    echo -e "${YELLOW}   • Check bandit warnings for security issues${NC}"
-    echo -e "${YELLOW}   • Fix failing tests or improve test coverage${NC}"
-    echo ""
-    exit 1
-fi