tests/src/test/scala/actionContainers/ActionContainer.scala - openwhisk - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package actionContainers

 import java.io.ByteArrayOutputStream
 import java.io.File
 import java.io.PrintWriter

 import scala.util.Try
 import scala.concurrent.Await
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import scala.concurrent.blocking
 import scala.concurrent.duration.Duration
 import scala.concurrent.duration.DurationInt
 import scala.sys.process.ProcessLogger
 import scala.sys.process.stringToProcess
 import scala.util.Random
 import scala.util.{Failure, Success}
 import org.apache.commons.lang3.StringUtils
 import org.scalatest.{FlatSpec, Matchers}
 import akka.actor.ActorSystem

 import scala.concurrent.ExecutionContext
 import spray.json._
 import common.StreamLogging
 import org.apache.openwhisk.common.Logging
 import org.apache.openwhisk.common.TransactionId
 import org.apache.openwhisk.core.entity.Exec
 import common.WhiskProperties
 import org.apache.openwhisk.core.containerpool.Container

 /**
  * For testing convenience, this interface abstracts away the REST calls to a
  * container as blocking method calls of this interface.
  */
 trait ActionContainer {
   def init(value: JsValue): (Int, Option[JsObject])
   def run(value: JsValue): (Int, Option[JsObject])
   def runMultiple(values: Seq[JsValue])(implicit ec: ExecutionContext): Seq[(Int, Option[JsObject])]
 }

 trait ActionProxyContainerTestUtils extends FlatSpec with Matchers with StreamLogging {
   import ActionContainer.{filterSentinel, sentinel}

   def initPayload(code: String, main: String = "main", env: Option[Map[String, JsString]] = None): JsObject =
     JsObject(
       "value" -> JsObject(
         "code" -> { if (code != null) JsString(code) else JsNull },
         "main" -> JsString(main),
         "binary" -> JsBoolean(Exec.isBinaryCode(code)),
         "env" -> env.map(JsObject(_)).getOrElse(JsNull)))

   def runPayload(args: JsValue, other: Option[JsObject] = None): JsObject =
     JsObject(Map("value" -> args) ++ (other map { _.fields } getOrElse Map.empty))

   def checkStreams(out: String,
                    err: String,
                    additionalCheck: (String, String) => Unit,
                    sentinelCount: Int = 1,
                    concurrent: Boolean = false): Unit = {
     withClue("expected number of stdout sentinels") {
       sentinelCount shouldBe StringUtils.countMatches(out, sentinel)
     }
     //sentinels should be all together
     if (concurrent) {
       withClue("expected grouping of stdout sentinels") {
         out should include((1 to sentinelCount).map(_ => sentinel + "\n").mkString)
       }
     }
     withClue("expected number of stderr sentinels") {
       sentinelCount shouldBe StringUtils.countMatches(err, sentinel)
     }
     //sentinels should be all together
     if (concurrent) {
       withClue("expected grouping of stderr sentinels") {
         err should include((1 to sentinelCount).map(_ => sentinel + "\n").mkString)
       }
     }

     val (o, e) = (filterSentinel(out), filterSentinel(err))
     o should not include sentinel
     e should not include sentinel
     additionalCheck(o, e)
   }
 }

 object ActionContainer {
   private lazy val dockerBin: String = {
     List("/usr/bin/docker", "/usr/local/bin/docker").find { bin =>
       new File(bin).isFile
     }.get // This fails if the docker binary couldn't be located.
   }

   lazy val dockerCmd: String = {
     /*
      * The docker host is set to a provided property 'docker.host' if it's
      * available; otherwise we check with WhiskProperties to see whether we are
      * running on a docker-machine.
      *
      * IMPLICATION:  The test must EITHER have the 'docker.host' system
      * property set OR the 'OPENWHISK_HOME' environment variable set and a
      * valid 'whisk.properties' file generated.  The 'docker.host' system
      * property takes precedence.
      *
      * WARNING:  Adding a non-docker-machine environment that contains 'mac'
      * (i.e. 'environments/local-mac') will likely break things.
      *
      * The plan is to move builds to using 'gradle-docker-plugin', which know
      * its docker socket and to have it pass the docker socket implicitly using
      * 'systemProperty "docker.host", docker.url'.  Eventually, we will also
      * need to handle TLS certificates here.  Again, 'gradle-docker-plugin'
      * knows where they are; we will just add system properties to get the
      * information onto the docker command line.
      */
     val dockerCmdString = dockerBin +
       sys.props
         .get("docker.host")
         .orElse(sys.env.get("DOCKER_HOST"))
         .orElse {
           Try { // whisk.properties file may not exist
             // Check if we are running on docker-machine env.
             Option(WhiskProperties.getProperty("environment.type"))
               .filter(_.toLowerCase.contains("docker-machine"))
               .map {
                 case _ => s"tcp://${WhiskProperties.getMainDockerEndpoint}"
               }
           }.toOption.flatten
         }
         .map(" --host " + _)
         .getOrElse("")

     // Test here that this actually works, otherwise throw a somewhat understandable error message
     proc(s"$dockerCmdString info").onComplete {
       case Success((v, _, _)) if v != 0 =>
         throw new RuntimeException(s"""
               |Unable to connect to docker host using $dockerCmdString as command string.
               |The docker host is determined using the Java property 'docker.host' or
               |the environment variable 'DOCKER_HOST'. Please verify that one or the
               |other is set for your build/test process.""".stripMargin)
       case Success((v, _, _)) if v == 0 => // Do nothing
       case Failure(t)                   => throw t
     }

     dockerCmdString
   }

   private def docker(command: String): String = s"$dockerCmd $command"

   // Runs a process asynchronously. Returns a future with (exitCode,stdout,stderr)
   private def proc(cmd: String): Future[(Int, String, String)] = Future {
     blocking {
       val out = new ByteArrayOutputStream
       val err = new ByteArrayOutputStream
       val outW = new PrintWriter(out)
       val errW = new PrintWriter(err)
       val v = cmd ! ProcessLogger(o => outW.println(o), e => errW.println(e))
       outW.close()
       errW.close()
       (v, out.toString, err.toString)
     }
   }

   // Tying it all together, we have a method that runs docker, waits for
   // completion for some time then returns the exit code, the output stream
   // and the error stream.
   private def awaitDocker(cmd: String, t: Duration): (Int, String, String) = {
     Await.result(proc(docker(cmd)), t)
   }

   // Filters out the sentinel markers inserted by the container (see relevant private code in Invoker)
   val sentinel = Container.ACTIVATION_LOG_SENTINEL
   def filterSentinel(str: String): String = str.replaceAll(sentinel, "").trim

   def withContainer(imageName: String, environment: Map[String, String] = Map.empty)(
     code: ActionContainer => Unit)(implicit actorSystem: ActorSystem, logging: Logging): (String, String) = {
     val rand = { val r = Random.nextInt; if (r < 0) -r else r }
     val name = imageName.toLowerCase.replaceAll("""[^a-z]""", "") + rand
     val envArgs = environment.toSeq
       .map {
         case (k, v) => s"-e $k=$v"
       }
       .mkString(" ")

     // We create the container... and find out its IP address...
     def createContainer(portFwd: Option[Int] = None): Unit = {
       val runOut = awaitDocker(
         s"run ${portFwd.map(p => s"-p $p:8080").getOrElse("")} --name $name $envArgs -d $imageName",
         60.seconds)
       assert(runOut._1 == 0, "'docker run' did not exit with 0: " + runOut)
     }

     // ...find out its IP address...
     val (ip, port) =
       if (System.getProperty("os.name").toLowerCase().contains("mac") && !sys.env
             .get("DOCKER_HOST")
             .exists(_.trim.nonEmpty)) {
         // on MacOSX, where docker for mac does not permit communicating with container directly
         val p = 8988 // port must be available or docker run will fail
         createContainer(Some(p))
         Thread.sleep(1500) // let container/server come up cleanly
         ("localhost", p)
       } else {
         // not "mac" i.e., docker-for-mac, use direct container IP directly (this is OK for Ubuntu, and docker-machine)
         createContainer()
         val ipOut = awaitDocker(s"""inspect --format '{{.NetworkSettings.IPAddress}}' $name""", 10.seconds)
         assert(ipOut._1 == 0, "'docker inspect did not exit with 0")
         (ipOut._2.replaceAll("""[^0-9.]""", ""), 8080)
       }

     // ...we create an instance of the mock container interface...
     val mock = new ActionContainer {
       def init(value: JsValue): (Int, Option[JsObject]) = syncPost(ip, port, "/init", value)
       def run(value: JsValue): (Int, Option[JsObject]) = syncPost(ip, port, "/run", value)
       def runMultiple(values: Seq[JsValue])(implicit ec: ExecutionContext): Seq[(Int, Option[JsObject])] =
         concurrentSyncPost(ip, port, "/run", values)
     }

     try {
       // ...and finally run the code with it.
       code(mock)
       // I'm told this is good for the logs.
       Thread.sleep(100)
       val (_, out, err) = awaitDocker(s"logs $name", 10.seconds)
       (out, err)
     } finally {
       awaitDocker(s"kill $name", 10.seconds)
       awaitDocker(s"rm $name", 10.seconds)
     }
   }

   private def syncPost(host: String, port: Int, endPoint: String, content: JsValue)(
     implicit logging: Logging,
     as: ActorSystem): (Int, Option[JsObject]) = {

     implicit val transid = TransactionId.testing

     org.apache.openwhisk.core.containerpool.AkkaContainerClient.post(host, port, endPoint, content, 30.seconds)
   }
   private def concurrentSyncPost(host: String, port: Int, endPoint: String, contents: Seq[JsValue])(
     implicit logging: Logging,
     as: ActorSystem): Seq[(Int, Option[JsObject])] = {

     implicit val transid = TransactionId.testing

     org.apache.openwhisk.core.containerpool.AkkaContainerClient
       .concurrentPost(host, port, endPoint, contents, 30.seconds)
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package actionContainers

	import java.io.ByteArrayOutputStream
	import java.io.File
	import java.io.PrintWriter

	import scala.util.Try
	import scala.concurrent.Await
	import scala.concurrent.ExecutionContext.Implicits.global
	import scala.concurrent.Future
	import scala.concurrent.blocking
	import scala.concurrent.duration.Duration
	import scala.concurrent.duration.DurationInt
	import scala.sys.process.ProcessLogger
	import scala.sys.process.stringToProcess
	import scala.util.Random
	import scala.util.{Failure, Success}
	import org.apache.commons.lang3.StringUtils
	import org.scalatest.{FlatSpec, Matchers}
	import akka.actor.ActorSystem

	import scala.concurrent.ExecutionContext
	import spray.json._
	import common.StreamLogging
	import org.apache.openwhisk.common.Logging
	import org.apache.openwhisk.common.TransactionId
	import org.apache.openwhisk.core.entity.Exec
	import common.WhiskProperties
	import org.apache.openwhisk.core.containerpool.Container

	/**
	* For testing convenience, this interface abstracts away the REST calls to a
	* container as blocking method calls of this interface.
	*/
	trait ActionContainer {
	def init(value: JsValue): (Int, Option[JsObject])
	def run(value: JsValue): (Int, Option[JsObject])
	def runMultiple(values: Seq[JsValue])(implicit ec: ExecutionContext): Seq[(Int, Option[JsObject])]
	}

	trait ActionProxyContainerTestUtils extends FlatSpec with Matchers with StreamLogging {
	import ActionContainer.{filterSentinel, sentinel}

	def initPayload(code: String, main: String = "main", env: Option[Map[String, JsString]] = None): JsObject =
	JsObject(
	"value" -> JsObject(
	"code" -> { if (code != null) JsString(code) else JsNull },
	"main" -> JsString(main),
	"binary" -> JsBoolean(Exec.isBinaryCode(code)),
	"env" -> env.map(JsObject(_)).getOrElse(JsNull)))

	def runPayload(args: JsValue, other: Option[JsObject] = None): JsObject =
	JsObject(Map("value" -> args) ++ (other map { _.fields } getOrElse Map.empty))

	def checkStreams(out: String,
	err: String,
	additionalCheck: (String, String) => Unit,
	sentinelCount: Int = 1,
	concurrent: Boolean = false): Unit = {
	withClue("expected number of stdout sentinels") {
	sentinelCount shouldBe StringUtils.countMatches(out, sentinel)
	}
	//sentinels should be all together
	if (concurrent) {
	withClue("expected grouping of stdout sentinels") {
	out should include((1 to sentinelCount).map(_ => sentinel + "\n").mkString)
	}
	}
	withClue("expected number of stderr sentinels") {
	sentinelCount shouldBe StringUtils.countMatches(err, sentinel)
	}
	//sentinels should be all together
	if (concurrent) {
	withClue("expected grouping of stderr sentinels") {
	err should include((1 to sentinelCount).map(_ => sentinel + "\n").mkString)
	}
	}

	val (o, e) = (filterSentinel(out), filterSentinel(err))
	o should not include sentinel
	e should not include sentinel
	additionalCheck(o, e)
	}
	}

	object ActionContainer {
	private lazy val dockerBin: String = {
	List("/usr/bin/docker", "/usr/local/bin/docker").find { bin =>
	new File(bin).isFile
	}.get // This fails if the docker binary couldn't be located.
	}

	lazy val dockerCmd: String = {
	/*
	* The docker host is set to a provided property 'docker.host' if it's
	* available; otherwise we check with WhiskProperties to see whether we are
	* running on a docker-machine.
	*
	* IMPLICATION: The test must EITHER have the 'docker.host' system
	* property set OR the 'OPENWHISK_HOME' environment variable set and a
	* valid 'whisk.properties' file generated. The 'docker.host' system
	* property takes precedence.
	*
	* WARNING: Adding a non-docker-machine environment that contains 'mac'
	* (i.e. 'environments/local-mac') will likely break things.
	*
	* The plan is to move builds to using 'gradle-docker-plugin', which know
	* its docker socket and to have it pass the docker socket implicitly using
	* 'systemProperty "docker.host", docker.url'. Eventually, we will also
	* need to handle TLS certificates here. Again, 'gradle-docker-plugin'
	* knows where they are; we will just add system properties to get the
	* information onto the docker command line.
	*/
	val dockerCmdString = dockerBin +
	sys.props
	.get("docker.host")
	.orElse(sys.env.get("DOCKER_HOST"))
	.orElse {
	Try { // whisk.properties file may not exist
	// Check if we are running on docker-machine env.
	Option(WhiskProperties.getProperty("environment.type"))
	.filter(_.toLowerCase.contains("docker-machine"))
	.map {
	case _ => s"tcp://${WhiskProperties.getMainDockerEndpoint}"
	}
	}.toOption.flatten
	}
	.map(" --host " + _)
	.getOrElse("")

	// Test here that this actually works, otherwise throw a somewhat understandable error message
	proc(s"$dockerCmdString info").onComplete {
	case Success((v, _, _)) if v != 0 =>
	throw new RuntimeException(s"""
	\|Unable to connect to docker host using $dockerCmdString as command string.
	\|The docker host is determined using the Java property 'docker.host' or
	\|the environment variable 'DOCKER_HOST'. Please verify that one or the
	\|other is set for your build/test process.""".stripMargin)
	case Success((v, _, _)) if v == 0 => // Do nothing
	case Failure(t) => throw t
	}

	dockerCmdString
	}

	private def docker(command: String): String = s"$dockerCmd $command"

	// Runs a process asynchronously. Returns a future with (exitCode,stdout,stderr)
	private def proc(cmd: String): Future[(Int, String, String)] = Future {
	blocking {
	val out = new ByteArrayOutputStream
	val err = new ByteArrayOutputStream
	val outW = new PrintWriter(out)
	val errW = new PrintWriter(err)
	val v = cmd ! ProcessLogger(o => outW.println(o), e => errW.println(e))
	outW.close()
	errW.close()
	(v, out.toString, err.toString)
	}
	}

	// Tying it all together, we have a method that runs docker, waits for
	// completion for some time then returns the exit code, the output stream
	// and the error stream.
	private def awaitDocker(cmd: String, t: Duration): (Int, String, String) = {
	Await.result(proc(docker(cmd)), t)
	}

	// Filters out the sentinel markers inserted by the container (see relevant private code in Invoker)
	val sentinel = Container.ACTIVATION_LOG_SENTINEL
	def filterSentinel(str: String): String = str.replaceAll(sentinel, "").trim

	def withContainer(imageName: String, environment: Map[String, String] = Map.empty)(
	code: ActionContainer => Unit)(implicit actorSystem: ActorSystem, logging: Logging): (String, String) = {
	val rand = { val r = Random.nextInt; if (r < 0) -r else r }
	val name = imageName.toLowerCase.replaceAll("""[^a-z]""", "") + rand
	val envArgs = environment.toSeq
	.map {
	case (k, v) => s"-e $k=$v"
	}
	.mkString(" ")

	// We create the container... and find out its IP address...
	def createContainer(portFwd: Option[Int] = None): Unit = {
	val runOut = awaitDocker(
	s"run ${portFwd.map(p => s"-p $p:8080").getOrElse("")} --name $name $envArgs -d $imageName",
	60.seconds)
	assert(runOut._1 == 0, "'docker run' did not exit with 0: " + runOut)
	}

	// ...find out its IP address...
	val (ip, port) =
	if (System.getProperty("os.name").toLowerCase().contains("mac") && !sys.env
	.get("DOCKER_HOST")
	.exists(_.trim.nonEmpty)) {
	// on MacOSX, where docker for mac does not permit communicating with container directly
	val p = 8988 // port must be available or docker run will fail
	createContainer(Some(p))
	Thread.sleep(1500) // let container/server come up cleanly
	("localhost", p)
	} else {
	// not "mac" i.e., docker-for-mac, use direct container IP directly (this is OK for Ubuntu, and docker-machine)
	createContainer()
	val ipOut = awaitDocker(s"""inspect --format '{{.NetworkSettings.IPAddress}}' $name""", 10.seconds)
	assert(ipOut._1 == 0, "'docker inspect did not exit with 0")
	(ipOut._2.replaceAll("""[^0-9.]""", ""), 8080)
	}

	// ...we create an instance of the mock container interface...
	val mock = new ActionContainer {
	def init(value: JsValue): (Int, Option[JsObject]) = syncPost(ip, port, "/init", value)
	def run(value: JsValue): (Int, Option[JsObject]) = syncPost(ip, port, "/run", value)
	def runMultiple(values: Seq[JsValue])(implicit ec: ExecutionContext): Seq[(Int, Option[JsObject])] =
	concurrentSyncPost(ip, port, "/run", values)
	}

	try {
	// ...and finally run the code with it.
	code(mock)
	// I'm told this is good for the logs.
	Thread.sleep(100)
	val (_, out, err) = awaitDocker(s"logs $name", 10.seconds)
	(out, err)
	} finally {
	awaitDocker(s"kill $name", 10.seconds)
	awaitDocker(s"rm $name", 10.seconds)
	}
	}

	private def syncPost(host: String, port: Int, endPoint: String, content: JsValue)(
	implicit logging: Logging,
	as: ActorSystem): (Int, Option[JsObject]) = {

	implicit val transid = TransactionId.testing

	org.apache.openwhisk.core.containerpool.AkkaContainerClient.post(host, port, endPoint, content, 30.seconds)
	}
	private def concurrentSyncPost(host: String, port: Int, endPoint: String, contents: Seq[JsValue])(
	implicit logging: Logging,
	as: ActorSystem): Seq[(Int, Option[JsObject])] = {

	implicit val transid = TransactionId.testing

	org.apache.openwhisk.core.containerpool.AkkaContainerClient
	.concurrentPost(host, port, endPoint, contents, 30.seconds)
	}

	}