A Distributed Code Execution Engine in Pekko with Scala

Architecture Diagram

Let’s explain step by step what happens during code execution:

• User sends request at POST http://localhost/python with python code attached to request body
• http processes the request
• Randomly chosen load balancer sends message to worker router on some node
• worker router sends message to worker which starts execution
• … skipping lots of details …
• worker responds to load balancer once the code is executed
• load balancer forwards message to http and user receives the output

Docker Runtime

There are a few options to my knowledge for executing the submitted code with the help of docker containers. All I can do is redirect you to the great StackOverflow post which expands on this issue.

Long story short, I decided to go with DooD - Docker out of Docker.

DooD Uses the host’s Docker daemon by mounting /var/run/docker.sock into the container.

It means that if the host machine has the docker engine installed on it:

• It can start container A using /var/run/docker.sock which listens for commands
• It can start “sibling” container B by executing commands such as docker run ... from container A, using /var/run/docker.sock

In our domain terms:

• Container A will be worker-1 container, which will be starting sibling containers
• Sibling containers will be short-lived, limited (CPU, RAM, timeout) containers which run python, java and such processes

Advantages of DooD:

• Efficient: Direct use of host resources.
• Simple: No extra Docker daemons.
• Fast: No nested virtualization overhead.
• Resource-efficient: Shared daemon optimizes resource use.

Disadvantages:

• Security risk: Access to host Docker daemon.
• Less isolation: Shared daemon.
• Overhead: Starting/Stopping containers all the time

Docker Compose

Since we want to run a pekko cluster we could use docker-compose and its declarative definitions for cluster components such as master and worker nodes, with shared volume, image names, ports mappings, environment variables and so on.

This is how docker-compose.yaml could look like in the root directory of our project:

version: "3"
services:
  master:
    image: braindrill:latest
    build:
      context: .
      dockerfile: Dockerfile
    container_name: "master"
    tty: true
    stdin_open: true
    ports:
      - "1600:1600"
      - "8080:8080"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - engine:/data
    working_dir: /app
    environment:
      CLUSTER_IP: "master"
      CLUSTER_PORT: "1600"
      CLUSTER_ROLE: "master"
      SEED_IP: "master"
  worker-1:
    image: braindrill:latest
    build:
      context: .
      dockerfile: Dockerfile
    tty: true
    stdin_open: true
    ports:
      - "17350:17350"
    security_opt:
      - "seccomp=unconfined"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - engine:/data
    working_dir: /app
    environment:
      CLUSTER_IP: "worker-1"
      CLUSTER_PORT: "17350"
      CLUSTER_ROLE: "worker"
      SEED_IP: "master"
      SEED_PORT: "1600"
  worker-2:
    image: braindrill:latest
    build:
      context: .
      dockerfile: Dockerfile
    tty: true
    stdin_open: true
    ports:
      - "17351:17351"
    security_opt:
      - "seccomp=unconfined"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - engine:/data
    working_dir: /app
    environment:
      CLUSTER_IP: "worker-2"
      CLUSTER_PORT: "17351"
      CLUSTER_ROLE: "worker"
      SEED_IP: "master"
      SEED_PORT: "1600"
  worker-3:
    image: braindrill:latest
    build:
      context: .
      dockerfile: Dockerfile
    tty: true
    stdin_open: true
    ports:
      - "17352:17352"
    security_opt:
      - "seccomp=unconfined"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - engine:/data
    working_dir: /app
    environment:
      CLUSTER_IP: "worker-3"
      CLUSTER_PORT: "17352"
      CLUSTER_ROLE: "worker"
      SEED_IP: "master"
      SEED_PORT: "1600"

volumes:
  engine:
    external: true

This docker-compose.yaml file sets up a Pekko cluster with one master and three worker services.

Key points include:

1. Version: Uses Docker Compose version 3.
2. Services: Defines four services - master, worker-1, worker-2, and worker-3.

Dockerfile

Dockerfile is a declarative file for building a docker image.

# Use an official OpenJDK runtime as a parent image
FROM hseeberger/scala-sbt:17.0.2_1.6.2_3.1.1

# Update the repository sources list and install required packages
RUN apt-get update && \
    apt-get install -y \
    docker.io # docker.io needed to run `docker` commands from the container itself

# Set the working directory to /tmp
WORKDIR /app

# Copy the current directory contents into the container at /app
COPY . /app

