Planning service revisited#

# Common configuration
with en.actions(roles=roles) as p:
    p.apt(task_name="Installing python", name="python3")

    p.command("apt update")

    p.apt(task_name="Installing pip", name="python3-pip")
    p.pip(task_name="Installing pika", name="pika")

    p.file(path="/tmp/rabbitmq", state="absent")
    p.file(path="/tmp/rabbitmq", state="directory")

username_monitoring = "user"
password_monitoring = "password"

# SETUP
## Server configuration
with en.actions(roles=roles["server"]) as p:
    # Setting the rabbimq server
    p.apt(task_name="Installing rabbitmq-server", name="rabbitmq-server")
    p.command("rabbitmq-plugins enable rabbitmq_management")
    p.command("systemctl start rabbitmq-server")
    p.command("systemctl enable rabbitmq-server")

    # For the management interface, adding a new admin
    p.command(f"rabbitmqctl add_user {username_monitoring} {password_monitoring}")
    p.command(f"rabbitmqctl set_user_tags {username_monitoring} administrator")
    p.command(f"rabbitmqctl set_permissions {username_monitoring} .* .* .* -p '/'")

    # Allow all connections (no credentials needed for consumers and producers)
    p.shell('echo "loopback_users.guest = false" | sudo tee -a /etc/rabbitmq/rabbitmq.conf')

    p.command("systemctl restart rabbitmq-server")

with en.actions(roles=roles["producer"]) as p:
    p.copy(
        src=HERE + "/producer.py",
        dest="/tmp/rabbitmq/producer.py",
        task_name="copying producer file",
    )

with en.actions(roles=roles["consumer"]) as p:
    p.copy(
        src=HERE + "/consumer.py",
        dest="/tmp/rabbitmq/consumer.py",
        task_name="copying consumer file",
    )

import pandas as pd
import logging
import json

from typing import List

from enoslib.log import DisableLogging
from enoslib.config import config_context
from IPython.display import clear_output


server = roles["server"][0]
ip_address_obj = server.filter_addresses(networks=networks["prod"])[0]
server_ip = ip_address_obj.ip.ip

# list all producer(s) ip(s)
producer_ips : List[str] = []
for p in roles["producer"]:
    ip_address_obj = p.filter_addresses(networks=networks["prod"])[0]
    producer_ips.append(str(ip_address_obj.ip.ip))

# list all consumer(s) ip(s)
consumer_ips : List[str] = []
for c in roles["consumer"]:
    ip_address_obj = c.filter_addresses(networks=networks["prod"])[0]
    consumer_ips.append(str(ip_address_obj.ip.ip))

def _get_queues_info() -> List:
    with DisableLogging(level=logging.ERROR):
        with config_context(ansible_stdout="noop"):
            result = en.run_command(
                "rabbitmqctl list_queues name messages --formatter json; echo '---'; "
                "rabbitmqctl list_connections name --formatter json",
                task_name="Gathering statistics from the queues",
                roles=roles["server"],
                gather_facts=False,
                on_error_continue=True,
            )

    queues_info : List = []
    r = result[0]
    if r.status == "FAILED" or r.rc != 0:
        return

    queues_info, connections = r.stdout.split('---')

    return (json.loads(queues_info), json.loads(connections))

def get_queues_info_for(duration: int) -> pd.DataFrame:
    results = {}
    results["Time"] = []
    results["nb_received_messages"] = []
    results["queue_depth"] = []
    results["nb_consumers"] = []
    results["nb_producers"] = []
    for _ in range(duration):
        time = str(datetime.now().strftime("%H:%M:%S"))
        results["Time"].append(time)

        queues_info, connections = _get_queues_info()

        queue_depth = 0
        nb_consumers = 0
        nb_recv_msg = 0
        nb_producers = 0

        for d in queues_info:
            if d["name"] == "fault_injection":
                queue_depth = int(d["messages"])
            elif d["name"] == "received_messages":
                nb_recv_msg = int(d["messages"])

        for actor in connections:
            actor_ip = actor["name"].split(":")[0]

            if actor_ip in producer_ips:
                nb_producers +=1
            elif actor_ip in consumer_ips:
                nb_consumers +=1

        results["queue_depth"].append(queue_depth)
        results["nb_consumers"].append(nb_consumers)
        results["nb_producers"].append(nb_producers)
        results["nb_received_messages"].append(nb_recv_msg)

        clear_output(wait=False)
        print(
            f"Time : {time}\n"
            f"nb_received_messages : {nb_recv_msg}\n"
            f"queue_depth : {queue_depth}\n"
            f"nb_consumers: {nb_consumers}\n"
            f"nb_producers: {nb_producers}\n"
        )

    df = pd.DataFrame(data=results)

    return df

def clean():
    """
    Kill all previouses launched processes,
    removes all cronjobs,
    purges the queue.
    """
    cleaning_registry = en.ProcessRegistry()
    cleaning_registry.build(
        "{producer,consumer}",
        roles["consumer"] + roles["producer"],
    )
    cleaning_registry.kill(signal.SIGKILL)

    with DisableLogging(level=logging.ERROR):
        with config_context(ansible_stdout="noop"):
            en.run_command(
                "rabbitmqctl purge_queue fault_injection & "\
                "rabbitmqctl purge_queue received_messages & ",
                task_name="purging the queue",
                roles=roles["server"],
                on_error_continue=True,
                gather_facts=False,
            )

            en.run_command(
                "crontab -r",
                task_name="purging crontab file",
                roles=roles["consumer"] + roles["producer"],
                on_error_continue=True,
                gather_facts=False,
            )