Distributed application with Akka Remoting

README.md

@@ -6,16 +6,21 @@ write good Scala or Akka code.
 
 It has no tests, but on the other side there isn't much code either!
 
-Run the application locally:
+Run the application locally (each command on different terminal):
 ```sh
-sbt frontend/run
+REMOTE_PORT=2551 sbt backend/run
+REMOTE_PORT=2552 sbt backend/run
+REMOTE_PORT=2553 BACKEND_NODES=127.0.0.1:2551,127.0.0.1:2552 sbt frontend/run
 curl http://localhost:9000/greet/World
 ```
 
 Run it on Kubernetes (Minikube):
 ```sh
 eval $(minikube docker-env)
 sbt docker:publishLocal
-kubectl create -f kubernetes/
+kubectl create -f kubernetes/backend-service.yaml
+kubectl create -f kubernetes/backend-deployment.yaml
+kubectl create -f kubernetes/frontend-service.yaml
+kubectl create -f kubernetes/frontend-deployment.yaml
 curl $(minikube service frontend-service --url)/greet/World
 ```

backend/src/main/scala/com/example/backend/BackendApp.scala

@@ -0,0 +1,10 @@
+package com.example.backend
+
+import com.example.ExampleApp
+
+object BackendApp extends App with ExampleApp {
+
+  system.actorOf(Backend.props, "backend")
+
+  log.info("Backend Service started")
+}

build.sbt

@@ -1,17 +1,57 @@
 import BuildSettings._
 import Dependencies._
 import Resolvers._
+import com.typesafe.sbt.packager.archetypes.JavaAppPackaging.autoImport.bashScriptExtraDefines
+import com.typesafe.sbt.packager.docker._
 
 lazy val commonSettings = buildSettings ++ Seq(
   resolvers += akkaHttpJsonResolvers
 )
 
+lazy val dockerSettings = Seq(
+  dockerExposedPorts := Seq(Docker.akkaTcpPort),
+  bashScriptExtraDefines := Seq(Docker.bashExports)
+)
+
+lazy val frontendDockerSettings = dockerSettings ++ Seq(
+  dockerCommands := Seq(
+    Cmd("FROM", "java:latest"),
+    Cmd("USER", "root"),
+    ExecCmd("RUN", "apt-get", "-qq", "update"),
+    ExecCmd("RUN", "apt-get", "-yq", "install", "dnsutils"),
+    ExecCmd("RUN", "apt-get", "clean"),
+    ExecCmd("RUN", "rm", "-rf", "/var/lib/apt/lists/*")
+  ) ++ dockerCommands.value.filterNot {
+    case Cmd("FROM", _) => true
+    case _ => false
+  },
+  bashScriptExtraDefines := Seq(Docker.frontendBashExports)
+)
+
 lazy val root = (project in file("."))
   .settings(buildSettings: _*)
-  .aggregate(frontend)
+  .aggregate(frontend, backend)
+
+lazy val common = project
+  .settings(commonSettings: _*)
+  .settings(libraryDependencies ++= akka)
 
 lazy val frontend = project
   .settings(commonSettings: _*)
   .settings(libraryDependencies ++= akkaHttp)
   .enablePlugins(JavaAppPackaging)
   .enablePlugins(DockerPlugin)
+  .settings(frontendDockerSettings)
+  .dependsOn(`backend-api`, common)
+
+lazy val backend = project
+  .settings(commonSettings: _*)
+  .settings(libraryDependencies ++= akka)
+  .enablePlugins(JavaAppPackaging)
+  .enablePlugins(DockerPlugin)
+  .settings(dockerSettings)
+  .dependsOn(`backend-api`, common)
+
+lazy val `backend-api` = project
+  .settings(commonSettings: _*)
+  .settings(libraryDependencies ++= akka)

common/src/main/resources/reference.conf

@@ -0,0 +1,23 @@
+akka {
+
+  actor {
+    provider = remote
+  }
+
+  remote {
+    enabled-transports = ["akka.remote.netty.tcp"]
+    netty.tcp {
+      hostname = ${app.remote.interface}
+      port = ${app.remote.port}
+    }
+  }
+}
+
+app {
+  remote {
+    interface = "127.0.0.1"
+    interface = ${?REMOTE_INTERFACE}
+    port = 0
+    port = ${?REMOTE_PORT}
+  }
+}

common/src/main/scala/com/example/ExampleApp.scala

@@ -0,0 +1,13 @@
+package com.example
+
+import akka.actor.ActorSystem
+import akka.event.LoggingAdapter
+import com.typesafe.config.ConfigFactory
+
+trait ExampleApp {
+
+  val actorSystemName = "example"
+  val config = ConfigFactory.load()
+  implicit val system = ActorSystem(actorSystemName, config)
+  val log: LoggingAdapter = system.log
+}

frontend/src/main/resources/application.conf

@@ -1,12 +1,10 @@
-akka {
-  actor {
-  }
-}
-
 app {
   http {
     interface = "0.0.0.0"
     port = 9000
     port = ${?HTTP_PORT}
   }
+  backend {
+    nodes = ${?BACKEND_NODES}
+  }
 }

