improve deployment plan computation perf by ~2000x

Summary:
import mesosphere.marathon.stream.Implicits._ brought in a method that
implicitly converted scala maps to java maps; we were calling the java
`containsKey` Java map method instead of `contains` scala map method,
and this led the map to be converted to a Java version, only to have
the containsKey method invoked, and entire data structure
discarded.

Also adds a flat DependencyPlan computation benchmark for apps and pods.

backport of 8e538f9

Test Plan: benchmark included

Reviewers: jdef, ichernetsky, jenkins

Reviewed By: jdef, ichernetsky, jenkins

Subscribers: marathon-dev

JIRA Issues: MARATHON-7479

Author: Tim Harper

benchmark/src/main/scala/mesosphere/marathon/upgrade/FlatDependencyBenchmark.scala

@@ -0,0 +1,91 @@
+package mesosphere.marathon
+package upgrade
+
+import java.util.concurrent.TimeUnit
+import mesosphere.marathon.core.pod.{ MesosContainer, BridgeNetwork }
+import mesosphere.marathon.raml.{ Endpoint, Image, ImageType, Resources }
+import mesosphere.marathon.state.Container
+
+import mesosphere.marathon.state.PathId._
+import mesosphere.marathon.state._
+import mesosphere.marathon.core.deployment.DeploymentPlan
+import mesosphere.marathon.core.pod.PodDefinition
+import org.openjdk.jmh.annotations.{ Group => _, _ }
+import org.openjdk.jmh.infra.Blackhole
+
+import scala.collection.breakOut
+
+@State(Scope.Benchmark)
+object FlatDependencyBenchmark {
+
+  val version = VersionInfo.forNewConfig(Timestamp(1))
+
+  def makeApp(path: PathId) =
+    AppDefinition(
+      id = path,
+      labels = Map("ID" -> path.toString),
+      versionInfo = version,
+      networks = Seq(BridgeNetwork()),
+      container = Some(
+        Container.Docker(Nil, "alpine", List(Container.PortMapping(2015, Some(0), 10000, "tcp", Some("thing")))))
+    )
+
+  def makePod(path: PathId) =
+    PodDefinition(
+      id = path,
+      networks = Seq(BridgeNetwork()),
+      labels = Map("ID" -> path.toString),
+      version = version.lastConfigChangeAt,
+
+      containers = Seq(
+        MesosContainer(
+          "container-1",
+          resources = Resources(1.0),
+          image = Some(Image(ImageType.Docker, "alpine")),
+          endpoints = List(
+            Endpoint(
+              "service",
+              Some(2015),
+              Some(0),
+              Seq("tcp"))))))
+
+  val ids = 0 to 900
+
+  val podPaths: Vector[PathId] = ids.map { podId =>
+    s"/pod-${podId}".toPath
+  }(breakOut)
+
+  val appPaths: Vector[PathId] = ids.map { appId =>
+    s"/app-${appId}".toPath
+  }(breakOut)
+
+  val appDefs: Map[PathId, AppDefinition] = appPaths.map { path =>
+    path -> makeApp(path)
+  }(breakOut)
+
+  val podDefs: Map[PathId, PodDefinition] = podPaths.map { path =>
+    path -> makePod(path)
+  }(breakOut)
+
+  val rootGroup = RootGroup(apps = appDefs, pods = podDefs)
+  def upgraded = {
+    val pathId = "/app-901".toPath
+    RootGroup(
+      apps = rootGroup.apps + (pathId -> makeApp(pathId)),
+      pods = rootGroup.pods + (pathId -> makePod(pathId))
+    )
+  }
+}
+
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@BenchmarkMode(Array(Mode.Throughput, Mode.AverageTime))
+@Fork(1)
+class FlatDependencyBenchmark {
+  import FlatDependencyBenchmark._
+
+  @Benchmark
+  def deploymentPlanDependencySpeed(hole: Blackhole): Unit = {
+    val deployment = DeploymentPlan(rootGroup, upgraded)
+    hole.consume(deployment)
+  }
+}

src/main/scala/mesosphere/marathon/core/pod/PodDefinition.scala

