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

SIENTIAPDE-1974: feat(read): slot v2 reads via SubscriptionEngine, OPC cert from MinIO

Browse Source 
Swaps the legacy OpcManager read path for the gateway's SubscriptionEngine over
the shared OpcConnectionPool. The OPC client certificate now always comes from
MinIO + Mongo (CertProvider, Basic256Sha256) — no more certs/ shipped in the
Gitea repository, so a new version branch no longer breaks the reads.

- io/slots.py: parse_slot accepts slot v2 (names[]) and converts the legacy
  v1 (tag_name + topics; cert_path => use_security, path dropped), so gateway
  and Orchestrator can be deployed in any order.
- managers/server_reader.py: new per-server reader with the OpcManager
  interface the IngestorManager already drives (connect, create_subscription,
  subscribe, unsubscribe, check_cycles, check_opc_listenning, shutdown).
- IngestorManager owns the read path: lazy-MinIO CertProvider, the pool and a
  read DataSink, built without touching the network. Slots from Redis are
  normalized on lease and refresh; a malformed slot is notified and the last
  good config is kept.
- GatewayRuntime borrows the pool (writes/gRPC) instead of building it, and no
  longer closes it on stop: a Postgres/NATS failure can't stop the reads.
- pool: pin/unpin per active subscription so release_idle never evicts a
  subscribed connection; public evict() for lost servers / changed config.
- SubscriptionEngine pins on subscribe, unpins on unsubscribe, and drops a
  subscription locally even when the server-side delete fails.
- DataSink: model_id only written when the name has one, keeping v1 documents
  identical to the legacy ones.
- config: CERT_BUCKET default is the legacy streamlit-connectors bucket (D7).
- Removes OpcManager, its tests and scripts/generate-and-push-opc-cert.sh;
  README/CLAUDE.md document the MinIO certificate and the slot v2 format.

Tests: 218 unit tests pass; ruff check/format clean. Local end-to-end against
an asyncua server: v1 slot with two topics -> one doc per collection, legacy
doc shape, idle sweep keeps the subscription, lost-server shutdown and
re-initialization work.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-10-06 13:27:18 -03:00
parent 4c85a9c5c2
commit 50578fdba8
25 changed files with 957 additions and 1534 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
ingestor/app.py
View File