# Build the Scala project
RUN sbt clean assembly

# Run application when the container launches
ENTRYPOINT ["java", "-jar", "target/scala-3.4.1/braindrill.jar"]

Local Deployment Shell Script

Deployment shell script is a handy tool that enables us to run the project locally with a single command.

The goal is to automate the following:

• Creating an image - braindrill
• Creating a volume - engine
• Pulling programming language runtime images (java, python and others)
• Redeploying the nodes with the new version of braindrill image if required

So, it could look like the following:

#!/bin/bash

# Function to print help message
print_help() {
    echo "Usage: $0 [rebuild]"
    echo "  rebuild   Add --build flag to docker-compose up"
}

# Check for the "rebuild" argument
DOCKER_COMPOSE_CMD="docker-compose up"
if [ "$1" == "rebuild" ]; then
    DOCKER_COMPOSE_CMD="docker-compose up --build"
elif [ -n "$1" ]; then
    print_help
    exit 1
fi

# Create Docker volume if it doesn't exist
VOLUME_NAME="engine"
if docker volume inspect "$VOLUME_NAME" > /dev/null 2>&1; then
  echo "Docker volume '$VOLUME_NAME' already exists."
else
  echo "Creating Docker volume '$VOLUME_NAME'."
  docker volume create "$VOLUME_NAME"
fi

# Docker images to be pulled
DOCKER_IMAGES=(
  "openjdk:17"
  "python"
  "node"
  "ruby"
  "perl"
  "php"
)

# Pull Docker images in parallel if they don't already exist
for IMAGE in "${DOCKER_IMAGES[@]}"; do
  if docker image inspect "$IMAGE" > /dev/null 2>&1; then
    echo "Docker image '$IMAGE' already exists."
  else
    echo "Pulling Docker image '$IMAGE' in the background."
    docker pull "$IMAGE" &
  fi
done

# Wait for all background jobs to complete
wait
echo "All Docker images pulled."

echo "Running command: $DOCKER_COMPOSE_CMD"
$DOCKER_COMPOSE_CMD

This deployment shell script (deploy.sh) sets up a Docker environment and optionally rebuilds the Docker images. Key points include:

1. Help Function:

   • Provides usage instructions for the script.

2. Rebuild Argument Check:

   • If “rebuild” is passed as an argument, the script adds --build to the docker-compose up command.
   • If an unrecognized argument is passed, it prints the help message and exits.

3. Docker Volume Management:

   • Checks if a Docker volume named engine exists.
   • If not, it creates the volume which is necessary for worker containers to access temporary files to be executed later

4. Docker Image Management:

   • Defines a list of Docker images to be pulled (openjdk:17, python, node, ruby, perl, php).
   • Checks if each image exists locally and pulls it in the background if it does not.

5. Parallel Image Pulling:

   • Uses background jobs to pull Docker images simultaneously for efficiency.
   • Waits for all background jobs to complete.

6. Run Docker Compose:

   • Executes the docker-compose up command, with or without the --build flag based on the initial argument.

Pekko Actor System Configuration

• provider = cluster: This setting specifies that the actor system will use clustering capabilities.
• serialization-bindings: This section defines serialization bindings for specific classes. Here, any class implementing serialization.CborSerializable will be serialized using the jackson-cbor serializer.

HTTP Configuration

• host-connection-pool.max-open-requests: This setting specifies the maximum number of open requests in the HTTP host connection pool. It is set to 256.

Remote Communication Configuration

• remote.artery: This section configures the Artery transport (a remoting mechanism in Pekko).
• canonical.hostname: This sets the hostname for the actor system, which is derived from clustering.ip.
• canonical.port: This sets the port for the actor system, which is derived from clustering.port.
• large-message-destinations: Specifies destinations for large messages. In this case, any destination matching the pattern /temp/load-balancer-* will be treated as a large message destination.

Cluster Configuration

• seed-nodes: Defines the initial contact points for the cluster, using placeholders for cluster name, seed IP, and seed port.
• roles: Specifies the roles of the cluster node, derived from clustering.role.
• downing-provider-class: Specifies the class for handling split-brain scenarios. Here, it’s set to SplitBrainResolverProvider.

Http Server Configuration

• http.port: Sets the HTTP server port to 8080.
• http.host: Sets the HTTP server host to 0.0.0.0, meaning it will bind to all available network interfaces.

Clustering Variables

