Skip to content

Commit

Permalink
Isolate Request in Call-Chain for REST Request Handling (#45130)
Browse files Browse the repository at this point in the history 
* Follow up to #44949
* Stop using a special code path for multi-line JSON and instead handle its detection like that of other XContent types when creating the request
* Only leave a single path that holds a reference to the full REST request
   * In the next step we can move the copying of request content to happen before the actual request handling and make it conditional on the handler in question to stop copying bulk requests as suggested in #44564
  • Loading branch information
@original-brownbear
original-brownbear authored Aug 10, 2019
1 parent 788fece commit 39142db
Show file tree
Hide file tree
Showing 5 changed files with 82 additions and 102 deletions.
4 changes: 4 additions & 0 deletions libs/x-content/src/main/java/org/elasticsearch/common/xcontent/XContentType.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -149,6 +149,10 @@ public static XContentType fromMediaType(String mediaType) {
return type;
}
}
// we also support newline delimited JSON: http://specs.okfnlabs.org/ndjson/
if (lowercaseMediaType.toLowerCase(Locale.ROOT).equals("application/x-ndjson")) {
return XContentType.JSON;
}

return null;
}
Expand Down
165 changes: 73 additions & 92 deletions server/src/main/java/org/elasticsearch/rest/RestController.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,15 +42,14 @@
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
import java.util.function.UnaryOperator;

