diff --git a/g5k/07_fault_injection_on_processes.ipynb b/g5k/07_fault_injection_on_processes.ipynb
index 5d75a550f80c78c77998adc47ff3d2d68d7d47a6..f0e38156f0ffb8224f473bdec0a7e5c33695782b 100644
--- a/g5k/07_fault_injection_on_processes.ipynb
+++ b/g5k/07_fault_injection_on_processes.ipynb
@@ -27,7 +27,7 @@
"cell_type": "markdown",
"metadata": {},
"source": [
- "## Rabbitmq / Cron / Pgrep\n",
+ "## Rabbitmq / Cron\n",
"\n",
"[RabbitMQ](https://www.rabbitmq.com/) is an open-source message broker that enables different software applications to communicate and exchange data in a reliable and scalable manner. It follows the Advanced Message Queuing Protocol (AMQP) and provides a flexible messaging model based on the concept of queues.\n",
"\n",
@@ -36,12 +36,7 @@
"\n",
"[Cron](https://man7.org/linux/man-pages/man8/cron.8.html) is a time-based job scheduler in Unix-like operating systems. It allows you to schedule and automate the execution of commands or scripts at specified intervals or specific times. Cron is commonly used for repetitive tasks, system maintenance, and scheduling periodic jobs.\n",
"\n",
- "All asynchronous tools shown here are based on cron. Because of that, for each event, the date is of the order of a minute and please take note that if a cron job is setup at HH:MM:50 for the minute after, it will be executed at HH:MM+1:00 (only 10 seconds after).\n",
- "\n",
- "\n",
- "[pgrep](https://man7.org/linux/man-pages/man1/pgrep.1.html) is a command-line utility in Unix-like operating systems that is used to search for and list the process IDs (PIDs) of running processes based on certain criteria. It allows you to find processes by their names, command lines, or other attributes.\n",
- "\n",
- "Each process research is based on pgrep which consider all passed string as a regexp."
+ "All asynchronous tools shown here are based on cron. Because of that, we can't schedule an event that will happen before the next minute, after this delay, we can be precise to the second (with a final accuracy of 1/2 seconds)."
]
},
{
@@ -185,17 +180,9 @@
"metadata": {},
"outputs": [],
"source": [
- "nproducer = 3\n",
- "nconsumer = 3\n",
- "\n",
"username_monitoring = \"user\"\n",
"password_monitoring = \"password\"\n",
"\n",
- "username_prod = \"prod\"\n",
- "username_cons = \"cons\"\n",
- "password_prod = \"pwd_prod\"\n",
- "password_cons = \"pwd_cons\"\n",
- "\n",
"# SETUP\n",
"## Server configuration\n",
"with en.actions(roles=roles[\"server\"]) as p:\n",
@@ -210,21 +197,11 @@
" p.command(f\"rabbitmqctl set_user_tags {username_monitoring} administrator\")\n",
" p.command(f\"rabbitmqctl set_permissions {username_monitoring} .* .* .* -p '/'\")\n",
"\n",
- " # For producers\n",
- " for idx in range(nproducer):\n",
- " # Add user's specifications (username + password)\n",
- " p.command(f\"rabbitmqctl add_user {username_prod}_{idx} {password_prod}\")\n",
- " # Allow users to connect to the default vhost ('/')\n",
- " p.command(f\"rabbitmqctl set_permissions {username_prod}_{idx} .* .* .* -p '/'\")\n",
- "\n",
- " # For consumers\n",
- " for idx in range(nconsumer):\n",
- " # Add user's specifications (username + password)\n",
- " p.command(f\"rabbitmqctl add_user {username_cons}_{idx} {password_cons}\")\n",
- " # Allow users to connect to the default vhost ('/')\n",
- " p.command(f\"rabbitmqctl set_permissions {username_cons}_{idx} .* .* .* -p '/'\")\n",
+ " # Allow all connections (no credentials needed for consumers and producers)\n",
+ " p.shell('echo \"loopback_users.guest = false\" | sudo tee -a /etc/rabbitmq/rabbitmq.conf')\n",
"\n",
- " p.command(\"systemctl restart rabbitmq-server\")"
+ " p.command(\"systemctl restart rabbitmq-server\")\n",
+ " "
]
},
{
@@ -303,7 +280,7 @@
"The only purpose of these functions is to facilitate and to make this experiment more readable. Their objectives are :\n",
"- to gather and show general statistics about the current state of the experiment (timestamp, number of received and processed messages, queue depth, number of consumer(s))\n",
"- to clear the experiment (kill all instances of producer(s)/consumer(s) if any, delete all output files if any, purges the rabbitmq queues)\n",
- "- to launch all producer(s) and consumer(s)\n",
+ "- to launch all producer(s) and consumer(s) by specifying the number of each\n",
"- to reset the experiment by going back to its initial state (clean + launch) "
]
},
@@ -318,6 +295,7 @@
"import pandas as pd\n",
"import logging\n",
"from enoslib.log import DisableLogging\n",
+ "from enoslib.config import config_context\n",
"\n",
"\n",
"server = roles[\"server\"][0]\n",
@@ -328,18 +306,19 @@
"\n",
"def _get_queues_info() -> List:\n",
" with DisableLogging(level=logging.ERROR):\n",
- " results = en.run_command(\n",
- " \"rabbitmqctl list_queues -p '/' messages consumers | \"\n",
- " \"awk 'NR>3 {printf \\\"%-15s %-15s\\\\n\\\", $1, $2}'\",\n",
- " task_name=\"Gathering statistics from the queues\",\n",
- " roles=roles[\"server\"],\n",
- " gather_facts=False,\n",
- " on_error_continue=True,\n",
- " )\n",
+ " with config_context(ansible_stdout=\"noop\"):\n",
+ " results = en.run_command(\n",
+ " \"rabbitmqctl list_queues -p '/' messages consumers | \"\n",
+ " \"awk 'NR>3 {printf \\\"%-15s %-15s\\\\n\\\", $1, $2}' ; \"\n",
+ " \"rabbitmqctl list_connections | grep guest | wc -l\",\n",
+ " task_name=\"Gathering statistics from the queues\",\n",
+ " roles=roles[\"server\"],\n",
+ " gather_facts=False,\n",
+ " on_error_continue=True,\n",
+ " )\n",
" queues_info : List = []\n",
" for r in results:\n",
" if r.status == \"FAILED\" or r.rc != 0:\n",
- " queues_info.append([-1, -1])\n",
" continue\n",
" lines = r.stdout.strip().split(\"\\n\")\n",
" for l in lines:\n",
@@ -353,16 +332,36 @@
" results[\"Time\"] = []\n",
" results[\"nb_received_messages\"] = []\n",
" results[\"queue_depth\"] = []\n",
- " results[\"nb_consumer\"] = []\n",
+ " results[\"nb_consumers\"] = []\n",
+ " results[\"nb_producers\"] = []\n",
" for _ in range(duration):\n",
- " results[\"Time\"].append(str(datetime.now()))\n",
- " \n",
+ " results[\"Time\"].append(str(datetime.now().strftime(\"%H:%M:%S\")))\n",
+ "\n",
" queues_info = _get_queues_info()\n",
- " queue_depth = queues_info[0][0]\n",
- " nb_consumer = queues_info[0][1]\n",
- " nb_recv_msg = queues_info[1][0]\n",
+ "\n",
+ " try:\n",
+ " queue_depth = queues_info[0][0]\n",
+ " except IndexError as e:\n",
+ " queue_depth = 0\n",
+ "\n",
+ " try:\n",
+ " nb_consumers = queues_info[0][1]\n",
+ " except IndexError as e:\n",
+ " nb_consumers = 0\n",
+ "\n",
+ " try:\n",
+ " nb_recv_msg = queues_info[1][0]\n",
+ " except IndexError as e:\n",
+ " nb_recv_msg = 0\n",
+ "\n",
+ " try:\n",
+ " nb_producers = int(queues_info[2][0]) - int(nb_consumers)\n",
+ " except IndexError as e:\n",
+ " nb_producers = 0\n",
+ "\n",
" results[\"queue_depth\"].append(int(queue_depth))\n",
- " results[\"nb_consumer\"].append(int(nb_consumer))\n",
+ " results[\"nb_consumers\"].append(int(nb_consumers))\n",
+ " results[\"nb_producers\"].append(int(nb_producers))\n",
" results[\"nb_received_messages\"].append(int(nb_recv_msg))\n",
"\n",
"\n",
@@ -383,24 +382,33 @@
" )\n",
" cleaning_registry.kill(signal.SIGKILL)\n",
"\n",
- " en.run_command(\n",
- " \"rabbitmqctl purge_queue fault_injection & \"\\\n",
- " \"rabbitmqctl purge_queue received_messages & \",\n",
- " task_name=\"purging the queue\",\n",
- " roles=roles[\"server\"],\n",
- " on_error_continue=True,\n",
- " gather_facts=False,\n",
- " )\n",
+ " with DisableLogging(level=logging.ERROR):\n",
+ " with config_context(ansible_stdout=\"noop\"):\n",
+ " en.run_command(\n",
+ " \"rabbitmqctl purge_queue fault_injection & \"\\\n",
+ " \"rabbitmqctl purge_queue received_messages & \",\n",
+ " task_name=\"purging the queue\",\n",
+ " roles=roles[\"server\"],\n",
+ " on_error_continue=True,\n",
+ " gather_facts=False,\n",
+ " )\n",
+ "\n",
+ " en.run_command(\n",
+ " \"crontab -r\",\n",
+ " task_name=\"purging crontab file\",\n",
+ " roles=roles[\"consumer\"] + roles[\"producer\"],\n",
+ " on_error_continue=True,\n",
+ " gather_facts=False,\n",
+ " )\n",
" \n",
- "def launch():\n",
+ "def launch(nconsumer, nproducer):\n",
" \"\"\"\n",
" Launch all consumers and producers.\n",
" \"\"\"\n",
"\n",
" for idx in range(nconsumer):\n",
" en.run_command(\n",
- " f\"python3 /tmp/rabbitmq/consumer.py {idx} {server_ip}\"\n",
- " f\" {username_cons} {password_cons}\",\n",
+ " f\"python3 /tmp/rabbitmq/consumer.py {server_ip}\",\n",
" task_name=f\"run consumer script number {idx}\",\n",
" roles=roles[\"consumer\"],\n",
" background=True,\n",
@@ -410,8 +418,7 @@
"\n",
" for idx in range(nproducer):\n",
" en.run_command(\n",
- " f\"python3 /tmp/rabbitmq/producer.py {idx} {server_ip}\"\n",
- " f\" {username_prod} {password_prod}\",\n",
+ " f\"python3 /tmp/rabbitmq/producer.py {server_ip}\",\n",
" task_name=f\"run producer script number {idx}\",\n",
" roles=roles[\"producer\"],\n",
" background=True,\n",
@@ -419,7 +426,7 @@
" ns=f\"tuto_producer_{idx}\",\n",
" )\n",
"\n",
- "def reset():\n",
+ "def reset(nconsumer, nproducer):\n",
" \"\"\"\n",
" Return to the initial state of the experiment.\n",
"\n",
@@ -435,7 +442,7 @@
" )\n",
"\n",
" clean()\n",
- " launch()\n",
+ " launch(nconsumer, nproducer)\n",
" \n",
" print(\n",
" \"\\n ------------------------------ \",\n",
@@ -529,7 +536,7 @@
"metadata": {},
"outputs": [],
"source": [
- "reset()"
+ "reset(nconsumer = 3, nproducer = 3)"
]
},
{
@@ -543,7 +550,6 @@
" \"tuto_\",\n",
" roles\n",
")\n",
- "registry\n",
"registry_on_consumers = registry.lookup(roles[\"consumer\"])\n",
"registry_on_consumers"
]
@@ -591,7 +597,7 @@
"metadata": {},
"outputs": [],
"source": [
- "reset()"
+ "reset(nconsumer = 3, nproducer = 3)"
]
},
{
@@ -667,7 +673,7 @@
"metadata": {},
"outputs": [],
"source": [
- "reset()"
+ "reset(nconsumer = 3, nproducer = 3)"
]
},
{
@@ -692,10 +698,10 @@
"source": [
"registry_on_consumers.kill_async(\n",
" signum = signal.SIGKILL,\n",
- " time_span = timedelta(minutes=1),\n",
+ " time_span = timedelta(minutes=1, seconds=20),\n",
")\n",
"# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
- "results = get_queues_info_for(25)\n",
+ "results = get_queues_info_for(40)\n",
"results"
]
},
@@ -713,7 +719,7 @@
"metadata": {},
"outputs": [],
"source": [
- "reset()"
+ "reset(nconsumer = 3, nproducer = 3)"
]
},
{
@@ -738,10 +744,10 @@
"source": [
"registry_on_producers.kill_async(\n",
" signum = signal.SIGKILL,\n",
- " time_span = datetime.now() + timedelta(minutes=1),\n",
+ " time_span = datetime.now() + timedelta(minutes=1, seconds=20),\n",
")\n",
"# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
- "results = get_queues_info_for(25)\n",
+ "results = get_queues_info_for(40)\n",
"results"
]
},
@@ -771,7 +777,7 @@
"metadata": {},
"outputs": [],
"source": [
- "reset()"
+ "reset(nconsumer = 3, nproducer = 3)"
]
},
{
@@ -797,11 +803,11 @@
"registry_on_consumers.kill_async_incr(\n",
" signum = signal.SIGKILL,\n",
" number = 2,\n",
- " beginning = datetime.now() + timedelta(minutes=1),\n",
- " interval = timedelta(minutes=1),\n",
+ " beginning = datetime.now() + timedelta(minutes=1, seconds=10),\n",
+ " interval = timedelta(seconds=20),\n",
")\n",
"# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
- "results = get_queues_info_for(50)\n",
+ "results = get_queues_info_for(45)\n",
"results"
]
},
@@ -828,7 +834,7 @@
"cell_type": "markdown",
"metadata": {},
"source": [
- "We can also restart all killed processes without acting on the others. In a way we go back to the registry's initial state. This ```registry.reset()``` has nothing to do with the ```reset()``` specificly implemented for this experiment ! "
+ "We can also restart all killed processes without acting on the others. In a way we go back to the registry's initial state. This ```registry.reset(nconsumer = 3, nproducer = 3)``` has nothing to do with the ```reset(nconsumer = 3, nproducer = 3)``` specificly implemented for this experiment ! "
]
},
{
@@ -873,7 +879,7 @@
"metadata": {},
"outputs": [],
"source": [
- "reset()"
+ "reset(nconsumer = 3, nproducer = 3)"
]
},
{
@@ -898,8 +904,8 @@
"source": [
"registry_on_consumers.restart_async(\n",
" signum = signal.SIGKILL,\n",
- " time_span = timedelta(minutes=1),\n",
- " interval = timedelta(minutes=1),\n",
+ " time_span = timedelta(minutes=1, seconds=10),\n",
+ " interval = timedelta(seconds=20),\n",
")\n",
"# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
"results = get_queues_info_for(45)\n",
@@ -920,7 +926,7 @@
"metadata": {},
"outputs": [],
"source": [
- "reset()"
+ "reset(nconsumer = 3, nproducer = 3)"
]
},
{
@@ -945,14 +951,205 @@
"source": [
"registry_on_producers.restart_async(\n",
" signum = signal.SIGKILL,\n",
- " time_span = datetime.now() + timedelta(minutes=1),\n",
- " interval = timedelta(minutes=1),\n",
+ " time_span = datetime.now() + timedelta(minutes=1, seconds=20),\n",
+ " interval = timedelta(seconds=20),\n",
")\n",
"# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
"results = get_queues_info_for(45)\n",
"results"
]
},
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Fifth example : Planning"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "We can schedule every event of the experiment in one shot using a ```Planning```. Set all the cron may take time depending on the amount of events scheduled, because of that, the first event must happen at least 5 min after the ```PlanningService.deploy()```"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset(nconsumer = 0, nproducer = 0)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "def launch_n_actors_cmd(actor:str, n:int):\n",
+ " cmd = []\n",
+ " for _ in range(n):\n",
+ " cmd.append(f\"python3 /tmp/rabbitmq/{actor}.py {server_ip}\")\n",
+ "\n",
+ " string = \" & \".join(cmd)\n",
+ " return string"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "After 6 minutes, we will launch 20 producers (5 each 10 seconds).\n",
+ "\n",
+ "After 8 minutes, we will launch 100 consumers (20 each 10 seconds).\n",
+ "\n",
+ "After 10 minutes, we will kill all the consumers consumers (20 each 10 seconds)."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "now = datetime.now()\n",
+ "planning = en.Planning()\n",
+ "\n",
+ "\n",
+ "cmd = launch_n_actors_cmd(actor = \"producer\", n = 5)\n",
+ "for m in range(5):\n",
+ " planning.add_event(\n",
+ " en.StartEvent(\n",
+ " date=now+timedelta(minutes=6, seconds=10*m),\n",
+ " host=roles[\"producer\"][0],\n",
+ " cmd=cmd,\n",
+ " ns=f\"tuto_producer_{m}\", \n",
+ " )\n",
+ " )\n",
+ "\n",
+ "\n",
+ "cmd = launch_n_actors_cmd(actor = \"consumer\", n = 5)\n",
+ "for m in range(5):\n",
+ " (\n",
+ " planning.add_event(\n",
+ " en.StartEvent(\n",
+ " date=now+timedelta(minutes=8, seconds=10*m),\n",
+ " host=roles[\"consumer\"][0],\n",
+ " cmd=cmd,\n",
+ " ns=f\"tuto_consumer_1_{m}\", \n",
+ " )\n",
+ " ).add_event(\n",
+ " en.StartEvent(\n",
+ " date=now+timedelta(minutes=8, seconds=10*m),\n",
+ " host=roles[\"consumer\"][0],\n",
+ " cmd=cmd,\n",
+ " ns=f\"tuto_consumer_2_{m}\", \n",
+ " )\n",
+ " ).add_event(\n",
+ " en.StartEvent(\n",
+ " date=now+timedelta(minutes=8, seconds=10*m),\n",
+ " host=roles[\"consumer\"][0],\n",
+ " cmd=cmd,\n",
+ " ns=f\"tuto_consumer_3_{m}\", \n",
+ " )\n",
+ " ).add_event(\n",
+ " en.StartEvent(\n",
+ " date=now+timedelta(minutes=8, seconds=10*m),\n",
+ " host=roles[\"consumer\"][0],\n",
+ " cmd=cmd,\n",
+ " ns=f\"tuto_consumer_4_{m}\", \n",
+ " )\n",
+ " )\n",
+ " )"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "for m in range(5):\n",
+ " (\n",
+ " planning.add_event(\n",
+ " en.KillEvent(\n",
+ " date=now+timedelta(minutes=10, seconds=10*m),\n",
+ " host=roles[\"consumer\"][0],\n",
+ " ns=f\"tuto_consumer_1_{m}\", \n",
+ " )\n",
+ " ).add_event(\n",
+ " en.KillEvent(\n",
+ " date=now+timedelta(minutes=10, seconds=10*m),\n",
+ " host=roles[\"consumer\"][0],\n",
+ " ns=f\"tuto_consumer_2_{m}\", \n",
+ " )\n",
+ " ).add_event(\n",
+ " en.KillEvent(\n",
+ " date=now+timedelta(minutes=10, seconds=10*m),\n",
+ " host=roles[\"consumer\"][0],\n",
+ " ns=f\"tuto_consumer_3_{m}\", \n",
+ " )\n",
+ " ).add_event(\n",
+ " en.KillEvent(\n",
+ " date=now+timedelta(minutes=10, seconds=10*m),\n",
+ " host=roles[\"consumer\"][0],\n",
+ " ns=f\"tuto_consumer_4_{m}\", \n",
+ " )\n",
+ " )\n",
+ " )"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "planning_service = en.PlanningService(planning)\n",
+ "planning_service.deploy()"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "We can have an up-to-date version of the planning, with both current alive and dead processes. The following can be used multiple times instead of ```get_queues_info()``` to see the evolution of the rabbitmq actors."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "planning_service.status()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "results = get_queues_info_for(300)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "results.plot(x=\"Time\", y=[\"nb_received_messages\", \"queue_depth\"], rot=45)\n",
+ "results.plot(x=\"Time\", y=[\"nb_consumers\", \"nb_producers\"], rot=45)"
+ ]
+ },
{
"attachments": {},
"cell_type": "markdown",
diff --git a/g5k/consumer.py b/g5k/consumer.py
index 6ebffc00ec90441aaf197e7bbb3ceb20e2795827..492eb60480c6e9a8fe380d0970884b7a01da9f5a 100644
--- a/g5k/consumer.py
+++ b/g5k/consumer.py
@@ -8,10 +8,10 @@ import pika
# defining what to do when a message is received
def callback(ch, method, properties, body):
body_str = body.decode("utf8").replace("'", '"')
- id_sender, number = body_str.split(";")
+ number = body_str
channel.basic_publish(
- exchange="", routing_key="received_messages", body=str(idx) + ";" + str(number) + ";" + str(id_sender)
+ exchange="", routing_key="received_messages", body=str(number)
)
time.sleep(1.0)
@@ -23,16 +23,10 @@ def callback(ch, method, properties, body):
if __name__ == "__main__":
global idx
- idx = int(sys.argv[1])
- host = sys.argv[2]
- user = sys.argv[3]
- password = sys.argv[4]
-
- username = user + "_" + str(idx)
+ host = sys.argv[1]
# getting a connection to the broker
- credentials = pika.PlainCredentials(username=username, password=password)
- parameters = pika.ConnectionParameters(host, 5672, credentials=credentials)
+ parameters = pika.ConnectionParameters(host, 5672)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
diff --git a/g5k/producer.py b/g5k/producer.py
index 6567c21c0fb3b8a879cb1f5f503af868d67e729a..1d1a27ac3107983f3084cea08f6b9e903dadab36 100644
--- a/g5k/producer.py
+++ b/g5k/producer.py
@@ -6,16 +6,10 @@ import time
import pika
if __name__ == "__main__":
- idx = int(sys.argv[1])
- host = sys.argv[2]
- user = sys.argv[3]
- password = sys.argv[4]
-
- username = user + "_" + str(idx)
+ host = sys.argv[1]
# getting a connection to the broker
- credentials = pika.PlainCredentials(username=username, password=password)
- parameters = pika.ConnectionParameters(host, 5672, credentials=credentials)
+ parameters = pika.ConnectionParameters(host, 5672)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
@@ -26,7 +20,7 @@ if __name__ == "__main__":
while 1:
number = random.randint(0, 100)
channel.basic_publish(
- exchange="", routing_key="fault_injection", body=str(idx) + ";" + str(number)
+ exchange="", routing_key="fault_injection", body=str(number)
)
time.sleep(1.0)