• ip: Default IP address for clustering is 127.0.0.1. It can be overridden by the environment variable CLUSTER_IP.
• port: Default port for clustering is 1600. It can be overridden by the environment variable CLUSTER_PORT.
• role: Role of the cluster node, which can be set using the environment variable CLUSTER_ROLE.
• seed-ip: Default seed IP address is 127.0.0.1. It can be overridden by CLUSTER_IP or SEED_IP.
• seed-port: Default seed port is 1600. It can be overridden by the environment variable SEED_PORT.
• cluster.name: Name of the cluster, set to ClusterSystem.

This configuration file is designed to be flexible, allowing for environment-specific overrides while providing sensible defaults, it plays crucial role with running the Pekko cluster within the containers.

transformation.conf

The project also requires a special configuration for capturing properties for domain objects such as worker, load-balancer and so on.

include "application"

transformation {
  workers-per-node = 32
  load-balancer = 3
}

Here, it means that each node will have 32 worker actors and master node will have 3 load balancer actors. In the real world, choosing those numbers would depend on multiple variables that must be collected and analyzed in production. In my opinion, those numbers are optimized based on empirical evidence rather than theoretical results.

Serialization

Since we’re building a cluster we need to keep in mind that actor messages must be serialized over wire. It means that messages could be sent from JVM to JVM which requires enabling the serialization protocol.

For that we need a simple trait definition under serialization package, as we’ve already defined in configuration.

package serialization

// just a marker trait to tell pekko to serialize messages into CBOR using Jackson for sending over the network
trait CborSerializable

Worker

Let’s quickly evaluate what we would expect from Worker.

Worker actor will live on worker node and will be reacting to StartExecution message.

Also, during runtime, it must be aware of which languages are supported, so that it quickly replies something like ExecutionFailed response with the message "unsupported language".

For successful case, we could have ExecutionSucceeded message that will be forwarded back to the original sender.

How about making Worker actor work within a cluster? We said earlier that the Worker actor will be expecting messages from master node, here enters remoting and other good stuff.

Somehow, Worker actor must be discovered within pekko cluster so that messages can be routed to it, for that, we’ll be using ServiceKey and later, Router-s which are special actors.

So, keeping all this in mind the code could look like this:

package workers

import org.apache.pekko.actor.typed.receptionist.ServiceKey
import workers.children.FileHandler.In.PrepareFile
import org.apache.pekko.actor.typed.{ActorRef, Behavior}
import org.apache.pekko.actor.typed.scaladsl.Behaviors
import serialization.CborSerializable
import workers.children.FileHandler

import scala.util._

object Worker {

   // this enables Worker to be discovered by actors living on other nodes
   val WorkerRouterKey = ServiceKey[Worker.StartExecution]("worker-router.StartExecution")

   // simple model for grouping compiler, file extension and docker image for the programming language
   private final case class LanguageSpecifics(
       compiler: String,
       extension: String,
       dockerImage: String
   )

   // language specifics per language
   private val languageSpecifics: Map[String, LanguageSpecifics] = Map(
      "java" -> LanguageSpecifics(
         compiler = "java",
         extension = ".java",
         dockerImage = "openjdk:17"
      ),
      "python" -> LanguageSpecifics(
         compiler = "python3",
         extension = ".py",
         dockerImage = "python"
      ),
      "ruby" -> LanguageSpecifics(
         compiler = "ruby",
         extension = ".ruby",
         dockerImage = "ruby"
      ),
      "perl" -> LanguageSpecifics(
         compiler = "perl",
         extension = ".pl",
         dockerImage = "perl"
      ),
      "javascript" -> LanguageSpecifics(
         compiler = "node",
         extension = ".js",
         dockerImage = "node"
      ),
      "php" -> LanguageSpecifics(
         compiler = "php",
         extension = ".php",
         dockerImage = "php"
      )
   )

   // a parent type for modeling incoming messages
   sealed trait In

   final case class StartExecution(
       code: String,
       language: String,
       replyTo: ActorRef[Worker.ExecutionResult]
   ) extends In with CborSerializable

   // a parent type that models successful and failed executions
   sealed trait ExecutionResult extends In {
      def value: String
   }

   final case class ExecutionSucceeded(value: String) extends ExecutionResult with CborSerializable
   final case class ExecutionFailed(value: String)    extends ExecutionResult with CborSerializable

