consumer-scheduler/app.py

#!/usr/bin/env python
import sentry_sdk
import time
import requests
import requests.exceptions
import os
import redis
import json
import logging

"""
Scheduler
"""

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

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


def main():
    logging.basicConfig(filename='', level=logging.DEBUG)

    r = redis.Redis(host='localhost', port=6379, db=0)

    ip_list = os.environ['INITIAL_SERVERS'].split(',')
    logging.DEBUG('Get consumer list from environ at first: Done')
    consumer_list_redis = json.loads((r.get('consumer_list') or b'{}').decode('utf-8'))
    logging.DEBUG('Get consumer list from redis at first: Done')
    temp_dict = {}

    for ip in ip_list:

        try:
            response = requests.post(f"http://{ip}/sync", json={'uuid': os.environ['LOCAL_UUID']})
        except requests.exceptions.ConnectionError:
            continue

        if response.status_code == 200:
            temp_dict[response.json()['uuid']] = {'ip': ip}

    consumer_list_redis.update(temp_dict)
    r.set('consumer_list', json.dumps(consumer_list_redis).encode('utf-8'))
    logging.DEBUG('Update redis consumers ip list from first answers: Done')
    while True:
        logging.DEBUG('Infinite Cycle start : Done')
        consumer_list_redis = json.loads((r.get('consumer_list') or b'{}').decode('utf-8'))
        logging.DEBUG('Get consumer list from redis: Done')
        for uuid, info in consumer_list_redis.items():
            ip = info['ip']
            try:
                response = requests.post(f"http://{ip}/sync", json={'uuid': os.environ['LOCAL_UUID']})
            except requests.exceptions.ConnectionError:
                continue

            if response.status_code == 200:
                temp_dict[response.json()['uuid']] = {'ip': ip}

        consumer_list_redis.update(temp_dict)
        r.set('consumer_list', json.dumps(consumer_list_redis).encode('utf-8'))
        logging.DEBUG('Update redis consumer ip list from answers: Done')

        logging.DEBUG('Waiting for next turn')
        time.sleep(30)

    LOG.ERROR('While get rekt')


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        pass