Merge branch 'master' of github.com:mesos/elasticsearch into feature/credentialsChange

Author: Sebastian Brandt

.travis.yml

@@ -0,0 +1,23 @@
+language: java
+
+jdk:
+  - oraclejdk8
+
+sudo: required
+
+install:
+# one liner installation of docker 1.9.1 below did not work (see https://github.com/moul/travis-docker/issues/38).
+#  - curl -sLo - http://j.mp/install-travis-docker | sh -xe
+# Therefore installing it through a script
+  - sudo sh -c 'echo "deb https://apt.dockerproject.org/repo ubuntu-precise main" > /etc/apt/sources.list.d/docker.list'
+  - sudo apt-key adv --keyserver hkp://p80.pool.sks-keyservers.net:80 --recv-keys 58118E89F3A912897C070ADBF76221572C52609D
+  - sudo apt-get update
+  - sudo apt-key update
+  - sudo apt-get -qqy install docker-engine=1.9.1-0~precise
+# Has to run this script with sudo because custom installation does not allow $USER to use docker and it's not possible to relogin
+
+# Has to run the build script with sudo because custom installation does not allow $USER to use docker and it's not possible to relogin
+script: chmod +x travis.sh && sudo ./travis.sh
+
+notifications:
+  email: true

build.gradle