   // constructor for creating Behavior[In]
   def apply(workerRouter: Option[ActorRef[Worker.ExecutionResult]] = None): Behavior[In] = {
      Behaviors.setup[In] { ctx =>
         val self = ctx.self

         Behaviors.receiveMessage[In] {
            case msg @ StartExecution(code, lang, replyTo) =>
               ctx.log.info(s"{} processing StartExecution", self)
               languageSpecifics get lang match {
                  case Some(specifics) =>
                     val fileHandler = ctx.spawn(FileHandler(), s"file-handler")
                     ctx.log.info(s"{} sending PrepareFile to {}", self, fileHandler)

                     fileHandler ! FileHandler.In.PrepareFile(
                        name =
                           s"$lang${Random.nextInt}${specifics.extension}", // random number for avoiding file overwrite/shadowing
                        compiler = specifics.compiler,
                        dockerImage = specifics.dockerImage,
                        code = code,
                        replyTo = ctx.self
                     )
                  case None =>
                     val reason = s"unsupported language: $lang"
                     ctx.log.warn(s"{} failed execution due to: {}", self, reason)

                     replyTo ! Worker.ExecutionFailed(reason)
               }

               // register original requester
               apply(workerRouter = Some(replyTo))

            case msg @ ExecutionSucceeded(result) =>
               ctx.log.info(s"{} sending ExecutionSucceeded to {}", self, workerRouter)
               workerRouter.foreach(_ ! msg)

               apply(workerRouter = None)

            case msg @ ExecutionFailed(reason) =>
               ctx.log.info(s"{} sending ExecutionFailed to {}", self, workerRouter)
               workerRouter.foreach(_ ! msg)

               apply(workerRouter = None)
         }
      }
   }
}

So, Worker actor is designed to handle code execution requests for various programming languages. Here’s a concise breakdown:

• ServiceKey: WorkerRouterKey enables the actor to be discovered across different nodes.
• LanguageSpecifics: A case class and a map define compiler, file extension, and Docker image specifics for each supported programming language.
• Messages: The In trait and its implementations (StartExecution, ExecutionResult, ExecutionSucceeded, ExecutionFailed) model the messages the actor can handle.
• Behavior Setup: The apply method sets up the actor’s behavior, processing incoming messages.
• On StartExecution, it logs the event, checks for language support, and spawns a FileHandler actor to handle file preparation. If the language isn’t supported, it responds with ExecutionFailed.
• On ExecutionSucceeded and ExecutionFailed, it logs the outcome and sends the result back to the requester.
• ActorRef Management: Manages the original requester (workerRouter which will be defined later) to send back execution results.
• The imports import workers.children.FileHandler.In.PrepareFile and import workers.children.FileHandler will be covered next, focusing on file preparation logic.

With this, we can move to FileHandler.

File Handler

FileHandler will be a local child actor for the Worker actor on the same worker node.

We’ve already seen PrepareFile message mentioned in Worker actor, but there must also be defined something like FilePrepared and FilePreparationFailed messages for covering both possible outcomes.

If file will be created, FileHandler should simply proceed with code execution, otherwise it should report immediately why the process failed.

With those expectations in mind, the code could look like the following:

package workers.children

import org.apache.pekko.actor.typed.{ActorRef, Terminated}
import org.apache.pekko.actor.typed.scaladsl.Behaviors
import org.apache.pekko.stream.scaladsl.{FileIO, Source}
import org.apache.pekko.util.ByteString
import workers.children.FileHandler.In.PrepareFile
import workers.Worker

import java.io.File
import java.nio.file.Path
import scala.concurrent.Future
import scala.util._

object FileHandler {

   sealed trait In
   object In {
      case class PrepareFile(
          name: String,
          code: String,
          compiler: String,
          dockerImage: String,
          replyTo: ActorRef[Worker.In]
      ) extends In

      case class FilePrepared(
          compiler: String,
          file: File,
          dockerImage: String,
          replyTo: ActorRef[Worker.In]
      ) extends In

      case class FilePreparationFailed(why: String, replyTo: ActorRef[Worker.In]) extends In
   }

