Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

allow rate-limiting on data nodes (for shards.tolerant=true) #239

Open
wants to merge 1 commit into
base: fs/branch_9_3
Choose a base branch
from
Open

allow rate-limiting on data nodes (for shards.tolerant=true) #239

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions solr/core/src/java/org/apache/solr/handler/component/HttpShardHandlerFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.impl.Http2SolrClient;
import org.apache.solr.client.solrj.impl.HttpClientUtil;
import org.apache.solr.client.solrj.impl.LBHttp2SolrClient;
Expand Down Expand Up @@ -285,6 +286,7 @@ public void init(PluginInfo info) {
.withIdleTimeout(soTimeout, TimeUnit.MILLISECONDS)
.withExecutor(commExecutor)
.withMaxConnectionsPerHost(maxConnectionsPerHost)
.withContext(SolrRequest.SolrClientContext.SERVER)
.build();
this.defaultClient.addListenerFactory(this.httpListenerFactory);
this.loadbalancer = new LBHttp2SolrClient.Builder(defaultClient).build();
Expand Down
14 changes: 10 additions & 4 deletions solr/core/src/java/org/apache/solr/servlet/RateLimitManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,10 +100,16 @@ public RequestRateLimiter.SlotReservation handleRequest(HttpServletRequest reque
}

// Do not throttle internal requests
if (requestContext != null
&& requestContext.equals(SolrRequest.SolrClientContext.SERVER.toString())) {
return RequestRateLimiter.UNLIMITED;
}
// TODO: the block below is disabled temporarily to evaluate datanode-level throttling,
// which is where the resources actually are. It should be re-enabled and fixed upstream
// to support datanode-level throttling in a more nuanced way. But for FS usecase, most
// requests will be `shards.tolerant=true`, and we shouldn't hit throttling unless there
// are real problems on a node, in which case we're probably better off rejecting the
// requests anyway.
// if (requestContext != null
// && requestContext.equals(SolrRequest.SolrClientContext.SERVER.toString())) {
// return RequestRateLimiter.UNLIMITED;
// }

RequestRateLimiter requestRateLimiter = requestRateLimiterMap.get(typeOfRequest);

Expand Down
94 changes: 65 additions & 29 deletions solr/core/src/test/org/apache/solr/servlet/TestRequestRateLimiter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
package org.apache.solr.servlet;

import static org.apache.solr.servlet.RateLimitManager.DEFAULT_SLOT_ACQUISITION_TIMEOUT_MS;
import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.containsStringIgnoringCase;
import static org.hamcrest.CoreMatchers.instanceOf;

import java.io.Closeable;
Expand All @@ -35,6 +35,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrRequest;
Expand All @@ -48,6 +49,7 @@
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.core.RateLimiterConfig;
import org.apache.solr.embedded.JettySolrRunner;
import org.hamcrest.MatcherAssert;
import org.junit.BeforeClass;
import org.junit.Test;
Expand All @@ -58,18 +60,21 @@ public class TestRequestRateLimiter extends SolrCloudTestCase {

@BeforeClass
public static void setupCluster() throws Exception {
configureCluster(1).addConfig(FIRST_COLLECTION, configset("cloud-minimal")).configure();
configureCluster(2).addConfig(FIRST_COLLECTION, configset("cloud-minimal")).configure();
}

@Test
public void testConcurrentQueries() throws Exception {
try (CloudSolrClient client =
cluster.basicSolrClientBuilder().withDefaultCollection(FIRST_COLLECTION).build()) {

CollectionAdminRequest.createCollection(FIRST_COLLECTION, 1, 1).process(client);
cluster.waitForActiveCollection(FIRST_COLLECTION, 1, 1);
CollectionAdminRequest.createCollection(FIRST_COLLECTION, 2, 1).process(client);
cluster.waitForActiveCollection(FIRST_COLLECTION, 2, 2);

SolrDispatchFilter solrDispatchFilter = cluster.getJettySolrRunner(0).getSolrDispatchFilter();
List<SolrDispatchFilter> solrDispatchFilters =
cluster.getJettySolrRunners().stream()
.map(JettySolrRunner::getSolrDispatchFilter)
.collect(Collectors.toList());

RateLimiterConfig rateLimiterConfig =
new RateLimiterConfig(
Expand All @@ -79,34 +84,50 @@ public void testConcurrentQueries() throws Exception {
DEFAULT_SLOT_ACQUISITION_TIMEOUT_MS,
5 /* allowedRequests */,
true /* isSlotBorrowing */);
// We are fine with a null FilterConfig here since we ensure that MockBuilder never invokes
// its parent here
RateLimitManager.Builder builder =
new MockBuilder(
null /* dummy SolrZkClient */, new MockRequestRateLimiter(rateLimiterConfig));
RateLimitManager rateLimitManager = builder.build();

solrDispatchFilter.replaceRateLimitManager(rateLimitManager);
List<RateLimitManager> rateLimitManagers = new ArrayList<>(solrDispatchFilters.size());

solrDispatchFilters.forEach(
(f) -> {
// We are fine with a null FilterConfig here since we ensure that MockBuilder never
// invokes
// its parent here
RateLimitManager.Builder builder =
new MockBuilder(
null /* dummy SolrZkClient */, new MockRequestRateLimiter(rateLimiterConfig));
RateLimitManager rateLimitManager = builder.build();
rateLimitManagers.add(rateLimitManager);
f.replaceRateLimitManager(rateLimitManager);
});

int numDocs = TEST_NIGHTLY ? 10000 : 100;

processTest(client, numDocs, 350 /* number of queries */);

MockRequestRateLimiter mockQueryRateLimiter =
(MockRequestRateLimiter)
rateLimitManager.getRequestRateLimiter(SolrRequest.SolrRequestType.QUERY);
List<MockRequestRateLimiter> mockQueryRateLimiters =
rateLimitManagers.stream()
.map(
(m) ->
(MockRequestRateLimiter)
m.getRequestRateLimiter(SolrRequest.SolrRequestType.QUERY))
.collect(Collectors.toList());

assertEquals(350, mockQueryRateLimiter.incomingRequestCount.get());
assertEquals(
350, mockQueryRateLimiters.stream().mapToInt((m) -> m.incomingRequestCount.get()).sum());

assertTrue(mockQueryRateLimiter.acceptedNewRequestCount.get() > 0);
assertTrue(
(mockQueryRateLimiter.acceptedNewRequestCount.get()
== mockQueryRateLimiter.incomingRequestCount.get()
|| mockQueryRateLimiter.rejectedRequestCount.get() > 0));
mockQueryRateLimiters.stream().mapToInt((m) -> m.acceptedNewRequestCount.get()).sum()
> 0);
assertTrue(
(mockQueryRateLimiters.stream().mapToInt((m) -> m.acceptedNewRequestCount.get()).sum()
== mockQueryRateLimiters.stream()
.mapToInt((m) -> m.incomingRequestCount.get())
.sum()
|| mockQueryRateLimiters.stream().mapToInt((m) -> m.rejectedRequestCount.get()).sum()
> 0));
assertEquals(
mockQueryRateLimiter.incomingRequestCount.get(),
mockQueryRateLimiter.acceptedNewRequestCount.get()
+ mockQueryRateLimiter.rejectedRequestCount.get());
mockQueryRateLimiters.stream().mapToInt((m) -> m.incomingRequestCount.get()).sum(),
mockQueryRateLimiters.stream().mapToInt((m) -> m.acceptedNewRequestCount.get()).sum()
+ mockQueryRateLimiters.stream().mapToInt((m) -> m.rejectedRequestCount.get()).sum());
}
}

Expand Down Expand Up @@ -183,9 +204,24 @@ private void processTest(SolrClient client, int numDocuments, int numQueries) th
callableList.add(
() -> {
try {
QueryResponse response = client.query(new SolrQuery("*:*"));

assertEquals(numDocuments, response.getResults().getNumFound());
// TODO: tolerant, when `true`, causes the Solr-Request-Type header to be added to
// shard requests. setting this to `true` shows proper behavior (quick-and-dirty),
// but there are some test assumptions that are invalidated by this. For now we'll
// proceed with this as always `false` (so that tests pass), but we need to circle
// back and do this properly.
boolean tolerant = false;
QueryResponse response =
client.query(
new SolrQuery("q", "*:*", "shards.tolerant", Boolean.toString(tolerant)));

try {
assertEquals(numDocuments, response.getResults().getNumFound());
} catch (AssertionError er) {
if (!tolerant
|| response.getResponseHeader().get("partialResults") != Boolean.TRUE) {
throw er;
}
}
} catch (Exception e) {
throw new RuntimeException(e.getMessage(), e);
}
Expand All @@ -200,11 +236,11 @@ private void processTest(SolrClient client, int numDocuments, int numQueries) th
try {
assertNotNull(future.get());
} catch (ExecutionException e) {
MatcherAssert.assertThat(e.getCause().getCause(), instanceOf(RemoteSolrException.class));
MatcherAssert.assertThat(getRootCause(e), instanceOf(RemoteSolrException.class));
RemoteSolrException rse = (RemoteSolrException) e.getCause().getCause();
assertEquals(SolrException.ErrorCode.TOO_MANY_REQUESTS.code, rse.code());
MatcherAssert.assertThat(
rse.getMessage(), containsString("non ok status: 429, message:Too Many Requests"));
rse.getMessage(), containsStringIgnoringCase("Too Many Requests"));
}
}
} finally {
Expand Down
27 changes: 26 additions & 1 deletion solr/solrj/src/java/org/apache/solr/client/solrj/impl/Http2SolrClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
import org.apache.solr.common.SolrException;
import org.apache.solr.common.params.CommonParams;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.ShardParams;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.params.UpdateParams;
import org.apache.solr.common.util.ContentStream;
Expand Down Expand Up @@ -586,6 +587,17 @@ private Request makeRequest(SolrRequest<?> solrRequest, String collection)
}