@@ -13,7 +13,7 @@ ext {
     commonsIOVersion = "2.4"
     commonsLangVersion = "3.4"
     commonsValidatorVersion = "1.5.0"
-    dockerJavaVersion = "1.4.0"
+    dockerJavaVersion = "3.0.0"
     elasticsearchVersion="2.2.0"
     gradleDownloadTaskVersion = "2.1.0"
     hamcrestVersion = "1.3"
@@ -25,7 +25,7 @@ ext {
     junitVersion = "4.12"
     log4jVersion = "1.2.17"
     mesosVer = "0.28.2"
-    minimesosVersion = "0.7.1"
+    minimesosVersion = "0.10.2"
     springBootVersion = "1.2.5.RELEASE" // Bumping SB version causes Jackson incompatabilities with Docker-Java
     unirestVersion = "1.4.8"
 

commons/src/main/java/org/apache/mesos/elasticsearch/common/zookeeper/model/ZKAddress.java

@@ -14,7 +14,7 @@ public class ZKAddress {
     public static final String ZK_PREFIX = "zk://";
     public static final String USER_AND_PASS_REG = "([^/@:]+):([^/@:]+)";
     public static final String HOST_AND_PORT_REG = "([A-z0-9-.]+)(?::)([0-9]+)";
-    public static final String ZK_NODE_REG = "/([^/]+)";
+    public static final String ZK_NODE_REG = "/([^/]+(?:/[^/]+)*)";
     public static final String ADDRESS_REGEX = "^(?:" + USER_AND_PASS_REG + "@)?" + HOST_AND_PORT_REG + "(?:" + ZK_NODE_REG + ")?";
     public static final String VALID_ZK_URL = "zk://host1:port1,user:pass@host2:port2/path,.../path";
     private String user;

commons/src/main/resources/log4j.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8" ?>
 <!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
 
-<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
+<log4j:configuration xmlns:log4j="```http://jakarta.apache.org/log4j/">
     <appender name="console" class="org.apache.log4j.ConsoleAppender">
         <param name="Target" value="System.out"/>
         <layout class="org.apache.log4j.PatternLayout">

commons/src/test/java/org/apache/mesos/elasticsearch/common/zookeeper/parser/ZKAddressParserTest.java

@@ -70,6 +70,30 @@ public void shouldAcceptIfMultiZKAddressWithPath() {
         assertZKEquals(zk.get(1), "", "", "10.4.52.3", "1234", "mesos");
     }
 
+    @Test
+    public void shouldAcceptIfSingleZKAddressWithSubpath() {
+        String add = "zk://192.168.0.1:2182/mesos/subpath";
+        List<ZKAddress> zk = new ZKAddressParser().validateZkUrl(add);
+        assertZKEquals(zk.get(0), "", "", "192.168.0.1", "2182", "mesos/subpath");
+    }
+
+    @Test
+    public void shouldAcceptIfMultiZKAddressWithSubpath() {
+        String add = "zk://192.168.0.1:2182,10.4.52.3:1234/mesos/sub_path";
+        List<ZKAddress> zk = new ZKAddressParser().validateZkUrl(add);
+        assertZKEquals(zk.get(0), "", "", "192.168.0.1", "2182", "");
+        assertZKEquals(zk.get(1), "", "", "10.4.52.3", "1234", "mesos/sub_path");
+    }
+
+    @Test
+    public void shouldAcceptIfMultiZKAddressWithMultiSubpath() {
+        String add = "zk://192.168.0.1:2182/mesos/sub_path1,10.4.52.3:1234/mesos/sub_path2";
+        List<ZKAddress> zk = new ZKAddressParser().validateZkUrl(add);
+        assertZKEquals(zk.get(0), "", "", "192.168.0.1", "2182", "mesos/sub_path1");
+        assertZKEquals(zk.get(1), "", "", "10.4.52.3", "1234", "mesos/sub_path2");
+    }
+
+
     @Test
     public void shouldAcceptIfSpacesInPath() {
         String add = "zk://192.168.0.1:2182, 10.4.52.3:1234/mesos";
@@ -108,4 +132,4 @@ private void assertZKEquals(ZKAddress zk, String user, String pass, String addr,
         assertEquals(pass, zk.getPassword());
         assertEquals(zkNode, zk.getZkNode());
     }
-}
+}

docs/index.md

@@ -158,6 +158,9 @@ Usage: (Options preceded by an asterisk are required) [options]
     --executorName
        The name given to the executor task.
        Default: elasticsearch-executor
+    --executorLabels
+       One or more labels given to the executor task. Example: 'environment=prod bananas=apples'
+       Default: <empty string>
     --externalVolumeDriver
        Use external volume storage driver. By default, nodes will use volumes on
        host.

scheduler/build.gradle

@@ -27,7 +27,9 @@ dependencies {
 
     compile project(":commons")
 
-    compile("org.springframework.boot:spring-boot-starter-web:${springBootVersion}")
+    compile("org.springframework.boot:spring-boot-starter-web:${springBootVersion}") {
+        exclude group: 'com.fasterxml.jackson.core', module: 'jackson-databind'
+    }
     compile("org.springframework.boot:spring-boot-starter-log4j:${springBootVersion}")
     compile "commons-io:commons-io:${commonsIOVersion}"
     compile "org.apache.commons:commons-collections4:${commonsCollectionsVersion}"

scheduler/src/main/java/org/apache/mesos/elasticsearch/scheduler/Configuration.java

@@ -19,6 +19,8 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
 import java.util.stream.Collectors;
 
 import static java.util.Arrays.asList;
@@ -36,6 +38,7 @@ public class Configuration {
     public static final String WEB_UI_PORT = "--webUiPort";
     public static final String FRAMEWORK_NAME = "--frameworkName";
     public static final String EXECUTOR_NAME = "--executorName";
+    public static final String EXECUTOR_LABELS = "--executorLabels";
     public static final String DATA_DIR = "--dataDir";
     public static final String DEFAULT_HOST_DATA_DIR = "/var/lib/mesos/slave/elasticsearch";
     // DCOS Certification requirement 01
@@ -85,6 +88,9 @@ public class Configuration {
     private String frameworkName = "elasticsearch";
     @Parameter(names = {EXECUTOR_NAME}, description = "The name given to the executor task.", validateWith = CLIValidators.NotEmptyString.class)
     private String executorName = "elasticsearch-executor";
+    @Parameter(names = {EXECUTOR_LABELS}, description = "One or more labels given to the executor task." +
+        "E.g. 'environment=prod bananas=apples'", variableArity = true)
+    private List<String> executorLabels = new ArrayList<>();
     @Parameter(names = {DATA_DIR}, description = "The host data directory used by Docker volumes in the executors. [DOCKER MODE ONLY]")
     private String dataDir = DEFAULT_HOST_DATA_DIR;
     @Parameter(names = {FRAMEWORK_FAILOVER_TIMEOUT}, description = "The time before Mesos kills a scheduler and tasks if it has not recovered (ms).", validateValueWith = CLIValidators.PositiveDouble.class)
@@ -176,6 +182,18 @@ public String getTaskName() {
         return executorName;
     }
 
+    public Map<String, String> getTaskLabels() {
+        HashMap<String, String> map = new HashMap<>();
+        for (String keyValue : executorLabels) {
+            String[] kvp = keyValue.split("=", 2);
+            if (kvp.length == 2) {
+              map.put(kvp[0], kvp[1]);
+            }
+        }
+
+        return map;
+    }
+
     public String getDataDir() {
         return dataDir;
     }

scheduler/src/main/java/org/apache/mesos/elasticsearch/scheduler/TaskInfoFactory.java

@@ -19,6 +19,7 @@
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.stream.Collectors;
@@ -69,6 +70,7 @@ private Protos.TaskInfo buildNativeTask(Protos.Offer offer, Configuration config
         final List<Integer> ports = getPorts(offer, configuration);
         final List<Protos.Resource> resources = getResources(configuration, ports);
         final Protos.DiscoveryInfo discovery = getDiscovery(ports, configuration);
+        final Protos.Labels labels = getLabels(configuration);
 
         final String hostAddress = resolveHostAddress(offer, ports);
 
@@ -83,6 +85,7 @@ private Protos.TaskInfo buildNativeTask(Protos.Offer offer, Configuration config
                 .setSlaveId(offer.getSlaveId())
                 .addAllResources(resources)
                 .setDiscovery(discovery)
+                .setLabels(labels)
                 .setCommand(nativeCommand(configuration, args, elasticSearchNodeId))
                 .build();
     }
@@ -91,6 +94,7 @@ private Protos.TaskInfo buildDockerTask(Protos.Offer offer, Configuration config
         final List<Integer> ports = getPorts(offer, configuration);
         final List<Protos.Resource> resources = getResources(configuration, ports);
         final Protos.DiscoveryInfo discovery = getDiscovery(ports, configuration);
+        final Protos.Labels labels = getLabels(configuration);
 
         final String hostAddress = resolveHostAddress(offer, ports);
 
@@ -107,6 +111,7 @@ private Protos.TaskInfo buildDockerTask(Protos.Offer offer, Configuration config
                 .setSlaveId(offer.getSlaveId())
                 .addAllResources(resources)
                 .setDiscovery(discovery)
+                .setLabels(labels)
                 .setCommand(dockerCommand(configuration, args, elasticSearchNodeId))
                 .setContainer(containerInfo)
                 .build();
@@ -125,8 +130,7 @@ private List<Integer> getPorts(Protos.Offer offer, Configuration configuration)
         if (elasticsearchPorts.isEmpty() || elasticsearchPorts.stream().allMatch(port -> port == 0)) {
             //No ports requested by user or two random ports requested
             ports = Resources.selectTwoPortsFromRange(offer.getResourcesList());
-        }
-        else {
+        } else {
             //Replace a user requested port 0 with a random port
             ports = elasticsearchPorts.stream().map(port -> port != 0 ?  port : Resources.selectOnePortFromRange(offer.getResourcesList())).collect(Collectors.toList());
         }
@@ -143,14 +147,29 @@ private List<Protos.Resource> getResources(Configuration configuration, List<Int
     private Protos.DiscoveryInfo getDiscovery(List<Integer> ports, Configuration configuration) {
         Protos.DiscoveryInfo.Builder discovery = Protos.DiscoveryInfo.newBuilder();
         Protos.Ports.Builder discoveryPorts = Protos.Ports.newBuilder();
-        discoveryPorts.addPorts(Discovery.CLIENT_PORT_INDEX, Protos.Port.newBuilder().setNumber(ports.get(0)).setName(Discovery.CLIENT_PORT_NAME).setProtocol("TCP"));
-        discoveryPorts.addPorts(Discovery.TRANSPORT_PORT_INDEX, Protos.Port.newBuilder().setNumber(ports.get(1)).setName(Discovery.TRANSPORT_PORT_NAME).setProtocol("TCP"));
+        discoveryPorts.addPorts(Discovery.CLIENT_PORT_INDEX, Protos.Port.newBuilder().setNumber(ports.get(0)).setName(Discovery.CLIENT_PORT_NAME).setProtocol("tcp"));
+        discoveryPorts.addPorts(Discovery.TRANSPORT_PORT_INDEX, Protos.Port.newBuilder().setNumber(ports.get(1)).setName(Discovery.TRANSPORT_PORT_NAME).setProtocol("tcp"));
         discovery.setPorts(discoveryPorts);
         discovery.setVisibility(Protos.DiscoveryInfo.Visibility.EXTERNAL);
         discovery.setName(configuration.getTaskName());
         return discovery.build();
     }
 
+    private Protos.Labels getLabels(Configuration configuration) {
+      Protos.Labels.Builder builder = Protos.Labels.newBuilder();
+      Map<String, String> labels = configuration.getTaskLabels();
+      for (Map.Entry<String, String> kvp : labels.entrySet()) {
+        Protos.Label label = Protos.Label.newBuilder()
+          .setKey(kvp.getKey())
+          .setValue(kvp.getValue())
+          .build();
+
+        builder.addLabels(label);
+      }
+
+      return builder.build();
+    }
+
     private Protos.ContainerInfo getContainer(Configuration configuration, Protos.TaskID taskID, Long elasticSearchNodeId, Protos.SlaveID slaveID) {
         final Protos.Environment environment = Protos.Environment.newBuilder().addAllVariables(new ExecutorEnvironmentalVariables(configuration, elasticSearchNodeId).getList()).build();
         final Protos.ContainerInfo.DockerInfo.Builder dockerInfo = Protos.ContainerInfo.DockerInfo.newBuilder()

scheduler/src/test/java/org/apache/mesos/elasticsearch/scheduler/ConfigurationTest.java

@@ -11,6 +11,7 @@
 import java.net.UnknownHostException;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 import static org.junit.Assert.*;
 
@@ -78,4 +79,20 @@ public void shouldCreateVolumeName() {
         Configuration configuration = new Configuration(ZookeeperCLIParameter.ZOOKEEPER_MESOS_URL, "aa", Configuration.FRAMEWORK_NAME, "test");
         assertEquals("test0data", configuration.dataVolumeName(0L));
     }
-}
+
+    @Test
+    public void shouldCreateTaskLabels() {
+        Configuration configuration = new Configuration(
+            ZookeeperCLIParameter.ZOOKEEPER_MESOS_URL, "aa", Configuration.EXECUTOR_LABELS,
+              "foo=bar",
+              "incomplete",
+              "empty=",
+              "separator=values=are=joined");
+        Map<String, String> labels = configuration.getTaskLabels();
+
+        assertEquals("bar", labels.get("foo"));
+        assertFalse(labels.containsKey("incomplete"));
+        assertEquals("", labels.get("empty"));
+        assertEquals("values=are=joined", labels.get("separator"));
+    }
+}