   def apply() = Behaviors
     .receive[In] { (ctx, msg) =>
       import CodeExecutor.In._
       import ctx.executionContext
       import ctx.system

       val self = ctx.self

       ctx.log.info(s"{}: processing {}", self, msg)

       msg match {
          case In.PrepareFile(name, code, compiler, dockerImage, replyTo) =>
             val filepath = s"/data/$name"
             val asyncFile = for {
                file <- Future(File(filepath))
                _ <- Source
                        .single(code)
                        .map(ByteString.apply)
                        .runWith(FileIO.toPath(Path.of(filepath)))
             } yield file

             ctx.pipeToSelf(asyncFile) {
                case Success(file) => In.FilePrepared(compiler, file, dockerImage, replyTo)
                case Failure(why)  => In.FilePreparationFailed(why.getMessage, replyTo)
             }

             Behaviors.same

          case In.FilePrepared(compiler, file, dockerImage, replyTo) =>
             val codeExecutor = ctx.spawn(CodeExecutor(), "code-executor")
             // observe child for self-destruction
             ctx.watch(codeExecutor)
             ctx.log.info("{} prepared file, sending Execute to {}", self, codeExecutor)
             codeExecutor ! Execute(compiler, file, dockerImage, replyTo)

             Behaviors.same

          case In.FilePreparationFailed(why, replyTo) =>
             ctx.log.warn(
                "{} failed during file preparation due to {}, sending ExecutionFailed to {}",
                self,
                why,
                replyTo
             )
             replyTo ! Worker.ExecutionFailed(why)

             Behaviors.stopped
       }
    }
    .receiveSignal { case (ctx, Terminated(ref)) =>
       ctx.log.info(s"{} is stopping because child actor: {} was stopped", ctx.self, ref)

       Behaviors.stopped
    }
  }

The details:

• Messages: The In enum defines messages the actor can handle, including PrepareFile, FilePrepared, and FilePreparationFailed.
• Behavior Setup: The apply method sets up the actor’s behavior
• On receiving PrepareFile, it logs the event, attempts to write the code to a file asynchronously, and uses pipeToSelf pattern to handle the result.
• On FilePrepared, it spawns a CodeExecutor actor to execute the code, watches it for termination, and sends an Execute message to the executor.
• On FilePreparationFailed, it logs the failure and sends an ExecutionFailed message back to the requester.

And we arrived at the point where we can talk about the most important actor - CodeExecutor.

Constants and Imports

1. Constants:

   • KiloByte, MegaByte, TwoMegabytes: Constants for byte size calculations.
   • AdjustedMaxSizeInBytes: Adjusted max size for output (approximately 409.6 KB).
   • MaxOutputSize: Maximum allowed output size for the code execution.

2. Imports:

   • Imports various Pekko and Scala utilities for actor behavior, stream handling, and future operations.

Messages (In)

The In enum defines messages the actor can handle:

• Execute: Message to execute a given file with a specific compiler and Docker image, and send the result to the specified ActorRef.
• Executed: Message indicating the execution result, including output and exit code.
• ExecutionFailed: Message indicating the execution failed with a reason.
• ExecutionSucceeded: Message indicating the execution succeeded with the output.

Throwable for Large Output

A custom TooLargeOutput throwable is defined to handle scenarios where the code output exceeds the allowed size.

Behavior Definition

The apply method defines the actor’s behavior using Akka’s Behaviors.receive method:

In.Execute case

1. Logging: Logs that code execution has started.
2. Running Docker Command: Constructs and runs a Docker command with a 2-second timeout and memory limits.
   • The Docker command binds the engine volume, sets the working directory to /data, and runs the provided compiler and file within the specified Docker image.
3. Handling Streams: Handles both the standard and error streams from the Docker process.
4. Reading Output: Reads the success and error streams and joins them with the exit code.
5. Deleting File: Schedules the file for deletion to free up memory.
6. Piping Result: Uses ctx.pipeToSelf to handle the asynchronous execution result.
   • On success, checks the exit code and constructs appropriate messages (ExecutionFailed for specific exit codes or ExecutionSucceeded).
   • On failure, logs the exception and constructs an ExecutionFailed message.

In.ExecutionSucceeded case

1. Logging: Logs successful execution.
2. Replying: Sends an ExecutionSucceeded message to the requester.
3. Stopping Behavior: Stops the actor.

In.ExecutionFailed case

1. Logging: Logs the failure reason.
2. Replying: Sends an ExecutionFailed message to the requester.
3. Stopping Behavior: Stops the actor.

Helper Methods

1. readOutput: Defines a sink that reads ByteString input, converts it to a UTF-8 string, and checks if the output size exceeds the maximum allowed size.
2. src: Creates a source from an input stream.
3. run: Executes a command using sys.runtime.exec and returns a future of the process.

