Misc improvements

.travis.yml

@@ -0,0 +1,4 @@
+before_install: 
+  - sudo add-apt-repository -y ppa:ubuntu-cloud-archive/icehouse-staging
+
+install: bin/build

pom.xml

@@ -9,16 +9,16 @@
     <version>${elasticsearch.version}-${mesos.version}-${elasticsearch-mesos.version}</version>
 
     <properties>
-        <mesos-utils.version>0.19.0-1</mesos-utils.version>
+        <mesos-utils.version>0.20.1-1</mesos-utils.version>
 
         <!-- Elasticsearch Mesos version. Looks like a build number from Maven's perspective-->
         <elasticsearch-mesos.version>1</elasticsearch-mesos.version>
 
         <!-- language & dependency versions -->
         <java.abi>1.7</java.abi>
         <scala.version>2.10.4</scala.version>
-        <elasticsearch.version>1.2.1</elasticsearch.version>
-        <mesos.version>0.19.0</mesos.version>
+        <elasticsearch.version>1.4.1</elasticsearch.version>
+        <mesos.version>0.20.1</mesos.version>
         <logback.version>1.0.9</logback.version>
 
     </properties>

src/main/scala/mesosphere/elasticsearch/ElasticSearchScheduler.scala

@@ -20,7 +20,8 @@ class ElasticSearchScheduler(masterUrl: String,
                          confServerHostName: String,
                          confServerPort: Int,
                          resources: mutable.Map[String, Float],
-                         numberOfHwNodes: Int)
+                         numberOfHwNodes: Int,
+                         requestedAttributes: mutable.Map[String, String])
   extends Scheduler with Runnable with Logger {
 
   val initialized = new CountDownLatch(1)
@@ -89,7 +90,7 @@ class ElasticSearchScheduler(masterUrl: String,
       s"&& rm logging.yml " +
       s"&& curl -sSfLO http://${confServerHostName}:${confServerPort}/logging.yml " +
       s"&& cd .. " +
-      s"&& bin/elasticsearch -f")
+      s"&& bin/elasticsearch")
 
 
 
@@ -141,6 +142,22 @@ class ElasticSearchScheduler(masterUrl: String,
   // Check if offer is reasonable
   def isOfferGood(offer: Offer) = {
 
+    // First of all check if we have all the requested 
+    // attributes in the offer.
+    // To do so we make sure the intersection of the
+    // offered and required attributes is actually 
+    // equal to the required attributes themselves.
+    val offeredAttributes = offer.getAttributesList.asScala.toList.map {
+      k => (k.getName, k.getText.getValue)
+    }.toSet
+
+    val requiredAttrs = requestedAttributes.toSet
+    val hasAllAttributes = offeredAttributes.intersect(requiredAttrs) == requiredAttrs
+
+    debug("attributes offered: " + offeredAttributes)
+    debug("attributes required: " + requiredAttrs)
+
+
     // Make a list of offered resources
     val offeredRes = offer.getResourcesList.asScala.toList.map {
       k => (k.getName, k.getScalar.getValue)
@@ -170,8 +187,8 @@ class ElasticSearchScheduler(masterUrl: String,
     }.size
 
     // don't start the same framework multiple times on the same host and
-    // make sure we got all resources we asked for
-    taskSet.forall(_.hostname != offer.getHostname) && offersTooSmall == 0
+    // make sure we got all resources and attributes we asked for
+    taskSet.forall(_.hostname != offer.getHostname) && offersTooSmall == 0 && hasAllAttributes
   }
 
   def reregistered(driver: SchedulerDriver, masterInfo: MasterInfo) {

src/main/scala/mesosphere/elasticsearch/Main.scala

@@ -43,6 +43,12 @@ object Main extends App with Logger {
     case (k, v) => k.replaceAllLiterally("resource.", "") -> v.toString.toFloat
   }
 
+  val requestedAttributes = mesosConf.filter {
+    _._1.startsWith("attribute.")
+  }.map {
+    case (k, v) => k.replaceAllLiterally("attribute.", "") -> v.toString
+  }
+
   //TODO load the ElasticSearch log-properties file
   BasicConfigurator.configure()
   getRootLogger.setLevel(Level.INFO)
@@ -55,7 +61,8 @@ object Main extends App with Logger {
     confServerHostName,
     confServerPort,
     resources,
-    numberOfHwNodes)
+    numberOfHwNodes,
+    requestedAttributes)
 
   val schedThred = new Thread(scheduler)
   schedThred.start()