Fixes #3723 - Fix validation of duplicate volume names

src/main/scala/mesosphere/marathon/core/externalvolume/impl/providers/DVDIProvider.scala

@@ -115,24 +115,39 @@ private[impl] object DVDIProviderValidations extends ExternalVolumeValidations {
   // task instance across the entire cluster.
   override lazy val rootGroup = new Validator[Group] {
     override def apply(g: Group): Result = {
-      val appsByVolume =
+      val appsByVolume: Map[String, Set[PathId]] =
         g.transitiveApps
           .flatMap { app => namesOfMatchingVolumes(app).map(_ -> app.id) }
           .groupBy { case (volumeName, _) => volumeName }
           .mapValues(_.map { case (volumeName, appId) => appId })
 
-      val groupViolations = g.apps.flatMap { app =>
-        val ruleViolations = namesOfMatchingVolumes(app).flatMap { name =>
-          for {
-            otherApp <- appsByVolume(name)
-            if otherApp != app.id // do not compare to self
-          } yield RuleViolation(app.id, s"Requested volume $name conflicts with a volume in app $otherApp", None)
+      val appValid: Validator[AppDefinition] = {
+        def volumeNameUnique(appId: PathId): Validator[ExternalVolume] = {
+          def conflictingApps(vol: ExternalVolume): Set[PathId] =
+            appsByVolume.getOrElse(vol.external.name, Set.empty).filter(_ != appId)
+
+          isTrue { (vol: ExternalVolume) =>
+            val conflictingAppIds = conflictingApps(vol).mkString(", ")
+            s"Volume name '${vol.external.name}' in $appId conflicts with volume(s) of same name in app(s): " +
+              s"$conflictingAppIds"
+          }{ vol => conflictingApps(vol).isEmpty }
+        }
+
+        validator[AppDefinition] { app =>
+          app.externalVolumes is every(volumeNameUnique(app.id))
         }
-        if (ruleViolations.isEmpty) None
-        else Some(GroupViolation(app, "app contains conflicting volumes", None, ruleViolations.toSet))
       }
-      group(groupViolations)
+
+      def groupValid: Validator[Group] = validator[Group] { group =>
+        group.apps is every(appValid)
+        group.groups is every(groupValid)
+      }
+
+      // We need to call the validators recursively such that the "description" of the rule violations
+      // is correctly calculated.
+      groupValid(g)
     }
+
   }
 
   override lazy val app = {

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

@@ -211,8 +211,18 @@ object Group {
   def defaultDependencies: Set[PathId] = Set.empty
   def defaultVersion: Timestamp = Timestamp.now()
 
-  private def validNestedGroup(base: PathId): Validator[Group] =
-    validator[Group] { group =>
+  def validRootGroup(maxApps: Option[Int]): Validator[Group] = {
+    case object doesNotExceedMaxApps extends Validator[Group] {
+      override def apply(group: Group): Result = {
+        maxApps.filter(group.transitiveApps.size > _).map { num =>
+          Failure(Set(RuleViolation(group,
+            s"""This Marathon instance may only handle up to $num Apps!
+                |(Override with command line option --max_apps)""".stripMargin, None)))
+        } getOrElse Success
+      }
+    }
+
+    def validNestedGroup(base: PathId): Validator[Group] = validator[Group] { group =>
       group.id is validPathWithBase(base)
       group.apps is every(AppDefinition.validNestedAppDefinition(group.id.canonicalPath(base)))
       group is noAppsAndGroupsWithSameName
@@ -221,24 +231,11 @@ object Group {
       group.groups is every(valid(validNestedGroup(group.id.canonicalPath(base))))
     }
 
-  implicit val validRootGroup: Validator[Group] = new Validator[Group] {
-    override def apply(group: Group): Result = {
-      group is ExternalVolumes.validRootGroup
-      validate(group)(validator =
-        validNestedGroup(PathId.empty))
-    }
-  }
-
-  def validGroupWithConfig(maxApps: Option[Int])(implicit validator: Validator[Group]): Validator[Group] = {
-    new Validator[Group] {
-      override def apply(group: Group): Result = {
-        maxApps.filter(group.transitiveApps.size > _).map { num =>
-          Failure(Set(RuleViolation(group,
-            s"""This Marathon instance may only handle up to $num Apps!
-                |(Override with command line option --max_apps)""".stripMargin, None)))
-        } getOrElse Success
-      } and validator(group)
-    }
+    // We do not want a "/value" prefix, therefore we do not create nested validators with validator[Group]
+    // but chain the validators directly.
+    doesNotExceedMaxApps and
+      validNestedGroup(PathId.empty) and
+      ExternalVolumes.validRootGroup()
   }
 
   private def noAppsAndGroupsWithSameName: Validator[Group] =

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

@@ -156,7 +156,7 @@ class GroupManager @Inject() (
       from <- rootGroup()
       (toUnversioned, resolve) <- resolveStoreUrls(assignDynamicServicePorts(from, change(from)))
       to = GroupVersioningUtil.updateVersionInfoForChangedApps(version, from, toUnversioned)
-      _ = validateOrThrow(to)(Group.validGroupWithConfig(config.maxApps.get))
+      _ = validateOrThrow(to)(Group.validRootGroup(config.maxApps.get))
       plan = DeploymentPlan(from, to, resolve, version, toKill)
       _ = validateOrThrow(plan)
       _ = log.info(s"Computed new deployment plan:\n$plan")

src/test/scala/mesosphere/marathon/api/v2/ModelValidationTest.scala

@@ -35,12 +35,12 @@ class ModelValidationTest
       createServicePortApp("/c".toPath, 0)
     ))
 
-    val failedResult = Group.validGroupWithConfig(Some(2)).apply(group)
+    val failedResult = Group.validRootGroup(maxApps = Some(2)).apply(group)
     failedResult.isFailure should be(true)
     ValidationHelper.getAllRuleConstrains(failedResult)
       .find(v => v.message.contains("This Marathon instance may only handle up to 2 Apps!")) should be ('defined)
 
-    val successfulResult = Group.validGroupWithConfig(Some(10)).apply(group)
+    val successfulResult = Group.validRootGroup(maxApps = Some(10)).apply(group)
     successfulResult.isSuccess should be(true)
   }
 
@@ -49,7 +49,7 @@ class ModelValidationTest
     val conflictingApp = createServicePortApp("/app2".toPath, 3200)
 
     val group = Group(id = PathId.empty, apps = Set(existingApp, conflictingApp))
-    val result = validate(group)
+    val result = validate(group)(Group.validRootGroup(maxApps = None))
 
     ValidationHelper.getAllRuleConstrains(result).exists(v =>
       v.message == "Requested service port 3200 conflicts with a service port in app /app2") should be(true)
@@ -61,7 +61,7 @@ class ModelValidationTest
     val conflictingApp = createServicePortApp("/app2".toPath, 3201)
 
     val group = Group(id = PathId.empty, apps = Set(existingApp, conflictingApp))
-    val result = validate(group)
+    val result = validate(group)(Group.validRootGroup(maxApps = None))
 
     result.isSuccess should be(true)
   }
@@ -71,7 +71,7 @@ class ModelValidationTest
     val conflictingApp = existingApp.copy(id = "/app2".toPath)
 
     val group = Group(id = PathId.empty, apps = Set(existingApp, conflictingApp))
-    val result = validate(group)
+    val result = validate(group)(Group.validRootGroup(maxApps = None))
 
     ValidationHelper.getAllRuleConstrains(result).exists(v =>
       v.message == "Requested service port 3200 conflicts with a service port in app /app2") should be(true)
@@ -98,7 +98,7 @@ class ModelValidationTest
       ),
       Group("/test/group2".toPath)))
 
-    validate(group) match {
+    validate(group)(Group.validRootGroup(maxApps = None)) match {
       case Success => fail()
       case f: Failure =>
         val errors = (Json.toJson(f) \ "details").as[Seq[JsObject]]

src/test/scala/mesosphere/marathon/api/v2/json/GroupUpdateTest.scala

@@ -26,7 +26,7 @@ class GroupUpdateTest extends FunSuite with Matchers with GivenWhenThen {
     When("The update is performed")
     val result = update(group, timestamp)
 
-    validate(result).isSuccess should be(true)
+    validate(result)(Group.validRootGroup(maxApps = None)).isSuccess should be(true)
 
     Then("The update is applied correctly")
     result.id should be(PathId.empty)
@@ -69,7 +69,7 @@ class GroupUpdateTest extends FunSuite with Matchers with GivenWhenThen {
     When("The update is performed")
     val result: Group = update(actual, timestamp)
 
-    validate(result).isSuccess should be(true)
+    validate(result)(Group.validRootGroup(maxApps = None)).isSuccess should be(true)
 
     Then("The update is applied correctly")
     result.id should be(PathId.empty)
@@ -115,7 +115,7 @@ class GroupUpdateTest extends FunSuite with Matchers with GivenWhenThen {
     val timestamp = Timestamp.now()
     val result = update(current, timestamp)
 
-    validate(result).isSuccess should be(true)
+    validate(result)(Group.validRootGroup(maxApps = None)).isSuccess should be(true)
 
     Then("The update is reflected in the current group")
     result.id.toString should be("/test")
@@ -156,7 +156,7 @@ class GroupUpdateTest extends FunSuite with Matchers with GivenWhenThen {
     When("The update is performed")
     val result = update(Group.empty, Timestamp.now())
 
-    validate(result).isSuccess should be(true)
+    validate(result)(Group.validRootGroup(maxApps = None)).isSuccess should be(true)
 
     Then("The update is applied correctly")
     val group = result.group("test-group".toRootPath)

src/test/scala/mesosphere/marathon/core/externalvolume/impl/providers/DVDIProviderRootGroupValidationTest.scala

@@ -0,0 +1,133 @@
+package mesosphere.marathon.core.externalvolume.impl.providers
+
+import com.wix.accord.{ RuleViolation, Success, Failure, Result }
+import mesosphere.marathon.api.v2.Validation
+import mesosphere.marathon.core.externalvolume.ExternalVolumes
+import mesosphere.marathon.state._
+import org.apache.mesos.{ Protos => MesosProtos }
+import org.scalatest.{ FunSuite, GivenWhenThen, Matchers }
+import play.api.libs.json.{ JsString, Json }
+
+import scala.collection.immutable.Seq
+
+class DVDIProviderRootGroupValidationTest extends FunSuite with Matchers with GivenWhenThen {
+
+  test("two volumes with different names should not result in an error") {
+    val f = new Fixture
+    Given("a root group with two apps and conflicting volumes")
+    val rootGroup = Group(
+      id = PathId.empty,
+      groups = Set(
+        Group(
+          id = PathId("/nested"),
+          apps = Set(
+            f.appWithDVDIVolume(appId = PathId("/nested/foo1"), volumeName = "vol1"),
+            f.appWithDVDIVolume(appId = PathId("/nested/foo2"), volumeName = "vol2")
+          )
+        )
+      )
+    )
+
+    f.checkResult(
+      rootGroup,
+      expectedViolations = Set.empty
+    )
+  }
+
+  test("two volumes with same name result in an error") {
+    val f = new Fixture
+    Given("a root group with two apps and conflicting volumes")
+    val app1 = f.appWithDVDIVolume(appId = PathId("/nested/app1"), volumeName = "vol")
+    val app2 = f.appWithDVDIVolume(appId = PathId("/nested/app2"), volumeName = "vol")
+    val rootGroup = Group(
+      id = PathId.empty,
+      groups = Set(
+        Group(
+          id = PathId("/nested"),
+          apps = Set(app1, app2)
+        )
+      )
+    )
+
+    f.checkResult(
+      rootGroup,
+      expectedViolations = Set(
+        RuleViolation(
+          value = app1.externalVolumes.head,
+          constraint = "Volume name 'vol' in /nested/app1 conflicts with volume(s) of same name in app(s): /nested/app2",
+          description = Some("/groups(0)/apps(0)/externalVolumes(0)")
+        ),
+        RuleViolation(
+          value = app2.externalVolumes.head,
+          constraint = "Volume name 'vol' in /nested/app2 conflicts with volume(s) of same name in app(s): /nested/app1",
+          description = Some("/groups(0)/apps(1)/externalVolumes(0)")
+        )
+      )
+    )
+  }
+
+  class Fixture {
+    def appWithDVDIVolume(appId: PathId, volumeName: String, provider: String = DVDIProvider.name): AppDefinition = {
+      AppDefinition(
+        id = appId,
+        cmd = Some("sleep 123"),
+        upgradeStrategy = UpgradeStrategy.forResidentTasks,
+        container = Some(
+          Container(
+            `type` = MesosProtos.ContainerInfo.Type.MESOS,
+            volumes = Seq(
+              ExternalVolume(
+                containerPath = "ignoreme",
+                external = ExternalVolumeInfo(
+                  name = volumeName,
+                  provider = provider,
+                  options = Map(
+                    DVDIProvider.driverOption -> "rexray"
+                  )
+                ),
+                mode = MesosProtos.Volume.Mode.RW
+              )
+            )
+          )
+        )
+      )
+    }
+
+    def jsonResult(result: Result): String = {
+      Json.prettyPrint(
+        result match {
+          case Success    => JsString("Success")
+          case f: Failure => Json.toJson(f)(Validation.failureWrites)
+        }
+      )
+    }
+
+    def checkResult(rootGroup: Group, expectedViolations: Set[RuleViolation]): Unit = {
+      val expectFailure = expectedViolations.nonEmpty
+
+      When("validating the root group")
+      val result = DVDIProviderValidations.rootGroup(rootGroup)
+
+      Then(s"we should ${if (expectFailure) "" else "NOT "}get an error")
+      withClue(jsonResult(result)) { result.isFailure should be(expectFailure) }
+
+      And("ExternalVolumes validation agrees")
+      val externalVolumesResult = ExternalVolumes.validRootGroup()(rootGroup)
+      withClue(jsonResult(externalVolumesResult)) { externalVolumesResult.isFailure should be(expectFailure) }
+
+      And("global validation agrees")
+      val globalResult: Result = Group.validRootGroup(maxApps = None)(rootGroup)
+      withClue(jsonResult(globalResult)) { globalResult.isFailure should be(expectFailure) }
+
+      val ruleViolations = globalResult match {
+        case Success    => Set.empty[RuleViolation]
+        case f: Failure => f.violations.flatMap(Validation.allRuleViolationsWithFullDescription(_))
+      }
+
+      And("the rule violations are the expected ones")
+      withClue(jsonResult(globalResult)) {
+        ruleViolations should equal(expectedViolations)
+      }
+    }
+  }
+}