Detailed Workflow

1. Execution Start: When the Execute message is received, the actor starts by logging the event.
2. Docker Command: It then constructs and runs the Docker command with specified limits.
3. Stream Handling: Both the success and error streams from the Docker process are handled and read concurrently.
4. Output Management: The output is read and checked against the maximum allowed size.
5. File Deletion: The file is scheduled for deletion to free up memory.
6. Result Handling: The result of the execution (success or failure) is handled and appropriate messages are constructed and sent to the requester.
7. Actor Lifecycle: Depending on the message (ExecutionSucceeded or ExecutionFailed), the actor stops itself after sending the reply.

This code ensures that code execution is managed efficiently within a controlled environment (Docker) and handles different outcomes (success, failure, timeouts, and memory limits) gracefully.

Now we can move on and talk about the clustering stuff.

Components and Their Roles

1. Cluster and Node Configuration:

   • The cluster is initialized using Cluster(ctx.system), and the node’s roles and configuration settings are obtained.

2. Worker Nodes:

   • Worker Router:
     • Each worker node should initialize a pool of worker actors using a Round Robin routing strategy. The pool size is configurable via transformation.workers-per-node.
     • The worker router is registered with the Receptionist, making it discoverable across the cluster.

3. Master Node:

   • Load Balancers:
     • A pool of load balancers forwards execution tasks to the worker routers. The pool size is configurable via transformation.load-balancer.
     • Each load balancer uses group routing to distribute tasks to remote worker routers.

4. HTTP Server:

   • The master node sets up an HTTP server to accept task submissions. It listens on a configurable host and port.
   • Incoming tasks are forwarded to a randomly selected load balancer, which then distributes them to the worker nodes.

Justification for the Components

1. Efficiency and Scalability:

   • Using a cluster setup with dedicated worker and master nodes allows for efficient task distribution and parallel processing.
   • The Round Robin routing of worker actors and load balancers ensures balanced workload distribution.

2. Dynamic Discovery and Registration:

   • The Receptionist enables dynamic discovery of worker routers across the cluster, facilitating seamless scaling and fault tolerance.

3. Robust and Resilient Execution:

   • Supervisory strategies for worker actors enhance fault tolerance by restarting actors on failure.
   • The use of ask patterns with timeouts ensures that task submissions are handled gracefully, even in the event of errors.

4. Configurability:

   • Key parameters like the number of workers, load balancers, and server settings are configurable, allowing for flexibility and adaptability to different environments and workloads.

By leveraging these components and strategies, the cluster system can efficiently manage and distribute tasks across multiple nodes, providing a robust solution for distributed computing scenarios.

Keeping in mind all this, we could translate those expectations in code:

package cluster

import workers.Worker
import org.apache.pekko
import org.apache.pekko.actor.typed.receptionist.Receptionist
import org.apache.pekko.actor.typed.scaladsl.Behaviors
import org.apache.pekko.cluster.typed.Cluster
import pekko.actor.typed.{ActorSystem, Behavior}
import pekko.http.scaladsl.Http
import pekko.http.scaladsl.server.Directives.*
import org.apache.pekko.util.Timeout
import pekko.actor.typed.scaladsl.AskPattern.schedulerFromActorSystem
import pekko.actor.typed.scaladsl.AskPattern.Askable
import pekko.actor.typed.*
import pekko.actor.typed.scaladsl.*
import workers.Worker.*

import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration.*
import scala.util.*

object ClusterSystem {

