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

SIENTIAPDE-1325

Browse Source 
Refactor Ingestor and Metrics Classes for Enhanced Asynchronous Operations

- Updated IngestorManager, OpcManager, and ResourceManager to utilize asynchronous methods for improved performance.
- Integrated MetricsController into various classes for better observability and monitoring.
- Adjusted unit tests to accommodate the new asynchronous behavior, ensuring proper mocking of async methods.
- Removed deprecated Redis metrics and streamlined resource management logic.
This commit is contained in:
vitor-aignosi
2025-11-03 09:41:49 -03:00
parent 85a371a38b
commit dfd08e883f
11 changed files with 525 additions and 479 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,4 +1,4 @@
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, MagicMock, call, patch
from pytest import fixture, mark
from sientia_do.notifications.models import NotificationLevel
@@ -31,9 +31,12 @@ def ingestor_manager(data_manager_mock, resource_manager_mock):
logger=MagicMock(),
notification_handler=MagicMock(),
metadata=metadata['metadata'],
metrics_controller=MagicMock(),
)
ingestor.send_notification = MagicMock()
ingestor.send_notification_async = AsyncMock()
ingestor.emit_metric = AsyncMock()
return ingestor
@@ -57,6 +60,7 @@ def test___init__(
logger=MagicMock(),
notification_handler=MagicMock(),
metadata=metadata['metadata'],
metrics_controller=MagicMock(),
)
opc_manager_mock.assert_not_called()
@@ -68,6 +72,7 @@ def test___init__(
metadata=metadata['metadata'],
logger=ingestor.logger,
notification_handler=ingestor.notification_handler,
metrics_controller=ingestor.metrics_controller,
)
resource_manager_mock.assert_called_once_with(
host='localhost',
@@ -79,6 +84,7 @@ def test___init__(
notification_handler=ingestor.notification_handler,
username=None,
password=None,
metrics_controller=ingestor.metrics_controller,
)
assert ingestor.poll_interval == 5
assert ingestor.managed_tags == {}
@@ -117,6 +123,7 @@ async def test_initialize_opc_from_config(opc_manager, ingestor_manager):
private_key_path=server_config['private_key_path'],
server_cert_path=server_config['server_cert_path'],
metadata=metadata['metadata'],
metrics_controller=ingestor_manager.metrics_controller,
)
assert result == opc_manager.return_value
@@ -155,6 +162,61 @@ async def test_initialize_opc_from_config_exception(traceback_mock, opc_manager,
)
@mark.asyncio
async def test_shutdown(ingestor_manager):
ingestor_manager.opc_managers = {'server1': AsyncMock(), 'server2': AsyncMock()}
ingestor_manager.data_manager.shutdown = MagicMock()
await ingestor_manager.shutdown()
ingestor_manager.opc_managers['server1'].shutdown.assert_called_once()
ingestor_manager.opc_managers['server2'].shutdown.assert_called_once()
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()
server2 = AsyncMock()
ingestor_manager.opc_managers = {'server1': server1, 'server2': server2}
ingestor_manager.managed_tags = {
'slot1': {'server1': {'config': 'config1'}, 'server2': {'config': 'config2'}}
}
await ingestor_manager.remove_server('server1')
server1.shutdown.assert_called_once()
server2.shutdown.assert_not_called()
assert 'server1' not in ingestor_manager.opc_managers
assert 'server2' in ingestor_manager.opc_managers
assert ingestor_manager.managed_tags == {'slot1': {'server2': {'config': 'config2'}}}
@mark.asyncio
async def test_remove_server_not_found(ingestor_manager):
server1 = AsyncMock()
server2 = AsyncMock()
ingestor_manager.opc_managers = {'server1': server1, 'server2': server2}
ingestor_manager.managed_tags = {
'slot1': {'server1': {'config': 'config1'}, 'server2': {'config': 'config2'}}
}
await ingestor_manager.remove_server('server3')
server1.shutdown.assert_not_called()
server2.shutdown.assert_not_called()
assert 'server1' in ingestor_manager.opc_managers
assert 'server2' in ingestor_manager.opc_managers
assert ingestor_manager.managed_tags == {
'slot1': {'server1': {'config': 'config1'}, 'server2': {'config': 'config2'}}
}
@mark.asyncio
@patch('ingestor.managers.ingestor_manager.OpcManager')
@patch('ingestor.managers.ingestor_manager.metrics')
@@ -174,6 +236,7 @@ async def test_update_opc_servers(metrics, opc_manager, ingestor_manager):
return None
ingestor_manager.initialize_opc_from_config = AsyncMock(side_effect=mock_initialize_from_config)
ingestor_manager.remove_server = AsyncMock()
ingestor_manager.managed_tags = {
'slot1': {
@@ -181,7 +244,10 @@ async def test_update_opc_servers(metrics, opc_manager, ingestor_manager):
'server2': {'config': 'config2'},
'server5': {'config': 'config5'},
},
'slot2': {'server3': {'config': 'config3'}, 'server1': {'config': 'config1'}},
'slot2': {
'server3': {'config': 'config3'},
'server1': {'config': 'config1'},
},
}
mock = AsyncMock(config={'config': 'old_config2'})
@@ -191,121 +257,220 @@ async def test_update_opc_servers(metrics, opc_manager, ingestor_manager):
await ingestor_manager.update_opc_servers()
assert len(ingestor_manager.opc_managers) == 3
ingestor_manager.initialize_opc_from_config.assert_any_call({'config': 'config1'})
ingestor_manager.initialize_opc_from_config.assert_any_call({'config': 'config2'})
ingestor_manager.initialize_opc_from_config.assert_any_call({'config': 'config5'})
assert ingestor_manager.initialize_opc_from_config.call_count == 3
assert ingestor_manager.opc_managers['server1'].config == {'config': 'config1'}
assert ingestor_manager.opc_managers['server2'].config == {'config': 'config2'}
assert ingestor_manager.opc_managers['server3'].config == {'config': 'config3'}
assert 'server4' not in ingestor_manager.opc_managers
assert 'server5' not in ingestor_manager.opc_managers
ingestor_manager.remove_server.assert_any_call('server4')
ingestor_manager.remove_server.assert_any_call('server5')
assert ingestor_manager.remove_server.call_count == 2
assert ingestor_manager.opc_managers['server2'] != mock
metrics.OPC_MANAGERS_ACTIVE.labels.assert_called_once_with(pod_id=ingestor_manager.pod_id)
metrics.OPC_MANAGERS_ACTIVE.labels.return_value.set.assert_called_once_with(
len(ingestor_manager.opc_managers)
ingestor_manager.emit_metric.assert_called_once_with(
metric_object=metrics.OPC_MANAGERS_ACTIVE,
method='set',
value=len(ingestor_manager.opc_managers),
tags={'pod_id': ingestor_manager.pod_id},
)
def test_declare_active(ingestor_manager):
ingestor_manager.resource_manager.ingestor_heartbeat = MagicMock()
ingestor_manager.declare_active()
@mark.asyncio
@patch('ingestor.managers.ingestor_manager.metrics')
async def test_check_opc_servers_integrity_all_healthy(metrics, ingestor_manager):
# Setup mock OPC managers
opc_manager1 = MagicMock()
opc_manager1.check_cycles.return_value = None
opc_manager1.check_opc_listenning.return_value = False
opc_manager1.config = {'config': 'config1'}
opc_manager2 = MagicMock()
opc_manager2.check_cycles.return_value = None
opc_manager2.check_opc_listenning.return_value = False
opc_manager2.config = {'config': 'config2'}
ingestor_manager.opc_managers = {'server1': opc_manager1, 'server2': opc_manager2}
# Mock the initialize_opc_from_config method
ingestor_manager.initialize_opc_from_config = MagicMock()
# Call the method
await ingestor_manager.check_opc_servers_integrity()
# Verify that check_cycles and check_opc_listenning were called for each server
opc_manager1.check_cycles.assert_called_once()
opc_manager1.check_opc_listenning.assert_called_once()
opc_manager2.check_cycles.assert_called_once()
opc_manager2.check_opc_listenning.assert_called_once()
# Verify that no reinitialization was needed
ingestor_manager.initialize_opc_from_config.assert_not_called()
ingestor_manager.emit_metric.assert_called_once_with(
metric_object=metrics.OPC_MANAGERS_ACTIVE,
method='set',
value=len(ingestor_manager.opc_managers),
tags={'pod_id': ingestor_manager.pod_id},
)
@mark.asyncio
async def test_check_opc_servers_integrity_server_lost(ingestor_manager):
# Setup mock OPC manager that will be lost
opc_manager = AsyncMock()
opc_manager.check_cycles.return_value = None
opc_manager.check_opc_listenning.return_value = True # Server is lost
opc_manager.config = {'config': 'config1'}
ingestor_manager.opc_managers = {'server1': opc_manager}
ingestor_manager.managed_tags = {'slot1': MagicMock()}
# Call the method
await ingestor_manager.check_opc_servers_integrity()
assert 'server1' not in ingestor_manager.opc_managers
ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with('server1', None)
def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager):
# Setup mock OPC manager that will be lost
opc_manager = MagicMock()
opc_manager.check_cycles.return_value = None
opc_manager.check_opc_listenning.return_value = True # Server is lost
opc_manager.config = {'config': 'config1'}
ingestor_manager.opc_managers = {'server1': opc_manager}
# Setup managed tags
ingestor_manager.managed_tags = {
'slot1': {'server1': {'config': 'config1', 'tags': {'tag1': 'value1'}}}
}
# Mock the initialize_opc_from_config method to return a new manager
new_manager = MagicMock()
ingestor_manager.initialize_opc_from_config = MagicMock(return_value=new_manager)
# Call the method
ingestor_manager.check_opc_servers_integrity()
@mark.asyncio
async def test_declare_active(ingestor_manager):
ingestor_manager.resource_manager.ingestor_heartbeat = AsyncMock()
await ingestor_manager.declare_active()
ingestor_manager.resource_manager.ingestor_heartbeat.assert_called_once()
def test_get_active_ingestors(ingestor_manager):
ingestor_manager.resource_manager.get_all_ingestors = MagicMock()
ingestor_manager.get_active_ingestors()
@mark.asyncio
async def test_get_active_ingestors(ingestor_manager):
ingestor_manager.resource_manager.get_all_ingestors = AsyncMock()
await ingestor_manager.get_active_ingestors()
ingestor_manager.resource_manager.get_all_ingestors.assert_called_once()
def test_get_active_ingestors_empty(ingestor_manager):
ingestor_manager.resource_manager.get_all_ingestors = MagicMock(return_value=None)
result = ingestor_manager.get_active_ingestors()
@mark.asyncio
async def test_get_active_ingestors_empty(ingestor_manager):
ingestor_manager.resource_manager.get_all_ingestors = AsyncMock(return_value=None)
result = await ingestor_manager.get_active_ingestors()
assert result == []
ingestor_manager.resource_manager.get_all_ingestors.assert_called_once()
@patch('ingestor.managers.ingestor_manager.metrics')
def test_get_number_of_leases_success(metrics, ingestor_manager):
ingestor_manager.resource_manager.get_all_leases = MagicMock(return_value=['lease1', 'lease2'])
result = ingestor_manager.get_number_of_leases()
@mark.asyncio
async def test_get_number_of_leases_success(metrics, ingestor_manager):
ingestor_manager.resource_manager.get_all_leases = AsyncMock(return_value=['lease1', 'lease2'])
result = await ingestor_manager.get_number_of_leases()
assert result == 2
ingestor_manager.emit_metric.assert_called_once_with(
metric_object=metrics.LEASES_TOTAL,
method='set',
value=2,
tags={'pod_id': ingestor_manager.pod_id},
)
ingestor_manager.resource_manager.get_all_leases.assert_called_once()
metrics.LEASES_TOTAL.set.assert_called_once_with(2)
@patch('ingestor.managers.ingestor_manager.metrics')
def test_get_number_of_leases_empty(metrics, ingestor_manager):
ingestor_manager.resource_manager.get_all_leases = MagicMock(return_value=None)
result = ingestor_manager.get_number_of_leases()
assert result == 0
ingestor_manager.resource_manager.get_all_leases.assert_called_once()
metrics.LEASES_TOTAL.set.assert_called_once_with(0)
@patch('ingestor.managers.ingestor_manager.metrics')
def test_get_number_of_slots_success(metrics, ingestor_manager):
ingestor_manager.resource_manager.get_all_slots = MagicMock(return_value=['slot1', 'slot2'])
result = ingestor_manager.get_number_of_slots()
@mark.asyncio
async def test_get_number_of_slots_success(metrics, ingestor_manager):
ingestor_manager.resource_manager.get_all_slots = AsyncMock(return_value=['slot1', 'slot2'])
result = await ingestor_manager.get_number_of_slots()
assert result == 2
ingestor_manager.resource_manager.get_all_slots.assert_called_once()
metrics.SLOTS_TOTAL.set.assert_called_once_with(2)
ingestor_manager.emit_metric.assert_called_once_with(
metric_object=metrics.SLOTS_TOTAL,
method='set',
value=2,
tags={'pod_id': ingestor_manager.pod_id},
)
@patch('ingestor.managers.ingestor_manager.metrics')
def test_get_number_of_slots_empty(metrics, ingestor_manager):
ingestor_manager.resource_manager.get_all_slots = MagicMock(return_value=None)
result = ingestor_manager.get_number_of_slots()
@mark.asyncio
async def test_get_number_of_slots_empty(metrics, ingestor_manager):
ingestor_manager.resource_manager.get_all_slots = AsyncMock(return_value=None)
result = await ingestor_manager.get_number_of_slots()
assert result == 0
ingestor_manager.resource_manager.get_all_slots.assert_called_once()
metrics.SLOTS_TOTAL.set.assert_called_once_with(0)
ingestor_manager.emit_metric.assert_called_once_with(
metric_object=metrics.SLOTS_TOTAL,
method='set',
value=0,
tags={'pod_id': ingestor_manager.pod_id},
)
def test_get_slot_leases_1_success(ingestor_manager):
ingestor_manager.resource_manager.lease_tag = MagicMock(return_value=True)
ingestor_manager.resource_manager.get_tag_slot = MagicMock(return_value={'tags': ['tag1']})
@mark.asyncio
async def test_get_slot_leases_1_success(ingestor_manager):
ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=True)
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(return_value={'tags': ['tag1']})
ingestor_manager.number_of_slots = 1
result = ingestor_manager.get_slot_leases()
result = await ingestor_manager.get_slot_leases()
assert result == {'1': {'tags': ['tag1']}}
def test_get_slot_leases_2_success(ingestor_manager):
ingestor_manager.resource_manager.lease_tag = MagicMock(side_effect=[True, True])
ingestor_manager.resource_manager.get_tag_slot = MagicMock(
@mark.asyncio
async def test_get_slot_leases_2_success(ingestor_manager):
ingestor_manager.resource_manager.lease_tag = AsyncMock(side_effect=[True, True])
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
side_effect=[{'tags': ['tag1']}, {'tags': ['tag2']}]
)
ingestor_manager.number_of_slots = 2
result = ingestor_manager.get_slot_leases(max_slots=2)
result = await ingestor_manager.get_slot_leases(max_slots=2)
assert result == {'1': {'tags': ['tag1']}, '2': {'tags': ['tag2']}}
def test_get_slot_leases_2_1_none(ingestor_manager):
ingestor_manager.resource_manager.lease_tag = MagicMock(side_effect=[True, True])
ingestor_manager.resource_manager.get_tag_slot = MagicMock(
@mark.asyncio
async def test_get_slot_leases_2_1_none(ingestor_manager):
ingestor_manager.resource_manager.lease_tag = AsyncMock(side_effect=[True, True])
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
side_effect=[None, {'tags': ['tag1']}]
)
ingestor_manager.number_of_slots = 1
result = ingestor_manager.get_slot_leases(max_slots=1)
result = await ingestor_manager.get_slot_leases(max_slots=1)
assert result == {}
def test_get_slot_leases_1_failure(ingestor_manager):
ingestor_manager.resource_manager.lease_tag = MagicMock(return_value=False)
ingestor_manager.resource_manager.get_tag_slot = MagicMock(return_value={'tags': ['tag1']})
@mark.asyncio
async def test_get_slot_leases_1_failure(ingestor_manager):
ingestor_manager.resource_manager.lease_tag = AsyncMock(return_value=False)
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(return_value={'tags': ['tag1']})
result = ingestor_manager.get_slot_leases()
result = await ingestor_manager.get_slot_leases()
ingestor_manager.resource_manager.get_tag_slot.assert_not_called()
assert result == {}
@@ -329,22 +494,20 @@ async def test_unsubscribe_slot(ingestor_manager):
ingestor_manager.opc_managers['server3'].unsubscribe.assert_not_called()
def test_update_slot_config(ingestor_manager):
@mark.asyncio
async def test_update_slot_config(ingestor_manager):
ingestor_manager.managed_tags = {
'slot1': {'config': 'old_config'},
'slot2': {'config': 'new_config'},
'slot3': {'config': 'old_config'},
}
ingestor_manager.resource_manager.get_tag_slot = MagicMock(
ingestor_manager.resource_manager.get_tag_slot = AsyncMock(
side_effect=[{'config': 'updated_config'}, {'config': 'new_config'}, None]
)
ingestor_manager.resource_manager.renew_tag_lease = AsyncMock()
ingestor_manager.update_opc_servers = MagicMock()
ingestor_manager.subscribe_to_tags = MagicMock()
ingestor_manager.unsubscribe_slot = MagicMock()
ingestor_manager.update_slot_config()
await ingestor_manager.update_slot_config()
assert ingestor_manager.managed_tags['slot1'] == {'config': 'updated_config'}
assert ingestor_manager.managed_tags['slot2'] == {'config': 'new_config'}
@@ -357,15 +520,30 @@ def test_update_slot_config(ingestor_manager):
@patch('ingestor.managers.ingestor_manager.metrics')
def test_drop_slot_leases(metrics, ingestor_manager):
ingestor_manager.resource_manager.drop_tag_lease = MagicMock()
ingestor_manager.drop_slot_leases(['1', '2'])
@mark.asyncio
async def test_drop_slot_leases(metrics, ingestor_manager):
ingestor_manager.resource_manager.drop_tag_lease = AsyncMock()
await ingestor_manager.drop_slot_leases(['1', '2'])
ingestor_manager.resource_manager.drop_tag_lease.assert_any_call('1')
ingestor_manager.resource_manager.drop_tag_lease.assert_any_call('2')
metrics.SLOTS_RELEASED.labels.assert_any_call(pod_id=ingestor_manager.pod_id)
metrics.SLOTS_RELEASED.labels.return_value.inc.assert_any_call()
ingestor_manager.emit_metric.assert_has_calls(
[
call(
metric_object=metrics.SLOTS_RELEASED,
method='inc',
value=1,
tags={'pod_id': ingestor_manager.pod_id},
),
call(
metric_object=metrics.SLOTS_RELEASED,
method='inc',
value=1,
tags={'pod_id': ingestor_manager.pod_id},
),
]
)
@mark.asyncio
@@ -432,7 +610,7 @@ async def test_manage_server_subscribe_failure(traceback_mock, ingestor_manager)
traceback_mock.format_exc.assert_called_once()
ingestor_manager.send_notification.assert_called_once_with(
ingestor_manager.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='OPC_SUBSCRIPTION_ERROR_slot1:server1',
message="Failed to subscribe to tags from slot1:server1\n{'tags': 'config1'}: Subscription error",
@@ -474,80 +652,3 @@ async def test_subscribe_to_tags(ingestor_manager):
assert ingestor_manager.manage_server.call_count == 3
ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with('server3', None)
@mark.asyncio
@patch('ingestor.managers.ingestor_manager.metrics')
async def test_check_opc_servers_integrity_all_healthy(metrics, ingestor_manager):
# Setup mock OPC managers
opc_manager1 = MagicMock()
opc_manager1.check_cycles.return_value = None
opc_manager1.check_opc_listenning.return_value = False
opc_manager1.config = {'config': 'config1'}
opc_manager2 = MagicMock()
opc_manager2.check_cycles.return_value = None
opc_manager2.check_opc_listenning.return_value = False
opc_manager2.config = {'config': 'config2'}
ingestor_manager.opc_managers = {'server1': opc_manager1, 'server2': opc_manager2}
# Mock the initialize_opc_from_config method
ingestor_manager.initialize_opc_from_config = MagicMock()
# Call the method
await ingestor_manager.check_opc_servers_integrity()
# Verify that check_cycles and check_opc_listenning were called for each server
opc_manager1.check_cycles.assert_called_once()
opc_manager1.check_opc_listenning.assert_called_once()
opc_manager2.check_cycles.assert_called_once()
opc_manager2.check_opc_listenning.assert_called_once()
# Verify that no reinitialization was needed
ingestor_manager.initialize_opc_from_config.assert_not_called()
metrics.OPC_MANAGERS_ACTIVE.labels.assert_called_once_with(pod_id=ingestor_manager.pod_id)
metrics.OPC_MANAGERS_ACTIVE.labels.return_value.set.assert_called_once_with(
len(ingestor_manager.opc_managers)
)
@mark.asyncio
async def test_check_opc_servers_integrity_server_lost(ingestor_manager):
# Setup mock OPC manager that will be lost
opc_manager = AsyncMock()
opc_manager.check_cycles.return_value = None
opc_manager.check_opc_listenning.return_value = True # Server is lost
opc_manager.config = {'config': 'config1'}
ingestor_manager.opc_managers = {'server1': opc_manager}
ingestor_manager.managed_tags = {'slot1': MagicMock()}
# Call the method
await ingestor_manager.check_opc_servers_integrity()
assert 'server1' not in ingestor_manager.opc_managers
ingestor_manager.managed_tags['slot1'].pop.assert_called_once_with('server1', None)
def test_check_opc_servers_integrity_server_lost_with_tags(ingestor_manager):
# Setup mock OPC manager that will be lost
opc_manager = MagicMock()
opc_manager.check_cycles.return_value = None
opc_manager.check_opc_listenning.return_value = True # Server is lost
opc_manager.config = {'config': 'config1'}
ingestor_manager.opc_managers = {'server1': opc_manager}
# Setup managed tags
ingestor_manager.managed_tags = {
'slot1': {'server1': {'config': 'config1', 'tags': {'tag1': 'value1'}}}
}
# Mock the initialize_opc_from_config method to return a new manager
new_manager = MagicMock()
ingestor_manager.initialize_opc_from_config = MagicMock(return_value=new_manager)
# Call the method
ingestor_manager.check_opc_servers_integrity()