consumer-scheduler/app.py

#!/usr/bin/env python3
"""
This is the top level module, containing the main application. Launching this file will launch the scheduler part of the consumer application.
"""
import sentry_sdk
import time
import os
import logging

from redis_super_storage import RedisSuperStorage
from communicators import ConsumerCommunicator, ProducerCommunicator
from ip_watchdog import IPWatchdog

__author__ = "@kocsisr"
__copyright__ = "Copyright 2020, GoldenPogácsa Team"
__module_name__ = "app"
__version__text__ = "1"

sentry_sdk.init("https://0a106e104e114bc9a3fa47f9cb0db2f4@sentry.kmlabz.com/10")


def get_initial_ip_list() -> list:
    """
    This method is used to parse the content of INITIAL_SERVERS environment variable.
    The contents of this variable is a list of ip addresses separated by commas.
    This mehod returns a Python native `list` object containing the addreseses provided.

    :return: A list of ip addresses provided by the environmental variable
    """
    ip_list = os.environ['INITIAL_SERVERS'].split(',')
    logging.debug('Initial ip list ' + ", ".join(ip_list))
    return ip_list


def main():
    """
    This is the main method of the scheduler application.

    This method does basically the followings:
     * Sets up logging
     * Creates the RedisSuperStorage object (connecting to Redis database)
     * Sets up communicators and IP watchdog
     * Performs an initial synchronization to all other consumers.
     * Starts the main loop which does roughly the following:
        * Syncrhronizes with all other consumers
        * Check if ip address changed. If yes, then push updates to all producers.
        * Wait `RUN_INTERVAL` seconds (provided by envvar)
    """
    # set logging preferences
    logging.basicConfig(filename='', level=logging.DEBUG)

    redis_storage = RedisSuperStorage(os.environ.get('REDIS_URL', "redis://localhost:6379/0"),
                                      int(os.environ.get("CUSTOMER_TIMEOUT", 30)))

    consumer_communicator = ConsumerCommunicator(redis_storage, bool(os.environ.get('FORCE_IP_OVERRIDE', False)))
    producer_communicator = ProducerCommunicator(redis_storage)
    ip_watchdog = IPWatchdog(redis_storage)

    logging.info("Syncing to initial consumer list")
    for ip in get_initial_ip_list():
        logging.debug(f"Syncing to {ip}")
        consumer_communicator.targeted_snyc(ip)

    while True:
        logging.debug("Doing a sync")
        consumer_communicator.sync_all()

        ip_changed, ipaddr = ip_watchdog.ip_changed()
        if ip_changed:
            producer_communicator.push_ip_update(ipaddr)

        time.sleep(int(os.environ.get("RUN_INTERVAL", 10)))


if __name__ == "__main__":
    try:

        main()
    except KeyboardInterrupt:
        pass