add a queue length metric for http post sinks (#168)

* add a queue length metric for http post sinks

Author: BrandonArp

src/main/docker/Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM openjdk:8u212-jre-alpine
+FROM openjdk:8u212-jre-slim-buster
 
 MAINTAINER arpnetworking
 
@@ -31,7 +31,7 @@ ENV CAGG_CONFIG="/opt/cluster-aggregator/config/config.conf"
 ENV JAVA_OPTS=""
 
 # Build
-RUN apk add --no-cache libstdc++ su-exec && \
+RUN apt update && apt install -y gosu && rm -rf /var/lib/apt/lists/* && \
     mkdir -p /opt/cluster-aggregator/lib/ext && \
     mkdir -p /opt/cluster-aggregator/logs && \
     mkdir -p /opt/cluster-aggregator/config

src/main/java/com/arpnetworking/tsdcore/sinks/HttpSinkActor.java

@@ -110,6 +110,8 @@ public HttpSinkActor(
         _evictedRequestsName = "sinks/http_post/" + _sink.getMetricSafeName() + "/evicted_requests";
         _requestLatencyName = "sinks/http_post/" + _sink.getMetricSafeName() + "/request_latency";
         _inQueueLatencyName = "sinks/http_post/" + _sink.getMetricSafeName() + "/queue_time";
+        _pendingRequestsQueueSizeName = "sinks/http_post/" + _sink.getMetricSafeName() + "/queue_size";
+        _inflightRequestsCountName = "sinks/http_post/" + _sink.getMetricSafeName() + "/inflight_count";
         _requestSuccessName = "sinks/http_post/" + _sink.getMetricSafeName() + "/success";
         _responseStatusName = "sinks/http_post/" + _sink.getMetricSafeName() + "/status";
         _httpSinkAttemptsName = "sinks/http_post/" + _sink.getMetricSafeName() + "/attempts";
@@ -309,6 +311,7 @@ private void processEmitAggregation(final EmitAggregation emitMessage) {
                 .addData("conditionsSize", periodicData.getConditions().size())
                 .addContext("actor", self())
                 .log();
+        _periodicMetrics.recordGauge(_inflightRequestsCountName, _inflightRequestsCount);
 
         if (!periodicData.getData().isEmpty() || !periodicData.getConditions().isEmpty()) {
             final Collection<RequestEntry.Builder> requestEntryBuilders = _sink.createRequests(_client, periodicData);
@@ -331,6 +334,8 @@ private void processEmitAggregation(final EmitAggregation emitMessage) {
                         .log();
             }
 
+            _periodicMetrics.recordGauge(_pendingRequestsQueueSizeName, _pendingRequests.size());
+
             if (_spreadingDelayMillis > 0) {
                 // If we don't currently have anything in-flight, we'll need to wait the spreading duration.
                 if (!_waiting && pendingWasEmpty) {
@@ -442,6 +447,8 @@ private void scheduleRetry(final RequestEntry requestEntry, final int attempt) {
     private final String _evictedRequestsName;
     private final String _requestLatencyName;
     private final String _inQueueLatencyName;
+    private final String _pendingRequestsQueueSizeName;
+    private final String _inflightRequestsCountName;
     private final String _requestSuccessName;
     private final String _responseStatusName;
     private final String _httpSinkAttemptsName;