Distributed Computation Framework

Table of Contents

1. Project Description and Provided Features
2. Framework Architecture
3. Getting Started
4. Usage
5. Limitations & Outlook
6. Licensing
7. Contacts

1) Project Description and Provided Features

The goal of this project is to provide a distributed computation framework (DCF) based on Kafka and Docker, which is designed to manage the lifecycle of Docker containers that generate sample data and deliver it via distinct Kafka topics.

Using a graphical user interface (GUI), users can easily submit jobs by providing a small set of parameters, such as the Docker container image name and the Kafka topic to which the generated data will be delivered. This simple interface is designed to be user-friendly, minimising the overhead for users to get started with data generation tasks. The complexity is abstracted and the user does not have to worry about the underlying infrastructure.

Once a job is submitted, the framework orchestrates the necessary processes to ensure that the user's requirements are met efficiently. The framework is designed to scale, allowing it to handle multiple concurrent jobs, each with unique parameters.

2) Framework Architecture

2.1) Architecture & Components

The proposed framework consists of six main components, which are described briefly subsequently:

1. Kafka: Distributed event streaming platform used to a) coordinate actions performed by the Job Handler and Container Handler (Agent), to b) provide agent monitoring mechanisms, and c) provide an event streaming endpoint for the Data Generator(s). While the topics for a) and b) are predefined and use schema validation, the topic(s) for c) can be defined individually for each submitted job and schema validation is not performed.
   The DCF uses Apache Kafka Raft (KRaft) as the consensus protocol, allowing the removal of Apache ZooKeper and its dependencies. The result is a simplified architecture, as the responsibility for metadata management can be consolidated into Kafka itself.
   To allow the framework to orchestrate the necessary processes, the three management topics 1) Job_Handling (creation & deletion of jobs), 2) Job_Status (status information of the jobs submitted) and 3) Agent_Status (status information of the Container Handler) are used. With a replication factor of 1 and only 1 partition, they all share the same configuration. More about the limitation of this configuration can be found in Section 5 (limitation 02).
2. Kafka Schema Registry: Centralized repository for the management and validation of schemes for message data that supports serialization and deserialization. The schema registry takes on a key role in ensuring data consistency and compatibility, especially as schemes evolve.
   The DCF uses Apache Avro as the serialization format. The three different Avro schemes 1) Job Handling, 2) Job Status and 3) Agent Status were defined and are in line with the topics described above.
3. Docker Daemon: Service responsible for orchestrating container lifecycle management. It handles tasks such as container creation, execution, deletion and monitoring required to fulfil the jobs submitted.
4. Job Handler: Service, which is responsible for managing and orchestrating jobs. The system consists of a web server, Celery workers, Kafka, and a persistence layer to ensure efficient orchestration. The Job Handler provides a user-friendly GUI for creating, monitoring, and managing jobs, providing real-time updates using Django Channels and web sockets. As the Job Handler plays a crucial role within the framework, it will be described in more detail in Section 2.2.
5. Container Handler (Agent): Agent, which is responsible for starting/stopping containers and providing monitoring capabilities for processing jobs submitted via the GUI. As the Container Handler plays a crucial role within the framework, it will be described in more detail in Section 2.3.
6. Data Generator(s): Docker containers that generate and expose sample data to a chosen/defined Kafka topic. As the logic for generating data is out of scope, the focus is on configuring the data generator containers to fit into the DCF. To do so, Section 3.5 can be consulted.

The graphical representation of the DCF architecture and the interactions between the six different components can be found in the Figure 01 below: Figure 01: Architecture Overview

2.2) Job Handler

The Job Handler is responsible for the management of the lifecycle of jobs, including their creation, monitoring, and termination at a high level. To ensure an efficient orchestration, the application interacts with various other components, such as the web server, Celery workers, Kafka, and the database. The Job Handler provides the following three main features:

1. User-friendly GUI: This interface provides a comprehensive management solution, which we refer to as job orchestration, for the lifecycle of jobs, encompassing their creation, monitoring and termination at a high level. Users can effortlessly create new jobs, monitor the status of running jobs/available agents in real-time by using Django Channels/web sockets, and manage job lifecycles within the GUI.
2. Job scheduling: The Job Handler facilitates efficient job scheduling, enabling users to schedule jobs to run for designated durations and at predetermined times. Users can track the elapsed time for jobs in seconds and monitor job progress by observing the job status.
3. Scalability: The Job Handler is designed to be scalable, allowing it to handle multiple concurrent jobs and multiple users. It can distribute tasks across multiple Celery workers and manage job orchestration efficiently.

