Prevent Marathon to scale down instances to zero

When :
- a deployment is stuck in an unsuccessful state (constraint
impossible to satisfy)
- maximumOverCapacity is used
- a new deployment has been requested and cancelled

The periodic scale thread removes all instances because number of
instances to decom is processed as max(goalRunning-scaleTo,
sentencedTasks).

This patch aims to change this behaviour by using number of actual
instances running instead of target number of instances (goalRunning).

JIRA: MESOS-5576

src/main/scala/mesosphere/marathon/core/deployment/ScalingProposition.scala

@@ -19,6 +19,10 @@ object ScalingProposition extends StrictLogging {
     killSelection: KillSelection,
     runSpecId: AbsolutePathId): ScalingProposition = {
 
+    val instancesRunning: scala.collection.immutable.Seq[mesosphere.marathon.core.instance.Instance] = instances
+      .filter(_.tasksMap.size != 0)
+      .filter(_.state.condition != UnreachableInactive)
+
     val instancesGoalRunning: Map[Instance.Id, Instance] = instances
       .filter(_.state.goal == Goal.Running)
       .filter(_.state.condition != UnreachableInactive)
@@ -30,7 +34,7 @@ object ScalingProposition extends StrictLogging {
         toDecommissionMap.contains(instanceId)
     }
     // overall number of tasks that need to be killed
-    val decommissionCount = math.max(instancesGoalRunning.size - scaleTo, sentencedAndRunningMap.size)
+    val decommissionCount = math.max(instancesRunning.size - scaleTo, sentencedAndRunningMap.size)
     // tasks that should be killed to meet constraints – pass notSentenced & consider the sentenced 'already killed'
     val killToMeetConstraints = meetConstraints(
       notSentencedAndRunningMap.values.to[Seq],

src/test/scala/mesosphere/marathon/core/deployment/impl/ScalingPropositionTest.scala

@@ -219,6 +219,24 @@ class ScalingPropositionTest extends UnitTest {
     }
   }
 
+  "Removing tasks is not based on goals" should {
+    "not generate decommission" in {
+      var f = new Fixture
+      val t1 = f.createInstance(1)
+      val t2 = f.createUnscheduledInstance()
+      val t3 = f.createUnscheduledInstance()
+      val proposition = ScalingProposition.propose(
+        instances = Seq(t1, t2, t3),
+        toDecommission = f.noTasks,
+        meetConstraints = f.noConstraintsToMeet,
+        scaleTo = 1,
+        killSelection = KillSelection.DefaultKillSelection,
+        f.appId
+      )
+      proposition.toDecommission.nonEmpty shouldBe false
+    }
+  }
+
   "ScalingProposition.sortByConditionAndDate" when {
     "sorting a unreachable, unhealthy, running, staging and healthy tasks" should {
       val f = new Fixture
@@ -309,6 +327,10 @@ class ScalingPropositionTest extends UnitTest {
       instance.copy(state = state)
     }
 
+    def createUnscheduledInstance(): Instance = {
+      TestInstanceBuilder.newBuilder(appId).getInstance()
+    }
+
     def noConstraintsToMeet(running: Seq[Instance], killCount: Int) = // linter:ignore:UnusedParameter
       Seq.empty[Instance]