Merge pull request #82 from delphi-hub/feature/dockerLogs

24santoshr · web-flow · commit 91318fcbf713 · 2019-01-17T14:16:12.000+01:00
Implemented endpoints to retrieve docker logs
diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf
@@ -1 +1,3 @@
+akka.http.client.idle-timeout = infinite
+akka.http.host-connection-pool.idle-timeout = infinite
 akka.http.server.websocket.periodic-keep-alive-max-idle = 10 seconds
diff --git a/src/main/scala/de/upb/cs/swt/delphi/instanceregistry/Docker/ContainerCommands.scala b/src/main/scala/de/upb/cs/swt/delphi/instanceregistry/Docker/ContainerCommands.scala
@@ -1,19 +1,28 @@
 package de.upb.cs.swt.delphi.instanceregistry.Docker
 
 
-import akka.NotUsed
-import akka.actor.ActorSystem
+import java.nio.ByteOrder
+
+import akka.{Done, NotUsed}
+import akka.actor.{ActorSystem, PoisonPill}
 import akka.http.scaladsl.client.RequestBuilding._
 import akka.http.scaladsl.model.MediaTypes.`application/json`
 import akka.http.scaladsl.model.Uri.{Path, Query}
 import akka.http.scaladsl.model.{HttpEntity, HttpResponse, StatusCodes}
 import akka.http.scaladsl.unmarshalling.Unmarshal
-import akka.stream.scaladsl.{Flow, Source}
+import akka.stream.scaladsl.{Flow, Framing, Keep, Sink, Source}
 import de.upb.cs.swt.delphi.instanceregistry.{AppLogging, Registry}
 import spray.json._
 import PostDataFormatting.commandJsonRequest
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.ws.{Message, TextMessage, WebSocketRequest}
+import akka.stream.OverflowStrategy
+import akka.util.ByteString
+import org.reactivestreams.Publisher
 
-import scala.concurrent.{ExecutionContext, Future}
+import scala.collection.mutable
+import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.util.{Failure, Success, Try}
 
 
 class ContainerCommands(connection: DockerConnection) extends JsonSupport with Commands with AppLogging {
@@ -184,25 +193,70 @@ class ContainerCommands(connection: DockerConnection) extends JsonSupport with C
     }
   }
 
