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

feat(gateway): wire engines+adapters via GatewayRuntime (write/gRPC live; subscription v2 gated on 1974)

Browse Source 
GatewayRuntime builds and owns the F4 object graph (CertProvider/pool ->
WriteEngine/NatsWriteConsumer -> ActiveReadEngine -> optional GrpcServer),
reusing the existing ingestor's logger/notification_handler/metrics_controller
and DataManager (CertProvider shares its Mongo connection instead of opening
a second one). app.py starts it after prepare_ingestor, sweeps idle OPC
connections each loop iteration, and stops it on shutdown. A NATS connect
failure at startup is logged/notified but never crashes the app -- the read
path keeps working and idle_sweep() retries the consumer start.

ingestor.py needed no changes: data_manager is already reachable via
ingestor.ingestor_manager.data_manager.

Also documents the new MinIO/Postgres env vars this wiring reads (via
sientia_do's build_postgres_config for Postgres) in .env.example, and
updates test_app.py's fixture to mock GatewayConfig/GatewayRuntime so the
existing app tests keep exercising only what they mocked before.
This commit is contained in:
PedroHMCosme
2026-07-15 13:51:18 -03:00
parent ff605eea93
commit 391268776c
5 changed files with 371 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
.env.example
View File

@@ -28,3 +28,19 @@ WRITE_QUALITY="true"
WRITE_MIN_PAUSE_S="0.0"
IDLE_CONN_MAX_S="300.0"
# OPC Gateway - MinIO (client cert storage, read by GatewayRuntime/CertProvider)
MINIO_ENDPOINT="localhost:9000"
MINIO_ACCESS_KEY="minioadmin"
MINIO_SECRET_KEY="minioadmin"
MINIO_SECURE="false"
# OPC Gateway - Postgres (write_status updates, read by GatewayRuntime/DataSink
# via sientia_do.utils.connectors_config.build_postgres_config)
POSTGRES_HOST="localhost"
POSTGRES_PORT="5432"
POSTGRES_USER="sientia"
POSTGRES_PASSWORD="sientia"
POSTGRES_DBNAME="sientia"
POSTGRES_MIN_CONNECTIONS="5"
POSTGRES_MAX_CONNECTIONS="20"

21
ingestor/app.py
View File

@@ -8,6 +8,8 @@ from time import time
from prometheus_client import start_http_server
import ingestor.metrics as metrics
from ingestor.config import GatewayConfig
from ingestor.gateway_runtime import GatewayRuntime
from ingestor.ingestor import Ingestor
exit_signal = Event()
@@ -48,10 +50,27 @@ async def main(): # NOSONAR
exit_signal.set()
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.
runtime: GatewayRuntime | None = None
if ingestor.ingestor_manager is not None:
runtime = GatewayRuntime(
config=GatewayConfig.from_env(),
data_manager=ingestor.ingestor_manager.data_manager,
logger=ingestor.logger,
notification_handler=ingestor.notification_handler,
metrics_controller=ingestor.metrics_controller,
metadata=ingestor.metadata,
)
await runtime.start()
while not exit_signal.is_set(): # NOSONAR
start_time = time() # Start loop timer
try:
await ingestor.loop()
if runtime is not None:
await runtime.idle_sweep()
metrics.APP_LOOP_COUNT.labels(pod_id=POD_ID).inc() # Increment loop counter
# Use asyncio.sleep instead of exit_signal.wait for better async compatibility
@@ -70,6 +89,8 @@ async def main(): # NOSONAR
duration = time() - start_time
metrics.APP_LOOP_DURATION.labels(pod_id=POD_ID).observe(duration)
if runtime is not None:
await runtime.stop()
await ingestor.shutdown()
metrics.APP_UP.labels(pod_id=POD_ID).set(0) # Mark app as DOWN

149
ingestor/gateway_runtime.py Normal file
View File

@@ -0,0 +1,149 @@
"""GatewayRuntime (Task 18 F4 capstone): builds and owns the write/active-read/gRPC
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).
# 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).
"""
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.connection.pool import OpcConnectionPool
from ingestor.engines.active_read import ActiveReadEngine
from ingestor.engines.write import WriteEngine
from ingestor.io.data_sink import DataSink
from ingestor.managers.data_manager import DataManager
class GatewayRuntime(SientiaMonitoring):
"""Owns the gateway write/active-read/gRPC object graph. Resilient to NATS
being unreachable at startup: the read path (existing subscription loop,
gRPC on-demand reads) must keep working even if the write path can't
connect yet -- so a NATS connect failure is logged/notified, not raised,
and retried from `idle_sweep()` on each main-loop iteration."""
def __init__(
self,
config: GatewayConfig,
data_manager: DataManager,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
metadata: dict | None = None,
):
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
self.config = config
self.metadata = metadata or {}
self._nats_started = False
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)
postgres = Postgres(
**build_postgres_config(),
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.sink = DataSink(data_manager, postgres)
self.write_engine = WriteEngine(
self.pool,
self.sink,
logger,
notification_handler,
metrics_controller,
write_timestamp=config.write_timestamp,
write_quality=config.write_quality,
min_pause_s=config.write_min_pause_s,
)
self.nats_consumer = NatsWriteConsumer(
config.nats_url,
self.write_engine,
logger,
subject=config.nats_write_subject,
queue=config.nats_queue,
)
self.active_read = ActiveReadEngine(
self.pool, self.sink, logger, notification_handler, metrics_controller
)
# slot_lookup=None for now: address -> names resolution needs slot v2
# (1974); until then ReadTag responses just aren't persisted.
self.grpc_server = (
GrpcServer(self.active_read, logger, port=config.grpc_port, slot_lookup=None)
if config.grpc_enabled
else None
)
async def start(self) -> None:
await self._start_nats()
if self.grpc_server is not None:
await self.grpc_server.start()
self.logger.info(
f'GatewayRuntime started: nats_started={self._nats_started} '
f'grpc_enabled={self.grpc_server is not None}'
)
async def stop(self) -> None:
if self.grpc_server is not None:
await self.grpc_server.stop()
if self._nats_started:
await self.nats_consumer.stop()
self._nats_started = False
await self.pool.close_all()
async def idle_sweep(self) -> None:
await self.pool.release_idle(self.config.idle_conn_max_s)
if not self._nats_started:
await self._start_nats()
async def _start_nats(self) -> None:
if self._nats_started:
return
try:
await self.nats_consumer.start()
self._nats_started = True
except Exception as e:
self.logger.error(f'GatewayRuntime: failed to start NATS write consumer: {e}')
await self.send_notification_async(
metadata=self.metadata,
notification_id='GATEWAY_NATS_CONSUMER_START_FAILED',
message=f'Failed to start NATS write consumer: {e}',
block='gateway_runtime',
level=NotificationLevel.ERROR,
)

29
tests/unit/test_app.py
View File

@@ -43,6 +43,15 @@ def mock_app_env(monkeypatch):
monkeypatch.setattr(app, 'Ingestor', mocks['Ingestor'])
monkeypatch.setattr(asyncio, 'sleep', mocks['asyncio_sleep'])
mocks['GatewayConfig'] = MagicMock()
mocks['GatewayRuntime'] = MagicMock()
monkeypatch.setattr(app, 'GatewayConfig', mocks['GatewayConfig'])
monkeypatch.setattr(app, 'GatewayRuntime', mocks['GatewayRuntime'])
mock_runtime_instance = mocks['GatewayRuntime'].return_value
mock_runtime_instance.start = AsyncMock()
mock_runtime_instance.idle_sweep = AsyncMock()
mock_runtime_instance.stop = AsyncMock()
monkeypatch.setattr(
app.metrics.APP_UP,
'labels',
@@ -77,6 +86,7 @@ def mock_app_env(monkeypatch):
mock_ingestor_instance = mocks['Ingestor'].return_value
mock_ingestor_instance.poll_interval = 0.01
mock_ingestor_instance.logger = MagicMock()
mock_ingestor_instance.ingestor_manager = MagicMock()
# Make async methods async mocks
mock_ingestor_instance.prepare_ingestor = AsyncMock()
@@ -124,6 +134,21 @@ async def test_main_successful_run_one_loop(mock_app_env, capsys):
mock_ingestor_instance.shutdown.assert_called_once()
mock_ingestor_instance.logger.info.assert_any_call('Main loop exit_signaled.')
# GatewayRuntime (Task 18): built once prepare_ingestor succeeds, started before
# the loop, idle-swept each iteration, stopped on shutdown.
mock_app_env['GatewayRuntime'].assert_called_once_with(
config=mock_app_env['GatewayConfig'].from_env.return_value,
data_manager=mock_ingestor_instance.ingestor_manager.data_manager,
logger=mock_ingestor_instance.logger,
notification_handler=mock_ingestor_instance.notification_handler,
metrics_controller=mock_ingestor_instance.metrics_controller,
metadata=mock_ingestor_instance.metadata,
)
mock_runtime_instance = mock_app_env['GatewayRuntime'].return_value
mock_runtime_instance.start.assert_awaited_once()
mock_runtime_instance.idle_sweep.assert_awaited_once()
mock_runtime_instance.stop.assert_awaited_once()
captured = capsys.readouterr()
assert 'Prometheus server started on port 9090.' in captured.out
@@ -297,12 +322,16 @@ async def test_main_prepare_ingestor_failure(mock_app_env, capsys):
mock_exit_signal = mock_app_env['mock_exit_signal']
mock_ingestor_instance.prepare_ingestor.side_effect = Exception('Preparation failed')
mock_ingestor_instance.ingestor_manager = None # prepare failed before assignment
mock_exit_signal.is_set.side_effect = [False, True]
with pytest.raises(OsExitCalledError) as excinfo:
await app.main()
assert excinfo.value.code == 0
# GatewayRuntime must not be built when prepare_ingestor never got a data_manager
mock_app_env['GatewayRuntime'].assert_not_called()
# Verify error metrics were incremented
app.metrics.APP_ERRORS_TOTAL.labels.assert_called_with(pod_id='test_pod')
mock_app_env['metrics_APP_ERRORS_TOTAL_labels_inc'].assert_called_once()

156
tests/unit/test_gateway_runtime.py Normal file
View File

@@ -0,0 +1,156 @@
from unittest.mock import AsyncMock, MagicMock
import pytest
from ingestor import gateway_runtime as gr
from ingestor.config import GatewayConfig
_PATCHED_CLASSES = (
'MinioRepository',
'CertProvider',
'OpcConnectionPool',
'Postgres',
'build_postgres_config',
'DataSink',
'WriteEngine',
'NatsWriteConsumer',
'ActiveReadEngine',
'GrpcServer',
)
@pytest.fixture
def mocks(monkeypatch):
patched = {}
for name in _PATCHED_CLASSES:
m = MagicMock()
monkeypatch.setattr(gr, name, m)
patched[name] = m
patched['build_postgres_config'].return_value = {
'host': 'localhost', 'port': 5432, 'user': 'u', 'password': 'p',
'dbname': 'd', 'min_connections': 5, 'max_connections': 20,
}
patched['OpcConnectionPool'].return_value.close_all = AsyncMock()
patched['OpcConnectionPool'].return_value.release_idle = AsyncMock()
patched['NatsWriteConsumer'].return_value.start = AsyncMock()
patched['NatsWriteConsumer'].return_value.stop = AsyncMock()
patched['GrpcServer'].return_value.start = AsyncMock()
patched['GrpcServer'].return_value.stop = AsyncMock()
return patched
def make_runtime(**config_overrides):
return gr.GatewayRuntime(
config=GatewayConfig(**config_overrides),
data_manager=MagicMock(),
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=MagicMock(),
)
def test_builds_object_graph_with_config_flags(mocks):
runtime = make_runtime(
write_timestamp=False, write_quality=False, write_min_pause_s=0.25,
cert_bucket='my-bucket', cert_cache_dir='certs-cache', grpc_enabled=False,
)
mocks['OpcConnectionPool'].assert_called_once()
mocks['DataSink'].assert_called_once()
mocks['WriteEngine'].assert_called_once()
_, kwargs = mocks['WriteEngine'].call_args
assert kwargs['write_timestamp'] is False
assert kwargs['write_quality'] is False
assert kwargs['min_pause_s'] == 0.25
_, cert_kwargs = mocks['CertProvider'].call_args
assert cert_kwargs['bucket'] == 'my-bucket'
assert cert_kwargs['cache_dir'] == 'certs-cache'
mocks['GrpcServer'].assert_not_called()
assert runtime.grpc_server is None
def test_builds_grpc_server_when_enabled(mocks):
runtime = make_runtime(grpc_enabled=True, grpc_port=12345)
mocks['GrpcServer'].assert_called_once()
assert runtime.grpc_server is mocks['GrpcServer'].return_value
@pytest.mark.asyncio
async def test_start_starts_nats_consumer(mocks):
runtime = make_runtime(grpc_enabled=False)
await runtime.start()
mocks['NatsWriteConsumer'].return_value.start.assert_awaited_once()
mocks['GrpcServer'].return_value.start.assert_not_called()
assert runtime._nats_started is True
@pytest.mark.asyncio
async def test_start_also_starts_grpc_when_enabled(mocks):
runtime = make_runtime(grpc_enabled=True)
await runtime.start()
mocks['NatsWriteConsumer'].return_value.start.assert_awaited_once()
mocks['GrpcServer'].return_value.start.assert_awaited_once()
@pytest.mark.asyncio
async def test_nats_start_failure_does_not_raise_and_idle_sweep_retries(mocks):
runtime = make_runtime()
runtime.send_notification_async = AsyncMock()
nats_mock = mocks['NatsWriteConsumer'].return_value
nats_mock.start = AsyncMock(side_effect=[Exception('boom'), None])
await runtime.start() # must not raise
assert runtime._nats_started is False
runtime.logger.error.assert_called_once()
runtime.send_notification_async.assert_awaited_once()
await runtime.idle_sweep()
assert runtime._nats_started is True
assert nats_mock.start.await_count == 2
@pytest.mark.asyncio
async def test_stop_stops_started_components_and_closes_pool(mocks):
runtime = make_runtime(grpc_enabled=True)
await runtime.start()
await runtime.stop()
mocks['GrpcServer'].return_value.stop.assert_awaited_once()
mocks['NatsWriteConsumer'].return_value.stop.assert_awaited_once()
mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once()
@pytest.mark.asyncio
async def test_stop_skips_nats_stop_if_never_started(mocks):
runtime = make_runtime()
runtime.send_notification_async = AsyncMock()
nats_mock = mocks['NatsWriteConsumer'].return_value
nats_mock.start = AsyncMock(side_effect=Exception('boom'))
await runtime.start() # nats fails to start
await runtime.stop()
nats_mock.stop.assert_not_called()
mocks['OpcConnectionPool'].return_value.close_all.assert_awaited_once()
@pytest.mark.asyncio
async def test_idle_sweep_releases_idle_connections(mocks):
runtime = make_runtime(idle_conn_max_s=42.0)
runtime._nats_started = True # isolate from the retry path
await runtime.idle_sweep()
mocks['OpcConnectionPool'].return_value.release_idle.assert_awaited_once_with(42.0)