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

Merge pull request #42 from Aignosi/feature-rc/SIENTIAPDE-1974-release-slot-keyerror

Browse Source 
SIENTIAPDE-1974: fix gateway main loop crash (KeyError) on slot unsubscribe
This commit is contained in:
Vitor Pimentel dos Santos
2026-10-13 13:59:26 -03:00
committed by GitHub
parent 6fa171c338 06caa88ed8
commit 54c854bb0a
6 changed files with 318 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
ingestor/ingestor.py
View File

@@ -268,12 +268,14 @@ class Ingestor(SientiaMonitoring):
overleases = list(self.ingestor_manager.managed_tags.keys())[1:]
await self.ingestor_manager.drop_slot_leases(overleases)
# Unsubscribe before giving the leases back: another pod may take a released
# slot right away, and must not find this pod still reading it
for lease in overleases:
await self.ingestor_manager.unsubscribe_slot(lease)
self.ingestor_manager.managed_tags.pop(lease)
await self.ingestor_manager.drop_slot_leases(overleases)
await self.emit_metric(
metric_object=metrics.SLOTS_MANAGED,
method='set',

89
ingestor/managers/ingestor_manager.py
View File

@@ -538,26 +538,84 @@ class IngestorManager(SientiaMonitoring):
async def unsubscribe_slot(self, slot: str):
"""
Unsubscribes a specific slot from all associated OPC servers.
Unsubscribes a specific slot from every OPC reader that holds a subscription for it.
This method removes all subscriptions for a given slot across all
OPC servers that were managing it. It ensures clean cleanup of
resources when slots are released or reconfigured.
The readers' own subscriptions are the source of truth, not managed_tags: the
main loop removes a slot from managed_tags before update_ingestor_manager asks
to unsubscribe it, both when manage_leases drops an extra lease (which already
unsubscribed it) and when update_slot_config finds the slot gone from Redis
(which did not). Looking the slot up in managed_tags raised KeyError in both
cases, which stopped the main loop and restarted the container; the second case
also left the subscription open on the OPC server. Unsubscribing a slot that no
reader holds is a no-op.
A reader that fails to unsubscribe is handled by _unsubscribe_reader: the other
readers are still unsubscribed and the main loop goes on.
Args:
slot (str): The name of the slot to unsubscribe.
Raises:
KeyError: If the specified slot does not exist in the managed tags.
Side Effects:
- Removes subscriptions from all OPC servers for the specified slot
- Removes the slot's subscription from every OPC reader that has one
- Cleans up subscription resources on the OPC servers
"""
for server in self.managed_tags[slot].keys():
if server in self.opc_managers:
await self.opc_managers[server].unsubscribe(slot)
# list() snapshots the readers: the loop awaits, so a reader added or removed
# meanwhile must not break the iteration. SonarQube flags it as an unnecessary
# list() (S7504); the issue is accepted in SonarQube for this reason.
for server, reader in list(self.opc_managers.items()):
if slot in reader.subscriptions:
await self._unsubscribe_reader(server, reader, slot)
async def _unsubscribe_reader(
self, server: str, reader: ServerReader | OpcManager, slot: str
) -> None:
"""
Unsubscribe one reader from a slot without letting a failure stop the main loop.
A failure (e.g. the legacy OpcManager, whose delete() raises when the server
session is gone) increments OPC_SUBSCRIPTION_ERRORS, sends
OPC_UNSUBSCRIBE_ERROR_<slot>:<server> and is logged. The slot is dropped from the
reader's books, as the v2 engine already does, so it can be subscribed again.
Known limitation: if delete() failed while the session is still alive (e.g. a
timeout), the server-side subscription may keep delivering data until that
session ends (reconnection or pod restart).
Args:
server (str): The reader's key in opc_managers
reader (ServerReader | OpcManager): The OPC reader
slot (str): The slot to unsubscribe
"""
try:
await reader.unsubscribe(slot)
except Exception as e:
await self.emit_metric(
metric_object=metrics.OPC_SUBSCRIPTION_ERRORS,
tags={
'pod_id': self.pod_id,
'server': server,
'slot': slot,
},
)
trace = traceback.format_exc()
await self.send_notification_async(
metadata=self.metadata,
notification_id=f'OPC_UNSUBSCRIBE_ERROR_{slot}:{server}',
message=f'Failed to unsubscribe slot {slot} from {server}: {e}',
block='opc_manager',
level=NotificationLevel.ERROR,
attachment_content=trace,
)
self.logger.error(trace)
reader.subscriptions.pop(slot, None)
async def _unsubscribe_removed_slots(self, slots: list[str]) -> None:
"""Unsubscribe and forget slots whose configuration is gone from Redis."""
for slot in slots:
self.logger.warning(f'Slot {slot} is no longer configured; unsubscribing it')
await self.unsubscribe_slot(slot)
self.managed_tags.pop(slot, None)
async def update_slot_config(self):
"""
@@ -567,7 +625,7 @@ class IngestorManager(SientiaMonitoring):
This method performs the following steps:
1. Renews the lease for each managed slot using the resource manager.
2. Fetches the latest configuration for each slot.
3. Logs and removes slots whose configurations are no longer available.
3. Logs, unsubscribes and removes slots whose configurations are no longer available.
4. Updates the configuration of slots if changes are detected.
5. Unsubscribes and re-subscribes to slots with updated configurations.
6. Removes slots from the managed tags if they are no longer valid.
@@ -600,8 +658,9 @@ class IngestorManager(SientiaMonitoring):
continue
self.managed_tags[slot] = update
for slot in removed_slots:
self.managed_tags.pop(slot, None)
# Unsubscribe right away: leaving it to update_ingestor_manager's snapshot diff
# would leak the subscription if anything in between failed
await self._unsubscribe_removed_slots(removed_slots)
def parse_slot_config(self, slot: str, raw: dict) -> dict | None:
"""
@@ -764,7 +823,7 @@ class IngestorManager(SientiaMonitoring):
self.logger.error(trace)
self.logger.warning(f'Removing subscription from server {server} for slot {slot}')
await self.opc_managers[server].unsubscribe(slot)
await self._unsubscribe_reader(server, self.opc_managers[server], slot)
return 2
return 0

2
ingestor/metrics.py
View File

@@ -93,7 +93,7 @@ OPC_MANAGERS_ACTIVE = Gauge(
)
OPC_SUBSCRIPTION_ERRORS = Counter(
'ingestor_opc_subscription_errors_total',
'Errors when trying to subscribe to OPC tags',
'Errors when trying to subscribe to or unsubscribe from OPC tags',
['pod_id', 'server', 'slot'],
)

20
tests/unit/conftest.py Normal file
View File

@@ -0,0 +1,20 @@
from unittest.mock import AsyncMock, MagicMock
from pytest import fixture
@fixture
def reader_double():
"""Factory of OPC reader doubles holding subscriptions for the given slots.
Like the real readers (ServerReader, OpcManager), `subscriptions` is keyed by slot
and `unsubscribe` drops the slot from it.
"""
def make(*slots):
reader = MagicMock()
reader.subscriptions = {slot: MagicMock() for slot in slots}
reader.unsubscribe = AsyncMock(side_effect=lambda slot: reader.subscriptions.pop(slot))
return reader
return make

169
tests/unit/managers/test_ingestor_manager.py
View File

@@ -1,13 +1,16 @@
import gc
from unittest.mock import AsyncMock, MagicMock, call, patch
from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
from pytest import fixture, mark
from sientia_do.notifications.models import NotificationLevel
import ingestor.metrics as metrics
from ingestor.config import GatewayConfig
from ingestor.connection.pool import OpcConnectionPool
from ingestor.io.slots import parse_slot
from ingestor.managers.ingestor_manager import IngestorManager
from ingestor.managers.opc_manager import OpcManager
from ingestor.managers.server_reader import ServerReader
metadata = {
'metadata': {
@@ -548,21 +551,153 @@ async def test_get_slot_leases_1_failure(ingestor_manager):
@mark.asyncio
async def test_unsubscribe_slot(ingestor_manager):
async def test_unsubscribe_slot(ingestor_manager, reader_double):
ingestor_manager.managed_tags = {
'slot1': {'server1': {'tags': 'config1'}, 'server2': {'tags': 'config2'}},
'slot2': {'server3': {'tags': 'config3'}, 'server1': {'tags': 'config1'}},
}
ingestor_manager.opc_managers = {
'server1': AsyncMock(),
'server2': AsyncMock(),
'server3': AsyncMock(),
'server1': reader_double('slot1', 'slot2'),
'server2': reader_double('slot1'),
'server3': reader_double('slot2'),
}
await ingestor_manager.unsubscribe_slot('slot1')
ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot1')
ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot1')
ingestor_manager.opc_managers['server3'].unsubscribe.assert_not_called()
assert list(ingestor_manager.opc_managers['server1'].subscriptions) == ['slot2']
@mark.asyncio
async def test_unsubscribe_slot_not_in_managed_tags(ingestor_manager, reader_double):
"""The slot is already gone from managed_tags: the readers still holding it are unsubscribed"""
ingestor_manager.managed_tags = {'slot1': {'server1': {'tags': 'config1'}}}
ingestor_manager.opc_managers = {
'server1': reader_double('slot1', 'slot3'),
'server2': reader_double('slot3'),
}
await ingestor_manager.unsubscribe_slot('slot3')
ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot3')
ingestor_manager.opc_managers['server2'].unsubscribe.assert_called_once_with('slot3')
@mark.asyncio
async def test_unsubscribe_slot_twice_is_a_no_op(ingestor_manager, reader_double):
"""A second unsubscribe of the same slot (lease already dropped) does nothing and never raises"""
ingestor_manager.managed_tags = {}
ingestor_manager.opc_managers = {'server1': reader_double('slot2')}
await ingestor_manager.unsubscribe_slot('slot2')
await ingestor_manager.unsubscribe_slot('slot2')
ingestor_manager.opc_managers['server1'].unsubscribe.assert_called_once_with('slot2')
@mark.asyncio
async def test_unsubscribe_slot_reader_failure_does_not_raise(ingestor_manager, reader_double):
"""A reader that fails to unsubscribe is notified and forgotten; the others still go"""
broken = reader_double('slot1')
broken.unsubscribe = AsyncMock(side_effect=ConnectionError('session closed'))
healthy = reader_double('slot1')
ingestor_manager.opc_managers = {'server1': broken, 'server2': healthy}
await ingestor_manager.unsubscribe_slot('slot1')
healthy.unsubscribe.assert_called_once_with('slot1')
assert 'slot1' not in broken.subscriptions
ingestor_manager.emit_metric.assert_called_once_with(
metric_object=metrics.OPC_SUBSCRIPTION_ERRORS,
tags={'pod_id': ingestor_manager.pod_id, 'server': 'server1', 'slot': 'slot1'},
)
ingestor_manager.send_notification_async.assert_called_once_with(
metadata=ingestor_manager.metadata,
notification_id='OPC_UNSUBSCRIBE_ERROR_slot1:server1',
message='Failed to unsubscribe slot slot1 from server1: session closed',
block='opc_manager',
level=NotificationLevel.ERROR,
attachment_content=ANY,
)
@mark.asyncio
async def test_unsubscribe_slot_with_real_readers(ingestor_manager):
"""Contract: both real readers keep `subscriptions` keyed by slot, and
unsubscribe_slot closes the slot's subscription on each of them"""
v2_subscription = AsyncMock()
client = MagicMock(create_subscription=AsyncMock(return_value=v2_subscription))
pool = MagicMock(get=AsyncMock(return_value=client), evict=AsyncMock())
server_reader = ServerReader(
name='kepserver',
server_config={
'server_id': '1',
'url': 'opc.tcp://kepserver:49320',
'subscription_period_ms': 1000.0,
},
pool=pool,
sink=AsyncMock(),
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
metadata={},
)
server_reader.engine.emit_metric = AsyncMock()
await server_reader.create_subscription('slot1')
with patch('ingestor.managers.opc_manager.metrics'):
legacy_reader = OpcManager(
name='legacy',
url='opc.tcp://legacy:4840',
data_manager=AsyncMock(),
subscription_period_ms=1000,
logger=MagicMock(),
server_uri='opc.tcp://legacy:4840',
notification_handler=MagicMock(),
metadata={},
metrics_controller=AsyncMock(),
)
legacy_reader.emit_metric = AsyncMock()
legacy_subscription = AsyncMock()
legacy_reader.client = MagicMock(
create_subscription=AsyncMock(return_value=legacy_subscription)
)
await legacy_reader.create_subscription('slot1')
assert 'slot1' in server_reader.subscriptions
assert 'slot1' in legacy_reader.subscriptions
ingestor_manager.opc_managers = {'kepserver': server_reader, 'legacy': legacy_reader}
await ingestor_manager.unsubscribe_slot('slot1')
v2_subscription.delete.assert_awaited_once()
legacy_subscription.delete.assert_awaited_once()
assert 'slot1' not in server_reader.subscriptions
assert 'slot1' not in legacy_reader.subscriptions
pool.pin.assert_called_once_with('1')
pool.unpin.assert_called_once_with('1')
@mark.asyncio
async def test_update_slot_config_unsubscribes_removed_slot(ingestor_manager, reader_double):
"""A slot deleted from Redis is unsubscribed right away, not left to the main loop"""
ingestor_manager.managed_tags = {
'slot1': parse_slot(raw_slot_v1('Same')),
'slot2': parse_slot(raw_slot_v1('Same')),
}
reader = reader_double('slot1', 'slot2')
ingestor_manager.opc_managers = {'test_server': reader}
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
side_effect=lambda slot: raw_slot_v1('Same') if slot == 'slot1' else None
)
ingestor_manager.resource_manager.renew_tag_lease = AsyncMock()
await ingestor_manager.update_slot_config()
reader.unsubscribe.assert_called_once_with('slot2')
assert list(ingestor_manager.managed_tags) == ['slot1']
assert list(reader.subscriptions) == ['slot1']
@mark.asyncio
@@ -646,6 +781,30 @@ async def test_manage_server_create_subscription_failure(ingestor_manager):
ingestor_manager.opc_managers['server1'].subscribe.assert_not_called()
@mark.asyncio
async def test_manage_server_subscribe_and_unsubscribe_failure(ingestor_manager):
"""subscribe() fails and so does the cleanup unsubscribe (legacy reader, session
gone): manage_server reports the failure and nothing escapes to the main loop"""
reader = AsyncMock()
reader.subscriptions = {}
reader.subscribe.side_effect = Exception('Subscription error')
reader.unsubscribe.side_effect = ConnectionError('session closed')
ingestor_manager.opc_managers = {'server1': reader}
server_config = {'tags': {'tag1': 'config1'}}
result = await ingestor_manager.manage_server('slot1', 'server1', server_config, server_config)
assert result == 2
reader.unsubscribe.assert_called_once_with('slot1')
notified = [
c.kwargs['notification_id'] for c in ingestor_manager.send_notification_async.call_args_list
]
assert notified == [
'OPC_SUBSCRIPTION_ERROR_slot1:server1',
'OPC_UNSUBSCRIBE_ERROR_slot1:server1',
]
@mark.asyncio
async def test_manage_server(ingestor_manager):
ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()}

55
tests/unit/test_ingestor.py
View File

@@ -1,8 +1,10 @@
from copy import deepcopy
from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
from pytest import fixture, mark
from ingestor.ingestor import Ingestor
from ingestor.managers.ingestor_manager import IngestorManager
@patch('ingestor.ingestor.getenv')
@@ -229,6 +231,10 @@ async def test_manage_leases_no_available_slots_extra_sltos(ingestor_manager_sta
ingestor_manager_started.ingestor_manager.drop_slot_leases.assert_called_once_with(
['tag2', 'tag3']
)
# The extra slots are unsubscribed before their leases are given back
calls = [c[0] for c in ingestor_manager_started.ingestor_manager.mock_calls]
assert calls.index('drop_slot_leases') > calls.index('unsubscribe_slot')
assert list(ingestor_manager_started.ingestor_manager.managed_tags) == ['tag1']
@mark.asyncio
@@ -332,3 +338,52 @@ async def test_update_ingestor_manager_no_ingestor_manager(ingestor_manager_star
ingestor_manager_started.ingestor_manager = None
await ingestor_manager_started.update_ingestor_manager({})
def _manager_with_real_unsubscribe(managed_tags, readers):
"""A mocked IngestorManager whose unsubscribe path (unsubscribe_slot and
_unsubscribe_reader) is the real one."""
manager = AsyncMock(update_opc_servers=AsyncMock(), subscribe_to_tags=AsyncMock())
manager.managed_tags = managed_tags
manager.opc_managers = readers
manager.unsubscribe_slot = lambda slot: IngestorManager.unsubscribe_slot(manager, slot)
manager._unsubscribe_reader = lambda server, reader, slot: IngestorManager._unsubscribe_reader(
manager, server, reader, slot
)
return manager
@mark.asyncio
async def test_dropped_extra_lease_does_not_stop_the_loop(ingestor_manager_started, reader_double):
"""Regression: manage_leases drops slot 2, then update_ingestor_manager sees slot 2 in
the snapshot only and unsubscribes it again. That used to raise KeyError and stop
the main loop (container restart)."""
reader = reader_double('1', '2')
ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe(
{'1': {'kepserver': {}}, '2': {'kepserver': {}}}, {'kepserver': reader}
)
snapshot = deepcopy(ingestor_manager_started.ingestor_manager.managed_tags)
await ingestor_manager_started.manage_leases(0, 0, 1)
await ingestor_manager_started.update_ingestor_manager(snapshot)
reader.unsubscribe.assert_called_once_with('2')
assert list(reader.subscriptions) == ['1']
assert list(ingestor_manager_started.ingestor_manager.managed_tags) == ['1']
@mark.asyncio
async def test_slot_removed_from_redis_is_unsubscribed(ingestor_manager_started, reader_double):
"""Regression: update_slot_config drops a slot deleted by the orchestrator from
managed_tags without unsubscribing it; update_ingestor_manager must close its
subscription instead of raising KeyError."""
reader = reader_double('1', '3')
ingestor_manager_started.ingestor_manager = _manager_with_real_unsubscribe(
{'1': {'kepserver': {}}}, {'kepserver': reader}
)
snapshot = {'1': {'kepserver': {}}, '3': {'kepserver': {}}}
await ingestor_manager_started.update_ingestor_manager(snapshot)
reader.unsubscribe.assert_called_once_with('3')
assert list(reader.subscriptions) == ['1']