Migration to fix unreachable strategy for resident apps.

Summary:
fixes #5209

forward port 809b9f9

Also-By: tharper@mesosphere.com

Test Plan: with older version of marathon, create resident app. Upgrade. Check that resident app unreachable strategy is disabled. Modify app and see if there are validation issues

Reviewers: jasongilanfarr, aquamatthias, jenkins

Reviewed By: jasongilanfarr, aquamatthias, jenkins

Subscribers: marathon-team

Differential Revision: https://phabricator.mesosphere.com/D557

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

@@ -8,7 +8,7 @@ import mesosphere.marathon.state.AppDefinition.AppKey
 import mesosphere.marathon.state.Group.GroupKey
 import mesosphere.marathon.state.PathId._
 import mesosphere.marathon.state._
-import org.openjdk.jmh.annotations.{Group => _, _}
+import org.openjdk.jmh.annotations.{ Group => _, _ }
 import org.openjdk.jmh.infra.Blackhole
 
 import scala.collection.breakOut

src/main/scala/mesosphere/marathon/storage/migration/Migration.scala

@@ -16,6 +16,7 @@ import scala.concurrent.duration._
 import scala.concurrent.{ Await, Future }
 import scala.util.control.NonFatal
 import scala.util.matching.Regex
+import mesosphere.marathon.storage.migration.legacy.MigrationTo_1_4_2
 
 /**
   * @param persistenceStore Optional "new" PersistenceStore for new migrations, the repositories
@@ -48,7 +49,14 @@ class Migration(
     * All the migrations, that have to be applied.
     * They get applied after the master has been elected.
     */
-  def migrations: List[MigrationAction] = List.empty
+  def migrations: List[MigrationAction] =
+    List(
+      StorageVersions(1, 4, 2, StorageVersion.StorageFormat.PERSISTENCE_STORE) -> { () =>
+        new MigrationTo_1_4_2(appRepository).migrate().recover {
+          case NonFatal(e) => throw new MigrationFailedException("while migrating storage to 1.4.2", e)
+        }
+      }
+    )
 
   def applyMigrationSteps(from: StorageVersion): Future[Seq[StorageVersion]] = {
     migrations.filter(_._1 > from).sortBy(_._1).foldLeft(Future.successful(Seq.empty[StorageVersion])) {

src/main/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo_1_4_2.scala

@@ -0,0 +1,52 @@
+package mesosphere.marathon
+package storage.migration.legacy
+
+import akka.Done
+import akka.stream.Materializer
+import akka.stream.scaladsl.{ Flow, Keep, Sink }
+import com.typesafe.scalalogging.StrictLogging
+import mesosphere.marathon.state._
+import mesosphere.marathon.storage.repository.AppRepository
+
+import scala.concurrent.{ ExecutionContext, Future }
+
+@SuppressWarnings(Array("ClassNames"))
+class MigrationTo_1_4_2(appRepository: AppRepository)(implicit
+  ctx: ExecutionContext,
+    mat: Materializer) extends StrictLogging {
+
+  import MigrationTo_1_4_2.migrationFlow
+  val sink =
+    Flow[AppDefinition]
+      .mapAsync(Int.MaxValue)(appRepository.store)
+      .toMat(Sink.ignore)(Keep.right)
+
+  def migrate(): Future[Done] = {
+    logger.info("Starting migration to 1.4.2")
+
+    appRepository.all()
+      .via(migrationFlow)
+      .runWith(sink)
+      .andThen {
+        case _ =>
+          logger.info("Finished 1.4.2 migration")
+      }
+  }
+}
+
+object MigrationTo_1_4_2 extends StrictLogging {
+  private def fixResidentApp(app: AppDefinition): AppDefinition = {
+    if (app.isResident)
+      app.copy(unreachableStrategy = UnreachableDisabled)
+    else
+      app
+  }
+
+  val migrationFlow =
+    Flow[AppDefinition]
+      .filter(_.isResident)
+      .map { app =>
+        logger.info(s"Disable Unreachable strategy for resident app: ${app.id}")
+        fixResidentApp(app)
+      }
+}

src/test/scala/mesosphere/marathon/storage/migration/MigrationTest.scala

@@ -129,6 +129,7 @@ class MigrationTest extends AkkaUnitTest with Mockito with GivenWhenThen {
       mockedStore.setStorageVersion(any) returns Future.successful(Done)
 
       val migrate = migration(persistenceStore = mockedStore)
+      migrate.appRepository.all() returns Source(Nil)
       val result = migrate.migrate()
       result should be ('nonEmpty)
       result should be(migrate.migrations.drop(1).map(_._1))

src/test/scala/mesosphere/marathon/storage/migration/legacy/MigrationTo_1_4_2Test.scala

@@ -0,0 +1,40 @@
+package mesosphere.marathon
+package storage.migration.legacy
+
+import akka.stream.scaladsl.Source
+import mesosphere.AkkaUnitTest
+import mesosphere.marathon.Protos.ResidencyDefinition.TaskLostBehavior
+import mesosphere.marathon.state.{ AppDefinition, PathId, Residency, UnreachableDisabled, UnreachableStrategy }
+import mesosphere.marathon.stream.Sink
+import mesosphere.marathon.test.GroupCreation
+
+class MigrationTo_1_4_2Test extends AkkaUnitTest with GroupCreation {
+  import MigrationTo_1_4_2.migrationFlow
+
+  "Migration to 1.4.2" should {
+    "do nothing if there are no resident apps" in {
+      val result = Source.single(AppDefinition(id = PathId("abc")))
+        .via(migrationFlow)
+        .runWith(Sink.seq)
+        .futureValue
+
+      result.shouldBe(Nil)
+    }
+
+    "fix wrong UnreachableStrategy for resident apps" in {
+      val badApp = AppDefinition(
+        id = PathId("/badApp"),
+        residency = Some(Residency(23L, TaskLostBehavior.WAIT_FOREVER)),
+        unreachableStrategy = UnreachableStrategy.default(resident = false))
+      val goodApp = AppDefinition(id = PathId("/goodApp"))
+      val fixedApp = badApp.copy(unreachableStrategy = UnreachableDisabled)
+
+      val result = Source(Seq(badApp, goodApp))
+        .via(migrationFlow)
+        .runWith(Sink.seq)
+        .futureValue
+
+      result shouldBe (Seq(fixedApp))
+    }
+  }
+}