This is a toy program that periodically makes HTTP requests and records the
latency. It's mostly an experiment to let me try writing concurrent code in
Python. At some point, if I find the time, I'll try to compare asyncio
,
threading
, multiprocessing
. For the time being, it uses multiprocessing
.
The real work -- making the requests and writing the results to Kafka -- is done with third-party libraries:
These are both Python wrappers around C libraries. If you choose your OS and Python release carefully, there are binary Wheels available with the libraries built-in.
latency-logger [-h] [-v] [-Q N] [-W N]
[--ca-cert FILE] [--auth-cert FILE] [--auth-key FILE]
--bootstrap-servers URLs --schema-registry URL
--topic TOPIC
FILE
Monitor URL request latency.
positional arguments:
FILE comma-separated file of delays (in seconds) and URLs
optional arguments:
-h, --help show this help message and exit
-v, --verbose produce more verbose output
--queue N maximum queue length
--workers N number of workers
--bootstrap-servers URLs Kafka bootstrap servers
--schema-registry URL schema registry URL
--ca-cert FILE TLS CA certificate file for Kafka
--auth-cert FILE TLS client certificate file
--auth-key FILE TLS client key file for
--topic TOPIC topic name
$ [ -f venv/bin/activate ] || python3 -m venv venv
$ . venv/bin/activate
(venv) $ export PYCURL_SSL_LIBRARY=openssl
(venv) $ pip install -e .[dev]
(venv) $ flake8 *.py src/ test/
(venv) $ pytest -vv
(venv) $ latency-logger --help
You can create the infrastructure services to run the application using the
Terraform configuration in terraform/
. You will need to assign values to the
variables in a terraform.tfvars
file or in appropriately named environment
variables.
Once the services are running, you can start the application using the run.sh
shell script. The script extracts parameters for the application from the
Terraform output values.
(venv) $ cd terraform/
(venv) $ terraform apply
(venv) $ cd -
(venv) $ ./run.sh demo.csv
The system is composed of three main components:
-
The main process spawns the workers, receives and handles signals, etc.
-
The scheduler process maintains the schedule and enqueues tasks for processing.
-
The worker processes dequeue tasks and process them, using PycURL to perform the requests and Confluent's Kafka client library to publish the results.
There are two ways processes communicate:
-
A shared variable is used by the main process to instruct children to shutdown. This is set in signal handlers for SIGINT and SIGTERM.
-
A shared queue of tasks is written by the scheduler and read by the workers.
The main process handles the command-line arguments, reads the configuration file, creates the IPC channels, and launches the child processes.
As part of its initialisation, the main process register a signal handler for
SIGINT
, SIGTERM
, and SIGCHLD
. When any of these signals are received,
shared variable is set and the children should terminate gracefully.
Once the system has started running the main process waits for the children to terminate.
A simple scheduler is used to maintain a queue of jobs ordered by their next execution time. The code is extremely naive -- a sorted array, resorted when changes are made -- and could easily be replaced with a heap to improve the asymptotics for large collections of sites to monitor.
The worker processes start a Kafka producer (configure to serialise messages to Avro) and then start their processing loop. In each iteration, they poll the Kafka client to process any pending events (e.g. logging Kafka errors to the worker process output), fetch a pending task from the queue, run the request, and write the results to Kafka.
This whole assembly more or less assumes deployment on a POSIX-ish system doing normal POSIX-y things. Like forking, handling signals, etc.
Importantly, the problems around fork() on MacOS makes some problems more
problematic. On platforms where it uses fork
, the main process should notice
when children die and terminate the system saftely. Unless forkserver
is
doing something clever under the covers (๐ค) and forwarding SIGCHLD
then we're
probably out of luck in that regard.