Merge pull request #54 from delphi-hub/feature/databaseIntegration

The registry now uses an SQL database for storing instance data

Author: johannesduesing

build.sbt

@@ -28,3 +28,12 @@ lazy val registry = (project in file(".")).
     dockerBaseImage := "openjdk:jre-alpine"
   )
 
+libraryDependencies ++= List(
+  "com.typesafe.slick" %% "slick" % "3.2.3",
+  "com.typesafe.slick" %% "slick-hikaricp" % "3.2.3",
+  "com.typesafe.slick" %% "slick-codegen" % "3.2.3",
+  "mysql" % "mysql-connector-java" % "5.1.34",
+  "org.slf4j" % "slf4j-nop" % "1.6.4"
+)
+
+trapExit := false

src/main/scala/de/upb/cs/swt/delphi/instanceregistry/Configuration.scala

@@ -1,6 +1,7 @@
 package de.upb.cs.swt.delphi.instanceregistry
 
 import akka.util.Timeout
+
 import scala.concurrent.duration.DurationInt
 
 class Configuration( ) {
@@ -32,6 +33,15 @@ class Configuration( ) {
 
   val dockerOperationTimeout: Timeout = Timeout(20 seconds)
 
-}
+  //Database configurations
+  val useInMemoryDB = false
+  val databaseHost = "jdbc:mysql://localhost/"
+  val databaseName = ""
+  val databaseDriver = "com.mysql.jdbc.Driver"
+  val databaseUsername = ""
+  val databasePassword = ""
+
+
+  }
 
 

src/main/scala/de/upb/cs/swt/delphi/instanceregistry/Registry.scala

@@ -4,6 +4,7 @@ import akka.actor.ActorSystem
 import akka.stream.ActorMaterializer
 import de.upb.cs.swt.delphi.instanceregistry.Docker._
 import de.upb.cs.swt.delphi.instanceregistry.connection.Server
+import de.upb.cs.swt.delphi.instanceregistry.daos.{DatabaseInstanceDAO, DynamicInstanceDAO, InstanceDAO}
 
 import scala.concurrent.ExecutionContext
 import scala.language.postfixOps
