A GitLab upgrade is scheduled for Monday, June 2, 2025. The service will be unavailable for a few minutes in the morning. We'll keep you posted on the progress of the upgrade on the Mattermost channel: https://mattermost.inria.fr/devel/channels/gitlab. We recommend that you do not work on the platform until an announcement indicates that maintenance is complete.
"Each node must have this minimal configuration :\n",
"- having python and pip\n",
"- having procps (to use pgrep)\n",
"- having procps (to use kill)\n",
"- having pika (for the rabbitmq connection)"
]
},
...
...
@@ -278,7 +278,7 @@
"metadata": {},
"source": [
"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 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) and producer(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) by specifying the number of each\n",
"- to reset the experiment by going back to its initial state (clean + launch) "
...
...
@@ -291,11 +291,12 @@
"outputs": [],
"source": [
"## Get server's IP address on the private network\n",
"Here, each kill is scheduled using the ```time_span``` parameter, which can be expressed either as a delay using a ```datetime.timedelta()``` object, or as a date using a ```datetime.datetime()``` object."
"Here, each kill is scheduled using the ```start_in``` or ```start_at``` parameters, which can be expressed either as a delay using a ```datetime.timedelta()``` object, or as a date using a ```datetime.datetime()``` object."
"# each iteration last for ~4 sec (2 requests + 1sec sleep)\n",
"# each iteration last for ~2 sec (2 requests + 1sec sleep)\n",
"results = get_queues_info_for(45)\n",
"results"
]
...
...
@@ -834,7 +845,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(nconsumer = 3, nproducer = 3)``` has nothing to do with the ```reset(nconsumer = 3, nproducer = 3)``` 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()``` has nothing to do with the ```reset(nconsumer, nproducer)``` specificly implemented for this experiment ! "
]
},
{
...
...
@@ -847,6 +858,16 @@
"registry_on_consumers"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"results = get_queues_info_for(2)\n",
"results"
]
},
{
"attachments": {},
"cell_type": "markdown",
...
...
@@ -862,7 +883,7 @@
"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. If asynchronous, the date of the kill(s) (or the time delta before they happen) and its interval with the start(s) of the docker container(s) can be specified."
"It can be done either synchronously with ```registry.restart()```, either asynchronously in the same way as the kills previously shown. If asynchronous, the date of the kill(s) (or the delay before they happen) and its interval with the start(s) of the process(es) can be specified."
Make sure you've run the one time setup for your environment
</div>
%% Cell type:markdown id: tags:
## Rabbitmq / Cron
[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.
For our experiment, we will deploy a publish / suscribe environment to demonstrate the impact of our api.
[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.
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).
Here, we does not launch anything yet, we just setup the server node to accept all our producer(s) and consumer(s). We also add a new administrator in order to have access to the management interface, the default one being blocked by the remote configuration.
# 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")
```
%% Cell type:markdown id: tags:
#### Producers' node configuration
%% Cell type:markdown id: tags:
The producers' node has to be configured such that it contains its specific script.
%% Cell type:code id: tags:
``` python
withen.actions(roles=roles["producer"])asp:
p.copy(
src=HERE+"/producer.py",
dest="/tmp/rabbitmq/producer.py",
task_name="copying producer file",
)
```
%% Cell type:markdown id: tags:
#### Consumers' node configuration
%% Cell type:markdown id: tags:
The consumers' node has to be configured such that it contains its specific script.
%% Cell type:code id: tags:
``` python
withen.actions(roles=roles["consumer"])asp:
p.copy(
src=HERE+"/consumer.py",
dest="/tmp/rabbitmq/consumer.py",
task_name="copying consumer file",
)
```
%% Cell type:markdown id: tags:
#### Utility functions
%% Cell type:markdown id: tags:
The only purpose of these functions is to facilitate and to make this experiment more readable. Their objectives are :
- 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))
- 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) and producer(s))
- to clear the experiment (kill all instances of producer(s)/consumer(s) if any, delete all output files if any, purges the rabbitmq queues)
- to launch all producer(s) and consumer(s) by specifying the number of each
- to reset the experiment by going back to its initial state (clean + launch)
Here, each kill is scheduled using the ```time_span``` parameter, which can be expressed either as a delay using a ```datetime.timedelta()``` object, or as a date using a ```datetime.datetime()``` object.
Here, each kill is scheduled using the ```start_in``` or ```start_at``` parameters, which can be expressed either as a delay using a ```datetime.timedelta()``` object, or as a date using a ```datetime.datetime()``` object.
%% Cell type:markdown id: tags:
### Killing all consumers
%% Cell type:code id: tags:
``` python
reset(nconsumer=3,nproducer=3)
```
%% Cell type:code id: tags:
``` python
registry_on_consumers=en.ProcessRegistry()
registry_on_consumers.build(
"tuto_",
roles["consumer"],
)
registry_on_consumers
```
%% Cell type:code id: tags:
``` python
registry_on_consumers.kill_async(
signum=signal.SIGKILL,
time_span=timedelta(minutes=1,seconds=20),
start_in=timedelta(minutes=1,seconds=20),
)
# each iteration last for ~4 sec (2 requests + 1sec sleep)
# each iteration last for ~2 sec (2 requests + 1sec sleep)
# each iteration last for ~4 sec (2 requests + 1sec sleep)
# each iteration last for ~2 sec (2 requests + 1sec sleep)
results=get_queues_info_for(40)
results
```
%% Cell type:markdown id: tags:
## Third example : Incremental case
%% Cell type:markdown id: tags:
Here, each kill is scheduled and the choice of the killed process(es) is done totally randomly among those registered, we have to specify how many process(es) we want to kill, the beginning (date of the first kill) and the interval between each of them.
The beginning can be expressed either as a delay using a ```datetime.timedelta()``` object, or as a date using a ```datetime.datetime()``` object.
The interval is a ```datetime.timedelta``` object.
# each iteration last for ~4 sec (2 requests + 1sec sleep)
# each iteration last for ~2 sec (2 requests + 1sec sleep)
results=get_queues_info_for(45)
results
```
%% Cell type:markdown id: tags:
We can have an updated version of the registry, with both dead and alive processes.
%% Cell type:code id: tags:
``` python
after_refresh=registry_on_consumers.refresh()
after_refresh
```
%% Cell type:markdown id: tags:
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 !
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(nconsumer, nproducer)``` specificly implemented for this experiment !
%% Cell type:code id: tags:
``` python
registry_on_consumers.reset()
registry_on_consumers
```
%% Cell type:code id: tags:
``` python
results=get_queues_info_for(2)
results
```
%% Cell type:markdown id: tags:
## Fourth example : Restart a registry
%% Cell type:markdown id: tags:
We can restart an entire registry, this means killing all of them (if alive) and starting them again.
It can be done either synchronously with ```registry.restart()```, either asynchronously in the same way as the kills previously shown. If asynchronous, the date of the kill(s) (or the time delta before they happen) and its interval with the start(s) of the docker container(s) can be specified.
It can be done either synchronously with ```registry.restart()```, either asynchronously in the same way as the kills previously shown. If asynchronous, the date of the kill(s) (or the delay before they happen) and its interval with the start(s) of the process(es) can be specified.
%% Cell type:markdown id: tags:
### Restarting all consumers
%% Cell type:code id: tags:
``` python
reset(nconsumer=3,nproducer=3)
```
%% Cell type:code id: tags:
``` python
registry_on_consumers=en.ProcessRegistry()
registry_on_consumers.build(
"tuto_",
roles["consumer"],
)
registry_on_consumers
```
%% Cell type:code id: tags:
``` python
registry_on_consumers.restart_async(
signum=signal.SIGKILL,
time_span=timedelta(minutes=1,seconds=10),
start_in=timedelta(minutes=1,seconds=10),
interval=timedelta(seconds=20),
)
# each iteration last for ~4 sec (2 requests + 1sec sleep)
# each iteration last for ~2 sec (2 requests + 1sec sleep)
# each iteration last for ~4 sec (2 requests + 1sec sleep)
# each iteration last for ~2 sec (2 requests + 1sec sleep)
results=get_queues_info_for(45)
results
```
%% Cell type:markdown id: tags:
## Fifth example : Planning
%% Cell type:markdown id: tags:
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()```
After 6 minutes, we will launch 20 producers (5 each 10 seconds).
After 8 minutes, we will launch 100 consumers (20 each 10 seconds).
After 10 minutes, we will kill all the consumers consumers (20 each 10 seconds).
%% Cell type:code id: tags:
``` python
now=datetime.now()
planning=en.Planning()
cmd=launch_n_actors_cmd(actor="producer",n=5)
forminrange(5):
planning.add_event(
en.StartEvent(
date=now+timedelta(minutes=6,seconds=10*m),
host=roles["producer"][0],
cmd=cmd,
ns=f"tuto_producer_{m}",
)
)
cmd=launch_n_actors_cmd(actor="consumer",n=5)
forminrange(5):
(
planning.add_event(
en.StartEvent(
date=now+timedelta(minutes=8,seconds=10*m),
host=roles["consumer"][0],
cmd=cmd,
ns=f"tuto_consumer_1_{m}",
)
).add_event(
en.StartEvent(
date=now+timedelta(minutes=8,seconds=10*m),
host=roles["consumer"][0],
cmd=cmd,
ns=f"tuto_consumer_2_{m}",
)
).add_event(
en.StartEvent(
date=now+timedelta(minutes=8,seconds=10*m),
host=roles["consumer"][0],
cmd=cmd,
ns=f"tuto_consumer_3_{m}",
)
).add_event(
en.StartEvent(
date=now+timedelta(minutes=8,seconds=10*m),
host=roles["consumer"][0],
cmd=cmd,
ns=f"tuto_consumer_4_{m}",
)
)
)
```
%% Cell type:code id: tags:
``` python
forminrange(5):
(
planning.add_event(
en.KillEvent(
date=now+timedelta(minutes=10,seconds=10*m),
host=roles["consumer"][0],
ns=f"tuto_consumer_1_{m}",
)
).add_event(
en.KillEvent(
date=now+timedelta(minutes=10,seconds=10*m),
host=roles["consumer"][0],
ns=f"tuto_consumer_2_{m}",
)
).add_event(
en.KillEvent(
date=now+timedelta(minutes=10,seconds=10*m),
host=roles["consumer"][0],
ns=f"tuto_consumer_3_{m}",
)
).add_event(
en.KillEvent(
date=now+timedelta(minutes=10,seconds=10*m),
host=roles["consumer"][0],
ns=f"tuto_consumer_4_{m}",
)
)
)
```
%% Cell type:code id: tags:
``` python
planning_service=en.PlanningService(planning)
planning_service.deploy()
```
%% Cell type:markdown id: tags:
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.
"Here, we does not launch anything yet, we just setup the server node to accept all our producer(s) and consumer(s). We also add a new administrator in order to have access to the management interface, the default one being blocked by the remote configuration."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"username_monitoring = \"user\"\n",
"password_monitoring = \"password\"\n",
"\n",
"# SETUP\n",
"## Server configuration\n",
"with en.actions(roles=roles[\"server\"]) as p:\n",
"The producers' node has to be configured such that it contains its specific script."
]
},
{
"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"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"The consumers' node has to be configured such that it contains its specific script."
]
},
{
"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"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"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) and producer(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) by specifying the number of each\n",
"- to reset the experiment by going back to its initial state (clean + launch) "
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"## Get server's IP address on the private network\n",
"An ```Event``` define a command that will run at a specific date (```datetime.datetime```) on a specific host (```Host```)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"#### StartEvent\n",
"\n",
"A ```StartEvent```, as its name suggests, will start a named process.\n",
"\n",
"In addtion of the base class ```Event``` attributes, the ```command``` to launch the process and its ```name``` must be specified and are of type ```str```.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Not supposed to return anything ! The only purpose of this code is to show how a StartEvent is created.\n",
"event_date = datetime.now()\n",
"event_host = en.Host(\"address\")\n",
"event_cmd = \"sleep infinity\"\n",
"process_name = \"name\"\n",
"\n",
"start_event = en.StartEvent(\n",
" date = event_date,\n",
" host = event_host,\n",
" cmd = event_cmd,\n",
" ns = process_name, \n",
")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"#### KillEvent\n",
"\n",
"A ```KillEvent```, as its name suggests, will kill a named process.\n",
"\n",
"In addtion of the base class ```Event``` attributes, only ```name``` of the process to kill must be specified, which is of type ```str```."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Not supposed to return anything ! The only purpose of this code is to show how a KillEvent is created.\n",
"event_date = datetime.now()\n",
"event_host = en.Host(\"address\")\n",
"process_name = \"name\"\n",
"\n",
"kill_event = en.KillEvent(\n",
" date = event_date,\n",
" host = event_host,\n",
" ns = process_name, \n",
")"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### Planning\n",
"\n",
"A ```Planning``` can be considered as an Event storage.\n",
"\n",
"A ```PlanningService``` records all processes specified in a ```Planning```. It can be seen as an Event registry.\n",
"\n",
"To define a ```PlanningService```, specifying a ```Planning``` is necessary."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"planning = en.Planning()\n",
"planning.add_event(start_event)\n",
"planning.add_event(kill_event)\n",
"\n",
"planning_service = en.PlanningService(planning)"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Examples"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"### A first deployment\n",
"\n",
"We are here gonna schedule :\n",
"- the start of 5 rabbitmq producers every 10 seconds after 2 minute 20 seconds\n",
"- the start of 5 rabbitmq consumers every 10 seconds, 1 minute after the last producer is launched\n",
"- the kill of 3 rabbitmq producers 4 minutes after the start of the experiment\n",
"- the kill of all the rabbitmq consumers 30 seconds after the producers are killed"
Here, we does not launch anything yet, we just setup the server node to accept all our producer(s) and consumer(s). We also add a new administrator in order to have access to the management interface, the default one being blocked by the remote configuration.
# 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")
```
%% Cell type:markdown id: tags:
#### Producers' node configuration
%% Cell type:markdown id: tags:
The producers' node has to be configured such that it contains its specific script.
%% Cell type:code id: tags:
``` python
withen.actions(roles=roles["producer"])asp:
p.copy(
src=HERE+"/producer.py",
dest="/tmp/rabbitmq/producer.py",
task_name="copying producer file",
)
```
%% Cell type:markdown id: tags:
#### Consumers' node configuration
%% Cell type:markdown id: tags:
The consumers' node has to be configured such that it contains its specific script.
%% Cell type:code id: tags:
``` python
withen.actions(roles=roles["consumer"])asp:
p.copy(
src=HERE+"/consumer.py",
dest="/tmp/rabbitmq/consumer.py",
task_name="copying consumer file",
)
```
%% Cell type:markdown id: tags:
#### Utility functions
%% Cell type:markdown id: tags:
The only purpose of these functions is to facilitate and to make this experiment more readable. Their objectives are :
- 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) and producer(s))
- to clear the experiment (kill all instances of producer(s)/consumer(s) if any, delete all output files if any, purges the rabbitmq queues)
- to launch all producer(s) and consumer(s) by specifying the number of each
- to reset the experiment by going back to its initial state (clean + launch)
An ```Event``` define a command that will run at a specific date (```datetime.datetime```) on a specific host (```Host```)
%% Cell type:markdown id: tags:
#### StartEvent
A ```StartEvent```, as its name suggests, will start a named process.
In addtion of the base class ```Event``` attributes, the ```command``` to launch the process and its ```name``` must be specified and are of type ```str```.
%% Cell type:code id: tags:
``` python
# Not supposed to return anything ! The only purpose of this code is to show how a StartEvent is created.
event_date=datetime.now()
event_host=en.Host("address")
event_cmd="sleep infinity"
process_name="name"
start_event=en.StartEvent(
date=event_date,
host=event_host,
cmd=event_cmd,
ns=process_name,
)
```
%% Cell type:markdown id: tags:
#### KillEvent
A ```KillEvent```, as its name suggests, will kill a named process.
In addtion of the base class ```Event``` attributes, only ```name``` of the process to kill must be specified, which is of type ```str```.
%% Cell type:code id: tags:
``` python
# Not supposed to return anything ! The only purpose of this code is to show how a KillEvent is created.
event_date=datetime.now()
event_host=en.Host("address")
process_name="name"
kill_event=en.KillEvent(
date=event_date,
host=event_host,
ns=process_name,
)
```
%% Cell type:markdown id: tags:
### Planning
A ```Planning``` can be considered as an Event storage.
A ```PlanningService``` records all processes specified in a ```Planning```. It can be seen as an Event registry.
To define a ```PlanningService```, specifying a ```Planning``` is necessary.
%% Cell type:code id: tags:
``` python
planning=en.Planning()
planning.add_event(start_event)
planning.add_event(kill_event)
planning_service=en.PlanningService(planning)
```
%% Cell type:markdown id: tags:
## Examples
%% Cell type:markdown id: tags:
### A first deployment
We are here gonna schedule :
- the start of 5 rabbitmq producers every 10 seconds after 2 minute 20 seconds
- the start of 5 rabbitmq consumers every 10 seconds, 1 minute after the last producer is launched
- the kill of 3 rabbitmq producers 4 minutes after the start of the experiment
- the kill of all the rabbitmq consumers 30 seconds after the producers are killed
