GoldenPogacsa/consumer-scheduler
GoldenPogacsa/consumer-scheduler
Archived
7
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Fixed stuff
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Pünkösd Marcell 2020-04-17 16:51:27 +02:00
parent aed27cba95
commit d0d6267c74
1 changed files with 27 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
app.py
View File

@ -22,12 +22,11 @@ sentry_sdk.init("https://0a106e104e114bc9a3fa47f9cb0db2f4@sentry.kmlabz.com/10")
def main(): def main():
# set logging preferences # set logging preferences
logging.basicConfig(filename = '', level = logging.DEBUG) logging.basicConfig(filename='', level=logging.DEBUG)
# connect to redis # connect to redis
r = redis.Redis(host = 'localhost', port = 6379, db = 0) r = redis.Redis(host='localhost', port=6379, db=0)
# set initial consumer addresses # set initial consumer addresses
ip_list = os.environ['INITIAL_SERVERS'].split(',') ip_list = os.environ['INITIAL_SERVERS'].split(',')
@ -35,20 +34,20 @@ def main():
# get the dictionary of the currently available consumers # get the dictionary of the currently available consumers
consumer_list_redis = json.loads((r.get('consumer_list') or b'{}').decode('utf-8')) consumer_list_redis = json.loads((r.get('consumer_list') or b'{}').decode('utf-8'))
logging.debug('Get consumer list from redis at first: Done') logging.debug('Get consumer list from redis at first: Done')
temp_dict = { } temp_dict = {}
host_name = socket.gethostname() host_name = socket.gethostname()
current_ip = socket.gethostbyname(host_name) current_ip = socket.gethostbyname(host_name)
for ip in ip_list: for ip in ip_list:
try: try:
# request synchronization # request synchronization
response = requests.post(f"http://{ip}/sync", json = { 'uuid': os.environ['LOCAL_UUID'] }) response = requests.post(f"http://{ip}/sync", json={'uuid': os.environ['LOCAL_UUID']})
except requests.exceptions.ConnectionError: except requests.exceptions.ConnectionError:
continue continue
if response.status_code == 200: if response.status_code == 200:
temp_dict[response.json()['uuid']] = { 'ip': ip } temp_dict[response.json()['uuid']] = {'ip': ip}
consumer_list_redis.update(temp_dict) consumer_list_redis.update(temp_dict)
r.set('consumer_list', json.dumps(consumer_list_redis).encode('utf-8')) r.set('consumer_list', json.dumps(consumer_list_redis).encode('utf-8'))
@ -76,18 +75,30 @@ def main():
logging.debug('Waiting for next turn') logging.debug('Waiting for next turn')
# wait for the next update time # wait for the next update time
if current_ip != socket.gethostbyname(host_name): if current_ip != socket.gethostbyname(host_name):
logging.debug('Check ip : Done') logging.debug('IP changed. Pushing updates...')
current_ip = socket.gethostbyname(host_name) current_ip = socket.gethostbyname(host_name)
keys = redis.keys('producer_*') keys = r.keys('producer_*')
logging.debug('Get producer list from redis: Done')
logging.debug(f'Pushing update to the following producers: ' + ', '.join(k.decode('utf-8') for k in keys))
for key in keys: for key in keys:
ip = redis.get(key) ip = r.get(key)
if not ip: if ip:
ip = ip.decode('utf-8')
else:
continue continue
response = requests.post(f"http://{ip}/ip", json={'uuid': os.environ['LOCAL_UUID'], 'ip': current_ip})
logging.debug(response.status_code) try:
response = requests.post(f"http://{ip}/ip", json={'uuid': os.environ['LOCAL_UUID'], 'ip': current_ip})
logging.debug(f"Pushed update to {key.decode('utf-8')} at {ip}. Response: {response.status_code}")
except requests.exceptions.ConnectionError as e:
logging.warning(f"Could not push update to {key.decode('utf-8')}: {str(e)}")
continue
else:
logging.debug('IP unchanged.')
time.sleep(30) time.sleep(30)