import static org.elasticsearch.rest.BytesRestResponse.TEXT_CONTENT_TYPE;
Expand Down Expand Up @@ -200,75 +199,53 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
/**
* Dispatch the request, if possible, returning true if a response was sent or false otherwise.
*/
boolean dispatchRequest(final RestRequest request, final RestChannel channel, final NodeClient client,
@Nullable final RestHandler handler) throws Exception {
if (handler == null) {
// Get the map of matching handlers for a request, for the full set of HTTP methods.
final Set<RestRequest.Method> validMethodSet = getValidHandlerMethodSet(request);
final RestRequest.Method method = request.method();
if (validMethodSet.contains(method) == false) {
if (method == RestRequest.Method.OPTIONS) {
handleOptionsRequest(channel, validMethodSet);
return true;
}
if (validMethodSet.isEmpty() == false) {
// If an alternative handler for an explicit path is registered to a
// different HTTP method than the one supplied - return a 405 Method
// Not Allowed error.
handleUnsupportedHttpMethod(request.uri(), method, channel, validMethodSet, null);
return true;
}
private boolean dispatchRequest(RestRequest request, RestChannel channel, RestHandler handler) throws Exception {
final int contentLength = request.contentLength();
if (contentLength > 0) {
final XContentType xContentType = request.getXContentType();
if (xContentType == null) {
sendContentTypeErrorMessage(request.getAllHeaderValues("Content-Type"), channel);
return true;
}
return false;
} else {
final int contentLength = request.contentLength();
if (contentLength > 0) {
if (hasContentType(request, handler) == false) {
sendContentTypeErrorMessage(request.getAllHeaderValues("Content-Type"), channel);
return true;
}
final XContentType xContentType = request.getXContentType();
if (handler.supportsContentStream() && xContentType != XContentType.JSON && xContentType != XContentType.SMILE) {
channel.sendResponse(BytesRestResponse.createSimpleErrorResponse(channel, RestStatus.NOT_ACCEPTABLE,
"Content-Type [" + xContentType + "] does not support stream parsing. Use JSON or SMILE instead"));
return true;
}
if (handler.supportsContentStream() && xContentType != XContentType.JSON && xContentType != XContentType.SMILE) {
channel.sendResponse(BytesRestResponse.createSimpleErrorResponse(channel, RestStatus.NOT_ACCEPTABLE,
"Content-Type [" + xContentType + "] does not support stream parsing. Use JSON or SMILE instead"));
return true;
}
RestChannel responseChannel = channel;
try {
if (handler.canTripCircuitBreaker()) {
inFlightRequestsBreaker(circuitBreakerService).addEstimateBytesAndMaybeBreak(contentLength, "<http_request>");
} else {
inFlightRequestsBreaker(circuitBreakerService).addWithoutBreaking(contentLength);
}
// iff we could reserve bytes for the request we need to send the response also over this channel
responseChannel = new ResourceHandlingHttpChannel(channel, circuitBreakerService, contentLength);
handler.handleRequest(request, responseChannel, client);
} catch (Exception e) {
responseChannel.sendResponse(new BytesRestResponse(responseChannel, e));
}
RestChannel responseChannel = channel;
try {
if (handler.canTripCircuitBreaker()) {
inFlightRequestsBreaker(circuitBreakerService).addEstimateBytesAndMaybeBreak(contentLength, "<http_request>");
} else {
inFlightRequestsBreaker(circuitBreakerService).addWithoutBreaking(contentLength);
}
return true;
// iff we could reserve bytes for the request we need to send the response also over this channel
responseChannel = new ResourceHandlingHttpChannel(channel, circuitBreakerService, contentLength);
handler.handleRequest(request, responseChannel, client);
} catch (Exception e) {
responseChannel.sendResponse(new BytesRestResponse(responseChannel, e));
}
return true;
}

/**
* If a request contains content, this method will return {@code true} if the {@code Content-Type} header is present, matches an
* {@link XContentType} or the handler supports a content stream and the content type header is for newline delimited JSON,
*/
private static boolean hasContentType(final RestRequest restRequest, final RestHandler restHandler) {
if (restRequest.getXContentType() == null) {
String contentTypeHeader = restRequest.header("Content-Type");
if (restHandler.supportsContentStream() && contentTypeHeader != null) {
final String lowercaseMediaType = contentTypeHeader.toLowerCase(Locale.ROOT);
// we also support newline delimited JSON: http://specs.okfnlabs.org/ndjson/
if (lowercaseMediaType.equals("application/x-ndjson")) {
restRequest.setXContentType(XContentType.JSON);
return true;
}
private boolean handleNoHandlerFound(String rawPath, RestRequest.Method method, String uri, RestChannel channel) {
// Get the map of matching handlers for a request, for the full set of HTTP methods.
final Set<RestRequest.Method> validMethodSet = getValidHandlerMethodSet(rawPath);
if (validMethodSet.contains(method) == false) {
if (method == RestRequest.Method.OPTIONS) {
handleOptionsRequest(channel, validMethodSet);
return true;
}
if (validMethodSet.isEmpty() == false) {
// If an alternative handler for an explicit path is registered to a
// different HTTP method than the one supplied - return a 405 Method
// Not Allowed error.
handleUnsupportedHttpMethod(uri, method, channel, validMethodSet, null);
return true;
}
return false;
}
return true;
return false;
}

private void sendContentTypeErrorMessage(@Nullable List<String> contentTypeHeader, RestChannel channel) throws IOException {
Expand All @@ -283,34 +260,29 @@ private void sendContentTypeErrorMessage(@Nullable List<String> contentTypeHeade
channel.sendResponse(BytesRestResponse.createSimpleErrorResponse(channel, NOT_ACCEPTABLE, errorMessage));
}

/**
* Checks the request parameters against enabled settings for error trace support
* @return true if the request does not have any parameters that conflict with system settings
*/
private static boolean checkErrorTraceParameter(final RestRequest request, final RestChannel channel) {
// error_trace cannot be used when we disable detailed errors
// we consume the error_trace parameter first to ensure that it is always consumed
return request.paramAsBoolean("error_trace", false) == false || channel.detailedErrorsEnabled();
}

private void tryAllHandlers(final RestRequest request, final RestChannel channel, final ThreadContext threadContext) throws Exception {
for (String key : headersToCopy) {
String httpHeader = request.header(key);
if (httpHeader != null) {
threadContext.putHeader(key, httpHeader);
}
}
if (checkErrorTraceParameter(request, channel) == false) {
// error_trace cannot be used when we disable detailed errors
// we consume the error_trace parameter first to ensure that it is always consumed
if (request.paramAsBoolean("error_trace", false) && channel.detailedErrorsEnabled() == false) {
channel.sendResponse(
BytesRestResponse.createSimpleErrorResponse(channel, BAD_REQUEST, "error traces in responses are disabled."));
return;
}

final String rawPath = request.rawPath();
final String uri = request.uri();
final RestRequest.Method requestMethod;
try {
// Resolves the HTTP method and fails if the method is invalid
final RestRequest.Method requestMethod = request.method();
requestMethod = request.method();
// Loop through all possible handlers, attempting to dispatch the request
Iterator<MethodHandlers> allHandlers = getAllHandlers(request);
Iterator<MethodHandlers> allHandlers = getAllHandlers(request.params(), rawPath);
while (allHandlers.hasNext()) {
final RestHandler handler;
final MethodHandlers handlers = allHandlers.next();
Expand All @@ -319,32 +291,41 @@ private void tryAllHandlers(final RestRequest request, final RestChannel channel
} else {
handler = handlers.getHandler(requestMethod);
}
if (dispatchRequest(request, channel, client, handler)) {
if (handler == null) {
if (handleNoHandlerFound(rawPath, requestMethod, uri, channel)) {
return;
}
} else if (dispatchRequest(request, channel, handler)) {
return;
}
}
} catch (final IllegalArgumentException e) {
handleUnsupportedHttpMethod(request.uri(), null, channel, getValidHandlerMethodSet(request), e);
handleUnsupportedHttpMethod(uri, null, channel, getValidHandlerMethodSet(rawPath), e);
return;
}
// If request has not been handled, fallback to a bad request error.
handleBadRequest(request.uri(), request.method(), channel);
handleBadRequest(uri, requestMethod, channel);
}

Iterator<MethodHandlers> getAllHandlers(final RestRequest request) {
// Between retrieving the correct path, we need to reset the parameters,
// otherwise parameters are parsed out of the URI that aren't actually handled.
final Map<String, String> originalParams = new HashMap<>(request.params());
Iterator<MethodHandlers> getAllHandlers(@Nullable Map<String, String> requestParamsRef, String rawPath) {
final Supplier<Map<String, String>> paramsSupplier;
if (requestParamsRef == null) {
paramsSupplier = () -> null;
} else {
// Between retrieving the correct path, we need to reset the parameters,
// otherwise parameters are parsed out of the URI that aren't actually handled.
final Map<String, String> originalParams = Map.copyOf(requestParamsRef);
paramsSupplier = () -> {
// PathTrie modifies the request, so reset the params between each iteration
requestParamsRef.clear();
requestParamsRef.putAll(originalParams);
return requestParamsRef;
};
}
// we use rawPath since we don't want to decode it while processing the path resolution
// so we can handle things like:
// my_index/my_type/http%3A%2F%2Fwww.google.com
final Map<String, String> requestParamsRef = request.params();
return handlers.retrieveAll(request.rawPath(), () -> {
// PathTrie modifies the request, so reset the params between each iteration
requestParamsRef.clear();
requestParamsRef.putAll(originalParams);
return requestParamsRef;
});
return handlers.retrieveAll(rawPath, paramsSupplier);
}

/**
Expand Down Expand Up @@ -416,9 +397,9 @@ private void handleBadRequest(String uri, RestRequest.Method method, RestChannel
/**
* Get the valid set of HTTP methods for a REST request.
*/
private Set<RestRequest.Method> getValidHandlerMethodSet(RestRequest request) {
private Set<RestRequest.Method> getValidHandlerMethodSet(String rawPath) {
Set<RestRequest.Method> validMethods = new HashSet<>();
Iterator<MethodHandlers> allHandlers = getAllHandlers(request);
Iterator<MethodHandlers> allHandlers = getAllHandlers(null, rawPath);
for (Iterator<MethodHandlers> it = allHandlers; it.hasNext(); ) {
Optional.ofNullable(it.next()).map(mh -> validMethods.addAll(mh.getValidMethods()));
}
Expand Down
10 changes: 2 additions & 8 deletions server/src/main/java/org/elasticsearch/rest/RestRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
Expand Down Expand Up @@ -248,13 +249,6 @@ public final XContentType getXContentType() {
return xContentType.get();
}

/**
* Sets the {@link XContentType}
*/
final void setXContentType(XContentType xContentType) {
this.xContentType.set(xContentType);
}

public HttpChannel getHttpChannel() {
return httpChannel;
}
Expand Down Expand Up @@ -294,7 +288,7 @@ public Map<String, String> params() {
* @return the list of currently consumed parameters.
*/
List<String> consumedParams() {
return consumedParams.stream().collect(Collectors.toList());
return new ArrayList<>(consumedParams);
}

/**
Expand Down
2 changes: 1 addition & 1 deletion server/src/test/java/org/elasticsearch/rest/RestControllerTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,7 @@ public void testApplyRelevantHeaders() throws Exception {
restHeaders.put("header.3", Collections.singletonList("false"));
RestRequest fakeRequest = new FakeRestRequest.Builder(xContentRegistry()).withHeaders(restHeaders).build();
final RestController spyRestController = spy(restController);
when(spyRestController.getAllHandlers(fakeRequest))
when(spyRestController.getAllHandlers(null, fakeRequest.rawPath()))
.thenReturn(new Iterator<MethodHandlers>() {
@Override
public boolean hasNext() {
Expand Down
3 changes: 2 additions & 1 deletion server/src/test/java/org/elasticsearch/rest/RestRequestTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,9 +91,10 @@ private <T extends Exception> void runConsumesContentTest(
final HttpRequest httpRequest = mock(HttpRequest.class);
when (httpRequest.uri()).thenReturn("");
when (httpRequest.content()).thenReturn(new BytesArray(new byte[1]));
when (httpRequest.getHeaders()).thenReturn(
Collections.singletonMap("Content-Type", Collections.singletonList(randomFrom("application/json", "application/x-ndjson"))));
final RestRequest request =
RestRequest.request(mock(NamedXContentRegistry.class), httpRequest, mock(HttpChannel.class));
request.setXContentType(XContentType.JSON);
assertFalse(request.isContentConsumed());
try {
consumer.accept(request);
Expand Down

0 comments on commit 39142db

Please sign in to comment.