[SPARK-19606] Support constraints in spark-dispatcher; applied recommended changes from PR

Author: Paul Mackles

docs/running-on-mesos.md

@@ -450,7 +450,9 @@ See the [configuration page](configuration.html) for information on Spark config
   <td><code>spark.mesos.constraints</code></td>
   <td>(none)</td>
   <td>
-    Attribute based constraints on mesos resource offers. By default, all resource offers will be accepted. Refer to <a href="http://mesos.apache.org/documentation/attributes-resources/">Mesos Attributes & Resources</a> for more information on attributes.
+    Attribute based constraints on mesos resource offers. By default, all resource offers will be accepted. This setting
+    applies only to executors. Refer to <a href="http://mesos.apache.org/documentation/attributes-resources/">Mesos
+    Attributes & Resources</a> for more information on attributes.
     <ul>
       <li>Scalar constraints are matched with "less than equal" semantics i.e. value in the constraint must be less than or equal to the value in the resource offer.</li>
       <li>Range constraints are matched with "contains" semantics i.e. value in the constraint must be within the resource offer's value.</li>
@@ -465,7 +467,7 @@ See the [configuration page](configuration.html) for information on Spark config
   <td>(none)</td>
   <td>
     Same as <code>spark.mesos.constraints</code> except applied to drivers when launched through the dispatcher. By default,
-    all resource offers will be accepted.
+    all offers with sufficient resources will be accepted.
   </td>
 </tr>
 <tr>

resource-managers/mesos/src/main/scala/org/apache/spark/deploy/mesos/config.scala

@@ -122,4 +122,11 @@ package object config {
         "Example: key1:val1,key2:val2")
       .stringConf
       .createOptional
+
+  private[spark] val DRIVER_CONSTRAINTS =
+    ConfigBuilder("spark.mesos.driver.constraints")
+      .doc("Attribute based constraints on mesos resource offers. Applied by the dispatcher " +
+        "when launching drivers. Default is to accept all offers with sufficient resources.")
+      .stringConf
+      .createWithDefault("")
 }

resource-managers/mesos/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterScheduler.scala

@@ -559,7 +559,7 @@ private[spark] class MesosClusterScheduler(
       var remainingResources: JList[Resource],
       var attributes: JList[Attribute]) {
     override def toString(): String = {
-      s"Offer id: ${offer.getId}, resources: ${remainingResources}"
+      s"Offer id: ${offer.getId}, resources: ${remainingResources}, attributes: ${attributes}"
     }
   }
 
@@ -603,7 +603,7 @@ private[spark] class MesosClusterScheduler(
       val driverCpu = submission.cores
       val driverMem = submission.mem
       val driverConstraints =
-        parseConstraintString(submission.conf.get("spark.mesos.driver.constraints", ""))
+        parseConstraintString(submission.conf.get(config.DRIVER_CONSTRAINTS))
       logTrace(s"Finding offer to launch driver with cpu: $driverCpu, mem: $driverMem, " +
         s"driverConstraints: $driverConstraints")
       val offerOption = currentOffers.find { offer =>

resource-managers/mesos/src/test/scala/org/apache/spark/scheduler/cluster/mesos/MesosClusterSchedulerSuite.scala

@@ -274,7 +274,7 @@ class MesosClusterSchedulerSuite extends SparkFunSuite with LocalSparkContext wi
           command,
           Map("spark.mesos.executor.home" -> "test",
             "spark.app.name" -> "test",
-            "spark.mesos.driver.constraints" -> driverConstraints),
+            config.DRIVER_CONSTRAINTS.key -> driverConstraints),
           "s1",
           new Date()))
       assert(response.success)