In order to achieve the features mentioned, ensuring reliability and maintainability, the architecture comprises the following key components:

• Celery: Responsible for the management of tasks and concurrency. This includes asynchronous tasks such as starting and stopping jobs, periodic tasks like monitoring job run lifecycles, and permanent tasks like observing agents and jobs.
• Django: Serves as the web framework, handling HTTP requests, rendering templates, and managing the database.
• Django Channels: Enable real-time communication using web sockets, providing live updates to the user interface.
• Redis: Used as the backend for Django Channels and Celery, providing fast in-memory storage.
• SQlite: The primary database for storing job, agent and container information (including further data models). More about the limitation of this setup can be found in Section 5 (limitation 05).

The graphical representation of the Job Handler architecture and the interactions necessary can be found in the Figure 02 below: Figure 02: Architecture Overview Job Handler

• Django Web Server: Offers a user-friendly experience for managing and deploying jobs. It handles HTTP requests, renders templates and manages the database. In addition to orchestrating job forwarding to Celery Workers, the web server is responsible for CRUD operations and first-level validation. Furthermore, the web server serves as the central configuration unit for Channels and Celery.
• Django Channel Consumer: Responsible for implementing a consumer group and web socket for the frontend. The backend components, web server and Celery Workers communicate with the Channel Consumer to update the user in the frontend. The backend components reach the clients via the consumer group.
• Celery Workers: Host independent processes (tasks) handling the job orchestration. All tasks are connected to Kafka to communicate with the Container Handler by consuming and producing messages. The tasks are connected to specific Kafka topics. All tasks are running in parallel and are distributed over multiple Celery Workers. The Celery Workers are running in a distributed environment and are connected to the Redis message broker. All tasks are handling business logic independently and are manipulating the main database (SQlite). The six celery workers available are introduced subsequently:
  • Agent Monitor: Plays a vital role in the system's operations. It is responsible for monitoring agents that have registered with the Job Handler via Kafka. The Agent Monitor consumes messages from the Kafka topic Agent_Status and is responsible for managing agents.
  • Job Monitor: Tasked with monitoring and managing the running jobs on Container Handler(s). It consumes messages from the Kafka topic Job_Status.
  • Job Starter: Asynchronous task that is triggered by the web server when a user requests the initiation of a job. If the conditions for starting the job are met, the task requests that the Container Handler start the relevant containers. The Job Starter produces messages to the Kafka topic Job_Instruction.
  • Job Start Scheduler: Responsible for monitoring the job-start conditions (DateTime in UCT format). If the conditions are fulfilled, the Container Handler is requested to initiate the jobs. The Job Start Scheduler produces messages to the Kafka topic Job_Instruction.
  • Job Stopper: Asynchronous task triggered by the web server when a user requests that a job be stopped. If the conditions for stopping the job are fulfilled, the task requests that the Container Handler stop and delete the corresponding containers. The Job Stopper produces messages to the Kafka topic Job_Instruction.
  • Job Stop Scheduler: Responsible to monitor running jobs. In the event that the jobs have reached their maximum computation time, the Container Handler will be requested to stop the jobs. The Job Stop Scheduler produces messages to the Kafka topic Job_Instruction.

The job handling, besides basic CRUD (no update), is handled fully asynchronously and concurrently with Celery.The business logic (tasks) is distributed over multiple Celery Workers, which operate independently and concurrently. Empirical testing showed that operating all tasks on a single Celery Worker resulted in communication and performance bottlenecks. Therefore, the tasks are distributed over multiple Celery Workers and Redis message queues. The Beat Worker is responsible for Celery Beat tasks, while the Manual Job Handling Worker executes asynchronous tasks manually triggered by users. As the Celery Worker workload for manual tasks is not permanent, these tasks are combined in one Celery Worker. The Agent Monitor and Job Monitor operate in separate Celery Workers since they run permanently.

2.3) Container Handler (Agent)

In order for the Container Handler to be able to start/stop containers and provide monitoring capabilities, several components are required - of particular interest is the Container Handler (Agent) process, which represents a multithreded environment, consisting of the following two threads:

