Compare commits

..

No commits in common. "957227b7150bab5758809566b71e835b88ea0905" and "0ab07cb1a24d972bb7431f92f9deb9a67c8cb3a9" have entirely different histories.

6 changed files with 49 additions and 142 deletions

10
app.py
View File

@ -10,7 +10,6 @@ import uuid
import logging
import sentry_sdk
import time
from consumerinformation import ConsumerInformation
from communicator import Communicator
from consumerlocator import ConsumerLocator
from messagesender import MessageSender
@ -43,14 +42,11 @@ if __name__ == "__main__":
"""
LOGGER.info("Producer started")
generateduuid = str(uuid.uuid4())
redisconnector = RedisConnector()
consumerinfomation = ConsumerInformation(redisconnector=redisconnector)
communicator = Communicator(currentconsumer=KNOWNCONSUMER, uuid=generateduuid,
consumerinformation=consumerinfomation)
communicator = Communicator(currentconsumer=KNOWNCONSUMER, uuid=generateduuid)
LOGGER.debug(f"My uuid is {generateduuid}")
messagesender = MessageSender(communicator=communicator, uuid=generateduuid)
consumerlocator = ConsumerLocator(communicator=communicator,
redisconnector=redisconnector)
consumerlocator = ConsumerLocator(uuid=generateduuid, communicator=communicator,
redisconnector=RedisConnector())
while True:
consumerlocator.learnconsumerlist()

View File

@ -7,7 +7,6 @@ Communicator module
import logging
import requests
import requests.exceptions
from consumerinformation import ConsumerInformation
__author__ = "@tormakris"
__copyright__ = "Copyright 2020, GoldenPogácsa Team"
@ -23,16 +22,15 @@ class Communicator:
Class handling low level communication with consumers.
"""
def __init__(self, currentconsumer: str, uuid: str, consumerinformation: ConsumerInformation):
def __init__(self, currentconsumer: str, uuid: str):
"""**Constructor:**
Initializes the object.
:param currentconsumer: the current consumer's IP address as a string
:param consumerlocator: the current consumer's IP address as a string
:param uuid: string typed UUID.
"""
self.currenctconsumer = currentconsumer
self.currenctconsumer=currentconsumer
self.uuid = uuid
self.consumerinformation = consumerinformation
def sendmessage(self, message: str) -> None:
"""Send message to the current consumer. Logs the process.
@ -40,18 +38,13 @@ class Communicator:
:param message: the message of type string that will be sent.
:return: None
"""
currentconsumer = self.currenctconsumer
currentconsumer=self.currenctconsumer
LOGGER.info(f"Sending message to {currentconsumer}")
for consumer in self.consumerinformation.getconsumerlist():
try:
postresponse = requests.post(f'http://{consumer}/log', json={'uuid': self.uuid, 'message': message},
timeout=5)
LOGGER.debug(f"Message status code is:{postresponse.status_code}")
if postresponse.status_code < 300:
return None
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
LOGGER.warning(f"Could not send message to {consumer}")
try:
postresponse=requests.post(f'http://{currentconsumer}/log', json={'uuid': self.uuid, 'message': message}, timeout=5)
LOGGER.debug(f"Message status code is:{postresponse.status_code}")
except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
LOGGER.exception(e) # Fun fact: ez azt jelenti, hogy elveszett az üzenet... ide valami retry kellene inkább más consumerek felé...
def discoveravailableconsumers(self) -> list:
"""Get the list of available consumer from the current primary consumer. Logs the received list.
@ -65,8 +58,8 @@ class Communicator:
LOGGER.info(f"List of currently available consumers: {json}")
return json
except Exception as e:
LOGGER.warning("Could not query available consumer list!")
# LOGGER.exception(e)
LOGGER.error("Could not query available consumer list!")
#LOGGER.exception(e)
return []
def isconsumeravailable(self) -> bool:
@ -79,7 +72,7 @@ class Communicator:
response = requests.get(f'http://{currentconsumer}/consumers', timeout=5)
isavailable = response.status_code == 200
except Exception as e:
# LOGGER.exception(e)
#LOGGER.exception(e)
isavailable = False
LOGGER.info(f"Current consumer availability: {isavailable}")
return isavailable
@ -94,15 +87,15 @@ class Communicator:
response = requests.get(f'http://{consumer}/consumers', timeout=5)
isavailable = response.status_code == 200
except Exception as e:
# LOGGER.exception(e)
#LOGGER.exception(e)
isavailable = False
LOGGER.info(f"Consumer {consumer} availability: {isavailable}")
return isavailable
def set_currentconsumer(self, currenctconsumer) -> None:
def set_currentconsumer(self,currenctconsumer) -> None:
"""Set current consumer
:param currenctconsumer: the consumer's IP address
:return: None
"""
self.currenctconsumer = currenctconsumer
self.currenctconsumer=currenctconsumer

View File

@ -1,37 +0,0 @@
#!/usr/bin/env python
"""
Consumer locator module, that manages the list of consumers.
"""
from redisconnector import RedisConnector
__author__ = "@ricsik52"
__copyright__ = "Copyright 2020, GoldenPogácsa Team"
__module_name__ = "consumerlocator"
__version__text__ = "1"
class ConsumerInformation:
"""
Component responsible for providing high level information about consumers
"""
def __init__(self, redisconnector: RedisConnector):
"""**Constructor:**
Initializes the object.
:param redisconnector: the :class:'redisconnector.RedisConnector' instance that will be used for Redis connection.
"""
self.red = redisconnector
def getconsumerlist(self) -> list:
"""
Gets the list of currently available consumers.
:return:
"""
toreturn = []
for consumer in self.red.consumerlist:
if consumer['State']:
toreturn.append(consumer['Host'])
return toreturn

View File

@ -20,16 +20,18 @@ LOGGER = logging.getLogger(__name__)
class ConsumerLocator:
"""
Component responsible for managing the list of consumers. Requires an instance of :class:`communicator.Communicator`
"""
def __init__(self, communicator: Communicator, redisconnector: RedisConnector):
def __init__(self, uuid: str, communicator: Communicator, redisconnector: RedisConnector):
"""**Constructor:**
Initializes the object.
Gets the known consumer's IP address from the PRODUCER_KNOWNCONSUMER envar.
:param uuid: Not used
:param communicator: the :class:'communicator.Communicator' instance that will be used for the low level communication.
"""
self.red = redisconnector
@ -49,6 +51,7 @@ class ConsumerLocator:
if not recievedconsumerlist:
return
consumer_list = self.red.consumerlist
for recconsumer in recievedconsumerlist:
@ -59,8 +62,7 @@ class ConsumerLocator:
if not contains:
LOGGER.info(f"Learned about new consumer at {recconsumer}")
consumer_list.append(
{"Host": recconsumer, "State": True, "LastOk": datetime.datetime.now().timestamp()})
consumer_list.append({"Host": recconsumer, "State": True, "LastOk": datetime.datetime.now().timestamp()})
self.red.consumerlist = consumer_list
self.updateconsumerlist()
@ -79,8 +81,7 @@ class ConsumerLocator:
for consumer in consumer_list:
if not self.communicator.checkconsumer(consumer["Host"]):
consumer["State"] = False
if datetime.datetime.now() - datetime.datetime.fromtimestamp(consumer["LastOk"]) > datetime.timedelta(
hours=1):
if datetime.datetime.now() - datetime.datetime.fromtimestamp(consumer["LastOk"]) > datetime.timedelta(hours=1):
removelist.append(consumer)
else:
consumer["LastOk"] = datetime.datetime.now().timestamp()

View File

@ -4,8 +4,9 @@
Integration test for the producer module.
"""
import os
import re
from consumerinformation import ConsumerInformation
import datetime
from communicator import Communicator
from consumerlocator import ConsumerLocator
from messagesender import MessageSender
@ -22,14 +23,11 @@ redis_proc = factories.redis_proc(host='cache', port=6379)
redis_db = factories.redisdb('redis_nooproc')
answer = ""
def answermethod(Request):
global answer
answer = Request
return ""
def test_integration(httpserver):
"""
Tests the whole system.
@ -47,14 +45,12 @@ def test_integration(httpserver):
method='POST').respond_with_handler(answermethod)
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnector = RedisConnector()
consumerinfomation = ConsumerInformation(redisconnector=redisconnector)
comm = Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid, consumerinformation=consumerinfomation)
uuid=generateduuid)
messagesender = MessageSender(communicator=comm, uuid=generateduuid)
consumerlocator = ConsumerLocator(communicator=comm,
redisconnector=redisconnector)
consumerlocator = ConsumerLocator(uuid=generateduuid, communicator=comm,
redisconnector=RedisConnector())
#First test method
consumerlocator.learnconsumerlist()
conslist = consumerlocator.red.consumerlist

82
test.py
View File

@ -9,7 +9,6 @@ import consumerlocator
import communicator
import messagesender
import redisconnector
import consumerinformation
from pytest_redis import factories
__author__ = "@tormakris"
@ -29,11 +28,9 @@ def test_generate_string(mocker):
:param mocker: patches the :class:`communicator.Communicator`.
"""
mocker.patch('communicator.Communicator')
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
comm = communicator.Communicator(
currentconsumer="localhost",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
mess = messagesender.MessageSender(communicator=comm, uuid=generateduuid)
msg = mess.randomstring(stringlength=32)
assert isinstance(msg, str)
@ -55,13 +52,9 @@ def test_sendmessage(httpserver):
"test": "ok"})
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
mess = "SENDING"
ret = comm.sendmessage(message=mess)
assert ret is None
@ -75,13 +68,9 @@ def test_send_message(mocker):
:return: None
"""
mocker.patch('communicator.Communicator')
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
redisconnect.currentconsumer = {"Host": f"127.0.0.1", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1", "State": True, "LastOk": "1589479202"}]
comm = communicator.Communicator(
currentconsumer="127.0.0.1",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
mess = messagesender.MessageSender(communicator=comm, uuid=generateduuid)
messa = "SENDING"
msg = mess.sendmessage(message=messa)
@ -102,13 +91,9 @@ def test_discoveravailableconsumers(httpserver):
["10.69.42.1", "10.10.10.10", "10.20.30.40"])
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
ret = comm.discoveravailableconsumers()
assert isinstance(ret, list)
assert ret == ["10.69.42.1", "10.10.10.10", "10.20.30.40"]
@ -128,13 +113,9 @@ def test_isconsumeravailable(httpserver):
["10.69.42.1", "10.10.10.10", "10.20.30.40"])
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
ret = comm.isconsumeravailable()
assert isinstance(ret, bool)
assert ret
@ -145,7 +126,7 @@ def test_isconsumeravailable(httpserver):
comm2 = communicator.Communicator(
currentconsumer="127.0.0.1:69",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
ret3 = comm2.isconsumeravailable()
assert isinstance(ret3, bool)
@ -166,13 +147,9 @@ def test_checkconsumer(httpserver):
["10.69.42.1", "10.10.10.10", "10.20.30.40"])
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
comm = communicator.Communicator(
currentconsumer="127.0.0.1",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
ret = comm.checkconsumer(f"127.0.0.1:{port}")
assert isinstance(ret, bool)
assert ret
@ -183,7 +160,8 @@ def test_checkconsumer(httpserver):
comm2 = communicator.Communicator(
currentconsumer="127.0.0.1",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
ret3 = comm2.checkconsumer(f"127.0.0.1:{port}")
assert isinstance(ret3, bool)
assert ret3 == False
@ -195,11 +173,9 @@ def test_setcurrentconsumer():
:return: None
"""
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
comm = communicator.Communicator(
currentconsumer="127.0.0.1",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
comm.set_currentconsumer("10.69.42.1")
assert comm.currenctconsumer == "10.69.42.1"
@ -218,16 +194,12 @@ def test_learnconsumerlist(httpserver):
["10.69.42.1", "10.10.10.10", "10.20.30.40"])
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
locator = consumerlocator.ConsumerLocator(
communicator=comm,
uuid=generateduuid, communicator=comm,
redisconnector=redisconnector.RedisConnector())
ret = locator.learnconsumerlist()
assert ret is None
@ -241,13 +213,11 @@ def test_getcurrentconsumer(mocker):
:return: None
"""
mocker.patch('communicator.Communicator')
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
comm = communicator.Communicator(
currentconsumer="127.0.0.1",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
locator = consumerlocator.ConsumerLocator(
communicator=comm,
uuid=generateduuid, communicator=comm,
redisconnector=redisconnector.RedisConnector())
assert locator.getcurrentconsumer() == consumerlocator.KNOWNCONSUMER
@ -266,17 +236,13 @@ def test_checkcurrentconsumer(httpserver):
["10.69.42.1", "10.10.10.10", "10.20.30.40"])
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
locator = consumerlocator.ConsumerLocator(
communicator=comm,
uuid=generateduuid, communicator=comm,
redisconnector=redisconnector.RedisConnector())
redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
ret = locator.checkcurrentconsumer()
assert ret == True
@ -295,18 +261,14 @@ def test_updateconsumer(httpserver):
["10.69.42.1", "10.10.10.10", "10.20.30.40"])
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
redisconn = redisconnector.RedisConnector()
locator = consumerlocator.ConsumerLocator(
communicator=comm,
uuid=generateduuid, communicator=comm,
redisconnector=redisconn)
redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
assert redisconn.currentconsumer is not None
ret = locator.updateconsumer()
assert ret == f"127.0.0.1:{port}"
@ -326,16 +288,12 @@ def test_updateconsumerlist(httpserver):
["10.69.42.1", "10.10.10.10", "10.20.30.40"])
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
redisconnect = redisconnector.RedisConnector()
consumerinfo = consumerinformation.ConsumerInformation(redisconnector=redisconnect)
comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid, consumerinformation=consumerinfo)
uuid=generateduuid)
consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
locator = consumerlocator.ConsumerLocator(
communicator=comm,
uuid=generateduuid, communicator=comm,
redisconnector=redisconnector.RedisConnector())
redisconnect.currentconsumer = {"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}
redisconnect.consumerlist = [{"Host": f"127.0.0.1:{port}", "State": True, "LastOk": "1589479202"}]
ret = locator.updateconsumerlist()
assert ret is None