From 391e6154d0bb4bcaedefcfa6d2dfecef3d445d66 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 13 Jul 2017 10:09:39 -0500 Subject: [PATCH 01/12] add configs to enable fast request failure on broker --- docs/content/configuration/broker.md | 2 + .../server/initialization/ServerConfig.java | 9 ++ .../initialization/jetty/JettyBindings.java | 9 ++ .../jetty/JettyServerModule.java | 18 ++- .../jetty/LimitRequestsFilterHolder.java | 115 ++++++++++++++++++ .../src/main/java/io/druid/cli/CliBroker.java | 13 ++ 6 files changed, 163 insertions(+), 3 deletions(-) create mode 100644 server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilterHolder.java diff --git a/docs/content/configuration/broker.md b/docs/content/configuration/broker.md index 349d96442ffd..6c46f9a02fb4 100644 --- a/docs/content/configuration/broker.md +++ b/docs/content/configuration/broker.md @@ -36,7 +36,9 @@ Druid uses Jetty to serve HTTP requests. |Property|Description|Default| |--------|-----------|-------| |`druid.server.http.numThreads`|Number of threads for HTTP requests.|max(10, (Number of cores * 17) / 16 + 2) + 30| +|`druid.server.http.queueSize`|Size of the worker queue used by Jetty server to temporarily store incoming client connections. If this value is set and a request is rejected by jetty because queue is full then client would observe request failure with TCP connection being closed immediately with a completely empty response from server.|Unbounded| |`druid.server.http.maxIdleTime`|The Jetty max idle time for a connection.|PT5m| +|`druid.server.http.maxActiveRequests`|Number of active requests that can be handled by broker concurrently, beyond which "HTTP 429 Too Many Requests" error response is sent. Note that this must be set smaller than `druid.server.http.numThreads` to be effective or requests might just get queued at jetty server.|Not Applied By Default| |`druid.server.http.defaultQueryTimeout`|Query timeout in millis, beyond which unfinished queries will be cancelled|300000| |`druid.server.http.maxScatterGatherBytes`|Maximum number of bytes gathered from data nodes such as historicals and realtime processes to execute a query. This is an advance configuration that allows to protect in case broker is under heavy load and not utilizing the data gathered in memory fast enough and leading to OOMs. This limit can be further reduced at query time using `maxScatterGatherBytes` in the context. Note that having large limit is not necessarily bad if broker is never under heavy concurrent load in which case data gathered is processed quickly and freeing up the memory used.|Long.MAX_VALUE| |`druid.broker.http.numConnections`|Size of connection pool for the Broker to connect to historical and real-time processes. If there are more queries than this number that all need to speak to the same node, then they will queue up.|20| diff --git a/server/src/main/java/io/druid/server/initialization/ServerConfig.java b/server/src/main/java/io/druid/server/initialization/ServerConfig.java index b299908bca4d..307e95aa7b08 100644 --- a/server/src/main/java/io/druid/server/initialization/ServerConfig.java +++ b/server/src/main/java/io/druid/server/initialization/ServerConfig.java @@ -33,6 +33,10 @@ public class ServerConfig @Min(1) private int numThreads = Math.max(10, (Runtime.getRuntime().availableProcessors() * 17) / 16 + 2) + 30; + @JsonProperty + @Min(1) + private int queueSize = Integer.MAX_VALUE; + @JsonProperty @NotNull private Period maxIdleTime = new Period("PT5m"); @@ -56,6 +60,11 @@ public int getNumThreads() return numThreads; } + public int getQueueSize() + { + return queueSize; + } + public Period getMaxIdleTime() { return maxIdleTime; diff --git a/server/src/main/java/io/druid/server/initialization/jetty/JettyBindings.java b/server/src/main/java/io/druid/server/initialization/jetty/JettyBindings.java index e05b664e14c5..0bd8359303e1 100644 --- a/server/src/main/java/io/druid/server/initialization/jetty/JettyBindings.java +++ b/server/src/main/java/io/druid/server/initialization/jetty/JettyBindings.java @@ -51,6 +51,15 @@ public static void addQosFilter(Binder binder, String path, int maxRequests) .toInstance(new QosFilterHolder(path, maxRequests)); } + public static void addLimitRequestsFilter(Binder binder, String path, int maxActiveRequests) + { + if (maxActiveRequests > 0) { + Multibinder.newSetBinder(binder, ServletFilterHolder.class) + .addBinding() + .toInstance(new LimitRequestsFilterHolder(path, maxActiveRequests)); + } + } + public static void addHandler(Binder binder, Class handlerClass) { Multibinder.newSetBinder(binder, Handler.class) diff --git a/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java b/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java index 424eb21d322b..062b6c116355 100644 --- a/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java +++ b/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java @@ -76,6 +76,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicInteger; /** @@ -169,9 +170,20 @@ public JacksonSmileProvider getJacksonSmileProvider(@Smile ObjectMapper objectMa static Server makeJettyServer(DruidNode node, ServerConfig config, TLSServerConfig tlsServerConfig) { - final QueuedThreadPool threadPool = new QueuedThreadPool(); - threadPool.setMinThreads(config.getNumThreads()); - threadPool.setMaxThreads(config.getNumThreads()); + final QueuedThreadPool threadPool; + if (config.getQueueSize() == Integer.MAX_VALUE) { + threadPool = new QueuedThreadPool(); + threadPool.setMinThreads(config.getNumThreads()); + threadPool.setMaxThreads(config.getNumThreads()); + } else { + threadPool = new QueuedThreadPool( + config.getNumThreads(), + config.getNumThreads(), + 60000, + new LinkedBlockingQueue<>(config.getQueueSize()) + ); + } + threadPool.setDaemon(true); final Server server = new Server(threadPool); diff --git a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilterHolder.java b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilterHolder.java new file mode 100644 index 000000000000..d2744b380b04 --- /dev/null +++ b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilterHolder.java @@ -0,0 +1,115 @@ +/* + * Licensed to Metamarkets Group Inc. (Metamarkets) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Metamarkets licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package io.druid.server.initialization.jetty; + +import com.google.common.collect.ImmutableMap; + +import javax.servlet.DispatcherType; +import javax.servlet.Filter; +import javax.servlet.FilterChain; +import javax.servlet.FilterConfig; +import javax.servlet.ServletException; +import javax.servlet.ServletRequest; +import javax.servlet.ServletResponse; +import javax.servlet.http.HttpServletResponse; +import java.io.IOException; +import java.util.EnumSet; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + +public class LimitRequestsFilterHolder implements ServletFilterHolder +{ + private final String path; + private final int maxActiveRequests; + + public LimitRequestsFilterHolder(String path, int maxActiveRequests) + { + this.path = path; + this.maxActiveRequests = maxActiveRequests; + } + + @Override + public Filter getFilter() + { + return new LimitRequestsFilter(maxActiveRequests); + } + + @Override + public Class getFilterClass() + { + return null; + } + + @Override + public Map getInitParameters() + { + return ImmutableMap.of(); + } + + @Override + public String getPath() + { + return path; + } + + @Override + public EnumSet getDispatcherType() + { + return null; + } + + private static class LimitRequestsFilter implements Filter + { + private final int maxActiveRequests; + + private AtomicInteger activeRequestsCount = new AtomicInteger(); + + public LimitRequestsFilter(int maxActiveRequests) + { + this.maxActiveRequests = maxActiveRequests; + } + + @Override + public void init(FilterConfig filterConfig) throws ServletException + { + } + + @Override + public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) + throws IOException, ServletException + { + + int curr = activeRequestsCount.incrementAndGet(); + try { + if (curr <= maxActiveRequests) { + chain.doFilter(request, response); + } else { + // See https://tools.ietf.org/html/rfc6585 for status code 429 explanation. + ((HttpServletResponse)response).sendError(429, "Too Many Requests"); + } + } finally { + activeRequestsCount.decrementAndGet(); + } + } + + @Override + public void destroy() { } + } +} diff --git a/services/src/main/java/io/druid/cli/CliBroker.java b/services/src/main/java/io/druid/cli/CliBroker.java index 0745c4d02a0d..e1dc20e03f9d 100644 --- a/services/src/main/java/io/druid/cli/CliBroker.java +++ b/services/src/main/java/io/druid/cli/CliBroker.java @@ -21,6 +21,7 @@ import com.google.common.collect.ImmutableList; import com.google.inject.Binder; +import com.google.inject.Inject; import com.google.inject.Module; import com.google.inject.name.Names; import io.airlift.airline.Command; @@ -50,6 +51,7 @@ import io.druid.server.ClientQuerySegmentWalker; import io.druid.server.coordination.broker.DruidBroker; import io.druid.server.http.BrokerResource; +import io.druid.server.initialization.jetty.JettyBindings; import io.druid.server.initialization.jetty.JettyServerInitializer; import io.druid.server.metrics.MetricsModule; import io.druid.server.metrics.QueryCountStatsProvider; @@ -58,6 +60,7 @@ import org.eclipse.jetty.server.Server; import java.util.List; +import java.util.Properties; /** */ @@ -69,6 +72,11 @@ public class CliBroker extends ServerRunnable { private static final Logger log = new Logger(CliBroker.class); + private static final String MAX_ACTIVE_REQUESTS_KEY = "druid.server.http.maxActiveRequests"; + + @Inject + private Properties properties; + public CliBroker() { super(log); @@ -109,6 +117,11 @@ public void configure(Binder binder) binder.bind(JettyServerInitializer.class).to(QueryJettyServerInitializer.class).in(LazySingleton.class); + if (properties.getProperty(MAX_ACTIVE_REQUESTS_KEY) != null) { + final int maxActiveRequests = Integer.parseInt(properties.getProperty(MAX_ACTIVE_REQUESTS_KEY)); + JettyBindings.addLimitRequestsFilter(binder, "/*", maxActiveRequests); + } + Jerseys.addResource(binder, BrokerQueryResource.class); binder.bind(QueryCountStatsProvider.class).to(BrokerQueryResource.class).in(LazySingleton.class); Jerseys.addResource(binder, BrokerResource.class); From 705849591a4c614af24b7a7e6358420406412601 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 13 Jul 2017 16:31:22 -0500 Subject: [PATCH 02/12] address review comments --- docs/content/configuration/broker.md | 2 +- docs/content/configuration/historical.md | 2 + .../server/initialization/ServerConfig.java | 9 ++ .../initialization/jetty/JettyBindings.java | 9 -- .../jetty/LimitRequestsFilter.java | 77 ++++++++++++ .../jetty/LimitRequestsFilterHolder.java | 115 ------------------ .../src/main/java/io/druid/cli/CliBroker.java | 13 -- .../cli/QueryJettyServerInitializer.java | 19 ++- 8 files changed, 107 insertions(+), 139 deletions(-) create mode 100644 server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java delete mode 100644 server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilterHolder.java diff --git a/docs/content/configuration/broker.md b/docs/content/configuration/broker.md index 6c46f9a02fb4..d98eb1a71bd6 100644 --- a/docs/content/configuration/broker.md +++ b/docs/content/configuration/broker.md @@ -38,7 +38,7 @@ Druid uses Jetty to serve HTTP requests. |`druid.server.http.numThreads`|Number of threads for HTTP requests.|max(10, (Number of cores * 17) / 16 + 2) + 30| |`druid.server.http.queueSize`|Size of the worker queue used by Jetty server to temporarily store incoming client connections. If this value is set and a request is rejected by jetty because queue is full then client would observe request failure with TCP connection being closed immediately with a completely empty response from server.|Unbounded| |`druid.server.http.maxIdleTime`|The Jetty max idle time for a connection.|PT5m| -|`druid.server.http.maxActiveRequests`|Number of active requests that can be handled by broker concurrently, beyond which "HTTP 429 Too Many Requests" error response is sent. Note that this must be set smaller than `druid.server.http.numThreads` to be effective or requests might just get queued at jetty server.|Not Applied By Default| +|`druid.server.http.maxActiveRequests`|Number of active requests that can be handled by this node concurrently, beyond which "HTTP 429 Too Many Requests" error response is sent. Note that this must be set smaller than `druid.server.http.numThreads` to be effective or requests might just get queued at jetty server.|Not Applied By Default| |`druid.server.http.defaultQueryTimeout`|Query timeout in millis, beyond which unfinished queries will be cancelled|300000| |`druid.server.http.maxScatterGatherBytes`|Maximum number of bytes gathered from data nodes such as historicals and realtime processes to execute a query. This is an advance configuration that allows to protect in case broker is under heavy load and not utilizing the data gathered in memory fast enough and leading to OOMs. This limit can be further reduced at query time using `maxScatterGatherBytes` in the context. Note that having large limit is not necessarily bad if broker is never under heavy concurrent load in which case data gathered is processed quickly and freeing up the memory used.|Long.MAX_VALUE| |`druid.broker.http.numConnections`|Size of connection pool for the Broker to connect to historical and real-time processes. If there are more queries than this number that all need to speak to the same node, then they will queue up.|20| diff --git a/docs/content/configuration/historical.md b/docs/content/configuration/historical.md index 321408e78e16..5bc6bafd3805 100644 --- a/docs/content/configuration/historical.md +++ b/docs/content/configuration/historical.md @@ -48,7 +48,9 @@ Druid uses Jetty to serve HTTP requests. |Property|Description|Default| |--------|-----------|-------| |`druid.server.http.numThreads`|Number of threads for HTTP requests.|max(10, (Number of cores * 17) / 16 + 2) + 30| +|`druid.server.http.queueSize`|Size of the worker queue used by Jetty server to temporarily store incoming client connections. If this value is set and a request is rejected by jetty because queue is full then client would observe request failure with TCP connection being closed immediately with a completely empty response from server.|Unbounded| |`druid.server.http.maxIdleTime`|The Jetty max idle time for a connection.|PT5m| +|`druid.server.http.maxActiveRequests`|Number of active requests that can be handled by this node concurrently, beyond which "HTTP 429 Too Many Requests" error response is sent. Note that this must be set smaller than `druid.server.http.numThreads` to be effective or requests might just get queued at jetty server.|Not Applied By Default| |`druid.server.http.defaultQueryTimeout`|Query timeout in millis, beyond which unfinished queries will be cancelled|300000| #### Processing diff --git a/server/src/main/java/io/druid/server/initialization/ServerConfig.java b/server/src/main/java/io/druid/server/initialization/ServerConfig.java index 307e95aa7b08..f96f68041dc4 100644 --- a/server/src/main/java/io/druid/server/initialization/ServerConfig.java +++ b/server/src/main/java/io/druid/server/initialization/ServerConfig.java @@ -37,6 +37,10 @@ public class ServerConfig @Min(1) private int queueSize = Integer.MAX_VALUE; + @JsonProperty + @Min(1) + private int maxActiveRequests = Integer.MAX_VALUE; + @JsonProperty @NotNull private Period maxIdleTime = new Period("PT5m"); @@ -65,6 +69,11 @@ public int getQueueSize() return queueSize; } + public int getMaxActiveRequests() + { + return maxActiveRequests; + } + public Period getMaxIdleTime() { return maxIdleTime; diff --git a/server/src/main/java/io/druid/server/initialization/jetty/JettyBindings.java b/server/src/main/java/io/druid/server/initialization/jetty/JettyBindings.java index 0bd8359303e1..e05b664e14c5 100644 --- a/server/src/main/java/io/druid/server/initialization/jetty/JettyBindings.java +++ b/server/src/main/java/io/druid/server/initialization/jetty/JettyBindings.java @@ -51,15 +51,6 @@ public static void addQosFilter(Binder binder, String path, int maxRequests) .toInstance(new QosFilterHolder(path, maxRequests)); } - public static void addLimitRequestsFilter(Binder binder, String path, int maxActiveRequests) - { - if (maxActiveRequests > 0) { - Multibinder.newSetBinder(binder, ServletFilterHolder.class) - .addBinding() - .toInstance(new LimitRequestsFilterHolder(path, maxActiveRequests)); - } - } - public static void addHandler(Binder binder, Class handlerClass) { Multibinder.newSetBinder(binder, Handler.class) diff --git a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java new file mode 100644 index 000000000000..4b0cda1193ad --- /dev/null +++ b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java @@ -0,0 +1,77 @@ +/* + * Licensed to Metamarkets Group Inc. (Metamarkets) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Metamarkets licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package io.druid.server.initialization.jetty; + +import com.google.common.base.Preconditions; + +import javax.servlet.Filter; +import javax.servlet.FilterChain; +import javax.servlet.FilterConfig; +import javax.servlet.ServletException; +import javax.servlet.ServletRequest; +import javax.servlet.ServletResponse; +import javax.servlet.http.HttpServletResponse; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; + +/** + */ +public class LimitRequestsFilter implements Filter +{ + private final int maxActiveRequests; + + private final AtomicInteger activeRequestsCount = new AtomicInteger(); + + public LimitRequestsFilter(int maxActiveRequests) + { + Preconditions.checkArgument( + maxActiveRequests > 0 && maxActiveRequests < Integer.MAX_VALUE, + "maxActiveRequests must be > 0 and < Integer.MAX_VALUE." + ); + this.maxActiveRequests = maxActiveRequests; + } + + @Override + public void init(FilterConfig filterConfig) throws ServletException + { + } + + @Override + public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) + throws IOException, ServletException + { + + int curr = activeRequestsCount.incrementAndGet(); + try { + if (curr <= maxActiveRequests) { + chain.doFilter(request, response); + } else { + // See https://tools.ietf.org/html/rfc6585 for status code 429 explanation. + ((HttpServletResponse) response).sendError(429, "Too Many Requests"); + } + } + finally { + activeRequestsCount.decrementAndGet(); + } + } + + @Override + public void destroy() { } +} diff --git a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilterHolder.java b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilterHolder.java deleted file mode 100644 index d2744b380b04..000000000000 --- a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilterHolder.java +++ /dev/null @@ -1,115 +0,0 @@ -/* - * Licensed to Metamarkets Group Inc. (Metamarkets) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. Metamarkets licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package io.druid.server.initialization.jetty; - -import com.google.common.collect.ImmutableMap; - -import javax.servlet.DispatcherType; -import javax.servlet.Filter; -import javax.servlet.FilterChain; -import javax.servlet.FilterConfig; -import javax.servlet.ServletException; -import javax.servlet.ServletRequest; -import javax.servlet.ServletResponse; -import javax.servlet.http.HttpServletResponse; -import java.io.IOException; -import java.util.EnumSet; -import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; - -public class LimitRequestsFilterHolder implements ServletFilterHolder -{ - private final String path; - private final int maxActiveRequests; - - public LimitRequestsFilterHolder(String path, int maxActiveRequests) - { - this.path = path; - this.maxActiveRequests = maxActiveRequests; - } - - @Override - public Filter getFilter() - { - return new LimitRequestsFilter(maxActiveRequests); - } - - @Override - public Class getFilterClass() - { - return null; - } - - @Override - public Map getInitParameters() - { - return ImmutableMap.of(); - } - - @Override - public String getPath() - { - return path; - } - - @Override - public EnumSet getDispatcherType() - { - return null; - } - - private static class LimitRequestsFilter implements Filter - { - private final int maxActiveRequests; - - private AtomicInteger activeRequestsCount = new AtomicInteger(); - - public LimitRequestsFilter(int maxActiveRequests) - { - this.maxActiveRequests = maxActiveRequests; - } - - @Override - public void init(FilterConfig filterConfig) throws ServletException - { - } - - @Override - public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) - throws IOException, ServletException - { - - int curr = activeRequestsCount.incrementAndGet(); - try { - if (curr <= maxActiveRequests) { - chain.doFilter(request, response); - } else { - // See https://tools.ietf.org/html/rfc6585 for status code 429 explanation. - ((HttpServletResponse)response).sendError(429, "Too Many Requests"); - } - } finally { - activeRequestsCount.decrementAndGet(); - } - } - - @Override - public void destroy() { } - } -} diff --git a/services/src/main/java/io/druid/cli/CliBroker.java b/services/src/main/java/io/druid/cli/CliBroker.java index e1dc20e03f9d..0745c4d02a0d 100644 --- a/services/src/main/java/io/druid/cli/CliBroker.java +++ b/services/src/main/java/io/druid/cli/CliBroker.java @@ -21,7 +21,6 @@ import com.google.common.collect.ImmutableList; import com.google.inject.Binder; -import com.google.inject.Inject; import com.google.inject.Module; import com.google.inject.name.Names; import io.airlift.airline.Command; @@ -51,7 +50,6 @@ import io.druid.server.ClientQuerySegmentWalker; import io.druid.server.coordination.broker.DruidBroker; import io.druid.server.http.BrokerResource; -import io.druid.server.initialization.jetty.JettyBindings; import io.druid.server.initialization.jetty.JettyServerInitializer; import io.druid.server.metrics.MetricsModule; import io.druid.server.metrics.QueryCountStatsProvider; @@ -60,7 +58,6 @@ import org.eclipse.jetty.server.Server; import java.util.List; -import java.util.Properties; /** */ @@ -72,11 +69,6 @@ public class CliBroker extends ServerRunnable { private static final Logger log = new Logger(CliBroker.class); - private static final String MAX_ACTIVE_REQUESTS_KEY = "druid.server.http.maxActiveRequests"; - - @Inject - private Properties properties; - public CliBroker() { super(log); @@ -117,11 +109,6 @@ public void configure(Binder binder) binder.bind(JettyServerInitializer.class).to(QueryJettyServerInitializer.class).in(LazySingleton.class); - if (properties.getProperty(MAX_ACTIVE_REQUESTS_KEY) != null) { - final int maxActiveRequests = Integer.parseInt(properties.getProperty(MAX_ACTIVE_REQUESTS_KEY)); - JettyBindings.addLimitRequestsFilter(binder, "/*", maxActiveRequests); - } - Jerseys.addResource(binder, BrokerQueryResource.class); binder.bind(QueryCountStatsProvider.class).to(BrokerQueryResource.class).in(LazySingleton.class); Jerseys.addResource(binder, BrokerResource.class); diff --git a/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java b/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java index d3d3f8b9fd43..a637490205ea 100644 --- a/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java +++ b/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java @@ -23,12 +23,16 @@ import com.google.inject.Inject; import com.google.inject.Injector; import com.google.inject.servlet.GuiceFilter; +import io.druid.java.util.common.logger.Logger; +import io.druid.server.initialization.ServerConfig; import io.druid.server.initialization.jetty.JettyServerInitUtils; import io.druid.server.initialization.jetty.JettyServerInitializer; +import io.druid.server.initialization.jetty.LimitRequestsFilter; import org.eclipse.jetty.server.Handler; import org.eclipse.jetty.server.Server; import org.eclipse.jetty.server.handler.HandlerList; import org.eclipse.jetty.servlet.DefaultServlet; +import org.eclipse.jetty.servlet.FilterHolder; import org.eclipse.jetty.servlet.ServletContextHandler; import org.eclipse.jetty.servlet.ServletHolder; @@ -39,12 +43,17 @@ */ public class QueryJettyServerInitializer implements JettyServerInitializer { + private static final Logger log = new Logger(QueryJettyServerInitializer.class); + private final List extensionHandlers; + private final ServerConfig serverConfig; + @Inject - public QueryJettyServerInitializer(Set extensionHandlers) + public QueryJettyServerInitializer(Set extensionHandlers, ServerConfig serverConfig) { this.extensionHandlers = ImmutableList.copyOf(extensionHandlers); + this.serverConfig = serverConfig; } @Override @@ -52,6 +61,14 @@ public void initialize(Server server, Injector injector) { final ServletContextHandler root = new ServletContextHandler(ServletContextHandler.SESSIONS); root.addServlet(new ServletHolder(new DefaultServlet()), "/*"); + + if (serverConfig.getMaxActiveRequests() < Integer.MAX_VALUE) { + log.info("Adding LimitRequestsFilter with maxActiveRequests [%d].", serverConfig.getMaxActiveRequests()); + root.addFilter(new FilterHolder(new LimitRequestsFilter(serverConfig.getMaxActiveRequests())), + "/*", null + ); + } + JettyServerInitUtils.addExtensionFilters(root, injector); root.addFilter(GuiceFilter.class, "/*", null); From 26e89db5c0756d19522ac91a1e7a026707bf23ca Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 5 Oct 2017 10:00:46 -0500 Subject: [PATCH 03/12] fix styling error --- .../server/initialization/jetty/LimitRequestsFilter.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java index 4b0cda1193ad..c7e453db3937 100644 --- a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java +++ b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java @@ -73,5 +73,8 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha } @Override - public void destroy() { } + public void destroy() + { + + } } From 12ece52f6b6782c6a73dcaeb4912a28bcafae72c Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 5 Oct 2017 10:02:41 -0500 Subject: [PATCH 04/12] fix style error --- .../druid/server/initialization/jetty/LimitRequestsFilter.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java index c7e453db3937..d6e30ba368ce 100644 --- a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java +++ b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java @@ -75,6 +75,6 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha @Override public void destroy() { - + } } From 7dedbd6f735a49be779a947a5b6ce0ef57b9a8d3 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 5 Oct 2017 13:04:02 -0500 Subject: [PATCH 05/12] have enableRequestLimit config instead of having user specify max limit --- docs/content/configuration/broker.md | 2 +- docs/content/configuration/historical.md | 2 +- .../server/initialization/ServerConfig.java | 29 +++++++++---------- .../jetty/JettyServerModule.java | 23 +++++++++------ .../cli/QueryJettyServerInitializer.java | 12 ++++++-- 5 files changed, 38 insertions(+), 30 deletions(-) diff --git a/docs/content/configuration/broker.md b/docs/content/configuration/broker.md index 0e785808c847..af97d2dffa50 100644 --- a/docs/content/configuration/broker.md +++ b/docs/content/configuration/broker.md @@ -38,7 +38,7 @@ Druid uses Jetty to serve HTTP requests. |`druid.server.http.numThreads`|Number of threads for HTTP requests.|max(10, (Number of cores * 17) / 16 + 2) + 30| |`druid.server.http.queueSize`|Size of the worker queue used by Jetty server to temporarily store incoming client connections. If this value is set and a request is rejected by jetty because queue is full then client would observe request failure with TCP connection being closed immediately with a completely empty response from server.|Unbounded| |`druid.server.http.maxIdleTime`|The Jetty max idle time for a connection.|PT5m| -|`druid.server.http.maxActiveRequests`|Number of active requests that can be handled by this node concurrently, beyond which "HTTP 429 Too Many Requests" error response is sent. Note that this must be set smaller than `druid.server.http.numThreads` to be effective or requests might just get queued at jetty server.|Not Applied By Default| +|`druid.server.http.enableRequestLimit`|If enabled, no requests would be queued in jetty queue and "HTTP 429 Too Many Requests" error response would be sent. |false| |`druid.server.http.defaultQueryTimeout`|Query timeout in millis, beyond which unfinished queries will be cancelled|300000| |`druid.server.http.maxScatterGatherBytes`|Maximum number of bytes gathered from data nodes such as historicals and realtime processes to execute a query. This is an advance configuration that allows to protect in case broker is under heavy load and not utilizing the data gathered in memory fast enough and leading to OOMs. This limit can be further reduced at query time using `maxScatterGatherBytes` in the context. Note that having large limit is not necessarily bad if broker is never under heavy concurrent load in which case data gathered is processed quickly and freeing up the memory used.|Long.MAX_VALUE| |`druid.broker.http.numConnections`|Size of connection pool for the Broker to connect to historical and real-time processes. If there are more queries than this number that all need to speak to the same node, then they will queue up.|20| diff --git a/docs/content/configuration/historical.md b/docs/content/configuration/historical.md index 21daa78c63c3..8ddb30c12590 100644 --- a/docs/content/configuration/historical.md +++ b/docs/content/configuration/historical.md @@ -49,7 +49,7 @@ Druid uses Jetty to serve HTTP requests. |`druid.server.http.numThreads`|Number of threads for HTTP requests.|max(10, (Number of cores * 17) / 16 + 2) + 30| |`druid.server.http.queueSize`|Size of the worker queue used by Jetty server to temporarily store incoming client connections. If this value is set and a request is rejected by jetty because queue is full then client would observe request failure with TCP connection being closed immediately with a completely empty response from server.|Unbounded| |`druid.server.http.maxIdleTime`|The Jetty max idle time for a connection.|PT5m| -|`druid.server.http.maxActiveRequests`|Number of active requests that can be handled by this node concurrently, beyond which "HTTP 429 Too Many Requests" error response is sent. Note that this must be set smaller than `druid.server.http.numThreads` to be effective or requests might just get queued at jetty server.|Not Applied By Default| +|`druid.server.http.enableRequestLimit`|If enabled, no requests would be queued in jetty queue and "HTTP 429 Too Many Requests" error response would be sent. |false| |`druid.server.http.defaultQueryTimeout`|Query timeout in millis, beyond which unfinished queries will be cancelled|300000| #### Processing diff --git a/server/src/main/java/io/druid/server/initialization/ServerConfig.java b/server/src/main/java/io/druid/server/initialization/ServerConfig.java index 3b263e70a818..d5db2cd2c530 100644 --- a/server/src/main/java/io/druid/server/initialization/ServerConfig.java +++ b/server/src/main/java/io/druid/server/initialization/ServerConfig.java @@ -39,8 +39,7 @@ public class ServerConfig private int queueSize = Integer.MAX_VALUE; @JsonProperty - @Min(1) - private int maxActiveRequests = Integer.MAX_VALUE; + private boolean enableRequestLimit = false; @JsonProperty @NotNull @@ -64,9 +63,9 @@ public int getQueueSize() return queueSize; } - public int getMaxActiveRequests() + public boolean isEnableRequestLimit() { - return maxActiveRequests; + return enableRequestLimit; } public Period getMaxIdleTime() @@ -95,6 +94,8 @@ public boolean equals(Object o) } ServerConfig that = (ServerConfig) o; return numThreads == that.numThreads && + queueSize == that.queueSize && + enableRequestLimit == that.enableRequestLimit && defaultQueryTimeout == that.defaultQueryTimeout && maxScatterGatherBytes == that.maxScatterGatherBytes && Objects.equals(maxIdleTime, that.maxIdleTime); @@ -103,17 +104,13 @@ public boolean equals(Object o) @Override public int hashCode() { - return Objects.hash(numThreads, maxIdleTime, defaultQueryTimeout, maxScatterGatherBytes); - } - - @Override - public String toString() - { - return "ServerConfig{" + - "numThreads=" + numThreads + - ", maxIdleTime=" + maxIdleTime + - ", defaultQueryTimeout=" + defaultQueryTimeout + - ", maxScatterGatherBytes=" + maxScatterGatherBytes + - '}'; + return Objects.hash( + numThreads, + queueSize, + enableRequestLimit, + maxIdleTime, + defaultQueryTimeout, + maxScatterGatherBytes + ); } } diff --git a/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java b/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java index c547808571fd..688a84bafb40 100644 --- a/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java +++ b/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java @@ -22,11 +22,9 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider; import com.fasterxml.jackson.jaxrs.smile.JacksonSmileProvider; -import com.google.common.collect.Iterables; import com.google.common.primitives.Ints; import com.google.inject.Binder; import com.google.inject.Binding; -import com.google.inject.ConfigurationException; import com.google.inject.Inject; import com.google.inject.Injector; import com.google.inject.Key; @@ -86,6 +84,9 @@ */ public class JettyServerModule extends JerseyServletModule { + // This is the maximum number of threads used by jetty acceptors and selectors + public static final int MAX_ACCEPTOR_SELECTOR_THREADS = 16; + private static final Logger log = new Logger(JettyServerModule.class); private static final AtomicInteger activeConnections = new AtomicInteger(); @@ -184,16 +185,20 @@ static Server makeJettyServer( Binding sslContextFactoryBinding ) { + // MAX_ACCEPTOR_SELECTOR_THREADS is added so that config.getNumThreads() has meaning, "number of threads + // that concurrently handle the requests". + int numServerThreads = config.getNumThreads() + MAX_ACCEPTOR_SELECTOR_THREADS; + final QueuedThreadPool threadPool; if (config.getQueueSize() == Integer.MAX_VALUE) { threadPool = new QueuedThreadPool(); - threadPool.setMinThreads(config.getNumThreads()); - threadPool.setMaxThreads(config.getNumThreads()); + threadPool.setMinThreads(numServerThreads); + threadPool.setMaxThreads(numServerThreads); } else { threadPool = new QueuedThreadPool( - config.getNumThreads(), - config.getNumThreads(), - 60000, + numServerThreads, + numServerThreads, + 60000, // same default is used in other case when threadPool = new QueuedThreadPool() new LinkedBlockingQueue<>(config.getQueueSize()) ); } @@ -273,8 +278,8 @@ static void initializeServer(Injector injector, Lifecycle lifecycle, final Serve try { initializer.initialize(server, injector); } - catch (ConfigurationException e) { - throw new ProvisionException(Iterables.getFirst(e.getErrorMessages(), null).getMessage()); + catch (Exception e) { + throw new ProvisionException(e.getMessage()); } lifecycle.addHandler( diff --git a/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java b/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java index 2cd2a9365555..7bedb403d213 100644 --- a/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java +++ b/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java @@ -20,6 +20,7 @@ package io.druid.cli; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableList; import com.google.inject.Inject; import com.google.inject.Injector; @@ -68,9 +69,14 @@ public void initialize(Server server, Injector injector) final ServletContextHandler root = new ServletContextHandler(ServletContextHandler.SESSIONS); root.addServlet(new ServletHolder(new DefaultServlet()), "/*"); - if (serverConfig.getMaxActiveRequests() < Integer.MAX_VALUE) { - log.info("Adding LimitRequestsFilter with maxActiveRequests [%d].", serverConfig.getMaxActiveRequests()); - root.addFilter(new FilterHolder(new LimitRequestsFilter(serverConfig.getMaxActiveRequests())), + // Add LimitRequestsFilter as first in the chain if enabled. + if (serverConfig.isEnableRequestLimit()) { + Preconditions.checkArgument( + serverConfig.getNumThreads() > 1, + "numThreads must be > 1 to enable Request Limit Filter." + ); + log.info("Enabling Request Limit Filter with limit [%d].", serverConfig.getNumThreads()-1); + root.addFilter(new FilterHolder(new LimitRequestsFilter(serverConfig.getNumThreads()-1)), "/*", null ); } From f0509d0ce47a1d952768700e3b2a7b64df83cc02 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 5 Oct 2017 14:22:26 -0500 Subject: [PATCH 06/12] add comment --- .../src/main/java/io/druid/cli/QueryJettyServerInitializer.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java b/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java index 7bedb403d213..99beed7da455 100644 --- a/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java +++ b/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java @@ -71,6 +71,8 @@ public void initialize(Server server, Injector injector) // Add LimitRequestsFilter as first in the chain if enabled. if (serverConfig.isEnableRequestLimit()) { + //To reject xth request, limit should be set to x-1 because (x+1)st request wouldn't reach filter + // but rather wait on jetty queue. Preconditions.checkArgument( serverConfig.getNumThreads() > 1, "numThreads must be > 1 to enable Request Limit Filter." From ea6dbe951e5c9d52f87174077bf996294c583a87 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 5 Oct 2017 15:20:48 -0500 Subject: [PATCH 07/12] fix style error --- .../main/java/io/druid/cli/QueryJettyServerInitializer.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java b/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java index 99beed7da455..e3988a29cae0 100644 --- a/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java +++ b/services/src/main/java/io/druid/cli/QueryJettyServerInitializer.java @@ -77,8 +77,8 @@ public void initialize(Server server, Injector injector) serverConfig.getNumThreads() > 1, "numThreads must be > 1 to enable Request Limit Filter." ); - log.info("Enabling Request Limit Filter with limit [%d].", serverConfig.getNumThreads()-1); - root.addFilter(new FilterHolder(new LimitRequestsFilter(serverConfig.getNumThreads()-1)), + log.info("Enabling Request Limit Filter with limit [%d].", serverConfig.getNumThreads() - 1); + root.addFilter(new FilterHolder(new LimitRequestsFilter(serverConfig.getNumThreads() - 1)), "/*", null ); } From ff7139175db3845d1344d9ac6cff24a2157aa09b Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 5 Oct 2017 15:52:16 -0500 Subject: [PATCH 08/12] add UT fo LimitRequestsFilter --- .../jetty/LimitRequestsFilterTest.java | 137 ++++++++++++++++++ 1 file changed, 137 insertions(+) create mode 100644 server/src/test/java/io/druid/server/initialization/jetty/LimitRequestsFilterTest.java diff --git a/server/src/test/java/io/druid/server/initialization/jetty/LimitRequestsFilterTest.java b/server/src/test/java/io/druid/server/initialization/jetty/LimitRequestsFilterTest.java new file mode 100644 index 000000000000..5c6344004256 --- /dev/null +++ b/server/src/test/java/io/druid/server/initialization/jetty/LimitRequestsFilterTest.java @@ -0,0 +1,137 @@ +/* + * Licensed to Metamarkets Group Inc. (Metamarkets) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. Metamarkets licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package io.druid.server.initialization.jetty; + +import io.druid.java.util.common.ISE; +import org.easymock.EasyMock; +import org.junit.Test; + +import javax.servlet.FilterChain; +import javax.servlet.ServletException; +import javax.servlet.ServletRequest; +import javax.servlet.ServletResponse; +import javax.servlet.http.HttpServletResponse; +import java.io.IOException; +import java.util.concurrent.CountDownLatch; + +/** + */ +public class LimitRequestsFilterTest +{ + @Test + public void testSimple() throws Exception + { + LimitRequestsFilter filter = new LimitRequestsFilter(2); + + CountDownLatch latch1 = createAndStartRequestThread( + filter, + EasyMock.createStrictMock(ServletRequest.class), + EasyMock.createStrictMock(HttpServletResponse.class) + ); + + CountDownLatch latch2 = createAndStartRequestThread( + filter, + EasyMock.createStrictMock(ServletRequest.class), + EasyMock.createStrictMock(HttpServletResponse.class) + ); + + //now further requests should fail + HttpServletResponse resp = EasyMock.createMock(HttpServletResponse.class); + resp.sendError(429, "Too Many Requests"); + EasyMock.expectLastCall().times(2); + EasyMock.replay(resp); + + filter.doFilter( + EasyMock.createStrictMock(ServletRequest.class), + resp, + EasyMock.createStrictMock(FilterChain.class) + ); + + filter.doFilter( + EasyMock.createStrictMock(ServletRequest.class), + resp, + EasyMock.createStrictMock(FilterChain.class) + ); + + EasyMock.verify(resp); + + //release one of the pending requests + latch1.countDown(); + + //now requests should go through + FilterChain chain = EasyMock.createMock(FilterChain.class); + chain.doFilter(EasyMock.anyObject(), EasyMock.anyObject()); + EasyMock.expectLastCall().times(2); + EasyMock.replay(chain); + + filter.doFilter( + EasyMock.createStrictMock(ServletRequest.class), + EasyMock.createStrictMock(HttpServletResponse.class), + chain + ); + + filter.doFilter( + EasyMock.createStrictMock(ServletRequest.class), + EasyMock.createStrictMock(HttpServletResponse.class), + chain + ); + + EasyMock.verify(chain); + + latch2.countDown(); + } + + private CountDownLatch createAndStartRequestThread(LimitRequestsFilter filter, ServletRequest req, HttpServletResponse resp) + { + CountDownLatch latch = new CountDownLatch(1); + new Thread( + () -> { + try { + filter.doFilter(req, resp, new TestFilterChain(latch)); + } + catch (Exception e) { + throw new ISE(e, "exception"); + } + } + ).start(); + return latch; + } + + private static class TestFilterChain implements FilterChain + { + private final CountDownLatch latch; + + TestFilterChain(CountDownLatch latch) + { + this.latch = latch; + } + + @Override + public void doFilter(ServletRequest request, ServletResponse response) throws IOException, ServletException + { + try { + latch.await(); + } + catch (InterruptedException ex) { + throw new ISE(ex, "exception"); + } + } + } +} From 1d16220414039ac43262c3042563607677a0f835 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 5 Oct 2017 16:17:47 -0500 Subject: [PATCH 09/12] address review comments --- .../jetty/JettyServerModule.java | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java b/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java index 688a84bafb40..6c5b7649e1c3 100644 --- a/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java +++ b/server/src/main/java/io/druid/server/initialization/jetty/JettyServerModule.java @@ -29,7 +29,6 @@ import com.google.inject.Injector; import com.google.inject.Key; import com.google.inject.Provides; -import com.google.inject.ProvisionException; import com.google.inject.Scopes; import com.google.inject.Singleton; import com.google.inject.multibindings.Multibinder; @@ -49,6 +48,7 @@ import io.druid.guice.annotations.Json; import io.druid.guice.annotations.Self; import io.druid.guice.annotations.Smile; +import io.druid.java.util.common.RE; import io.druid.java.util.common.lifecycle.Lifecycle; import io.druid.java.util.common.logger.Logger; import io.druid.server.DruidNode; @@ -84,9 +84,6 @@ */ public class JettyServerModule extends JerseyServletModule { - // This is the maximum number of threads used by jetty acceptors and selectors - public static final int MAX_ACCEPTOR_SELECTOR_THREADS = 16; - private static final Logger log = new Logger(JettyServerModule.class); private static final AtomicInteger activeConnections = new AtomicInteger(); @@ -185,9 +182,9 @@ static Server makeJettyServer( Binding sslContextFactoryBinding ) { - // MAX_ACCEPTOR_SELECTOR_THREADS is added so that config.getNumThreads() has meaning, "number of threads + // adjusting to make config.getNumThreads() mean, "number of threads // that concurrently handle the requests". - int numServerThreads = config.getNumThreads() + MAX_ACCEPTOR_SELECTOR_THREADS; + int numServerThreads = config.getNumThreads() + getMaxJettyAcceptorsSelectorsNum(node); final QueuedThreadPool threadPool; if (config.getQueueSize() == Integer.MAX_VALUE) { @@ -279,7 +276,7 @@ static void initializeServer(Injector injector, Lifecycle lifecycle, final Serve initializer.initialize(server, injector); } catch (Exception e) { - throw new ProvisionException(e.getMessage()); + throw new RE(e, "server initialization exception"); } lifecycle.addHandler( @@ -305,6 +302,14 @@ public void stop() ); } + private static int getMaxJettyAcceptorsSelectorsNum(DruidNode druidNode) + { + // This computation is based on Jetty v9.3.19 which uses upto 8(4 acceptors and 4 selectors) threads per + // ServerConnector + int numServerConnector = (druidNode.isEnablePlaintextPort() ? 1 : 0) + (druidNode.isEnableTlsPort() ? 1 : 0); + return numServerConnector * 8; + } + @Provides @Singleton public JettyMonitor getJettyMonitor( From 57298e97f6f2aee21ab1a55163f8cc05cc0c32ab Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Thu, 5 Oct 2017 19:14:43 -0500 Subject: [PATCH 10/12] fix test --- .../test/java/io/druid/server/initialization/JettyQosTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/src/test/java/io/druid/server/initialization/JettyQosTest.java b/server/src/test/java/io/druid/server/initialization/JettyQosTest.java index 3bdb417ad644..b23489845c3e 100644 --- a/server/src/test/java/io/druid/server/initialization/JettyQosTest.java +++ b/server/src/test/java/io/druid/server/initialization/JettyQosTest.java @@ -96,7 +96,7 @@ public void testNumThreads() { // Just make sure the injector stuff for this test is actually working. Assert.assertEquals( - 10, + 18, ((QueuedThreadPool) server.getThreadPool()).getMaxThreads() ); } From c24360739487936a83cd4e4d40328e6d1e565533 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Fri, 6 Oct 2017 09:55:03 -0500 Subject: [PATCH 11/12] make LimitRequestsFilterTest more robust --- .../initialization/jetty/LimitRequestsFilter.java | 5 +++++ .../jetty/LimitRequestsFilterTest.java | 14 +++++++++++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java index d6e30ba368ce..8dfe499fb6b8 100644 --- a/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java +++ b/server/src/main/java/io/druid/server/initialization/jetty/LimitRequestsFilter.java @@ -77,4 +77,9 @@ public void destroy() { } + + public int getActiveRequestsCount() + { + return activeRequestsCount.get(); + } } diff --git a/server/src/test/java/io/druid/server/initialization/jetty/LimitRequestsFilterTest.java b/server/src/test/java/io/druid/server/initialization/jetty/LimitRequestsFilterTest.java index 5c6344004256..87358070c910 100644 --- a/server/src/test/java/io/druid/server/initialization/jetty/LimitRequestsFilterTest.java +++ b/server/src/test/java/io/druid/server/initialization/jetty/LimitRequestsFilterTest.java @@ -35,7 +35,7 @@ */ public class LimitRequestsFilterTest { - @Test + @Test(timeout = 5000L) public void testSimple() throws Exception { LimitRequestsFilter filter = new LimitRequestsFilter(2); @@ -52,6 +52,10 @@ public void testSimple() throws Exception EasyMock.createStrictMock(HttpServletResponse.class) ); + while (filter.getActiveRequestsCount() != 2) { + Thread.sleep(100); + } + //now further requests should fail HttpServletResponse resp = EasyMock.createMock(HttpServletResponse.class); resp.sendError(429, "Too Many Requests"); @@ -75,6 +79,10 @@ public void testSimple() throws Exception //release one of the pending requests latch1.countDown(); + while (filter.getActiveRequestsCount() != 1) { + Thread.sleep(100); + } + //now requests should go through FilterChain chain = EasyMock.createMock(FilterChain.class); chain.doFilter(EasyMock.anyObject(), EasyMock.anyObject()); @@ -96,6 +104,10 @@ public void testSimple() throws Exception EasyMock.verify(chain); latch2.countDown(); + + while (filter.getActiveRequestsCount() != 0) { + Thread.sleep(100); + } } private CountDownLatch createAndStartRequestThread(LimitRequestsFilter filter, ServletRequest req, HttpServletResponse resp) From 253458634e12ad6739274c905d90ca29d85acd0d Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Fri, 6 Oct 2017 12:25:18 -0500 Subject: [PATCH 12/12] fix JettyQosTest --- .../java/io/druid/server/initialization/JettyQosTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/server/src/test/java/io/druid/server/initialization/JettyQosTest.java b/server/src/test/java/io/druid/server/initialization/JettyQosTest.java index b23489845c3e..ffb67b6a84e7 100644 --- a/server/src/test/java/io/druid/server/initialization/JettyQosTest.java +++ b/server/src/test/java/io/druid/server/initialization/JettyQosTest.java @@ -80,7 +80,7 @@ public void configure(Binder binder) binder.bind(AuthorizerMapper.class).toInstance(AuthTestUtils.TEST_AUTHORIZER_MAPPER); JettyBindings.addQosFilter(binder, "/slow/*", 2); final ServerConfig serverConfig = new ObjectMapper().convertValue( - ImmutableMap.of("numThreads", "10"), + ImmutableMap.of("numThreads", "2"), ServerConfig.class ); binder.bind(ServerConfig.class).toInstance(serverConfig); @@ -96,7 +96,7 @@ public void testNumThreads() { // Just make sure the injector stuff for this test is actually working. Assert.assertEquals( - 18, + 10, ((QueuedThreadPool) server.getThreadPool()).getMaxThreads() ); }