b2d106e5c4
SonarQube S7504 flagged `for slot in list(self.engine.subscriptions)` in ServerReader.shutdown as an unnecessary list() call. It is necessary: engine.unsubscribe() pops from engine.subscriptions, so iterating the dict directly raises "dictionary changed size during iteration" and leaves the remaining slots subscribed. Kept the snapshot with an explanatory comment and `# NOSONAR`, the same convention as OpcConnectionPool.close_all(). Adds a regression test where unsubscribe() really pops: it fails with RuntimeError without the list() and passes with it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
108 lines
4.4 KiB
Python
108 lines
4.4 KiB
Python
"""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
|
|
from ingestor.io.slots import CONNECTION_FIELDS
|
|
|
|
|
|
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, evict: bool = True):
|
|
"""Drop every subscription of this server.
|
|
|
|
With `evict` (lost server or changed server config) the pooled connection is
|
|
also dropped, so the next initialization reconnects. Without it (the slot just
|
|
moved to another pod) the connection stays for writes/gRPC: the connection is
|
|
shared, and the idle sweep closes it if nobody uses it anymore."""
|
|
# list() snapshots the keys: unsubscribe() pops from engine.subscriptions, so
|
|
# iterating the dict directly would raise "changed size during iteration".
|
|
for slot in list(self.engine.subscriptions): # NOSONAR
|
|
await self.engine.unsubscribe(slot)
|
|
if evict:
|
|
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},
|
|
)
|