• Agent Monitor: Responsible for monitoring the health of the Docker Daemon and reporting on the containers under the responsibility of the specific agent. The Agent Monitor provides the monitoring result to the Job Handler via the Kafka topic Agent_Status at a predefined interval. If access to the Docker Daemon fails, not only is an unsuccessful monitoring result sent, but the container handler (thread) is also blocked, meaning that no further messages are consumed.
• Container Controller: Responsible for starting/stopping containers by consuming messages from the Job_Instruction Kafka topic and communicating with the Docker Daemon. After each and every job is processed, a status message targeting the specific processed job is created in the Job_Status Kafka topic. Sophisticated exception handling ensures that the Container Controller runs as reliably as possible.

The graphical representation of the Container Handler (Agent) architecture and the interactions necessary can be found in the Figure 03 below: Figure 03: Architecture Overview Container Handler (Agent)

2.4) Technologies used

To implement the DCF, the following technology stack is used:

3) Getting Started

All commands outlined in this chapter must be run from the root directory of this project, otherwise adapt them accordingly.

3.1) Prerequisites

Ensure that the following requirements are met:

• Installed Python v3.13.
• Up and running Docker Daemon on all host machines on which the Container Handler (Agent) is to run.
• (Optional, but highly recommended) A Python virtual environment (venv). To do so, proceed as follows:
  1. Create the Python virtual environment named venv:

     python3.13 -m venv venv

  2. Activate the virtual environment in every terminal used and make sure it has been activated correctly:

     source venv/bin/activate
     which python3.13 && which pip3.13
     # Ensure, the following ouput is returned to check for correct activation:
     # ./venv/bin/python3.13
     # ./venv/bin/pip3.13

• Required Python dependencies:
  1. Install the required dependencies:

     python3.13 -m pip install -r requirements.txt

• It should be noted that the available .env File provides fundamental configuration options. Despite the possibility that this may not align with best practice, it was deemed an appropriate course of action in order to facilitate more efficient handling of the DCF.

3.2) Kafka

With the provided Docker Compose File, the installation of Kafka is straightforward. As additional configuration is performed automatically during startup (see initialize.py and initializer container within Docker Compose File), only the subsequent steps must be completed:

1. Open a new terminal or reuse the one from Section 3.1 (make sure the venv is activated properly, use source venv/bin/activate otherwise).

2. Start Kafka, the GUI, the Schema registry and the initialization by running the following command:

   Important: As we use Kafka in KRaft mode (possible from v2.8+), it is essential to ensure that no Kafka image with an older version is present on your host machine. Therefore, the initial step (which is highly recommended) is to delete the old image with docker rmi apache/kafka-native --force.

    docker compose -f kafka/docker-compose.yaml up -d
    # be aware: Kafka might take a few seconds to be deployed
   
    # if it should be necessary to stop the cluster, use docker compose -f kafka/docker-compose.yaml down

   (Optional, in case of issues with Kafka) Should any issues arise in spite of the aforementioned procedure, a different configuration of Kafka can be utilised by calling docker compose -f kafka/docker-compose-alternative.yaml up -d.

3. (Optional, but highly recommended) After the startup of Kafka, you can access:
   3.1 Kafka Grapical User Interface: http://localhost:8080
   3.2 Kafka Schema registry: http://localhost:8081

3.3) Container Handler (Agent)

To start up an instance of the Container Handler (Agent), proceed as follows:

1. Open a new terminal or reuse the one from Section 3.2 (make sure the venv is activated properly, use source venv/bin/activate otherwise).
2. Start the Container Handler (Agent) by performing the subsequent command :

   python3.13 -m orchestrator.container_handler.container_handler

3. (Optional, but highly recommended) Refer to the log file to ensure that the system has booted up correctly.

3.4) Job Handler

Each command should run in its own terminal window or session to ensure all processes operate concurrently. For efficiency during development, tools like tmux or screen to manage multiple terminal sessions in one window can be used.

To set up and run the Job Handler follow the steps subsequently:

1. Open a new terminal (make sure the venv is activated properly, use source venv/bin/activate otherwise).

2. Start Redis with the provided Docker Compose File by running the following command:

   docker-compose -f orchestrator/control_server/docker-compose.yaml up -d
   # be aware: Redis might take a few seconds to be deployed

   This command will spin up the following two Redis instances:

   • redis_channels: Enabling asynchronous parallelized web sockets and consumer groups with Django Channels.
   • redis_celery: Serving celery workers. The Broker and the Result Backend use two seperate Databases.

3. Apply Django migrations (create database schema) by running:

   python3.13 orchestrator/control_server/manage.py migrate

4. Start the Django server with the subsequent command:

   python3.13 orchestrator/control_server/manage.py runserver

   (Optional, but highly recommended) Check if the server is accessible under http://localhost:8000/job_handler.

   Please note: Do not perform any actions yet, as the subsequent steps 5 and 6 must be completed first for the system to be fully operational.

