Enable configuration of monitoring sinks. (#219)

* Enable configuration of monitoring sinks.

* Addressed PR feedback.

* Restored Collectd record TLB.

Author: vjkoskela

config/config.conf

@@ -13,8 +13,30 @@ httpPort=7090
 # Metrics
 # ~~~~
 monitoringCluster="mad"
+#monitoringService="mad"
+#monitoringSinks=[
+#  {
+#    class = "com.arpnetworking.metrics.impl.ApacheHttpSink"
+#    bufferSize = 10000
+#    uri = "http://localhost:7090/metrics/v3/application"
+#    parallelism = 2
+#    maxBatchSize = 500
+#    emptyQueueInterval = "PT0.5S"
+#    eventsDroppedLoggingInterval = "PT1M"
+#    dispatchErrorLoggingInterval = "PT1M"
+#    unsupportedDataLoggingInterval = "PT1M"
+#  }
+#]
 #jvmMetricsCollectionInterval="PT1.0S"
 
+# NOTES:
+# - metricsClientHost and metricsClientPort are deprecated
+# - metricsClientHost and metricsClientPort no longer default from httpHost and httpPort
+# - specifying metricsClientHost or metricsClientPort disables monitoringSinks
+# - for backwards compatibility the legacy fields metricsClientHost and metricsClientPort
+#   take precedence over monitoringSinks
+# - migrate to monitoringSinks as soon as possible
+
 # Logging
 # ~~~~
 logDirectory="logs"

jdk-wrapper.sh

@@ -17,6 +17,12 @@
 # For documentation please refer to:
 # https://github.com/KoskiLabs/jdk-wrapper/blob/master/README.md
 
+HTTP_PROTOCOL="http"
+FILE_PROTOCOL="file"
+
+LATEST_RELEASE="latest"
+SNAPSHOT_RELEASE="snapshot"
+
 log_err() {
   l_prefix=$(date  +'%H:%M:%S')
   printf "[%s] %s\\n" "${l_prefix}" "$@" 1>&2;
@@ -42,37 +48,63 @@ safe_command() {
 
 checksum() {
   l_file="$1"
-  checksum_exec=""
-  checksum_args=""
+  l_checksum_exec=""
+  l_checksum_args=""
   if command -v sha256sum > /dev/null; then
-    checksum_exec="sha256sum"
+    l_checksum_exec="sha256sum"
   elif command -v shasum > /dev/null; then
-    checksum_exec="shasum"
-    checksum_args="-a 256"
+    l_checksum_exec="shasum"
+    l_checksum_args="-a 256"
   elif command -v sha1sum > /dev/null; then
-    checksum_exec="sha1sum"
+    l_checksum_exec="sha1sum"
   elif command -v md5 > /dev/null; then
-    checksum_exec="md5"
+    l_checksum_exec="md5"
   fi
-  if [ -z "${checksum_exec}" ]; then
+  if [ -z "${l_checksum_exec}" ]; then
     log_err "ERROR: No supported checksum command found!"
     exit 1
   fi
-  ${checksum_exec} ${checksum_args} < "${l_file}"
+  ${l_checksum_exec} ${l_checksum_args} < "${l_file}"
 }
 
 rand() {
   awk 'BEGIN {srand();printf "%d\n", (rand() * 10^8);}'
 }
 
-download_if_needed() {
-  file="$1"
-  path="$2"
-  if [ ! -f "${path}/${file}" ]; then
-    jdkw_url="${JDKW_BASE_URI}/releases/download/${JDKW_RELEASE}/${file}"
-    log_out "Downloading ${file} from ${jdkw_url}"
-    safe_command "curl ${curl_options} -f -k -L -o \"${path}/${file}\" \"${jdkw_url}\""
-    safe_command "chmod +x \"${path}/${file}\""
+get_protocol() {
+  case "${JDKW_BASE_URI}" in
+  http://*|https://*)
+    printf "%s" "${HTTP_PROTOCOL}"
+    ;;
+  file://*)
+    printf "%s" "${FILE_PROTOCOL}"
+    ;;
+  *)
+    log_err "ERROR: Unsupported protocol in JDKW_BASE_URI: ${JDKW_BASE_URI}"
+    exit 1
+  esac
+}
+
+obtain_if_needed() {
+  l_file="$1"
+  l_target_path="$2"
+  if [ ! -f "${l_target_path}/${l_file}" ]; then
+    case "${JDKW_BASE_URI}" in
+    http://*|https://*)
+      l_jdkw_url="${JDKW_BASE_URI}/releases/download/${JDKW_RELEASE}/${l_file}"
+      log_out "Downloading ${l_file} from ${l_jdkw_url}"
+      safe_command "curl ${curl_options} -f -k -L -o \"${l_target_path}/${l_file}\" \"${l_jdkw_url}\""
+      ;;
+    file://*)
+      l_jdkw_path="${JDKW_BASE_URI#file://}/${l_file}"
+      log_out "Copying ${l_file} from ${l_jdkw_path}"
+      safe_command "cp \"${l_jdkw_path}\" \"${l_target_path}/${l_file}\""
+      ;;
+    *)
+      log_err "ERROR: Unsupported protocol in JDKW_BASE_URI: ${JDKW_BASE_URI}"
+      exit 1
+    esac
+    safe_command "chmod +x \"${l_target_path}/${l_file}\""
   fi
 }
 
@@ -98,7 +130,6 @@ done < "${l_fifo}"
 safe_command "rm \"${l_fifo}\""
 
 # Process (but do not load) properties from command line arguments
-command=
 cmd_configuration=
 for arg in "$@"; do
   jdkw_arg=$(echo "${arg}" | grep '^JDKW_.*')
@@ -109,13 +140,6 @@ for arg in "$@"; do
   if [ -n "${jdkw_arg}" ]; then
     cmd_configuration="${cmd_configuration}${arg} "
   fi
-  case "${arg}" in
-    *\'*)
-       arg=$(printf "%s" "$arg" | sed "s/'/'\"'\"'/g")
-       ;;
-    *) : ;;
-  esac
-  command="${command} '${arg}'"
 done
 
 # Default base directory to current working directory
@@ -144,7 +168,10 @@ if [ -z "${JDKW_BASE_URI}" ]; then
     JDKW_BASE_URI="https://github.com/KoskiLabs/jdk-wrapper"
 fi
 if [ -z "${JDKW_RELEASE}" ]; then
-  JDKW_RELEASE="latest"
+  JDKW_RELEASE="${LATEST_RELEASE}"
+  if [ $(get_protocol) = "${FILE_PROTOCOL}" ]; then
+    JDKW_RELEASE="${SNAPSHOT_RELEASE}"
+  fi
   log_out "Defaulted to version ${JDKW_RELEASE}"
 fi
 if [ -z "${JDKW_TARGET}" ]; then
@@ -156,7 +183,7 @@ if [ -z "${JDKW_VERBOSE}" ]; then
 fi
 
 # Resolve latest version
-if [ "${JDKW_RELEASE}" = "latest" ]; then
+if [ "${JDKW_RELEASE}" = "${LATEST_RELEASE}" ]; then
   latest_version_json="${TMPDIR:-/tmp}/jdkw-latest-version-$$.$(rand)"
   safe_command "curl ${curl_options} -f -k -L -o \"${latest_version_json}\" -H 'Accept: application/json' \"${JDKW_BASE_URI}/releases/latest\""
   JDKW_RELEASE=$(sed -e 's/.*"tag_name":"\([^"]*\)".*/\1/' < "${latest_version_json}")
@@ -166,6 +193,10 @@ fi
 
 # Ensure target directory exists
 jdkw_path="${JDKW_TARGET}/jdkw/${JDKW_RELEASE}"
+if [ -d "${jdkw_path}" ] && [ "${JDKW_RELEASE}" = "${SNAPSHOT_RELEASE}" ]; then
+  log_out "Removing target snapshot directory ${jdkw_path}"
+  safe_command "rm -rf \"${jdkw_path}\""
+fi
 if [ ! -d "${jdkw_path}" ]; then
   log_out "Creating target directory ${jdkw_path}"
   safe_command "mkdir -p \"${jdkw_path}\""
@@ -174,8 +205,8 @@ fi
 # Download the jdk wrapper version
 jdkw_impl="jdkw-impl.sh"
 jdkw_wrapper="jdk-wrapper.sh"
-download_if_needed "${jdkw_impl}" "${jdkw_path}"
-download_if_needed "${jdkw_wrapper}" "${jdkw_path}"
+obtain_if_needed "${jdkw_impl}" "${jdkw_path}"
+obtain_if_needed "${jdkw_wrapper}" "${jdkw_path}"
 
 # Check whether this wrapper is the one specified for this version
 jdkw_download="${jdkw_path}/${jdkw_wrapper}"

pom.xml

@@ -92,10 +92,10 @@
     <log4j.over.slf4j.version>1.7.25</log4j.over.slf4j.version>
     <metrics.aggregator.protocol.prometheus.version>1.0.0</metrics.aggregator.protocol.prometheus.version>
     <metrics.aggregator.protocol.version>1.0.10</metrics.aggregator.protocol.version>
-    <metrics.client.version>0.11.1</metrics.client.version>
-    <metrics.client.http.version>0.11.1</metrics.client.http.version>
-    <metrics.client.incubator.version>0.11.0</metrics.client.incubator.version>
-    <metrics.jvm.extra.version>0.11.0</metrics.jvm.extra.version>
+    <metrics.client.version>0.11.2</metrics.client.version>
+    <metrics.client.http.version>0.11.2</metrics.client.http.version>
+    <metrics.client.incubator.version>0.11.2</metrics.client.incubator.version>
+    <metrics.jvm.extra.version>0.11.2</metrics.jvm.extra.version>
     <oval.version>1.90</oval.version>
     <protobuf.version>3.8.0</protobuf.version>
     <scala.version>2.11</scala.version>

src/main/java/com/arpnetworking/metrics/mad/Main.java

@@ -57,6 +57,7 @@
 import com.arpnetworking.utility.Configurator;
 import com.arpnetworking.utility.Launchable;
 import com.arpnetworking.utility.ScalaForkJoinPoolAdapter;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
@@ -75,7 +76,7 @@
 
 import java.io.File;
 import java.net.URI;
-import java.util.Collections;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -88,6 +89,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
 
 /**
  * Class containing entry point for Metrics Data Aggregator (MAD).
@@ -264,6 +266,7 @@ private void launchActors(final Injector injector) {
                 .run(materializer);
     }
 
+    @SuppressWarnings("deprecation")
     private Injector launchGuice(final ActorSystem actorSystem) {
         LOGGER.info().setMessage("Launching guice").log();
 
@@ -282,17 +285,28 @@ private Injector launchGuice(final ActorSystem actorSystem) {
         }
 
         // Instantiate the metrics factory
-        final URI sinkUrl = URI.create(
-                "http://" + _configuration.getMetricsClientHost() + ":"
-                        + _configuration.getMetricsClientPort() + "/metrics/v3/application");
+        final ImmutableList.Builder<com.arpnetworking.metrics.Sink> monitoringSinksBuilder =
+                new ImmutableList.Builder<>();
+        if (_configuration.getMetricsClientHost().isPresent()
+                || _configuration.getMetricsClientPort().isPresent()) {
+            final String endpoint = String.format(
+                    "http://%s:%d/metrics/v3/application",
+                    _configuration.getMetricsClientHost().orElse("localhost"),
+                    _configuration.getMetricsClientPort().orElse(7090));
+
+            monitoringSinksBuilder.add(
+                    new ApacheHttpSink.Builder()
+                            .setUri(URI.create(endpoint))
+                            .build());
+
+        } else {
+            monitoringSinksBuilder.addAll(createSinks(_configuration.getMonitoringSinks()));
+        }
+
         final MetricsFactory metricsFactory = new TsdMetricsFactory.Builder()
                 .setClusterName(_configuration.getMonitoringCluster())
                 .setServiceName(_configuration.getMonitoringService())
-                .setSinks(
-                        Collections.singletonList(
-                                new ApacheHttpSink.Builder()
-                                        .setUri(sinkUrl)
-                                        .build()))
+                .setSinks(monitoringSinksBuilder.build())
                 .build();
 
         final AppShutdown shutdown = new AppShutdown();
@@ -302,6 +316,36 @@ private Injector launchGuice(final ActorSystem actorSystem) {
         return Guice.createInjector(new MainModule(actorSystem, metricsFactory, shutdown));
     }
 
+    @SuppressFBWarnings("REC_CATCH_EXCEPTION")
+    static List<com.arpnetworking.metrics.Sink> createSinks(
+            final ImmutableList<JsonNode> monitoringSinks) {
+        // Until we implement the Commons Builder pattern in the metrics client
+        // library we need to resort to a more brute-force deserialization
+        // style. The benefit of this approach is that it will be forwards
+        // compatible with the Commons Builder approach. The drawbacks are
+        // the ugly way the configuration is passed around (as JsonNode) and
+        // then two-step deserialized.
+        final List<com.arpnetworking.metrics.Sink> sinks = new ArrayList<>();
+        for (final JsonNode sinkNode : monitoringSinks) {
+            @Nullable final JsonNode classNode = sinkNode.get("class");
+            try {
+                if (classNode != null) {
+                    final Class<?> builderClass = Class.forName(classNode.textValue() + "$Builder");
+                    final Object builder = OBJECT_MAPPER.treeToValue(sinkNode, builderClass);
+                    @SuppressWarnings("unchecked")
+                    final com.arpnetworking.metrics.Sink sink =
+                            (com.arpnetworking.metrics.Sink) builderClass.getMethod("build").invoke(builder);
+                    sinks.add(sink);
+                }
+                // CHECKSTYLE.OFF: IllegalCatch - There are so many ways this hack can fail!
+            } catch (final Exception e) {
+                // CHECKSTYLE.ON: IllegalCatch
+                throw new RuntimeException("Unable to create sink from: " + sinkNode.toString(), e);
+            }
+        }
+        return sinks;
+    }
+
     private ActorSystem launchAkka() {
         final Config akkaConfiguration = ConfigFactory.parseMap(_configuration.getAkkaConfiguration());
         return ActorSystem.create("MAD", ConfigFactory.load(akkaConfiguration));