-  def logs(
-            containerId: String
-          )(implicit ec: ExecutionContext): Source[String, NotUsed] = {
-    val query = Query("stdout" -> "true" )
+  def retrieveLogs(
+            containerId: String,
+            stdErrSelected: Boolean
+          )(implicit ec: ExecutionContext): Future[String] = {
+
+    val query = Query("stdout" -> (!stdErrSelected).toString, "stderr" -> stdErrSelected.toString, "follow" -> "false", "tail" -> "all", "timestamps" -> "true")
     val request = Get(buildUri(containersPath / containerId.substring(0,11) / "logs", query))
 
-    val flow =
-      Flow[HttpResponse].map {
-        case HttpResponse(StatusCodes.OK, _, HttpEntity.Chunked(_, chunks), _) =>
-          chunks.map(_.data().utf8String)
-        case HttpResponse(StatusCodes.NotFound, _, HttpEntity.Strict(_, data), _) =>
-          log.warning(s"DOCKER LOGS FAILED: ${data.utf8String}")
+    connection.sendRequest(request).flatMap {response =>
+      response.status match {
+        case StatusCodes.OK =>
+          Unmarshal(response.entity).to[String]
+        case StatusCodes.UpgradeRequired =>
+          log.warning(s"Unexpected upgrade response while reading logs for container $containerId")
+          log.warning(s"Got $response")
+          unknownResponseFuture(response)
+        case StatusCodes.NotFound =>
           throw ContainerNotFoundException(containerId)
-        case response =>
-          unknownResponse(response)
-      }.flatMapConcat(identity)
+        case _ =>
+          unknownResponseFuture(response)
+      }
+    }
+  }
+
+  def streamLogs(containerId: String, stdErrSelected: Boolean) (implicit ec: ExecutionContext) : Try[Publisher[Message]] = {
+
+    // Select stdout / stderr in query params
+    val queryParams = Query("stdout" -> (!stdErrSelected).toString, "stderr" -> stdErrSelected.toString, "follow" -> "true", "tail" -> "all", "timestamps" -> "false")
+
+    // Create actor-publisher pair, publisher will be returned
+    val (streamActor, streamPublisher) = Source.actorRef[Message](bufferSize = 10, OverflowStrategy.dropNew)
+      .toMat(Sink.asPublisher(fanout = true))(Keep.both)
+      .run()
+
+    // Delimiter flow splits incoming traffic into lines based on dockers multiplex-protocol
+    // Docker prepends an 8-byte header, where the last 4 byte encode line length in big endian
+    // See https://docs.docker.com/engine/api/v1.30/#operation/ContainerAttach
+    val delimiter: Flow[ByteString, ByteString, NotUsed] = Framing.lengthField(4, 4, 100000, ByteOrder.BIG_ENDIAN)
+
+    // Flow that removes header bytes from payload
+    val removeHeaderFlow: Flow[ByteString, ByteString, NotUsed] = Flow.fromFunction(in => in.slice(8, in.size))
+
+    // Build request
+    val request = Get(buildUri(containersPath / containerId.substring(0,11) / "logs", queryParams))
+
+    // Execute request
+    val res = connection.sendRequest(request).flatMap { res =>
+      // Extract payload ByteString from data stream using above flows. Map to string.
+      val logLines = res.entity.dataBytes.via(delimiter).via(removeHeaderFlow).map(_.utf8String)
+      logLines.runForeach { line =>
+        // Send each log line to the stream actor, which will publish them
+        log.debug(s"Streaming log message $line")
+        streamActor ! TextMessage(line)
+      }
+    }
+
+    // Kill actor on completion
+    res.onComplete{ _ =>
+      log.info("Log stream finished successfully.")
+      streamActor ! PoisonPill
+    }
 
-    Source.fromFuture(connection.sendRequest(request))
-      .via(flow)
+    // Return publish so server can subscribe to it
+    Success(streamPublisher)
   }
 
   def commandCreate(
diff --git a/src/main/scala/de/upb/cs/swt/delphi/instanceregistry/Docker/DockerActor.scala b/src/main/scala/de/upb/cs/swt/delphi/instanceregistry/Docker/DockerActor.scala
@@ -1,10 +1,12 @@
 package de.upb.cs.swt.delphi.instanceregistry.Docker
 
 import akka.actor.{Actor, ActorLogging, ActorSystem, Props, Status}
+import akka.http.scaladsl.model.ws.Message
 import akka.stream.ActorMaterializer
 import de.upb.cs.swt.delphi.instanceregistry.Docker.DockerActor._
 import de.upb.cs.swt.delphi.instanceregistry.Registry
 import de.upb.cs.swt.delphi.instanceregistry.io.swagger.client.model.InstanceEnums.ComponentType
+import org.reactivestreams.Publisher
 
 import scala.concurrent.duration._
 import scala.concurrent.{Await, ExecutionContext, Future}
@@ -120,9 +122,23 @@ class DockerActor(connection: DockerConnection) extends Actor with ActorLogging
           }
       }
 
-    case logs(containerId: String) =>
-      log.info(s"Fetching Container logs")
-      sender ! container.logs(containerId)
+    case logs(containerId: String, stdErrSelected: Boolean, stream: Boolean) =>
+
+      log.info(s"Fetching Container logs: stdErrSelected -> $stdErrSelected, stream -> $stream")
+
+      if(!stream){
+        val logResult = Try(Await.result(container.retrieveLogs(containerId, stdErrSelected), Duration.Inf))
+        logResult match {
+          case Failure(ex) =>
+            log.warning(s"Failed to get container logs with ${ex.getMessage}")
+            sender ! Failure(ex)
+          case Success(logContent) =>
+            sender ! Success(logContent)
+        }
+      } else {
+        sender ! container.streamLogs(containerId, stdErrSelected)
+      }
+
 
     case x => log.warning("Received unknown message: [{}] ", x)
   }