private void decorateRequest(Request req, SolrRequest<?> solrRequest) {
SolrRequest.SolrClientContext context = getContext();
req.header(CommonParams.SOLR_REQUEST_CONTEXT_PARAM, context.toString());
if (context == SolrRequest.SolrClientContext.CLIENT
|| solrRequest.getParams().getBool(ShardParams.SHARDS_TOLERANT, false)) {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Whole purpose of ratelimiter to limit the resources on all nodes. Not sure we really need to make any special case here.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The reason for differentiating between shards.tolarant=true vs. shards.tolerant=false is explained in the comment immediately below:

      // NOTE: if `shards.tolerant=false`, do _not_ set the `Solr-Request-Type` header, because we
      // could end up doing a lot of extra work at the cluster level, retrying requests that may
      // only have failed to obtain a ratelimit permit on a single shard.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For our case in particular, practically we actually do want to avoid failing any requests that are shards.tolerant=false. Notably these are also the most likely requests to be retried on failure, so if we end up repeatedly executing requests on all nodes, only to repeatedly fail because of 1 struggling node (for example), that could easily cause load to increase on the other nodes to the point where the problem spreads to the entire cluster.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess these are two different problems. Rate limiting should be agnostic to any parameters.
if we see any issue with shards.tolerant=false or single node, then we need to track that in that context. We don't want to increase the load on other node same time.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure what you're suggesting to do here. If we see an issue with shards.tolerant=false on a single node, we will already have increased the load on other nodes (requests to nodes are issued in parallel), and if the top-level request fails due to the one node rate-limiting, then the client is likely to retry, increasing the load on the cluster overall (the exact situation that we both agree we want to avoid).

Rate limiting should be agnostic to any parameters

Why do you say this? I think the status quo (evaluate rate limiting only on the coordinator node) is due to the potential for rate-limiting evaluation on data-nodes to amplify request load. So if we really want rate limiting to be agnostic to any parameters, then I think rate-limiting on data nodes must be avoided entirely (due to the request amplification issue).

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's chat on this.

// automatically set requestType on top-level requests (CLIENT), or if `shards.tolerant=true`.
// NOTE: if `shards.tolerant=false`, do _not_ set the `Solr-Request-Type` header, because we
// could end up doing a lot of extra work at the cluster level, retrying requests that may
// only
// have failed to obtain a ratelimit permit on a single shard.
req.header(CommonParams.SOLR_REQUEST_TYPE_PARAM, solrRequest.getRequestType());
}
req.header(HttpHeader.ACCEPT_ENCODING, null);
if (requestTimeoutMillis > 0) {
req.timeout(requestTimeoutMillis, TimeUnit.MILLISECONDS);
Expand Down Expand Up @@ -1034,6 +1046,7 @@ public static class Builder {
private Long connectionTimeoutMillis;
private Long requestTimeoutMillis;
private Integer maxConnectionsPerHost;
private SolrRequest.SolrClientContext context = SolrRequest.SolrClientContext.CLIENT;
private String basicAuthAuthorizationStr;
private boolean useHttp1_1 = Boolean.getBoolean("solr.http1");
private Boolean followRedirects;
Expand All @@ -1051,7 +1064,14 @@ public Builder(String baseSolrUrl) {
}

public Http2SolrClient build() {
Http2SolrClient client = new Http2SolrClient(baseSolrUrl, this);
final SolrRequest.SolrClientContext context = this.context;
Http2SolrClient client =
new Http2SolrClient(baseSolrUrl, this) {
@Override
public SolrRequest.SolrClientContext getContext() {
return context;
}
};
try {
httpClientBuilderSetup(client);
} catch (RuntimeException e) {
Expand Down Expand Up @@ -1199,6 +1219,11 @@ public Builder withMaxConnectionsPerHost(int max) {
return this;
}

public Builder withContext(SolrRequest.SolrClientContext context) {
this.context = context;
return this;
}

/**
* @deprecated Please use {@link #withIdleTimeout(long, TimeUnit)}
*/
Expand Down
6 changes: 6 additions & 0 deletions solr/solrj/src/java/org/apache/solr/client/solrj/impl/LBHttp2SolrClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.client.solrj.ResponseParser;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.request.IsUpdateRequest;
import org.apache.solr.client.solrj.request.RequestWriter;
Expand Down Expand Up @@ -104,6 +105,11 @@ private LBHttp2SolrClient(Http2SolrClient solrClient, List<String> baseSolrUrls)
this.solrClient = solrClient;
}

@Override
public SolrRequest.SolrClientContext getContext() {
return solrClient.getContext();
}

@Override
protected SolrClient getClient(String baseUrl) {
return solrClient;
Expand Down
Loading