@@ -33,6 +33,12 @@ case class PodDefinition(
     override val killSelection: KillSelection = KillSelection.DefaultKillSelection
 ) extends RunSpec with plugin.PodSpec with MarathonState[Protos.Json, PodDefinition] {
 
+  /**
+    * As an optimization, we precompute and cache the hash of this object
+    * This is done to speed up deployment plan computation.
+    */
+  override val hashCode: Int = scala.util.hashing.MurmurHash3.productHash(this)
+
   val endpoints: Seq[Endpoint] = containers.flatMap(_.endpoints)
   val resources = aggregateResources()
 

src/main/scala/mesosphere/marathon/state/AppDefinition.scala

@@ -86,6 +86,11 @@ case class AppDefinition(
   override val killSelection: KillSelection = KillSelection.DefaultKillSelection) extends RunSpec
     with plugin.ApplicationSpec with MarathonState[Protos.ServiceDefinition, AppDefinition] {
 
+  /**
+    * As an optimization, we precompute and cache the hash of this object
+    * This is done to speed up deployment plan computation.
+    */
+  override val hashCode: Int = scala.util.hashing.MurmurHash3.productHash(this)
   import mesosphere.mesos.protos.Implicits._
 
   require(

src/main/scala/mesosphere/marathon/state/PathId.scala

@@ -83,12 +83,12 @@ case class PathId(path: Seq[String], absolute: Boolean = true) extends Ordered[P
 
   override def equals(obj: Any): Boolean = {
     obj match {
-      case that: PathId => (that eq this) || (that.toString == toString)
+      case that: PathId => (that eq this) || (that.hashCode == hashCode && that.absolute == absolute && that.path == path)
       case _ => false
     }
   }
 
-  override def hashCode(): Int = toString.hashCode()
+  override val hashCode: Int = scala.util.hashing.MurmurHash3.productHash(this)
 }
 
 object PathId {

src/main/scala/mesosphere/marathon/upgrade/DeploymentPlan.scala

@@ -13,7 +13,6 @@ import mesosphere.marathon.raml.{ ArgvCommand, ShellCommand }
 import mesosphere.marathon.state._
 import mesosphere.marathon.storage.TwitterZk
 import mesosphere.marathon.storage.repository.legacy.store.{ CompressionConf, ZKData }
-import mesosphere.marathon.stream._
 import org.slf4j.LoggerFactory
 
 import scala.collection.SortedMap
@@ -96,19 +95,20 @@ case class DeploymentPlan(
 
   lazy val nonEmpty: Boolean = !isEmpty
 
-  lazy val affectedRunSpecs: Set[RunSpec] = steps.flatMap(_.actions.map(_.runSpec)).toSet
+  lazy val affectedRunSpecs: Set[RunSpec] = steps.flatMap(_.actions.map(_.runSpec))(collection.breakOut)
 
   /** @return all ids of apps which are referenced in any deployment actions */
-  lazy val affectedRunSpecIds: Set[PathId] = steps.flatMap(_.actions.map(_.runSpec.id)).toSet
+  lazy val affectedRunSpecIds: Set[PathId] = steps.flatMap(_.actions.map(_.runSpec.id))(collection.breakOut)
 
-  def affectedAppIds: Set[PathId] = affectedRunSpecs.collect{ case app: AppDefinition => app }.map(_.id)
-  def affectedPodIds: Set[PathId] = affectedRunSpecs.collect{ case pod: PodDefinition => pod }.map(_.id)
+  def affectedAppIds: Set[PathId] = affectedRunSpecs.collect{ case app: AppDefinition => app.id }
+  def affectedPodIds: Set[PathId] = affectedRunSpecs.collect{ case pod: PodDefinition => pod.id }
 
   def isAffectedBy(other: DeploymentPlan): Boolean =
     // FIXME: check for group change conflicts?
     affectedRunSpecIds.intersect(other.affectedRunSpecIds).nonEmpty
 
   lazy val createdOrUpdatedApps: Seq[AppDefinition] = {
+    import mesosphere.marathon.stream.toRichTraversableLike
     target.transitiveApps.filterAs(app => affectedRunSpecIds(app.id))(collection.breakOut)
   }
 
@@ -117,6 +117,7 @@ case class DeploymentPlan(
   }
 
   lazy val createdOrUpdatedPods: Seq[PodDefinition] = {
+    import mesosphere.marathon.stream.toRichTraversableLike
     target.transitivePodsById.values.filterAs(pod => affectedRunSpecIds(pod.id))(collection.breakOut)
   }
 
@@ -213,6 +214,8 @@ object DeploymentPlan {
     import org.jgrapht.graph.DefaultEdge
 
     def longestPathFromVertex[V](g: DirectedGraph[V, DefaultEdge], vertex: V): Seq[V] = {
+      import mesosphere.marathon.stream.RichSet
+
       val outgoingEdges: Set[DefaultEdge] =
         if (g.containsVertex(vertex)) g.outgoingEdgesOf(vertex)
         else Set.empty[DefaultEdge]
@@ -313,17 +316,13 @@ object DeploymentPlan {
     )
 
     // applications that are either new or the specs are different should be considered for the dependency graph
-    val addedOrChanged: Set[PathId] = targetRuns.flatMap {
-      case (runSpecId, spec) =>
-        if (!originalRuns.containsKey(runSpecId) ||
-          (originalRuns.containsKey(runSpecId) && originalRuns(runSpecId) != spec)) {
-          // the above could be optimized/refined further by checking the version info. The tests are actually
-          // really bad about structuring this correctly though, so for now, we just make sure that
-          // the specs are different (or brand new)
-          Some(runSpecId)
-        } else {
-          None
-        }
+    val addedOrChanged: Set[PathId] = targetRuns.collect {
+      case (runSpecId, spec) if (!originalRuns.contains(runSpecId) ||
+        (originalRuns.contains(runSpecId) && originalRuns(runSpecId) != spec)) =>
+        // the above could be optimized/refined further by checking the version info. The tests are actually
+        // really bad about structuring this correctly though, so for now, we just make sure that
+        // the specs are different (or brand new)
+        runSpecId
     }(collection.breakOut)
     val affectedApplications = addedOrChanged ++ (originalRuns.keySet -- targetRuns.keySet)