@@ -58,16 +58,17 @@ async def main(): # NOSONAR
ingestor.logger.info('Ingestor prepared. Starting main loop.')
# Gateway write/active-read/gRPC wiring (Task 18). Only built if prepare_ingestor
# got far enough to have a data_manager (needed by CertProvider/DataSink) --
# skipped entirely if prepare failed before that point. Construction can hit
# the network (MinIO bucket check, Postgres engine): a failure there must not
# kill the read path, so the app continues without the gateway runtime.
# got far enough to have an ingestor_manager (its data_manager and connection
# pool are shared) -- skipped entirely if prepare failed before that point. The
# read path (subscriptions) lives in the ingestor_manager and is already running:
# a failure here only disables writes/gRPC, the app continues without the runtime.
runtime: GatewayRuntime | None = None
if ingestor.ingestor_manager is not None:
try:
runtime = GatewayRuntime(
config=GatewayConfig.from_env(),
data_manager=ingestor.ingestor_manager.data_manager,
pool=ingestor.ingestor_manager.pool,
logger=ingestor.logger,
notification_handler=ingestor.notification_handler,
metrics_controller=ingestor.metrics_controller,

4
ingestor/config.py
View File

@@ -22,7 +22,9 @@ class GatewayConfig:
nats_queue: str = 'gateway-writers'
grpc_enabled: bool = False
grpc_port: int = 50051
cert_bucket: str = 'sientia-dataops'
# Legacy frontend bucket the platform already provisions (SIENTIAPDE-1974 D7);
# the design's `sientia-dataops` bucket does not exist in the platform MinIO.
cert_bucket: str = 'streamlit-connectors'
cert_cache_dir: str = '/tmp/opc-certs' # noqa: S108 # nosec B108 - default is spec'd, overridable via env
write_timestamp: bool = True
write_quality: bool = True

48
ingestor/connection/certs.py
View File

@@ -1,5 +1,10 @@
from os import getenv
from pathlib import Path
from sientia_do.repository.minio_repository import MinioRepository
from ingestor.config import _env_bool
COLLECTION_NAME = 'platform_certificates'
@@ -27,3 +32,46 @@ class CertProvider:
local_path = str(Path(self.cache_dir) / Path(object_name).name)
Path(local_path).write_bytes(data)
return local_path
class LazyMinio:
"""Builds the MinIO client on first use. MinioRepository's constructor hits the
network (bucket check), so building it eagerly would let a MinIO outage at pod
start take the whole read path down. Built lazily, only the servers that need a
certificate fail — and they are retried on the next loop."""
def __init__(self, factory):
self._factory = factory
self._client = None
async def download_file(self, object_name: str, bucket: str, metadata: dict):
if self._client is None:
self._client = self._factory()
return await self._client.download_file(object_name, bucket, metadata)
def build_cert_provider(
mongo, config, logger, notification_handler, metrics_controller, metadata=None
) -> CertProvider:
"""CertProvider shared by the read path (IngestorManager) and the write path
(GatewayRuntime): one certificate source — MinIO + Mongo — for every connection."""
def minio_factory() -> MinioRepository:
return MinioRepository(
endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'),
access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'),
secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'),
secure=_env_bool('MINIO_SECURE', True),
bucket=config.cert_bucket,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
return CertProvider(
mongo=mongo,
minio=LazyMinio(minio_factory),
cache_dir=config.cert_cache_dir,
bucket=config.cert_bucket,
metadata=metadata,
)

31
ingestor/connection/pool.py
View File

@@ -26,6 +26,10 @@ class OpcConnectionPool(SientiaMonitoring):
self.cert_provider = cert_provider
self.clock = MonotonicClock()
self._connections: dict[str, tuple[Client, float]] = {}
# Active subscriptions per server. A subscribed connection only receives
# callbacks and never calls get(), so without a pin release_idle() would
# evict it as idle and silently stop the reads (SIENTIAPDE-1974 G4).
self._pins: dict[str, int] = {}
# ponytail: one Lock per server_id, never cleaned up — server set is
# small and bounded. Add eviction of stale locks only if it grows.
self._locks: dict[str, asyncio.Lock] = {}
@@ -96,15 +100,36 @@ class OpcConnectionPool(SientiaMonitoring):
tags={'pod_id': self.pod_id, 'server_name': server_id, 'server_url': url},
)
def pin(self, server_id: str) -> None:
"""Mark one more active subscription on `server_id` (never evicted as idle)."""
self._pins[server_id] = self._pins.get(server_id, 0) + 1
def unpin(self, server_id: str) -> None:
"""Drop one active subscription mark from `server_id`."""
count = self._pins.get(server_id, 0) - 1
if count > 0:
self._pins[server_id] = count
else:
self._pins.pop(server_id, None)
async def release_idle(self, max_idle_s: float):
"""Disconnect and drop clients unused for longer than `max_idle_s`."""
"""Disconnect and drop clients unused for longer than `max_idle_s`,
except the ones pinned by an active subscription."""
now = self.clock.elapsed()
idle_ids = [
sid for sid, (_, last_used) in self._connections.items() if now - last_used > max_idle_s
sid
for sid, (_, last_used) in self._connections.items()
if now - last_used > max_idle_s and not self._pins.get(sid)
]
for server_id in idle_ids:
await self._evict(server_id)
async def evict(self, server_id: str):
"""Disconnect `server_id` now (lost server or changed config); the next
get() reconnects. No-op when the server has no pooled client."""
if server_id in self._connections:
await self._evict(server_id)
async def close_all(self):
"""Disconnect every pooled client (e.g. on shutdown)."""
# list() snapshots the keys: _evict() pops from self._connections,
@@ -122,7 +147,7 @@ class OpcConnectionPool(SientiaMonitoring):
async def _disconnect_with_retries(self, client: Client, server_id: str) -> None:
"""Tries 5 times to disconnect, with a delay of 100ms x try (mirrors
OpcManager.disconnection_fallback)."""
the legacy OpcManager.disconnection_fallback)."""
for i in range(5):
try:
await client.disconnect()

14
ingestor/engines/subscription.py
View File

@@ -57,6 +57,7 @@ class SubscriptionEngine(SientiaMonitoring):
raise
self._clients[name] = client
self._servers[name] = server_id
self.pool.pin(server_id)
self.logger.info(f'Subscription {name} created on {server_id}.')
await self.emit_metric(
metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED,
@@ -96,10 +97,17 @@ class SubscriptionEngine(SientiaMonitoring):
if not self.subscriptions.get(subscription):
self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.")
return
await self.subscriptions[subscription].delete()
del self.subscriptions[subscription]
sub = self.subscriptions.pop(subscription)
self._clients.pop(subscription, None)
self._servers.pop(subscription, None)
server_id = self._servers.pop(subscription, None)
if server_id is not None:
self.pool.unpin(server_id)
try:
await sub.delete()
except Exception as e:
# The connection may already be gone (lost server); the subscription is
# dropped from the books either way so the slot can be re-subscribed.
self.logger.warning(f'Failed to delete subscription {subscription}: {e}')
self.logger.info(f'Unsubscribed from {subscription}.')
async def datachange_notification(self, node, _val, data):

42
ingestor/gateway_runtime.py
View File

@@ -2,30 +2,25 @@
object graph and wires it into the app's lifecycle.
Reuses the existing ingestor's logger/notification_handler/metrics_controller and
its already-connected DataManager (so CertProvider shares the existing Mongo
connection instead of opening a second one -- MongoDBRepository.find_one has the
exact (collection_name, filters, metadata) signature CertProvider.fetch calls).
its already-connected DataManager. The OpcConnectionPool (and its CertProvider,
MinIO + Mongo) belongs to the read path in IngestorManager (SIENTIAPDE-1974):
reads start in prepare_ingestor, before this runtime exists, and must keep running
if anything here fails. The runtime only borrows the pool for writes and gRPC.
# ponytail: subscription v2 swap + scheduled_tick wiring happen at 1974
# integration; when swapping, build ONE SubscriptionEngine per OPC server (its
# non_receive_count is engine-wide -- see plan ledger F1 carry-forward).
# ponytail: scheduled_tick + gRPC slot_lookup wiring land in the next 1974 RC.
"""
from os import getenv
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.repository.minio_repository import MinioRepository
from sientia_do.temporal.activities.postgres import Postgres
from sientia_do.utils.connectors_config import build_postgres_config
from ingestor.adapters.grpc_server import GrpcServer
from ingestor.adapters.nats_consumer import NatsWriteConsumer
from ingestor.config import GatewayConfig, _env_bool
from ingestor.connection.certs import CertProvider
from ingestor.config import GatewayConfig
from ingestor.connection.pool import OpcConnectionPool
from ingestor.engines.active_read import ActiveReadEngine
from ingestor.engines.write import WriteEngine
@@ -44,6 +39,7 @@ class GatewayRuntime(SientiaMonitoring):
self,
config: GatewayConfig,
data_manager: DataManager,
pool: OpcConnectionPool,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
@@ -55,26 +51,7 @@ class GatewayRuntime(SientiaMonitoring):
self._nats_started = False
self._nats_notified = False # notify once per outage, not per 5s retry
minio = MinioRepository(
endpoint=getenv('MINIO_ENDPOINT', 'localhost:9000'),
access_key=getenv('MINIO_ACCESS_KEY', 'minioadmin'),
secret_key=getenv('MINIO_SECRET_KEY', 'minioadmin'),
secure=_env_bool('MINIO_SECURE', True),
bucket=config.cert_bucket,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
cert_provider = CertProvider(
mongo=data_manager.mongo_repository,
minio=minio,
cache_dir=config.cert_cache_dir,
bucket=config.cert_bucket,
metadata=self.metadata,
)
self.pool = OpcConnectionPool(
cert_provider, logger, notification_handler, metrics_controller
)
self.pool = pool # owned by IngestorManager (read path); closed there
postgres = Postgres(
**build_postgres_config(),
@@ -128,7 +105,8 @@ class GatewayRuntime(SientiaMonitoring):
if self._nats_started:
await self.nats_consumer.stop()
self._nats_started = False
await self.pool.close_all()
# The pool is not closed here: IngestorManager.shutdown() owns it and runs
# after this, once its readers have dropped their subscriptions.
async def idle_sweep(self) -> None:
await self.pool.release_idle(self.config.idle_conn_max_s)

19
ingestor/io/data_sink.py
View File

@@ -21,15 +21,16 @@ class DataSink:
await self.data_manager.emit_metric(**kwargs)
async def publish_reading(self, tag: str, value, timestamp, names: list[dict]) -> None:
"""Publish one MongoDB document per name entry and record its tag value metric."""
"""Publish one MongoDB document per name entry and record its tag value metric.
`model_id` is only written when the name carries one: names converted from a
slot v1 have none, and their documents must stay identical to the legacy ones."""
for name in names:
doc = {
'tag': tag,
'name': name['tag_name'],
'model_id': name['model_id'],
'timestamp': timestamp,
'value': value,
}
doc = {'tag': tag, 'name': name['tag_name']}
if 'model_id' in name:
doc['model_id'] = name['model_id']
doc['timestamp'] = timestamp
doc['value'] = value
await self.data_manager.publish(name['collection'], doc)
await self.emit_metric(
metric_object=metrics.OPC_TAG_VALUE,
@@ -38,7 +39,7 @@ class DataSink:
tags={
'pod_id': self.data_manager.pod_id,
'tag_name': name['tag_name'],
'model_id': name['model_id'],
'model_id': name.get('model_id', ''),
},
)

36
ingestor/io/slots.py
View File

@@ -1,25 +1,45 @@
"""Parse the slot v2 structure written by the Orchestrator to Redis (contract C1)."""
"""Parse the slot written by the Orchestrator to Redis into the slot v2 structure (contract C1).
Both shapes are accepted so the gateway and the Orchestrator can be deployed in
any order (SIENTIAPDE-1974):
- v2: each tag carries `names` (one entry per collection / tag name / model);
- v1 (legacy): each tag carries `tag_name` + `topics`, and the server carries the
file-based `cert_path`. v1 is converted to v2: one `names` entry per topic, and
`use_security` is true when the server had a certificate path (the certificate
itself now always comes from MinIO via CertProvider, never from that path).
"""
SERVER_FIELDS = ('server_id', 'url', 'server_uri', 'use_security', 'subscription_period_ms')
def _parse_tag(address: str, tag: dict) -> dict:
def _parse_names(address: str, tag: dict) -> list[dict]:
names = tag.get('names')
if not names:
raise ValueError(f'tag {address!r} has no names')
if names:
return names
topics = tag.get('topics')
if topics and tag.get('tag_name'):
return [{'collection': topic, 'tag_name': tag['tag_name']} for topic in topics]
raise ValueError(f'tag {address!r} has no names')
def _parse_tag(address: str, tag: dict) -> dict:
return {
'frequency': tag['frequency'],
'names': names,
'names': _parse_names(address, tag),
'active_read': tag.get('active_read', {'enabled': False}),
}
def _parse_server(server: dict) -> dict:
def _parse_server(name: str, server: dict) -> dict:
parsed = {field: server[field] for field in SERVER_FIELDS if field in server}
parsed['name'] = server.get('name', name)
if 'use_security' not in parsed:
parsed['use_security'] = bool(server.get('cert_path'))
parsed['tags'] = {addr: _parse_tag(addr, tag) for addr, tag in server['tags'].items()}
return parsed
def parse_slot(raw: dict) -> dict:
"""Normalize a raw slot v2 dict into `{server_name: {..., tags: {addr: tag}}}`."""
return {name: _parse_server(server) for name, server in raw.items()}
"""Normalize a raw slot (v1 or v2) into `{server_name: {..., tags: {addr: tag}}}`."""
return {name: _parse_server(name, server) for name, server in raw.items()}

117
ingestor/managers/ingestor_manager.py
View File

@@ -8,9 +8,14 @@ from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
import ingestor.metrics as metrics
from ingestor.config import GatewayConfig
from ingestor.connection.certs import build_cert_provider
from ingestor.connection.pool import OpcConnectionPool
from ingestor.io.data_sink import DataSink
from ingestor.io.slots import parse_slot
from ingestor.managers.data_manager import DataManager
from ingestor.managers.opc_manager import OpcManager
from ingestor.managers.resource_manager import ResourceManager
from ingestor.managers.server_reader import ServerReader
class IngestorManager(SientiaMonitoring):
@@ -47,7 +52,10 @@ class IngestorManager(SientiaMonitoring):
Attributes:
data_manager (DataManager): Manages data persistence
opc_managers (dict): Dictionary of OPC managers keyed by server name
opc_managers (dict): ServerReader per OPC server, keyed by server name
pool (OpcConnectionPool): OPC connections shared by reads, writes and gRPC;
owned here because reads start in prepare_ingestor, before GatewayRuntime
sink (DataSink): read egress (fan-out to every slot v2 `names` entry)
resource_manager (ResourceManager): Manages Redis-based resource coordination
number_of_slots (int): Total number of slots configured in the system
poll_interval (int): Main loop polling interval
@@ -68,6 +76,7 @@ class IngestorManager(SientiaMonitoring):
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
gateway_config: GatewayConfig | None = None,
):
redis_host: str = redis_data['host']
redis_port: int = int(redis_data['port'])
@@ -90,6 +99,22 @@ class IngestorManager(SientiaMonitoring):
metrics_controller=metrics_controller,
)
self.opc_managers: dict = {}
# Read path (SIENTIAPDE-1974): pooled connections with the certificate from
# MinIO. Nothing here touches the network at construction (MinIO is lazy), so
# a MinIO/Postgres/NATS outage can't stop the reads before they start.
config = gateway_config or GatewayConfig.from_env()
self.cert_provider = build_cert_provider(
mongo=self.data_manager.mongo_repository,
config=config,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
metadata=metadata,
)
self.pool = OpcConnectionPool(
self.cert_provider, logger, notification_handler, metrics_controller
)
self.sink = DataSink(self.data_manager, None)
self.resource_manager = ResourceManager(
host=redis_host,
port=redis_port,
@@ -109,48 +134,34 @@ class IngestorManager(SientiaMonitoring):
self.metadata = metadata
async def initialize_opc_from_config(self, server_config: dict) -> OpcManager | None:
async def initialize_opc_from_config(self, server_config: dict) -> ServerReader | None:
"""
Initializes an OPC Manager instance using the provided server configuration.
Initializes the reader of one OPC UA server from its slot v2 configuration.
This method creates and configures an OPC Manager for a specific OPC UA server,
establishing the connection and preparing it for tag subscriptions.
Creates a ServerReader and opens its pooled connection, so the server is ready
for tag subscriptions. With `use_security`, the certificate comes from MinIO.
Args:
server_config (dict): A dictionary containing the OPC server configuration.
Expected keys include:
- 'name' (str): The name of the OPC server.
- 'url' (str): The URL of the OPC server.
- 'server_uri' (str): The URI of the OPC server.
- 'cert_path' (str, optional): Path to the client certificate file.
- 'private_key_path' (str, optional): Path to the private key file.
- 'server_cert_path' (str, optional): Path to the server certificate file.
server_config (dict): The server configuration of a parsed slot (parse_slot),
without the 'tags' key. Expected keys: 'name', 'server_id', 'url',
'server_uri', 'use_security', 'subscription_period_ms'.
Returns:
OpcManager | None: An initialized OpcManager instance if successful,
otherwise None if an error occurs during initialization.
Raises:
Exception: If OPC manager initialization fails, the error is logged and
a notification is sent, but the method returns None to allow
the system to continue operating with other servers.
ServerReader | None: The connected reader, or None if the connection fails
(logged and notified, so the other servers keep working).
"""
try:
self.logger.info(f'Initializing OpcManager at {server_config["url"]}')
manager = OpcManager(
self.logger.info(f'Initializing OPC reader at {server_config["url"]}')
manager = ServerReader(
name=server_config['name'],
url=server_config['url'],
subscription_period_ms=server_config['subscription_period_ms'],
data_manager=self.data_manager,
server_config=server_config,
pool=self.pool,
sink=self.sink,
logger=self.logger,
server_uri=server_config['server_uri'],
notification_handler=self.notification_handler,
metadata=self.metadata,
cert_path=server_config.get('cert_path'),
private_key_path=server_config.get('private_key_path'),
server_cert_path=server_config.get('server_cert_path'),
metrics_controller=self.metrics_controller,
metadata=self.metadata,
)
manager.config = server_config
@@ -176,15 +187,15 @@ class IngestorManager(SientiaMonitoring):
Gracefully shuts down the IngestorManager and all its components.
This method ensures proper cleanup of:
- All OPC manager instances and their connections
- All OPC readers, their subscriptions and the pooled connections
- Data manager connections and resources
- Active subscriptions and server connections
The shutdown process is performed asynchronously to allow proper cleanup
of all managed resources before termination.
"""
for _server_name, server in self.opc_managers.items():
await server.shutdown()
await self.pool.close_all()
self.data_manager.shutdown()
async def remove_server(self, server: str):
@@ -419,14 +430,18 @@ class IngestorManager(SientiaMonitoring):
Notes:
- If a slot is leased but its details cannot be retrieved
(i.e., `get_tag_slot` returns None), that slot is skipped.
(i.e., `get_tag_slot` returns None) or are malformed, that slot is skipped.
- Slot details are normalized to the slot v2 structure (parse_slot).
"""
acquired = {}
for i in range(1, self.number_of_slots + 1):
if await self.resource_manager.lease_tag(str(i)):
self.logger.info(f'Leased slot {i}')
slots = await self.resource_manager.get_tag_slot(str(i))
raw = await self.resource_manager.get_tag_slot(str(i))
if raw is None:
continue
slots = self.parse_slot_config(str(i), raw)
if slots is None:
continue
acquired[str(i)] = slots
@@ -518,16 +533,44 @@ class IngestorManager(SientiaMonitoring):
removed_slots: list[str] = []
for slot, _slot_config in self.managed_tags.items():
await self.resource_manager.renew_tag_lease(slot)
update = await self.resource_manager.get_tag_slot(slot)
if update is None:
raw = await self.resource_manager.get_tag_slot(slot)
if raw is None:
removed_slots.append(slot)
continue
update = self.parse_slot_config(slot, raw)
if update is None:
# Malformed config: keep reading with the last good one.
continue
self.managed_tags[slot] = update
for slot in removed_slots:
self.managed_tags.pop(slot, None)
def parse_slot_config(self, slot: str, raw: dict) -> dict | None:
"""
Normalizes the raw slot read from Redis to the slot v2 structure.
Accepts slot v1 (`topics`) and v2 (`names`), so the gateway and the
Orchestrator can be deployed in any order (SIENTIAPDE-1974).
Returns:
dict | None: The parsed slot, or None when the config is malformed
(logged and notified).
"""
try:
return parse_slot(raw)
except (AttributeError, KeyError, TypeError, ValueError) as e:
self.send_notification(
metadata=self.metadata,
notification_id=f'OPC_SLOT_CONFIG_ERROR_{slot}',
message=f'Slot {slot} has an invalid configuration: {e!r}',
block='ingestor_manager',
level=NotificationLevel.ERROR,
)
self.logger.error(f'Slot {slot} has an invalid configuration: {e!r}')
return None
async def drop_slot_leases(self, ids: list[str]) -> None:
"""
Releases the leases associated with the specified slot IDs.

588
ingestor/managers/opc_manager.py
View File

@@ -1,588 +0,0 @@
import asyncio
import json
import traceback
from pathlib import Path
from asyncua import Client
from asyncua.crypto.security_policies import SecurityPolicyBasic256
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ, OPC_TIMEZONE
import ingestor.metrics as metrics
from ingestor.managers.data_manager import DataManager
class OpcManager(SientiaMonitoring):
"""
Manages OPC UA server connections and tag subscriptions.
The OpcManager is responsible for:
- Establishing and maintaining secure connections to OPC UA servers
- Managing tag subscriptions and data collection
- Handling server reconnection and error recovery
- Processing OPC data and forwarding it to the data manager
- Monitoring connection health and performance metrics
The manager supports both secure and unsecured connections, with optional
certificate-based authentication for enhanced security.
Args:
name (str): Unique identifier for the OPC server
url (str): OPC UA server endpoint URL
data_manager (DataManager): Manager for data persistence and export
logger (Logger): Logger instance for application logging
server_uri (str): OPC UA server application URI
notification_handler (NotificationHandler): Handler for sending notifications
metadata (dict): Application metadata for notifications and tracking
cert_path (str, optional): Path to client certificate file for secure connections
private_key_path (str, optional): Path to client private key file
server_cert_path (str, optional): Path to server certificate file for validation
Attributes:
url (str): OPC UA server endpoint URL
name (str): Unique identifier for the OPC server
server_uri (str): OPC UA server application URI
data_queue (dict): Queue for buffering OPC data before processing
non_receive_count (int): Counter for cycles without data reception
client (Client): OPC UA client instance
cert_path (str): Path to client certificate file
private_key_path (str): Path to client private key file
server_cert_path (str): Path to server certificate file
nodes (dict): Dictionary of OPC node references
subscriptions (dict): Active OPC subscriptions
data_manager (DataManager): Manager for data persistence and export
metadata (dict): Application metadata
"""
def __init__(
self,
name: str,
url: str,
subscription_period_ms: int,
data_manager: DataManager,
logger: Logger,
server_uri: str,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
metadata: dict,
cert_path: str | None = None,
private_key_path: str | None = None,
server_cert_path: str | None = None,
):
self.url = url
self.name = name
self.server_uri = server_uri
self.data_queue: dict = {}
self.non_receive_count = 0
self.client: Client | None = None
self.subscription_period_ms = subscription_period_ms
self.cert_path = cert_path
self.private_key_path = private_key_path
self.server_cert_path = server_cert_path
self.nodes: dict = {}
self.subscriptions: dict = {}
self.data_manager = data_manager
self.metadata = metadata
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
metrics.OPC_CONNECTION_STATUS.labels(
pod_id=self.pod_id, server_name=self.name, server_url=self.url
).set(0)
metrics.OPC_TAGS_SUBSCRIBED.labels(pod_id=self.pod_id, server_name=self.name).set(0)
def __str__(self):
"""
String representation of the OPC Manager.
Returns:
str: Human-readable representation showing server details and current state.
"""
return (
f'OpcManager(name={self.name}, url={self.url}, server_uri={self.server_uri})\n'
f'nodes={self.nodes}, subscriptions={self.subscriptions}'
)
async def shutdown(self):
"""
Comprehensive cleanup method for graceful shutdown.
This method ensures proper cleanup of all OPC UA resources:
- Closes active subscriptions
- Disconnects from the OPC server
- Releases allocated resources
Should be called before the application terminates to prevent resource leaks
and ensure clean disconnection from OPC servers.
"""
try:
await self.disconnect()
except Exception as e:
self.logger.error(f'Error during cleanup: {e}')
async def set_security(self):
"""
Configures the security settings for the OPC UA client.
This method sets up the security policy, certificates, and timeouts
required for establishing a secure connection with the OPC UA server.
It implements Basic256 security policy with certificate-based authentication.
Raises:
ValueError: If either the certificate path or private key path is not provided.
Security Settings:
- Security Policy: Basic256
- Secure Channel Timeout: 10,000,000 ms
- Session Timeout: 10,000,000 ms
The method configures:
- Client application URI
- Certificate-based authentication
- Server certificate validation (if provided)
- Connection timeouts for stability
"""
if not all([self.cert_path, self.private_key_path]):
raise ValueError(
'Certificate and private key paths must be provided for secure connection.'
)
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
self.logger.info('Setting security...')
await self.client.set_security(
SecurityPolicyBasic256,
certificate=cert,
private_key=private_key,
server_certificate=server_cert,
)
self.client.secure_channel_timeout = 10000000
self.client.session_timeout = 10000000
async def connect(self):
"""
Establishes a connection to the OPC server.
This method initializes the OPC client using the provided URL and
sets up security if a certificate path is specified. It then
attempts to connect to the server and logs the connection status.
The connection process includes:
1. Client initialization with server URL
2. Security configuration (if certificates are provided)
3. Connection establishment
4. Metrics recording for monitoring
Raises:
Exception: If the connection to the OPC server fails.
Metrics:
- OPC_CONNECTIONS_TOTAL: Incremented on connection attempt
- OPC_CONNECTION_STATUS: Set to 1 on successful connection
"""
await self.emit_metric(
metric_object=metrics.OPC_CONNECTIONS_TOTAL,
tags={
'pod_id': self.pod_id,
'server_name': self.name,
},
)
try:
self.client = Client(self.url, timeout=10, watchdog_intervall=3600000)
assert self.client is not None # Informa ao mypy que client não é None
self.client.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
if self.cert_path:
await self.set_security()
self.logger.info(f'Starting connection to {self.name}...')
await self.client.connect()
await self.emit_metric(
metric_object=metrics.OPC_CONNECTION_STATUS,
method='set',
value=1,
tags={
'pod_id': self.pod_id,
'server_name': self.name,
'server_url': self.url,
},
)
self.logger.info(f'Connection to {self.name} successful.')
except Exception:
await self.disconnect()
raise
async def create_subscription(self, name: str):
"""
Creates a subscription with the specified monitoring period.
This method establishes a subscription to monitor data changes or events
from the OPC UA server. If the client is not connected, an exception is raised.
Args:
name (str): The name identifier for the subscription
period (int, optional): The monitoring period in milliseconds. Defaults to 500 ms.
Raises:
ValueError: If the client is not connected.
Side Effects:
- Sets the `self.period` attribute to the specified or default period.
- Creates a subscription and assigns it to `self.subscriptions[name]`.
- Logs the creation of the subscription.
- Increments subscription creation metrics.
"""
if not self.client:
raise ValueError('Client not connected. Call connect first.')
try:
self.subscriptions[name] = await self.client.create_subscription(
self.subscription_period_ms, self
)
self.logger.info(f'Subscription {name} created on {self.name}.')
await self.emit_metric(
metric_object=metrics.OPC_SUBSCRIPTIONS_CREATED,
method='inc',
value=1,
tags={
'pod_id': self.pod_id,
'server_name': self.name,
'slot_name': name,
},
)
except Exception as e:
self.logger.error(f'Failed to create subscription {name} on {self.name}: {e}')
raise
async def subscribe(self, subscription: str, nodes: dict, collect_period: int):
"""
Subscribes to a set of OPC UA nodes for data change notifications.
This method adds the specified nodes to the subscription and configures
their data collection rules based on the provided collection period and
node-specific frequency.
Args:
subscription (str): The name of the subscription to use
nodes (dict): A dictionary where keys are node identifiers and values are
configurations for each node. Each configuration must include a 'frequency'
key indicating the frequency of data collection in Hz.
collect_period (int): The data collection period in seconds.
Raises:
ValueError: If the subscription has not been created by calling
`create_subscription` prior to this method.
Side Effects:
- Updates internal node tracking and cycle rules
- Establishes data change monitoring for specified nodes
- Updates metrics for subscribed tags count
"""
if not self.subscriptions.get(subscription):
raise ValueError('Subscription not created. Call create_subscription first.')
self.logger.info(f'Subscribing to {subscription} on {self.name}...')
self.logger.info(f'Subscribing to nodes: {nodes}')
assert self.client is not None # Informa ao mypy que client não é None
addr_nodes = [self.client.get_node(n) for n in nodes]
self.logger.debug(f'Addr nodes: {addr_nodes}')
self.nodes.update(nodes)
self.logger.debug(f'Nodes: {self.nodes}')
self.collect_period = collect_period
for node, config in self.nodes.items():
self.nodes[node]['cycle_rule'] = {
'cycle_increment': collect_period * 1000 / float(config['frequency']),
'cycle_count': 0,
}
await self.subscriptions[subscription].subscribe_data_change(addr_nodes)
await self.emit_metric(
metric_object=metrics.OPC_TAGS_SUBSCRIBED,
method='set',
value=len(self.nodes),
tags={
'pod_id': self.pod_id,
'server_name': self.name,
},
)
async def unsubscribe(self, subscription: str):
"""
Unsubscribes from a given subscription.
This method removes the specified subscription and cleans up associated
resources. It handles cases where the subscription doesn't exist gracefully.
Args:
subscription (str): The name of the subscription to unsubscribe from.
Logs:
- A warning if the specified subscription does not exist.
- An info message upon successful unsubscription.
Behavior:
- If the subscription exists, it is deleted and removed from the
subscriptions dictionary.
- If the subscription does not exist, no action is taken.
"""
if not self.subscriptions.get(subscription):
self.logger.warning(f"Subscription '{subscription}' not found. Cannot unsubscribe.")
return
await self.subscriptions[subscription].delete()
del self.subscriptions[subscription]
self.logger.info(f'Unsubscribed from {subscription}.')
async def disconnection_fallback(self) -> list:
"""
Tries 5 times to disconnect from the OPC UA server, with a delay of 100ms x try.
"""
assert self.client is not None
error_stack = []
for i in range(5):
try:
self.logger.info(f'Disconnecting from OPC UA server, attempt {i + 1} of 5')
await self.client.disconnect()
return []
except Exception as e:
self.logger.error(
f'Failed to disconnect from OPC UA serve in attempt {i + 1} of 5: {e}'
)
error_stack.append(
{
'attempt': i + 1,
'error': str(e),
'traceback': traceback.format_exc(),
}
)
await asyncio.sleep(0.1 * i)
return error_stack
async def disconnect(self):
"""
Disconnects from the OPC UA server.
This method handles the disconnection process by deleting all subscriptions
and disconnecting the client from the OPC UA server. It logs the disconnection
process and handles any exceptions that may occur during cleanup.
Raises:
Exception: If an error occurs while deleting the subscription or disconnecting
from the OPC UA server, it logs the error details.
Side Effects:
- Deletes all active subscriptions
- Disconnects the OPC client
- Updates connection status metrics
- Clears internal client reference
"""
self.logger.warning('Disconnecting from OPC server')
if self.client is None:
self.logger.warning('Client already disconnected.')
return
try:
for sub in self.subscriptions:
await self.subscriptions[sub].delete()
self.logger.warning('Deleted all subscriptions.')
except Exception as sub_error:
self.logger.error(f'Failed to clean up subscription: {sub_error}')
errors = await self.disconnection_fallback()
if errors:
await self.send_notification_async(
metadata=self.metadata,
notification_id=f'OPC_DISCONNECTION_ERROR_{self.name}',
message=f'Failed to disconnect from OPC UA server {self.name} after 5 attempts',
block='opc_manager',
level=NotificationLevel.ERROR,
attachment_content=json.dumps(errors, indent=4),
)
else:
self.logger.warning('Disconnected from OPC UA server.')
del self.client
self.client = None
await self.emit_metric(
metric_object=metrics.OPC_CONNECTION_STATUS,
method='set',
value=0,
tags={
'pod_id': self.pod_id,
'server_name': self.name,
'server_url': self.url,
},
)
await self.emit_metric(
metric_object=metrics.OPC_TAGS_SUBSCRIBED,
method='set',
value=0,
tags={
'pod_id': self.pod_id,
'server_name': self.name,
},
)
async def datachange_notification(self, node, _val, data):
"""
Handles data change notifications for monitored OPC UA nodes.
This method is triggered when a monitored node's value changes. It processes
the notification, updates internal state, and publishes the data to the
appropriate topics.
Args:
node (NodeId): The OPC UA node that triggered the data change notification.
_val (Any): The new value of the node (unused in this implementation).
data (DataChangeNotification): The data change notification object containing
details about the change.
Behavior:
- Extracts the value and source timestamp from the monitored item.
- Resets the cycle count for the node's cycle rule.
- Resets the non-receive count.
- Constructs a data dictionary containing the tag, tag name, timestamp, and value.
- Publishes the data to all topics associated with the node.
"""
# get data value
monitored_item = data.monitored_item
value = monitored_item.Value.Value.Value
# source_timestamp
source_timestamp = monitored_item.Value.SourceTimestamp.replace(tzinfo=OPC_TIMEZONE)
tag = str(node)
self.logger.debug(
f'Data change notification received for tag:'
f'{tag} after {self.nodes[tag]["cycle_rule"]["cycle_count"]} cycles'
)
data = {
'tag': tag,
'name': self.nodes[str(node)]['tag_name'],
'timestamp': source_timestamp.strftime(DATETIME_FORMAT_WITH_TZ),
'value': value,
}
for topic in self.nodes[tag]['topics']:
await self.data_manager.publish(topic, data)
self.nodes[tag]['cycle_rule']['cycle_count'] = 0
self.non_receive_count = 0
await self.emit_metric(
metric_object=metrics.OPC_CYCLES_WITHOUT_DATA,
method='set',
value=0,
tags={
'pod_id': self.pod_id,
'server_name': self.name,
},
)
async def check_cycles(self):
"""
Checks the cycle counts for all monitored nodes and sends
notifications if thresholds are exceeded.
This method iterates through all monitored nodes and updates their cycle counts based on
configured increments. If a node's cycle count exceeds a threshold (5 cycles), it triggers
a warning notification.
Side Effects:
- Updates cycle counts for all monitored nodes
- Sends warning notifications for nodes exceeding cycle thresholds
"""
for node, config in self.nodes.items():
self.nodes[node]['cycle_rule']['cycle_count'] += config['cycle_rule']['cycle_increment']
if self.nodes[node]['cycle_rule']['cycle_count'] >= 5:
name = config['tag_name']
cycles = self.nodes[node]['cycle_rule']['cycle_count']
await self.send_notification_async(
metadata=self.metadata,
notification_id=f'TAG_{node}:{name}_LISTENNING_STOPPED',
message=f'{cycles} cycles without receive from {node}:{name}',
block='opc_manager',
level=NotificationLevel.WARNING,
)
async def check_opc_listenning(self) -> bool:
"""
Checks the OPC connection and triggers notifications if the connection is lost.
This method monitors the data reception health by tracking cycles without
data. It sends notifications at different thresholds and can trigger
reconnection attempts.
Returns:
bool: True if the connection is lost and reconnection should be attempted,
False otherwise.
Side Effects:
- Increments non-receive count
- Updates metrics for cycles without data
- Sends warning notifications at 5 cycles
- Sends error notifications and triggers reconnection at 15 cycles
"""
self.non_receive_count += 1
await self.emit_metric(
metric_object=metrics.OPC_CYCLES_WITHOUT_DATA,
method='set',
value=self.non_receive_count,
tags={
'pod_id': self.pod_id,
'server_name': self.name,
},
)
if self.non_receive_count >= 5:
await self.send_notification_async(
metadata=self.metadata,
notification_id=f'OPC_LISTENNING_STOPPED__{self.name}',
message=f'{self.non_receive_count} cycles without '
f'receive from OPC {self.name}. Tags: {json.dumps(self.nodes)}',
block='opc_manager',
level=NotificationLevel.ERROR,
)
if self.non_receive_count >= 15:
await self.emit_metric(
metric_object=metrics.OPC_RECONNECTIONS_TOTAL,
tags={
'pod_id': self.pod_id,
'server_name': self.name,
},
)
await self.send_notification_async(
metadata=self.metadata,
notification_id=f'OPC_CONNECTION_RETRY__{self.name}',
message=f'Retrying to connect to server {self.name}',
block='opc_manager',
level=NotificationLevel.ERROR,
)
return True
return False

101
ingestor/managers/server_reader.py Normal file
View File

@@ -0,0 +1,101 @@
"""Per-server read path driven by the IngestorManager (SIENTIAPDE-1974).
Replaces the legacy OpcManager behind the same interface the IngestorManager
already uses (connect / create_subscription / subscribe / unsubscribe /
check_cycles / check_opc_listenning / shutdown), so the slot lease loop is
unchanged. The difference is underneath:
- the connection comes from the shared OpcConnectionPool, so the certificate is
fetched from MinIO by CertProvider (Basic256Sha256) — never from a file shipped
in the Gitea repository;
- readings go through SubscriptionEngine, which fans each value out to every
`names` entry of the slot v2 (one document per collection / tag name / model).
"""
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
import ingestor.metrics as metrics
from ingestor.connection.pool import OpcConnectionPool
from ingestor.engines.subscription import SubscriptionEngine
from ingestor.io.data_sink import DataSink
_CONNECTION_FIELDS = ('url', 'server_uri', 'use_security', 'subscription_period_ms')
class ServerReader:
"""Reads one OPC server: one SubscriptionEngine (its `non_receive_count` is
engine-wide, hence one per server) with one subscription per leased slot."""
def __init__(
self,
name: str,
server_config: dict,
pool: OpcConnectionPool,
sink: DataSink,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
metadata: dict | None = None,
):
self.name = name
self.server_id = str(server_config['server_id'])
self.url = server_config['url']
self.server_cfg = {k: server_config[k] for k in _CONNECTION_FIELDS if k in server_config}
self.pool = pool
self.logger = logger
self.config: dict | None = None
self.engine = SubscriptionEngine(
pool, sink, logger, notification_handler, metrics_controller, metadata
)
def __str__(self):
return f'ServerReader(name={self.name}, server_id={self.server_id}, url={self.url})'
@property
def subscriptions(self) -> dict:
return self.engine.subscriptions
async def connect(self):
"""Open (or reuse) the pooled connection; raises if the server is unreachable
or the certificate can't be fetched, like OpcManager.connect did."""
await self.pool.get(self.server_id, self.server_cfg)
async def create_subscription(self, slot: str):
await self.engine.create_subscription(self.server_id, self.server_cfg, slot)
async def subscribe(self, slot: str, nodes: dict, collect_period: int):
await self.engine.subscribe(slot, nodes, collect_period)
async def unsubscribe(self, slot: str):
await self.engine.unsubscribe(slot)
async def check_cycles(self):
await self.engine.check_cycles()
async def check_opc_listenning(self) -> bool:
return await self.engine.check_opc_listenning(self.server_id)
async def shutdown(self):
"""Drop every subscription and the pooled connection, so the next
initialization reconnects (lost server or changed server config)."""
for slot in list(self.engine.subscriptions):
await self.engine.unsubscribe(slot)
await self.pool.evict(self.server_id)
await self.engine.emit_metric(
metric_object=metrics.OPC_CONNECTION_STATUS,
method='set',
value=0,
tags={
'pod_id': self.engine.pod_id,
'server_name': self.server_id,
'server_url': self.url,
},
)
await self.engine.emit_metric(
metric_object=metrics.OPC_TAGS_SUBSCRIBED,
method='set',
value=0,
tags={'pod_id': self.engine.pod_id, 'server_name': self.server_id},
)