   def apply(): Behavior[Nothing] = Behaviors.setup[Nothing] { ctx =>

      import ctx.executionContext

      val cluster = Cluster(ctx.system)
      val node = cluster.selfMember
      val cfg = ctx.system.settings.config

      if (node hasRole "worker") {
        val numberOfWorkers = Try(cfg.getInt("transformation.workers-per-node")).getOrElse(50)
        // actor that sends StartExecution message to local Worker actors in a round robin fashion
        val workerRouter = ctx.spawn(
          behavior = Routers
            .pool(numberOfWorkers) {
              Behaviors
                .supervise(Worker().narrow[StartExecution])
                .onFailure(SupervisorStrategy.restart)
            }
            .withRoundRobinRouting(),
          name = "worker-router"
        )
        // actors are registered to the ActorSystem receptionist using a special ServiceKey.
        // All remote worker-routers will be registered to ClusterBootstrap actor system receptionist.
        // When the "worker" node starts it registers the local worker-router to the Receptionist which is cluster-wide
        // As a result "master" node can have access to remote worker-router and receive any updates about workers through worker-router
        ctx.system.receptionist ! Receptionist.Register(Worker.WorkerRouterKey, workerRouter)
      }

      if (node hasRole "master") {
        given system: ActorSystem[Nothing] = ctx.system

        given ec: ExecutionContextExecutor = ctx.executionContext

        given timeout: Timeout = Timeout(3.seconds)

        val numberOfLoadBalancers = Try(cfg.getInt("transformation.load-balancer")).getOrElse(3)
        // pool of load balancers that forward StartExecution message to the remote worker-router actors in a round robin fashion
        val loadBalancers = (1 to numberOfLoadBalancers).map { n =>
           ctx.spawn(
              behavior = Routers
                      .group(Worker.WorkerRouterKey)
                      .withRoundRobinRouting(), // routes StartExecution message to the remote worker-router
              name = s"load-balancer-$n"
           )
        }
      }

      val route =
         pathPrefix("lang" / Segment) { lang =>
            post {
               entity(as[String]) { code =>
                  val loadBalancer = Random.shuffle(loadBalancers).head
                  val asyncResponse = loadBalancer
                          .ask[ExecutionResult](StartExecution(code, lang, _))
                          .map(_.value)
                          .recover(_ => "something went wrong")
                  complete(asyncResponse)
               }
            }
         }

      val host = Try(cfg.getString("http.host")).getOrElse("0.0.0.0")
      val port = Try(cfg.getInt("http.port")).getOrElse(8080)

      Http()
        .newServerAt(host, port)
        .bind(route)

      ctx.log.info("Server is listening on {}:{}", host, port)

      Behaviors.empty[Nothing]
   }
}

The code above sets up an Pekko cluster with worker and master nodes.

1. Receptionist: Registers the worker routers so they can be discovered cluster-wide.
2. HTTP Server: On the master node, it starts an HTTP server to accept code execution requests.
3. Configuration: Reads settings for the number of workers, load balancers, and HTTP server details.
4. Worker Router: Each worker node creates a pool of “worker actors using round-robin routing.
5. Load-Balancer: The master node creates load balancers to forward tasks to worker routers.

Algorithm for Code Execution:

• HTTP POST request with code is received.
• A random load balancer forwards the request to a worker router.
• The worker router assigns the task to a worker actor, which executes the code and returns the result.

With that we finish the part which is concerned with code.

Let’s move on to the final part of the project - running it all locally.

Key Components and Features

1. Cluster Awareness with Receptionist and Router:

   • By using Receptionist and Router, we ensured our system is cluster-aware, allowing dynamic scaling and efficient task distribution among worker nodes.

2. Actors for Task Management:

   • Worker: Coordinates tasks on the worker nodes, ensuring seamless processing and communication with the master node.
   • FileHandler: Manages file preparation required for code execution.
   • CodeExecutor: Executes the code snippets in a sibling docker container and collects the output.

3. Docker-out-of-Docker (DooD) Approach:

   • Each worker node runs as a Docker container, and within these containers, we utilize the DooD approach to execute sibling containers. This method provides isolation and security, ensuring each code execution happens in a controlled and disposable environment.

4. Resource Management and Security:

   • We’ve set stringent resource limits (CPU, RAM) and a 2-second timeout for code execution. These measures protect the system from malicious code, such as infinite loops, ensuring stability and security.

5. Support for Multiple Languages:

   • Our engine supports six programming languages, demonstrating its versatility and adaptability to various use cases.

Scalability and Efficiency

The design choices made in this project ensure that our remote code execution engine can scale efficiently with the demands. By distributing the workload across multiple worker nodes and dynamically managing these nodes using cluster-aware techniques, we can handle a large number of concurrent code execution requests without compromising performance.

Final Thoughts

Building this distributed system with Scala 3 and Apache Pekko has been an enlightening experience. We’ve harnessed the power of actor-based concurrency, cluster management, and containerization to create a resilient and secure remote code execution engine. This project exemplifies how modern technologies can be integrated to solve complex problems in a scalable and efficient manner.

Whether you’re looking to implement a similar system or seeking insights into distributed computing with Scala and Pekko, we hope this blog post has provided valuable knowledge and inspiration.