Refactor/241 executor cli

commons/build.gradle

@@ -1 +1,5 @@
-apply plugin: 'java'
+apply plugin: 'java'
+
+dependencies {
+    compile "com.beust:jcommander:1.48"
+}

commons/src/main/java/org/apache/mesos/elasticsearch/common/cli/validators/CLIValidators.java

@@ -0,0 +1,68 @@
+package org.apache.mesos.elasticsearch.common.cli.validators;
+
+import com.beust.jcommander.IParameterValidator;
+import com.beust.jcommander.IValueValidator;
+import com.beust.jcommander.ParameterException;
+
+/**
+ * Holds CLI validators
+ */
+public class CLIValidators {
+
+    /**
+     * Abstract class to validate a number.
+     * @param <T> A numeric type
+     */
+    public abstract static class PositiveValue<T> implements IValueValidator<T> {
+        @Override
+        public void validate(String name, T value) throws ParameterException {
+            if (notValid(value)) {
+                throw new ParameterException("Parameter " + name + " should be greater than zero (found " + value + ")");
+            }
+        }
+
+        public abstract Boolean notValid(T value);
+    }
+
+    /**
+     * Validates a positive number. For type Long
+     */
+    public static class PositiveLong extends PositiveValue<Long> {
+        @Override
+        public Boolean notValid(Long value) {
+            return value <= 0;
+        }
+    }
+
+    /**
+     * Validates a positive number. For type Double
+     */
+    public static class PositiveDouble extends PositiveValue<Double> {
+        @Override
+        public Boolean notValid(Double value) {
+            return value <= 0;
+        }
+    }
+
+    /**
+     * Validates a positive number. For type Integer
+     */
+    public static class PositiveInteger extends PositiveValue<Integer> {
+        @Override
+        public Boolean notValid(Integer value) {
+            return value <= 0;
+        }
+    }
+
+    /**
+     * Ensures that the string is not empty. Will strip spaces.
+     */
+    public static class NotEmptyString implements IParameterValidator {
+        @Override
+        public void validate(String name, String value) throws ParameterException {
+            if (value.replace(" ", "").isEmpty()) {
+                throw new ParameterException("Parameter " + name + " cannot be empty");
+            }
+        }
+    }
+}

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

@@ -0,0 +1,23 @@
+package org.apache.mesos.elasticsearch.common.zookeeper;
+
+import com.beust.jcommander.Parameter;
+import org.apache.mesos.elasticsearch.common.cli.validators.CLIValidators;
+
+/**
+ * Class to reuse ZooKeeper CLI Parameters
+ */
+public class ZookeeperCLIParameter {
+    public static final String ZOOKEEPER_URL = "--zookeeperUrl";
+    @Parameter(names = {ZOOKEEPER_URL}, required = true, description = "Zookeeper urls in the format zk://IP:PORT,IP:PORT,...)", validateWith = CLIValidators.NotEmptyString.class)
+    private String zookeeperUrl = "zk://mesos.master:2181";
+    public String getZookeeperUrl() {
+        return zookeeperUrl;
+    }
+
+    public static final String ZOOKEEPER_TIMEOUT = "--zookeeperTimeout";
+    @Parameter(names = {ZOOKEEPER_TIMEOUT}, description = "The timeout for connecting to zookeeper (ms).", validateValueWith = CLIValidators.PositiveLong.class)
+    private long zookeeperTimeout = 20000L;
+    public long getZookeeperTimeout() {
+        return zookeeperTimeout;
+    }
+}

executor/build.gradle

