diff --git a/g5k/07_fault_injection_on_processes.ipynb b/g5k/07_fault_injection_on_processes.ipynb
new file mode 100644
index 0000000000000000000000000000000000000000..8522f648e6f27221658e04ebe8174522671fa62c
--- /dev/null
+++ b/g5k/07_fault_injection_on_processes.ipynb
@@ -0,0 +1,1027 @@
+{
+ "cells": [
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# Fault-injection on processes\n",
+ "\n",
+ "---\n",
+ "\n",
+ "- Website: https://discovery.gitlabpages.inria.fr/enoslib/index.html\n",
+ "- Instant chat: https://framateam.org/enoslib\n",
+ "- Source code: https://gitlab.inria.fr/discovery/enoslib\n",
+ "\n",
+ "---\n",
+ "\n",
+ "## Prerequisites\n",
+ "\n",
+ "<div class=\"alert alert-block alert-warning\">\n",
+ " Make sure you've run the one time setup for your environment\n",
+ "</div>"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Rabbitmq / Cron / Pgrep\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",
+ "For our experiment, we will deploy a publish / suscribe environment to demonstrate the impact of our api.\n",
+ "\n",
+ "\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.\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."
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Setup"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "import signal\n",
+ "import os\n",
+ "from datetime import datetime, timedelta\n",
+ "from pathlib import Path\n",
+ "\n",
+ "import enoslib as en\n",
+ "\n",
+ "en.init_logging()\n",
+ "en.check()\n"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Reservation"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "CLUSTER = \"nova\"\n",
+ "HERE = os.getcwd()\n",
+ "# claim the resources\n",
+ "conf = (\n",
+ " en.G5kConf.from_settings(\n",
+ " job_name=\"fault-injection tutorial\",\n",
+ " job_type=[],\n",
+ " )\n",
+ " .add_machine(roles=[\"server\"], cluster=CLUSTER, nodes=1)\n",
+ " .add_machine(\n",
+ " roles=[\"producer\"], cluster=CLUSTER, nodes=1\n",
+ " ) # all the producers are running on the same machine\n",
+ " .add_machine(\n",
+ " roles=[\"consumer\"], cluster=CLUSTER, nodes=1\n",
+ " ) # all the consumers are running on the same machine\n",
+ ")\n",
+ "\n",
+ "provider = en.G5k(conf)\n",
+ "\n",
+ "roles, networks = provider.init()\n",
+ "\n",
+ "# Fill in network information from nodes\n",
+ "roles = en.sync_info(roles, networks)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "roles"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Rabbitmq configuration"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "#### Common node's configuration"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "# Common configuration\n",
+ "with en.actions(roles=roles) as p:\n",
+ " p.apt(task_name=\"Installing python\", name=\"python3\")\n",
+ " p.apt(task_name=\"Installing procps\", name=\"procps\")\n",
+ "\n",
+ " p.command(\"apt update\")\n",
+ "\n",
+ " p.apt(task_name=\"Installing pip\", name=\"python3-pip\")\n",
+ " p.pip(task_name=\"Installing pika\", name=\"pika\")\n",
+ "\n",
+ " p.file(path=\"/tmp/rabbitmq\", state=\"absent\")\n",
+ " p.file(path=\"/tmp/rabbitmq\", state=\"directory\")\n"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "#### Server configuration"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "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",
+ " # Setting the rabbimq server\n",
+ " p.apt(task_name=\"Installing rabbitmq-server\", name=\"rabbitmq-server\")\n",
+ " p.command(\"rabbitmq-plugins enable rabbitmq_management\")\n",
+ " p.command(\"systemctl start rabbitmq-server\")\n",
+ " p.command(\"systemctl enable rabbitmq-server\")\n",
+ " \n",
+ " # For the management interface, adding a new admin\n",
+ " p.command(f\"rabbitmqctl add_user {username_monitoring} {password_monitoring}\")\n",
+ " 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",
+ "\n",
+ " p.command(\"systemctl restart rabbitmq-server\")"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "#### Producers' node configuration"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "with en.actions(roles=roles[\"producer\"]) as p:\n",
+ " p.copy(\n",
+ " src=HERE + \"/producer.py\",\n",
+ " dest=\"/tmp/rabbitmq/producer.py\",\n",
+ " task_name=\"copying producer file\",\n",
+ " )"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "#### Consumers' node configuration"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "with en.actions(roles=roles[\"consumer\"]) as p:\n",
+ " p.copy(\n",
+ " src=HERE + \"/consumer.py\",\n",
+ " dest=\"/tmp/rabbitmq/consumer.py\",\n",
+ " task_name=\"copying consumer file\",\n",
+ " )"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "#### Utility functions"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "## Get server's IP address on the private network\n",
+ "from ast import List\n",
+ "import pandas as pd\n",
+ "\n",
+ "\n",
+ "server = roles[\"server\"][0]\n",
+ "ip_address_obj = server.filter_addresses(networks=networks[\"prod\"])[0]\n",
+ "## This may seem weird: ip_address_obj.ip is a `netaddr.IPv4Interface`\n",
+ "## which itself has an `ip` attribute.\n",
+ "server_ip = ip_address_obj.ip.ip\n",
+ "\n",
+ "def get_recv_msg(file: str) -> int:\n",
+ " \"\"\"\n",
+ " Shows the total number of processed messages.\n",
+ " \"\"\"\n",
+ " results = en.run_command(\n",
+ " f\"wc -l {file}\",\n",
+ " task_name=\"getting total number of received messages\",\n",
+ " roles=roles[\"consumer\"],\n",
+ " gather_facts=False,\n",
+ " on_error_continue=True,\n",
+ " )\n",
+ " totalnbmsg = 0\n",
+ " for r in results:\n",
+ " if r.status == \"FAILED\" or r.rc != 0:\n",
+ " print(f\"Actual number of received message : 0\")\n",
+ " continue\n",
+ " _lines = r.stdout.split(\"\\n\")\n",
+ " _total = _lines[-1].strip().split(\" \") # last line contain the total number of line if multiple files, else\n",
+ " totalnbmsg += int(_total[0])\n",
+ "\n",
+ " return totalnbmsg\n",
+ "\n",
+ "def get_queue_size() -> List:\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=\"getting number of messages waiting for processing\",\n",
+ " roles=roles[\"server\"],\n",
+ " gather_facts=False,\n",
+ " on_error_continue=True,\n",
+ " )\n",
+ " for r in results:\n",
+ " if r.status == \"FAILED\" or r.rc != 0:\n",
+ " print(\"Queue is empty\")\n",
+ " continue\n",
+ " lines = r.stdout.strip().split(\"\\n\")\n",
+ " line = lines[0].strip().split(\" \")\n",
+ " return [v for v in line if v!= \"\"]\n",
+ "\n",
+ "def get_stats(duration: int) -> pd.DataFrame:\n",
+ " \"\"\"\n",
+ " Retreive general statistics using the rabbitmq management tool.\n",
+ " \"\"\"\n",
+ " results = {}\n",
+ " results[\"Time\"] = []\n",
+ " results[\"nb_received_messages\"] = []\n",
+ " results[\"queue_depth\"] = []\n",
+ " results[\"nb_consumer\"] = []\n",
+ " for _ in range(duration):\n",
+ " results[\"Time\"].append(str(datetime.now()))\n",
+ " \n",
+ " results[\"nb_received_messages\"].append(get_recv_msg(\"/tmp/rabbitmq/*_output.txt\"))\n",
+ "\n",
+ " queue_depth, nb_consumer = get_queue_size()\n",
+ " results[\"queue_depth\"].append(int(queue_depth))\n",
+ " results[\"nb_consumer\"].append(int(nb_consumer))\n",
+ "\n",
+ "\n",
+ " df = pd.DataFrame(data=results)\n",
+ "\n",
+ " return df\n",
+ "\n",
+ "def clean():\n",
+ " \"\"\"\n",
+ " Kill all previouses launched processes, \n",
+ " removes all previouses results,\n",
+ " purges the queue.\n",
+ " \"\"\"\n",
+ " cleaning_registry = en.ProcessRegistry()\n",
+ " cleaning_registry.build(\n",
+ " \"'python3 /tmp/rabbitmq/'\",\n",
+ " roles[\"consumer\"] + roles[\"producer\"],\n",
+ " )\n",
+ " cleaning_registry.kill(signal.SIGKILL)\n",
+ "\n",
+ " en.run_command(\n",
+ " \"rm /tmp/rabbitmq/*_output.txt\",\n",
+ " task_name=\"cleaning output files\",\n",
+ " roles=roles[\"consumer\"] + roles[\"producer\"],\n",
+ " on_error_continue=True,\n",
+ " gather_facts=False,\n",
+ " )\n",
+ "\n",
+ " en.run_command(\n",
+ " \"rabbitmqctl purge_queue fault_injection\",\n",
+ " task_name=\"purging the queue\",\n",
+ " roles=roles[\"server\"],\n",
+ " on_error_continue=True,\n",
+ " gather_facts=False,\n",
+ " )\n",
+ " \n",
+ "def launch():\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",
+ " task_name=f\"run consumer script number {idx}\",\n",
+ " roles=roles[\"consumer\"],\n",
+ " background=True,\n",
+ " gather_facts=False,\n",
+ " )\n",
+ "\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",
+ " task_name=f\"run producer script number {idx}\",\n",
+ " roles=roles[\"producer\"],\n",
+ " background=True,\n",
+ " gather_facts=False,\n",
+ " )\n",
+ "\n",
+ "def reset():\n",
+ " \"\"\"\n",
+ " Return to the initial state of the experiment.\n",
+ " \"\"\"\n",
+ " print(\n",
+ " \"\\n ------------------------------------ \",\n",
+ " \"\\n| RESETING THE EXPERIMENT PARAMETERS |\",\n",
+ " \"\\n ------------------------------------ \",\n",
+ " )\n",
+ "\n",
+ " clean()\n",
+ " launch()\n",
+ " \n",
+ " print(\n",
+ " \"\\n ------------------------------ \",\n",
+ " \"\\n| DONE - INITIAL STATE REACHED |\",\n",
+ " \"\\n ------------------------------ \",\n",
+ " )"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## General knowledge\n",
+ "\n",
+ "A ```ProcessRegistry``` records all processes that follows a regexp on specific roles. "
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry = en.ProcessRegistry()\n",
+ "registry.build(\"regexp\", roles)"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "One ```Process``` is defined by its pid, its host and its command."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "process = en.Process(1234, roles, \"cmd\")"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "For each case below, we will act on either the consumers, either the producers, never both. It can be the entire group or only a subset.\n",
+ "\n",
+ "When acting on consumers, we will observe an increase of the queue depth, meaning that the messages are not processed as fast as they are produced.\n",
+ "\n",
+ "When acting on consumers, we will observe no evolution regarding the number of processed messages."
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## First example : Synchronous case"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "Using this type of kill, the user must wait for the end before doing anything else."
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Killing all consumers"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers = en.ProcessRegistry()\n",
+ "registry_on_consumers.build(\n",
+ " \"'python3 /tmp/rabbitmq/consumer.py' | grep -v bin\",\n",
+ " roles[\"consumer\"]\n",
+ ")\n",
+ "print(registry_on_consumers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "results = get_stats(5)\n",
+ "results\n",
+ "\n",
+ "registry_on_consumers.kill(signal.SIGKILL)\n",
+ "\n",
+ "results = get_stats(5)\n",
+ "results"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Killing all producers"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_producers = en.ProcessRegistry()\n",
+ "registry_on_producers.build(\n",
+ " \"'python3 /tmp/rabbitmq/producer.py' | grep -v bin\",\n",
+ " roles[\"producer\"]\n",
+ ")\n",
+ "print(registry_on_producers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "results = get_stats(5)\n",
+ "results\n",
+ "registry_on_producers.kill(signal.SIGKILL)\n",
+ "results = get_stats(5)\n",
+ "results"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Second example : asynchronous case with delta"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "Here, each kill is scheduled, we have to specify the delay before they happen.\n",
+ "\n",
+ "\n",
+ "The delay is a ```datetime.timedelta()``` object."
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Killing all consumers"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers = en.ProcessRegistry()\n",
+ "registry_on_consumers.build(\n",
+ " \"'python3 /tmp/rabbitmq/consumer.py' | grep -v bin\",\n",
+ " roles[\"consumer\"],\n",
+ ")\n",
+ "print(registry_on_consumers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers.kill_async_after(\n",
+ " signum = signal.SIGKILL,\n",
+ " delta = timedelta(minutes=1),\n",
+ ")\n",
+ "# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
+ "results = get_stats(20)\n",
+ "results"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Killing all producers"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_producers = en.ProcessRegistry()\n",
+ "registry_on_producers.build(\n",
+ " \"'python3 /tmp/rabbitmq/producer.py' | grep -v bin\",\n",
+ " roles[\"producer\"],\n",
+ ")\n",
+ "print(registry_on_producers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_producers.kill_async_after(\n",
+ " signum = signal.SIGKILL,\n",
+ " delta = timedelta(minutes=1),\n",
+ ")\n",
+ "# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
+ "results = get_stats(20)\n",
+ "results"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Third example : Asynchronous case specifying a date"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "Here, each kill is scheduled, we have to specify the exact date at which the kill(s) happen.\n",
+ "\n",
+ "\n",
+ "The date is a ```datetime.datetime()``` object."
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Killing all consumers"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers = en.ProcessRegistry()\n",
+ "registry_on_consumers.build(\n",
+ " \"'python3 /tmp/rabbitmq/consumer.py' | grep -v bin\",\n",
+ " roles[\"consumer\"],\n",
+ ")\n",
+ "print(registry_on_consumers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers.kill_async_at(\n",
+ " signum = signal.SIGKILL,\n",
+ " date = datetime.now() + timedelta(minutes=1),\n",
+ ")\n",
+ "# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
+ "results = get_stats(20)\n",
+ "results"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Killing all producers"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_producers = en.ProcessRegistry()\n",
+ "registry_on_producers.build(\n",
+ " \"'python3 /tmp/rabbitmq/producer.py' | grep -v bin\",\n",
+ " roles[\"producer\"],\n",
+ ")\n",
+ "print(registry_on_producers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_producers.kill_async_at(\n",
+ " signum = signal.SIGKILL,\n",
+ " date = datetime.now() + timedelta(minutes=1),\n",
+ ")\n",
+ "# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
+ "results = get_stats(20)\n",
+ "results"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Fourth example : Incremental case"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "Here, each kill is scheduled, we have to specify the number, the beginning and the interval between each one of them.\n",
+ "\n",
+ "The beginning is a ```datetime.datetime``` object.\n",
+ "\n",
+ "The interval is a ```datetime.timedelta``` object."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers = en.ProcessRegistry()\n",
+ "registry_on_consumers.build(\n",
+ " \"'python3 /tmp/rabbitmq/consumer.py' | grep -v bin\",\n",
+ " roles[\"consumer\"],\n",
+ ")\n",
+ "print(registry_on_consumers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "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",
+ ")\n",
+ "# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
+ "results = get_stats(50)\n",
+ "results"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "We can have an updated version of the registry, with both dead and alive processes."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "after_refresh = registry_on_consumers.refresh()\n",
+ "print(after_refresh)"
+ ]
+ },
+ {
+ "attachments": {},
+ "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."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers.reset()\n",
+ "print(registry_on_consumers)"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Fifth example : Restart a registry"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "We can restart an entire registry, this means killing all of them (if alive) and starting them again.\n",
+ "\n",
+ "It can be done either synchronously with '''registry.restart()''', either asynchronously in the same way as the kills previously shown."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers = en.ProcessRegistry()\n",
+ "registry_on_consumers.build(\n",
+ " \"'python3 /tmp/rabbitmq/consumer.py' | grep -v bin\",\n",
+ " roles[\"consumer\"],\n",
+ ")\n",
+ "print(registry_on_consumers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_consumers.restart_async_after(\n",
+ " signum = signal.SIGKILL,\n",
+ " delta = timedelta(minutes=1),\n",
+ " interval = timedelta(minutes=1),\n",
+ ")\n",
+ "# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
+ "results = get_stats(40)\n",
+ "results"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "reset()"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_producers = en.ProcessRegistry()\n",
+ "registry_on_producers.build(\n",
+ " \"'python3 /tmp/rabbitmq/producer.py' | grep -v bin\",\n",
+ " roles[\"producer\"],\n",
+ ")\n",
+ "print(registry_on_producers)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "registry_on_producers.restart_async_at(\n",
+ " signum = signal.SIGKILL,\n",
+ " date = datetime.now() + timedelta(minutes=1),\n",
+ " interval = timedelta(minutes=1),\n",
+ ")\n",
+ "# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
+ "results = get_stats(40)\n",
+ "results"
+ ]
+ },
+ {
+ "attachments": {},
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Cleaning"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "clean()\n",
+ "provider.destroy()"
+ ]
+ }
+ ],
+ "metadata": {
+ "kernelspec": {
+ "display_name": "enoslib",
+ "language": "python",
+ "name": "python3"
+ },
+ "language_info": {
+ "codemirror_mode": {
+ "name": "ipython",
+ "version": 3
+ },
+ "file_extension": ".py",
+ "mimetype": "text/x-python",
+ "name": "python",
+ "nbconvert_exporter": "python",
+ "pygments_lexer": "ipython3",
+ "version": "3.10.4"
+ },
+ "orig_nbformat": 4
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}
diff --git a/g5k/consumer.py b/g5k/consumer.py
new file mode 100644
index 0000000000000000000000000000000000000000..2b3ce3669ac0b645a6b8da8465b0bd65f5ff3deb
--- /dev/null
+++ b/g5k/consumer.py
@@ -0,0 +1,53 @@
+#!/usr/bin/env python
+import sys
+import time
+
+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(";")
+
+ with open(f"/tmp/rabbitmq/consumer_{idx}_output.txt", "a") as f: # append mode
+ f.write(
+ " [x] Consumer "
+ + str(idx)
+ + " has received "
+ + str(number)
+ + " from "
+ + str(id_sender)
+ + "\n"
+ )
+
+ time.sleep(1.0)
+
+ # Manually acknowledge the message
+ ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+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)
+
+ # getting a connection to the broker
+ credentials = pika.PlainCredentials(username=username, password=password)
+ parameters = pika.ConnectionParameters(host, 5672, credentials=credentials)
+ connection = pika.BlockingConnection(parameters)
+ channel = connection.channel()
+
+ # declaring the queue again (to be sure)
+ channel.queue_declare(queue="fault_injection")
+
+ # auto_ack: as soon as collected, a message is considered as acked
+ channel.basic_consume(queue="fault_injection", auto_ack=False, on_message_callback=callback)
+
+ # wait for messages
+ channel.start_consuming()
diff --git a/g5k/producer.py b/g5k/producer.py
new file mode 100644
index 0000000000000000000000000000000000000000..d44ca60cd07bc792999960355e7117313b82e8d7
--- /dev/null
+++ b/g5k/producer.py
@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+import random
+import sys
+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)
+
+ # getting a connection to the broker
+ credentials = pika.PlainCredentials(username=username, password=password)
+ parameters = pika.ConnectionParameters(host, 5672, credentials=credentials)
+ connection = pika.BlockingConnection(parameters)
+ channel = connection.channel()
+
+ # declaring the queue
+ channel.queue_declare(queue="fault_injection")
+ # send the message, through the exchange ''
+ # which simply delivers to the queue having the key as name
+ while 1:
+ number = random.randint(0, 100)
+ channel.basic_publish(
+ exchange="", routing_key="fault_injection", body=str(idx) + ";" + str(number)
+ )
+
+ with open(f"/tmp/rabbitmq/producer_{idx}_output.txt", "a") as f:
+ f.write(" [x] Producer " + str(idx) + " has sent " + str(number) + "\n")
+
+ time.sleep(1.0)
+
+ # gently close (flush)
+ connection.close()