5. Run the following commands to start the Celery Workers. Open a new terminal for every command, four in total (make sure the venv is activated properly, use source venv/bin/activate otherwise):

   • Worker for manual job handling:

     cd orchestrator/control_server/ && celery -A control_server_project worker --queues manual_job_handling --loglevel=info

   • Worker for Celery Beat tasks:

     cd orchestrator/control_server/ && celery -A control_server_project worker --queues beat --loglevel=info

   • Worker for monitoring agent status:

     cd orchestrator/control_server/ && celery -A control_server_project worker --queues monitor_agent_status --loglevel=info

   • Worker for monitoring job status:

     cd orchestrator/control_server/ && celery -A control_server_project worker --queues monitor_job_status --loglevel=info

6. Start the Celery Beat Scheduler by running the following command in a new terminal (make sure the venv is activated properly, use source venv/bin/activate otherwise):

   cd orchestrator/control_server/ && celery -A control_server_project beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler

(Optional, but highly recommended) Refer to the log file to ensure that the system has booted up correctly.

Once all the services are running, the application can be accessed and used under http://localhost:8000/job_handler.

3.5) Preparing the Data Generator(s)

In order to integrate the Data Generator(s) into the DCF, some minor adjustments (within the relevant parts of the application source code) are required, as shown below:

1. Install the required dependency (confluent-kafka) by means of the following command:

   python3.13 -m pip install confluent-kafka

2. Use the following code snippet to import the Producer class and the os package:

   from confluent_kafka import Producer
   import os

3. Initialise the Kafka Producer with the following snippet of code:

   # load environment variables necessary for Kafka accessability
   kafka_bootstrap_servers_docker = os.getenv("KAFKA_BOOTSTRAP_SERVERS_DOCKER")
   kafka_topic = os.getenv("KAFKA_TOPIC")
   # initialize Kafka Producer if environment variables loaded successfully
   if kafka_bootstrap_servers and kafka_topic:
       kafka_producer = Producer({"bootstrap.servers": kafka_bootstrap_servers_docker})

4. At the correct position (ideally within a while true-loop with the appropriate interval), stream the data to the Kafka topic previously defined in the GUI by integrating the following bit of code:

   kafka_producer.produce(topic=kafka_topic, value=json.dumps(output_data).encode('utf-8'))

   Note: Make sure that output_data is a serialisable Python object (e.g. dict, list)

Once the integration described in the four steps above has been completed, please proceed as usual. This involves building the container and making it accessible via Docker Hub. Now the container image is ready to be used.

Please note that, in addition to confluent-kafka, other libraries can be utilised. If you do not wish to work with Python, you may use Java instead. In both cases, adjust the source code accordingly.

Last but not least, it is important to mention that, depending on the needs, the Kafka schema registry and Avro schemes can be implemented as well. Furthermore, we do not provide guidance on consuming messages created by the data generators, but there are many respected resources which explain how to do this.

For an exemplary implementation, please refer to the following Github repository.

4) Usage

Once the steps outlined in sections 3.1-3.4 have been successfully completed and a Data Generator has been prepared in accordance with section 3.5 the DCF can be utilised. Please use the following URLs to access the relevant components:

• Job Handler GUI: http://localhost:8000/job_handler
• Kafka GUI: http://localhost:8080

4.1) Create a Job(s)

To create a new job, follow the subsequent steps:

1. Navigate to the Job Handler GUI -> Create Job.
2. Fill in the form (the input fields are self-explanatory). For the container image name, christianbieri/data-simulator-temperature:1.0.1 can be used as it has been prepared in accordance with section 3.5.
3. Submit the job.

An overview of the points described above is provided by the following recording:

4.2) Start, Stop, Restart Job(s)

As soon as the job has been created according to section 4.1, it can be started, stopped and restarted within the Job Handler GUI as visible in the following recording:

Note: Please be aware that stopping a job may take some time as it is shut down gracefully.

4.3) Delete Job(s)

When a job created in section 4.1 is no longer needed, it can be easily deleted within the Job Handler GUI as shown in the following recording:

Note: Please be aware that only jobs with a status "created" or "completed" can be deleted.

5) Limitations & Outlook

Although the DCF is already quite powerful, it does have limitations, which will be presented below. For each limitation, a possible solution is given:

