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

import ingestor.metrics as metrics
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.')

    while not exit_signal.is_set():  # NOSONAR
        start_time = time()  # Start loop timer
        try:
            await ingestor.loop()
            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)

    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()