[service_discovery] enabling service discovery via named pipe. (#113)

* [service_discovery] enabling service discovery via named pipe.

[service_discovery] fixing configuration.

[service_discovery] store SD configs in its own cache.

[service_discovery] use named pipes to receive service discovery configs.

[service_discovery] parsing pipe input into config.

[service_discovery] move relevant options to AppConfig.

[service_discovery] fixing up service discovery config traversal.

[service_discovery] only remove config from list if not a service discovery received config.

[logging] prettify logging statement.

* [service_discovery] making Java 6 compatible.

* [service_discovery] cleanup.

* [jmxfetch] camelcase those vars...

* [jmxfetch][service_discovery] adding service discovery configuration test.

* [jmxfetch][service_discovery] regex lookbehind was seriously broken in older java 1.6.

[jmxfetch][service_discovery] fixing regex to be more Java6 friendly.

* [travis-ci] we should probably start testing against jdk8.

* [jmxfetch][skip ci] change scope of mockito dependency to test only.

Author: truthbk

.travis.yml

@@ -1,5 +1,6 @@
 language: java
 jdk:
+- oraclejdk8
 - oraclejdk7
 - openjdk7
 - openjdk6

pom.xml

@@ -14,11 +14,13 @@
 	<properties>
 		<commons-io.version>2.4</commons-io.version>
 		<commons-lang.version>2.6</commons-lang.version>
+		<apache-commons-lang3.version>3.5</apache-commons-lang3.version>
 		<guava.version>17.0</guava.version>
 		<java-dogstatsd-client.version>2.1.0</java-dogstatsd-client.version>
 		<jcommander.version>1.35</jcommander.version>
 		<junit.version>4.11</junit.version>
 		<log4j.version>1.2.17</log4j.version>
+        <mockito.version>2.2.27</mockito.version>
 		<maven-surefire-plugin.version>2.9</maven-surefire-plugin.version>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<snakeyaml.version>1.13</snakeyaml.version>
@@ -59,7 +61,13 @@
 			<groupId>commons-lang</groupId>
 			<artifactId>commons-lang</artifactId>
 			<version>${commons-lang.version}</version>
-		</dependency>
+        </dependency>
+        <dependency>
+            <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+			<version>${apache-commons-lang3.version}</version>
+        </dependency>
 		<dependency>
 			<groupId>com.datadoghq</groupId>
 			<artifactId>java-dogstatsd-client</artifactId>
@@ -95,6 +103,12 @@
 			<artifactId>snakeyaml</artifactId>
 			<version>${snakeyaml.version}</version>
 		</dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <version>${mockito.version}</version>
+			<scope>test</scope>
+        </dependency>
 	</dependencies>
 	<scm>
 		<connection>scm:git:git@github.com:Datadog/jmxfetch.git</connection>

src/main/java/org/datadog/jmxfetch/App.java

@@ -1,12 +1,21 @@
 package org.datadog.jmxfetch;
 
+import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.FileFilter;
 import java.io.FileNotFoundException;
+import java.io.InputStream;
 import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Arrays;
 import java.util.ArrayList;
 import java.util.Enumeration;
 import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.regex.Pattern;
+import java.util.regex.Matcher;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
@@ -19,6 +28,7 @@
 import org.apache.log4j.Appender;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
+import org.apache.commons.lang3.CharEncoding;
 import org.datadog.jmxfetch.reporter.Reporter;
 import org.datadog.jmxfetch.util.CustomLogger;
 
@@ -29,9 +39,13 @@
 @SuppressWarnings("unchecked")
 public class App {
     private final static Logger LOGGER = Logger.getLogger(App.class.getName());
+    private final static String SERVICE_DISCOVERY_PREFIX = "SD-";
     public static final String CANNOT_CONNECT_TO_INSTANCE = "Cannot connect to instance ";
+    private static final String SD_CONFIG_SEP = "#### SERVICE-DISCOVERY ####";
     private static int loopCounter;
-    private HashMap<String, YamlParser> configs;
+    private AtomicBoolean reinit = new AtomicBoolean(false);
+    private ConcurrentHashMap<String, YamlParser> configs;
+    private ConcurrentHashMap<String, YamlParser> sdConfigs = new ConcurrentHashMap<String, YamlParser>();
     private ArrayList<Instance> instances = new ArrayList<Instance>();
     private LinkedList<Instance> brokenInstances = new LinkedList<Instance>();
     private AppConfig appConfig;
@@ -132,6 +146,10 @@ public void run() {
         new ShutdownHook().attachShutDownHook();
     }
 
+    public void setReinit(boolean reinit) {
+        this.reinit.set(reinit);
+    }
+
     public static int getLoopCounter() {
         return loopCounter;
     }
@@ -145,16 +163,84 @@ private static void clearInstances(List<Instance> instances) {
         }
     }
 
+    private String getSDName(String config){
+      String[] splitted = config.split(System.getProperty("line.separator"), 2);
+
+      return SERVICE_DISCOVERY_PREFIX + splitted[0].substring(2, splitted[0].length());
+    }
+
+    public boolean processServiceDiscovery(byte[] buffer) {
+      boolean reinit = false;
+      String[] discovered;
+
+      try {
+        String configs = new String(buffer, CharEncoding.UTF_8);
+        discovered = configs.split(App.SD_CONFIG_SEP + System.getProperty("line.separator"));
+      } catch(UnsupportedEncodingException e) {
+        LOGGER.debug("Unable to parse byte buffer to UTF-8 String.");
+        return false;
+      }
+
+      for (String config : discovered) {
+        if (config == null || config.isEmpty()) {
+          continue;
+        }
+
+        try{
+          String name = getSDName(config);
+          LOGGER.debug("Attempting to apply config. Name: " + name + "\nconfig: \n" + config);
+          InputStream stream = new ByteArrayInputStream(config.getBytes(CharEncoding.UTF_8));
+          YamlParser yaml = new YamlParser(stream);
+
+          if (this.addConfig(name, yaml)){
+            reinit = true;
+            LOGGER.debug("Configuration added succesfully reinit in order");
+          }
+        } catch(UnsupportedEncodingException e) {
+          LOGGER.debug("Unable to parse byte buffer to UTF-8 String.");
+        }
+      }
+
+      return reinit;
+    }
+
     void start() {
         // Main Loop that will periodically collect metrics from the JMX Server
+        long start_ms = System.currentTimeMillis();
+        long delta_s = 0;
+        FileInputStream sdPipe = null;
+
+        try {
+          sdPipe = new FileInputStream(appConfig.getServiceDiscoveryPipe()); //Should we use RandomAccessFile?
+        } catch (FileNotFoundException e) {
+          LOGGER.warn("Unable to open named pipe - Service Discovery disabled.");
+          sdPipe = null;
+        }
+
         while (true) {
-            // Exit on exit file trigger
+            // Exit on exit file trigger...
             if (appConfig.getExitWatcher().shouldExit()){
                 LOGGER.info("Exit file detected: stopping JMXFetch.");
                 System.exit(0);
             }
 
+            // any SD configs waiting in pipe?
+            try {
+              if(sdPipe != null && sdPipe.available() > 0) {
+                int len = sdPipe.available();
+                byte[] buffer = new byte[len];
+                sdPipe.read(buffer);
+                setReinit(processServiceDiscovery(buffer));
+              }
+            } catch(IOException e) {
+              LOGGER.warn("Unable to read from pipe - Service Discovery configuration may have been skipped.");
+            }
+
             long start = System.currentTimeMillis();
+            if (this.reinit.get()) {
+                init(true);
+            }
+
             if (instances.size() > 0) {
                 doIteration();
             } else {
@@ -279,10 +365,39 @@ public void doIteration() {
         }
     }
 
-    private HashMap<String, YamlParser> getConfigs(AppConfig config) {
-        HashMap<String, YamlParser> configs = new HashMap<String, YamlParser>();
+    public boolean addConfig(String name, YamlParser config) {
+        // named groups not supported with Java6: "(?<check>.{1,30})_(?<version>\\d{0,30})"
+        Pattern pattern = Pattern.compile(SERVICE_DISCOVERY_PREFIX+"(.{1,30})_(\\d{0,30})");
+
+        Matcher matcher = pattern.matcher(name);
+        if (!matcher.find()) {
+            // bad name.
+            return false;
+        }
+
+        // Java 6 doesn't allow name matching - group 1 is "check"
+        String check = matcher.group(1);
+        if (this.configs.containsKey(check)) {
+            // there was already a file config for the check.
+            return false;
+        }
+
+        this.sdConfigs.put(name, config);
+        this.setReinit(true);
+
+        return true;
+    }
+
+    private ConcurrentHashMap<String, YamlParser> getConfigs(AppConfig config) {
+        ConcurrentHashMap<String, YamlParser> configs = new ConcurrentHashMap<String, YamlParser>();
         YamlParser fileConfig;
-        for (String fileName : config.getYamlFileList()) {
+
+        List<String> fileList = config.getYamlFileList();
+        if (fileList == null) {
+            return configs;
+        }
+
+        for (String fileName : fileList) {
             File f = new File(config.getConfdDirectory(), fileName);
             String name = f.getName().replace(".yaml", "");
             FileInputStream yamlInputStream = null;
@@ -306,6 +421,7 @@ private HashMap<String, YamlParser> getConfigs(AppConfig config) {
                 }
             }
         }
+
         LOGGER.info("Found " + configs.size() + " config files");
         return configs;
     }
@@ -336,11 +452,23 @@ public void init(boolean forceNewConnection) {
 
 
         Iterator<Entry<String, YamlParser>> it = configs.entrySet().iterator();
-        while (it.hasNext()) {
-            Map.Entry<String, YamlParser> entry = it.next();
+        // SD config cache doesn't remove configs - it just overwrites.
+        Iterator<Entry<String, YamlParser>> itSD = sdConfigs.entrySet().iterator();
+        while (it.hasNext() || itSD.hasNext()) {
+            Map.Entry<String, YamlParser> entry;
+            boolean sdIterator = false;
+            if (it.hasNext()) {
+                entry = it.next();
+            } else {
+                entry = itSD.next();
+                sdIterator = true;
+            }
+
             String name = entry.getKey();
             YamlParser yamlConfig = entry.getValue();
-            it.remove();
+            if(!sdIterator) {
+                it.remove();
+            }
 
             ArrayList<LinkedHashMap<String, Object>> configInstances = ((ArrayList<LinkedHashMap<String, Object>>) yamlConfig.getYamlInstances());
             if (configInstances == null || configInstances.size() == 0) {

src/main/java/org/datadog/jmxfetch/AppConfig.java

@@ -28,6 +28,9 @@ class AppConfig {
     public static final HashSet<String> ACTIONS = new HashSet<String>(Arrays.asList(ACTION_COLLECT, ACTION_LIST_EVERYTHING,
             ACTION_LIST_COLLECTED, ACTION_LIST_MATCHING, ACTION_LIST_NOT_MATCHING, ACTION_LIST_LIMITED, ACTION_HELP, ACTION_LIST_JVMS));
 
+    private static final String SD_WIN_PIPE_PATH = "\\\\.\\pipe\\";
+    private static final String SD_PIPE_NAME = "dd-service_discovery";
+
     @Parameter(names = {"--help", "-h"},
             description = "Display this help page",
             help = true)
@@ -49,6 +52,11 @@ class AppConfig {
             required = true)
     private String confdDirectory;
 
+    @Parameter(names = {"--tmp_directory", "-T"},
+            description = "Absolute path to a temporary directory",
+            required = false)
+    private String tmpDirectory = "/tmp";
+
     @Parameter(names = {"--reporter", "-r"},
             description = "Reporter to use: should be either \"statsd:[STATSD_PORT]\" or \"console\"",
             validateWith = ReporterValidator.class,
@@ -58,7 +66,7 @@ class AppConfig {
 
     @Parameter(names = {"--check", "-c"},
             description = "Yaml file name to read (must be in the confd directory)",
-            required = true,
+            required = false,
             variableArity = true)
     private List<String> yamlFileList;
 
@@ -68,6 +76,16 @@ class AppConfig {
             required = false)
     private int checkPeriod = 15000;
 
+    @Parameter(names = {"--sd_standby", "-w"},
+            description = "Service Discovery standby.",
+            required = false)
+    private boolean sdStandby = false;
+
+    @Parameter(names = {"--sd_pipe", "-S"},
+            description = "Service Discovery pipe name.",
+            required = false)
+    private String sdPipe = SD_PIPE_NAME;
+
     @Parameter(names = {"--status_location", "-s"},
             description = "Absolute path of the status file. (default to null = no status file written)",
             converter = StatusConverter.class,
@@ -110,6 +128,10 @@ public int getCheckPeriod() {
         return checkPeriod;
     }
 
+    public boolean getSDStandby() {
+        return sdStandby;
+    }
+
     public Reporter getReporter() {
         return reporter;
     }
@@ -122,11 +144,26 @@ public String getConfdDirectory() {
         return confdDirectory;
     }
 
+    public String getTmpDirectory() {
+        return tmpDirectory;
+    }
+
     public String getLogLevel() {
         return logLevel;
     }
 
     public String getLogLocation() {
         return logLocation;
     }
+
+    public String getServiceDiscoveryPipe() {
+        String pipePath;
+
+        if (System.getProperty("os.name").startsWith("Windows")) {
+            pipePath = SD_WIN_PIPE_PATH + "/" + sdPipe;
+        } else {
+            pipePath = getTmpDirectory() + "/" + sdPipe;
+        }
+        return pipePath;
+    }
 }