You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ke...@apache.org on 2025/03/25 22:00:20 UTC
(tinkerpop) branch master updated (08c0fab828 -> a6b017e20a)
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
from 08c0fab828 Merge branch '3.8-dev'
add cbafd31f28 TINKERPOP-3144 Close traversal on failure (#3075)
add d7ea77e55c Merge branch '3.7-dev' into 3.8-dev
new a6b017e20a Merge branch '3.8-dev'
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../server/handler/HttpGremlinEndpointHandler.java | 27 +++++++++++++++++-----
1 file changed, 21 insertions(+), 6 deletions(-)
(tinkerpop) 01/01: Merge branch '3.8-dev'
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kenhuuu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit a6b017e20aade6299ecd20f2ca21f8eba3e4af22
Merge: 08c0fab828 d7ea77e55c
Author: Ken Hu <10...@users.noreply.github.com>
AuthorDate: Tue Mar 25 14:56:04 2025 -0700
Merge branch '3.8-dev'
.../server/handler/HttpGremlinEndpointHandler.java | 27 +++++++++++++++++-----
1 file changed, 21 insertions(+), 6 deletions(-)
diff --cc gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
index 81388de4ed,665cf1ef3c..269e436d0d
--- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
+++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/HttpGremlinEndpointHandler.java
@@@ -18,46 -18,35 +18,47 @@@
*/
package org.apache.tinkerpop.gremlin.server.handler;
+import com.codahale.metrics.Meter;
import com.codahale.metrics.Timer;
import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.channel.ChannelPromise;
+import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.TooLongFrameException;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.FullHttpRequest;
-import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpContent;
+import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
-import io.netty.handler.codec.http.HttpUtil;
-import io.netty.util.ReferenceCountUtil;
+import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor;
-import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptChecker;
-import org.apache.tinkerpop.gremlin.process.remote.traversal.DefaultRemoteTraverser;
+import org.apache.tinkerpop.gremlin.groovy.jsr223.TimedInterruptTimeoutException;
+import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptEngine;
+import org.apache.tinkerpop.gremlin.language.grammar.GremlinParserException;
+import org.apache.tinkerpop.gremlin.process.traversal.Failure;
+import org.apache.tinkerpop.gremlin.process.traversal.Operator;
+import org.apache.tinkerpop.gremlin.process.traversal.Order;
+import org.apache.tinkerpop.gremlin.process.traversal.Pop;
+import org.apache.tinkerpop.gremlin.process.traversal.Scope;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.AbstractTraverser;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalInterruptedException;
+import org.apache.tinkerpop.gremlin.server.Context;
import org.apache.tinkerpop.gremlin.server.GraphManager;
import org.apache.tinkerpop.gremlin.server.GremlinServer;
+import org.apache.tinkerpop.gremlin.server.ProcessingException;
import org.apache.tinkerpop.gremlin.server.Settings;
import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
+import org.apache.tinkerpop.gremlin.server.util.GremlinError;
import org.apache.tinkerpop.gremlin.server.util.MetricManager;
-import org.apache.tinkerpop.gremlin.server.util.TextPlainMessageSerializer;
-import org.apache.tinkerpop.gremlin.structure.Element;
+import org.apache.tinkerpop.gremlin.server.util.TraverserIterator;
+import org.apache.tinkerpop.gremlin.structure.Column;
import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory;
+import org.apache.tinkerpop.gremlin.structure.T;
++import org.apache.tinkerpop.gremlin.structure.util.CloseableIterator;
+import org.apache.tinkerpop.gremlin.structure.util.TemporaryException;
import org.apache.tinkerpop.gremlin.util.ExceptionHelper;
import org.apache.tinkerpop.gremlin.util.MessageSerializer;
import org.apache.tinkerpop.gremlin.util.Tokens;
@@@ -206,145 -186,138 +207,159 @@@ public class HttpGremlinEndpointHandle
String address = ctx.channel().remoteAddress().toString();
if (address.startsWith("/") && address.length() > 1) address = address.substring(1);
auditLogger.info("User {} with address {} requested: {}", user.getName(), address,
- requestMessage.getArgOrDefault(Tokens.ARGS_GREMLIN, ""));
- }
- final ChannelPromise promise = ctx.channel().newPromise();
- final AtomicReference<Object> resultHolder = new AtomicReference<>();
- promise.addListener(future -> {
- // if failed then the error was already written back to the client as part of the eval future
- // processing of the exception
- if (future.isSuccess()) {
- logger.debug("Preparing HTTP response for request with script [{}] and bindings of [{}] with result of [{}] on [{}]",
- requestMessage.getArgOrDefault(Tokens.ARGS_GREMLIN, ""),
- requestMessage.getArgOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap()),
- resultHolder.get(), Thread.currentThread().getName());
- final FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK, (ByteBuf) resultHolder.get());
- response.headers().set(HttpHeaderNames.CONTENT_TYPE, serializer.getValue0());
-
- // handle cors business
- if (origin != null) response.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, origin);
-
- HttpHandlerUtil.sendAndCleanupConnection(ctx, keepAlive, response);
- }
- });
-
- final Timer.Context timerContext = evalOpTimer.time();
-
- final Bindings bindings;
- try {
- bindings = createBindings(requestMessage.getArgOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap()),
- requestMessage.getArgOrDefault(Tokens.ARGS_ALIASES, Collections.emptyMap()));
- } catch (IllegalStateException iae) {
- HttpHandlerUtil.sendError(ctx, BAD_REQUEST, requestId, iae.getMessage(), keepAlive);
- ReferenceCountUtil.release(msg);
- return;
+ requestMessage.getGremlin());
}
- // provide a transform function to serialize to message - this will force serialization to occur
- // in the same thread as the eval. after the CompletableFuture is returned from the eval the result
- // is ready to be written as a ByteBuf directly to the response. nothing should be blocking here.
- final CompletableFuture<Object> evalFuture = gremlinExecutor.eval(
- requestMessage.getArg(Tokens.ARGS_GREMLIN), requestMessage.getArg(Tokens.ARGS_LANGUAGE), bindings,
- requestMessage.getArgOrDefault(Tokens.ARGS_EVAL_TIMEOUT, null),
- FunctionUtils.wrapFunction(o -> {
- // stopping the timer here is roughly equivalent to where the timer would have been stopped for
- // this metric in other contexts. we just want to measure eval time not serialization time.
- timerContext.stop();
-
- logger.debug("Transforming result of request with script [{}] and bindings of [{}] with result of [{}] on [{}]",
- requestMessage.getArg(Tokens.ARGS_GREMLIN),
- requestMessage.getArg(Tokens.ARGS_BINDINGS), o, Thread.currentThread().getName());
-
- final Optional<String> mp = requestMessage.getArg(Tokens.ARGS_GREMLIN) instanceof String
- ? GremlinScriptChecker.parse(requestMessage.getArg(Tokens.ARGS_GREMLIN)).getMaterializeProperties()
- : Optional.empty();
-
- // need to replicate what TraversalOpProcessor does with the bytecode op. it converts
- // results to Traverser so that GLVs can handle the results. don't quite get the same
- // benefit here because the bulk has to be 1 since we've already resolved the result,
- // but at least http is compatible
- final List<Object> results = requestMessage.getOp().equals(Tokens.OPS_BYTECODE) ?
- (List<Object>) IteratorUtils.asList(o).stream().map(r -> new DefaultRemoteTraverser<Object>(r, 1)).collect(Collectors.toList()) :
- IteratorUtils.asList(o);
-
- if (mp.isPresent() && mp.get().equals(Tokens.MATERIALIZE_PROPERTIES_TOKENS)) {
- final Object firstElement = results.get(0);
-
- if (firstElement instanceof Element) {
- for (int i = 0; i < results.size(); i++)
- results.set(i, ReferenceFactory.detach((Element) results.get(i)));
- } else if (firstElement instanceof AbstractTraverser) {
- for (final Object item : results)
- ((AbstractTraverser) item).detach();
- }
- }
-
- final ResponseMessage responseMessage = ResponseMessage.build(requestId)
- .code(ResponseStatusCode.SUCCESS)
- .result(results).create();
-
- // http server is sessionless and must handle commit on transactions. the commit occurs
- // before serialization to be consistent with how things work for websocket based
- // communication. this means that failed serialization does not mean that you won't get
- // a commit to the database
- attemptCommit(requestMessage.getArg(Tokens.ARGS_ALIASES), graphManager, settings.strictTransactionManagement);
-
- try {
- return Unpooled.wrappedBuffer(serializer.getValue1().serializeResponseAsBinary(responseMessage, ctx.alloc()));
- } catch (Exception ex) {
- logger.warn(String.format("Error during serialization for %s", responseMessage), ex);
-
- // creating a new SerializationException will clear the cause which will allow the
- // future to report a better error message. if the cause is present, then
- // GremlinExecutor will prefer the cause and we'll get a low level Jackson sort of
- // error in the response.
- if (ex instanceof SerializationException) {
- throw new SerializationException(String.format(
- "Could not serialize the result with %s - %s",
- serializer.getValue0(),
- ex.getMessage()));
- }
-
- throw ex;
- }
- }));
-
- evalFuture.exceptionally(t -> {
- if (t.getMessage() != null)
- HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, t.getMessage(), Optional.of(t), keepAlive);
- else
- HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, String.format("Error encountered evaluating script: %s",
- requestMessage.getArg(Tokens.ARGS_GREMLIN))
- , Optional.of(t), keepAlive);
- promise.setFailure(t);
- return null;
- });
-
- evalFuture.thenAcceptAsync(r -> {
- // now that the eval/serialization is done in the same thread - complete the promise so we can
- // write back the HTTP response on the same thread as the original request
- resultHolder.set(r);
- promise.setSuccess();
- }, gremlinExecutor.getExecutorService());
- } catch (Exception ex) {
- // send the error response here and don't rely on exception caught because it might not have the
- // context on whether to close the connection or not, based on keepalive.
- final Throwable t = ExceptionHelper.getRootCause(ex);
- if (t instanceof TooLongFrameException) {
- HttpHandlerUtil.sendError(ctx, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, requestId, t.getMessage() + " - increase the maxContentLength", keepAlive);
- } else if (t != null){
- HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, t.getMessage(), keepAlive);
- } else {
- HttpHandlerUtil.sendError(ctx, INTERNAL_SERVER_ERROR, requestId, ex.getMessage(), keepAlive);
+ // Send back the 200 OK response header here since the response is always chunk transfer encoded. Any
+ // failures that follow this will show up in the response body instead.
+ final HttpResponse responseHeader = new DefaultHttpResponse(HTTP_1_1, OK);
+ if (acceptsDeflateEncoding(ctx.attr(StateKey.REQUEST_HEADERS).get().getAll(ACCEPT_ENCODING))) {
+ responseHeader.headers().add(CONTENT_ENCODING, DEFLATE);
}
+ responseHeader.headers().set(TRANSFER_ENCODING, CHUNKED);
+ responseHeader.headers().set(HttpHeaderNames.CONTENT_TYPE, serializer.getValue0());
+ ctx.writeAndFlush(responseHeader);
+ ctx.channel().attr(StateKey.HTTP_RESPONSE_SENT).set(true);
+
+ iterateScriptEvalResult(requestCtx, serializer.getValue1(), requestMessage);
+ } catch (Throwable t) {
+ writeError(requestCtx, formErrorResponseMessage(t, requestMessage), serializer.getValue1());
+ } finally {
+ timerContext.stop();
+
+ // There is a race condition that this query may have finished before the timeoutFuture was created,
+ // though this is very unlikely. This is handled in the settor, if this has already been grabbed.
+ // If we passed this point and the setter hasn't been called, it will cancel the timeoutFuture inside
+ // the setter to compensate.
+ final ScheduledFuture<?> timeoutFuture = requestCtx.getTimeoutExecutor();
+ if (null != timeoutFuture)
+ timeoutFuture.cancel(true);
+ }
+
+ return null;
+ });
+
+ try {
+ final Future<?> executionFuture = requestCtx.getGremlinExecutor().getExecutorService().submit(evalFuture);
+ if (seto > 0) {
+ // Schedule a timeout in the thread pool for future execution
+ requestCtx.setTimeoutExecutor(requestCtx.getScheduledExecutorService().schedule(() -> {
+ executionFuture.cancel(true);
+ if (!requestCtx.getStartedResponse()) {
+ writeError(requestCtx, GremlinError.timeout(requestMessage), serializer.getValue1());
+ }
+ }, seto, TimeUnit.MILLISECONDS));
+ }
+ } catch (RejectedExecutionException ree) {
+ writeError(requestCtx, GremlinError.rateLimiting(), serializer.getValue1());
+ }
+ }
+
+ private GremlinError formErrorResponseMessage(Throwable t, RequestMessage requestMessage) {
+ if (t instanceof UndeclaredThrowableException) t = t.getCause();
+
+ // if any exception in the chain is TemporaryException or Failure then we should respond with the
+ // right error code so that the client knows to retry
+ final Optional<Throwable> possibleSpecialException = determineIfSpecialException(t);
+ if (possibleSpecialException.isPresent()) {
+ final Throwable special = possibleSpecialException.get();
+ if (special instanceof TemporaryException) {
+ return GremlinError.temporary(special);
+ }
+ if (special instanceof Failure) {
+ return GremlinError.failStep((Failure) special);
+ }
+ return GremlinError.general(special);
+ }
+ if (t instanceof ProcessingException) {
+ return ((ProcessingException) t).getError();
+ }
+ t = ExceptionHelper.getRootCause(t);
+
+ if (t instanceof TooLongFrameException) {
+ return GremlinError.longFrame(t);
+ }
+ if (t instanceof InterruptedException || t instanceof TraversalInterruptedException) {
+ return GremlinError.timeout(requestMessage);
+ }
+ if (t instanceof TimedInterruptTimeoutException) {
+ // occurs when the TimedInterruptCustomizerProvider is in play
+ logger.warn(String.format("A timeout occurred within the script during evaluation of [%s] - consider increasing the limit given to TimedInterruptCustomizerProvider", requestMessage));
+ return GremlinError.timedInterruptTimeout();
+ }
+ if (t instanceof TimeoutException) {
+ logger.warn(String.format("Script evaluation exceeded the configured threshold for request [%s]", requestMessage));
+ return GremlinError.timeout(requestMessage);
+ }
+ if (t instanceof MultipleCompilationErrorsException && t.getMessage().contains("Method too large") &&
+ ((MultipleCompilationErrorsException) t).getErrorCollector().getErrorCount() == 1) {
+ final GremlinError error = GremlinError.longRequest(requestMessage);
+ logger.warn(error.getMessage());
+ return error;
+ }
+ if (t instanceof GremlinParserException) {
+ return GremlinError.parsing((GremlinParserException) t);
+ }
+
+ logger.warn(String.format("Exception processing request [%s].", requestMessage));
+ return GremlinError.general(t);
+ }
+
+ private void iterateScriptEvalResult(final Context context, MessageSerializer<?> serializer, final RequestMessage message)
+ throws ProcessingException, InterruptedException, ScriptException {
+ if (message.optionalField(Tokens.ARGS_BINDINGS).isPresent()) {
+ final Map bindings = (Map) message.getFields().get(Tokens.ARGS_BINDINGS);
+ if (IteratorUtils.anyMatch(bindings.keySet().iterator(), k -> null == k || !(k instanceof String))) {
+ throw new ProcessingException(GremlinError.binding());
+ }
+
+ final Set<String> badBindings = IteratorUtils.set(IteratorUtils.<String>filter(bindings.keySet().iterator(), INVALID_BINDINGS_KEYS::contains));
+ if (!badBindings.isEmpty()) {
+ throw new ProcessingException(GremlinError.binding(badBindings));
+ }
+
+ // ignore control bindings that get passed in with the "#jsr223" prefix - those aren't used in compilation
+ if (IteratorUtils.count(IteratorUtils.filter(bindings.keySet().iterator(), k -> !k.toString().startsWith("#jsr223"))) > settings.maxParameters) {
+ throw new ProcessingException(GremlinError.binding(bindings.size(), settings.maxParameters));
+ }
+ }
+
+ final Map<String, Object> args = message.getFields();
+ final String language = args.containsKey(Tokens.ARGS_LANGUAGE) ? (String) args.get(Tokens.ARGS_LANGUAGE) : "gremlin-lang";
+ final GremlinScriptEngine scriptEngine = gremlinExecutor.getScriptEngineManager().getEngineByName(language);
+
+ final Bindings mergedBindings = mergeBindingsFromRequest(context, new SimpleBindings(graphManager.getAsBindings()));
+ final Object result = scriptEngine.eval(message.getGremlin(), mergedBindings);
+
+ final String bulkingSetting = context.getChannelHandlerContext().channel().attr(StateKey.REQUEST_HEADERS).get().get(Tokens.BULK_RESULTS);
+ // bulking only applies if it's gremlin-lang, and per request token setting takes precedence over header setting.
+ // The serializer check is temporarily needed because GraphSON hasn't been removed yet and doesn't support bulking.
+ final boolean bulking = language.equals("gremlin-lang") && serializer instanceof GraphBinaryMessageSerializerV4 ?
+ (args.containsKey(Tokens.BULK_RESULTS) ?
+ Objects.equals(args.get(Tokens.BULK_RESULTS), "true") :
+ Objects.equals(bulkingSetting, "true")) :
+ false;
+
- if (bulking) {
- // optimization for driver requests
- ((Traversal.Admin<?, ?>) result).applyStrategies();
- handleIterator(context, new TraverserIterator((Traversal.Admin<?, ?>) result), serializer, true);
- } else {
- handleIterator(context, IteratorUtils.asIterator(result), serializer, false);
++ Iterator itty = null;
++ try {
++ if (bulking) {
++ // optimization for driver requests
++ ((Traversal.Admin<?, ?>) result).applyStrategies();
++ itty = new TraverserIterator((Traversal.Admin<?, ?>) result);
++ handleIterator(context, itty, serializer, true);
++ } else {
++ itty = IteratorUtils.asIterator(result);
++ handleIterator(context, itty, serializer, false);
+ }
++ } catch (Exception ex) {
++ // TINKERPOP-3144 ensure Traversals are closed when exception thrown.
++ if (itty instanceof TraverserIterator) {
++ CloseableIterator.closeIterator(((TraverserIterator) itty).getTraversal());
++ } else if (itty != null) {
++ CloseableIterator.closeIterator(itty);
++ }
++
++ throw ex;
}
}