@@ -15,14 +16,24 @@ object Registry extends AppLogging {
 
 
   val configuration = new Configuration()
-  val requestHandler = new RequestHandler(configuration, DockerConnection.fromEnvironment())
 
+  private val dao : InstanceDAO  = {
+    if (configuration.useInMemoryDB) {
+      new DynamicInstanceDAO(configuration)
+    } else {
+      new DatabaseInstanceDAO(configuration)
+    }
+  }
 
-  def main(args: Array[String]): Unit = {
+  private val requestHandler = new RequestHandler(configuration, dao, DockerConnection.fromEnvironment())
+
+  private val server: Server = new Server(requestHandler)
 
+
+  def main(args: Array[String]): Unit = {
     requestHandler.initialize()
     log.info("Starting server ...")
-    Server.startServer(configuration.bindHost, configuration.bindPort)
+    server.startServer(configuration.bindHost, configuration.bindPort)
     log.info("Shutting down ...")
     requestHandler.shutdown()
     system.terminate()

src/main/scala/de/upb/cs/swt/delphi/instanceregistry/RequestHandler.scala

@@ -10,7 +10,7 @@ import de.upb.cs.swt.delphi.instanceregistry.Docker.{ContainerAlreadyStoppedExce
 import akka.stream.scaladsl.{Keep, Sink, Source}
 import akka.stream.{ActorMaterializer, Materializer, OverflowStrategy}
 import de.upb.cs.swt.delphi.instanceregistry.connection.RestClient
-import de.upb.cs.swt.delphi.instanceregistry.daos.{DynamicInstanceDAO, InstanceDAO}
+import de.upb.cs.swt.delphi.instanceregistry.daos.InstanceDAO
 import de.upb.cs.swt.delphi.instanceregistry.io.swagger.client.model.InstanceEnums.{ComponentType, InstanceState}
 import de.upb.cs.swt.delphi.instanceregistry.io.swagger.client.model.LinkEnums.LinkState
 import de.upb.cs.swt.delphi.instanceregistry.io.swagger.client.model._
@@ -20,15 +20,14 @@ import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.language.postfixOps
 import scala.util.{Failure, Success, Try}
 
-class RequestHandler(configuration: Configuration, connection: DockerConnection) extends AppLogging {
+class RequestHandler(configuration: Configuration, instanceDao: InstanceDAO, connection: DockerConnection) extends AppLogging {
 
 
 
   implicit val system: ActorSystem = Registry.system
   implicit val materializer : Materializer = ActorMaterializer()
   implicit val ec: ExecutionContext = system.dispatcher
 
-  private[instanceregistry] val instanceDao: InstanceDAO = new DynamicInstanceDAO(configuration)
 
   val (eventActor, eventPublisher) = Source.actorRef[RegistryEvent](10, OverflowStrategy.dropNew)
     .toMat(Sink.asPublisher(fanout = true))(Keep.both)
@@ -67,20 +66,18 @@ class RequestHandler(configuration: Configuration, connection: DockerConnection)
     * @return Newly assigned ID if successful
     */
   def handleRegister(instance: Instance): Try[Long] = {
-    val newID = generateNextId()
 
-    log.info(s"Assigned new id $newID to registering instance with name ${instance.name}.")
-
-    val newInstance = Instance(id = Some(newID), name = instance.name, host = instance.host,
+    val noIdInstance = Instance(id = None, name = instance.name, host = instance.host,
       portNumber = instance.portNumber, componentType = instance.componentType,
       dockerId = None, instanceState = InstanceState.Running, labels = instance.labels,
       linksTo = List.empty[InstanceLink], linksFrom = List.empty[InstanceLink])
 
-    instanceDao.addInstance(newInstance) match {
-      case Success(_) =>
-        fireNumbersChangedEvent(newInstance.componentType)
-        fireInstanceAddedEvent(newInstance)
-        Success(newID)
+    instanceDao.addInstance(noIdInstance) match {
+      case Success(id) =>
+        fireNumbersChangedEvent(instanceDao.getInstance(id).get.componentType)
+        fireInstanceAddedEvent(instanceDao.getInstance(id).get)
+        log.info(s"Assigned new id $id to registering instance with name ${instance.name}.")
+        Success(id)
       case Failure(x) => Failure(x)
     }
   }
@@ -243,48 +240,68 @@ class RequestHandler(configuration: Configuration, connection: DockerConnection)
   }
 
   def handleDeploy(componentType: ComponentType, name: Option[String]): Try[Long] = {
-    val newId = generateNextId()
-
     log.info(s"Deploying container of type $componentType")
+    val instance = Instance(None,
+      "",
+      -1L,
+      name.getOrElse(s"Generic $componentType"),
+      componentType,
+      None,
+      InstanceState.Deploying,
+      List.empty[String],
+      List.empty[InstanceLink],
+      List.empty[InstanceLink]
+    )
+
+    instanceDao.addInstance(instance) match {
+      case Success(id) =>
+        implicit val timeout: Timeout = configuration.dockerOperationTimeout
 
-    implicit val timeout: Timeout = configuration.dockerOperationTimeout
-
-    val future: Future[Any] = dockerActor ? create(componentType, newId)
-    val deployResult = Await.result(future, timeout.duration).asInstanceOf[Try[(String, String, Int)]]
-
-    deployResult match {
-      case Failure(ex) =>
-        log.error(s"Failed to deploy container, docker host not reachable.")
-        fireDockerOperationErrorEvent(None, s"Deploy failed with message: ${ex.getMessage}")
-        Failure(new RuntimeException(s"Failed to deploy container, docker host not reachable (${ex.getMessage})."))
-      case Success((dockerId, host, port)) =>
-        val normalizedHost = host.substring(1,host.length - 1)
-        log.info(s"Deployed new container with id $dockerId, host $normalizedHost and port $port.")
-
-        val newInstance = Instance(Some(newId),
-          normalizedHost,
-          port,
-          name.getOrElse(s"Generic $componentType"),
-          componentType,
-          Some(dockerId),
-          InstanceState.Deploying,
-          List.empty[String],
-          List.empty[InstanceLink],
-          List.empty[InstanceLink]
-        )
-        log.info(s"Registering instance $newInstance....")
-
-        instanceDao.addInstance(newInstance) match {
-          case Success(_) =>
-            log.info("Successfully registered.")
-            fireInstanceAddedEvent(newInstance)
-            fireNumbersChangedEvent(newInstance.componentType)
-            Success(newId)
-          case Failure(x) =>
-            log.info(s"Failed to register. Exception: $x")
-            Failure(x)
+        val future: Future[Any] = dockerActor ? create(componentType, id)
+        val deployResult = Await.result(future, timeout.duration).asInstanceOf[Try[(String, String, Int)]]
+
+        deployResult match {
+          case Failure(ex) =>
+            log.error(s"Failed to deploy container, docker host not reachable.")
+            instanceDao.removeInstance(id)
+            fireDockerOperationErrorEvent(None, s"Deploy failed with message: ${ex.getMessage}")
+            Failure(new RuntimeException(s"Failed to deploy container, docker host not reachable (${ex.getMessage})."))
+          case Success((dockerId, host, port)) =>
+            val normalizedHost = host.substring(1, host.length - 1)
+            log.info(s"Deployed new container with id $dockerId, host $normalizedHost and port $port.")
+
+            val newInstance = Instance(Some(id),
+              normalizedHost,
+              port,
+              name.getOrElse(s"Generic $componentType"),
+              componentType,
+              Some(dockerId),
+              InstanceState.Deploying,
+              List.empty[String],
+              List.empty[InstanceLink],
+              List.empty[InstanceLink]
+            )
+
+            instanceDao.updateInstance(newInstance) match {
+              case Success(_) =>
+                log.info("Successfully registered.")
+                fireInstanceAddedEvent(newInstance)
+                fireNumbersChangedEvent(newInstance.componentType)
+                Success(id)
+              case Failure(x) =>
+                log.info(s"Failed to register. Exception: $x")
+                Failure(x)
+            }
         }
+      case Failure(ex) =>
+        Failure(ex)
     }
+
+
+
+
+
+
   }
 
   /** *
@@ -962,14 +979,6 @@ class RequestHandler(configuration: Configuration, connection: DockerConnection)
 
   }
 
-  private def generateNextId(): Long = {
-    if (instanceDao.allInstances().isEmpty) {
-      0L
-    } else {
-      (instanceDao.allInstances().map(i => i.id.getOrElse(0L)) max) + 1L
-    }
-  }
-
   private def assignmentAllowed(instanceType: ComponentType) : Boolean = {
     instanceType == ComponentType.Crawler || instanceType == ComponentType.WebApi || instanceType == ComponentType.WebApp
   }

src/main/scala/de/upb/cs/swt/delphi/instanceregistry/connection/Server.scala

@@ -20,7 +20,7 @@ import scala.util.{Failure, Success}
 /**
   * Web server configuration for Instance Registry API.
   */
-object Server extends HttpApp
+class Server (handler: RequestHandler) extends HttpApp
   with InstanceJsonSupport
   with EventJsonSupport
   with InstanceLinkJsonSupport
@@ -30,8 +30,6 @@ object Server extends HttpApp
   implicit val materializer : ActorMaterializer = ActorMaterializer()
   implicit val ec : ExecutionContext = system.dispatcher
 
-  private val handler : RequestHandler = Registry.requestHandler
-
   //Routes that map http endpoints to methods in this object
   override def routes : server.Route =
       /****************BASIC OPERATIONS****************/