added request timeouts

puppet/manifests/init.pp

@@ -8,6 +8,7 @@
   $reader_host = '127.0.0.1',
   $reader_port = '8080',
   $reader_threads = '16',
+  $reader_request_timeout = '30',
   $reader_rollups = ['60s:5356800s','900s:62208000s'],
 
   $store_cluster = [$::ipaddress],

puppet/templates/disthene-reader.yaml.erb

@@ -2,6 +2,7 @@ reader:
   bind: "<%= @reader_host %>"
   port: <%= @reader_port %>
   threads: <%= @reader_threads %>
+  requestTimeout: <%= @reader_request_timeout %>
   rollups:
 <% @reader_rollups.each do |rollup| -%>
     - <%= rollup %>

src/main/java/net/iponweb/disthene/reader/DistheneReader.java

@@ -108,7 +108,7 @@ private void run() {
             readerServer.registerHandler(PING_PATH, pingHandler);
 
             logger.info("Creating render handler");
-            RenderHandler renderHandler = new RenderHandler(metricService, statsService, throttlingService);
+            RenderHandler renderHandler = new RenderHandler(metricService, statsService, throttlingService, distheneReaderConfiguration.getReader());
             readerServer.registerHandler(RENDER_PATH, renderHandler);
 
             logger.info("Starting reader");

src/main/java/net/iponweb/disthene/reader/config/ReaderConfiguration.java

@@ -10,6 +10,7 @@ public class ReaderConfiguration {
     private String bind;
     private int port;
     private int threads = 32;
+    private int requestTimeout = 30;
     private List<Rollup> rollups = new ArrayList<>();
 
     public String getBind() {
@@ -44,12 +45,21 @@ public void setThreads(int threads) {
         this.threads = threads;
     }
 
+    public int getRequestTimeout() {
+        return requestTimeout;
+    }
+
+    public void setRequestTimeout(int requestTimeout) {
+        this.requestTimeout = requestTimeout;
+    }
+
     @Override
     public String toString() {
         return "ReaderConfiguration{" +
                 "bind='" + bind + '\'' +
                 ", port=" + port +
                 ", threads=" + threads +
+                ", requestTimeout=" + requestTimeout +
                 ", rollups=" + rollups +
                 '}';
     }

src/main/java/net/iponweb/disthene/reader/handler/RenderHandler.java

@@ -1,12 +1,16 @@
 package net.iponweb.disthene.reader.handler;
 
 import com.google.common.base.Stopwatch;
-import io.netty.handler.codec.http.FullHttpResponse;
-import io.netty.handler.codec.http.HttpRequest;
-import io.netty.handler.codec.http.QueryStringDecoder;
+import com.google.common.util.concurrent.SimpleTimeLimiter;
+import com.google.common.util.concurrent.TimeLimiter;
+import com.google.common.util.concurrent.UncheckedTimeoutException;
+import io.netty.handler.codec.http.*;
 import net.iponweb.disthene.reader.beans.TimeSeries;
-import net.iponweb.disthene.reader.exceptions.*;
-import net.iponweb.disthene.reader.format.Format;
+import net.iponweb.disthene.reader.config.ReaderConfiguration;
+import net.iponweb.disthene.reader.exceptions.EvaluationException;
+import net.iponweb.disthene.reader.exceptions.InvalidParameterValueException;
+import net.iponweb.disthene.reader.exceptions.LogarithmicScaleNotAllowed;
+import net.iponweb.disthene.reader.exceptions.ParameterParsingException;
 import net.iponweb.disthene.reader.format.ResponseFormatter;
 import net.iponweb.disthene.reader.graphite.Target;
 import net.iponweb.disthene.reader.graphite.TargetVisitor;
@@ -22,14 +26,10 @@
 import org.antlr.v4.runtime.misc.ParseCancellationException;
 import org.antlr.v4.runtime.tree.ParseTree;
 import org.apache.log4j.Logger;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
-import sun.reflect.generics.reflectiveObjects.NotImplementedException;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 /**
  * @author Andrei Ivanov
@@ -41,16 +41,22 @@ public class RenderHandler implements DistheneReaderHandler {
     private TargetEvaluator evaluator;
     private StatsService statsService;
     private ThrottlingService throttlingService;
+    private ReaderConfiguration readerConfiguration;
 
-    public RenderHandler(MetricService metricService, StatsService statsService, ThrottlingService throttlingService) {
+    private static final ExecutorService executor = Executors.newCachedThreadPool();
+    private TimeLimiter timeLimiter = new SimpleTimeLimiter(executor);
+
+
+    public RenderHandler(MetricService metricService, StatsService statsService, ThrottlingService throttlingService, ReaderConfiguration readerConfiguration) {
         this.evaluator = new TargetEvaluator(metricService);
         this.statsService = statsService;
         this.throttlingService = throttlingService;
+        this.readerConfiguration = readerConfiguration;
     }
 
     @Override
-    public FullHttpResponse handle(HttpRequest request) throws ParameterParsingException, ExecutionException, InterruptedException, EvaluationException, LogarithmicScaleNotAllowed {
-        RenderParameters parameters = RenderParameters.parse(request);
+    public FullHttpResponse handle(final HttpRequest request) throws ParameterParsingException, ExecutionException, InterruptedException, EvaluationException, LogarithmicScaleNotAllowed {
+        final RenderParameters parameters = RenderParameters.parse(request);
 
         logger.debug("Got request: " + parameters);
         Stopwatch timer = Stopwatch.createStarted();
@@ -63,7 +69,7 @@ public FullHttpResponse handle(HttpRequest request) throws ParameterParsingExcep
             statsService.incThrottleTime(parameters.getTenant(), throttled);
         }
 
-        List<Target> targets = new ArrayList<>();
+        final List<Target> targets = new ArrayList<>();
 
         // Let's parse the targets
         for(String targetString : parameters.getTargets()) {
@@ -81,16 +87,35 @@ public FullHttpResponse handle(HttpRequest request) throws ParameterParsingExcep
             }
         }
 
+        FullHttpResponse response;
+        try {
+            response = timeLimiter.callWithTimeout(new Callable<FullHttpResponse>() {
+                @Override
+                public FullHttpResponse call() throws EvaluationException, LogarithmicScaleNotAllowed {
+                    return handleInternal(targets, parameters);
+                }
+            }, readerConfiguration.getRequestTimeout(), TimeUnit.SECONDS, true);
+        } catch (UncheckedTimeoutException e) {
+            logger.debug("Request timed out: " + parameters);
+            response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE);
+        } catch (Exception e) {
+            logger.error(e);
+            response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR);
+        }
+
+        timer.stop();
+        logger.debug("Request took " + timer.elapsed(TimeUnit.MILLISECONDS) + " milliseconds (" + parameters + ")");
+
+        return response;
+    }
+    private FullHttpResponse handleInternal(List<Target> targets, RenderParameters parameters) throws EvaluationException, LogarithmicScaleNotAllowed {
         // now evaluate each target producing list of TimeSeries
         List<TimeSeries> results = new ArrayList<>();
 
         for(Target target : targets) {
             results.addAll(evaluator.eval(target));
         }
 
-        timer.stop();
-        logger.debug("Request took " + timer.elapsed(TimeUnit.MILLISECONDS) + " milliseconds (" + parameters + ")");
-
         return ResponseFormatter.formatResponse(results, parameters);
     }