From b9c5380690cc7f04401c67050b239f95b851f05d Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 11:49:28 -0300 Subject: [PATCH 01/16] SAM0123-231: Refactor Python dependencies to use published sientia_do package, introduce requirements-local.txt for local development and quality gates, and add script for automated OPC UA client certificate generation and deployment. --- .github/workflows/quality-gate.yml | 1 + requirements-local.txt | 5 ++ requirements.txt | 2 +- scripts/generate-and-push-opc-cert.sh | 95 +++++++++++++++++++++++++++ 4 files changed, 102 insertions(+), 1 deletion(-) create mode 100644 requirements-local.txt create mode 100755 scripts/generate-and-push-opc-cert.sh diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index a8f9b2a..6bcf476 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -13,4 +13,5 @@ jobs: with: project_name: 'ingestor' repositories: 'sientia-dataops-library' + requirements-file: 'requirements-local.txt' secrets: inherit \ No newline at end of file diff --git a/requirements-local.txt b/requirements-local.txt new file mode 100644 index 0000000..bb176bf --- /dev/null +++ b/requirements-local.txt @@ -0,0 +1,5 @@ +asyncua==1.1.5 +redis +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 +prometheus_client +pymongo \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index bb176bf..0959294 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ asyncua==1.1.5 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 +sientia_do==1.5.3 prometheus_client pymongo \ No newline at end of file diff --git a/scripts/generate-and-push-opc-cert.sh b/scripts/generate-and-push-opc-cert.sh new file mode 100755 index 0000000..713965e --- /dev/null +++ b/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='' ./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 - <&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 < ${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 From c209796ffb493a5a8b06545a68279ca05f75bbef Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 11:51:55 -0300 Subject: [PATCH 02/16] SAM0123-231: Update sientia-dataops-library dependency to version 1.12.1. --- requirements-local.txt | 2 +- requirements.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/requirements-local.txt b/requirements-local.txt index bb176bf..190e40b 100644 --- a/requirements-local.txt +++ b/requirements-local.txt @@ -1,5 +1,5 @@ asyncua==1.1.5 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1 prometheus_client pymongo \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 0959294..ed65ae3 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ asyncua==1.1.5 redis -sientia_do==1.5.3 +sientia_do==1.12.1 prometheus_client pymongo \ No newline at end of file From 675f30c60ebe4acc0fb2e97ce377975a097c47f9 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 11:58:26 -0300 Subject: [PATCH 03/16] SAM0123-231: Update GitHub Actions quality gate workflow to use python-quality-gate template and correct requirements file parameter name. --- .github/workflows/quality-gate.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 6bcf476..05565bd 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -8,10 +8,10 @@ on: 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 with: project_name: 'ingestor' repositories: 'sientia-dataops-library' - requirements-file: 'requirements-local.txt' + requirements_file: 'requirements-local.txt' secrets: inherit \ No newline at end of file From 0d9d7b2ebe207eaac0c31a8aa01a2a27dacfaac2 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 12:00:54 -0300 Subject: [PATCH 04/16] SAM0123-231: Update quality gate workflow to use requirements.txt for the ingestor project. --- .github/workflows/quality-gate.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 05565bd..2f3002f 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -13,5 +13,5 @@ jobs: with: project_name: 'ingestor' repositories: 'sientia-dataops-library' - requirements_file: 'requirements-local.txt' + requirements_file: 'requirements.txt' secrets: inherit \ No newline at end of file From c157d1649a8e395567ac29767896c33203145aa1 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 12:04:13 -0300 Subject: [PATCH 05/16] SAM0123-231: Broaden quality gate workflow to include release and feature branches; correct asynchronous call for OPC server updates in ingestor. --- .github/workflows/quality-gate.yml | 2 ++ ingestor/ingestor.py | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 2f3002f..61b07c5 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -4,6 +4,8 @@ on: pull_request: branches: - main + - 'release/**' + - 'feature/**' types: [ opened, synchronize, reopened ] jobs: diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 4b76888..4ddb063 100644 --- a/ingestor/ingestor.py +++ b/ingestor/ingestor.py @@ -156,7 +156,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): From b10c226e4e791704e1a840a01762163fd35f8116 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 12:25:29 -0300 Subject: [PATCH 06/16] SAM0123-231: Remove Kafka integration from Ingestor service. --- ingestor/ingestor.py | 14 +- ingestor/managers/data_manager.py | 160 +-------------- ingestor/managers/ingestor_manager.py | 10 +- ingestor/metrics.py | 17 -- tests/unit/managers/test_data_manager.py | 195 +------------------ tests/unit/managers/test_ingestor_manager.py | 6 - tests/unit/test_ingestor.py | 6 - tests/unit/test_metrics.py | 24 --- 8 files changed, 18 insertions(+), 414 deletions(-) diff --git a/ingestor/ingestor.py b/ingestor/ingestor.py index 4ddb063..2244d61 100644 --- a/ingestor/ingestor.py +++ b/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 - - Distributing data to Kafka and MongoDB + - Distributing data to MongoDB - Providing health monitoring and metrics collection The ingestor uses a slot-based architecture where each slot represents @@ -27,8 +27,6 @@ class Ingestor(SientiaMonitoring): 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) @@ -43,7 +41,6 @@ class Ingestor(SientiaMonitoring): 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) @@ -54,7 +51,6 @@ class Ingestor(SientiaMonitoring): 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 @@ -66,17 +62,12 @@ class Ingestor(SientiaMonitoring): Initializes the ingestor with configuration values retrieved from environment variables. Sets up all necessary connections and configurations for: - - Kafka connectivity (if enabled) - Redis for slot management and coordination - MongoDB for data persistence and notifications - OPC UA server management - Metrics collection and monitoring """ - kafka_servers = getenv('KAFKA_SERVERS', 'localhost:9092') - export_to_kafka: bool = getenv('EXPORT_TO_KAFKA', 'false') == 'true' - - self.export_to_kafka = export_to_kafka self.redis_host = getenv('REDIS_HOST', 'localhost') self.redis_port = int(getenv('REDIS_PORT', '6379')) self.redis_username = getenv('REDIS_USERNAME', None) @@ -91,7 +82,6 @@ class Ingestor(SientiaMonitoring): self.mongo_database = getenv('MONGODB_DATABASE', 'sientia') self.mongo_connection_string = f'mongodb://{mongo_username}:{mongo_password}@{mongo_url}' - self.kafka_servers = kafka_servers.split(',') self.logger = get_logger(__name__) self.notification_handler = NotificationHandler( connection_string=self.mongo_connection_string, @@ -179,7 +169,6 @@ class Ingestor(SientiaMonitoring): """ self.ingestor_manager = IngestorManager( - kafka_servers=','.join(self.kafka_servers), redis_data={ 'host': self.redis_host, 'port': self.redis_port, @@ -195,7 +184,6 @@ class Ingestor(SientiaMonitoring): logger=self.logger, notification_handler=self.notification_handler, metrics_controller=self.metrics_controller, - export_to_kafka=self.export_to_kafka, ) assert self.ingestor_manager is not None diff --git a/ingestor/managers/data_manager.py b/ingestor/managers/data_manager.py index 44d248e..ef3b5ac 100644 --- a/ingestor/managers/data_manager.py +++ b/ingestor/managers/data_manager.py @@ -1,10 +1,6 @@ -import json import os import traceback -from time import sleep -from kafka import KafkaProducer -from kafka.errors import NoBrokersAvailable from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.models import NotificationLevel from sientia_do.observability.logger import Logger @@ -18,31 +14,22 @@ import ingestor.metrics as metrics class DataManager(SientiaMonitoring): """ - Manages data persistence and export operations for the OPC Ingestor. + Manages data persistence 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 @@ -51,47 +38,25 @@ class DataManager(SientiaMonitoring): def __init__( self, - kafka_servers: str, mongo_connection_string: str, mongo_database: str, - export_to_kafka: bool, metadata: dict, logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, ) -> None: """ - Initializes the DataManager instance with Kafka and MongoDB connections. - - This constructor attempts to establish connections to the specified services: - 1. Kafka: Initializes producer with retry logic (up to 3 attempts) - 2. MongoDB: Establishes connection and verifies server availability - - The initialization process includes: - - Kafka producer setup with JSON serialization - - MongoDB client initialization and connection testing - - Metrics recording for connection status - - Error handling with notifications + Initializes the DataManager instance with a MongoDB connection. 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 metadata (dict): Application metadata logger (Logger): Logger instance notification_handler (NotificationHandler): Notification handler - - Raises: - NoBrokersAvailable: If the connection to Kafka servers fails after 3 attempts. - - Metrics: - - KAFKA_CONNECTION_STATUS: Set to 1 on successful connection, 0 on failure """ self.pod_id = os.getenv('HOSTNAME', 'localhost') - self.kafka_producer = None - self.export_to_kafka = export_to_kafka SientiaMonitoring.__init__( self, @@ -100,37 +65,6 @@ class DataManager(SientiaMonitoring): notification_handler=notification_handler, ) - if self.export_to_kafka: - for i in range(0, 3): - logger.info( - f'Trying ({i}) to initializing DataManager with Kafka servers: {kafka_servers}' - ) - try: - self.kafka_producer = KafkaProducer( - bootstrap_servers=kafka_servers, - value_serializer=lambda v: json.dumps(v).encode( - 'utf-8' - ), # Serialize JSON messages - key_serializer=lambda k: str(k).encode('utf-8') if k else None, - ) - # Kafka connected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(1) - break - except NoBrokersAvailable: - logger.error(f'Kafka servers {kafka_servers} are not available. Retrying...') - sleep(5) - else: - # Kafka not connected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0) - logger.error( - f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.' - ) - raise NoBrokersAvailable( - f'Failed to connect to Kafka servers {kafka_servers} after 3 attempts.' - ) - - logger.info(f'DataManager initialized with Kafka servers: {kafka_servers}') - logger.info( f'Trying to initializing DataManager with MongoDB servers: {mongo_connection_string}' ) @@ -153,26 +87,7 @@ class DataManager(SientiaMonitoring): def shutdown(self): """ 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: - try: - self.kafka_producer.flush(timeout=10) - self.kafka_producer.close() - # Mark as disconnected - metrics.KAFKA_CONNECTION_STATUS.labels(pod_id=self.pod_id).set(0) - except Exception as e: - self.logger.error(f'Error closing Kafka producer: {e}') - else: - self.logger.warning('Kafka producer is already closed or not initialized.') - try: self.mongo_repository.close() except Exception as e: @@ -181,83 +96,18 @@ class DataManager(SientiaMonitoring): def __del__(self): self.shutdown() - def delivery_report(self, msg): - """ - 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( - f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}' - ) - - def delivery_error(self, err): - """ - 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}') - async def publish(self, topic: str, data: dict) -> None: """ - Publishes a message to a specified Kafka topic. + Persists a message to MongoDB. Args: - topic (str): The name of the Kafka topic to which the message will be published. - data (dict): The message data to be sent to the Kafka topic. + topic (str): The name of the MongoDB collection to which the message will be written. + data (dict): The message data to be stored. Returns: None - - Raises: - Exception: If there is an error during message delivery, it will be handled by the `delivery_error` callback. """ - if self.export_to_kafka and self.kafka_producer: - try: - self.logger.debug(f'Publishing message to topic {topic}: {data}') - self.kafka_producer.send(topic=topic, value=data).add_callback( - self.delivery_report - ).add_errback(self.delivery_error) - - self.kafka_producer.flush(timeout=10) - await self.emit_metric( - metric_object=metrics.KAFKA_MESSAGES_SENT, - tags={ - 'pod_id': self.pod_id, - 'topic': topic, - }, - ) - - except Exception as e: - await self.emit_metric( - metric_object=metrics.KAFKA_MESSAGES_ERRORS, - tags={ - 'pod_id': self.pod_id, - 'topic': topic, - }, - ) - trace = traceback.format_exc() - await self.send_notification_async( - metadata=self.metadata, - notification_id=f'KAFKA_PRODUCER_ERROR_{topic}', - message=f'Error publishing message to topic {topic}: {e}', - block='kafka_producer', - level=NotificationLevel.ERROR, - attachment_content=trace, - ) - self.logger.error(trace) - try: await self.mongo_repository.insert( collection_name=topic, diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 030e305..34a09ca 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/ingestor/managers/ingestor_manager.py @@ -19,7 +19,7 @@ class IngestorManager(SientiaMonitoring): Central coordinator for managing OPC data ingestion operations. The IngestorManager orchestrates the interaction between different components: - - DataManager: Handles data persistence and Kafka export + - DataManager: Handles data persistence - OPC Managers: Manage individual OPC UA server connections - ResourceManager: Coordinates slot leasing and load balancing @@ -36,7 +36,6 @@ class IngestorManager(SientiaMonitoring): - 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 @@ -46,10 +45,9 @@ class IngestorManager(SientiaMonitoring): 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 + data_manager (DataManager): Manages data persistence 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 @@ -61,7 +59,6 @@ class IngestorManager(SientiaMonitoring): def __init__( self, - kafka_servers: str, redis_data: dict, lease_ttl: int, heartbeat_ttl: int, @@ -72,7 +69,6 @@ class IngestorManager(SientiaMonitoring): logger: Logger, notification_handler: NotificationHandler, metrics_controller: MetricsController, - export_to_kafka: bool = False, ): redis_host: str = redis_data['host'] redis_port: int = int(redis_data['port']) @@ -87,10 +83,8 @@ class IngestorManager(SientiaMonitoring): ) self.data_manager = DataManager( - kafka_servers=kafka_servers, mongo_connection_string=mongo_connection_string, mongo_database=mongo_database, - export_to_kafka=export_to_kafka, metadata=metadata, logger=logger, notification_handler=notification_handler, diff --git a/ingestor/metrics.py b/ingestor/metrics.py index 503d299..cb18652 100644 --- a/ingestor/metrics.py +++ b/ingestor/metrics.py @@ -19,7 +19,6 @@ from prometheus_client import Counter, Gauge, Histogram # Metric label definitions for consistent labeling across all metrics POD_ID_LABEL = ['pod_id'] SERVER_LABELS = ['pod_id', 'server_name', 'server_url'] -KAFKA_LABELS = ['pod_id', 'topic'] REDIS_LABELS = ['pod_id', 'operation'] NOTIFICATION_LABELS = ['pod_id', 'level', 'block'] @@ -134,22 +133,6 @@ OPC_RECONNECTIONS_TOTAL = Counter( ['pod_id', 'server_name'], ) -# --- Data Manager (Kafka) Metrics --- -KAFKA_MESSAGES_SENT = Counter( - 'kafka_messages_sent_total', 'Total messages sent to Kafka', KAFKA_LABELS -) -KAFKA_MESSAGES_ERRORS = Counter( - 'kafka_messages_errors_total', - 'Total errors sending messages to Kafka', - KAFKA_LABELS, -) -KAFKA_CONNECTION_STATUS = Gauge( - 'kafka_connection_status', - 'Connection status with Kafka (1=connected, 0=disconnected)', - POD_ID_LABEL, -) - - # --- Notification Metrics --- NOTIFICATIONS_SENT = Counter( 'notifications_sent_total', diff --git a/tests/unit/managers/test_data_manager.py b/tests/unit/managers/test_data_manager.py index 2cb4ae9..4c6b2b6 100644 --- a/tests/unit/managers/test_data_manager.py +++ b/tests/unit/managers/test_data_manager.py @@ -1,6 +1,5 @@ from unittest.mock import ANY, AsyncMock, MagicMock, patch -from kafka.errors import NoBrokersAvailable from pytest import fixture, mark from sientia_do.notifications.models import NotificationLevel @@ -18,14 +17,11 @@ metadata = { @fixture -@patch('ingestor.managers.data_manager.KafkaProducer') @patch('ingestor.managers.data_manager.MongoDBRepository') -def data_manager(mongodb_repository, kafka): +def data_manager(mongodb_repository): data_manager = DataManager( - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=True, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -38,141 +34,34 @@ def data_manager(mongodb_repository, kafka): return data_manager -@patch('ingestor.managers.data_manager.KafkaProducer') @patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___success(mongodb_repository, kafka): +def test___init___success(mongodb_repository): logger_mock = MagicMock() data_manager = DataManager( metadata=metadata['metadata'], - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=True, logger=logger_mock, notification_handler=MagicMock(), metrics_controller=MagicMock(), ) - kafka.assert_called_once_with( - bootstrap_servers='localhost:9092', value_serializer=ANY, key_serializer=ANY - ) - assert data_manager.kafka_producer is not None logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' + 'DataManager initialized with MongoDB servers: mongodb://localhost:27017' ) - logger_mock.info.assert_any_call('DataManager initialized with Kafka servers: localhost:9092') logger_mock.error.assert_not_called() - assert logger_mock.info.call_count == 4 + assert data_manager.connection_string == 'mongodb://localhost:27017' -@patch('ingestor.managers.data_manager.KafkaProducer') -@patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___second_attempt(mongodb_repository, kafka): - kafka.side_effect = [NoBrokersAvailable, MagicMock()] - logger_mock = MagicMock() - - data_manager = DataManager( - kafka_servers='localhost:9092', - mongo_connection_string='mongodb://localhost:27017', - mongo_database='sientia', - export_to_kafka=True, - logger=logger_mock, - notification_handler=MagicMock(), - metadata=metadata['metadata'], - metrics_controller=MagicMock(), - ) - - kafka.assert_any_call( - bootstrap_servers='localhost:9092', value_serializer=ANY, key_serializer=ANY - ) - assert kafka.call_count == 2 - assert data_manager.kafka_producer is not None - logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (1) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call('DataManager initialized with Kafka servers: localhost:9092') - logger_mock.error.assert_called_once_with( - 'Kafka servers localhost:9092 are not available. Retrying...' - ) - assert logger_mock.info.call_count == 5 - - -@patch('ingestor.managers.data_manager.KafkaProducer') -@patch('ingestor.managers.data_manager.MongoDBRepository') -def test___init___failure_max_attempts(mongodb_repository, kafka): - kafka.side_effect = NoBrokersAvailable - logger_mock = MagicMock() - - try: - DataManager( - kafka_servers='localhost:9092', - mongo_connection_string='mongodb://localhost:27017', - mongo_database='sientia', - export_to_kafka=True, - logger=logger_mock, - notification_handler=MagicMock(), - metadata=metadata['metadata'], - metrics_controller=MagicMock(), - ) - except NoBrokersAvailable as e: - assert ( - str(e) - == 'NoBrokersAvailable: Failed to connect to Kafka servers localhost:9092 after 3 attempts.' - ) - - assert kafka.call_count == 3 - logger_mock.info.assert_any_call( - 'Trying (0) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (1) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.info.assert_any_call( - 'Trying (2) to initializing DataManager with Kafka servers: localhost:9092' - ) - logger_mock.error.assert_called_with( - 'Failed to connect to Kafka servers localhost:9092 after 3 attempts.' - ) - assert logger_mock.info.call_count == 3 - - else: - raise AssertionError('Expected NoBrokersAvailable exception was not raised.') - - -def test_shutdown_has_producer(data_manager): - flush_mock = MagicMock() +def test_shutdown(data_manager): close_mock = MagicMock() - - data_manager.kafka_producer.flush = flush_mock - data_manager.kafka_producer.close = close_mock + data_manager.mongo_repository.close = close_mock data_manager.shutdown() - flush_mock.assert_called_once() close_mock.assert_called_once() -def test_shutdown_no_producer(data_manager): - data_manager.kafka_producer = None - - data_manager.shutdown() - - data_manager.logger.warning.assert_any_call( - 'Kafka producer is already closed or not initialized.' - ) - - -def test_shutdown_exception(data_manager): - data_manager.kafka_producer.flush = MagicMock(side_effect=Exception('Test error')) - data_manager.kafka_producer.close = MagicMock() - - data_manager.shutdown() - data_manager.logger.error.assert_called_once_with('Error closing Kafka producer: Test error') - - def test_shutdown_exception_mongo(data_manager): data_manager.mongo_repository.close = MagicMock(side_effect=Exception('Test error')) @@ -187,88 +76,24 @@ def test___del__(data_manager): data_manager.shutdown.assert_called_once() -def test_delivery_report(data_manager): - msg = MagicMock() - msg.topic = 'test_topic' - msg.partition = 0 - msg.offset = 1 - - data_manager.delivery_report(msg) - - data_manager.logger.debug.assert_called_once_with( - f'Record successfully produced to {msg.topic} [{msg.partition}] at offset {msg.offset}' - ) - - -def test_delivery_error(data_manager): - err = 'Test error' - data_manager.delivery_error(err) - - data_manager.logger.error.assert_called_once_with(f'Delivery failed for record : {err}') - - @mark.asyncio async def test_publish(data_manager): - topic = 'test_topic' - data = {'key': 'value'} - - # Mock the send method of the Kafka producer - send_mock = MagicMock() - data_manager.kafka_producer.send = send_mock - - # Call the publish method - await data_manager.publish(topic, data) - - # Check if the send method was called with the correct arguments - send_mock.assert_called_once_with(topic=topic, value=data) - - send_mock.return_value.add_callback.assert_called_once() - - data_manager.kafka_producer.flush.assert_called_once() - - -def test_publish_no_kafka(data_manager): - data_manager.export_to_kafka = False - topic = 'test_topic' - data = {'key': 'value'} - - data_manager.publish(topic, data) - - data_manager.kafka_producer.send.assert_not_called() - - -@patch('ingestor.managers.data_manager.traceback') -@mark.asyncio -async def test_publish_error(traceback, data_manager): topic = 'test_topic' data = {'key': 'value', 'name': 'test_tag'} - # Mock the send method of the Kafka producer to raise an exception - send_mock = MagicMock(side_effect=Exception('Test error')) - data_manager.kafka_producer.send = send_mock + data_manager.mongo_repository.insert = AsyncMock() - data_manager.mongo_repository = AsyncMock() - - # Call the publish method await data_manager.publish(topic, data) - # Check if the send method was called with the correct arguments - send_mock.assert_called_once_with(topic=topic, value=data) - - # Check if the error was logged - data_manager.send_notification_async.assert_called_once_with( - notification_id=f'KAFKA_PRODUCER_ERROR_{topic}', - message=f'Error publishing message to topic {topic}: Test error', - block='kafka_producer', - level=NotificationLevel.ERROR, - attachment_content=traceback.format_exc.return_value, + data_manager.mongo_repository.insert.assert_called_once_with( + collection_name=topic, + document={**data, 'inserted_at': ANY}, metadata=metadata['metadata'], ) @mark.asyncio async def test_publish_error_mongo(data_manager): - data_manager.export_to_kafka = False data_manager.mongo_repository.insert = AsyncMock(side_effect=Exception('Test error')) await data_manager.publish('test_topic', {'key': 'value'}) diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 32ac827..9dcdd21 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -20,14 +20,12 @@ metadata = { @patch('ingestor.managers.ingestor_manager.ResourceManager') def ingestor_manager(data_manager_mock, resource_manager_mock): ingestor = IngestorManager( - kafka_servers='localhost:9092', redis_data={'host': 'localhost', 'port': 6379}, lease_ttl=60, heartbeat_ttl=60, poll_interval=5, mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -49,14 +47,12 @@ def test___init__( notification_handler_mock, resource_manager_mock, data_manager_mock, opc_manager_mock ): ingestor = IngestorManager( - kafka_servers='localhost:9092', redis_data={'host': 'localhost', 'port': 6379}, lease_ttl=60, heartbeat_ttl=60, poll_interval=5, mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, logger=MagicMock(), notification_handler=MagicMock(), metadata=metadata['metadata'], @@ -65,10 +61,8 @@ def test___init__( opc_manager_mock.assert_not_called() data_manager_mock.assert_called_once_with( - kafka_servers='localhost:9092', mongo_connection_string='mongodb://localhost:27017', mongo_database='sientia', - export_to_kafka=False, metadata=metadata['metadata'], logger=ingestor.logger, notification_handler=ingestor.notification_handler, diff --git a/tests/unit/test_ingestor.py b/tests/unit/test_ingestor.py index 63ea638..8bbccc4 100644 --- a/tests/unit/test_ingestor.py +++ b/tests/unit/test_ingestor.py @@ -9,8 +9,6 @@ from ingestor.ingestor import Ingestor @patch('ingestor.ingestor.NotificationHandler') def test___init__(notification_handler, getenv): getenv.side_effect = [ - 'localhost:9092,localhost:35', # KAFKA_SERVERS - 'true', # EXPORT_TO_KAFKA 'localhost', # REDIS_HOST '63790', # REDIS_PORT 'user', # REDIS_USERNAME @@ -27,7 +25,6 @@ def test___init__(notification_handler, getenv): ingestor = Ingestor() - getenv.assert_any_call('KAFKA_SERVERS', 'localhost:9092') getenv.assert_any_call('REDIS_HOST', 'localhost') getenv.assert_any_call('REDIS_PORT', '6379') getenv.assert_any_call('REDIS_USERNAME', None) @@ -37,7 +34,6 @@ def test___init__(notification_handler, getenv): getenv.assert_any_call('HOSTNAME', 'localhost') getenv.assert_any_call('POLL_INTERVAL', '5') - assert ingestor.kafka_servers == ['localhost:9092', 'localhost:35'] assert ingestor.redis_host == 'localhost' assert ingestor.redis_port == 63790 assert ingestor.redis_username == 'user' @@ -124,7 +120,6 @@ async def test_prepare_ingestor(ingestor_manager_mock, ingestor): await ingestor.prepare_ingestor() ingestor_manager_mock.assert_called_once_with( - kafka_servers=','.join(ingestor.kafka_servers), redis_data={ 'host': ingestor.redis_host, 'port': ingestor.redis_port, @@ -139,7 +134,6 @@ async def test_prepare_ingestor(ingestor_manager_mock, ingestor): metadata=ingestor.metadata, logger=ingestor.logger, notification_handler=ingestor.notification_handler, - export_to_kafka=ingestor.export_to_kafka, metrics_controller=ingestor.metrics_controller, ) ingestor_manager.declare_active.assert_called_once() diff --git a/tests/unit/test_metrics.py b/tests/unit/test_metrics.py index 809abdf..3c50dea 100644 --- a/tests/unit/test_metrics.py +++ b/tests/unit/test_metrics.py @@ -183,30 +183,6 @@ def test_opc_reconnections_total(): assert set(metrics.OPC_RECONNECTIONS_TOTAL._labelnames) == {'pod_id', 'server_name'} -def test_kafka_messages_sent(): - """Verify the definition of KAFKA_MESSAGES_SENT.""" - assert metrics.KAFKA_MESSAGES_SENT is not None - assert isinstance(metrics.KAFKA_MESSAGES_SENT, Counter) - assert metrics.KAFKA_MESSAGES_SENT._name == 'kafka_messages_sent' # REMOVED _total - assert set(metrics.KAFKA_MESSAGES_SENT._labelnames) == {'pod_id', 'topic'} - - -def test_kafka_messages_errors(): - """Verify the definition of KAFKA_MESSAGES_ERRORS.""" - assert metrics.KAFKA_MESSAGES_ERRORS is not None - assert isinstance(metrics.KAFKA_MESSAGES_ERRORS, Counter) - assert metrics.KAFKA_MESSAGES_ERRORS._name == 'kafka_messages_errors' # REMOVED _total - assert set(metrics.KAFKA_MESSAGES_ERRORS._labelnames) == {'pod_id', 'topic'} - - -def test_kafka_connection_status(): - """Verify the definition of KAFKA_CONNECTION_STATUS.""" - assert metrics.KAFKA_CONNECTION_STATUS is not None - assert isinstance(metrics.KAFKA_CONNECTION_STATUS, Gauge) - assert metrics.KAFKA_CONNECTION_STATUS._name == 'kafka_connection_status' - assert set(metrics.KAFKA_CONNECTION_STATUS._labelnames) == {'pod_id'} - - def test_notifications_sent(): """Verify the definition of NOTIFICATIONS_SENT.""" assert metrics.NOTIFICATIONS_SENT is not None From 648458510d0fe78e07c3dbbd3ae98d4359d3c664 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 13:37:56 -0300 Subject: [PATCH 07/16] SAM0123-231: Remove Kafka integration for data streaming and add comprehensive OPC UA client certificate management. --- README.md | 67 ++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 47 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index d7ef4ad..15f3669 100644 --- a/README.md +++ b/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 │ - │ - Slot Leasing │ │ - MongoDB Store │ - │ - Heartbeats │ │ - Data Pipeline │ + │ - Redis Coord. │ │ - MongoDB Store │ + │ - Slot Leasing │ │ - 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 @@ -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='' ./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,6 +323,7 @@ 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 +351,7 @@ sientia-dataops-opc-ingestor/ - 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** +3. **Performance Issues** - Monitor Prometheus metrics for bottlenecks - Adjust polling intervals and lease TTLs - Review OPC server performance and network latency @@ -350,7 +377,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 From f75aa322cb189d13ecae5fade8f271d64b1a1587 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 13:41:56 -0300 Subject: [PATCH 08/16] SAM0123-231: Remove Kafka configuration and infrastructure references; simplify README documentation. --- .env.example | 4 ---- README.md | 18 +----------------- 2 files changed, 1 insertion(+), 21 deletions(-) diff --git a/.env.example b/.env.example index d6fcde0..54ea343 100644 --- a/.env.example +++ b/.env.example @@ -1,7 +1,3 @@ -# Kafka -KAFKA_SERVERS="localhost:9092" -EXPORT_TO_KAFKA="false" - # Redis REDIS_HOST="localhost" REDIS_PORT="6379" diff --git a/README.md b/README.md index 15f3669..0f2db31 100644 --- a/README.md +++ b/README.md @@ -156,7 +156,7 @@ pytest # Run with coverage pytest --cov=ingestor --cov-report=html - +``` ### Populating Redis with OPC Configuration @@ -325,7 +325,6 @@ sientia-dataops-opc-ingestor/ ├── 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 ``` @@ -395,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: @@ -410,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. \ No newline at end of file From 08a76a9293617b889ae10295ad97d8a32e2e74dd Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 16:10:51 -0300 Subject: [PATCH 09/16] SAM0123-231: Remove version pin for asyncua in requirements.txt --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index ed65ae3..4db22cf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -asyncua==1.1.5 +asyncua redis sientia_do==1.12.1 prometheus_client From a239a1a706ed701c08ab5dfbdd09851884af21a0 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 23 Jul 2026 16:18:28 -0300 Subject: [PATCH 10/16] SAM0123-231: Enhance OPC UA client setup by asserting certificate and private key presence and assigning pod ID to client description. --- ingestor/managers/opc_manager.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 7b446c9..3505757 100644 --- a/ingestor/managers/opc_manager.py +++ b/ingestor/managers/opc_manager.py @@ -162,6 +162,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 +210,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 From ef37122f56ee1ba7660d21c5ca204206c69caf52 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 5 Aug 2026 15:08:00 -0300 Subject: [PATCH 11/16] ci(workflow): Enhance quality gate, streamline validation, and add AI guidance - **Quality Gate (.github/workflows/quality-gate.yml):** - Expanded push triggers to include main, release/**, and feature/** branches. - Added concurrency control to cancel in-progress runs, preventing redundant checks. - Refined workflow permissions to be more specific (contents: read, pull-requests: write, issues: write), enhancing security. - **Local Validation (validate.sh):** - Removed the validate.sh script, as its checks are now integrated into or superseded by the CI pipeline. - **AI Guidance (CLAUDE.md):** - Introduced CLAUDE.md to provide comprehensive guidance for Claude Code, detailing project architecture, commands, and conventions for AI assistants. - **SonarQube Configuration (sonar-project.properties):** - Removed hardcoded sonar.projectVersion. --- .github/workflows/quality-gate.yml | 14 +++- CLAUDE.md | 74 +++++++++++++++++ sonar-project.properties | 1 - validate.sh | 124 ----------------------------- 4 files changed, 87 insertions(+), 126 deletions(-) create mode 100644 CLAUDE.md delete mode 100755 validate.sh diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 61b07c5..fa5f41d 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -1,6 +1,11 @@ name: Quality gate on: + push: + branches: + - main + - 'release/**' + - 'feature/**' pull_request: branches: - main @@ -8,10 +13,17 @@ on: - '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/python-quality-gate.yml@main - permissions: write-all + permissions: + contents: read + pull-requests: write + issues: write with: project_name: 'ingestor' repositories: 'sientia-dataops-library' diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..7330de9 --- /dev/null +++ b/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:`, 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. diff --git a/sonar-project.properties b/sonar-project.properties index 081bdfa..fc43878 100644 --- a/sonar-project.properties +++ b/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 diff --git a/validate.sh b/validate.sh deleted file mode 100755 index 5e837c6..0000000 --- a/validate.sh +++ /dev/null @@ -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 \ No newline at end of file From 810bf54d5675779ad03096ac245cede590d04bbe Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 5 Aug 2026 15:25:06 -0300 Subject: [PATCH 12/16] fix(ingestor): Remove __del__ methods for async shutdown --- ingestor/managers/ingestor_manager.py | 4 ---- ingestor/managers/opc_manager.py | 3 --- tests/unit/managers/test_ingestor_manager.py | 7 ------- 3 files changed, 14 deletions(-) diff --git a/ingestor/managers/ingestor_manager.py b/ingestor/managers/ingestor_manager.py index 34a09ca..0034d0a 100644 --- a/ingestor/managers/ingestor_manager.py +++ b/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. diff --git a/ingestor/managers/opc_manager.py b/ingestor/managers/opc_manager.py index 3505757..c08d5ab 100644 --- a/ingestor/managers/opc_manager.py +++ b/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. diff --git a/tests/unit/managers/test_ingestor_manager.py b/tests/unit/managers/test_ingestor_manager.py index 9dcdd21..fe8308c 100644 --- a/tests/unit/managers/test_ingestor_manager.py +++ b/tests/unit/managers/test_ingestor_manager.py @@ -170,13 +170,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() From 68dc8bd6d502e9835388c3dfe3611c2bf5fae261 Mon Sep 17 00:00:00 2001 From: Eduardo Rios Date: Mon, 17 Aug 2026 16:42:16 -0300 Subject: [PATCH 13/16] SIENTIAPDE-2072: bump sientia_do pin to 1.12.2 Picks up the notification timestamp -> native datetime fix. --- requirements-local.txt | 2 +- requirements.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/requirements-local.txt b/requirements-local.txt index 190e40b..063fdd4 100644 --- a/requirements-local.txt +++ b/requirements-local.txt @@ -1,5 +1,5 @@ asyncua==1.1.5 redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2 prometheus_client pymongo \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 4db22cf..134a075 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ asyncua redis -sientia_do==1.12.1 +sientia_do==1.12.2 prometheus_client pymongo \ No newline at end of file From 0aa39e3b0c5b3fdc866473ccf95ead7989e936d9 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Thu, 10 Sep 2026 11:12:53 -0300 Subject: [PATCH 14/16] SIENTIAPDE-2068: Implement DataOps Worker deployment workflow, remove legacy release workflow, and enforce 'chart-ref' label on pull requests. --- .github/workflows/deploy.yml | 29 +++++++++++++++++++ .github/workflows/release.yml | 16 ---------- .github/workflows/require-chart-ref-label.yml | 16 ++++++++++ 3 files changed, 45 insertions(+), 16 deletions(-) create mode 100644 .github/workflows/deploy.yml delete mode 100644 .github/workflows/release.yml create mode 100644 .github/workflows/require-chart-ref-label.yml diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml new file mode 100644 index 0000000..fd078b4 --- /dev/null +++ b/.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 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml deleted file mode 100644 index 45d63bb..0000000 --- a/.github/workflows/release.yml +++ /dev/null @@ -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 \ No newline at end of file diff --git a/.github/workflows/require-chart-ref-label.yml b/.github/workflows/require-chart-ref-label.yml new file mode 100644 index 0000000..20f8be3 --- /dev/null +++ b/.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 From 7c28c147cac094d95d73d61ea6d905171cdb3446 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 10 Sep 2026 14:29:54 +0000 Subject: [PATCH 15/16] chore: bump version to 1.4.4 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index e4841ad..8a7dbab 100644 --- a/pyproject.toml +++ b/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" From 6df354f505a5e4c84826f6ec9fb1a8181ac4f3b1 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Fri, 11 Sep 2026 00:33:34 -0300 Subject: [PATCH 16/16] ci(deploy): Configure env_list_path as a list --- .github/workflows/deploy.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index fd078b4..998130d 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -20,7 +20,7 @@ jobs: branch_name: ${{ github.event.pull_request.head.ref }} project_name: 'ingestor' chart_name: 'opc-ingestor' - env_list_path: '.env' + env_list_path: '[".env"]' update_version_in: '["pyproject"]' create_tarball: ${{ github.event_name == 'pull_request' }} mirror: true