• Limitation 01: At this stage, the project is only considered to run on a single host setup, meaning that all six components introduced in Section 2.1 run on the same host machine. While this setup may be appropriate in the context of the current project, and well suited for Kafka, the Kafka Schema Registry, and the Job Handler, for deployment in a production environment, it is not appropriate for the Container Handler (Agent) and the Data Generator(s).

• Outlook 01: For the DCF to be "truly" distributed, it is necessary to containerise the Container Handler (Agent) and make it available via a container registry. Containerisation makes it easy to run the Container Handler (Agent) on multiple host machines, so that submitted jobs can be distributed across multiple host machines. The small amount of configuration required can be easily provided via environment variables that are passed to th
  
  

• Limitation 02: As outlined in Section 2.1, the three management topics Job_Handling, Job_Status and Agent_Status each have one partition and a replication factor of one. While the replication factor enhances durability, availability and is not the main focus of the limitation (and not considered subsequently), it is the partitioning that presents a challenge. The reason is straightforward: When working with a single partition, only one consumer from a consumer group can consume messages from that specific partition. What does this mean? While, this is not an issue for the Job_Status and Agent_Status topics, as they are only consumed by the Job Handler (one consumer), the Job_Handling topic is problematic because, when scaling the Container Handler (Agent), it does not have any effect, as only one Container Handler (Agent) is able to processes submitted jobs.

• Outlook 02: The following changes are required to overcome the limitation:

  • The Kafka topic Job_Handling must be split into two, for example Job_Creation and Job_Deletion:
    • The Kafka topic Job_Creation must have as many partitions as Container Handlers (Agents). This ensures that jobs (by default) are distributed in a round robin manner on the different Container Handlers (Agents). While adding partitions at runtime is possible, deleting them is not. However, if the number of partitions exceeds the number of Container Handlers (Agents), one Container Handler (Agent) will consume messages from multiple partitions, ensuring that all messages from the Kafka topic Job_Creation are handled accordingly.
    • The Kafka topic Job_Deletion requires only one partition, but further configuration is necessary when it comes to the group_id (see below).
  • It is necessary to adjust the Container Handler (thread within the Container Handler (Agent)). Rather than working with only one thread, it is necessary to introduce another thread. These threads will be referred to as Create Thread and Delete Thread (which replace the Container Handler thread):
    • Create Thread: The purpose of this thread is to consume messages from the Kafka topic Job_Creation. It is imperative that all of these consumers (on the different Container Handlers (Agents)) reside in the same group ID, to ensure that creation jobs are only handled once. Obviously, the thread must be able to create Data Generators
    • Delete Thread: The purpose of this thread is to consume messages from the Kafka topic Job_Deletion. All of these consumers (on the different Container Handlers (Agents)) must reside in a different group ID, to ensure each and every Container Handler (Agent) is able to check deletion message and verify the relevance of the respective message to the agent (implemented via an Agent-ID). Obviously, the thread must be able to delete Data Generators.

  With such a configuration, it is guaranteed that tasks are allocated to various Container Handlers (Agents) in a round-robin manner (by default), with each Container Handler (Agent) capable of reading deletion messages and determining the relevance of a specific message and whether action is necessary.
  
  

• Limitation 03: As of now, images of Data Generators can only be fetched from DockerHub.

• Outlook 03: In order to facilitate access to multiple container registries, it is necessary to adjust the Job Handling Avro Schema. This process involves the addition of information regarding the container registry, as well as any relevant authentication details if applicable. This information is then utilised by the Container Handler (Agent) to retrieve images from a range of container registries and start the corresponding containers.
  
  

• Limitation 04: At present, the Container Handler (Agent) is subject to only one active check: the Docker Daemon Checker.

• Outlook 04: In order to enhance the reliability of the Container Handler (Agent), it is possible to incorporate supplementary checks, such as those pertaining to resource availability. These checks can be simply be integrated into the Container Handler (Agent) (), (see observe_agent() and checker() within the source code), with all other processes being managed automatically.
  
  

• Limitation 05: The Job Handler currently uses a SQlite database. While this is sufficient for development purposes, it is not recommended for production environments.

• Outlook 05: Future work should focus on migrating the database to a more robust solution, such as PostgreSQL. This can be achieved by adjusting the Django settings file accordingly and deploying a PostgreSQL instance.

6) Licensing

The DCF is available under the MIT license.

7) Contacts

• Christian Bieri, Site Reliability Engineer, info@christianbieri.ch
• Frederico Fischer, DevOps Engineer, fredae14@hotmail.com
• Leandro Hoenen, DevOps Engineer, exponent_stooge805@perfunc.ch