@@ -146,7 +162,7 @@ object DockerActor {
 
   case class restart(containerId: String)
 
-  case class logs(containerId: String)
+  case class logs(containerId: String, stdErrSelected: Boolean, stream: Boolean)
 
   case class runCommand(
                          containerId: String,
diff --git a/src/main/scala/de/upb/cs/swt/delphi/instanceregistry/RequestHandler.scala b/src/main/scala/de/upb/cs/swt/delphi/instanceregistry/RequestHandler.scala
@@ -1,8 +1,8 @@
 package de.upb.cs.swt.delphi.instanceregistry
 
-import akka.NotUsed
 import akka.actor._
 import akka.http.scaladsl.model.StatusCodes
+import akka.http.scaladsl.model.ws.Message
 import akka.pattern.{AskTimeoutException, ask}
 import akka.util.Timeout
 import de.upb.cs.swt.delphi.instanceregistry.Docker.DockerActor._
@@ -14,6 +14,7 @@ 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._
+import org.reactivestreams.Publisher
 
 import scala.concurrent.duration._
 import scala.concurrent.{Await, ExecutionContext, Future}
@@ -28,7 +29,6 @@ class RequestHandler(configuration: Configuration, instanceDao: InstanceDAO, con
   implicit val materializer : Materializer = ActorMaterializer()
   implicit val ec: ExecutionContext = system.dispatcher
 
-
   val (eventActor, eventPublisher) = Source.actorRef[RegistryEvent](10, OverflowStrategy.dropNew)
     .toMat(Sink.asPublisher(fanout = true))(Keep.both)
     .run()
@@ -674,23 +674,59 @@ class RequestHandler(configuration: Configuration, instanceDao: InstanceDAO, con
     * @param id Id of the instance
     * @return Tuple of OperationResult and Option[Source[...] ]
     */
-  def handleGetLogs(id: Long) : (OperationResult.Value, Option[Source[String, NotUsed]]) = {
+  def handleGetLogs(id: Long, stdErrSelected: Boolean) : (OperationResult.Value, Option[String]) = {
     if(!instanceDao.hasInstance(id)){
       (OperationResult.IdUnknown, None)
     } else if(!isInstanceDockerContainer(id)){
       (OperationResult.NoDockerContainer, None)
     } else {
       val instance = instanceDao.getInstance(id).get
 
-      val f : Future[(OperationResult.Value, Option[Source[String, NotUsed]])]= (dockerActor ? logs(instance.dockerId.get))(configuration.dockerOperationTimeout).map{
-        source: Any =>
-          (OperationResult.Ok, Option(source.asInstanceOf[Source[String, NotUsed]]))
+      val f : Future[(OperationResult.Value, Option[String])] = (dockerActor ? logs(instance.dockerId.get, stdErrSelected, stream = false))(configuration.dockerOperationTimeout).map{
+        logVal: Any =>
+          val logResult = logVal.asInstanceOf[Try[String]]
+          logResult match {
+            case Success(logContent) =>
+              (OperationResult.Ok, Some(logContent))
+            case Failure(ex) =>
+              log.warning(s"Failed to get logs from actor, exception: ${ex.getMessage}")
+              (OperationResult.InternalError, None)
+          }
+
       }.recover{
         case ex: Exception =>
           fireDockerOperationErrorEvent(Some(instance), errorMessage = s"Failed to get logs with message: ${ex.getMessage}")
           (OperationResult.InternalError, None)
       }
-      Await.result(f, Duration.Inf)
+      Await.result(f, configuration.dockerOperationTimeout.duration)
+    }
+  }
+
+  def handleStreamLogs(id: Long, stdErrSelected: Boolean) : (OperationResult.Value, Option[Publisher[Message]]) = {
+    if(!instanceDao.hasInstance(id)){
+      (OperationResult.IdUnknown, None)
+    } else if(!isInstanceDockerContainer(id)){
+      (OperationResult.NoDockerContainer, None)
+    } else {
+      val instance = instanceDao.getInstance(id).get
+
+      val f : Future[(OperationResult.Value, Option[Publisher[Message]])] = (dockerActor ? logs(instance.dockerId.get, stdErrSelected, stream = true))(configuration.dockerOperationTimeout).map{
+        publisherVal: Any =>
+          val publisherResult = publisherVal.asInstanceOf[Try[Publisher[Message]]]
+          publisherResult match {
+            case Success(publisher) =>
+              (OperationResult.Ok, Some(publisher))
+            case Failure(ex) =>
+              log.warning(s"Failed to stream logs from actor, exception: ${ex.getMessage}")
+              (OperationResult.InternalError, None)
+          }
+
+      }.recover{
+        case ex: Exception =>
+          fireDockerOperationErrorEvent(Some(instance), errorMessage = s"Failed to stream logs with message: ${ex.getMessage}")
+          (OperationResult.InternalError, None)
+      }
+      Await.result(f, configuration.dockerOperationTimeout.duration)
     }
   }
 
diff --git a/src/main/scala/de/upb/cs/swt/delphi/instanceregistry/connection/Server.scala b/src/main/scala/de/upb/cs/swt/delphi/instanceregistry/connection/Server.scala
@@ -67,6 +67,8 @@ class Server (handler: RequestHandler) extends HttpApp
       path("delete") { deleteContainer()} ~
       path("assignInstance") { assignInstance()} ~
       path("command") { runCommandInContainer()} ~
+      path("logs") { retrieveLogs()} ~
+      path("attach") { streamLogs()} ~
       /****************EVENT OPERATIONS****************/
       path("events") { streamEvents()}
 
@@ -675,6 +677,64 @@ class Server (handler: RequestHandler) extends HttpApp
     }
   }
 
+  def retrieveLogs(): server.Route = parameters('Id.as[Long], 'StdErr.as[Boolean].?) { (id, stdErrOption) =>
+    authenticateOAuth2[AccessToken]("Secure Site", AuthProvider.authenticateOAuthRequire(_, userType = UserType.Admin)){ token =>
+      get {
+        log.debug(s"GET /logs?Id=$id has been called")
+
+        val stdErrSelected = stdErrOption.isDefined && stdErrOption.get
+
+        handler.handleGetLogs(id, stdErrSelected) match {
+          case (handler.OperationResult.IdUnknown, _) =>
+            log.warning(s"Cannot get logs, id $id not found.")
+            complete{HttpResponse(StatusCodes.NotFound, entity = s"Cannot get logs, id $id not found.")}
+          case (handler.OperationResult.NoDockerContainer, _) =>
+            log.warning(s"Cannot get logs, id $id is no docker container.")
+            complete{HttpResponse(StatusCodes.BadRequest,entity = s"Cannot get logs, id $id is no docker container.")}
+          case (handler.OperationResult.Ok, Some(logString)) =>
+            complete{logString}
+          case (handler.OperationResult.InternalError, _) =>
+            complete{HttpResponse(StatusCodes.InternalServerError, entity = s"Internal server error")}
+          case _ =>
+            complete{HttpResponse(StatusCodes.InternalServerError, entity = s"Internal server error")}
+        }
+      }
+    }
+  }
+
+  def streamLogs(): server.Route = parameters('Id.as[Long], 'StdErr.as[Boolean].?) { (id, stdErrOption) =>
+
+
+    val stdErrSelected = stdErrOption.isDefined && stdErrOption.get
+
+    handler.handleStreamLogs(id, stdErrSelected) match {
+      case (handler.OperationResult.IdUnknown, _) =>
+        complete{HttpResponse(StatusCodes.NotFound, entity = s"Cannot stream logs, id $id not found.")}
+      case (handler.OperationResult.NoDockerContainer, _) =>
+        complete{HttpResponse(StatusCodes.BadRequest, entity = s"Cannot stream logs, id $id is no docker container.")}
+      case (handler.OperationResult.Ok, Some(publisher)) =>
+        handleWebSocketMessages {
+          Flow[Message]
+            .via(
+              Flow.fromSinkAndSource(Sink.ignore, Source.fromPublisher(publisher))
+            )
+            .watchTermination() {(_, done) =>
+              done.onComplete {
+                case Success(_) =>
+                  log.info("Log stream route completed successfully")
+                case Failure(ex) =>
+                  log.error(s"Log stream route completed with failure : $ex")
+              }
+            }
+        }
+      case (handler.OperationResult.InternalError, _) =>
+        complete{HttpResponse(StatusCodes.InternalServerError, entity = s"Internal server error")}
+      case _ =>
+        complete{HttpResponse(StatusCodes.InternalServerError, entity = s"Internal server error")}
+    }
+
+  }
+
   /**
     * Creates a WebSocketConnection that streams events that are issued by the registry to all connected clients.
     * @return Server route that maps to the WebSocketConnection

Original file line number	Diff line number	Diff line change
`@@ -1 +1,3 @@`
	`1`	`+akka.http.client.idle-timeout = infinite`
	`2`	`+akka.http.host-connection-pool.idle-timeout = infinite`
`1`	`3`	`akka.http.server.websocket.periodic-keep-alive-max-idle = 10 seconds`