Configurable accept http status codes. (#229)

Author: vjkoskela

config/pipelines/pipeline.conf

@@ -214,6 +214,7 @@ sinks=[
     sink={
       type="com.arpnetworking.tsdcore.sinks.AggregationServerHttpSink"
       name="cluster_http_sink"
+      #acceptedStatusCodes=[200, 201, 202, 204]
       uri="http://"${cluster_aggregator_host}":"${cluster_aggregator_port}"/metrics/v1/data/persist"
     }
   }

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

@@ -20,6 +20,7 @@
 import akka.actor.PoisonPill;
 import akka.http.javadsl.model.HttpMethods;
 import akka.http.javadsl.model.MediaTypes;
+import akka.http.javadsl.model.StatusCodes;
 import com.arpnetworking.logback.annotations.LogValue;
 import com.arpnetworking.metrics.MetricsFactory;
 import com.arpnetworking.steno.LogValueMapFactory;
@@ -28,6 +29,7 @@
 import com.arpnetworking.tsdcore.model.PeriodicData;
 import com.arpnetworking.tsdcore.model.RequestEntry;
 import com.fasterxml.jackson.annotation.JacksonInject;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Lists;
 import net.sf.oval.constraint.Min;
 import net.sf.oval.constraint.NotNull;
@@ -145,6 +147,15 @@ protected Uri getAysncHttpClientUri() {
         return _aysncHttpClientUri;
     }
 
+    /**
+     * Accessor for the status codes accepted as successful.
+     *
+     * @return the status codes accepted as success
+     */
+    ImmutableSet<Integer> getAcceptedStatusCodes() {
+        return _acceptedStatusCodes;
+    }
+
     /**
      * Serialize the {@link PeriodicData} instances for posting.
      *
@@ -161,6 +172,7 @@ protected Uri getAysncHttpClientUri() {
     protected HttpPostSink(final Builder<?, ?> builder) {
         super(builder);
         _uri = builder._uri;
+        _acceptedStatusCodes = builder._acceptedStatusCodes;
         _aysncHttpClientUri = Uri.create(_uri.toString());
 
         _sinkActor = builder._actorSystem.actorOf(
@@ -176,6 +188,7 @@ protected HttpPostSink(final Builder<?, ?> builder) {
     private final URI _uri;
     private final Uri _aysncHttpClientUri;
     private final ActorRef _sinkActor;
+    private final ImmutableSet<Integer> _acceptedStatusCodes;
 
     private static final Logger LOGGER = LoggerFactory.getLogger(HttpPostSink.class);
     private static final AsyncHttpClient CLIENT;
@@ -266,6 +279,18 @@ public B setMetricsFactory(final MetricsFactory value) {
             return self();
         }
 
+        /**
+         * Sets the http status codes accepted as success. Optional. Cannot be null.
+         * Default is: [200, 201, 202, 204]
+         *
+         * @param value the status codes accepted as success
+         * @return this builder
+         */
+        public B setAcceptedStatusCodes(final ImmutableSet<Integer> value) {
+            _acceptedStatusCodes = value;
+            return self();
+        }
+
         /**
          * Protected constructor for subclasses.
          *
@@ -291,6 +316,19 @@ protected Builder(final Function<B, S> targetConstructor) {
         @JacksonInject
         @NotNull
         private MetricsFactory _metricsFactory;
+        @NotNull
+        private ImmutableSet<Integer> _acceptedStatusCodes = DEFAULT_ACCEPTED_STATUS_CODES;
+
+        private static final ImmutableSet<Integer> DEFAULT_ACCEPTED_STATUS_CODES;
+
+        static {
+            DEFAULT_ACCEPTED_STATUS_CODES = ImmutableSet.<Integer>builder()
+                    .add(StatusCodes.OK.intValue())
+                    .add(StatusCodes.CREATED.intValue())
+                    .add(StatusCodes.ACCEPTED.intValue())
+                    .add(StatusCodes.NO_CONTENT.intValue())
+                    .build();
+        }
     }
 
     static final class SerializedDatum {

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

@@ -17,7 +17,6 @@
 
 import akka.actor.AbstractActor;
 import akka.actor.Props;
-import akka.http.javadsl.model.StatusCodes;
 import akka.pattern.PatternsCS;
 import com.arpnetworking.logback.annotations.LogValue;
 import com.arpnetworking.metrics.Metrics;
@@ -31,7 +30,7 @@
 import com.google.common.base.Charsets;
 import com.google.common.collect.EvictingQueue;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Sets;
+import com.google.common.collect.ImmutableSet;
 import org.asynchttpclient.AsyncCompletionHandler;
 import org.asynchttpclient.AsyncHttpClient;
 import org.asynchttpclient.Request;
@@ -44,7 +43,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Random;
-import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
@@ -95,6 +93,7 @@ public HttpPostSinkActor(
             final MetricsFactory metricsFactory) {
         _client = client;
         _sink = sink;
+        _acceptedStatusCodes = sink.getAcceptedStatusCodes();
         _maximumConcurrency = maximumConcurrency;
         _pendingRequests = EvictingQueue.create(maximumQueueSize);
         if (Objects.equals(Duration.ZERO, spreadPeriod)) {
@@ -121,6 +120,7 @@ public HttpPostSinkActor(
     public Object toLogValue() {
         return LogValueMapFactory.builder(this)
                 .put("sink", _sink)
+                .put("acceptedStatusCodes", _acceptedStatusCodes)
                 .put("maximumConcurrency", _maximumConcurrency)
                 .put("spreadingDelayMillis", _spreadingDelayMillis)
                 .put("waiting", _waiting)
@@ -293,7 +293,7 @@ private void fireNextRequest() {
                                     String.format("%s/%dxx", _responseStatusName, i),
                                     responseStatusClass == i ? 1 : 0);
                         }
-                        if (ACCEPTED_STATUS_CODES.contains(responseStatusCode)) {
+                        if (_acceptedStatusCodes.contains(responseStatusCode)) {
                              returnValue =  new PostSuccess(result);
                              metrics.incrementCounter(_samplesSentName, requestEntry.getPopulationSize());
                         } else {
@@ -330,6 +330,7 @@ public void postStop() throws Exception {
     private final HttpPostSink _sink;
     private final int _spreadingDelayMillis;
     private final MetricsFactory _metricsFactory;
+    private final ImmutableSet<Integer> _acceptedStatusCodes;
 
     private final String _evictedRequestsName;
     private final String _requestLatencyName;
@@ -339,22 +340,11 @@ public void postStop() throws Exception {
     private final String _samplesDroppedName;
     private final String _samplesSentName;
 
-
     private static final Logger LOGGER = LoggerFactory.getLogger(HttpPostSink.class);
     private static final Logger POST_ERROR_LOGGER = LoggerFactory.getRateLimitLogger(HttpPostSink.class, Duration.ofSeconds(30));
     private static final Logger EVICTED_LOGGER = LoggerFactory.getRateLimitLogger(HttpPostSink.class, Duration.ofSeconds(30));
-    private static final Set<Integer> ACCEPTED_STATUS_CODES = Sets.newHashSet();
     private static final com.google.common.collect.ImmutableList<Integer> STATUS_CLASSES = ImmutableList.of(2, 3, 4, 5);
 
-
-    static {
-        // TODO(vkoskela): Make accepted status codes configurable [AINT-682]
-        ACCEPTED_STATUS_CODES.add(StatusCodes.OK.intValue());
-        ACCEPTED_STATUS_CODES.add(StatusCodes.CREATED.intValue());
-        ACCEPTED_STATUS_CODES.add(StatusCodes.ACCEPTED.intValue());
-        ACCEPTED_STATUS_CODES.add(StatusCodes.NO_CONTENT.intValue());
-    }
-
     /**
      * Message class to wrap a list of {@link AggregatedData}.
      */