frontend/src/main/scala/com/example/BackendPathsConfig.scala

@@ -0,0 +1,41 @@
+package com.example
+
+import com.typesafe.config.Config
+
+trait BackendPathsConfig {
+
+  def actorSystemName: String
+  def config: Config
+
+  def backendPaths() = remoteNodesAddress.map(node => s"$node/user/backend")
+
+  private def remoteNodesAddress: List[String] = {
+    val remoteInterface = config.getString("app.remote.interface")
+    val remotePort = config.getInt("app.remote.port")
+    if (config.hasPath("app.backend.nodes")) {
+      config.getString("app.backend.nodes")
+        .split(',').toList
+        .map(toAkkaNodeRef(remotePort))
+        .flatten
+    } else if (remotePort > 0) {
+      List(toAkkaNodeRef(remoteInterface, remotePort))
+    } else {
+      List.empty
+    }
+  }
+
+  private def toAkkaNodeRef(defaultPort: Int)(peer: String): Option[String] = {
+    peer.trim.split(':') match {
+      case Array(host, port) =>
+        Some(toAkkaNodeRef(host, port.toInt))
+      case Array(host) =>
+        if (defaultPort == 0) None
+        else Some(toAkkaNodeRef(host, defaultPort))
+    }
+  }
+
+  private def toAkkaNodeRef(host: String, port: Int) = {
+    s"akka.tcp://$actorSystemName@$host:$port"
+  }
+
+}

frontend/src/main/scala/com/example/frontend/FrontendApp.scala

@@ -1,25 +1,19 @@
 package com.example.frontend
 
-import akka.actor.{ActorRef, ActorSystem}
-import akka.event.LoggingAdapter
+import akka.actor.ActorRef
+import akka.routing.RoundRobinGroup
 import akka.util.Timeout
-import com.example.backend.Backend
-import com.typesafe.config.ConfigFactory
+import com.example.{BackendPathsConfig, ExampleApp}
 
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
-object FrontendApp extends App
+object FrontendApp extends App with ExampleApp with BackendPathsConfig
   with HttpServerStartup with FrontendRestApi {
 
-  val actorSystemName = "example"
-  val config = ConfigFactory.load()
-
-  implicit val system = ActorSystem(actorSystemName, config)
   override implicit val executionContext: ExecutionContext = system.dispatcher
-  val log: LoggingAdapter = system.log
 
-  override val backend: ActorRef = system.actorOf(Backend.props, "backend")
+  override val backend: ActorRef = system.actorOf(RoundRobinGroup(backendPaths()).props(), "backend")
   override implicit val backendTimeout: Timeout = 5 seconds
 
   val interface = config.getString("app.http.interface")

kubernetes/backend-deployment.yaml

@@ -0,0 +1,15 @@
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: backend-deployment
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: backend
+        cluster: example
+    spec:
+      containers:
+      - name: backend
+        image: backend:0.1

kubernetes/backend-service.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: backend-service
+spec:
+  type: ClusterIP
+  clusterIP: None
+  selector:
+    app: backend
+  ports:
+  - port: 2551

kubernetes/frontend-deployment.yaml

@@ -14,3 +14,6 @@ spec:
         image: frontend:0.1
         ports:
         - containerPort: 9000
+        env:
+        - name: BACKEND_DISCOVERY_SERVICE
+          value: backend-service

project/Build.scala

@@ -15,7 +15,8 @@ object Dependencies {
   private val akkaHttpV = "2.4.11"
 
   val akka = Seq(
-    "com.typesafe.akka" %% "akka-actor"             % akkaV
+    "com.typesafe.akka" %% "akka-actor"             % akkaV,
+    "com.typesafe.akka" %% "akka-remote"            % akkaV
   )
   val akkaHttp = Seq(
     "com.typesafe.akka" %% "akka-http-experimental" % akkaHttpV,

project/Docker.scala

@@ -0,0 +1,32 @@
+object Docker {
+
+  val akkaTcpPort = 2551
+
+  val bashExports = """
+export REMOTE_INTERFACE=$(hostname --ip-address)
+
+# Configure the default remote port unless passed
+if [ -z "$REMOTE_PORT" ]; then
+  export REMOTE_PORT=""" + akkaTcpPort + """
+fi
+"""
+
+  val frontendBashExports = """
+# Configure backend nodes from service if passed
+# otherwise keep them as they are
+if [ ! -z "$BACKEND_DISCOVERY_SERVICE" ]; then
+  BACKEND_NODES=$(
+    host -t A $BACKEND_DISCOVERY_SERVICE | \
+    grep 'has address' |
+    cut -d ' ' -f 4 |
+    xargs |
+    sed -e 's/ /,/g'
+  )
+  if [ ! -z "$BACKEND_NODES" ]; then
+    export BACKEND_NODES
+  fi
+fi
+
+""" + bashExports
+
+}