Http post sink (#99)

* Http post sink

* Fixed copyright comments and year.

Author: vjkoskela

README.md

@@ -147,9 +147,9 @@ For example:
             "name": "my_application_telemetry_sink"
         },
         {
-            "type": "com.arpnetworking.tsdcore.sinks.AggregationServerSink",
-            "name": "my_application_aggregation_server_sink",
-            "serverAddress": "192.168.0.2"
+            "type": "com.arpnetworking.tsdcore.sinks.AggregationServerHttpSink",
+            "name": "my_application_aggregation_server_http_sink",
+            "uri": "http://192.168.0.2:7066/metrics/v1/data"
         }
     ]
 }
@@ -284,6 +284,27 @@ interval="10s"
 [[inputs.kernel]]
 ```
 
+### Graphite
+
+Example MAD source configuration:
+
+```json
+{
+  type="com.arpnetworking.metrics.mad.sources.MappingSource"
+  name="graphitetcp_mapping_source"
+  actorName="graphite-tcp-source"
+  findAndReplace={
+    "\\."=["/"]
+  }
+  source={
+    type="com.arpnetworking.metrics.common.sources.TcpLineSource"
+    name="graphitetcp_source"
+    host="0.0.0.0"
+    port="2003"
+  }
+}
+```
+
 Development
 -----------
 

jdk-wrapper.sh

@@ -40,45 +40,109 @@ safe_command() {
   fi
 }
 
-download() {
+checksum() {
+  l_file="$1"
+  checksum_exec=""
+  if command -v sha256sum > /dev/null; then
+    checksum_exec="sha256sum"
+  elif command -v shasum > /dev/null; then
+    checksum_exec="shasum -a 256"
+  elif command -v sha1sum > /dev/null; then
+    checksum_exec="sha1sum"
+  elif command -v md5 > /dev/null; then
+    checksum_exec="md5"
+  fi
+  if [ -z "${checksum_exec}" ]; then
+    log_err "ERROR: No supported checksum command found!"
+    exit 1
+  fi
+  cat "${l_file}" | ${checksum_exec}
+}
+
+rand() {
+  awk 'BEGIN {srand();printf "%d\n", (rand() * 10^8);}'
+}
+
+download_if_needed() {
   file="$1"
-  if [ ! -f "${JDKW_PATH}/${file}" ]; then
-    jdkw_url="${JDKW_URI}/${file}"
+  path="$2"
+  if [ ! -f "${path}/${file}" ]; then
+    jdkw_url="${JDKW_BASE_URI}/releases/download/${JDKW_RELEASE}/${file}"
     log_out "Downloading ${file} from ${jdkw_url}"
-    safe_command "curl ${CURL_OPTIONS} -f -k -L -o \"${JDKW_PATH}/${file}\" \"${jdkw_url}\""
-    safe_command "chmod +x \"${JDKW_PATH}/${file}\""
+    safe_command "curl ${curl_options} -f -k -L -o \"${path}/${file}\" \"${jdkw_url}\""
+    safe_command "chmod +x \"${path}/${file}\""
   fi
 }
 
 # Default curl options
-CURL_OPTIONS=""
+curl_options=""
+
+# Process (but do not load) properties from environment
+env_configuration=
+l_fifo="${TMPDIR:-/tmp}/$$.$(rand)"
+safe_command "mkfifo \"${l_fifo}\""
+env > "${l_fifo}" &
+while IFS='=' read -r name value
+do
+  jdkw_arg=$(echo "${name}" | grep '^JDKW_.*')
+  jdkw_base_dir_arg=$(echo "${name}" | grep '^JDKW_BASE_DIR')
+  if [ -n "${jdkw_base_dir_arg}" ]; then
+    eval "${name}=\"${value}\""
+  fi
+  if [ -n "${jdkw_arg}" ]; then
+    env_configuration="${env_configuration}${name}=\"${value}\" "
+  fi
+done < "${l_fifo}"
+safe_command "rm \"${l_fifo}\""
+
+# Process (but do not load) properties from command line arguments
+command=
+cmd_configuration=
+for arg in "$@"; do
+  if [ -z ${in_command} ]; then
+    jdkw_arg=$(echo "${arg}" | grep '^JDKW_.*')
+    jdkw_base_dir_arg=$(echo "${arg}" | grep '^JDKW_BASE_DIR.*')
+    if [ -n "${jdkw_base_dir_arg}" ]; then
+      eval ${arg}
+    fi
+    if [ -n "${jdkw_arg}" ]; then
+      cmd_configuration="${cmd_configuration}${arg} "
+    fi
+  fi
+  case "${arg}" in
+    *\'*)
+       arg=`printf "%s" "$arg" | sed "s/'/'\"'\"'/g"`
+       ;;
+    *) : ;;
+  esac
+  command="${command} '${arg}'"
+done
+
+# Default base directory to current working directory
+if [ -z "${JDKW_BASE_DIR}" ]; then
+    JDKW_BASE_DIR="."
+fi
 
 # Load properties file in home directory
 if [ -f "${HOME}/.jdkw" ]; then
   . "${HOME}/.jdkw"
 fi
 
-# Load properties file in working directory
-if [ -f ".jdkw" ]; then
-  . "./.jdkw"
+# Load properties file in base directory
+if [ -f "${JDKW_BASE_DIR}/.jdkw" ]; then
+  . "${JDKW_BASE_DIR}/.jdkw"
 fi
 
-# Process command line arguments
-for ARG in "$@"; do
-  JDKW_ARG=$(echo "${ARG}" | grep 'JDKW_.*')
-  if [ -n "${JDKW_ARG}" ]; then
-    eval ${ARG}
-  else
-    break
-  fi
-done
+# Load properties from environment
+eval "${env_configuration}"
 
-# Globals
-JDKW_BASE_URI="https://github.com/KoskiLabs/jdk-wrapper"
-JDKW_IMPL="jdkw-impl.sh"
-JDKW_WRAPPER="jdk-wrapper.sh"
+# Load properties from command line arguments
+eval "${cmd_configuration}"
 
 # Process configuration
+if [ -z "${JDKW_BASE_URI}" ]; then
+    JDKW_BASE_URI="https://github.com/KoskiLabs/jdk-wrapper"
+fi
 if [ -z "${JDKW_RELEASE}" ]; then
   JDKW_RELEASE="latest"
   log_out "Defaulted to version ${JDKW_RELEASE}"
@@ -88,38 +152,42 @@ if [ -z "${JDKW_TARGET}" ]; then
   log_out "Defaulted to target ${JDKW_TARGET}"
 fi
 if [ -z "${JDKW_VERBOSE}" ]; then
-  CURL_OPTIONS="${CURL_OPTIONS} --silent"
+  curl_options="${curl_options} --silent"
 fi
 
 # Resolve latest version
 if [ "${JDKW_RELEASE}" = "latest" ]; then
-  JDKW_RELEASE=$(curl ${CURL_OPTIONS} -f -k -L -H 'Accept: application/json' "${JDKW_BASE_URI}/releases/latest" | sed -e 's/.*"tag_name":"\([^"]*\)".*/\1/')
+  latest_version_json="${TMPDIR:-/tmp}/jdkw-latest-version-$$.$(rand)"
+  safe_command "curl ${curl_options} -f -k -L -o \"${latest_version_json}\" -H 'Accept: application/json' \"${JDKW_BASE_URI}/releases/latest\""
+  JDKW_RELEASE=$(cat "${latest_version_json}" | sed -e 's/.*"tag_name":"\([^"]*\)".*/\1/')
+  rm -f "${latest_version_json}"
   log_out "Resolved latest version to ${JDKW_RELEASE}"
 fi
 
-# Define source and target
-JDKW_URI="${JDKW_BASE_URI}/releases/download/${JDKW_RELEASE}"
-JDKW_PATH="${JDKW_TARGET}/jdkw/${JDKW_RELEASE}"
-
 # Ensure target directory exists
-if [ ! -d "${JDKW_PATH}" ]; then
-  log_out "Creating target directory ${JDKW_PATH}"
-  safe_command "mkdir -p \"${JDKW_PATH}\""
+jdkw_path="${JDKW_TARGET}/jdkw/${JDKW_RELEASE}"
+if [ ! -d "${jdkw_path}" ]; then
+  log_out "Creating target directory ${jdkw_path}"
+  safe_command "mkdir -p \"${jdkw_path}\""
 fi
 
 # Download the jdk wrapper version
-download "${JDKW_IMPL}"
-download "${JDKW_WRAPPER}"
+jdkw_impl="jdkw-impl.sh"
+jdkw_wrapper="jdk-wrapper.sh"
+download_if_needed "${jdkw_impl}" "${jdkw_path}"
+download_if_needed "${jdkw_wrapper}" "${jdkw_path}"
+
+# Execute the provided command
+eval ${jdkw_path}/${jdkw_impl} ${command}
+result=$?
 
 # Check whether this wrapper is the one specified for this version
-jdkw_download="${JDKW_PATH}/${JDKW_WRAPPER}"
+jdkw_download="${jdkw_path}/${jdkw_wrapper}"
 jdkw_current="$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)/$(basename "$0")"
-if [ "$(cat "${jdkw_download}" | sha1sum )" != "$(cat "${jdkw_current}" | sha1sum)" ]; then
+if [ "$(checksum "${jdkw_download}")" != "$(checksum "${jdkw_current}")" ]; then
   printf "\e[0;31m[WARNING]\e[0m Your jdk-wrapper.sh file does not match the one in your JDKW_RELEASE.\n"
   printf "\e[0;32mUpdate your jdk-wrapper.sh to match by running:\e[0m\n"
   printf "cp \"%s\" \"%s\"\n" "${jdkw_download}" "${jdkw_current}"
 fi
 
-# Execute the provided command
-${JDKW_PATH}/${JDKW_IMPL} $@
-exit $?
+exit ${result}

pom.xml

@@ -77,6 +77,7 @@
     <apache.httpcore.version>4.4.3</apache.httpcore.version>
     <arpnetworking.commons.version>1.16.0</arpnetworking.commons.version>
     <aspectjrt.version>1.8.9</aspectjrt.version>
+    <asynchttpclient.version>2.4.8</asynchttpclient.version>
     <cglib.version>3.2.5</cglib.version>
     <client.protocol.version>0.10.0</client.protocol.version>
     <fastutil.version>8.1.1</fastutil.version>
@@ -576,16 +577,21 @@
       <artifactId>scala-library</artifactId>
       <version>${scala.library.version}</version>
     </dependency>
-    <dependency>
-      <groupId>io.vertx</groupId>
-      <artifactId>vertx-core</artifactId>
-      <version>${vertx.core.version}</version>
-    </dependency>
     <dependency>
       <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpclient</artifactId>
       <version>${apache.httpclient.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.asynchttpclient</groupId>
+      <artifactId>async-http-client</artifactId>
+      <version>${asynchttpclient.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>io.vertx</groupId>
+      <artifactId>vertx-core</artifactId>
+      <version>${vertx.core.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpcore</artifactId>
@@ -769,7 +775,7 @@
                   <directory>/opt/mad</directory>
                   <sources>
                     <source>
-                      <location>target/appassembler</location>
+                      <location>${buildDirectory}/appassembler</location>
                     </source>
                   </sources>
                 </mapping>

src/main/docker/Dockerfile

@@ -43,7 +43,7 @@ ENTRYPOINT [ \
     "-XX:HeapDumpPath=/opt/mad/logs/mad.oom.hprof", \
     "-XX:+PrintGCDetails", \
     "-XX:+PrintGCDateStamps", \
-    "-Xloggc:logs/mad.gc.log", \
+    "-Xloggc:/opt/mad/logs/mad.gc.log", \
     "-XX:NumberOfGCLogFiles=2", \
     "-XX:GCLogFileSize=50M", \
     "-XX:+UseGCLogFileRotation", \

src/main/java/com/arpnetworking/tsdcore/model/AggregationMessage.java

@@ -15,12 +15,17 @@
  */
 package com.arpnetworking.tsdcore.model;
 
+import akka.util.ByteString;
+import akka.util.ByteStringBuilder;
 import com.arpnetworking.metrics.aggregation.protocol.Messages;
 import com.google.protobuf.GeneratedMessage;
 import org.vertx.java.core.buffer.Buffer;
 
+import java.io.IOException;
+import java.nio.ByteOrder;
+
 /**
- * Class for building messages from the raw, on-the-wire bytes in the TCP stream.
+ * Class for building on-the-wire bytes for messages.
  *
  * @author Brandon Arp (brandon dot arp at inscopemetrics dot com)
  */
@@ -41,7 +46,7 @@ public static AggregationMessage create(final GeneratedMessage message) {
      *
      * @return <code>Buffer</code> containing serialized message.
      */
-    public Buffer serialize() {
+    public Buffer serializeToBuffer() {
         final Buffer b = new Buffer();
         b.appendInt(0);
         if (_message instanceof Messages.HostIdentification) {
@@ -64,6 +69,40 @@ public Buffer serialize() {
         return b;
     }
 
+    /**
+     * Serialize the message into a <code>ByteString</code>.
+     *
+     * @return <code>Buffer</code> containing serialized message.
+     */
+    public ByteString serializeToByteString() {
+        final ByteStringBuilder b = ByteString.createBuilder();
+        if (_message instanceof Messages.HostIdentification) {
+            b.putByte((byte) 0x01);
+        } else if (_message instanceof Messages.HeartbeatRecord) {
+            b.putByte((byte) 0x03);
+        } else if (_message instanceof Messages.StatisticSetRecord) {
+            b.putByte((byte) 0x04);
+        } else if (_message instanceof Messages.SamplesSupportingData) {
+            b.putByte((byte) 0x05);
+            b.putByte((byte) 0x01);
+        } else if (_message instanceof Messages.SparseHistogramSupportingData) {
+            b.putByte((byte) 0x05);
+            b.putByte((byte) 0x02);
+        } else {
+            throw new IllegalArgumentException(
+                    String.format("Unsupported message; messageClass=%s", _message.getClass()));
+        }
+        try {
+            _message.writeTo(b.asOutputStream());
+        } catch (final IOException e) {
+            throw new RuntimeException(e);
+        }
+        final ByteString bs = b.result();
+        final ByteStringBuilder sizePrefix = ByteString.createBuilder();
+        sizePrefix.putInt(bs.size() + INTEGER_SIZE_IN_BYTES, ByteOrder.BIG_ENDIAN);
+        return sizePrefix.result().concat(bs);
+    }
+
     public GeneratedMessage getMessage() {
         return _message;
     }