sientia-dataops-opc-ingestor/ingestor/app.py

import asyncio
import os
import signal
import traceback
from threading import Event
from time import time

from prometheus_client import start_http_server

try:  # optional: local dev reads .env; in-cluster env comes from Helm (no .env file)
    from dotenv import load_dotenv

    load_dotenv()
except ImportError:
    pass

import ingestor.metrics as metrics
from ingestor.config import GatewayConfig
from ingestor.gateway_runtime import GatewayRuntime
from ingestor.ingestor import Ingestor

exit_signal = Event()
POD_ID = os.getenv('HOSTNAME', 'localhost')


async def main():  # NOSONAR
    """
    Main asynchronous function that orchestrates the OPC Ingestor application.

    This function performs the following operations:
    1. Starts the Prometheus metrics server for monitoring
    2. Initializes the Ingestor instance
    3. Prepares the ingestor (connects to services, acquires slot leases)
    4. Runs the main processing loop until shutdown is requested
    5. Handles graceful shutdown and cleanup

    The main loop continuously:
    - Processes OPC data from subscribed tags
    - Manages slot leases and resource allocation
    - Monitors OPC server connections
    - Records metrics for monitoring and observability

    Environment Variables:
        HOSTNAME: Pod identifier for metrics labeling (default: "localhost")
        HTTP_METRICS_PORT: Port for Prometheus metrics server (default: 9090)

    Raises:
        Exception: If ingestor preparation fails, the application will exit
    """
    start_prometheus_server()
    ingestor = Ingestor()
    try:
        await ingestor.prepare_ingestor()
    except Exception as e:
        metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()  # Increment errors
        ingestor.logger.error(f'Failed to prepare ingestor: {e}')
        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 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,
                metadata=ingestor.metadata,
            )
            await runtime.start()
        except Exception as e:
            runtime = None
            metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()
            ingestor.logger.error(
                f'GatewayRuntime failed to initialize, continuing without gateway: {e}'
            )

    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
            await asyncio.sleep(ingestor.poll_interval)  # NOSONAR

        except KeyboardInterrupt:  # Handle Ctrl+C gracefully
            print('KeyboardInterrupt received. Setting exit_signal flag.')
            exit_signal.set()
        except Exception:
            print('Exception in main loop. Setting exit_signal flag.')
            traceback.print_exc()
            metrics.APP_ERRORS_TOTAL.labels(pod_id=POD_ID).inc()  # Increment errors
            exit_signal.set()
        finally:
            # Record loop duration
            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

    ingestor.logger.info('Main loop exit_signaled.')

    # Give Prometheus a chance to scrape one last time before exiting (optional)
    await asyncio.sleep(5)

    os._exit(0)


def signal_handler(_signum, _frame):
    """
    Signal handler for graceful application shutdown.

    This function handles system signals (SIGINT, SIGTERM, SIGHUP) by setting
    the exit_signal flag, which triggers the main loop to complete its current
    iteration and then shut down gracefully.

    Args:
        _signum: The signal number received
        _frame: The current stack frame (unused)
    """
    print(f'Received signal {_signum}. Setting exit_signal flag.')
    exit_signal.set()


def start_prometheus_server():
    """
    Starts the Prometheus metrics HTTP server.

    This function initializes a Prometheus metrics server on the configured port
    to expose application metrics for monitoring and alerting. The server provides
    metrics about application health, performance, and operational status.

    Environment Variables:
        HTTP_METRICS_PORT: Port number for the metrics server (default: 9090)

    Raises:
        Exception: If the server fails to start, the application will exit
    """
    try:
        port = int(os.getenv('HTTP_METRICS_PORT', 9090))
        start_http_server(port)
        print(f'Prometheus server started on port {port}.')
        metrics.APP_UP.labels(pod_id=POD_ID).set(1)  # Mark app as UP
    except Exception as e:
        print(f'Failed to start Prometheus server: {e}')
        os._exit(1)


def run_async_main():
    """
    Run the async main function with proper event loop setup.

    This function sets up the asyncio event loop and runs the main async function.
    It handles KeyboardInterrupt gracefully and ensures proper cleanup of the event loop.

    The function is designed to work with both direct execution and containerized
    environments, providing consistent behavior across different deployment scenarios.
    """
    try:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        loop.run_until_complete(main())
    except KeyboardInterrupt:
        print('KeyboardInterrupt received in main thread.')
        exit_signal.set()
    finally:
        loop.close()


if __name__ == '__main__':
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGHUP, signal_handler)

    run_async_main()