Release version 2.2.0

.gitignore

@@ -8,4 +8,5 @@ logs/
 server/pki/
 client/src/main/python/setup.py
 grafana/timely-app/yarn-error.log
+grafana/timely-app/data/
 

analytics/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0</version>
   </parent>
   <artifactId>timely-analytics</artifactId>
   <!--

balancer/bin/create-timely-balancer-profiles.sh

@@ -1,40 +0,0 @@
-#!/bin/bash
-
-if [[ `uname` == "Darwin" ]]; then
-    THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
-else
-    THIS_SCRIPT=`readlink -f $0`
-fi
-
-THIS_DIR="${THIS_SCRIPT%/*}"
-BASE_DIR=$(cd $THIS_DIR/.. && pwd)
-BIN_DIR=${BASE_DIR}/bin
-CONF_DIR=${BASE_DIR}/conf
-
-. ${BIN_DIR}/timely-balancer-env.sh
-
-for i in $(seq 1 $NUM_PROFILES); do
-    if [ "$NUM_PROFILES" -eq "1" ]; then
-        FILENAME=timely-balancer.yml
-    else
-        FILENAME=timely-balancer-${i}.yml
-    fi
-
-    echo "creating configuration file ${CONF_DIR}/${FILENAME}"
-    cp ${CONF_DIR}/timely-balancer-template.yml ${CONF_DIR}/${FILENAME}
-#    if [ "$NUM_PROFILES" -eq "1" ]; then
-#        sed -i "s/\${INSTANCE}//g" ${CONF_DIR}/${FILENAME}
-#    else
-#        sed -i "s/\${INSTANCE}/${i}/g" ${CONF_DIR}/${FILENAME}
-#    fi
-    MULTIPLE=$(($i - 1))
-    TCP_PORT=$(($TCP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
-    sed -i "s/\${TCP_PORT}/${TCP_PORT}/g" ${CONF_DIR}/${FILENAME}
-    HTTP_PORT=$(($HTTP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
-    sed -i "s/\${HTTP_PORT}/${HTTP_PORT}/g" ${CONF_DIR}/${FILENAME}
-    WEBSOCKET_PORT=$(($WEBSOCKET_PORT_START + $MULTIPLE * $PORT_INCREMENT))
-    sed -i "s/\${WEBSOCKET_PORT}/${WEBSOCKET_PORT}/g" ${CONF_DIR}/${FILENAME}
-    UDP_PORT=$(($UDP_PORT_START + $MULTIPLE * $PORT_INCREMENT))
-    sed -i "s/\${UDP_PORT}/${UDP_PORT}/g" ${CONF_DIR}/${FILENAME}
-done
-

balancer/bin/start-all.sh

@@ -12,12 +12,7 @@ BIN_DIR=${BASE_DIR}/bin
 
 . ${BIN_DIR}/timely-balancer-env.sh
 
-if [ "$NUM_PROFILES" -eq "1" ]; then
-    echo ${BIN_DIR}/timely-balancer.sh
-    ${BIN_DIR}/timely-balancer.sh
-else
-    for i in $(seq 1 $NUM_PROFILES); do
-        echo ${BIN_DIR}/timely-balancer.sh $i
-        ${BIN_DIR}/timely-balancer.sh $i
-    done
-fi
+for i in $(seq 1 $NUM_INSTANCES); do
+    echo ${BIN_DIR}/timely-balancer.sh $i
+    ${BIN_DIR}/timely-balancer.sh $i
+done

balancer/bin/stop-all.sh

@@ -1,3 +1,21 @@
 #!/bin/bash
 
-pkill -9 -f 'timely-balancer-.*-exec.jar'
+if [[ `uname` == "Darwin" ]]; then
+    THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
+else
+    THIS_SCRIPT=`readlink -f $0`
+fi
+
+THIS_DIR="${THIS_SCRIPT%/*}"
+BASE_DIR=$(cd $THIS_DIR/.. && pwd)
+BIN_DIR=${BASE_DIR}/bin
+
+. ${BIN_DIR}/timely-balancer-env.sh
+
+for i in $(seq 1 $NUM_INSTANCES); do
+    PID=`jps -m | grep -E 'timely-balancer-.*-exec.jar' | grep "instance=${i}" | awk '{print $1}'`
+    if [ "$PID" != "" ]; then
+        echo "stopping timely-balancer instance ${i} [${PID}]"
+        kill -9 $PID
+    fi
+done

balancer/bin/testing/load-test-env.sh

@@ -1,11 +0,0 @@
-#!/bin/bash
-
-export TRUSTSTORE_FILE=
-export TRUSTSTORE_TYPE="jks";
-export TRUSTSTORE_PASS=
-export KEYSTORE_FILE=
-export KEYSTORE_TYPE="pkcs12";
-export KEYSTORE_PASS=
-
-export NUM_QUERY_THREADS=4
-export NUM_INGEST_THREADS=4

balancer/bin/testing/load-test-ingest.sh

@@ -1,21 +0,0 @@
-#!/bin/bash
-
-if [[ `uname` == "Darwin" ]]; then
-        THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
-else
-        THIS_SCRIPT=`readlink -f $0`
-fi
-
-THIS_DIR="${THIS_SCRIPT%/*}"
-BASE_DIR=${THIS_DIR}/../..
-TMP_DIR="${BASE_DIR}/tmp"
-CONF_DIR="${BASE_DIR}/conf"
-LIB_DIR="${BASE_DIR}/lib"
-
-. ${THIS_DIR}/load-test-env.sh
-
-export CLASSPATH="${CONF_DIR}:${THIS_DIR}:${LIB_DIR}/*"
-JVM_ARGS="-Xmx1G -Xms1G"
-
-echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricPutLoadTest $*"
-$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricPutLoadTest $*

balancer/bin/testing/load-test-query.sh

@@ -1,21 +0,0 @@
-#!/bin/bash
-
-if [[ `uname` == "Darwin" ]]; then
-        THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
-else
-        THIS_SCRIPT=`readlink -f $0`
-fi
-
-THIS_DIR="${THIS_SCRIPT%/*}"
-BASE_DIR=${THIS_DIR}/../..
-TMP_DIR="${BASE_DIR}/tmp"
-CONF_DIR="${BASE_DIR}/conf"
-LIB_DIR="${BASE_DIR}/lib"
-
-. ${THIS_DIR}/load-test-env.sh
-
-export CLASSPATH="${CONF_DIR}:${THIS_DIR}:${LIB_DIR}/*"
-JVM_ARGS="-Xmx1G -Xms1G"
-
-echo "$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricQueryLoadTest $*"
-$JAVA_HOME/bin/java ${JVM_ARGS} timely.testing.MetricQueryLoadTest $*

balancer/bin/timely-balancer-env.sh

@@ -1,16 +1,17 @@
 #!/bin/bash
 
-NUM_PROFILES=1
+NUM_INSTANCES=1
 ACCUMULO_ZOOKEEPERS=localhost:2181
-TCP_PORT_START=4242
-HTTP_PORT_START=4243
-WEBSOCKET_PORT_START=4244
-UDP_PORT_START=4245
-PORT_INCREMENT=100
+HADOOP_CONF_DIR=${HADOOP_CONF_DIR}
+HADOOP_DEFAULT_FS=hdfs://localhost:8020
+TCP_BASE_PORT=4242
+HTTP_BASE_PORT=4243
+WEBSOCKET_BASE_PORT=4244
+UDP_BASE_PORT=4245
 
 HOST_IP=0.0.0.0
 HOST=$(hostname -s)
-HOST_FQDN=$(hostname -f)
+HOST_VERIFICATION_ENABLED=true
 
 GENERATE_KEYS=true
 TRUST_STORE_TYPE=
@@ -22,6 +23,5 @@ KEY_STORE_PASSWORD=
 AUTHORIZATION_URL=
 AUTHORIZATION_CERTIFICATE_TYPE=
 AUTHORIZATION_CERTIFICATE_FILE=
-
-REQUIRED_ROLES=
-REQUIRED_AUTHS=
+ANONYMOUS_HTTP=true
+ANONYMOUS_WEBSOCKET=true

balancer/bin/timely-balancer.sh

@@ -1,69 +1,64 @@
 #!/bin/bash
 
-if [[ `uname` == "Darwin" ]]; then
-        THIS_SCRIPT=`python -c 'import os,sys; print os.path.realpath(sys.argv[1])' $0`
-        TCNATIVE_SUFFIX="jnilib"
+if [[ $(uname) == "Darwin" ]]; then
+    THIS_SCRIPT=$(python -c 'import os,sys; print os.path.realpath(sys.argv[1])' "$0")
 else
-        THIS_SCRIPT=`readlink -f $0`
-        TCNATIVE_SUFFIX="so"
+    THIS_SCRIPT=$(readlink -f "$0")
+    HW=`uname -m`
+    OS_PATH=linux-${HW}
 fi
 
-THIS_DIR="${THIS_SCRIPT%/*}"
-
-set -a
-. ${THIS_DIR}/timely-balancer-env.sh
-set +a
-
-if [ -n "$1" ]; then
-  export INSTANCE="$1"
-  PROFILE_ARG="--spring.profiles.active=${INSTANCE}"
-  PID=`jps -m | grep -E 'timely-balancer-.*-exec' | grep "spring.profiles.active=${INSTANCE}" | awk '{print $1}'`
+if [ -z "$1" ]; then
+    export INSTANCE="1"
 else
-  export INSTANCE="1"
-  PID=`jps -m | grep -E 'timely-balancer-.*-exec' | awk '{print $1}'`
+    export INSTANCE="$1"
 fi
 
+PID=$(jps -m | grep -E 'timely-balancer-.*-exec.jar' | grep "instance=${INSTANCE}" | awk '{print $1}')
+
 if [ "$PID" == "" ]; then
-
-    NATIVE_DIR="${THIS_DIR}/META-INF/native"
-    BASE_DIR=$(cd $THIS_DIR/.. && pwd)
-    TMP_DIR="${BASE_DIR}/tmp"
-    export CONF_DIR="${BASE_DIR}/conf"
+    THIS_DIR="${THIS_SCRIPT%/*}"
+    BASE_DIR=$(cd "$THIS_DIR"/.. && pwd)
     LIB_DIR="${BASE_DIR}/lib"
-    export LOG_DIR="${BASE_DIR}/logs"
-    NUM_SERVER_THREADS=4
+    BIN_DIR="${BASE_DIR}/bin"
+    export CONF_DIR="${BASE_DIR}/conf"
+    NATIVE_DIR="${BIN_DIR}/META-INF/native"
 
-    if [[ -e ${TMP_DIR} ]]; then
-      rm -rf ${TMP_DIR}
-    fi
-    mkdir ${TMP_DIR}
+    set -a
+    . "${BIN_DIR}/timely-balancer-env.sh"
+    set +a
 
-    if [[ -e ${NATIVE_DIR} ]]; then
-      rm -rf ${NATIVE_DIR}
+    # use either a value from timely-balancer-env.sh or the default
+    export LOG_DIR="${TIMELY_LOG_DIR:-${BASE_DIR}/logs}"
+
+    if [[ ! -e ${NATIVE_DIR} ]]; then
+        mkdir -p "${NATIVE_DIR}"
     fi
-    mkdir -p ${NATIVE_DIR}
 
     if [[ ! -e ${LOG_DIR} ]]; then
-      mkdir ${LOG_DIR}
+        mkdir "${LOG_DIR}"
     fi
 
-    pushd ${BASE_DIR}/bin
-    $JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-tcnative-boringssl-static*.jar META-INF/native/libnetty_tcnative_linux_x86_64.${TCNATIVE_SUFFIX}
-    $JAVA_HOME/bin/jar xf ${LIB_DIR}/netty-all*.jar META-INF/native/libnetty_transport_native_epoll_x86_64.${TCNATIVE_SUFFIX}
-    popd
+    JVM_ARGS="-Xmx2G -Xms2G -XX:NewSize=1G -XX:MaxNewSize=1G"
+    JVM_ARGS="${JVM_ARGS} -Dlogging.config=${CONF_DIR}/log4j2.yml"
+    JVM_ARGS="${JVM_ARGS} -Dlog4j2.contextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
+    JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:+UseStringDeduplication"
+    JVM_ARGS="${JVM_ARGS} -Djava.net.preferIPv4Stack=true"
+    JVM_ARGS="${JVM_ARGS} -XX:+UseNUMA"
 
-    export CLASSPATH="${CONF_DIR}:${LIB_DIR}/*:${HADOOP_CONF_DIR}"
-    JVM_ARGS="-Dio.netty.eventLoopThreads=${NUM_SERVER_THREADS} -Dlog4j.configurationFile=${CONF_DIR}/log4j2.yml"
-    JVM_ARGS="${JVM_ARGS} -DLog4jContextSelector=org.apache.logging.log4j.core.async.AsyncLoggerContextSelector"
-    JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
-
-    JVM_ARGS="${JVM_ARGS} -Xmx2G -Xms2G -XX:NewSize=1G -XX:MaxNewSize=1G"
-    JVM_ARGS="${JVM_ARGS} -XX:+UseCompressedOops -XX:+UseNUMA -Djava.net.preferIPv4Stack=true"
-    JVM_ARGS="${JVM_ARGS} -XX:+UseG1GC -XX:MaxGCPauseMillis=4000 -XX:ParallelGCThreads=20 -XX:+UseStringDeduplication"
-
-    echo "$JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-balancer-*-exec.jar ${PROFILE_ARG}"
-    nohup $JAVA_HOME/bin/java ${JVM_ARGS} -jar ${THIS_DIR}/timely-balancer-*-exec.jar ${PROFILE_ARG} >> ${LOG_DIR}/timely-balancer${INSTANCE}.out 2>&1 &
+    # for linux varieties of OS, extract native libraries if not already present
+    if [[ -n "$OS_PATH" ]]; then
+        if [[ ! -d ${NATIVE_DIR} || -z "$(ls -A ${NATIVE_DIR})" ]]; then
+            pushd "${BASE_DIR}/bin" || exit
+            "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-tcnative-boringssl-static-*.Final-"${OS_PATH}".jar META-INF/native/libnetty_tcnative_linux_"${HW}".so
+            "${JAVA_HOME}"/bin/jar xf "${LIB_DIR}"/netty-transport-native-epoll-*-"${OS_PATH}".jar META-INF/native/libnetty_transport_native_epoll_"${HW}".so
+            popd || exit
+        fi
+        JVM_ARGS="${JVM_ARGS} -Djava.library.path=${NATIVE_DIR}"
+    fi
 
+    echo "${JAVA_HOME}/bin/java ${JVM_ARGS} -jar ${BIN_DIR}/timely-balancer-*-exec.jar --instance=${INSTANCE}"
+    nohup "${JAVA_HOME}"/bin/java ${JVM_ARGS} -jar "${BIN_DIR}"/timely-balancer-*-exec.jar --instance="${INSTANCE}" >> "${LOG_DIR}/timely-balancer${INSTANCE}.out" 2>&1 &
 else
-    echo "timely-balancer ${INSTANCE} already running with pid ${PID}"
+    echo "timely-balancer --instance=${INSTANCE} already running with pid ${PID}"
 fi

balancer/conf/log4j2.yml

@@ -9,16 +9,9 @@ Configuration:
       - name: PID
         value: "????"
       - name: LOG_PATTERN
-        value: "%clr{%d{yyyy-MM-dd HH:mm:ss.SSS}}{faint} %clr{%5p} %clr{${sys:PID}}{magenta} %clr{---}{faint} %clr{[%15.15t]}{faint} %clr{%-40.40c{1.}}{cyan} %clr{:}{faint} %m%n%wEx"
+        value: "%d{ISO8601} %-5p [%-40.40t] [%-40.40c{1.}] %m%n%wEx"
 
   Appenders:
-    Console:
-      name: Console
-      target: SYSTEM_OUT
-      follow: true
-      PatternLayout:
-        pattern: "${LOG_PATTERN}"
-
     RollingFile:
       - name: File
         fileName: "${sys:logDir}/timely-balancer${env:INSTANCE}.log"

balancer/conf/security.properties

@@ -1,9 +0,0 @@
-trust.store.type=${TRUST_STORE_TYPE}
-trust.store.file=${TRUST_STORE_FILE}
-trust.store.password=${TRUST_STORE_PASSWORD}
-key.store.type=${KEY_STORE_TYPE}
-key.store.file=${KEY_STORE_FILE}
-key.store.password=${KEY_STORE_PASSWORD}
-authorization.url=${AUTHORIZATION_URL}
-authorization.certificate.type=${AUTHORIZATION_CERTIFICATE_TYPE}
-authorization.certificate.file=${AUTHORIZATION_CERTIFICATE_FILE}

balancer/conf/timely-balancer-template.yml

@@ -1,142 +0,0 @@
-timely:
-  zookeeper:
-    servers: ${ACCUMULO_ZOOKEEPERS}
-    timeout: 30s
-  security:
-    session-max-age: 86400
-    cache-expiration-minutes: 60
-    cache-refresh-minutes: 5
-    allow-anonymous-http-access: true
-    allow-anonymous-ws-access: true
-    serverSsl:
-      key-store-type: ${KEY_STORE_TYPE}
-      key-store-file: ${KEY_STORE_FILE}
-      key-store-password: ${KEY_STORE_PASSWORD}
-      use-generated-keypair: ${GENERATE_KEYS}
-      ignore-ssl-handshake-errors: true
-      trust-store-type: ${TRUST_STORE_TYPE}
-      trust-store-file: ${TRUST_STORE_FILE}
-      trust-store-password: ${TRUST_STORE_PASSWORD}
-    clientSsl:
-      key-store-type: ${KEY_STORE_TYPE}
-      key-store-file: ${KEY_STORE_FILE}
-      key-store-password: ${KEY_STORE_PASSWORD}
-      trust-store-type: ${TRUST_STORE_TYPE}
-      trust-store-file: ${TRUST_STORE_FILE}
-      trust-store-password: ${TRUST_STORE_PASSWORD}
-    jwtSsl:
-      key-store-type: ${AUTHORIZATION_CERTIFICATE_TYPE}
-      key-store-file: ${AUTHORIZATION_CERTIFICATE_FILE}
-      key-store-password: ${AUTHORIZATION_CERTIFICATE_PASSWORD}
-  cache:
-    enabled: false
-    metric-age-off-hours:
-      default: 26
-    nonCachedMetrics:
-  balancer:
-    checkServerHealthInterval: 10000
-    serverFailuresBeforeDown: 6
-    serverSuccessesBeforeUp: 1
-    assignmentFile: /timely/balancer/assignments.csv
-    server:
-      ip: ${HOST_IP}
-      tcpPort: ${TCP_PORT}
-      numTcpPools: 1
-      tcpBufferSize: 100
-      tcpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-      udpPort: ${UDP_PORT}
-      udpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    http:
-      host: ${HOST}
-      ip: ${HOST_IP}
-      port: ${HTTP_PORT}
-      redirect-path: /secure-me
-      strict-transport-max-age: 604800
-      cors:
-        allow-any-origin: false
-        allow-null-origin: false
-        allow-credentials: true
-        allowed-origins:
-        allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
-        allowed-headers: content-type
-      httpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-    websocket:
-      ip: ${HOST_IP}
-      port: ${WEBSOCKET_PORT}
-      timeout: 60
-      subscription-lag: 120
-      scanner-batch-size: 10000
-      flush-interval-seconds: 30
-      scanner-read-ahead: 1
-      subscription-batch-size: 5000
-      incomingBufferSize: 50000000
-      wsClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
-        lifo: true
-        fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-
-logging:
-  level:
-    ROOT: info
-    timey: debug

balancer/conf/timely-balancer.yml

@@ -1,4 +1,5 @@
 timely:
+  instance: ${INSTANCE}
   zookeeper:
     servers: ${ACCUMULO_ZOOKEEPERS}
     timeout: 30s
@@ -6,82 +7,88 @@ timely:
     session-max-age: 86400
     cache-expiration-minutes: 60
     cache-refresh-minutes: 5
-    allow-anonymous-http-access: true
-    allow-anonymous-ws-access: true
-    serverSsl:
-      key-store-type:
-      key-store-file:
-      key-store-password:
-      use-generated-keypair: true
+    authorization-url: ${AUTHORIZATION_URL}
+    allow-anonymous-http-access: ${ANONYMOUS_HTTP:true}
+    allow-anonymous-ws-access: ${ANONYMOUS_WEBSOCKET:true}
+    server-ssl:
+      key-store-type: ${KEY_STORE_TYPE}
+      key-store-file: ${KEY_STORE_FILE}
+      key-store-password: ${KEY_STORE_PASSWORD}
+      use-generated-keypair: ${GENERATE_KEYS}
       ignore-ssl-handshake-errors: true
-      trust-store-type:
-      trust-store-file:
-      trust-store-password:
-    clientSsl:
-      key-store-type:
-      key-store-file:
-      key-store-password:
-      trust-store-type:
-      trust-store-file:
-      trust-store-password:
-    jwtSsl:
-      key-store-type:
-      key-store-file:
-      key-store-password:
+      trust-store-type: ${TRUST_STORE_TYPE}
+      trust-store-file: ${TRUST_STORE_FILE}
+      trust-store-password: ${TRUST_STORE_PASSWORD}
+    client-ssl:
+      key-store-type: ${KEY_STORE_TYPE}
+      key-store-file: ${KEY_STORE_FILE}
+      key-store-password: ${KEY_STORE_PASSWORD}
+      trust-store-type: ${TRUST_STORE_TYPE}
+      trust-store-file: ${TRUST_STORE_FILE}
+      trust-store-password: ${TRUST_STORE_PASSWORD}
+      hostVerificationEnabled: ${HOST_VERIFICATION_ENABLED:true}
+    jwt-ssl:
+      key-store-type: ${AUTHORIZATION_CERTIFICATE_TYPE}
+      key-store-file: ${AUTHORIZATION_CERTIFICATE_FILE}
+      key-store-password: ${AUTHORIZATION_CERTIFICATE_PASSWORD}
   cache:
     enabled: false
     metric-age-off-hours:
       default: 26
-    nonCachedMetrics:
+    non-cached-metrics:
   balancer:
-    checkServerHealthInterval: 10000
-    serverFailuresBeforeDown: 6
-    serverSuccessesBeforeUp: 1
-    assignmentFile: /timely/balancer/assignments.csv
+    check-server-health-interval: 10000
+    server-failures-before-down: 6
+    server-successes-before-up: 1
+    default-fs: ${HADOOP_DEFAULT_FS}
+    fs-config-resources:
+      - ${HADOOP_CONF_DIR}/core-site.xml
+      - ${HADOOP_CONF_DIR}/hdfs-site.xml
+    assignment-file: hdfs:///timely/balancer/assignments.csv
     server:
-      ip: 0.0.0.0
-      tcpPort: 4242
-      numTcpPools: 1
-      tcpBufferSize: 100
-      tcpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
+      ip: ${HOST_IP}
+      tcp-base-port: ${TCP_BASE_PORT}
+      num-tcp-pools: 1
+      tcp-buffer-size: 100
+      tcp-client-pool:
+        min-idle-per-key: 0
+        max-idle-per-key: 40
+        max-total-per-key: 40
+        max-total: -1
         lifo: true
         fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-      udpPort: 4245
-      udpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
+        max-wait-millis: -1
+        min-evictable-idle-time-millis: 1800000
+        evictor-shutdown-timeout-millis: 10000
+        soft-min-evictable-idle-time-millis: -1
+        num-tests-per-eviction-run: 3
+        time-between-eviction-runs-millis: -1
+        block-when-exhausted: true
+        jmx-enabled: true
+        test-on-borrow: false
+        test-on-return: false
+      udp-base-port: ${UDP_BASE_PORT}
+      udp-client-pool:
+        min-idle-per-key: 0
+        max-idle-per-key: 40
+        max-total-per-key: 40
+        max-total: -1
         lifo: true
         fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
+        max-wait-millis: -1
+        min-evictable-idle-time-millis: 1800000
+        evictor-shutdown-timeout-millis: 10000
+        soft-min-evictable-idle-time-millis: -1
+        num-tests-per-eviction-run: 3
+        time-between-eviction-runs-millis: -1
+        block-when-exhausted: true
+        jmx-enabled: true
+        test-on-borrow: false
+        test-on-return: false
     http:
-      host: localhost
-      ip: 0.0.0.0
-      port: 4243
+      host: ${HOST}
+      ip: ${HOST_IP}
+      base-port: ${HTTP_BASE_PORT}
       redirect-path: /secure-me
       strict-transport-max-age: 604800
       cors:
@@ -92,51 +99,51 @@ timely:
         allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
         allowed-headers: content-type
       httpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
+        min-idle-per-key: 0
+        max-idle-per-key: 40
+        max-total-per-key: 40
+        max-total: -1
         lifo: true
         fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
+        max-wait-millis: -1
+        min-evictable-idle-time-millis: 1800000
+        evictor-shutdown-timeout-millis: 10000
+        soft-min-evictable-idle-time-millis: -1
+        num-tests-per-eviction-run: 3
+        time-between-eviction-runs-millis: -1
+        block-when-exhausted: true
+        jmx-enabled: true
+        test-on-borrow: false
+        test-on-return: false
     websocket:
-      ip: 0.0.0.0
-      port: 5244
+      ip: ${HOST_IP}
+      base-port: ${WEBSOCKET_BASE_PORT}
       timeout: 60
       subscription-lag: 120
       scanner-batch-size: 10000
       flush-interval-seconds: 30
       scanner-read-ahead: 1
       subscription-batch-size: 5000
-      incomingBufferSize: 50000000
-      wsClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
+      incoming-buffer-size: 50000000
+      ws-client-pool:
+        min-idle-per-key: 0
+        max-idle-per-key: 40
+        max-total-per-key: 40
+        max-total: -1
         lifo: true
         fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
+        max-wait-millis: -1
+        min-evictable-idle-time-millis: 1800000
+        evictor-shutdown-timeout-millis: 10000
+        soft-min-evictable-idle-time-millis: -1
+        num-tests-per-eviction-run: 3
+        time-between-eviction-runs-millis: -1
+        block-when-exhausted: true
+        jmx-enabled: true
+        test-on-borrow: false
+        test-on-return: false
 
 logging:
   level:
     ROOT: info
-    timey: debug
+    timely: debug

balancer/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0</version>
   </parent>
   <artifactId>timely-balancer</artifactId>
   <description>Timely Balancer</description>
@@ -40,18 +40,10 @@
       <groupId>commons-beanutils</groupId>
       <artifactId>commons-beanutils</artifactId>
     </dependency>
-    <dependency>
-      <groupId>commons-logging</groupId>
-      <artifactId>commons-logging</artifactId>
-    </dependency>
     <dependency>
       <groupId>io.github.classgraph</groupId>
       <artifactId>classgraph</artifactId>
     </dependency>
-    <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty-all</artifactId>
-    </dependency>
     <dependency>
       <groupId>net.sourceforge.javacsv</groupId>
       <artifactId>javacsv</artifactId>
@@ -60,10 +52,6 @@
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-core</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.accumulo</groupId>
-      <artifactId>accumulo-minicluster</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-collections4</artifactId>
@@ -100,10 +88,6 @@
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.hadoop</groupId>
-      <artifactId>hadoop-minicluster</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-1.2-api</artifactId>
@@ -166,10 +150,6 @@
       <groupId>timely</groupId>
       <artifactId>timely-common</artifactId>
     </dependency>
-    <dependency>
-      <groupId>timely</groupId>
-      <artifactId>timely-server</artifactId>
-    </dependency>
     <dependency>
       <groupId>com.lmax</groupId>
       <artifactId>disruptor</artifactId>
@@ -178,6 +158,30 @@
     <dependency>
       <groupId>io.netty</groupId>
       <artifactId>netty-tcnative-boringssl-static</artifactId>
+      <classifier>linux-aarch_64</classifier>
+      <scope>runtime</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-tcnative-boringssl-static</artifactId>
+      <classifier>linux-x86_64</classifier>
+      <scope>runtime</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-native-epoll</artifactId>
+      <classifier>linux-aarch_64</classifier>
+      <scope>runtime</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-transport-native-epoll</artifactId>
+      <classifier>linux-x86_64</classifier>
+      <scope>runtime</scope>
+    </dependency>
+    <dependency>
+      <groupId>javax.activation</groupId>
+      <artifactId>activation</artifactId>
       <scope>runtime</scope>
     </dependency>
     <dependency>
@@ -200,6 +204,11 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-minicluster</artifactId>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.jsoup</groupId>
       <artifactId>jsoup</artifactId>
@@ -215,6 +224,11 @@
       <artifactId>spring-boot-test-autoconfigure</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>timely</groupId>
+      <artifactId>timely-test</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

balancer/src/main/dist/assembly.xml

@@ -19,8 +19,10 @@
       <directory>${project.build.directory}/lib</directory>
       <outputDirectory>lib</outputDirectory>
       <includes>
-        <include>netty-tcnative-boringssl-static*.jar</include>
-        <include>netty-all*.jar</include>
+        <include>netty-tcnative-boringssl-static*linux-x86_64.jar</include>
+        <include>netty-tcnative-boringssl-static*linux-aarch_64.jar</include>
+        <include>netty-transport-native-epoll*linux-x86_64.jar</include>
+        <include>netty-transport-native-epoll*linux-aarch_64.jar</include>
       </includes>
     </fileSet>
     <fileSet>
@@ -52,4 +54,4 @@
 	  </includes>
 	</fileSet>
   </fileSets>
-</assembly>
+</assembly>

balancer/src/main/java/timely/balancer/Balancer.java

@@ -1,4 +1,4 @@
-package timely.balancer.component;
+package timely.balancer;
 
 import static org.apache.accumulo.core.conf.ConfigurationTypeHelper.getTimeInMillis;
 import static timely.Constants.NON_CACHED_METRICS;
@@ -9,30 +9,24 @@ import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Date;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
 import javax.net.ssl.SSLContext;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.retry.RetryForever;
-import org.apache.hadoop.util.StringUtils;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.data.Stat;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.ApplicationContext;
-import org.springframework.stereotype.Component;
 
 import io.netty.bootstrap.AbstractBootstrap;
 import io.netty.bootstrap.Bootstrap;
@@ -64,12 +58,9 @@ import io.netty.handler.codec.http.HttpResponseEncoder;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
 import io.netty.handler.logging.LoggingHandler;
-import io.netty.handler.ssl.ClientAuth;
 import io.netty.handler.ssl.OpenSslServerContext;
 import io.netty.handler.ssl.OpenSslServerSessionContext;
 import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.SslContextBuilder;
-import io.netty.handler.ssl.util.SelfSignedCertificate;
 import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.NettyRuntime;
@@ -87,11 +78,9 @@ import timely.balancer.netty.http.HttpRelayHandler;
 import timely.balancer.netty.tcp.TcpRelayHandler;
 import timely.balancer.netty.udp.UdpRelayHandler;
 import timely.balancer.netty.ws.WsRelayHandler;
-import timely.client.http.HttpClient;
 import timely.common.component.AuthenticationService;
 import timely.common.configuration.SecurityProperties;
 import timely.common.configuration.SslClientProperties;
-import timely.common.configuration.SslHelper;
 import timely.common.configuration.SslServerProperties;
 import timely.common.configuration.ZookeeperProperties;
 import timely.netty.http.HttpStaticFileServerHandler;
@@ -107,8 +96,6 @@ import timely.netty.websocket.WebSocketFullRequestHandler;
 import timely.netty.websocket.WebSocketRequestDecoder;
 import timely.netty.websocket.subscription.WSTimelyExceptionHandler;
 
-@Component
-@ConditionalOnMissingBean(Balancer.class)
 public class Balancer {
 
     final public static String LEADER_LATCH_PATH = "/timely/balancer/leader";
@@ -120,8 +107,6 @@ public class Balancer {
     private static final String OS_NAME = "os.name";
     private static final String OS_VERSION = "os.version";
     private static final String WS_PATH = "/websocket";
-    protected static final CountDownLatch LATCH = new CountDownLatch(1);
-
     private BalancerProperties balancerProperties;
     private SecurityProperties securityProperties;
     private SslServerProperties sslServerProperties;
@@ -134,7 +119,8 @@ public class Balancer {
     private AuthenticationService authenticationService;
     private CuratorFramework curatorFramework;
     private int quietPeriod;
-
+    private SslContext serverSslContext;
+    private SSLContext clientSSLContext;
     private EventLoopGroup tcpWorkerGroup = null;
     private EventLoopGroup tcpBossGroup = null;
     private EventLoopGroup httpWorkerGroup = null;
@@ -157,20 +143,26 @@ public class Balancer {
         final String osName = SystemPropertyUtil.get(OS_NAME).toLowerCase().trim();
         final String osVersion = SystemPropertyUtil.get(OS_VERSION).toLowerCase();
         // split at periods, keep the first two, join with a period
-        final String majMinVers = StringUtils.join(".", Arrays.stream(osVersion.split("\\.")).limit(2).collect(Collectors.toList()));
+        final String majMinVers = StringUtils.join(Arrays.stream(osVersion.split("\\.")).limit(2).collect(Collectors.toList()), ".");
         if (osName.startsWith("linux")) {
-            Runtime.Version currentVersion = Runtime.Version.parse(majMinVers);
-            Runtime.Version epollMinVersion = Runtime.Version.parse(EPOLL_MIN_VERSION);
-            return currentVersion.compareTo(epollMinVersion) > 0;
-        } else {
-            return false;
+            try {
+                Runtime.Version currentVersion = Runtime.Version.parse(majMinVers);
+                Runtime.Version epollMinVersion = Runtime.Version.parse(EPOLL_MIN_VERSION);
+                return currentVersion.compareTo(epollMinVersion) > 0;
+            } catch (Exception e) {
+                log.error(e.getMessage(), e);
+            }
         }
+        return false;
     }
 
-    public Balancer(ApplicationContext applicationContext, AuthenticationService authenticationService, MetricResolver metricResolver,
-                    BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
-                    SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
-                    BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
+    public Balancer(ApplicationContext applicationContext, SslContext serverSslContext, SSLContext clientSSLContext,
+                    AuthenticationService authenticationService, MetricResolver metricResolver, BalancerProperties balancerProperties,
+                    ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties, SslServerProperties sslServerProperties,
+                    SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties, BalancerHttpProperties balancerHttpProperties,
+                    BalancerWebsocketProperties balancerWebsocketProperties) {
+        this.serverSslContext = serverSslContext;
+        this.clientSSLContext = clientSSLContext;
         this.quietPeriod = balancerServerProperties.getShutdownQuietPeriod();
         this.applicationContext = applicationContext;
         this.authenticationService = authenticationService;
@@ -186,8 +178,7 @@ public class Balancer {
         DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
     }
 
-    @PostConstruct
-    public void startup() {
+    public void start() {
         log.info("Starting {}", this.getClass().getSimpleName());
         try {
             // start curator framework
@@ -246,9 +237,9 @@ public class Balancer {
 
             final int httpPort = balancerHttpProperties.getPort();
             final String httpIp = balancerHttpProperties.getIp();
-            SslContext sslCtx = createSSLContext(sslServerProperties);
-            if (sslCtx instanceof OpenSslServerContext) {
-                OpenSslServerContext openssl = (OpenSslServerContext) sslCtx;
+
+            if (serverSslContext instanceof OpenSslServerContext) {
+                OpenSslServerContext openssl = (OpenSslServerContext) serverSslContext;
                 String application = "Timely_" + httpPort;
                 OpenSslServerSessionContext opensslCtx = openssl.sessionContext();
                 opensslCtx.setSessionCacheEnabled(true);
@@ -256,13 +247,12 @@ public class Balancer {
                 opensslCtx.setSessionIdContext(application.getBytes(StandardCharsets.UTF_8));
                 opensslCtx.setSessionTimeout(securityProperties.getSessionMaxAge());
             }
-            SSLContext clientSSLContext = createSSLClientContext(sslClientProperties);
             final ServerBootstrap httpServer = new ServerBootstrap();
             httpServer.group(httpBossGroup, httpWorkerGroup);
             httpServer.channel(channelClass);
             httpServer.handler(new LoggingHandler());
-            HttpClientPool httpClientPool = new HttpClientPool(securityProperties, sslClientProperties, balancerHttpProperties, clientSSLContext);
-            httpServer.childHandler(setupHttpChannel(balancerHttpProperties, sslServerProperties, sslCtx, metricResolver, httpClientPool));
+            HttpClientPool httpClientPool = new HttpClientPool(balancerHttpProperties, clientSSLContext, sslClientProperties);
+            httpServer.childHandler(setupHttpChannel(balancerHttpProperties, sslServerProperties, serverSslContext, metricResolver, httpClientPool));
             httpServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
             httpServer.option(ChannelOption.SO_BACKLOG, 128);
             httpServer.childOption(ChannelOption.SO_KEEPALIVE, true);
@@ -276,7 +266,7 @@ public class Balancer {
             wsServer.channel(channelClass);
             wsServer.handler(new LoggingHandler());
             WsClientPool wsClientPool = new WsClientPool(balancerProperties, balancerWebsocketProperties, sslClientProperties, clientSSLContext);
-            wsServer.childHandler(setupWSChannel(balancerWebsocketProperties, securityProperties, sslCtx, metricResolver, wsClientPool));
+            wsServer.childHandler(setupWSChannel(balancerWebsocketProperties, securityProperties, serverSslContext, metricResolver, wsClientPool));
             wsServer.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
             wsServer.option(ChannelOption.SO_BACKLOG, 128);
             wsServer.childOption(ChannelOption.SO_KEEPALIVE, true);
@@ -317,7 +307,6 @@ public class Balancer {
         return channel;
     }
 
-    @PreDestroy
     public void shutdown() {
         List<ChannelFuture> channelFutures = new ArrayList<>();
 
@@ -410,31 +399,6 @@ public class Balancer {
         log.info("{} shut down.", this.getClass().getSimpleName());
     }
 
-    protected SslContext createSSLContext(SslServerProperties sslServerProperties) throws Exception {
-
-        Boolean generate = sslServerProperties.isUseGeneratedKeypair();
-        SslContextBuilder ssl;
-        if (generate) {
-            log.warn("Using generated self signed server certificate");
-            Date begin = new Date();
-            Date end = new Date(begin.getTime() + 86400000);
-            SelfSignedCertificate ssc = new SelfSignedCertificate("localhost", begin, end);
-            ssl = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey());
-        } else {
-            ssl = SslHelper.getSslContextBuilder(sslServerProperties);
-        }
-        // Can't set to REQUIRE because the CORS pre-flight requests will fail.
-        ssl.clientAuth(ClientAuth.OPTIONAL);
-        return ssl.build();
-    }
-
-    protected SSLContext createSSLClientContext(SslClientProperties sslClientProperties) {
-
-        SslClientProperties ssl = sslClientProperties;
-        return HttpClient.getSSLContext(ssl.getTrustStoreFile(), ssl.getTrustStoreType(), ssl.getTrustStorePassword(), ssl.getKeyStoreFile(),
-                        ssl.getKeyStoreType(), ssl.getKeyStorePassword());
-    }
-
     private void ensureZkPaths(CuratorFramework curatorFramework, String[] paths) {
         for (String s : paths) {
             try {

balancer/src/main/java/timely/balancer/HealthChecker.java

@@ -1,4 +1,4 @@
-package timely.balancer.component;
+package timely.balancer;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -6,13 +6,8 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.stereotype.Component;
 
 import timely.balancer.configuration.BalancerProperties;
 import timely.balancer.configuration.BalancerServerProperties;
@@ -20,8 +15,6 @@ import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.tcp.TcpClientPool;
 import timely.balancer.healthcheck.TCPServerHealthCheck;
 
-@Component
-@ConditionalOnMissingBean(HealthChecker.class)
 public class HealthChecker {
 
     private static final Logger log = LoggerFactory.getLogger(HealthChecker.class);
@@ -36,8 +29,7 @@ public class HealthChecker {
         this.balancerServerProperties = balancerServerProperties;
     }
 
-    @PostConstruct
-    public void setup() {
+    public void start() {
         TcpClientPool tcpClientPool = new TcpClientPool(balancerServerProperties);
         this.tcpServerHealthCheck = new TCPServerHealthCheck(tcpClientPool);
         this.executorService.scheduleAtFixedRate(() -> {
@@ -58,8 +50,7 @@ public class HealthChecker {
         }, 0, balancerProperties.getCheckServerHealthInterval(), TimeUnit.MILLISECONDS);
     }
 
-    @PreDestroy
-    public void close() throws Exception {
+    public void shutdown() {
         this.executorService.shutdown();
         try {
             this.executorService.awaitTermination(5, TimeUnit.SECONDS);

balancer/src/main/java/timely/balancer/MetricResolver.java

@@ -1,13 +1,15 @@
-package timely.balancer.component;
+package timely.balancer;
 
+import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY;
 import static timely.Constants.NON_CACHED_METRICS;
 import static timely.Constants.NON_CACHED_METRICS_LOCK_PATH;
 import static timely.Constants.SERVICE_DISCOVERY_PATH;
-import static timely.balancer.component.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
-import static timely.balancer.component.Balancer.ASSIGNMENTS_LOCK_PATH;
-import static timely.balancer.component.Balancer.LEADER_LATCH_PATH;
+import static timely.balancer.Balancer.ASSIGNMENTS_LAST_UPDATED_PATH;
+import static timely.balancer.Balancer.ASSIGNMENTS_LOCK_PATH;
+import static timely.balancer.Balancer.LEADER_LATCH_PATH;
 
 import java.io.FileNotFoundException;
+import java.io.IOException;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -29,9 +31,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-
 import org.apache.commons.lang3.SerializationUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.curator.framework.CuratorFramework;
@@ -53,25 +52,21 @@ import org.apache.curator.x.discovery.details.ServiceCacheListener;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.zookeeper.CreateMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.stereotype.Component;
 
 import com.csvreader.CsvReader;
 import com.csvreader.CsvWriter;
 
 import timely.ServerDetails;
-import timely.balancer.ArrivalRate;
 import timely.balancer.configuration.BalancerProperties;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.common.configuration.CacheProperties;
 
-@Component
-@ConditionalOnMissingBean(MetricResolver.class)
 public class MetricResolver {
 
     private static final Logger log = LoggerFactory.getLogger(MetricResolver.class);
@@ -96,26 +91,34 @@ public class MetricResolver {
     final protected BalancerProperties balancerProperties;
     final protected CacheProperties cacheProperties;
     final protected HealthChecker healthChecker;
+    final protected Path assignmentFile;
+    final protected FileSystem fs;
 
     private enum BalanceType {
         HIGH_LOW, HIGH_AVG, AVG_LOW
     }
 
     public MetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         this.curatorFramework = curatorFramework;
         this.balancerProperties = balancerProperties;
         this.cacheProperties = cacheProperties;
         this.healthChecker = healthChecker;
+        this.assignmentFile = new Path(balancerProperties.getAssignmentFile());
+        this.fs = getFileSystem(this.assignmentFile);
     }
 
-    @PostConstruct
-    public void setup() {
+    public void start() {
         assignmentsIPRWLock = new InterProcessReadWriteLock(curatorFramework, ASSIGNMENTS_LOCK_PATH);
         testIPRWLock(curatorFramework, assignmentsIPRWLock, ASSIGNMENTS_LOCK_PATH);
         startLeaderLatch(curatorFramework);
         startServiceListener(curatorFramework);
         assignmentsLastUpdatedInHdfs = new DistributedAtomicLong(curatorFramework, ASSIGNMENTS_LAST_UPDATED_PATH, new RetryForever(1000));
+        try {
+            assignmentsLastUpdatedInHdfs.forceSet(getLastWrittenToHdfsTimestamp());
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+        }
 
         TreeCacheListener assignmentListener = (curframework, event) -> {
             log.debug("Handling assignments event {}. assignmentsLastUpdatedInHdfs:{}", event.getType().toString(),
@@ -191,8 +194,7 @@ public class MetricResolver {
         }, 10, 60, TimeUnit.SECONDS);
     }
 
-    @PreDestroy
-    public void close() throws Exception {
+    public void shutdown() {
 
         this.assignmentExecutor.shutdown();
         try {
@@ -358,8 +360,13 @@ public class MetricResolver {
                         if (isLeader.get()) {
                             for (String s : reassignMetrics) {
                                 TimelyBalancedHost h = getRoundRobinHost(null);
-                                assignMetric(s, h);
-                                log.debug("Assigned server removed.  Assigning {} to server {}:{}", s, h.getTcpPort());
+                                if (h == null) {
+                                    unassignMetric(s);
+                                    log.debug("Assigned server removed and none available. Unassigning {}", s);
+                                } else {
+                                    assignMetric(s, h);
+                                    log.debug("Assigned server removed. Assigning {} to server {}:{}", s, h.getTcpPort());
+                                }
                             }
                         }
 
@@ -787,20 +794,19 @@ public class MetricResolver {
                 tbh = metricToHostMap.get(metric);
                 if (tbh == null) {
                     tbh = getRoundRobinHost(null);
-                    if (!balancerLock.isWriteLockedByCurrentThread()) {
-                        balancerLock.readLock().unlock();
-                        balancerLock.writeLock().lock();
-                    }
-                    assignMetric(metric, tbh);
                 } else if (!tbh.isUp()) {
                     TimelyBalancedHost oldTbh = tbh;
                     tbh = getLeastUsedHost();
-                    log.debug("rebalancing from host that is down: reassigning metric {} from server {}:{} to {}:{}", metric, oldTbh.getHost(),
-                                    oldTbh.getTcpPort(), tbh.getHost(), tbh.getTcpPort());
-                    if (!balancerLock.isWriteLockedByCurrentThread()) {
-                        balancerLock.readLock().unlock();
-                        balancerLock.writeLock().lock();
+                    if (tbh != null) {
+                        log.debug("rebalancing from host that is down: reassigning metric {} from server {}:{} to {}:{}", metric, oldTbh.getHost(),
+                                        oldTbh.getTcpPort(), tbh.getHost(), tbh.getTcpPort());
                     }
+                }
+                if (!balancerLock.isWriteLockedByCurrentThread()) {
+                    balancerLock.readLock().unlock();
+                    balancerLock.writeLock().lock();
+                }
+                if (tbh != null) {
                     assignMetric(metric, tbh);
                 }
             }
@@ -899,18 +905,41 @@ public class MetricResolver {
         return tbh;
     }
 
+    protected FileSystem getFileSystem(Path path) throws IOException {
+        Configuration configuration = new Configuration();
+        if (StringUtils.isNotBlank(balancerProperties.getDefaultFs())) {
+            configuration.set(FS_DEFAULT_NAME_KEY, balancerProperties.getDefaultFs());
+        }
+        if (balancerProperties.getFsConfigResources() != null) {
+            for (String resource : balancerProperties.getFsConfigResources()) {
+                configuration.addResource(new Path(resource));
+            }
+        }
+        return FileSystem.get(path.toUri(), configuration);
+    }
+
     protected void createAssignmentFile() {
         try {
-            log.info("Creating assignment file: " + balancerProperties.getAssignmentFile());
-            Configuration configuration = new Configuration();
-            FileSystem fs = FileSystem.get(configuration);
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            fs.create(assignmentFile);
+            log.info("Creating assignment file: " + this.assignmentFile);
+            this.fs.create(this.assignmentFile);
         } catch (Exception e) {
             log.error(e.getMessage(), e);
         }
     }
 
+    protected long getLastWrittenToHdfsTimestamp() {
+        try {
+            FileStatus fileStatus = this.fs.getFileStatus(this.assignmentFile);
+            return fileStatus.getModificationTime();
+        } catch (FileNotFoundException e) {
+            createAssignmentFile();
+            return System.currentTimeMillis();
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+            return System.currentTimeMillis();
+        }
+    }
+
     private void readAssignmentsFromHdfs(boolean checkIfNecessary) {
 
         try {
@@ -931,17 +960,15 @@ public class MetricResolver {
         // proceed with reading from HDFS
 
         Map<String,TimelyBalancedHost> assignedMetricToHostMap = new TreeMap<>();
+        CsvReader reader = null;
         try {
             boolean acquired = false;
             while (!acquired) {
                 acquired = assignmentsIPRWLock.readLock().acquire(60, TimeUnit.SECONDS);
             }
             balancerLock.writeLock().lock();
-            Configuration configuration = new Configuration();
-            FileSystem fs = FileSystem.get(configuration);
-            Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-            FSDataInputStream iStream = fs.open(assignmentFile);
-            CsvReader reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
+            FSDataInputStream iStream = this.fs.open(this.assignmentFile);
+            reader = new CsvReader(iStream, ',', Charset.forName("UTF-8"));
             reader.setUseTextQualifier(false);
 
             // skip the headers
@@ -961,7 +988,9 @@ public class MetricResolver {
                     } else {
                         log.trace("Found assigment: {} to {}:{}", metric, host, tcpPort);
                     }
-                    assignedMetricToHostMap.put(metric, tbh);
+                    if (tbh != null) {
+                        assignedMetricToHostMap.put(metric, tbh);
+                    }
                 }
             }
 
@@ -1001,6 +1030,9 @@ public class MetricResolver {
         } catch (Exception e) {
             log.error(e.getMessage(), e);
         } finally {
+            if (reader != null) {
+                reader.close();
+            }
             balancerLock.writeLock().unlock();
             try {
                 assignmentsIPRWLock.readLock().release();
@@ -1022,13 +1054,10 @@ public class MetricResolver {
             nonCachedMetricsLocalLock.readLock().lock();
             try {
                 if (!metricToHostMap.isEmpty()) {
-                    Configuration configuration = new Configuration();
-                    FileSystem fs = FileSystem.get(configuration);
-                    Path assignmentFile = new Path(balancerProperties.getAssignmentFile());
-                    if (!fs.exists(assignmentFile.getParent())) {
-                        fs.mkdirs(assignmentFile.getParent());
+                    if (!this.fs.exists(this.assignmentFile.getParent())) {
+                        this.fs.mkdirs(this.assignmentFile.getParent());
                     }
-                    FSDataOutputStream oStream = fs.create(assignmentFile, true);
+                    FSDataOutputStream oStream = this.fs.create(this.assignmentFile, true);
                     writer = new CsvWriter(oStream, ',', Charset.forName("UTF-8"));
                     writer.setUseTextQualifier(false);
                     writer.write("metric");
@@ -1036,7 +1065,7 @@ public class MetricResolver {
                     writer.write("tcpPort");
                     writer.endRecord();
                     for (Map.Entry<String,TimelyBalancedHost> e : metricToHostMap.entrySet()) {
-                        if (!nonCachedMetrics.contains(e.getKey())) {
+                        if (e.getValue() != null && !nonCachedMetrics.contains(e.getKey())) {
                             writer.write(e.getKey());
                             writer.write(e.getValue().getHost());
                             writer.write(Integer.toString(e.getValue().getTcpPort()));
@@ -1051,7 +1080,7 @@ public class MetricResolver {
                     if (!assignmentsLastUpdatedInHdfs.get().succeeded()) {
                         assignmentsLastUpdatedInHdfs.forceSet(now);
                     }
-                    log.debug("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
+                    log.info("Wrote {} assignments to hdfs lastHdfsUpdate = lastLocalUpdate ({})", metricToHostMap.size(),
                                     new Date(assignmentsLastUpdatedLocal.get()));
                     // remove metric from metricMap (ArrivalRate) if not being cached
                     metricMap.entrySet().removeIf(e -> !metricToHostMap.containsKey(e.getKey()));
@@ -1090,4 +1119,18 @@ public class MetricResolver {
             }
         }
     }
+
+    protected void unassignMetric(String metric) {
+        if (isLeader.get()) {
+            if (StringUtils.isNotBlank(metric)) {
+                balancerLock.writeLock().lock();
+                try {
+                    metricToHostMap.remove(metric);
+                    assignmentsLastUpdatedLocal.set(System.currentTimeMillis());
+                } finally {
+                    balancerLock.writeLock().unlock();
+                }
+            }
+        }
+    }
 }

balancer/src/main/java/timely/balancer/configuration/BalancerConfiguration.java

@@ -1,10 +1,58 @@
 package timely.balancer.configuration;
 
+import javax.net.ssl.SSLContext;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import io.netty.handler.ssl.SslContext;
+import timely.balancer.Balancer;
+import timely.balancer.HealthChecker;
+import timely.balancer.MetricResolver;
+import timely.common.component.AuthenticationService;
+import timely.common.configuration.CacheProperties;
+import timely.common.configuration.SecurityProperties;
+import timely.common.configuration.SslClientProperties;
+import timely.common.configuration.SslServerProperties;
+import timely.common.configuration.ZookeeperProperties;
+
 @Configuration
 @EnableConfigurationProperties({BalancerProperties.class, BalancerServerProperties.class, BalancerWebsocketProperties.class, BalancerHttpProperties.class})
 public class BalancerConfiguration {
 
+    @Bean
+    @ConditionalOnMissingBean
+    public Balancer balancer(ApplicationContext applicationContext, @Qualifier("nettySslContext") SslContext serverSslContext,
+                    @Qualifier("outboundJDKSslContext") SSLContext clientSSLContext, AuthenticationService authenticationService, MetricResolver metricResolver,
+                    BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
+                    SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
+                    BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
+        Balancer balancer = new Balancer(applicationContext, serverSslContext, clientSSLContext, authenticationService, metricResolver, balancerProperties,
+                        zookeeperProperties, securityProperties, sslServerProperties, sslClientProperties, balancerServerProperties, balancerHttpProperties,
+                        balancerWebsocketProperties);
+        balancer.start();
+        return balancer;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public HealthChecker healthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
+        HealthChecker healthChecker = new HealthChecker(balancerProperties, balancerServerProperties);
+        healthChecker.start();
+        return healthChecker;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
+                    HealthChecker healthChecker) throws Exception {
+        MetricResolver metricResolver = new MetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
+        metricResolver.start();
+        return metricResolver;
+    }
 }

balancer/src/main/java/timely/balancer/configuration/BalancerProperties.java

@@ -1,5 +1,7 @@
 package timely.balancer.configuration;
 
+import java.util.List;
+
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.stereotype.Component;
 import org.springframework.validation.annotation.Validated;
@@ -13,6 +15,10 @@ public class BalancerProperties {
 
     private String assignmentFile;
 
+    private String defaultFs;
+
+    private List<String> fsConfigResources;
+
     private Double controlBandPercentage = 0.05;
 
     private long checkServerHealthInterval = 10000;
@@ -37,6 +43,22 @@ public class BalancerProperties {
         this.assignmentFile = assignmentFile;
     }
 
+    public String getDefaultFs() {
+        return defaultFs;
+    }
+
+    public void setDefaultFs(String defaultFs) {
+        this.defaultFs = defaultFs;
+    }
+
+    public List<String> getFsConfigResources() {
+        return fsConfigResources;
+    }
+
+    public void setFsConfigResources(List<String> fsConfigResources) {
+        this.fsConfigResources = fsConfigResources;
+    }
+
     public void setControlBandPercentage(Double controlBandPercentage) {
         this.controlBandPercentage = controlBandPercentage;
     }

balancer/src/main/java/timely/balancer/configuration/SpringBootstrap.java

@@ -1,6 +0,0 @@
-package timely.balancer.configuration;
-
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-
-@SpringBootApplication(scanBasePackageClasses = {SpringBootstrap.class})
-public class SpringBootstrap {}

balancer/src/main/java/timely/balancer/connection/http/HttpClientFactory.java

@@ -8,17 +8,15 @@ import org.apache.commons.pool2.impl.DefaultPooledObject;
 import org.apache.http.impl.client.CloseableHttpClient;
 
 import timely.balancer.connection.TimelyBalancedHost;
-import timely.common.configuration.SecurityProperties;
+import timely.client.http.HttpClient;
 import timely.common.configuration.SslClientProperties;
 
 public class HttpClientFactory implements KeyedPooledObjectFactory<TimelyBalancedHost,CloseableHttpClient> {
 
     private final SSLContext sslContext;
-    private final SecurityProperties security;
     private SslClientProperties sslClientProperties;
 
-    public HttpClientFactory(SecurityProperties security, SslClientProperties sslClientProperties, SSLContext sslContext) {
-        this.security = security;
+    public HttpClientFactory(SSLContext sslContext, SslClientProperties sslClientProperties) {
         this.sslContext = sslContext;
         this.sslClientProperties = sslClientProperties;
     }
@@ -26,8 +24,8 @@ public class HttpClientFactory implements KeyedPooledObjectFactory<TimelyBalance
     @Override
     public PooledObject<CloseableHttpClient> makeObject(TimelyBalancedHost k) throws Exception {
         // disable cookie management because we are sharing connections
-        return new DefaultPooledObject<>(timely.client.http.HttpClient.get(this.sslContext, null, sslClientProperties.isHostVerificationEnabled(),
-                        sslClientProperties.isUseClientCert()));
+        return new DefaultPooledObject<>(
+                        HttpClient.get(this.sslContext, null, sslClientProperties.isHostVerificationEnabled(), sslClientProperties.isUseClientCert()));
     }
 
     @Override

balancer/src/main/java/timely/balancer/connection/http/HttpClientPool.java

@@ -7,12 +7,11 @@ import org.apache.http.impl.client.CloseableHttpClient;
 
 import timely.balancer.configuration.BalancerHttpProperties;
 import timely.balancer.connection.TimelyBalancedHost;
-import timely.common.configuration.SecurityProperties;
 import timely.common.configuration.SslClientProperties;
 
 public class HttpClientPool extends GenericKeyedObjectPool<TimelyBalancedHost,CloseableHttpClient> {
 
-    public HttpClientPool(SecurityProperties securityProperties, SslClientProperties sslClientProperties, BalancerHttpProperties http, SSLContext sslContext) {
-        super(new HttpClientFactory(securityProperties, sslClientProperties, sslContext), http.getHttpClientPool());
+    public HttpClientPool(BalancerHttpProperties http, SSLContext sslContext, SslClientProperties sslClientProperties) {
+        super(new HttpClientFactory(sslContext, sslClientProperties), http.getHttpClientPool());
     }
 }

balancer/src/main/java/timely/balancer/netty/http/HttpRelayHandler.java

@@ -43,7 +43,7 @@ import timely.api.response.TimelyException;
 import timely.auth.TimelyAuthenticationToken;
 import timely.auth.util.HttpHeaderUtils;
 import timely.auth.util.ProxiedEntityUtils;
-import timely.balancer.component.MetricResolver;
+import timely.balancer.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.http.HttpClientPool;
 import timely.netty.http.TimelyHttpHandler;

balancer/src/main/java/timely/balancer/netty/tcp/TcpRelayHandler.java

@@ -14,7 +14,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import timely.api.request.MetricRequest;
 import timely.api.request.TcpRequest;
-import timely.balancer.component.MetricResolver;
+import timely.balancer.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.tcp.TcpClientPool;
 import timely.client.tcp.TcpClient;

balancer/src/main/java/timely/balancer/netty/udp/UdpRelayHandler.java

@@ -12,7 +12,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import timely.api.request.MetricRequest;
 import timely.api.request.UdpRequest;
-import timely.balancer.component.MetricResolver;
+import timely.balancer.MetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.udp.UdpClientPool;
 import timely.client.udp.UdpClient;

balancer/src/main/java/timely/balancer/netty/ws/WsRelayHandler.java

@@ -28,7 +28,7 @@ import timely.api.request.websocket.RemoveSubscription;
 import timely.api.request.websocket.SubscriptionRequest;
 import timely.api.response.TimelyException;
 import timely.auth.util.ProxiedEntityUtils;
-import timely.balancer.component.MetricResolver;
+import timely.balancer.MetricResolver;
 import timely.balancer.configuration.BalancerWebsocketProperties;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.balancer.connection.ws.WsClientPool;

balancer/src/main/resources/banner.txt

@@ -0,0 +1,11 @@
+===============================================================================
+  _______ _                _         ____        _
+ |__   __(_)              | |       |  _ \      | |
+    | |   _ _ __ ___   ___| |_   _  | |_) | __ _| | __ _ _ __   ___ ___ _ __
+    | |  | | '_ ` _ \ / _ \ | | | | |  _ < / _` | |/ _` | '_ \ / __/ _ \ '__|
+    | |  | | | | | | |  __/ | |_| | | |_) | (_| | | (_| | | | | (_|  __/ |
+    |_|  |_|_| |_| |_|\___|_|\__, | |____/ \__,_|_|\__,_|_| |_|\___\___|_|
+                              __/ |
+                             |___/
+===============================================================================
+TimelyBalancer ${application.formatted-version}

balancer/src/main/spotbugs/excludes.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <FindBugsFilter>
   <Match>
-    <Class name="timely.balancer.component.Balancer" />
+    <Class name="timely.balancer.Balancer" />
     <Method name="shutdown" />
     <Bug pattern="UC_USELESS_OBJECT" />
   </Match>

balancer/src/test/java/timely/balancer/component/TestBalancer.java

@@ -1,33 +0,0 @@
-package timely.balancer.component;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.annotation.Primary;
-import org.springframework.stereotype.Component;
-
-import timely.balancer.configuration.BalancerHttpProperties;
-import timely.balancer.configuration.BalancerProperties;
-import timely.balancer.configuration.BalancerServerProperties;
-import timely.balancer.configuration.BalancerWebsocketProperties;
-import timely.common.component.AuthenticationService;
-import timely.common.configuration.SecurityProperties;
-import timely.common.configuration.SslClientProperties;
-import timely.common.configuration.SslServerProperties;
-import timely.common.configuration.ZookeeperProperties;
-
-@Primary
-@Component
-public class TestBalancer extends Balancer {
-    private static final Logger log = LoggerFactory.getLogger(TestBalancer.class);
-
-    public TestBalancer(ApplicationContext applicationContext, AuthenticationService authenticationService, MetricResolver metricResolver,
-                    BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
-                    SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
-                    BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
-        super(applicationContext, authenticationService, metricResolver, balancerProperties, zookeeperProperties, securityProperties, sslServerProperties,
-                        sslClientProperties, balancerServerProperties, balancerHttpProperties, balancerWebsocketProperties);
-        DEFAULT_EVENT_LOOP_THREADS = 1;
-        log.info("Starting TestBalancer");
-    }
-}

balancer/src/test/java/timely/balancer/component/TestTimelyAuthenticationManager.java

@@ -1,20 +0,0 @@
-package timely.balancer.component;
-
-import java.util.Collections;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.annotation.Primary;
-import org.springframework.security.authentication.ProviderManager;
-import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationProvider;
-import org.springframework.stereotype.Component;
-
-@Primary
-@Component
-public class TestTimelyAuthenticationManager extends ProviderManager {
-
-    @Autowired
-    public TestTimelyAuthenticationManager(@Qualifier("file") PreAuthenticatedAuthenticationProvider file) {
-        super(Collections.singletonList(file));
-    }
-}

balancer/src/test/java/timely/balancer/configuration/TestBalancerConfiguration.java

@@ -0,0 +1,56 @@
+package timely.balancer.configuration;
+
+import javax.net.ssl.SSLContext;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+
+import io.netty.handler.ssl.SslContext;
+import timely.balancer.HealthChecker;
+import timely.balancer.MetricResolver;
+import timely.balancer.test.TestBalancer;
+import timely.balancer.test.TestHealthChecker;
+import timely.balancer.test.TestMetricResolver;
+import timely.common.component.AuthenticationService;
+import timely.common.configuration.CacheProperties;
+import timely.common.configuration.SecurityProperties;
+import timely.common.configuration.SslClientProperties;
+import timely.common.configuration.SslServerProperties;
+import timely.common.configuration.ZookeeperProperties;
+
+@Configuration
+@ComponentScan(basePackages = {"timely.balancer", "timely.common", "timely.test"})
+public class TestBalancerConfiguration {
+
+    @Bean
+    public TestBalancer balancer(ApplicationContext applicationContext, @Qualifier("nettySslContext") SslContext serverSslContext,
+                    @Qualifier("outboundJDKSslContext") SSLContext clientSSLContext, AuthenticationService authenticationService, MetricResolver metricResolver,
+                    BalancerProperties balancerProperties, ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties,
+                    SslServerProperties sslServerProperties, SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties,
+                    BalancerHttpProperties balancerHttpProperties, BalancerWebsocketProperties balancerWebsocketProperties) {
+        TestBalancer balancer = new TestBalancer(applicationContext, serverSslContext, clientSSLContext, authenticationService, metricResolver,
+                        balancerProperties, zookeeperProperties, securityProperties, sslServerProperties, sslClientProperties, balancerServerProperties,
+                        balancerHttpProperties, balancerWebsocketProperties);
+        balancer.start();
+        return balancer;
+    }
+
+    @Bean
+    public TestHealthChecker healthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
+        TestHealthChecker healthChecker = new TestHealthChecker(balancerProperties, balancerServerProperties);
+        healthChecker.start();
+        return healthChecker;
+    }
+
+    @Bean
+    public TestMetricResolver metricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
+                    HealthChecker healthChecker) throws Exception {
+        TestMetricResolver metricResolver = new TestMetricResolver(curatorFramework, balancerProperties, cacheProperties, healthChecker);
+        metricResolver.start();
+        return metricResolver;
+    }
+}

balancer/src/test/java/timely/balancer/configuration/TestConfiguration.java

@@ -1,21 +0,0 @@
-package timely.balancer.configuration;
-
-import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-import timely.common.configuration.AccumuloProperties;
-
-@Configuration
-public class TestConfiguration {
-
-    @Bean
-    @ConditionalOnMissingBean
-    public AccumuloProperties accumuloProperties() {
-        AccumuloProperties accumuloProperties = new AccumuloProperties();
-        accumuloProperties.setInstanceName("default");
-        accumuloProperties.setUsername("root");
-        accumuloProperties.setPassword("secret");
-        return accumuloProperties;
-    }
-}

balancer/src/test/java/timely/balancer/configuration/TestCuratorConfiguration.java

@@ -1,49 +0,0 @@
-package timely.balancer.configuration;
-
-import static org.apache.accumulo.core.conf.ConfigurationTypeHelper.getTimeInMillis;
-
-import org.apache.curator.RetryPolicy;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.retry.RetryForever;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.DependsOn;
-
-import timely.common.configuration.ZookeeperProperties;
-
-/*
- * The purpose of this class is to wait for the creation of either ZooKeeperServerMain (in-memory)
- * or MiniAccumuloCluster (mini accumulo) before creating a CuratorFramework so that the components
- * that need CuratorFramework autowired will wait.
- */
-@Configuration
-public class TestCuratorConfiguration {
-
-    private static final Logger log = LoggerFactory.getLogger(TestCuratorConfiguration.class);
-
-    @Bean(destroyMethod = "close")
-    @ConditionalOnProperty(name = "timely.in-memory-accumulo.enabled", havingValue = "true")
-    @DependsOn("ZooKeeperServerMain")
-    public CuratorFramework inMemoryCuratorFramework(ZookeeperProperties zookeeperProperties) {
-        RetryPolicy retryPolicy = new RetryForever(1000);
-        int timeout = Long.valueOf(getTimeInMillis(zookeeperProperties.getTimeout())).intValue();
-        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServers(), timeout, 10000, retryPolicy);
-        curatorFramework.start();
-        return curatorFramework;
-    }
-
-    @Bean(destroyMethod = "close")
-    @ConditionalOnProperty(name = "timely.mini-accumulo.enabled", havingValue = "true")
-    @DependsOn("MiniAccumuloCluster")
-    public CuratorFramework miniAccumuloCuratorFramework(ZookeeperProperties zookeeperProperties) {
-        RetryPolicy retryPolicy = new RetryForever(1000);
-        int timeout = Long.valueOf(getTimeInMillis(zookeeperProperties.getTimeout())).intValue();
-        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServers(), timeout, 10000, retryPolicy);
-        curatorFramework.start();
-        return curatorFramework;
-    }
-}

balancer/src/test/java/timely/balancer/configuration/TestWatcherConfiguration.java

@@ -1,23 +0,0 @@
-package timely.balancer.configuration;
-
-import org.apache.accumulo.core.client.AccumuloClient;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Scope;
-
-import timely.balancer.test.BalancerTestRule;
-import timely.common.configuration.MiniAccumuloProperties;
-import timely.common.configuration.TimelyProperties;
-
-@Configuration
-@EnableConfigurationProperties({MiniAccumuloProperties.class, TimelyProperties.class})
-public class TestWatcherConfiguration {
-
-    @Bean
-    @Scope("prototype")
-    public BalancerTestRule timelyTestWatcherAccumulo(AccumuloClient accumuloClient, TimelyProperties timelyProperties) throws Exception {
-
-        return new BalancerTestRule(timelyProperties, accumuloClient);
-    }
-}

balancer/src/test/java/timely/balancer/integration/MetricResolverIT.java

@@ -1,4 +1,4 @@
-package timely.balancer.test;
+package timely.balancer.integration;
 
 import static timely.Constants.SERVICE_DISCOVERY_PATH;
 
@@ -27,10 +27,11 @@ import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
 
 import timely.ServerDetails;
-import timely.balancer.component.TestHealthChecker;
-import timely.balancer.component.TestMetricResolver;
 import timely.balancer.connection.TimelyBalancedHost;
+import timely.balancer.test.TestHealthChecker;
+import timely.balancer.test.TestMetricResolver;
 import timely.test.IntegrationTest;
+import timely.test.TimelyTestRule;
 
 @Category(IntegrationTest.class)
 @RunWith(SpringRunner.class)
@@ -41,7 +42,7 @@ public class MetricResolverIT {
 
     @Autowired
     @Rule
-    public BalancerTestRule balancerTestRule;
+    public TimelyTestRule testRule;
 
     @Autowired
     private TestMetricResolver metricResolver;

balancer/src/test/java/timely/balancer/test/BalancerTestRule.java

@@ -1,61 +0,0 @@
-package timely.balancer.test;
-
-import org.apache.accumulo.core.client.AccumuloClient;
-import org.junit.rules.TestWatcher;
-import org.junit.runner.Description;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import timely.common.configuration.TimelyProperties;
-import timely.util.TestUtils;
-
-public class BalancerTestRule extends TestWatcher {
-
-    private static final Logger log = LoggerFactory.getLogger(BalancerTestRule.class);
-    private TimelyProperties timelyProperties;
-    private AccumuloClient accumuloClient;
-
-    public BalancerTestRule(TimelyProperties timelyProperties, AccumuloClient accumuloClient) {
-        this.timelyProperties = timelyProperties;
-        this.accumuloClient = accumuloClient;
-    }
-
-    @Override
-    protected void failed(Throwable e, Description description) {
-        log.error("Test {}:{} failed", description.getTestClass().getCanonicalName(), description.getMethodName());
-        log.error(e.getMessage(), e);
-        if (accumuloClient != null) {
-            if (accumuloClient.tableOperations().exists(timelyProperties.getMetricsTable())) {
-                separator();
-                log.info("Contents of {}", timelyProperties.getMetricsTable());
-                separator();
-                TestUtils.printTable(accumuloClient, timelyProperties.getMetricsTable(), true);
-                separator();
-            }
-            if (accumuloClient.tableOperations().exists(timelyProperties.getMetaTable())) {
-                log.info("Contents of {}", timelyProperties.getMetaTable());
-                separator();
-                TestUtils.printTable(accumuloClient, timelyProperties.getMetaTable(), false);
-                separator();
-            }
-        }
-    }
-
-    private void separator() {
-        log.info("-----------------------------------------------------------------------------------");
-    }
-
-    @Override
-    protected void starting(Description description) {
-        separator();
-        log.info("Starting test {}:{}", description.getTestClass().getCanonicalName(), description.getMethodName());
-        separator();
-    }
-
-    @Override
-    protected void finished(Description description) {
-        separator();
-        log.info("Finished test {}:{}", description.getTestClass().getCanonicalName(), description.getMethodName());
-        separator();
-    }
-}

balancer/src/test/java/timely/balancer/test/TestBalancer.java

@@ -0,0 +1,36 @@
+package timely.balancer.test;
+
+import javax.net.ssl.SSLContext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationContext;
+
+import io.netty.handler.ssl.SslContext;
+import timely.balancer.Balancer;
+import timely.balancer.MetricResolver;
+import timely.balancer.configuration.BalancerHttpProperties;
+import timely.balancer.configuration.BalancerProperties;
+import timely.balancer.configuration.BalancerServerProperties;
+import timely.balancer.configuration.BalancerWebsocketProperties;
+import timely.common.component.AuthenticationService;
+import timely.common.configuration.SecurityProperties;
+import timely.common.configuration.SslClientProperties;
+import timely.common.configuration.SslServerProperties;
+import timely.common.configuration.ZookeeperProperties;
+
+public class TestBalancer extends Balancer {
+    private static final Logger log = LoggerFactory.getLogger(TestBalancer.class);
+
+    public TestBalancer(ApplicationContext applicationContext, SslContext serverSslContext, SSLContext clientSSLContext,
+                    AuthenticationService authenticationService, MetricResolver metricResolver, BalancerProperties balancerProperties,
+                    ZookeeperProperties zookeeperProperties, SecurityProperties securityProperties, SslServerProperties sslServerProperties,
+                    SslClientProperties sslClientProperties, BalancerServerProperties balancerServerProperties, BalancerHttpProperties balancerHttpProperties,
+                    BalancerWebsocketProperties balancerWebsocketProperties) {
+        super(applicationContext, serverSslContext, clientSSLContext, authenticationService, metricResolver, balancerProperties, zookeeperProperties,
+                        securityProperties, sslServerProperties, sslClientProperties, balancerServerProperties, balancerHttpProperties,
+                        balancerWebsocketProperties);
+        DEFAULT_EVENT_LOOP_THREADS = 1;
+        log.info("Starting TestBalancer");
+    }
+}

balancer/src/test/java/timely/balancer/test/TestHealthChecker.java

@@ -1,32 +1,18 @@
-package timely.balancer.component;
+package timely.balancer.test;
 
 import java.util.List;
 
-import org.springframework.context.annotation.Primary;
-import org.springframework.stereotype.Component;
-
+import timely.balancer.HealthChecker;
 import timely.balancer.configuration.BalancerProperties;
 import timely.balancer.configuration.BalancerServerProperties;
 import timely.balancer.connection.TimelyBalancedHost;
 
-@Primary
-@Component
 public class TestHealthChecker extends HealthChecker {
 
     public TestHealthChecker(BalancerProperties balancerProperties, BalancerServerProperties balancerServerProperties) {
         super(balancerProperties, balancerServerProperties);
     }
 
-    @Override
-    public void setup() {
-
-    }
-
-    @Override
-    public void close() {
-
-    }
-
     public List<TimelyBalancedHost> getTimelyHosts() {
         return this.timelyHosts;
     }

balancer/src/test/java/timely/balancer/test/TestMetricResolver.java

@@ -1,4 +1,4 @@
-package timely.balancer.component;
+package timely.balancer.test;
 
 import java.util.List;
 
@@ -8,21 +8,19 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.context.annotation.Primary;
-import org.springframework.stereotype.Component;
 
+import timely.balancer.HealthChecker;
+import timely.balancer.MetricResolver;
 import timely.balancer.configuration.BalancerProperties;
 import timely.balancer.connection.TimelyBalancedHost;
 import timely.common.configuration.CacheProperties;
 
-@Primary
-@Component
 public class TestMetricResolver extends MetricResolver {
 
     private static final Logger log = LoggerFactory.getLogger(TestMetricResolver.class);
 
     public TestMetricResolver(CuratorFramework curatorFramework, BalancerProperties balancerProperties, CacheProperties cacheProperties,
-                    HealthChecker healthChecker) {
+                    HealthChecker healthChecker) throws Exception {
         super(curatorFramework, balancerProperties, cacheProperties, healthChecker);
     }
 

balancer/src/test/java/timely/balancer/test/TestServerInfo.java

@@ -1,48 +0,0 @@
-package timely.balancer.test;
-
-public class TestServerInfo {
-
-    public String host;
-    public String ip;
-    public Integer tcpPort;
-    public Integer httpPort;
-    public Integer wsPort;
-    public Integer udpPort;
-
-    public TestServerInfo(String host, String ip, Integer tcpPort, Integer httpPort, Integer wsPort, Integer udpPort) {
-        this.host = host;
-        this.ip = ip;
-        this.tcpPort = tcpPort;
-        this.httpPort = httpPort;
-        this.wsPort = wsPort;
-        this.udpPort = udpPort;
-    }
-
-    static public TestServerInfo of(String host, String ip, Integer tcpPort, Integer httpPort, Integer wsPort, Integer udpPort) {
-        return new TestServerInfo(host, ip, tcpPort, httpPort, wsPort, udpPort);
-    }
-
-    public String getHost() {
-        return host;
-    }
-
-    public String getIp() {
-        return ip;
-    }
-
-    public Integer getTcpPort() {
-        return tcpPort;
-    }
-
-    public Integer getHttpPort() {
-        return httpPort;
-    }
-
-    public Integer getWsPort() {
-        return wsPort;
-    }
-
-    public Integer getUdpPort() {
-        return udpPort;
-    }
-}

balancer/src/test/java/timely/balancer/test/TestTimelyAuthenticationManager.java

@@ -1,4 +1,4 @@
-package timely.common.component;
+package timely.balancer.test;
 
 import java.util.Collections;
 
@@ -9,8 +9,8 @@ import org.springframework.security.authentication.ProviderManager;
 import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationProvider;
 import org.springframework.stereotype.Component;
 
-@Component
 @Primary
+@Component
 public class TestTimelyAuthenticationManager extends ProviderManager {
 
     @Autowired

balancer/src/test/resources/config/application.yml

@@ -6,6 +6,11 @@ logging:
 
 timely:
   test: true
+  instance: 1
+  accumulo:
+    instance-name: default
+    username: root
+    password: secret
   mini-accumulo:
     enabled: true
     rootAuths: A,B,C,D,E,F,G,H,I
@@ -18,74 +23,87 @@ timely:
     cache-refresh-minutes: 5
     allow-anonymous-http-access: true
     allow-anonymous-ws-access: true
-    authorizedUsers:
-      - subjectDn: cn=example.com
+    authorized-users:
+      - subject-dn: cn=example.com
         roles: G,H,I
         auths: D,E,F
-        userType: USER
-      - subjectDn: cn=test a. user, ou=example developers, o=example corp, c=us
-        issuerDn: cn=example corp ca, o=example corp, c=us
+        user-type: USER
+      - subject-dn: cn=test a. user, ou=example developers, o=example corp, c=us
+        issuer-dn: cn=example corp ca, o=example corp, c=us
         roles: G,H,I
         auths: A,B,C
-        userType: USER
-    serverSsl:
-      useGeneratedKeypair: true
-      ignoreSslHandshakeErrors: true
-      useOpenssl: true
+        user-type: USER
+    server-ssl:
+      key-store-type: PKCS12
+      key-store-file: classpath:ssl/testServer.p12
+      key-store-password: ChangeIt
+      trust-store-type: JKS
+      trust-store-file: classpath:ssl/ca.jks
+      trust-store-password: ChangeIt
+      use-generated-keypair: false
+      ignore-ssl-handshake-errors: true
+      use-openssl: true
+    client-ssl:
+      key-store-type: PKCS12
+      key-store-file: classpath:ssl/testUser.p12
+      key-store-password: ChangeIt
+      trust-store-type: JKS
+      trust-store-file: classpath:ssl/ca.jks
+      trust-store-password: ChangeIt
   cache:
     enabled: false
     metric-age-off-hours:
       default: 26
-    nonCachedMetrics:
+    non-cached-metrics:
   balancer:
-    checkServerHealthInterval: 10000
-    serverFailuresBeforeDown: 6
-    serverSuccessesBeforeUp: 1
-    assignmentFile: file://${java.io.tmpdir}/timely/balancer/assignments.csv
+    check-server-health-interval: 10000
+    server-failures-before-down: 6
+    server-successes-before-up: 1
+    assignment-file: file://${java.io.tmpdir}/timely/balancer/assignments.csv
     server:
       ip: 127.0.0.1
-      tcpPort: 4242
-      numTcpPools: 1
-      tcpBufferSize: 100
-      tcpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
+      tcp-base-port: 9242
+      num-tcp-pools: 1
+      tcp-buffer-size: 100
+      tcp-client-pool:
+        min-idle-per-key: 0
+        max-idle-per-key: 40
+        max-total-per-key: 40
+        max-total: -1
         lifo: true
         fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
-      udpPort: 4245
-      udpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
+        max-wait-millis: -1
+        min-evictable-idle-time-millis: 1800000
+        evictor-shutdown-timeout-millis: 10000
+        soft-min-evictable-idle-time-millis: -1
+        num-tests-per-eviction-run: 3
+        time-between-eviction-runs-millis: -1
+        block-when-exhausted: true
+        jmx-enabled: true
+        test-on-borrow: false
+        test-on-return: false
+      udp-base-port: 9245
+      udp-client-pool:
+        min-idle-per-key: 0
+        max-idle-per-key: 40
+        max-total-per-key: 40
+        max-total: -1
         lifo: true
         fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
+        max-wait-millis: -1
+        min-evictable-idle-time-millis: 1800000
+        evictor-shutdown-timeout-millis: 10000
+        soft-min-evictable-idle-time-millis: -1
+        num-tests-per-eviction-run: 3
+        time-between-eviction-runs-millis: -1
+        block-when-exhausted: true
+        jmx-enabled: true
+        test-on-borrow: false
+        test-on-return: false
     http:
       host: localhost
       ip: 127.0.0.1
-      port: 4243
+      base-port: 9243
       redirect-path: /secure-me
       strict-transport-max-age: 604800
       cors:
@@ -95,47 +113,47 @@ timely:
         allowed-origins:
         allowed-methods: DELETE,GET,HEAD,OPTIONS,PUT,POST
         allowed-headers: content-type
-      httpClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
+      http-client-pool:
+        min-idle-per-key: 0
+        max-idle-per-key: 40
+        max-total-per-key: 40
+        max-total: -1
         lifo: true
         fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
+        max-wait-millis: -1
+        min-evictable-idle-time-millis: 1800000
+        evictor-shutdown-timeout-millis: 10000
+        soft-min-evictable-idle-time-millis: -1
+        num-tests-per-eviction-run: 3
+        time-between-eviction-runs-millis: -1
+        block-when-exhausted: true
+        jmx-enabled: true
+        test-on-borrow: false
+        test-on-return: false
     websocket:
       ip: 127.0.0.1
-      port: 5244
+      base-port: 9244
       timeout: 60
       subscription-lag: 120
       scanner-batch-size: 10000
       flush-interval-seconds: 30
       scanner-read-ahead: 1
       subscription-batch-size: 5000
-      incomingBufferSize: 50000000
-      wsClientPool:
-        minIdlePerKey: 0
-        maxIdlePerKey: 40
-        maxTotalPerKey: 40
-        maxTotal: -1
+      incoming-buffer-size: 50000000
+      ws-client-pool:
+        min-idle-per-key: 0
+        max-idle-per-key: 40
+        max-total-per-key: 40
+        max-total: -1
         lifo: true
         fairness: false
-        maxWaitMillis: -1
-        minEvictableIdleTimeMillis: 1800000
-        evictorShutdownTimeoutMillis: 10000
-        softMinEvictableIdleTimeMillis: -1
-        numTestsPerEvictionRun: 3
-        timeBetweenEvictionRunsMillis: -1
-        blockWhenExhausted: true
-        jmxEnabled: true
-        testOnBorrow: false
-        testOnReturn: false
+        max-wait-millis: -1
+        min-evictable-idle-time-millis: 1800000
+        evictor-shutdown-timeout-millis: 10000
+        soft-min-evictable-idle-time-millis: -1
+        num-tests-per-eviction-run: 3
+        time-between-eviction-runs-millis: -1
+        block-when-exhausted: true
+        jmx-enabled: true
+        test-on-borrow: false
+        test-on-return: false

balancer/src/test/resources/ssl/ca.cert

@@ -0,0 +1,23 @@
+Bag Attributes: <No Attributes>
+subject=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
+issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
+-----BEGIN CERTIFICATE-----
+MIIDTjCCAjagAwIBAgIURUj8l+ms/i9jeabIBBi19KxEhAgwDQYJKoZIhvcNAQEL
+BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
+AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
+NTAwWjA+MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRgwFgYD
+VQQDEw9FWEFNUExFIENPUlAgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
+AoIBAQDVBr2XF20SKmGXfvp2BUY44BF1fm3Qw83GkC66qIHgpgQxzJ8u44P34vV1
+DYTj4XzrBpx23J0nCMEiAW8YW9MxZJPEokZafRNKFQAngVdgmAW9fHemHRQfaYwe
+ySVz2UZ+z28DJ3C0QshvqXskrYstwLVs4Ss2uoX+XpeFbW9IVVg4aW+GX3LHPNTk
+7Hji13Capp6BCiS61AGQAiMWNBW2aRgHVLOb7RDQug2Bka/dcJ44XxJDh26C6kV1
+pNEZm0ToqWAiVfvyzionPec5im6NKX9kStZc+C/iikfx5vdmrvGIF5zxy6phdhJ7
+AGd6cF6UBkkYm/SsIP3WTXpBek5LAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAP
+BgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBSPLNlCmmp9PH+crzCx9TNp5BoCajAN
+BgkqhkiG9w0BAQsFAAOCAQEAkjLnZKqLMjbu5Hy8YOVoKaEKahP+t/tNYZW/NtrC
+xSxKmGLz5kVCkjpLgCykkcpu4h2wZW4MGZJ4nc87+tN22mJOg1WnqEpBzRsEPajk
+7duVjdU2FDCarERj53hxF41l/u8BxvMGbFnrAijv5SzspeLi+VCGF3wu53Bi6UHi
+7WO1dIhZLMHepqgtwbDDudxwlKbG/JWjJ7OspCNqOpktpY8BFUAQ42ctYZcOHy2w
+kdZDcK5IhRztkBH2aSWOhudYh/CF7BOu2nSUdtR5NwhXnsStmSbud7ahw3W5Jc9/
+IVAFXf2brkrfIzwrY5kmjnEPJ8mjUyLepwSlOlMTBrehzQ==
+-----END CERTIFICATE-----

balancer/src/test/resources/ssl/testServer.cert

@@ -0,0 +1,51 @@
+Bag Attributes
+    localKeyID: 22 89 1C EA 4E FF 9A F6 11 1B E7 9B 5E 4F 59 B7 B1 0F AA FA 
+    friendlyName: example-server
+subject=/C=US/O=Example Corp/OU=Servers/CN=testserver.example.com
+issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
+-----BEGIN CERTIFICATE-----
+MIIDwjCCAqqgAwIBAgIUe6yBgkY1m/oXuPXA9j/c/7+w16UwDQYJKoZIhvcNAQEL
+BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
+AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
+NTAwWjBXMQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRAwDgYD
+VQQLEwdTZXJ2ZXJzMR8wHQYDVQQDExZ0ZXN0c2VydmVyLmV4YW1wbGUuY29tMIIB
+IjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAmb1pEWg9a6ZChQULazK75ppc
+XScU5Ap29L70/5zcaOcbaBwqKhpKhdZCoZ1TD+Sdkk9kp7IlabCQjM2vVjknb6NY
+msl3PbQFNva0EkH6s5cvpX4WhQDsWK77/4eztHT+0fpaG1YoKQn+Gy3VuW0CDyR6
+GzGKifnI/aNLevgUBebbiiTPLEdGlo0jp3SXlZfoDZdbvnhBWI8Xc+aDekNlrXxW
+X2TVnc02Fi914PECYJzkMpmxnBRPAcljU24WSqaUVgRoSbafkm4OXDw0xtvxnIrJ
+s3iO0uZfRf7+OxAlqlvh4ZJ9/PufPMEKvmr6zea05KkFjBotKMip5Md43H3lQQID
+AQABo4GcMIGZMA4GA1UdDwEB/wQEAwIFoDAdBgNVHSUEFjAUBggrBgEFBQcDAQYI
+KwYBBQUHAwIwDAYDVR0TAQH/BAIwADAdBgNVHQ4EFgQUDWwMrl9/uc1wplERWxas
+bDROB4MwHwYDVR0jBBgwFoAUjyzZQppqfTx/nK8wsfUzaeQaAmowGgYDVR0RBBMw
+EYIJbG9jYWxob3N0hwR/AAABMA0GCSqGSIb3DQEBCwUAA4IBAQBfG7StgQf3WNk/
+EnaYuurwvaOOpcOV5NDcZ6tmgaXUrISPVDpyIWfsDGetNEXc+4gpefLQtGF2wuiW
+odiLc4H9JwQ/yEkVYwzrFViPdZKVvxNU2EjiwO/e/3sl+SxxkTDy6o838uTXX+X5
+Q8DabmL5L8ukcnqUHVOqW38oA1vRNBKiJX58JzOr8ezX8crZmqU9h8cNwG/V+ZXE
+5U7+hSRzLm7gRwtixj8ZsTDUiGJxOdbyUKImPClt1Se9TizB2Q/xOPTdmYB1qiiB
++XVKkGwQPgIbHckLsZWhBor2u5cgdrBGW09Q+H88maagwq5sBHNMs33vkkgbUDpB
+6WSEBJMi
+-----END CERTIFICATE-----
+Bag Attributes: <No Attributes>
+subject=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
+issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
+-----BEGIN CERTIFICATE-----
+MIIDTjCCAjagAwIBAgIURUj8l+ms/i9jeabIBBi19KxEhAgwDQYJKoZIhvcNAQEL
+BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
+AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
+NTAwWjA+MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRgwFgYD
+VQQDEw9FWEFNUExFIENPUlAgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
+AoIBAQDVBr2XF20SKmGXfvp2BUY44BF1fm3Qw83GkC66qIHgpgQxzJ8u44P34vV1
+DYTj4XzrBpx23J0nCMEiAW8YW9MxZJPEokZafRNKFQAngVdgmAW9fHemHRQfaYwe
+ySVz2UZ+z28DJ3C0QshvqXskrYstwLVs4Ss2uoX+XpeFbW9IVVg4aW+GX3LHPNTk
+7Hji13Capp6BCiS61AGQAiMWNBW2aRgHVLOb7RDQug2Bka/dcJ44XxJDh26C6kV1
+pNEZm0ToqWAiVfvyzionPec5im6NKX9kStZc+C/iikfx5vdmrvGIF5zxy6phdhJ7
+AGd6cF6UBkkYm/SsIP3WTXpBek5LAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAP
+BgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBSPLNlCmmp9PH+crzCx9TNp5BoCajAN
+BgkqhkiG9w0BAQsFAAOCAQEAkjLnZKqLMjbu5Hy8YOVoKaEKahP+t/tNYZW/NtrC
+xSxKmGLz5kVCkjpLgCykkcpu4h2wZW4MGZJ4nc87+tN22mJOg1WnqEpBzRsEPajk
+7duVjdU2FDCarERj53hxF41l/u8BxvMGbFnrAijv5SzspeLi+VCGF3wu53Bi6UHi
+7WO1dIhZLMHepqgtwbDDudxwlKbG/JWjJ7OspCNqOpktpY8BFUAQ42ctYZcOHy2w
+kdZDcK5IhRztkBH2aSWOhudYh/CF7BOu2nSUdtR5NwhXnsStmSbud7ahw3W5Jc9/
+IVAFXf2brkrfIzwrY5kmjnEPJ8mjUyLepwSlOlMTBrehzQ==
+-----END CERTIFICATE-----

balancer/src/test/resources/ssl/testServer.key

@@ -0,0 +1,32 @@
+Bag Attributes
+    localKeyID: 22 89 1C EA 4E FF 9A F6 11 1B E7 9B 5E 4F 59 B7 B1 0F AA FA 
+    friendlyName: example-server
+Key Attributes: <No Attributes>
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCZvWkRaD1rpkKF
+BQtrMrvmmlxdJxTkCnb0vvT/nNxo5xtoHCoqGkqF1kKhnVMP5J2ST2SnsiVpsJCM
+za9WOSdvo1iayXc9tAU29rQSQfqzly+lfhaFAOxYrvv/h7O0dP7R+lobVigpCf4b
+LdW5bQIPJHobMYqJ+cj9o0t6+BQF5tuKJM8sR0aWjSOndJeVl+gNl1u+eEFYjxdz
+5oN6Q2WtfFZfZNWdzTYWL3Xg8QJgnOQymbGcFE8ByWNTbhZKppRWBGhJtp+Sbg5c
+PDTG2/GcismzeI7S5l9F/v47ECWqW+Hhkn38+588wQq+avrN5rTkqQWMGi0oyKnk
+x3jcfeVBAgMBAAECggEAZEHC7NkfhL4NjxV1Ftkk9AMhzJWZ6ARzArQsrSysB7T+
+iGwliwNOKe1J0n74FdsubqVbrIheIYhFG61edEodT44kT/ylyBVHeUp0r5tbVBZ5
+ueB56w5pEoQ6Ta8/H6VUL7alYEJVOj4L+clwINXXi+xrvKQYBlZgHheJN43j8NxS
+9lCrVcm58rzX3i29S7ErBf7RqDabgD6j2m7rehnSq5Ps2wxwV/LXMl4Lxlwa0jz/
+LV8szNMZViPOWAqOCy7NZbwCCaCfKVUAwmNGBkE3p4JVPeLNLzCBY9G0rHVmFJLK
+yPnCB7PtCZeFTsU+8OHHWoxMFsaOO9Pp8FtB/r8PYQKBgQDI0OzEo3t/QKvbwAdz
+9Eze0LQvstzZIidTWbYMTxH0g7DIt+3p11d61GKtfeuYUTPlLodufdv/BaP8G9sI
+NzxisB3uH/787M9fwi1OzO7zMMZZCVa7Q1vZk6ta1qY3PekcYMPqjBJhxP1X3S7x
+8+mrW5vTq/l8ZRVFeLvEkh6bpQKBgQDD/MCRJoq6nzd2DoTyT9m5qO+Jlztmu9+T
+vqVOg7q0NO7U7dzyMbWgMvMFoEBidnyllU1rEj2YOo4OrI2v/JmSPUVS8VXsBkkG
+UqInV4F/g0Wp3hzrkZFMqHLa6AydugmRNkOpgurRKOiwAL0Pgvk8+bOIbKMM7JBE
+tZ6Zvi0gbQKBgD4S5r0505jYQsdgqm0bXQ+a77e20DGGLyXCm1Ves6wjcOf6gt4/
+pFjhgccjH0or/6CeK1rqKLDB65ioe6IkgNP4oKfU2BsSkF+HRD4RduNRdl5bVqX4
+AlJVJSilJSf6SS5ucuDzf1MdczpPso0z2k0PujFJ/YqILHUXBuOST0fJAoGAZAdX
+nX3HZpKmsumcLzRUEq9HerkmRzvyU60FH/qZyyoND75Le42lfpsicCenVNnWqoh2
+WASaTVztbk3hjwXkt7/v5vJoASwAD2sTefLulE9Cuf00U4MbXr3GyTWQYtw1z9ib
+ViQMhy5S/KmT1Rvq65UibanM2mbPB1KPHTb0P00CgYEAgieJ1YXYYHPjABWJ7/S9
+3Z/8eJY0kuUe1DRFaKOlfej8SOCFFIa90IW2mAnujh55H4N2lww4s/N+FIRQoQpA
+1Hrch9tYRiXyouQc67dFlLxmCOVJIf31ZUt7I4HcPhsfr4fuCJXUPBs1jaDmwQsk
+90VWrXaTwSjmP+yHMiPfaA0=
+-----END PRIVATE KEY-----

balancer/src/test/resources/ssl/testUser.cert

@@ -0,0 +1,50 @@
+Bag Attributes
+    localKeyID: 57 F6 9F DF B2 94 61 47 B0 B0 2E 3A F7 79 98 C8 15 5C 64 FA 
+    friendlyName: example-client
+subject=/C=US/O=Example Corp/OU=Example Developers/CN=Test A. User
+issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
+-----BEGIN CERTIFICATE-----
+MIIDmzCCAoOgAwIBAgIUFp4IuIaZ2XId7UNvzP6ZB9hChyUwDQYJKoZIhvcNAQEL
+BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
+AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
+NTAwWjBYMQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRswGQYD
+VQQLExJFeGFtcGxlIERldmVsb3BlcnMxFTATBgNVBAMTDFRlc3QgQS4gVXNlcjCC
+ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAMOMeN0ElMDxYKST2VD1wP6Z
+jmy543Zv/QgRLTYizg3U9t3uZU0C3HqUJnK+c4+OJnRIyHuMXuJWEYVdairyAdYW
+ZDF5sfiL9ikG9rZ2WxR2z7nsYLJcDGwKAt/Duzuoed6KSccEpiQc+BkoOTs9p87I
+l4s7saX/anESMUY7YOomLSqS2zvQZh/zTjIdP1/9kG9PLzXPClJELuiC3pJCmr7Y
+xdJIXecu6YKFpApy0y+rKcfYUVPML15gyRu9pIqH3U39y7mCkn+YP9Qc9GcZAoqR
+iYdUEl1rFAgV/a3o0pmf1tFFRDOsoPSMUGNNQlaCjv1uZIbRm25nvvQlPfstFDEC
+AwEAAaN1MHMwDgYDVR0PAQH/BAQDAgWgMBMGA1UdJQQMMAoGCCsGAQUFBwMCMAwG
+A1UdEwEB/wQCMAAwHQYDVR0OBBYEFPAe3n59DcNwRcnpIwQY2kPawbCVMB8GA1Ud
+IwQYMBaAFI8s2UKaan08f5yvMLH1M2nkGgJqMA0GCSqGSIb3DQEBCwUAA4IBAQBo
++zadU9DwR1rnHTBZF6CJXmiIdrQmArojH4a3NOHbNjFjRXm7dkd/tAy5LtgDJkww
+yg+ZrOyOKHbFgk6IXRZ8nw2b1Ck9I9MDW1vxfjid9YLW2nhDaMcIeTXz3aIBO5i6
+gMCV+xMcFaU/O57Ohia+ADUqrqu14nG9Moi171UWPbMr5N3gYn3Fs4VrTGYs4i0Z
+5x2itm88WB3bWj+cd06zKQopU9b0Y76mQv9h+w4EHxq4uWA7GmIDDkvtI61MQo8+
+twdX159Ms6o4lW43SRqVBHeMgSJbiLDkdkypYJvd9PvAjWgmQK/sD1SYKxfgYhtx
+hqIvAjlGeDwdf+55EGjf
+-----END CERTIFICATE-----
+Bag Attributes: <No Attributes>
+subject=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
+issuer=/C=US/O=Example Corp/CN=EXAMPLE CORP CA
+-----BEGIN CERTIFICATE-----
+MIIDTjCCAjagAwIBAgIURUj8l+ms/i9jeabIBBi19KxEhAgwDQYJKoZIhvcNAQEL
+BQAwPjELMAkGA1UEBhMCVVMxFTATBgNVBAoTDEV4YW1wbGUgQ29ycDEYMBYGA1UE
+AxMPRVhBTVBMRSBDT1JQIENBMCAXDTIzMDIwMTIzMTUwMFoYDzIxMjMwMTA4MjMx
+NTAwWjA+MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRXhhbXBsZSBDb3JwMRgwFgYD
+VQQDEw9FWEFNUExFIENPUlAgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK
+AoIBAQDVBr2XF20SKmGXfvp2BUY44BF1fm3Qw83GkC66qIHgpgQxzJ8u44P34vV1
+DYTj4XzrBpx23J0nCMEiAW8YW9MxZJPEokZafRNKFQAngVdgmAW9fHemHRQfaYwe
+ySVz2UZ+z28DJ3C0QshvqXskrYstwLVs4Ss2uoX+XpeFbW9IVVg4aW+GX3LHPNTk
+7Hji13Capp6BCiS61AGQAiMWNBW2aRgHVLOb7RDQug2Bka/dcJ44XxJDh26C6kV1
+pNEZm0ToqWAiVfvyzionPec5im6NKX9kStZc+C/iikfx5vdmrvGIF5zxy6phdhJ7
+AGd6cF6UBkkYm/SsIP3WTXpBek5LAgMBAAGjQjBAMA4GA1UdDwEB/wQEAwIBBjAP
+BgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBSPLNlCmmp9PH+crzCx9TNp5BoCajAN
+BgkqhkiG9w0BAQsFAAOCAQEAkjLnZKqLMjbu5Hy8YOVoKaEKahP+t/tNYZW/NtrC
+xSxKmGLz5kVCkjpLgCykkcpu4h2wZW4MGZJ4nc87+tN22mJOg1WnqEpBzRsEPajk
+7duVjdU2FDCarERj53hxF41l/u8BxvMGbFnrAijv5SzspeLi+VCGF3wu53Bi6UHi
+7WO1dIhZLMHepqgtwbDDudxwlKbG/JWjJ7OspCNqOpktpY8BFUAQ42ctYZcOHy2w
+kdZDcK5IhRztkBH2aSWOhudYh/CF7BOu2nSUdtR5NwhXnsStmSbud7ahw3W5Jc9/
+IVAFXf2brkrfIzwrY5kmjnEPJ8mjUyLepwSlOlMTBrehzQ==
+-----END CERTIFICATE-----

balancer/src/test/resources/ssl/testUser.key

@@ -0,0 +1,32 @@
+Bag Attributes
+    localKeyID: 57 F6 9F DF B2 94 61 47 B0 B0 2E 3A F7 79 98 C8 15 5C 64 FA 
+    friendlyName: example-client
+Key Attributes: <No Attributes>
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDDjHjdBJTA8WCk
+k9lQ9cD+mY5sueN2b/0IES02Is4N1Pbd7mVNAtx6lCZyvnOPjiZ0SMh7jF7iVhGF
+XWoq8gHWFmQxebH4i/YpBva2dlsUds+57GCyXAxsCgLfw7s7qHneiknHBKYkHPgZ
+KDk7PafOyJeLO7Gl/2pxEjFGO2DqJi0qkts70GYf804yHT9f/ZBvTy81zwpSRC7o
+gt6SQpq+2MXSSF3nLumChaQKctMvqynH2FFTzC9eYMkbvaSKh91N/cu5gpJ/mD/U
+HPRnGQKKkYmHVBJdaxQIFf2t6NKZn9bRRUQzrKD0jFBjTUJWgo79bmSG0ZtuZ770
+JT37LRQxAgMBAAECggEBALFWsT1vLCAWRxi55Hkhhhbe12UO55O/V+969E+GHOw6
+WG9IBaJjl+Oqe3vP8CkeNw15/CbfVCVa/KS/5Mpxkuy+fbrIrHfFJkZqxE8Q4uSa
+IyRvk/nstG8+XQDlFuWxkmQZcXMFak6R3wWhjtWEwqwrChQA3Mi366YDY6BXtTpi
+MuLJVWXqYVF2RirWX3a7eosR4MYBCJ1S8NnpI1w4oZwuew6iYiCJNkURctxnGKOe
+hoop/711PEyIb/qHETBIlReY9scvZgJQZMgGe5mTGgFmAA8Bzo4e1/F2QrVIi7wy
+REOSj4U+tP/QrYMSPk0cMpH5ft5+2mfh/o+D1ugm/AkCgYEA2qt3P4vlm1TMmJru
+vMYc8iZjfe7UQTnD+J1EgrASBI0JP565zSZKBRIXnheumWhWS+2KR1MP8N6KAu4Q
+xE/E7TKfLuE1CPzn7Rh/WZAxDqGNeQthk0qTbeBZ4/mHPKvt8V7c+N4vLjZ00uul
+bKitlLpb49T+8FpSgRmk+nrBsesCgYEA5O6LmxH1+ip5NoMHoZVBhLbgJkxoXEzn
+xHT7On8CjQ9B0T85NW9O38/a2gyIb5dIavcrh4dMvOKX8hdrVlal3chrXFRemyvy
+VzpW7HZpMwthJb/QSim+GQ4u2Jsm+eXQAswq/yfowls9mp7K81McBgdfcdWgWuqc
+Ttf3dBRS71MCgYBCPpu7ppgCfKgWuykEC0iNm0+cwROQRv21fX9Z0BlFQ6y/l0hW
+e/kEnDyHakk8tD0gd2/it6kmoWRSCukNb9UKvsE2pIc6l9NzIm7CJUqXUh3Xw7jl
+tDwBl2vkAp4TpUpKSyDKTBUi/OBbXYQZ7OfZHLPly4sIKXTBI5G1SCcRzQKBgE3Y
+Y7ZmRBs+CBumMJkcVLwZJ10+jBoeuIzQLoF0oNxwfLP3aK+n2s02M7gDnP09Fw3w
+YGfkdQ6hioNyvTgNnLcuqHsr4UOojO5mUxpzJVCLJQyeDei6Cv8gGIPlNjqxj6rz
+3AcJfhdgyTDjoCVfdF8EcMsXlkB0GJSi0W0yYAUrAoGAVK5v0cbEEV+gxdQgh848
+FwSQXx7i9jVOB2BUTLX2Q+gLUJibtFzLnh03sGCXtTa8sK+hLW4FCMWZa9Ev5ZjG
+ylFyA96U97CNpS+2/ZW5+hvRZ/GuY/0yB8w26BeS38lI8RnmxSPi5cnexdl32NIX
+5FkU9WJ4HKmi1n0304CkF+4=
+-----END PRIVATE KEY-----

client/pom.xml

@@ -4,13 +4,13 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0</version>
   </parent>
   <artifactId>timely-client</artifactId>
   <properties>
     <!-- PEP 440 compliant version gets substituted into src/main/python/template/setup.py -->
     <!-- use .dev0 instead of -SNAPSHOT -->
-    <python.client.version>2.1.0.dev0</python.client.version>
+    <python.client.version>2.2.0</python.client.version>
   </properties>
   <dependencies>
     <dependency>

client/src/main/java/timely/client/tcp/TcpClient.java

@@ -18,18 +18,27 @@ public class TcpClient implements AutoCloseable {
     private Socket sock = null;
     private BufferedWriter out = null;
     private long connectTime = 0L;
-    private long backoff = 2000;
-    private int bufferSize = -1;
+    private long INIT_BACKOFF = 1000;
+    private long MAX_BACKOFF = 60000;
+    private long backoff = INIT_BACKOFF;
+    private int bufferSize;
+    private long latency;
     private int writesSinceFlush = 0;
+    private long lastFlushTime = System.currentTimeMillis();
 
     public TcpClient(String hostname, int port) {
         this(hostname, port, -1);
     }
 
     public TcpClient(String hostname, int port, int bufferSize) {
+        this(hostname, port, bufferSize, -1);
+    }
+
+    public TcpClient(String hostname, int port, int bufferSize, long latency) {
         this.host = hostname;
         this.port = port;
         this.bufferSize = bufferSize;
+        this.latency = latency;
     }
 
     /**
@@ -54,20 +63,54 @@ public class TcpClient implements AutoCloseable {
         if (connect() != 0) {
             throw new IOException();
         }
-        out.write(metric);
-        writesSinceFlush++;
-        if (bufferSize > 0 && writesSinceFlush >= bufferSize) {
-            flush();
+        try {
+            out.write(metric);
+            writesSinceFlush++;
+            if (bufferSize > 0 && writesSinceFlush >= bufferSize) {
+                if (log.isTraceEnabled()) {
+                    log.trace(String.format("Flushing buffer writesSinceFlush >= %d", writesSinceFlush));
+                }
+                flush();
+            }
+            if (latency > 0 && (System.currentTimeMillis() - lastFlushTime) >= latency) {
+                if (log.isTraceEnabled()) {
+                    log.trace(String.format("Flushing buffer timeSinceFlush >= %d", latency));
+                }
+                flush();
+            }
+        } catch (IOException e) {
+            log.error(e.getMessage(), e);
+            reset();
+            throw e;
         }
     }
 
     public synchronized void flush() throws IOException {
         if (null != out) {
-            out.flush();
-            writesSinceFlush = 0;
+            try {
+                long start = System.currentTimeMillis();
+                out.flush();
+                lastFlushTime = System.currentTimeMillis();
+                if (log.isTraceEnabled()) {
+                    log.trace(String.format("Flushed %d metrics to socket in %d ms", writesSinceFlush, (lastFlushTime - start)));
+                }
+                writesSinceFlush = 0;
+            } catch (IOException e) {
+                log.error(String.format("Failed to flush %d metrics to socket", writesSinceFlush), e);
+                reset();
+                throw e;
+            }
         }
     }
 
+    private synchronized void reset() {
+        out = null;
+        sock = null;
+        backoff = INIT_BACKOFF;
+        writesSinceFlush = 0;
+        lastFlushTime = System.currentTimeMillis();
+    }
+
     /**
      * Closes the tcp connection to Timely
      *
@@ -75,8 +118,8 @@ public class TcpClient implements AutoCloseable {
      *             if an error occurs
      */
     @Override
-    public void close() throws IOException {
-        log.trace("Shutting down connection to Timely at {}:{}", host, port);
+    public synchronized void close() throws IOException {
+        log.info(String.format("Shutting down connection to Timely at %s:%d", host, port));
         if (null != sock) {
             try {
                 if (null != out) {
@@ -87,7 +130,7 @@ public class TcpClient implements AutoCloseable {
                 }
                 sock.close();
             } catch (IOException e) {
-                log.error("Error closing connection to Timely at " + host + ":" + port + ". Error: " + e.getMessage());
+                log.error(String.format("Error closing connection to Timely at %s:%d", host, port), e);
             } finally {
                 sock = null;
             }
@@ -103,10 +146,10 @@ public class TcpClient implements AutoCloseable {
                     sock = new Socket(host, port);
                     osw = new OutputStreamWriter(sock.getOutputStream(), StandardCharsets.UTF_8);
                     out = new BufferedWriter(osw);
-                    backoff = 2000;
-                    log.trace("Connected to Timely at {}:{}", host, port);
+                    backoff = INIT_BACKOFF;
+                    log.info(String.format("Connected to Timely at %s:%d", host, port));
                 } catch (Exception e) {
-                    log.error("Error connecting to Timely at {}:{} - {}", host, port, e.getMessage());
+                    log.error(String.format("Error connecting to Timely at %s:%d", host, port), e);
                     if (sock != null) {
                         try {
                             sock.close();
@@ -132,12 +175,13 @@ public class TcpClient implements AutoCloseable {
                             out = null;
                         }
                     }
-                    backoff = backoff * 2;
-                    log.info("Will retry connection in {} ms.", backoff);
+                    backoff = Math.min(MAX_BACKOFF, backoff * 2);
+                    if (log.isTraceEnabled()) {
+                        log.trace(String.format("Will retry connection in %d ms", backoff));
+                    }
                     return -1;
                 }
             } else {
-                log.warn("Not writing to Timely, waiting to reconnect");
                 return -1;
             }
         }
@@ -147,4 +191,22 @@ public class TcpClient implements AutoCloseable {
     public long getConnectTime() {
         return connectTime;
     }
+
+    public int getWritesSinceFlush() {
+        return writesSinceFlush;
+    }
+
+    public long getTimeSinceFlush() {
+        return System.currentTimeMillis() - lastFlushTime;
+    }
+
+    public synchronized boolean isConnected() {
+        boolean isConnected = (null != sock && sock.isConnected());
+        if (isConnected) {
+            return true;
+        } else {
+            int status = connect();
+            return status == 0;
+        }
+    }
 }

collectd-nsq-plugin/pom.xml

@@ -1,59 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-  <modelVersion>4.0.0</modelVersion>
-  <parent>
-    <groupId>timely</groupId>
-    <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
-  </parent>
-  <artifactId>collectd-nsq-plugin</artifactId>
-  <dependencies>
-    <dependency>
-      <groupId>collectd</groupId>
-      <artifactId>collectd-api</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.commons</groupId>
-      <artifactId>commons-pool2</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.httpcomponents</groupId>
-      <artifactId>httpclient</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>timely</groupId>
-      <artifactId>collectd-plugin-parent</artifactId>
-    </dependency>
-  </dependencies>
-  <build>
-    <plugins>
-      <plugin>
-        <groupId>org.apache.maven.plugins</groupId>
-        <artifactId>maven-shade-plugin</artifactId>
-        <configuration>
-          <artifactSet>
-            <includes />
-          </artifactSet>
-          <filters>
-            <filter>
-              <artifact>*:*</artifact>
-              <excludes>
-                <exclude>**/module-info.class</exclude>
-              </excludes>
-            </filter>
-          </filters>
-          <createDependencyReducedPom>false</createDependencyReducedPom>
-        </configuration>
-        <executions>
-          <execution>
-            <id>shade</id>
-            <goals>
-              <goal>shade</goal>
-            </goals>
-            <phase>package</phase>
-          </execution>
-        </executions>
-      </plugin>
-    </plugins>
-  </build>
-</project>

collectd-plugin-parent/pom.xml

@@ -1,32 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-  <modelVersion>4.0.0</modelVersion>
-  <parent>
-    <groupId>timely</groupId>
-    <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
-  </parent>
-  <artifactId>collectd-plugin-parent</artifactId>
-  <dependencies>
-    <dependency>
-      <groupId>collectd</groupId>
-      <artifactId>collectd-api</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>net.jcip</groupId>
-      <artifactId>jcip-annotations</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.powermock</groupId>
-      <artifactId>powermock-easymock-release-full</artifactId>
-      <classifier>full</classifier>
-      <scope>test</scope>
-    </dependency>
-  </dependencies>
-</project>

collectd-plugin-parent/src/main/java/timely/collectd/plugin/CollectDPluginParent.java

@@ -1,254 +0,0 @@
-package timely.collectd.plugin;
-
-import java.io.OutputStream;
-import java.text.MessageFormat;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import org.collectd.api.Collectd;
-import org.collectd.api.DataSource;
-import org.collectd.api.OConfigItem;
-import org.collectd.api.ValueList;
-
-public abstract class CollectDPluginParent {
-
-    private static final String PUT = "put {0} {1} {2}{3}\n";
-    private static final Pattern HADOOP_STATSD_PATTERN = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
-    private static final String INSTANCE = " instance=";
-    private static final String NAME = " name=";
-    private static final String SAMPLE = " sample=";
-    private static final String CODE = " code=";
-    private static final String PERIOD = ".";
-    private static final String SAMPLE_TYPE = " sampleType=";
-    private static final String COUNTER = "COUNTER";
-    private static final String GAUGE = "GAUGE";
-    private static final String DERIVE = "DERIVE";
-    private static final String ABSOLUTE = "ABSOLUTE";
-    private static final Pattern NSQ_PATTERN1 = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
-    private static final Pattern NSQ_PATTERN2 = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
-    private static final Pattern NSQ_PATTERN3 = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_#]+)\\.([\\w-_]+)");
-    private static final Pattern ETHSTAT_PATTERN = Pattern.compile("([\\w-_]+)_queue_([\\w-_]+)_([\\w-_]+)");
-    private static final String STATSD_PREFIX = "statsd";
-    private static final String NSQ_PREFIX = STATSD_PREFIX + ".nsq.";
-    private static final Pattern HAPROXY_PATTERN = Pattern.compile("\\[([\\w-_=]+),([\\w-_=]+)\\]");
-
-    private final SMARTCodeMapping smart = new SMARTCodeMapping();
-
-    protected Set<String> addlTags = new HashSet<>();
-
-    public int config(OConfigItem config) {
-        for (OConfigItem child : config.getChildren()) {
-            switch (child.getKey()) {
-                case "Tags":
-                case "tags":
-                    String[] additionalTags = child.getValues().get(0).getString().split(",");
-                    for (String t : additionalTags) {
-                        addlTags.add(" " + t.trim());
-                    }
-                default:
-            }
-        }
-        return 0;
-    }
-
-    public abstract void write(String metric, OutputStream out);
-
-    public void process(ValueList vl, OutputStream out) {
-        debugValueList(vl);
-
-        StringBuilder metric = new StringBuilder();
-        Long timestamp = vl.getTime();
-        StringBuilder tags = new StringBuilder();
-        String host = vl.getHost();
-        int idx = host.indexOf(PERIOD);
-        if (-1 != idx) {
-            tags.append(" host=").append(host.substring(0, idx));
-        } else {
-            tags.append(" host=").append(host);
-        }
-        int nIdx = host.indexOf('n');
-        if (-1 != nIdx) {
-            tags.append(" rack=").append(host.substring(0, nIdx));
-        }
-        for (String tag : addlTags) {
-            tags.append(tag);
-        }
-        if (vl.getPlugin().equals(STATSD_PREFIX)) {
-            Matcher m = HADOOP_STATSD_PATTERN.matcher(vl.getTypeInstance());
-            Matcher n1 = NSQ_PATTERN1.matcher(vl.getTypeInstance());
-            Matcher n2 = NSQ_PATTERN2.matcher(vl.getTypeInstance());
-            Matcher n3 = NSQ_PATTERN3.matcher(vl.getTypeInstance());
-            String instance = null;
-            if (m.matches() && !vl.getTypeInstance().startsWith("nsq")) {
-                // Here we are processing the statsd metrics coming from the
-                // Hadoop Metrics2 StatsDSink without the host name.
-                // The format of metric is:
-                // serviceName.contextName.recordName.metricName. The recordName
-                // is typically duplicative and is dropped here. The serviceName
-                // is used as the instance.
-                metric.append(STATSD_PREFIX).append(PERIOD).append(m.group(2)).append(PERIOD).append(m.group(4));
-                instance = m.group(1);
-            } else if (n1.matches() && vl.getTypeInstance().startsWith("nsq")) {
-                metric.append(NSQ_PREFIX).append(n1.group(2)).append(PERIOD).append(n1.group(3));
-            } else if (n2.matches() && vl.getTypeInstance().startsWith("nsq")) {
-                metric.append(NSQ_PREFIX).append(n2.group(2)).append(PERIOD).append(n2.group(4));
-                instance = n2.group(3);
-            } else if (n3.matches() && vl.getTypeInstance().startsWith("nsq")) {
-                metric.append(NSQ_PREFIX).append(n3.group(4)).append(PERIOD).append(n3.group(6));
-                instance = n3.group(5);
-            } else {
-                // Handle StatsD metrics of unknown formats. If there is a
-                // period in the metric name, use everything up to that as
-                // the instance.
-                int period = vl.getTypeInstance().indexOf('.');
-                if (-1 == period) {
-                    metric.append(STATSD_PREFIX).append(PERIOD).append(vl.getTypeInstance());
-                } else {
-                    instance = vl.getTypeInstance().substring(0, period);
-                    metric.append(STATSD_PREFIX).append(PERIOD).append(vl.getTypeInstance().substring(period + 1));
-                }
-            }
-            timestamp = vl.getTime();
-            if (null != instance) {
-                tags.append(INSTANCE).append(instance);
-            }
-        } else if (vl.getPlugin().equals("ethstat")) {
-            metric.append("sys.ethstat.");
-            if (vl.getTypeInstance().contains("queue")) {
-                Matcher m = ETHSTAT_PATTERN.matcher(vl.getTypeInstance());
-                if (m.matches()) {
-                    metric.append(m.group(1)).append("_").append(m.group(3));
-                    tags.append(" queue=").append(m.group(2));
-                } else {
-                    metric.append(vl.getTypeInstance());
-                }
-            } else {
-                metric.append(vl.getTypeInstance());
-            }
-            tags.append(INSTANCE).append(vl.getPluginInstance());
-        } else if (vl.getPlugin().equals("hddtemp")) {
-            metric.append("sys.hddtemp.").append(vl.getType());
-            tags.append(INSTANCE).append(vl.getTypeInstance());
-        } else if (vl.getPlugin().equals("smart")) {
-            int code = -1;
-            String name = null;
-            if (vl.getTypeInstance().startsWith("attribute-")) {
-                int hyphen = vl.getTypeInstance().indexOf('-');
-                code = Integer.parseInt(vl.getTypeInstance().substring(hyphen + 1));
-                name = smart.get(code);
-            }
-            if (code == -1) {
-                if (notEmpty(vl.getTypeInstance())) {
-                    metric.append("sys.smart.").append(vl.getTypeInstance());
-                } else {
-                    metric.append("sys.smart.").append(vl.getType());
-                }
-            } else {
-                metric.append("sys.smart.").append(name);
-                tags.append(CODE).append(code);
-            }
-            tags.append(INSTANCE).append(vl.getPluginInstance());
-        } else if (vl.getPlugin().equals("sensors")) {
-            String instance = "";
-            if (vl.getTypeInstance().startsWith("temp")) {
-                instance = vl.getTypeInstance().substring(4);
-            }
-            metric.append("sys.sensors.").append(vl.getType()).append(PERIOD).append(vl.getPluginInstance());
-            tags.append(INSTANCE).append(instance);
-        } else if (vl.getPlugin().equals("haproxy")) {
-            metric.append("sys.haproxy.").append(vl.getTypeInstance());
-            Matcher m = HAPROXY_PATTERN.matcher(vl.getPluginInstance());
-            if (m.matches()) {
-                tags.append(" ").append(m.group(1));
-                tags.append(" ").append(m.group(2));
-            }
-        } else if (vl.getPlugin().equals("ipmi")) {
-            metric.append("sys.ipmi.").append(vl.getType());
-            tags.append(INSTANCE).append(vl.getTypeInstance().replaceAll(" ", "_"));
-        } else if (vl.getPlugin().equals("snmp")) {
-            metric.append("sys.snmp.").append(vl.getType());
-            tags.append(INSTANCE).append(vl.getTypeInstance().replaceAll(" ", "_"));
-        } else if (vl.getPlugin().equals("GenericJMX")) {
-            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD).append(vl.getTypeInstance());
-            String[] pluginInstanceSplit = vl.getPluginInstance().split("-");
-            if (pluginInstanceSplit.length > 0) {
-                tags.append(INSTANCE).append(pluginInstanceSplit[0].replaceAll(" ", "_"));
-            }
-            if (pluginInstanceSplit.length > 1) {
-                tags.append(NAME).append(pluginInstanceSplit[1].replaceAll(" ", "_"));
-            }
-        } else if (notEmpty(vl.getTypeInstance()) && notEmpty(vl.getType()) && notEmpty(vl.getPluginInstance()) && notEmpty(vl.getPlugin())) {
-            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD).append(vl.getTypeInstance());
-            tags.append(INSTANCE).append(vl.getPluginInstance().replaceAll(" ", "_"));
-        } else if (notEmpty(vl.getTypeInstance()) && notEmpty(vl.getType()) && notEmpty(vl.getPlugin())) {
-            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType()).append(PERIOD).append(vl.getTypeInstance());
-        } else if (notEmpty(vl.getType()) && notEmpty(vl.getPluginInstance()) && notEmpty(vl.getPlugin())) {
-            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType());
-            tags.append(INSTANCE).append(vl.getPluginInstance().replaceAll(" ", "_"));
-        } else if (notEmpty(vl.getType()) && notEmpty(vl.getPlugin())) {
-            metric.append("sys.").append(vl.getPlugin()).append(PERIOD).append(vl.getType());
-        } else {
-            Collectd.logWarning("Unhandled metric: " + vl.toString());
-            return;
-        }
-        final String metricName = metric.toString().replaceAll(" ", "_");
-        for (int i = 0; i < vl.getValues().size(); i++) {
-            StringBuilder tagsWithSample = new StringBuilder(tags.toString());
-            String sampleName = vl.getDataSet().getDataSources().get(i).getName();
-            int type = vl.getDataSet().getDataSources().get(i).getType();
-            String sampleType = convertType(type);
-            if (null != sampleName) {
-                tagsWithSample.append(SAMPLE).append(sampleName);
-            }
-            if (null != sampleType) {
-                tagsWithSample.append(SAMPLE_TYPE).append(sampleType);
-            }
-            Double value = vl.getValues().get(i).doubleValue();
-            String datapoint = MessageFormat.format(PUT, metricName, timestamp.toString(), value.toString(), tagsWithSample.toString());
-            Collectd.logDebug("Writing: " + datapoint);
-            write(datapoint, out);
-        }
-    }
-
-    private void debugValueList(ValueList vl) {
-        Collectd.logDebug("Input: " + vl.toString());
-        Collectd.logDebug("Plugin: " + vl.getPlugin());
-        Collectd.logDebug("PluginInstance: " + vl.getPluginInstance());
-        Collectd.logDebug("Type: " + vl.getType());
-        Collectd.logDebug("TypeInstance: " + vl.getTypeInstance());
-        for (int i = 0; i < vl.getValues().size(); i++) {
-            Number value = vl.getValues().get(i);
-            DataSource ds = vl.getDataSet().getDataSources().get(i);
-            Collectd.logDebug(convertType(ds.getType()) + " " + ds.getName() + " = " + value);
-        }
-    }
-
-    private String convertType(int type) {
-        String result = null;
-        switch (type) {
-            case 0:
-                result = COUNTER;
-                break;
-            case 1:
-                result = GAUGE;
-                break;
-            case 2:
-                result = DERIVE;
-                break;
-            case 3:
-                result = ABSOLUTE;
-                break;
-            default:
-                result = GAUGE;
-                break;
-        }
-        return result;
-    }
-
-    private boolean notEmpty(String arg) {
-        return (null != arg) && !(arg.equals(""));
-    }
-
-}

collectd-plugin-parent/src/main/java/timely/collectd/plugin/PooledCloseableHttpClientFactory.java

@@ -0,0 +1,97 @@
+package timely.collectd.plugin;
+
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.PooledObjectFactory;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.collectd.api.OConfigItem;
+
+import java.util.concurrent.TimeUnit;
+
+public class PooledCloseableHttpClientFactory implements PooledObjectFactory<CloseableHttpClient> {
+
+    private int connectionRequestTimeout = 10000; // 10 seconds in milliseconds
+    private int connectTimeout = 30000; // 30 seconds in milliseconds
+    private int socketTimeout = 60000; // 60 seconds in milliseconds
+    private long connectionTimeToLive = 300000; // 300 seconds in milliseconds
+
+    public int config(OConfigItem config) {
+        for (OConfigItem child : config.getChildren()) {
+            switch (child.getKey()) {
+                case "connectionRequestTimeout":
+                case "ConnectionRequestTimeout":
+                    connectionRequestTimeout = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "connectTimeout":
+                case "ConnectTimeout":
+                    connectTimeout = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "socketTimeout":
+                case "SocketTimeout":
+                    socketTimeout = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "connectionTimeToLive":
+                case "ConnectionTimeToLive":
+                    connectionTimeToLive = Long.parseLong(child.getValues().get(0).getString());
+                    break;
+                default:
+            }
+        }
+        return 0;
+    }
+
+    @Override
+    public PooledObject<CloseableHttpClient> makeObject() throws Exception {
+        RequestConfig requestConfig = RequestConfig.custom().setConnectionRequestTimeout(connectionRequestTimeout).setConnectTimeout(connectTimeout)
+                        .setSocketTimeout(socketTimeout).build();
+        HttpClientBuilder builder = HttpClientBuilder.create();
+        builder.setDefaultRequestConfig(requestConfig);
+        builder.setConnectionTimeToLive(connectionTimeToLive, TimeUnit.MILLISECONDS);
+        return new DefaultPooledObject<>(builder.build());
+    }
+
+    @Override
+    public void destroyObject(PooledObject<CloseableHttpClient> pooledObject) throws Exception {
+        try {
+            pooledObject.getObject().close();
+        } catch (Exception e) {
+            // do nothing
+        } finally {
+            pooledObject.invalidate();
+            ;
+        }
+    }
+
+    @Override
+    public boolean validateObject(PooledObject<CloseableHttpClient> pooledObject) {
+        return pooledObject.getObject() != null;
+    }
+
+    @Override
+    public void activateObject(PooledObject<CloseableHttpClient> pooledObject) throws Exception {
+        // do nothing
+    }
+
+    @Override
+    public void passivateObject(PooledObject<CloseableHttpClient> pooledObject) throws Exception {
+        // do nothing
+    }
+
+    public void setConnectionRequestTimeout(int connectionRequestTimeout) {
+        this.connectionRequestTimeout = connectionRequestTimeout;
+    }
+
+    public void setConnectTimeout(int connectTimeout) {
+        this.connectTimeout = connectTimeout;
+    }
+
+    public void setSocketTimeout(int socketTimeout) {
+        this.socketTimeout = socketTimeout;
+    }
+
+    public void setConnectionTimeToLive(long connectionTimeToLive) {
+        this.connectionTimeToLive = connectionTimeToLive;
+    }
+}

collectd-plugin-parent/src/main/java/timely/collectd/plugin/PooledSocketFactory.java

@@ -1,13 +1,13 @@
 package timely.collectd.plugin;
 
-import java.net.Socket;
-
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.PooledObjectFactory;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
 import org.collectd.api.Collectd;
 import org.collectd.api.OConfigItem;
 
+import java.net.Socket;
+
 public class PooledSocketFactory implements PooledObjectFactory<Socket> {
 
     private String host = null;

collectd-plugin-parent/src/main/java/timely/collectd/plugin/WriteNSQPlugin.java

@@ -1,11 +1,5 @@
 package timely.collectd.plugin;
 
-import java.io.ByteArrayOutputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.util.Random;
-
 import org.apache.commons.pool2.impl.GenericObjectPool;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.http.HttpStatus;
@@ -21,6 +15,12 @@ import org.collectd.api.CollectdWriteInterface;
 import org.collectd.api.OConfigItem;
 import org.collectd.api.ValueList;
 
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Random;
+
 /**
  *
  * CollectD plugin that writes metrics collected by the StatsD plugin to NSQ. NSQ can be configured to write statsd metrics to CollectD. If you do this, set the
@@ -79,6 +79,7 @@ public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConf
                 } catch (IOException e1) {}
                 try {
                     clientPool.invalidateObject(client);
+                    client = null;
                 } catch (Exception e2) {}
             }
         } finally {

collectd-plugin-parent/src/main/java/timely/collectd/plugin/WriteTimelyPlugin.java

@@ -0,0 +1,99 @@
+package timely.collectd.plugin;
+
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.collectd.api.Collectd;
+import org.collectd.api.CollectdConfigInterface;
+import org.collectd.api.CollectdShutdownInterface;
+import org.collectd.api.CollectdWriteInterface;
+import org.collectd.api.OConfigItem;
+import org.collectd.api.ValueList;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.io.PrintWriter;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+
+/**
+ *
+ * CollectD plugin that writes metrics collected by the StatsD plugin to Timely.
+ *
+ */
+public class WriteTimelyPlugin extends CollectDPluginParent implements CollectdConfigInterface, CollectdShutdownInterface, CollectdWriteInterface {
+
+    private GenericObjectPool<Socket> socketPool = null;
+
+    // pool should be limited by the number of WriteThreads configured in
+    // collectd
+    static final private int POOL_MAX_SIZE = Integer.MAX_VALUE;
+
+    public WriteTimelyPlugin() {
+        Collectd.registerConfig(WriteTimelyPlugin.class.getName(), this);
+        Collectd.registerShutdown(WriteTimelyPlugin.class.getName(), this);
+        Collectd.registerWrite(WriteTimelyPlugin.class.getName(), this);
+    }
+
+    public int write(ValueList vl) {
+
+        Socket socket = null;
+        try {
+            socket = socketPool.borrowObject();
+            if (socket == null) {
+                return -1;
+            }
+            super.process(vl, socket.getOutputStream());
+        } catch (Exception e) {
+            Collectd.logWarning(e.getMessage());
+            try {
+                // close socket so that the object pool will discard it and
+                // reconnect
+                if (socket != null) {
+                    socket.close();
+                }
+            } catch (IOException e1) {
+                Collectd.logError(e1.getMessage());
+            }
+            return -1;
+        } finally {
+            if (socket != null) {
+                socketPool.returnObject(socket);
+            }
+        }
+        return 0;
+    }
+
+    @Override
+    public void write(String metric, OutputStream out) {
+        PrintWriter printWriter = new PrintWriter(new OutputStreamWriter(out, StandardCharsets.UTF_8), false);
+        printWriter.write(metric);
+        printWriter.flush();
+        if (printWriter.checkError()) {
+            throw new RuntimeException("Error writing to Timely");
+        }
+    }
+
+    public int shutdown() {
+        socketPool.close();
+        return 0;
+    }
+
+    public int config(OConfigItem config) {
+        super.config(config);
+        int retval = 0;
+        PooledSocketFactory socketFactory = new PooledSocketFactory();
+        retval = socketFactory.config(config);
+        if (retval == 0) {
+            GenericObjectPoolConfig<Socket> poolConfig = new GenericObjectPoolConfig<>();
+            // use max size for maxTotal and maxIdle
+            // no need to activate and passivate
+            poolConfig.setMaxTotal(POOL_MAX_SIZE);
+            poolConfig.setMaxIdle(POOL_MAX_SIZE);
+            poolConfig.setTestOnReturn(true);
+            socketPool = new GenericObjectPool<>(socketFactory, poolConfig);
+        }
+        return retval;
+    }
+
+}

collectd-plugin-parent/src/test/java/timely/collectd/plugin/WriteTimelyPluginTest.java

@@ -1,12 +1,6 @@
 package timely.collectd.plugin;
 
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.util.Collections;
-
+import net.jcip.annotations.NotThreadSafe;
 import org.collectd.api.Collectd;
 import org.collectd.api.DataSet;
 import org.collectd.api.DataSource;
@@ -21,7 +15,12 @@ import org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
-import net.jcip.annotations.NotThreadSafe;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.Collections;
 
 @RunWith(PowerMockRunner.class)
 @PrepareForTest({Collectd.class})

collectd-timely-plugin/src/main/spotbugs/excludes.xml

@@ -1,8 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<FindBugsFilter>
-  <Match>
-    <Bug pattern="DM_DEFAULT_ENCODING"/>
-    <Method name="connect" />
-    <Class name="timely.collectd.plugin.WriteTimelyPlugin"/>
-  </Match>
-</FindBugsFilter>

collectd/pom.xml

@@ -4,9 +4,9 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0</version>
   </parent>
-  <artifactId>collectd-timely-plugin</artifactId>
+  <artifactId>timely-collectd</artifactId>
   <dependencies>
     <dependency>
       <groupId>collectd</groupId>
@@ -16,25 +16,23 @@
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-pool2</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpclient</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>jcl-over-slf4j</artifactId>
+    </dependency>
     <dependency>
       <groupId>timely</groupId>
-      <artifactId>collectd-plugin-parent</artifactId>
+      <artifactId>timely-common</artifactId>
     </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>net.jcip</groupId>
-      <artifactId>jcip-annotations</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.javassist</groupId>
-      <artifactId>javassist</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.powermock</groupId>
       <artifactId>powermock-easymock-release-full</artifactId>
@@ -49,7 +47,17 @@
         <artifactId>maven-shade-plugin</artifactId>
         <configuration>
           <artifactSet>
-            <includes />
+            <includes>
+              <include>com.google.guava:guava</include>
+              <include>commons-codec:commons-codec</include>
+              <include>org.apache.commons:commons-lang3</include>
+              <include>org.apache.commons:commons-pool2</include>
+              <include>org.apache.httpcomponents:httpclient</include>
+              <include>org.apache.httpcomponents:httpcore</include>
+              <include>org.slf4j:jcl-over-slf4j</include>
+              <include>org.slf4j:slf4j-api</include>
+              <include>timely:timely-common</include>
+            </includes>
           </artifactSet>
           <filters>
             <filter>

collectd/src/main/java/timely/collectd/plugin/CollectDPluginParent.java

@@ -0,0 +1,424 @@
+package timely.collectd.plugin;
+
+import java.io.OutputStream;
+import java.text.MessageFormat;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang3.StringUtils;
+import org.collectd.api.Collectd;
+import org.collectd.api.OConfigItem;
+import org.collectd.api.ValueList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import timely.util.Exclusions;
+
+public abstract class CollectDPluginParent {
+
+    private static final String PUT = "put {0} {1} {2} {3}";
+    private static final String INSTANCE = "instance";
+    private static final String NAME = "name";
+    private static final String SAMPLE_TYPE = "sampleType";
+    private static final String CODE = "code";
+    private static final String PERIOD = ".";
+    private static final String COUNTER = "COUNTER";
+    private static final String GAUGE = "GAUGE";
+    private static final String DERIVE = "DERIVE";
+    private static final String ABSOLUTE = "ABSOLUTE";
+    private static final Collection<String> HADOOP_CONTEXTS = Arrays.asList("NameNode", "DataNode", "JobManager", "NodeManager", "JobHistoryServer",
+                    "ResourceManager", "MRAppMaster");
+    private static final Pattern HADOOP_PATTERN_4_GROUPS = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
+    private static final Pattern STATSD_PATTERN_3_GROUPS = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
+    private static final Pattern STATSD_PATTERN_4_GROUPS = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)");
+    private static final Pattern STATSD_PATTERN_6_GROUPS = Pattern.compile("([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_]+)\\.([\\w-_#]+)\\.([\\w-_]+)");
+    private static final Pattern ETHSTAT_QUEUE_PATTERN1 = Pattern.compile("([\\w-_]*)queue_([0-9]+)_([\\w-_]+)");
+    private static final Pattern ETHSTAT_QUEUE_PATTERN2 = Pattern.compile("([\\w-_\\.]*)(tx|rx)([-_]?)([0-9]+)([._])([\\w-_]+)");
+    private static final Pattern ETHSTAT_TRAFFIC_CLASS_PATTERN = Pattern.compile("([\\w-_\\.]*)(tc)([-_]?)([0-9]+)([._])([\\w-_]+)");
+    private static final Pattern ETHSTAT_CHANNEL_PATTERN = Pattern.compile("(ch)([0-9]+)([._])([\\w-_]+)");
+    private static final String STATSD_PREFIX = "statsd";
+    private static final String NSQ_PREFIX = STATSD_PREFIX + ".nsq.";
+    private static final Pattern HAPROXY_PATTERN = Pattern.compile("\\[([\\w-_=]+),([\\w-_=]+)\\]");
+    private static final Logger log = LoggerFactory.getLogger(CollectDPluginParent.class);
+
+    private final SMARTCodeMapping smart = new SMARTCodeMapping();
+    protected Map<String,String> addlTags = new TreeMap<>();
+    private boolean debug = false;
+    private Exclusions exclusions = new Exclusions();
+
+    public int config(OConfigItem config) {
+        for (OConfigItem child : config.getChildren()) {
+            switch (child.getKey().toLowerCase()) {
+                case "tags":
+                    String tagValue = child.getValues().get(0).getString();
+                    Arrays.stream(tagValue.split(",")).map(String::trim).forEach(p -> {
+                        String[] pair = p.split("=");
+                        if (pair.length == 2) {
+                            addTag(addlTags, pair[0], pair[1]);
+                        }
+                    });
+                    break;
+                case "filteredmetricsfile":
+                    setFilteredMetricsFile(child.getValues().get(0).getString());
+                    break;
+                case "filteredtagsfile":
+                    setFilteredTagsFile(child.getValues().get(0).getString());
+                    break;
+                case "debug":
+                    debug = child.getValues().get(0).getBoolean();
+                    break;
+                default:
+            }
+        }
+        return 0;
+    }
+
+    public void setFilteredMetricsFile(String filteredMetricsFile) {
+        if (StringUtils.isNotBlank(filteredMetricsFile)) {
+            exclusions.setFilteredMetricsFile(filteredMetricsFile);
+        }
+    }
+
+    public void setFilteredTagsFile(String filteredTagsFile) {
+        if (StringUtils.isNotBlank(filteredTagsFile)) {
+            exclusions.setFilteredTagsFile(filteredTagsFile);
+        }
+    }
+
+    public abstract void write(String metric, OutputStream out);
+
+    protected void addTag(Map<String,String> m, String k, String v) {
+        if (k != null && !k.isBlank() && v != null && !v.isBlank()) {
+            k = k.replaceAll("=", ":");
+            v = v.replaceAll("=", ":");
+            m.put(k.trim(), v.trim());
+        }
+    }
+
+    protected void addTag(Map<String,String> m, String kv) {
+        if (kv != null && !kv.isBlank()) {
+            String[] parts = kv.split("=");
+            if (parts.length == 2) {
+                addTag(m, parts[0], parts[1]);
+            }
+        }
+    }
+
+    public void process(ValueList vl, OutputStream out) {
+        try {
+            processInternal(vl, out);
+        } catch (Exception e) {
+            Collectd.logError(e.getMessage());
+            e.printStackTrace(System.err);
+        }
+    }
+
+    private void processInternal(ValueList vl, OutputStream out) {
+        debugValueList(vl);
+
+        StringBuilder metric = new StringBuilder();
+        Long timestamp = vl.getTime();
+        Map<String,String> tagMap = new TreeMap<>();
+        String host = vl.getHost();
+        int idx = host.indexOf(PERIOD);
+        if (-1 != idx) {
+            addTag(tagMap, "host", host.substring(0, idx));
+        } else {
+            addTag(tagMap, "host", host);
+        }
+        String[] hostSplit = host.split("\\.");
+        int nIdx = hostSplit[0].lastIndexOf('n');
+        if (-1 != nIdx) {
+            addTag(tagMap, "rack", hostSplit[0].substring(0, nIdx));
+        }
+        tagMap.putAll(addlTags);
+
+        String plugin = vl.getPlugin();
+        String pluginInstance = vl.getPluginInstance();
+        String type = vl.getType();
+        String typeInstance = vl.getTypeInstance();
+
+        switch (plugin) {
+            case STATSD_PREFIX:
+                Matcher hadoop_4_groups = HADOOP_PATTERN_4_GROUPS.matcher(typeInstance);
+                Matcher statsd_3_groups = STATSD_PATTERN_3_GROUPS.matcher(typeInstance);
+                Matcher statsd_4_groups = STATSD_PATTERN_4_GROUPS.matcher(typeInstance);
+                Matcher statsd_6_groups = STATSD_PATTERN_6_GROUPS.matcher(typeInstance);
+                String instance = null;
+                boolean defaultCase = false;
+                if (!typeInstance.startsWith("nsq")) {
+                    String[] parts = typeInstance.split("\\.");
+                    if (parts.length >= 4 && (HADOOP_CONTEXTS.contains(parts[0]) || hadoop_4_groups.matches())) {
+                        // Here we are processing the statsd metrics coming from the Hadoop Metrics2 StatsDSink without the host name.
+                        // The format of metric is: serviceName.contextName.recordName.metricName
+                        // The serviceName is used as the instance.
+                        instance = parts[0];
+                        String contextName = parts[1];
+                        int firstPeriod = typeInstance.indexOf(".");
+                        int lastPeriod = typeInstance.lastIndexOf(".");
+                        String metricName = typeInstance.substring(lastPeriod + 1);
+                        String recordName = typeInstance.substring(firstPeriod + 1, lastPeriod);
+                        // the recordName may contain useful tags such as op=<operation> and user<user>
+                        // in addition to a possibly duplicative recordName
+                        String[] recordNameSplits = recordName.split("\\.");
+                        metric.append(STATSD_PREFIX).append(PERIOD).append(contextName).append(PERIOD);
+                        for (String split : recordNameSplits) {
+                            // avoid duplication with either the contextName or instance
+                            if (!split.equalsIgnoreCase(contextName) && !split.equalsIgnoreCase(instance)) {
+                                if (split.startsWith("NNTopUserOpCounts") && split.contains("windowMs=")) {
+                                    metric.append("NNTopUserOpCounts").append(PERIOD);
+                                    addTag(tagMap, split.substring(split.indexOf("windowMs=")));
+                                } else if (split.contains("=")) {
+                                    addTag(tagMap, split);
+                                } else {
+                                    addTag(tagMap, "record", split);
+                                }
+                            }
+                        }
+                        metric.append(metricName);
+                    } else if (parts.length >= 2 && parts[1].equals("dwquery")) {
+                        // These come from Datawave's QueryIterator metrics - the first split part is the queryId
+                        // Use statsd plus the rest of the typeInstance after the first period as the metric name
+                        int x = typeInstance.indexOf(".");
+                        metric.append(STATSD_PREFIX).append(PERIOD).append(typeInstance.substring(x + 1));
+                        addTag(tagMap, "queryId", parts[0]);
+                    } else if (parts.length % 2 == 1) {
+                        // EtsyStatsD format -- metric.(tagName.tagValue)*
+                        metric.append(STATSD_PREFIX).append(PERIOD).append(parts[0]);
+                        for (int x = 1; x < parts.length; x += 2) {
+                            addTag(tagMap, parts[x], parts[x + 1]);
+                        }
+                    } else if (statsd_4_groups.matches()) {
+                        // Here we are processing the statsd metrics coming from the Hadoop Metrics2 StatsDSink without the host name.
+                        // The format of metric is: serviceName.contextName.recordName.metricName. The recordName is typically duplicative
+                        // and is dropped here. The serviceName is used as the instance.
+                        metric.append(STATSD_PREFIX).append(PERIOD).append(statsd_4_groups.group(2)).append(PERIOD).append(statsd_4_groups.group(4));
+                        instance = statsd_4_groups.group(1);
+                    } else {
+                        defaultCase = true;
+                    }
+                } else {
+                    if (statsd_3_groups.matches()) {
+                        metric.append(NSQ_PREFIX).append(statsd_3_groups.group(2)).append(PERIOD).append(statsd_3_groups.group(3));
+                    } else if (statsd_4_groups.matches()) {
+                        metric.append(NSQ_PREFIX).append(statsd_4_groups.group(2)).append(PERIOD).append(statsd_4_groups.group(4));
+                        instance = statsd_4_groups.group(3);
+                    } else if (statsd_6_groups.matches()) {
+                        metric.append(NSQ_PREFIX).append(statsd_6_groups.group(4)).append(PERIOD).append(statsd_6_groups.group(6));
+                        instance = statsd_6_groups.group(5);
+                    } else {
+                        defaultCase = true;
+                    }
+                }
+                if (defaultCase) {
+                    // Handle StatsD metrics of unknown formats. If there is a
+                    // period in the metric name, use everything up to that as
+                    // the instance.
+                    int period = typeInstance.indexOf('.');
+                    if (-1 == period) {
+                        metric.append(STATSD_PREFIX).append(PERIOD).append(typeInstance);
+                    } else {
+                        instance = typeInstance.substring(0, period);
+                        metric.append(STATSD_PREFIX).append(PERIOD).append(typeInstance.substring(period + 1));
+                    }
+                }
+                if (null != instance) {
+                    addTag(tagMap, INSTANCE, instance);
+                }
+                break;
+            case "ethstat":
+                metric.append("sys.ethstat.");
+                if (typeInstance.matches(ETHSTAT_QUEUE_PATTERN1.pattern())) {
+                    Matcher ethstat_m = ETHSTAT_QUEUE_PATTERN1.matcher(typeInstance);
+                    if (ethstat_m.matches()) {
+                        metric.append(ethstat_m.group(1));
+                        metric.append("queue_");
+                        metric.append(ethstat_m.group(3));
+                        addTag(tagMap, "queue", ethstat_m.group(2));
+                    }
+                } else if (typeInstance.matches(ETHSTAT_QUEUE_PATTERN2.pattern())) {
+                    Matcher ethstat_m = ETHSTAT_QUEUE_PATTERN2.matcher(typeInstance);
+                    if (ethstat_m.matches()) {
+                        metric.append(ethstat_m.group(1));
+                        metric.append(ethstat_m.group(2));
+                        metric.append("_queue_");
+                        metric.append(ethstat_m.group(6));
+                        addTag(tagMap, "queue", ethstat_m.group(4));
+                    }
+                } else if (typeInstance.matches(ETHSTAT_TRAFFIC_CLASS_PATTERN.pattern())) {
+                    Matcher ethstat_m = ETHSTAT_TRAFFIC_CLASS_PATTERN.matcher(typeInstance);
+                    if (ethstat_m.matches()) {
+                        metric.append(ethstat_m.group(1));
+                        metric.append(ethstat_m.group(2));
+                        metric.append(ethstat_m.group(5));
+                        metric.append(ethstat_m.group(6));
+                        addTag(tagMap, "trafficClass", ethstat_m.group(4));
+                    }
+                } else if (typeInstance.matches(ETHSTAT_CHANNEL_PATTERN.pattern())) {
+                    Matcher ethstat_m = ETHSTAT_CHANNEL_PATTERN.matcher(typeInstance);
+                    if (ethstat_m.matches()) {
+                        metric.append(ethstat_m.group(1));
+                        metric.append("_");
+                        metric.append(ethstat_m.group(4));
+                        addTag(tagMap, "channel", ethstat_m.group(2));
+                    }
+                } else {
+                    metric.append(typeInstance);
+                }
+                addTag(tagMap, INSTANCE, pluginInstance);
+                break;
+            case "hddtemp":
+                metric.append("sys.hddtemp.").append(type);
+                addTag(tagMap, INSTANCE, typeInstance);
+                break;
+            case "smart":
+                int code = -1;
+                String name = null;
+                if (typeInstance.startsWith("attribute-")) {
+                    int hyphen = typeInstance.indexOf('-');
+                    code = Integer.parseInt(typeInstance.substring(hyphen + 1));
+                    name = smart.get(code);
+                }
+                if (code == -1) {
+                    if (notEmpty(typeInstance)) {
+                        metric.append("sys.smart.").append(typeInstance);
+                    } else {
+                        metric.append("sys.smart.").append(type);
+                    }
+                } else {
+                    metric.append("sys.smart.").append(name);
+                    addTag(tagMap, CODE, Integer.toString(code));
+                }
+                addTag(tagMap, INSTANCE, pluginInstance);
+                break;
+            case "sensors":
+                if (typeInstance.startsWith("temp")) {
+                    addTag(tagMap, INSTANCE, typeInstance.substring(4));
+                }
+                metric.append("sys.sensors.").append(type).append(PERIOD).append(pluginInstance);
+                break;
+            case "haproxy":
+                metric.append("sys.haproxy.").append(typeInstance);
+                Matcher hadoop_m2 = HAPROXY_PATTERN.matcher(pluginInstance);
+                if (hadoop_m2.matches()) {
+                    addTag(tagMap, hadoop_m2.group(1));
+                    addTag(tagMap, hadoop_m2.group(2));
+                }
+                break;
+            case "ipmi":
+            case "snmp":
+                metric.append("sys.").append(plugin).append(PERIOD).append(type);
+                addTag(tagMap, INSTANCE, typeInstance.replaceAll(" ", "_"));
+                break;
+            case "load":
+                metric.append("sys.load.").append(plugin).append(PERIOD).append(type);
+                // plugin returns 1 min, 5 min, 15 min averages, but Timely can
+                // downsample on its own so we only use the first (1 min) value
+                vl.setValues(Arrays.asList(vl.getValues().get(0)));
+                break;
+            case "GenericJMX":
+                metric.append("sys.").append(plugin).append(PERIOD).append(type).append(PERIOD).append(typeInstance);
+                String[] pluginInstanceSplit = pluginInstance.split("-");
+                if (pluginInstanceSplit.length > 0) {
+                    addTag(tagMap, INSTANCE, pluginInstanceSplit[0].replaceAll(" ", "_"));
+                }
+                if (pluginInstanceSplit.length > 1) {
+                    addTag(tagMap, NAME, pluginInstanceSplit[1].replaceAll(" ", "_"));
+                }
+                break;
+            default:
+                if (notEmpty(type) && notEmpty(typeInstance) && notEmpty(plugin) && notEmpty(pluginInstance)) {
+                    metric.append("sys.").append(plugin).append(PERIOD).append(type).append(PERIOD).append(typeInstance);
+                    addTag(tagMap, INSTANCE, pluginInstance.replaceAll(" ", "_"));
+                } else if (notEmpty(type) && notEmpty(typeInstance) && notEmpty(plugin)) {
+                    metric.append("sys.").append(plugin).append(PERIOD).append(type).append(PERIOD).append(typeInstance);
+                } else if (notEmpty(type) && notEmpty(plugin) && notEmpty(pluginInstance)) {
+                    metric.append("sys.").append(plugin).append(PERIOD).append(type);
+                    addTag(tagMap, INSTANCE, pluginInstance.replaceAll(" ", "_"));
+                } else if (notEmpty(type) && notEmpty(plugin)) {
+                    metric.append("sys.").append(plugin).append(PERIOD).append(type);
+                } else {
+                    Collectd.logWarning("Unhandled metric: " + vl);
+                    return;
+                }
+                break;
+        }
+        final String metricName = metric.toString().replaceAll(" ", "_");
+        for (int i = 0; i < vl.getValues().size(); i++) {
+            int dataSourceSampleType = vl.getDataSet().getDataSources().get(i).getType();
+            String sampleType = convertType(dataSourceSampleType);
+            if (null != sampleType) {
+                addTag(tagMap, SAMPLE_TYPE, sampleType);
+            }
+            Double value = vl.getValues().get(i).doubleValue();
+            String tagString = tagMap.entrySet().stream().map(e -> e.getKey() + "=" + e.getValue()).collect(Collectors.joining(" "));
+
+            if (metricName.isBlank() || timestamp == null || value == null) {
+                logDebug(String.format("Not writing unhandled metric: plugin:%s pluginInstance:%s type:%s typeInstance:%s", plugin, pluginInstance, type,
+                                typeInstance));
+            } else {
+                String datapoint = MessageFormat.format(PUT, metricName, timestamp.toString(), value.toString(), tagString);
+                if (exclusions.filterMetric(datapoint)) {
+                    logDebug(String.format("Not writing excluded metric: %s", datapoint));
+                } else {
+                    datapoint = exclusions.filterExcludedTags(datapoint);
+                    logDebug(String.format("Writing: %s", datapoint));
+                    write(datapoint + "\n", out);
+                }
+            }
+        }
+    }
+
+    private void logDebug(String s) {
+        // collectd must be compiled with debug enabled in order for log level debug to work
+        if (debug) {
+            Collectd.logInfo(s);
+        } else {
+            Collectd.logDebug(s);
+        }
+    }
+
+    private void debugValueList(ValueList vl) {
+        try {
+            StringBuilder sb = new StringBuilder();
+            sb.append(String.format("Plugin:%s", vl.getPlugin()));
+            if (vl.getPluginInstance() != null && !vl.getPluginInstance().isBlank()) {
+                sb.append(String.format(" PluginInstance:%s", vl.getPluginInstance()));
+            }
+            sb.append(String.format(" Type:%s", vl.getType()));
+            sb.append(String.format(" TypeInstance:%s", vl.getTypeInstance()));
+            for (int i = 0; i < vl.getValues().size(); i++) {
+                Number value = vl.getValues().get(i);
+                sb.append(String.format(" Value:%s", value.toString()));
+            }
+            logDebug(sb.toString());
+        } catch (Exception e) {
+            e.printStackTrace();
+            logDebug(e.getMessage());
+        }
+    }
+
+    private String convertType(int type) {
+        switch (type) {
+            case 0:
+                return COUNTER;
+            case 1:
+                return GAUGE;
+            case 2:
+                return DERIVE;
+            case 3:
+                return ABSOLUTE;
+            default:
+                return GAUGE;
+        }
+    }
+
+    private boolean notEmpty(String arg) {
+        return (null != arg) && !(arg.equals(""));
+    }
+}

collectd/src/main/java/timely/collectd/plugin/PooledSocketFactory.java

@@ -0,0 +1,122 @@
+package timely.collectd.plugin;
+
+import java.net.Socket;
+
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.PooledObjectFactory;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.collectd.api.Collectd;
+import org.collectd.api.OConfigItem;
+
+public class PooledSocketFactory implements PooledObjectFactory<Socket> {
+
+    private String host = null;
+    private int port = 0;
+    private static final long initialBackoff = 2000;
+    private int socketTimeout = 60000; // 60 seconds in milliseconds
+    private long connectionTimeToLive = 300000; // 300 seconds in milliseconds
+
+    public int config(OConfigItem config) {
+        for (OConfigItem child : config.getChildren()) {
+            switch (child.getKey()) {
+                case "host":
+                case "hostname":
+                case "Host":
+                case "HostName":
+                    host = child.getValues().get(0).getString();
+                    break;
+                case "Port":
+                case "port":
+                    port = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "socketTimeout":
+                case "SocketTimeout":
+                    socketTimeout = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "connectionTimeToLive":
+                case "ConnectionTimeToLive":
+                    connectionTimeToLive = Long.parseLong(child.getValues().get(0).getString());
+                    break;
+                default:
+
+            }
+        }
+        if (null != host) {
+            return 0;
+        }
+        if (host == null) {
+            Collectd.logError("Timely host must be configured");
+        }
+        if (port == 0) {
+            Collectd.logError("Timely port must be configured");
+        }
+        return -1;
+    }
+
+    private Socket connect() {
+
+        Socket socket = null;
+        long currentBackoff = initialBackoff;
+        while (socket == null) {
+            try {
+                socket = new Socket(host, port);
+                if (socketTimeout > -1) {
+                    socket.setSoTimeout(socketTimeout);
+                }
+                Collectd.logInfo("Connected to Timely at " + host + ":" + port + " from local port:" + socket.getLocalPort());
+            } catch (Exception e) {
+                Collectd.logError("Error connecting to Timely at " + host + ":" + port + ". Error: " + e.getMessage() + ".  Will retry connection in "
+                                + currentBackoff + " ms.");
+                try {
+                    Thread.sleep(currentBackoff);
+                } catch (InterruptedException e1) {
+
+                }
+                // max out reconnect period at one minute
+                currentBackoff = (currentBackoff * 2 > 60000) ? 60000 : currentBackoff * 2;
+            }
+        }
+        return socket;
+    }
+
+    @Override
+    public PooledObject<Socket> makeObject() {
+        return new DefaultPooledObject<>(connect());
+    }
+
+    @Override
+    public void destroyObject(PooledObject<Socket> pooledObject) {
+        try {
+            Collectd.logDebug("Shutting down connection to Timely at " + host + ":" + port);
+            Socket socket = pooledObject.getObject();
+            if (socket != null) {
+                socket.close();
+            }
+        } catch (Exception e) {
+            Collectd.logError("Error closing connection to Timely at " + host + ":" + port + ". Error: " + e.getMessage());
+        } finally {
+            pooledObject.invalidate();
+        }
+    }
+
+    @Override
+    public boolean validateObject(PooledObject<Socket> pooledObject) {
+        if (connectionTimeToLive > -1) {
+            if (System.currentTimeMillis() > pooledObject.getCreateTime() + connectionTimeToLive) {
+                return false;
+            }
+        }
+        Socket socket = pooledObject.getObject();
+        return socket != null && !socket.isClosed();
+    }
+
+    @Override
+    public void activateObject(PooledObject<Socket> pooledObject) {
+        // do nothing
+    }
+
+    @Override
+    public void passivateObject(PooledObject<Socket> pooledObject) {
+        // do nothing
+    }
+}

collectd/src/main/java/timely/collectd/plugin/WriteNSQPlugin.java

@@ -0,0 +1,161 @@
+package timely.collectd.plugin;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Random;
+
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.entity.EntityBuilder;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.util.EntityUtils;
+import org.collectd.api.Collectd;
+import org.collectd.api.CollectdConfigInterface;
+import org.collectd.api.CollectdShutdownInterface;
+import org.collectd.api.CollectdWriteInterface;
+import org.collectd.api.OConfigItem;
+import org.collectd.api.ValueList;
+
+/**
+ *
+ * CollectD plugin that writes metrics collected by the StatsD plugin to NSQ. NSQ can be configured to write statsd metrics to CollectD. If you do this, set the
+ * statsd prefix in NSQ to 'nsq'.
+ *
+ */
+public class WriteNSQPlugin extends CollectDPluginParent implements CollectdConfigInterface, CollectdShutdownInterface, CollectdWriteInterface {
+
+    private String host = null;
+    private int port = 0;
+    private String topic = "metrics#ephemeral";
+    private String endpoint = null;
+    private GenericObjectPool<CloseableHttpClient> clientPool = null;
+    private Random random = new Random();
+
+    // pool should be limited by the number of WriteThreads configured in
+    // collectd
+    static final private int POOL_MAX_SIZE = Integer.MAX_VALUE;
+
+    public WriteNSQPlugin() {
+        Collectd.registerConfig(WriteNSQPlugin.class.getName(), this);
+        Collectd.registerShutdown(WriteNSQPlugin.class.getName(), this);
+        Collectd.registerWrite(WriteNSQPlugin.class.getName(), this);
+    }
+
+    @Override
+    public int write(ValueList vl) {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream out = new DataOutputStream(baos);
+        super.process(vl, out);
+        CloseableHttpClient client = null;
+        try {
+            out.flush();
+            HttpPost post = new HttpPost(endpoint);
+            EntityBuilder request = EntityBuilder.create();
+            request.setBinary(baos.toByteArray());
+            post.setEntity(request.build());
+            client = clientPool.borrowObject();
+            CloseableHttpResponse response = client.execute(post);
+            try {
+                int code = response.getStatusLine().getStatusCode();
+                String msg = response.getStatusLine().getReasonPhrase();
+                // Consume the entire response so the connection will be reused
+                EntityUtils.consume(response.getEntity());
+                if (code != HttpStatus.SC_OK) {
+                    Collectd.logInfo("Code: " + code + ", msg: " + msg);
+                }
+            } finally {
+                response.close();
+            }
+        } catch (Exception e) {
+            Collectd.logError("Error sending metrics to NSQ: " + e.getMessage());
+            if (client != null) {
+                try {
+                    client.close();
+                } catch (IOException e1) {}
+                try {
+                    clientPool.invalidateObject(client);
+                    client = null;
+                } catch (Exception e2) {}
+            }
+        } finally {
+            if (client != null) {
+                clientPool.returnObject(client);
+            }
+            try {
+                out.close();
+            } catch (IOException e) {}
+        }
+        return 0;
+    }
+
+    @Override
+    public void write(String metric, OutputStream out) {
+        try {
+            ((DataOutputStream) out).writeBytes(metric);
+        } catch (IOException e) {
+            Collectd.logError("Error writing metric: " + e.getMessage());
+            try {
+                out.close(); // Should cause IOException in write.
+            } catch (IOException e1) {}
+        }
+    }
+
+    @Override
+    public int shutdown() {
+        Collectd.logInfo("Shutting down connection to NSQ at " + host + ":" + port);
+        clientPool.close();
+        return 0;
+    }
+
+    @Override
+    public int config(OConfigItem config) {
+        super.config(config);
+        GenericObjectPoolConfig<CloseableHttpClient> poolConfig = new GenericObjectPoolConfig<>();
+        // use max size for maxTotal and maxIdle
+        // no need to activate or passivate
+        poolConfig.setMaxTotal(POOL_MAX_SIZE);
+        poolConfig.setMaxIdle(POOL_MAX_SIZE);
+        poolConfig.setTestOnReturn(true);
+        PooledCloseableHttpClientFactory pooledCloseableHttpClientFactory = new PooledCloseableHttpClientFactory();
+        pooledCloseableHttpClientFactory.config(config);
+
+        for (OConfigItem child : config.getChildren()) {
+            switch (child.getKey()) {
+                case "host":
+                case "hostname":
+                case "Host":
+                case "HostName":
+                    String hostString = child.getValues().get(0).getString();
+                    String[] hosts = hostString.split(",");
+                    if (hosts.length == 1) {
+                        host = hosts[0];
+                    } else {
+                        host = hosts[random.nextInt(hosts.length)];
+                    }
+                    break;
+                case "Port":
+                case "port":
+                    port = Integer.parseInt(child.getValues().get(0).getString());
+                    break;
+                case "topic":
+                case "Topic":
+                    topic = child.getValues().get(0).getString();
+                    break;
+                default:
+            }
+        }
+        if (host == null || port == 0 || topic == null) {
+            Collectd.logError("NSQ host, port, and topic must be configured");
+            return -1;
+        }
+        this.clientPool = new GenericObjectPool<>(pooledCloseableHttpClientFactory, poolConfig);
+        this.endpoint = "http://" + host + ":" + port + "/mpub?topic=" + topic;
+        return 0;
+    }
+
+}

collectd/src/test/java/timely/collectd/plugin/PluginParentTest.java

@@ -3,6 +3,7 @@ package timely.collectd.plugin;
 import static org.junit.Assert.*;
 
 import java.io.OutputStream;
+import java.net.URL;
 import java.util.Collections;
 
 import org.collectd.api.Collectd;
@@ -13,20 +14,19 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.powermock.api.easymock.PowerMock;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
-import net.jcip.annotations.NotThreadSafe;
-
 @RunWith(PowerMockRunner.class)
+@PowerMockIgnore("jdk.internal.reflect.*")
 @PrepareForTest({Collectd.class})
-@NotThreadSafe
 public class PluginParentTest {
 
     public class TestPlugin extends CollectDPluginParent {
 
         public TestPlugin() {
-            super.addlTags.add(" addl1=foo");
+            super.addlTags.put("addl1", "foo");
         }
 
         @Override
@@ -41,12 +41,13 @@ public class PluginParentTest {
     public String result = null;
 
     @Before
-    public void setup() throws Exception {
+    public void setup() {
+        result = null;
         PowerMock.suppress(PowerMock.everythingDeclaredIn(Collectd.class));
     }
 
     @Test
-    public void testStatsDHadoopFormat() throws Exception {
+    public void testStatsDHadoopFormat() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("statsd");
@@ -60,11 +61,11 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put statsd.dfs.BlocksRead 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo instance=DataNode sample=value sampleType=GAUGE\n", result);
+        assertEquals("put statsd.dfs.BlocksRead 1456156976840 1.0 addl1=foo host=r01n01 instance=DataNode rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
-    public void testStatsDUnknownFormat() throws Exception {
+    public void testStatsDUnknownFormat() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("statsd");
@@ -78,11 +79,11 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put statsd.baz 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo sample=value sampleType=GAUGE\n", result);
+        assertEquals("put statsd.baz 1456156976840 1.0 addl1=foo host=r01n01 rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
-    public void testStatsDUnknownFormat2() throws Exception {
+    public void testStatsDUnknownFormat2() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("statsd");
@@ -96,11 +97,11 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put statsd.baz 1456156976840 1.0 host=r01n01 rack=r01 addl1=foo instance=bar sample=value sampleType=GAUGE\n", result);
+        assertEquals("put statsd.baz 1456156976840 1.0 addl1=foo host=r01n01 instance=bar rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
-    public void testHddTemp() throws Exception {
+    public void testHddTemp() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("hddtemp");
@@ -114,11 +115,11 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.hddtemp.temperature 1456156976840 35.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n", result);
+        assertEquals("put sys.hddtemp.temperature 1456156976840 35.0 addl1=foo host=r01n01 instance=sda rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
-    public void testSmart1() throws Exception {
+    public void testSmart1() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("smart");
@@ -132,11 +133,11 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.smart.smart_badsectors 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n", result);
+        assertEquals("put sys.smart.smart_badsectors 1456156976840 0.0 addl1=foo host=r01n01 instance=sda rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
-    public void testSmart2() throws Exception {
+    public void testSmart2() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("smart");
@@ -151,11 +152,11 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.smart.raw-read-error-rate 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=sda sample=value sampleType=GAUGE\n", result);
+        assertEquals("put sys.smart.raw-read-error-rate 1456156976840 0.0 addl1=foo host=r01n01 instance=sda rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
-    public void testSmartCode() throws Exception {
+    public void testSmartCode() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("smart");
@@ -170,12 +171,11 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.smart.Total_LBAs_Read 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo code=242 instance=sda sample=value sampleType=GAUGE\n",
-                        result);
+        assertEquals("put sys.smart.Total_LBAs_Read 1456156976840 0.0 addl1=foo code=242 host=r01n01 instance=sda rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
-    public void testSnmp() throws Exception {
+    public void testSnmp() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("snmp");
@@ -190,7 +190,7 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.snmp.if_octets 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo instance=Ethernet1 sample=value sampleType=GAUGE\n", result);
+        assertEquals("put sys.snmp.if_octets 1456156976840 0.0 addl1=foo host=r01n01 instance=Ethernet1 rack=r01 sampleType=GAUGE\n", result);
 
     }
 
@@ -210,8 +210,7 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.sensors.temperature.coretemp-isa-0000 1456156976840 35.0 host=r01n01 rack=r01 addl1=foo instance=1 sample=value sampleType=GAUGE\n",
-                        result);
+        assertEquals("put sys.sensors.temperature.coretemp-isa-0000 1456156976840 35.0 addl1=foo host=r01n01 instance=1 rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
@@ -229,7 +228,7 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.haproxy.run_queue 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo sample=value sampleType=GAUGE\n", result);
+        assertEquals("put sys.haproxy.run_queue 1456156976840 0.0 addl1=foo host=r01n01 rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
@@ -248,7 +247,7 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.haproxy.queue_current 1456156976840 0.0 host=r01n01 rack=r01 addl1=foo proxy_name=proxy1 service_name=server1 sample=value sampleType=GAUGE\n",
+        assertEquals("put sys.haproxy.queue_current 1456156976840 0.0 addl1=foo host=r01n01 proxy_name=proxy1 rack=r01 sampleType=GAUGE service_name=server1\n",
                         result);
     }
 
@@ -268,20 +267,19 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.ethstat.tx_comp_queue_full 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo instance=eth0 sample=value sampleType=GAUGE\n",
-                        result);
+        assertEquals("put sys.ethstat.tx_comp_queue_full 1456156976840 6.0 addl1=foo host=r01n01 instance=eth0 rack=r01 sampleType=GAUGE\n", result);
 
     }
 
     @Test
-    public void testEthStatWithQueue() {
+    public void testEthstatWithQueueStyle1() {
         ValueList vl = new ValueList();
         vl.setHost(HOST);
         vl.setPlugin("ethstat");
         vl.setPluginInstance("eth0");
         vl.setTime(TIME);
         vl.setType("derive");
-        vl.setTypeInstance("rx_queue_1_bytes");
+        vl.setTypeInstance("rx_queue_15_bytes");
         vl.setValues(Collections.singletonList((Number) 6.0D));
         DataSet ds = new DataSet("DERIVE");
         ds.addDataSource(new DataSource("value", 1, 6.0D, 100.0D));
@@ -289,8 +287,83 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.ethstat.rx_bytes 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo queue=1 instance=eth0 sample=value sampleType=GAUGE\n", result);
+        assertEquals("put sys.ethstat.rx_queue_bytes 1456156976840 6.0 addl1=foo host=r01n01 instance=eth0 queue=15 rack=r01 sampleType=GAUGE\n", result);
 
+        result = null;
+        vl.setTypeInstance("queue_7_tx_bytes");
+        test.process(vl, null);
+        assertEquals("put sys.ethstat.queue_tx_bytes 1456156976840 6.0 addl1=foo host=r01n01 instance=eth0 queue=7 rack=r01 sampleType=GAUGE\n", result);
+
+        result = null;
+        vl.setTypeInstance("queue_7_rx_xdp_drop");
+        test.process(vl, null);
+        assertEquals("put sys.ethstat.queue_rx_xdp_drop 1456156976840 6.0 addl1=foo host=r01n01 instance=eth0 queue=7 rack=r01 sampleType=GAUGE\n", result);
+    }
+
+    @Test
+    public void testEthstatWithQueueStyle2() {
+        ValueList vl = new ValueList();
+        vl.setHost(HOST);
+        vl.setPlugin("ethstat");
+        vl.setPluginInstance("eth0");
+        vl.setTime(TIME);
+        vl.setType("derive");
+        vl.setTypeInstance("rx-15.bytes");
+        vl.setValues(Collections.singletonList((Number) 6.0D));
+        DataSet ds = new DataSet("DERIVE");
+        ds.addDataSource(new DataSource("value", 1, 6.0D, 100.0D));
+        vl.setDataSet(ds);
+
+        TestPlugin test = new TestPlugin();
+
+        test.process(vl, null);
+        assertEquals("put sys.ethstat.rx_queue_bytes 1456156976840 6.0 addl1=foo host=r01n01 instance=eth0 queue=15 rack=r01 sampleType=GAUGE\n", result);
+
+        result = null;
+        vl.setTypeInstance("tx15_xdp_err");
+        test.process(vl, null);
+        assertEquals("put sys.ethstat.tx_queue_xdp_err 1456156976840 6.0 addl1=foo host=r01n01 instance=eth0 queue=15 rack=r01 sampleType=GAUGE\n", result);
+    }
+
+    @Test
+    public void testEthstatWithTrafficClass() {
+        ValueList vl = new ValueList();
+        vl.setHost(HOST);
+        vl.setPlugin("ethstat");
+        vl.setPluginInstance("eth0");
+        vl.setTime(TIME);
+        vl.setType("derive");
+        vl.setTypeInstance("veb.tc_15_tx_bytes");
+        vl.setValues(Collections.singletonList((Number) 6.0D));
+        DataSet ds = new DataSet("DERIVE");
+        ds.addDataSource(new DataSource("value", 1, 6.0D, 100.0D));
+        vl.setDataSet(ds);
+
+        TestPlugin test = new TestPlugin();
+
+        test.process(vl, null);
+        assertEquals("put sys.ethstat.veb.tc_tx_bytes 1456156976840 6.0 addl1=foo host=r01n01 instance=eth0 rack=r01 sampleType=GAUGE trafficClass=15\n",
+                        result);
+    }
+
+    @Test
+    public void testEthstatWithChannel() {
+        ValueList vl = new ValueList();
+        vl.setHost(HOST);
+        vl.setPlugin("ethstat");
+        vl.setPluginInstance("eth0");
+        vl.setTime(TIME);
+        vl.setType("derive");
+        vl.setTypeInstance("ch44_events");
+        vl.setValues(Collections.singletonList((Number) 6.0D));
+        DataSet ds = new DataSet("DERIVE");
+        ds.addDataSource(new DataSource("value", 1, 6.0D, 100.0D));
+        vl.setDataSet(ds);
+
+        TestPlugin test = new TestPlugin();
+
+        test.process(vl, null);
+        assertEquals("put sys.ethstat.ch_events 1456156976840 6.0 addl1=foo channel=44 host=r01n01 instance=eth0 rack=r01 sampleType=GAUGE\n", result);
     }
 
     @Test
@@ -308,9 +381,37 @@ public class PluginParentTest {
 
         TestPlugin test = new TestPlugin();
         test.process(vl, null);
-        assertEquals("put sys.ipmi.temperature 1456156976840 6.0 host=r01n01 rack=r01 addl1=foo instance=LAN_NIC_Temp_system_board_(3.2) sample=value sampleType=GAUGE\n",
+        assertEquals("put sys.ipmi.temperature 1456156976840 6.0 addl1=foo host=r01n01 instance=LAN_NIC_Temp_system_board_(3.2) rack=r01 sampleType=GAUGE\n",
                         result);
 
     }
 
+    @Test
+    public void testExclusions() {
+        ValueList vl = new ValueList();
+        vl.setHost(HOST);
+        vl.setPlugin("ethstat");
+        vl.setPluginInstance("eth0");
+        vl.setTime(TIME);
+        vl.setType("derive");
+        vl.setTypeInstance("rx_queue_1_bytes");
+        vl.setValues(Collections.singletonList((Number) 6.0D));
+        DataSet ds = new DataSet("DERIVE");
+        ds.addDataSource(new DataSource("value", 1, 6.0D, 100.0D));
+        vl.setDataSet(ds);
+
+        TestPlugin test = new TestPlugin();
+        URL filteredMetricsFile = getClass().getClassLoader().getResource("filteredMetrics.txt");
+        test.setFilteredMetricsFile(filteredMetricsFile.getPath());
+        URL filteredTagsFile = getClass().getClassLoader().getResource("filteredTags.txt");
+        test.setFilteredTagsFile(filteredTagsFile.getPath());
+
+        test.process(vl, null);
+        assertEquals("put sys.ethstat.rx_queue_bytes 1456156976840 6.0 addl1=foo host=r01n01 queue=1 rack=r01\n", result);
+
+        result = null;
+        vl.setPlugin("interface");
+        vl.setTypeInstance("if_octets");
+        assertNull(result);
+    }
 }

collectd/src/test/java/timely/collectd/plugin/WriteTimelyPluginTest.java

@@ -0,0 +1,203 @@
+package timely.collectd.plugin;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.Collections;
+
+import org.collectd.api.Collectd;
+import org.collectd.api.DataSet;
+import org.collectd.api.DataSource;
+import org.collectd.api.OConfigItem;
+import org.collectd.api.ValueList;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.easymock.PowerMock;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({Collectd.class})
+@PowerMockIgnore({"javax.management.*", "jdk.internal.reflect.*"})
+public class WriteTimelyPluginTest {
+
+    private WriteTimelyPlugin plugin = null;
+    private FakeServer server = null;
+
+    @Before
+    public void setup() throws Exception {
+        PowerMock.suppress(PowerMock.everythingDeclaredIn(Collectd.class));
+        server = new FakeServer();
+    }
+
+    private ValueList createMetric() {
+        ValueList val = new ValueList();
+        val.setHost("r001n01.localdomain");
+        val.setTime(System.currentTimeMillis());
+        val.setPlugin("hddtemp");
+        val.setPluginInstance(null);
+        val.setType("temperature");
+        val.setTypeInstance("sda");
+        val.setValues(Collections.singletonList((Number) new Double(2.0)));
+        DataSet ds = new DataSet("test");
+        ds.addDataSource(new DataSource("test", DataSource.TYPE_COUNTER, 0, 100));
+        val.setDataSet(ds);
+        return val;
+    }
+
+    private void setupPlugin() throws Exception {
+        OConfigItem host = new OConfigItem("Host");
+        host.addValue(server.getHost());
+        OConfigItem port = new OConfigItem("Port");
+        port.addValue(Integer.toString(server.getPort()));
+        OConfigItem config = new OConfigItem("");
+        config.addChild(host);
+        config.addChild(port);
+        plugin = new WriteTimelyPlugin();
+        Assert.assertEquals(0, plugin.config(config));
+        // Seed with one metric to create the connection
+        Assert.assertEquals(0, plugin.write(createMetric()));
+        Thread.sleep(100);
+        Assert.assertTrue(server.messageReceived());
+    }
+
+    @Test
+    public void testWrite() throws Exception {
+        Thread t = new Thread(server);
+        t.start();
+        setupPlugin();
+        while (!server.ready()) {
+            Thread.sleep(1000);
+        }
+        Assert.assertEquals(0, plugin.write(createMetric()));
+        Thread.sleep(100);
+        Assert.assertTrue(server.messageReceived());
+        plugin.shutdown();
+        server.shutdown();
+        t.join();
+    }
+
+    @Test
+    public void testWriteAfterServerRestart() throws Exception {
+        Thread t = new Thread(server);
+        t.start();
+        setupPlugin();
+        while (!server.ready()) {
+            Thread.sleep(1000);
+        }
+        Assert.assertEquals(0, plugin.write(createMetric()));
+        Thread.sleep(100);
+        Assert.assertTrue(server.messageReceived());
+        server.shutdown();
+        t.join();
+        Thread.sleep(2000);
+
+        server.create();
+        Thread t2 = new Thread(server);
+        t2.start();
+        // Need to call this again because the server is not guaranteed to be
+        // listening on the same local port as the first time that it was
+        // started
+        setupPlugin();
+        while (!server.ready()) {
+            Thread.sleep(1000);
+            // Keep sending metrics to plugin to force reconnect
+            int result = plugin.write(createMetric());
+            System.out.println("Wrote to client, result: " + result);
+            Assert.assertEquals(0, result);
+        }
+        Assert.assertEquals(0, plugin.write(createMetric()));
+        Thread.sleep(1000);
+        Assert.assertTrue(server.messageReceived());
+        plugin.shutdown();
+        server.shutdown();
+        t2.join();
+    }
+
+    private static class FakeServer implements Runnable {
+
+        private ServerSocket server = null;
+        private BufferedReader reader = null;
+        private String host = null;
+        private int port = 0;
+        private Socket s = null;
+        private boolean received = false;
+
+        public FakeServer() throws Exception {
+            create();
+        }
+
+        public void create() throws Exception {
+            server = new ServerSocket(port);
+            server.setReceiveBufferSize(8192);
+            host = server.getInetAddress().getHostAddress();
+            port = server.getLocalPort();
+            System.out.println("Local port " + port);
+        }
+
+        public String getHost() {
+            return host;
+        }
+
+        public int getPort() {
+            return port;
+        }
+
+        public void shutdown() throws Exception {
+            System.out.println("Closing server.");
+            s.close();
+            s = null;
+            reader.close();
+            reader = null;
+            server.close();
+            server = null;
+        }
+
+        public boolean ready() throws Exception {
+            return (s != null && s.isConnected());
+        }
+
+        synchronized public boolean messageReceived() {
+            try {
+                return Boolean.valueOf(this.received);
+            } finally {
+                this.received = false;
+            }
+        }
+
+        @Override
+        public void run() {
+            try {
+                s = server.accept();
+                System.out.println("Client connected.");
+                reader = new BufferedReader(new InputStreamReader(s.getInputStream()));
+                while (server != null && !server.isClosed() && s != null && s.isConnected()) {
+                    String line;
+                    try {
+                        line = reader.readLine();
+                    } catch (Exception e) {
+                        // Might have been closed
+                        System.out.println("Closed by Exception: " + e.getMessage());
+                        return;
+                    }
+                    if (null != line) {
+                        synchronized (this) {
+                            this.received = true;
+                        }
+                        System.out.println(line);
+                    }
+                }
+            } catch (IOException e) {
+                Assert.fail(e.getMessage());
+            }
+            System.out.println("Exiting run method.");
+        }
+
+    }
+
+}

collectd/src/test/resources/filteredMetrics.txt

@@ -0,0 +1,3 @@
+sys.ethstat.*_to_.*
+sys.interface.if_packets
+sys.interface.if_octets

collectd/src/test/resources/filteredTags.txt

@@ -0,0 +1,3 @@
+sys.interface.if_packets sampleType,instance
+sys.interface.if_octets sampleType,instance
+sys.ethstat.rx_queue_bytes sampleType,instance

common/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <groupId>timely</groupId>
     <artifactId>timely</artifactId>
-    <version>2.1.0-SNAPSHOT</version>
+    <version>2.2.0</version>
   </parent>
   <artifactId>timely-common</artifactId>
   <dependencies>
@@ -72,14 +72,14 @@
       <groupId>io.jsonwebtoken</groupId>
       <artifactId>jjwt-jackson</artifactId>
     </dependency>
-    <dependency>
-      <groupId>io.netty</groupId>
-      <artifactId>netty-all</artifactId>
-    </dependency>
     <dependency>
       <groupId>io.projectreactor.netty</groupId>
       <artifactId>reactor-netty</artifactId>
     </dependency>
+    <dependency>
+      <groupId>javax.xml.bind</groupId>
+      <artifactId>jaxb-api</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-core</artifactId>
@@ -94,10 +94,6 @@
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-core</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.accumulo</groupId>
-      <artifactId>accumulo-minicluster</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.commons</groupId>
       <artifactId>commons-lang3</artifactId>

common/src/main/java/timely/accumulo/MetaAdapter.java

@@ -1,4 +1,4 @@
-package timely.adapter.accumulo;
+package timely.accumulo;
 
 import org.apache.accumulo.core.data.Key;
 

common/src/main/java/timely/accumulo/MetricAdapter.java

@@ -1,4 +1,4 @@
-package timely.adapter.accumulo;
+package timely.accumulo;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
@@ -206,4 +206,9 @@ public class MetricAdapter {
     public static Pair<String,Long> decodeRowKey(Key k) {
         return rowCoder.decode(k.getRow().getBytes());
     }
+
+    public static void main(String[] args) {
+        long l = 1707859146644L;
+        System.out.println(roundTimestampToLastHour(l));
+    }
 }

common/src/main/java/timely/api/request/timeseries/QueryRequest.java

@@ -32,6 +32,7 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
         private boolean counter = false;
         private long counterMax = 0;
         private long resetValue = 0;
+        private String interval = null;
 
         public RateOption() {}
 
@@ -84,12 +85,21 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
             this.resetValue = resetValue;
         }
 
+        public String getInterval() {
+            return interval;
+        }
+
+        public void setInterval(String interval) {
+            this.interval = interval;
+        }
+
         @Override
         public String toString() {
             ToStringBuilder tsb = new ToStringBuilder(this);
             tsb.append("counter", counter);
             tsb.append("counterMax", counterMax);
             tsb.append("resetValue", resetValue);
+            tsb.append("interval", interval);
             return tsb.toString();
         }
 
@@ -99,6 +109,7 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
             hcb.append(counter);
             hcb.append(counterMax);
             hcb.append(resetValue);
+            hcb.append(interval);
             return hcb.toHashCode();
         }
 
@@ -210,7 +221,7 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
         private String aggregator;
         private String metric;
         private boolean rate = false;
-        private RateOption rateOptions = null;
+        private RateOption rateOptions = new RateOption();
         private Optional<String> downsample = Optional.empty();
         private Map<String,String> tags = new LinkedHashMap<>();
         private Collection<Filter> filters = new ArrayList<>();
@@ -536,9 +547,12 @@ public class QueryRequest extends AuthenticatedRequest implements HttpGetRequest
                                         options.setCounter(rateOptions[x].endsWith("counter"));
                                         break;
                                     case 1:
-                                        options.setCounterMax(Long.parseLong(rateOptions[x]));
+                                        options.setInterval(rateOptions[x]);
                                         break;
                                     case 2:
+                                        options.setCounterMax(Long.parseLong(rateOptions[x]));
+                                        break;
+                                    case 3:
                                         options.setResetValue(Long.parseLong(rateOptions[x]));
                                         break;
                                     default:

common/src/main/java/timely/auth/JWTTokenHandler.java

@@ -34,7 +34,7 @@ import timely.common.configuration.SecurityProperties;
  */
 public class JWTTokenHandler {
 
-    private static final Logger logger = LoggerFactory.getLogger(JWTTokenHandler.class);
+    private static final Logger log = LoggerFactory.getLogger(JWTTokenHandler.class);
     private static Key signatureCheckKey;
     private static ObjectMapper objectMapper;
     private static Collection<String> accumuloAuths = new TreeSet<>();
@@ -68,17 +68,17 @@ public class JWTTokenHandler {
                 Authorizations currentAccumuloAuths = accumuloClient.securityOperations().getUserAuthorizations(accumuloClient.whoami());
                 currentAccumuloAuths.iterator().forEachRemaining(a -> accumuloAuths.add(new String(a, Charset.forName("UTF-8"))));
             } catch (Exception e) {
-                logger.error(e.getMessage(), e);
+                log.error(e.getMessage(), e);
             }
         }
     }
 
     public static Collection<TimelyUser> createUsersFromToken(String token, String claimName) {
-        logger.trace("Attempting to parse JWT {}", token);
+        log.trace("Attempting to parse JWT {}", token);
         JwtParser jwtParser = Jwts.parser().setSigningKey(signatureCheckKey).build();
         Jws<Claims> claimsJws = jwtParser.parseClaimsJws(token);
         Claims claims = claimsJws.getBody();
-        logger.trace("Resulting claims: {}", claims);
+        log.trace("Resulting claims: {}", claims);
         List<?> principalsClaim = claims.get(claimName, List.class);
         if (principalsClaim == null || principalsClaim.isEmpty()) {
             throw new IllegalArgumentException("JWT for " + claims.getSubject() + " does not contain any proxied principals.");

common/src/main/java/timely/auth/RemoteAuthenticationProvider.java

@@ -1,4 +1,4 @@
-package timely.common.component;
+package timely.auth;
 
 import static timely.common.component.AuthenticationService.PRINCIPALS_CLAIM;
 
@@ -14,14 +14,8 @@ import org.springframework.security.core.AuthenticationException;
 import org.springframework.security.core.userdetails.AuthenticationUserDetailsService;
 import org.springframework.security.core.userdetails.UsernameNotFoundException;
 import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationToken;
-import org.springframework.stereotype.Component;
 import org.springframework.web.reactive.function.client.WebClient;
 
-import timely.auth.JWTTokenHandler;
-import timely.auth.SubjectIssuerDNPair;
-import timely.auth.TimelyAuthenticationToken;
-import timely.auth.TimelyPrincipal;
-import timely.auth.TimelyUser;
 import timely.auth.util.ProxiedEntityUtils;
 import timely.common.configuration.SecurityProperties;
 
@@ -34,22 +28,23 @@ import timely.common.configuration.SecurityProperties;
  * production, it is likely better for the JWT to be retrieved by a load balancer or other API gateway and inject the JWT header before calling a service.
  * Therefore, this service is mostly useful for debugging and if there are problems with the suggested approach.
  */
-@Component
 public class RemoteAuthenticationProvider implements AuthenticationProvider {
 
-    private final Logger logger = LoggerFactory.getLogger(getClass());
+    private static final Logger log = LoggerFactory.getLogger(RemoteAuthenticationProvider.class);
     private final WebClient webClient;
 
     public static final String ENTITIES_HEADER = "X-ProxiedEntitiesChain";
     public static final String ISSUERS_HEADER = "X-ProxiedIssuersChain";
 
     public RemoteAuthenticationProvider(WebClient.Builder builder, SecurityProperties securityProperties) {
-        this.webClient = builder.baseUrl(securityProperties.getAuthorizationUri()).build();
+        log.debug("creating RemoteAuthenticationProvider with authorizationUrl:{} anonymousHttp:{} anonymousWs:{} requiredRoles:{} requiredAuths{}",
+                        securityProperties.getAuthorizationUrl(), securityProperties.isAllowAnonymousHttpAccess(),
+                        securityProperties.isAllowAnonymousWsAccess(), securityProperties.getRequiredRoles(), securityProperties.getRequiredAuths());
+        this.webClient = builder.baseUrl(securityProperties.getAuthorizationUrl()).build();
     }
 
     public TimelyPrincipal fetchTimelyPrincipal(TimelyAuthenticationToken token) throws UsernameNotFoundException {
-        logger.debug("Authenticating {} via the authorization microservice", token);
-
+        log.debug("Authenticating {}", token.getTimelyPrincipal().getUsername());
         TimelyPrincipal principal = token.getTimelyPrincipal();
         try {
             // @formatter:off
@@ -60,7 +55,7 @@ public class RemoteAuthenticationProvider implements AuthenticationProvider {
                 .bodyToMono(String.class)
                 .block();
             // @formatter:on
-            logger.debug("Authentication {} on remote authentication service with {} result", principal.getUsername(), (jwt == null) ? "null" : "non-null");
+            log.debug("Authenticated {} with {} result", principal.getUsername(), (jwt == null) ? "null" : "non-null");
             if (jwt != null) {
                 Collection<TimelyUser> principals = JWTTokenHandler.createUsersFromToken(jwt, PRINCIPALS_CLAIM);
                 long createTime = principals.stream().map(TimelyUser::getCreationTime).min(Long::compareTo).orElse(System.currentTimeMillis());
@@ -69,7 +64,7 @@ public class RemoteAuthenticationProvider implements AuthenticationProvider {
                 throw new UsernameNotFoundException("No entities found for " + principal.getUsername());
             }
         } catch (Exception e) {
-            logger.error("Failed performing lookup of {}: {}", principal.getUsername(), e);
+            log.error("Failed performing lookup of {}: {}", principal.getUsername(), e);
             throw new UsernameNotFoundException(e.getMessage(), e);
         }
     }

common/src/main/java/timely/auth/TimelyAuthenticationManager.java

@@ -0,0 +1,13 @@
+package timely.auth;
+
+import java.util.List;
+
+import org.springframework.security.authentication.AuthenticationProvider;
+import org.springframework.security.authentication.ProviderManager;
+
+public class TimelyAuthenticationManager extends ProviderManager {
+
+    public TimelyAuthenticationManager(List<AuthenticationProvider> providers) {
+        super(providers);
+    }
+}

common/src/main/java/timely/auth/TimelyUser.java

@@ -3,7 +3,7 @@ package timely.auth;
 import java.io.Serializable;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.LinkedHashSet;
+import java.util.TreeSet;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnore;
@@ -51,10 +51,10 @@ public class TimelyUser implements Serializable {
         this.userType = DnUtils.isServerDN(subjectDn) ? UserType.SERVER : UserType.USER;
         this.email = null;
         this.login = ProxiedEntityUtils.getShortName(dn.subjectDN());
-        this.auths = auths == null ? Collections.emptyList() : new LinkedHashSet<>(auths);
-        this.unmodifiableAuths = Collections.unmodifiableCollection(this.auths);
-        this.roles = roles == null ? Collections.emptyList() : new LinkedHashSet<>(roles);
-        this.unmodifiableRoles = Collections.unmodifiableCollection(this.roles);
+        this.auths = auths == null ? Collections.emptySortedSet() : new TreeSet<>(auths);
+        this.unmodifiableAuths = Collections.unmodifiableSortedSet(new TreeSet<>(this.auths));
+        this.roles = roles == null ? Collections.emptySortedSet() : new TreeSet<>(roles);
+        this.unmodifiableRoles = Collections.unmodifiableSortedSet(new TreeSet(this.roles));
         this.roleToAuthMapping = null;
         this.creationTime = System.currentTimeMillis();
         this.expirationTime = -1L;
@@ -89,10 +89,10 @@ public class TimelyUser implements Serializable {
         this.email = email;
         this.dn = dn;
         this.userType = userType;
-        this.auths = auths == null ? Collections.emptyList() : new LinkedHashSet<>(auths);
-        this.unmodifiableAuths = Collections.unmodifiableCollection(this.auths);
-        this.roles = roles == null ? Collections.emptyList() : new LinkedHashSet<>(roles);
-        this.unmodifiableRoles = Collections.unmodifiableCollection(this.roles);
+        this.auths = auths == null ? Collections.emptySortedSet() : new TreeSet<>(auths);
+        this.unmodifiableAuths = Collections.unmodifiableSortedSet(new TreeSet<>(this.auths));
+        this.roles = roles == null ? Collections.emptySortedSet() : new TreeSet<>(roles);
+        this.unmodifiableRoles = Collections.unmodifiableSortedSet(new TreeSet<>(this.roles));
         this.roleToAuthMapping = roleToAuthMapping == null ? LinkedHashMultimap.create() : LinkedHashMultimap.create(roleToAuthMapping);
         this.creationTime = creationTime;
         this.expirationTime = expirationTime;

common/src/main/java/timely/common/component/AuthenticationService.java

@@ -16,7 +16,7 @@ import org.apache.accumulo.core.security.Authorizations;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.security.authentication.AuthenticationManager;
+import org.springframework.security.authentication.ProviderManager;
 import org.springframework.security.core.Authentication;
 import org.springframework.stereotype.Component;
 
@@ -32,6 +32,7 @@ import timely.api.request.AuthenticatedWebSocketRequest;
 import timely.api.response.TimelyException;
 import timely.auth.JWTTokenHandler;
 import timely.auth.SubjectIssuerDNPair;
+import timely.auth.TimelyAuthenticationManager;
 import timely.auth.TimelyAuthenticationToken;
 import timely.auth.TimelyPrincipal;
 import timely.auth.TimelyUser;
@@ -46,7 +47,7 @@ public class AuthenticationService {
     public static final String AUTH_HEADER = "Authorization";
     public static final String PRINCIPALS_CLAIM = "principals";
 
-    private AuthenticationManager authenticationManager;
+    private ProviderManager authenticationManager;
     private ConcurrentHashMap<String,Object> fetchingEntity = new ConcurrentHashMap<>();
     private ConcurrentHashMap<String,Object> fetchingPrincipal = new ConcurrentHashMap<>();
 
@@ -153,7 +154,7 @@ public class AuthenticationService {
                                     }
                                     verifiedPrincipal = new TimelyPrincipal(authenticatedTimelyUsers);
                                     this.authCache.put(verifiedPrincipal.getName(), verifiedPrincipal);
-                                    log.debug("Authenticated user {} for request {} with authorizations {}", verifiedPrincipal.getName(), request.toString(),
+                                    log.debug("Authenticated user {} with authorizations {}", verifiedPrincipal.getName(),
                                                     verifiedPrincipal.getAuthorizationsString());
                                 } else {
                                     log.trace("Verified principal {} in AuthCache on 2nd attempt", entity);
@@ -204,10 +205,9 @@ public class AuthenticationService {
             String message = "";
             if (!missingRoles.isEmpty() && !missingAuths.isEmpty()) {
                 message = "User:" + primaryUser.getName() + " is missing role(s):" + missingRoles + " and auth(s):" + missingAuths;
-
             } else if (!missingRoles.isEmpty()) {
                 message = "User:" + primaryUser.getName() + " is missing role(s):" + missingRoles;
-            } else if (missingAuths.isEmpty()) {
+            } else if (!missingAuths.isEmpty()) {
                 message = "User:" + primaryUser.getName() + " is missing auths(s):" + missingAuths;
             }
             log.debug(message);

common/src/main/java/timely/common/component/ClientCertRestTemplateCustomizer.java

@@ -1,63 +0,0 @@
-package timely.common.component;
-
-import javax.net.ssl.SSLContext;
-
-import org.apache.http.client.HttpClient;
-import org.apache.http.conn.ssl.NoopHostnameVerifier;
-import org.apache.http.impl.client.HttpClientBuilder;
-import org.apache.http.impl.client.HttpClients;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.boot.web.client.RestTemplateCustomizer;
-import org.springframework.core.annotation.Order;
-import org.springframework.http.client.ClientHttpRequestFactory;
-import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
-import org.springframework.stereotype.Component;
-import org.springframework.web.client.RestTemplate;
-
-import timely.common.configuration.RestClientProperties;
-
-/**
- * Customizes Spring {@link RestTemplate} instances by using our configured SSL certificate to present a client certificate whenever asked by remote services.
- */
-@Component
-@Order(100) // execute this after standard customizers so we're sure to overwrite the client
-            // request factory
-@ConditionalOnBean(name = "outboundJDKSslContext")
-public class ClientCertRestTemplateCustomizer implements RestTemplateCustomizer {
-
-    private final SSLContext sslContext;
-    private final int maxConnectionsTotal;
-    private final int maxConnectionsPerRoute;
-
-    @Autowired
-    public ClientCertRestTemplateCustomizer(@Qualifier("outboundJDKSslContext") SSLContext sslContext, RestClientProperties restClientProperties) {
-        this.sslContext = sslContext;
-        this.maxConnectionsTotal = restClientProperties.getMaxConnectionsTotal();
-        this.maxConnectionsPerRoute = restClientProperties.getMaxConnectionsPerRoute();
-    }
-
-    @Override
-    public void customize(RestTemplate restTemplate) {
-        restTemplate.setRequestFactory(clientHttpRequestFactory());
-    }
-
-    protected ClientHttpRequestFactory clientHttpRequestFactory() {
-        HttpClient httpClient = customizeHttpClient(HttpClients.custom(), sslContext).build();
-        return new HttpComponentsClientHttpRequestFactory(httpClient);
-    }
-
-    protected HttpClientBuilder customizeHttpClient(HttpClientBuilder httpClientBuilder, SSLContext sslContext) {
-        if (sslContext != null) {
-            httpClientBuilder.setSSLContext(sslContext);
-        }
-        httpClientBuilder.setMaxConnTotal(maxConnectionsTotal);
-        httpClientBuilder.setMaxConnPerRoute(maxConnectionsPerRoute);
-        // TODO: We're allowing all hosts, since the cert presented by the service we're
-        // calling likely won't match its hostname (e.g., a docker host name)
-        // Instead, we could list the expected cert as a property (or use our server
-        // cert), and verify that the presented name matches.
-        return httpClientBuilder.setSSLHostnameVerifier(NoopHostnameVerifier.INSTANCE);
-    }
-}

common/src/main/java/timely/common/component/ClientCertWebClientCustomizer.java

@@ -1,36 +0,0 @@
-package timely.common.component;
-
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
-import org.springframework.boot.web.reactive.function.client.WebClientCustomizer;
-import org.springframework.core.annotation.Order;
-import org.springframework.http.client.reactive.ReactorClientHttpConnector;
-import org.springframework.stereotype.Component;
-import org.springframework.web.reactive.function.client.WebClient;
-
-import io.netty.handler.ssl.SslContext;
-import reactor.netty.http.client.HttpClient;
-
-/**
- * Customizes the Spring-provided {@link org.springframework.web.reactive.function.client.WebClient.Builder} in order to supply an {@link SslContext} that will
- * provide a client certificate to the remote server if one is requested.
- */
-@Component
-@Order(100) // execute this after standard customizers so we're sure to overwrite the client
-            // connector
-@ConditionalOnBean(name = "outboundNettySslContext")
-public class ClientCertWebClientCustomizer implements WebClientCustomizer {
-
-    private final SslContext sslContext;
-
-    public ClientCertWebClientCustomizer(@Qualifier("outboundNettySslContext") SslContext sslContext) {
-        this.sslContext = sslContext;
-    }
-
-    @Override
-    public void customize(WebClient.Builder webClientBuilder) {
-        HttpClient httpClient = HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext));
-        ReactorClientHttpConnector clientHttpConnector = new ReactorClientHttpConnector(httpClient);
-        webClientBuilder.clientConnector(clientHttpConnector);
-    }
-}

common/src/main/java/timely/common/component/TimelyAuthenticationManager.java

@@ -1,18 +0,0 @@
-package timely.common.component;
-
-import java.util.Arrays;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.security.authentication.ProviderManager;
-import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationProvider;
-import org.springframework.stereotype.Component;
-
-@Component
-public class TimelyAuthenticationManager extends ProviderManager {
-
-    @Autowired
-    public TimelyAuthenticationManager(RemoteAuthenticationProvider remoteAuthenticationProvider,
-                    PreAuthenticatedAuthenticationProvider fileAuthenticationProvider) {
-        super(Arrays.asList(remoteAuthenticationProvider, fileAuthenticationProvider));
-    }
-}

common/src/main/java/timely/common/configuration/AccumuloConfiguration.java

@@ -2,6 +2,7 @@ package timely.common.configuration;
 
 import org.apache.accumulo.core.client.Accumulo;
 import org.apache.accumulo.core.client.AccumuloClient;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
@@ -9,6 +10,7 @@ import org.springframework.context.annotation.Configuration;
 
 @Configuration
 @EnableConfigurationProperties(AccumuloProperties.class)
+@ConditionalOnExpression("T(org.apache.commons.lang3.StringUtils).isNotEmpty('${timely.accumulo.instance-name:}')")
 public class AccumuloConfiguration {
 
     @Bean

common/src/main/java/timely/common/configuration/AccumuloProperties.java

@@ -15,7 +15,6 @@ public class AccumuloProperties {
     private String username;
     @NotBlank
     private String password;
-    private int accumuloClientPoolSize = 16;
     @Valid
     @NestedConfigurationProperty
     private WriteProperties writeProperties = new WriteProperties();
@@ -54,12 +53,4 @@ public class AccumuloProperties {
     public void setPassword(String password) {
         this.password = password;
     }
-
-    public void setAccumuloClientPoolSize(int accumuloClientPoolSize) {
-        this.accumuloClientPoolSize = accumuloClientPoolSize;
-    }
-
-    public int getAccumuloClientPoolSize() {
-        return accumuloClientPoolSize;
-    }
 }

common/src/main/java/timely/common/configuration/AuthorizationProperties.java

@@ -1,46 +0,0 @@
-package timely.common.configuration;
-
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.cloud.context.config.annotation.RefreshScope;
-
-@RefreshScope
-@ConfigurationProperties(prefix = "timely.security.authorization.remote")
-public class AuthorizationProperties {
-
-    private boolean enabled = false;
-    private String authorizationUrl;
-    private String publicCertificateFile;
-    private String publicCertificateType;
-
-    public boolean isEnabled() {
-        return enabled;
-    }
-
-    public void setEnabled(boolean enabled) {
-        this.enabled = enabled;
-    }
-
-    public String getAuthorizationUrl() {
-        return authorizationUrl;
-    }
-
-    public void setAuthorizationUrl(String authorizationUrl) {
-        this.authorizationUrl = authorizationUrl;
-    }
-
-    public String getPublicCertificateFile() {
-        return publicCertificateFile;
-    }
-
-    public void setPublicCertificateFile(String publicCertificateFile) {
-        this.publicCertificateFile = publicCertificateFile;
-    }
-
-    public String getPublicCertificateType() {
-        return publicCertificateType;
-    }
-
-    public void setPublicCertificateType(String publicCertificateType) {
-        this.publicCertificateType = publicCertificateType;
-    }
-}

common/src/main/java/timely/common/configuration/CuratorConfiguration.java

@@ -11,7 +11,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Scope;
 
 @Configuration
 @ConditionalOnExpression("!T(org.springframework.util.StringUtils).isEmpty('${timely.zookeeper.servers:}')")
@@ -19,7 +18,6 @@ import org.springframework.context.annotation.Scope;
 public class CuratorConfiguration {
 
     @Bean(destroyMethod = "close")
-    @Scope("singleton")
     @ConditionalOnMissingBean
     public CuratorFramework curatorFramework(ZookeeperProperties zookeeperProperties) {
         RetryPolicy retryPolicy = new RetryForever(1000);

common/src/main/java/timely/common/configuration/HttpProperties.java

@@ -4,6 +4,7 @@ import javax.validation.Valid;
 import javax.validation.constraints.NotBlank;
 import javax.validation.constraints.NotNull;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.NestedConfigurationProperty;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
@@ -11,11 +12,12 @@ import org.springframework.cloud.context.config.annotation.RefreshScope;
 @RefreshScope
 @ConfigurationProperties(prefix = "timely.http")
 public class HttpProperties {
-
+    @Autowired
+    TimelyProperties timelyProperties;
     @NotBlank
     private String ip;
     @NotNull
-    private Integer port;
+    private Integer basePort;
     @NotNull
     private String host;
     private String redirectPath = "/secure-me";
@@ -24,6 +26,10 @@ public class HttpProperties {
     @NestedConfigurationProperty
     private CorsProperties corsProperties = new CorsProperties();
 
+    private int getPortOffset() {
+        return (timelyProperties.getInstance() - 1) * timelyProperties.getPortIncrement();
+    }
+
     public String getIp() {
         return ip;
     }
@@ -33,11 +39,15 @@ public class HttpProperties {
     }
 
     public int getPort() {
-        return port;
+        return basePort + getPortOffset();
     }
 
-    public void setPort(int port) {
-        this.port = port;
+    public void setBasePort(int basePort) {
+        this.basePort = basePort;
+    }
+
+    public Integer getBasePort() {
+        return basePort;
     }
 
     public String getHost() {

common/src/main/java/timely/common/configuration/SecurityConfiguration.java

@@ -0,0 +1,73 @@
+package timely.common.configuration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.client.reactive.ReactorClientHttpConnector;
+import org.springframework.security.web.authentication.preauth.PreAuthenticatedAuthenticationProvider;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import io.netty.handler.ssl.SslContext;
+import reactor.netty.http.client.HttpClient;
+import timely.auth.FileUserDetailsService;
+import timely.auth.RemoteAuthenticationProvider;
+import timely.auth.SubjectIssuerDNPair;
+import timely.auth.TimelyAuthenticationManager;
+import timely.auth.TimelyUser;
+
+@Configuration
+@EnableConfigurationProperties({SecurityProperties.class})
+public class SecurityConfiguration {
+
+    @Bean
+    @ConditionalOnExpression("T(org.apache.commons.lang3.StringUtils).isNotEmpty('${timely.security.authorization-url:}')")
+    public TimelyAuthenticationManager remoteAuthenticationManager(RemoteAuthenticationProvider remoteAuthenticationProvider) {
+        return new TimelyAuthenticationManager(Arrays.asList(remoteAuthenticationProvider));
+    }
+
+    @Bean
+    @ConditionalOnExpression("T(org.apache.commons.lang3.StringUtils).isNotEmpty('${timely.security.authorization-url:}')")
+    public RemoteAuthenticationProvider remoteAuthenticationProvider(SecurityProperties securityProperties, WebClient.Builder builder,
+                    @Qualifier("outboundNettySslContext") SslContext sslContext) {
+        HttpClient httpClient = HttpClient.create().secure(sslContextSpec -> sslContextSpec.sslContext(sslContext));
+        ReactorClientHttpConnector clientHttpConnector = new ReactorClientHttpConnector(httpClient);
+        builder.clientConnector(clientHttpConnector);
+        return new RemoteAuthenticationProvider(builder, securityProperties);
+    }
+
+    @Bean
+    @Qualifier("file")
+    @ConditionalOnExpression("T(org.apache.commons.lang3.StringUtils).isEmpty('${timely.security.authorization-url:}')")
+    public PreAuthenticatedAuthenticationProvider fileAuthenticationProvider(FileUserDetailsService fileUserDetailsService) {
+        PreAuthenticatedAuthenticationProvider authenticationProvider = new PreAuthenticatedAuthenticationProvider();
+        authenticationProvider.setPreAuthenticatedUserDetailsService(fileUserDetailsService);
+        return authenticationProvider;
+    }
+
+    @Bean
+    @ConditionalOnExpression("T(org.apache.commons.lang3.StringUtils).isEmpty('${timely.security.authorization-url:}')")
+    public TimelyAuthenticationManager fileAuthenticationManager(@Qualifier("file") PreAuthenticatedAuthenticationProvider fileAuthenticationProvider) {
+        return new TimelyAuthenticationManager(Arrays.asList(fileAuthenticationProvider));
+    }
+
+    @Bean
+    public FileUserDetailsService fileUserDetailsService(SecurityProperties securityProperties) {
+        List<TimelyUser> timelyUsers = new ArrayList<>();
+        List<AuthorizedUser> users = securityProperties.getAuthorizedUsers();
+        for (AuthorizedUser user : users) {
+            TimelyUser timelyUser = new TimelyUser(SubjectIssuerDNPair.of(user.getSubjectDn(), user.getIssuerDn()),
+                            TimelyUser.UserType.valueOf(user.getUserType()), "", user.getAuthsCollection(), user.getRolesCollection(), null,
+                            System.currentTimeMillis(), -1L);
+            timelyUsers.add(timelyUser);
+        }
+        FileUserDetailsService fileUserDetailsService = new FileUserDetailsService();
+        fileUserDetailsService.setUsers(timelyUsers);
+        return fileUserDetailsService;
+    }
+}

common/src/main/java/timely/common/configuration/SecurityProperties.java

@@ -24,7 +24,7 @@ public class SecurityProperties {
     @NestedConfigurationProperty
     private SslServerProperties serverSsl = new SslServerProperties();
 
-    private String authorizationUri = "https://authorization:8443/authorization/v1/authorize";
+    private String authorizationUrl;
     private List<String> requiredRoles;
     private List<String> requiredAuths;
 
@@ -86,12 +86,12 @@ public class SecurityProperties {
         return authorizedUsers;
     }
 
-    public void setAuthorizationUri(String authorizationUri) {
-        this.authorizationUri = authorizationUri;
+    public void setAuthorizationUrl(String authorizationUrl) {
+        this.authorizationUrl = authorizationUrl;
     }
 
-    public String getAuthorizationUri() {
-        return authorizationUri;
+    public String getAuthorizationUrl() {
+        return authorizationUrl;
     }
 
     public List<String> getRequiredAuths() {
@@ -109,4 +109,5 @@ public class SecurityProperties {
     public void setRequiredRoles(List<String> requiredRoles) {
         this.requiredRoles = requiredRoles;
     }
+
 }

common/src/main/java/timely/common/configuration/ServerProperties.java

@@ -3,21 +3,27 @@ package timely.common.configuration;
 import javax.validation.constraints.NotBlank;
 import javax.validation.constraints.NotNull;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
 
 @RefreshScope
 @ConfigurationProperties(prefix = "timely.server")
 public class ServerProperties {
-
+    @Autowired
+    TimelyProperties timelyProperties;
     @NotBlank
     private String ip;
     @NotNull
-    private Integer tcpPort;
+    private Integer tcpBasePort;
     @NotNull
-    private Integer udpPort;
+    private Integer udpBasePort;
     private Integer shutdownQuietPeriod = 5;
 
+    private int getPortOffset() {
+        return (timelyProperties.getInstance() - 1) * timelyProperties.getPortIncrement();
+    }
+
     @NotNull
     public String getIp() {
         return ip;
@@ -28,19 +34,27 @@ public class ServerProperties {
     }
 
     public int getTcpPort() {
-        return tcpPort;
+        return tcpBasePort + getPortOffset();
     }
 
-    public void setTcpPort(int tcpPort) {
-        this.tcpPort = tcpPort;
+    public void setTcpBasePort(int tcpPort) {
+        this.tcpBasePort = tcpPort;
+    }
+
+    public Integer getTcpBasePort() {
+        return tcpBasePort;
     }
 
     public int getUdpPort() {
-        return udpPort;
+        return udpBasePort + getPortOffset();
     }
 
-    public void setUdpPort(Integer udpPort) {
-        this.udpPort = udpPort;
+    public void setUdpBasePort(Integer udpBasePort) {
+        this.udpBasePort = udpBasePort;
+    }
+
+    public Integer getUdpBasePort() {
+        return udpBasePort;
     }
 
     /**

common/src/main/java/timely/common/configuration/SslClientConfig.java

@@ -2,6 +2,8 @@ package timely.common.configuration;
 
 import javax.net.ssl.SSLContext;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -15,12 +17,16 @@ import io.netty.handler.ssl.SslContext;
  */
 @Configuration
 @EnableConfigurationProperties({SslClientProperties.class})
-@ConditionalOnExpression("!T(org.springframework.util.StringUtils).isEmpty('${timely.security.clientSsl.keyStoreFile:}')")
+@ConditionalOnExpression("!T(org.springframework.util.StringUtils).isEmpty('${timely.security.client-ssl.key-store-file:}')")
 public class SslClientConfig {
 
+    private static final Logger log = LoggerFactory.getLogger(SslClientConfig.class);
+
     @Bean
     @Qualifier("outboundJDKSslContext")
     public SSLContext outboundJDKSslContext(SslClientProperties sslClientProperties) throws Exception {
+        log.debug("Creating outboundJDKSslContext using keyStoreFile:{} trustStoreFile:{}", sslClientProperties.getKeyStoreFile(),
+                        sslClientProperties.getTrustStoreFile());
         SSLContext sslContext = SSLContext.getInstance("TLS");
         sslContext.init(SslHelper.getKeyManagerFactory(sslClientProperties).getKeyManagers(),
                         SslHelper.getTrustManagerFactory(sslClientProperties).getTrustManagers(), null);
@@ -30,6 +36,8 @@ public class SslClientConfig {
     @Bean
     @Qualifier("outboundNettySslContext")
     public SslContext outboundNettySslContext(SslClientProperties sslClientProperties) throws Exception {
+        log.debug("Creating outboundNettySslContext using keyStoreFile:{} trustStoreFile:{}", sslClientProperties.getKeyStoreFile(),
+                        sslClientProperties.getTrustStoreFile());
         return SslHelper.getSslContext(sslClientProperties);
     }
 }

common/src/main/java/timely/common/configuration/SslClientProperties.java

@@ -7,8 +7,7 @@ import org.springframework.cloud.context.config.annotation.RefreshScope;
 @ConfigurationProperties(prefix = "timely.security.client-ssl")
 public class SslClientProperties implements SslProperties {
 
-    private String[] ciphers = new String[] {"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256", "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA", "TLS_RSA_WITH_AES_128_GCM_SHA256",
-            "TLS_RSA_WITH_AES_128_CBC_SHA", "SSL_RSA_WITH_3DES_EDE_CBC_SHA"};
+    private String[] ciphers;
     private String[] enabledProtocols;
     private String keyStoreType;
     private String keyStoreFile;

common/src/main/java/timely/common/configuration/SslServerConfig.java

@@ -27,7 +27,7 @@ public class SslServerConfig {
     private static final Logger log = LoggerFactory.getLogger(SslServerConfig.class);
 
     @Bean
-    protected SelfSignedCertificate selfSignedCertificate() throws Exception {
+    public SelfSignedCertificate selfSignedCertificate() throws Exception {
         Date begin = new Date();
         Date end = new Date(begin.getTime() + TimeUnit.DAYS.toMillis(7));
         return new SelfSignedCertificate("localhost", begin, end);
@@ -35,13 +35,15 @@ public class SslServerConfig {
 
     @Bean
     @Qualifier("nettySslContext")
-    protected SslContext nettySslContext(SslServerProperties sslServerProperties, SelfSignedCertificate ssc) throws Exception {
+    public SslContext nettySslContext(SslServerProperties sslServerProperties, SelfSignedCertificate ssc) throws Exception {
         Boolean generate = sslServerProperties.isUseGeneratedKeypair();
         SslContextBuilder sslContextBuilder;
         if (generate) {
-            log.warn("Using generated self signed server certificate");
+            log.debug("Creating nettySslContext using generated self signed server certificate");
             sslContextBuilder = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey());
         } else {
+            log.debug("Creating nettySslContext using keyStoreFile:{} trustStoreFile:{}", sslServerProperties.getKeyStoreFile(),
+                            sslServerProperties.getTrustStoreFile());
             sslContextBuilder = SslHelper.getSslContextBuilder(sslServerProperties);
         }
 

common/src/main/java/timely/common/configuration/SslServerProperties.java

@@ -1,7 +1,15 @@
 package timely.common.configuration;
 
+import java.net.URL;
+
+import javax.annotation.PostConstruct;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
+import org.springframework.util.ResourceUtils;
 
 import timely.validator.NotEmptyIfFieldSet;
 
@@ -11,6 +19,8 @@ import timely.validator.NotEmptyIfFieldSet;
 @ConfigurationProperties(prefix = "timely.security.server-ssl")
 public class SslServerProperties implements SslProperties {
 
+    private static final Logger log = LoggerFactory.getLogger(SslServerProperties.class);
+
     private String[] ciphers = new String[] {"TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256", "TLS_ECDHE_RSA_WITH_AES_128_CBC_SHA", "TLS_RSA_WITH_AES_128_GCM_SHA256",
             "TLS_RSA_WITH_AES_128_CBC_SHA", "SSL_RSA_WITH_3DES_EDE_CBC_SHA"};
     private String[] enabledProtocols;
@@ -25,6 +35,27 @@ public class SslServerProperties implements SslProperties {
     private boolean ignoreSslHandshakeErrors = false;
     private boolean useOpenssl = true;
 
+    @PostConstruct
+    public void setTrustStoreSystemProperties() {
+        try {
+            if (StringUtils.isNotBlank(trustStoreFile)) {
+                // This step is to resolve any classpath: prefixes used in trustStoreFile
+                URL url = ResourceUtils.getURL(trustStoreFile);
+                if (url != null) {
+                    System.setProperty("javax.net.ssl.trustStore", url.getFile());
+                    if (StringUtils.isNotBlank(trustStoreType)) {
+                        System.setProperty("javax.net.ssl.trustStoreType", trustStoreType);
+                    }
+                    if (StringUtils.isNotBlank(trustStorePassword)) {
+                        System.setProperty("javax.net.ssl.trustStorePassword", trustStorePassword);
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+        }
+    }
+
     public String[] getCiphers() {
         return ciphers;
     }

common/src/main/java/timely/common/configuration/TimelyCommonConfiguration.java

@@ -1,8 +1,12 @@
 package timely.common.configuration;
 
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import timely.util.Exclusions;
+
 @Configuration
 @EnableConfigurationProperties({AccumuloProperties.class, CacheProperties.class, CorsProperties.class, HttpProperties.class, MetaCacheProperties.class,
         RestClientProperties.class, ScanProperties.class, SecurityProperties.class, ServerProperties.class, SslClientProperties.class,
@@ -10,4 +14,15 @@ import org.springframework.context.annotation.Configuration;
 
 public class TimelyCommonConfiguration {
 
+    @Bean
+    public Exclusions exclusions(TimelyProperties timelyProperties) {
+        Exclusions exclusions = new Exclusions();
+        if (StringUtils.isNotBlank(timelyProperties.getFilteredMetricsFile())) {
+            exclusions.setFilteredMetricsFile(timelyProperties.getFilteredMetricsFile());
+        }
+        if (StringUtils.isNotBlank(timelyProperties.getFilteredTagsFile())) {
+            exclusions.setFilteredTagsFile(timelyProperties.getFilteredTagsFile());
+        }
+        return exclusions;
+    }
 }