@@ -29,6 +29,7 @@ dependencies {
     compile "log4j:log4j:1.2.16"
     compile "org.apache.zookeeper:zookeeper:3.4.6"
     compile "com.github.containersolutions:elasticsearch-zookeeper:v${elasticsearchVersion}"
+    compile "com.beust:jcommander:1.48"
 }
 
 shadowJar {

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

@@ -0,0 +1,58 @@
+package org.apache.mesos.elasticsearch.executor;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import org.apache.log4j.Logger;
+import org.apache.mesos.elasticsearch.common.cli.validators.CLIValidators;
+import org.apache.mesos.elasticsearch.common.zookeeper.ZookeeperCLIParameter;
+import org.apache.mesos.elasticsearch.common.zookeeper.formatter.ElasticsearchZKFormatter;
+import org.apache.mesos.elasticsearch.common.zookeeper.parser.ZKAddressParser;
+
+import java.net.URISyntaxException;
+
+/**
+ * Executor configuration
+ */
+public class Configuration {
+    private static final Logger LOGGER = Logger.getLogger(Configuration.class);
+    public static final String ELASTICSEARCH_YML = "elasticsearch.yml";
+
+    // **** ZOOKEEPER
+    private final ZookeeperCLIParameter zookeeperCLI = new ZookeeperCLIParameter();
+
+    public Configuration(String[] args) {
+        final JCommander jCommander = new JCommander();
+        jCommander.addObject(zookeeperCLI);
+        jCommander.addObject(this);
+        try {
+            jCommander.parse(args); // Parse command line args into configuration class.
+        } catch (com.beust.jcommander.ParameterException ex) {
+            System.out.println(ex);
+            jCommander.setProgramName("(Options preceded by an asterisk are required)");
+            jCommander.usage();
+            throw ex;
+        }
+    }
+
+    // ******* ELASTICSEARCH
+    public static final String ELASTICSEARCH_SETTINGS_LOCATION = "--elasticsearchSettingsLocation";
+    @Parameter(names = {ELASTICSEARCH_SETTINGS_LOCATION}, description = "Local path to custom elasticsearch.yml settings file", validateWith = CLIValidators.NotEmptyString.class)
+    private String elasticsearchSettingsLocation = getElasticsearchSettingsPath();
+    public String getElasticsearchSettingsLocation() {
+        return elasticsearchSettingsLocation;
+    }
+    private String getElasticsearchSettingsPath() {
+        String path = "";
+        try {
+            path = getClass().getClassLoader().getResource(ELASTICSEARCH_YML).toURI().toString();
+        } catch (NullPointerException | URISyntaxException ex) {
+            LOGGER.error("Unable to read default settings file from resources", ex);
+        }
+        return path;
+    }
+
+    public String getElasticsearchZKURL() {
+        ElasticsearchZKFormatter zkFormatter = new ElasticsearchZKFormatter(new ZKAddressParser());
+        return zkFormatter.format(zookeeperCLI.getZookeeperUrl());
+    }
+}

executor/src/main/java/org/apache/mesos/elasticsearch/executor/elasticsearch/ElasticsearchSettings.java

@@ -6,15 +6,8 @@
  * Builds the ES settings from the provided settings.
  */
 public class ElasticsearchSettings {
-    // Todo: Make ES settings, settings.
     public ImmutableSettings.Builder defaultSettings() {
         return ImmutableSettings.settingsBuilder()
-                .put("node.local", false)
-                .put("cluster.name", "mesos-elasticsearch")
-                .put("node.master", true)
-                .put("node.data", true)
-                .put("index.number_of_shards", 5)
-                .put("index.number_of_replicas", 1)
                 .put("discovery.type", "com.sonian.elasticsearch.zookeeper.discovery.ZooKeeperDiscoveryModule")
                 .put("sonian.elasticsearch.zookeeper.settings.enabled", true)
                 .put("sonian.elasticsearch.zookeeper.discovery.state_publishing.enabled", true);

executor/src/main/java/org/apache/mesos/elasticsearch/executor/mesos/ElasticsearchExecutor.java

@@ -5,14 +5,19 @@
 import org.apache.mesos.Executor;
 import org.apache.mesos.ExecutorDriver;
 import org.apache.mesos.Protos;
+import org.apache.mesos.elasticsearch.executor.Configuration;
 import org.apache.mesos.elasticsearch.executor.elasticsearch.Launcher;
 import org.apache.mesos.elasticsearch.executor.model.PortsModel;
 import org.apache.mesos.elasticsearch.executor.model.RunTimeSettings;
 import org.apache.mesos.elasticsearch.executor.model.ZooKeeperModel;
+import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.node.Node;
 
+import java.net.MalformedURLException;
+import java.net.URL;
 import java.security.InvalidParameterException;
 import java.util.Arrays;
+import java.util.List;
 
 /**
  * Executor for Elasticsearch.
@@ -21,6 +26,7 @@ public class ElasticsearchExecutor implements Executor {
     private final Launcher launcher;
     public static final Logger LOGGER = Logger.getLogger(ElasticsearchExecutor.class.getCanonicalName());
     private final TaskStatus taskStatus;
+    private Configuration configuration;
 
     public ElasticsearchExecutor(Launcher launcher, TaskStatus taskStatus) {
         this.launcher = launcher;
@@ -54,12 +60,24 @@ public void launchTask(final ExecutorDriver driver, final Protos.TaskInfo task)
         driver.sendStatusUpdate(taskStatus.starting());
 
         try {
+            // Parse CommandInfo arguments
+            List<String> list = task.getExecutor().getCommand().getArgumentsList();
+            String[] args = list.toArray(new String[list.size()]);
+            LOGGER.debug("Using arguments: " + Arrays.toString(args));
+            configuration = new Configuration(args);
+
+            // Add settings provided in es Settings file
+            URL elasticsearchSettingsPath = java.net.URI.create(configuration.getElasticsearchSettingsLocation()).toURL();
+            LOGGER.debug("Using elasticsearch settings file: " + elasticsearchSettingsPath);
+            ImmutableSettings.Builder esSettings = ImmutableSettings.builder().loadFromUrl(elasticsearchSettingsPath);
+            launcher.addRuntimeSettings(esSettings);
+
             // Parse ports
             RunTimeSettings ports = new PortsModel(task);
             launcher.addRuntimeSettings(ports.getRuntimeSettings());
 
             // Parse ZooKeeper address
-            RunTimeSettings zk = new ZooKeeperModel(task);
+            RunTimeSettings zk = new ZooKeeperModel(configuration.getElasticsearchZKURL());
             launcher.addRuntimeSettings(zk.getRuntimeSettings());
 
             // Launch Node
@@ -76,7 +94,7 @@ public void run() {
 
             // Send status update, running
             driver.sendStatusUpdate(taskStatus.running());
-        } catch (InvalidParameterException e) {
+        } catch (InvalidParameterException | MalformedURLException e) {
             driver.sendStatusUpdate(taskStatus.failed());
             LOGGER.error(e);
         }

executor/src/main/java/org/apache/mesos/elasticsearch/executor/model/ZooKeeperModel.java

@@ -1,20 +1,16 @@
 package org.apache.mesos.elasticsearch.executor.model;
 
-import org.apache.mesos.Protos;
-import org.apache.mesos.elasticsearch.executor.parser.ParseZooKeeper;
-import org.apache.mesos.elasticsearch.executor.parser.TaskParser;
 import org.elasticsearch.common.settings.ImmutableSettings;
 
 /**
  * Model representing ZooKeeper information
  */
 public class ZooKeeperModel implements RunTimeSettings {
     public static final String ZOOKEEPER_ADDRESS_KEY = "sonian.elasticsearch.zookeeper.client.host";
-    private final TaskParser<String> parser = new ParseZooKeeper();
     private final String address;
 
-    public ZooKeeperModel(Protos.TaskInfo taskInfo) {
-        address = parser.parse(taskInfo);
+    public ZooKeeperModel(String address) {
+        this.address = address;
     }
 
     private ImmutableSettings.Builder getAddress() {