[
https://issues.apache.org/jira/browse/TINKERPOP-2245?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17316442#comment-17316442
]
ASF GitHub Bot commented on TINKERPOP-2245:
-------------------------------------------
spmallette commented on a change in pull request #1414:
URL: https://github.com/apache/tinkerpop/pull/1414#discussion_r608781109
##########
File path:
gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/AbstractRexster.java
##########
@@ -0,0 +1,758 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.tinkerpop.gremlin.server.handler;
+
+import groovy.lang.GroovyRuntimeException;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.tinkerpop.gremlin.driver.MessageSerializer;
+import org.apache.tinkerpop.gremlin.driver.Tokens;
+import org.apache.tinkerpop.gremlin.driver.message.RequestMessage;
+import org.apache.tinkerpop.gremlin.driver.message.ResponseMessage;
+import org.apache.tinkerpop.gremlin.driver.message.ResponseStatusCode;
+import org.apache.tinkerpop.gremlin.driver.ser.MessageTextSerializer;
+import
org.apache.tinkerpop.gremlin.groovy.jsr223.TimedInterruptTimeoutException;
+import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptEngine;
+import org.apache.tinkerpop.gremlin.jsr223.JavaTranslator;
+import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource;
+import
org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.VerificationException;
+import org.apache.tinkerpop.gremlin.process.traversal.util.BytecodeHelper;
+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.Settings;
+import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
+import org.apache.tinkerpop.gremlin.server.util.ExceptionHelper;
+import org.apache.tinkerpop.gremlin.server.util.TraverserIterator;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+import org.apache.tinkerpop.gremlin.structure.util.TemporaryException;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
+import org.codehaus.groovy.control.MultipleCompilationErrorsException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.script.Bindings;
+import javax.script.ScriptException;
+import javax.script.SimpleBindings;
+import java.io.InterruptedIOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
+
+/**
+ * A base implementation of {@link Rexster} which offers some common
functionality that matches typical Gremlin Server
+ * request response expectations for script, bytecode and graph operations.
The class is designed to be extended but
+ * take care in understanding the way that different methods are called as
they do depend on one another a bit. It
+ * maybe best to examine the source code to determine how best to use this
class or to extend from the higher order
+ * classes of {@link SingleRexster} or {@link MultiRexster}.
+ */
+public abstract class AbstractRexster implements Rexster, AutoCloseable {
+ private static final Logger logger =
LoggerFactory.getLogger(AbstractRexster.class);
+ private static final Logger auditLogger =
LoggerFactory.getLogger(GremlinServer.AUDIT_LOGGER_NAME);
+
+ private final boolean sessionIdOnRequest;
+ private final Channel initialChannel;
+ private final boolean transactionManaged;
+ private final String sessionId;
+ private final AtomicReference<ScheduledFuture<?>> sessionCancelFuture =
new AtomicReference<>();
+ private final AtomicReference<Future<?>> sessionFuture = new
AtomicReference<>();
+ private long actualTimeoutLengthWhenClosed = 0;
+ private Thread sessionThread;
+ protected final boolean maintainStateAfterException;
+ protected final AtomicReference<CloseReason> closeReason = new
AtomicReference<>();
+ protected final GraphManager graphManager;
+ protected final ConcurrentMap<String, Rexster> sessions;
+ protected final Set<String> aliasesUsedByRexster = new HashSet<>();
+
+ protected enum CloseReason { UNDETERMINED, CHANNEL_CLOSED,
SESSION_TIMEOUT, REQUEST_TIMEOUT, NORMAL }
+
+ AbstractRexster(final Context gremlinContext, final String sessionId,
+ final boolean transactionManaged,
+ final ConcurrentMap<String, Rexster> sessions) {
+ // this only applies to sessions
+ this.maintainStateAfterException = (boolean)
gremlinContext.getRequestMessage().
+
optionalArgs(Tokens.ARGS_MAINTAIN_STATE_AFTER_EXCEPTION).orElse(false);
+ this.sessionIdOnRequest =
gremlinContext.getRequestMessage().optionalArgs(Tokens.ARGS_SESSION).isPresent();
+ this.transactionManaged = transactionManaged;
+ this.sessionId = sessionId;
+ this.initialChannel =
gremlinContext.getChannelHandlerContext().channel();
+
+ // close Rexster if the channel closes to cleanup and close
transactions
+ this.initialChannel.closeFuture().addListener(f -> {
+ if (closeReason.compareAndSet(null, CloseReason.CHANNEL_CLOSED)) {
+ // cancel session worker or it will keep waiting for items to
appear in the session queue
+ cancel(true);
+ close();
+ }
+ });
+ this.sessions = sessions;
+ this.graphManager = gremlinContext.getGraphManager();
+ }
+
+ protected synchronized void cancel(final boolean mayInterruptIfRunning) {
+ final FutureTask<?> sf = (FutureTask) sessionFuture.get();
+ if (sf != null && !sf.isDone()) {
+ sf.cancel(mayInterruptIfRunning);
+ }
+ }
+
+ public boolean isTransactionManaged() {
+ return transactionManaged;
+ }
+
+ public String getSessionId() {
+ return sessionId;
+ }
+
+ public boolean isBoundTo(final Channel channel) {
+ return channel == initialChannel;
+ }
+
+ public long getActualTimeoutLengthWhenClosed() {
+ return actualTimeoutLengthWhenClosed;
+ }
+
+ public Optional<CloseReason> getCloseReason() {
+ return Optional.ofNullable(closeReason.get());
+ }
+
+ public GremlinScriptEngine getScriptEngine(final Context context, final
String language) {
+ return
context.getGremlinExecutor().getScriptEngineManager().getEngineByName(language);
+ }
+
+ public void setSessionThread(final Thread runner) {
+ this.sessionThread = runner;
+ }
+
+ @Override
+ public void setSessionCancelFuture(final ScheduledFuture<?> f) {
+ if (!sessionCancelFuture.compareAndSet(null, f))
+ throw new IllegalStateException("Session cancellation future is
already set");
+ }
+
+ @Override
+ public void setSessionFuture(final Future<?> f) {
+ if (!sessionFuture.compareAndSet(null, f))
+ throw new IllegalStateException("Session future is already set");
+ }
+
+ @Override
+ public synchronized void triggerTimeout(final long timeout, final boolean
causedBySession) {
+ // triggering timeout triggers the stop of the Rexster Runnable which
will end in close()
+ // for final cleanup
+ final Future<?> f = sessionFuture.get();
+ if (f != null && !f.isDone()) {
+ if (closeReason.compareAndSet(null, causedBySession ?
CloseReason.SESSION_TIMEOUT : CloseReason.REQUEST_TIMEOUT)) {
+ actualTimeoutLengthWhenClosed = timeout;
+
+ // if caused by a session timeout for a session OR if it is a
request timeout for a sessionless
+ // request then we can just straight cancel() the Rexster
instance
+ if (causedBySession || !sessionIdOnRequest)
+ cancel(true);
+ else
+ sessionThread.interrupt();
+ }
+ }
+ }
+
+ protected void process(final Context gremlinContext) throws
RexsterException {
+ final RequestMessage msg = gremlinContext.getRequestMessage();
+ final Map<String, Object> args = msg.getArgs();
+ final Object gremlinToExecute = args.get(Tokens.ARGS_GREMLIN);
+
+ // for strict transactions track the aliases used so that we can
commit them and only them on close()
+ if (gremlinContext.getSettings().strictTransactionManagement)
+ msg.optionalArgs(Tokens.ARGS_ALIASES).ifPresent(m ->
aliasesUsedByRexster.addAll(((Map<String,String>) m).values()));
+
+ try {
+ // itty is optional as Bytecode could be a "graph operation"
rather than a Traversal. graph operations
+ // don't need to be iterated and handle their own lifecycle
+ final Optional<Iterator<?>> itty = gremlinToExecute instanceof
Bytecode ?
+ fromBytecode(gremlinContext, (Bytecode) gremlinToExecute) :
+ Optional.of(fromScript(gremlinContext, (String)
gremlinToExecute));
+
+ processAuditLog(gremlinContext.getSettings(),
gremlinContext.getChannelHandlerContext(), gremlinToExecute);
+
+ if (itty.isPresent())
+ handleIterator(gremlinContext, itty.get());
+ } catch (Exception ex) {
+ handleException(gremlinContext, ex);
+ }
+ }
+
+ protected void handleException(final Context gremlinContext, final
Throwable t) throws RexsterException {
+ if (t instanceof RexsterException) throw (RexsterException) t;
+
+ final Optional<Throwable> possibleTemporaryException =
determineIfTemporaryException(t);
+ if (possibleTemporaryException.isPresent()) {
+ final Throwable temporaryException =
possibleTemporaryException.get();
+ throw new RexsterException(temporaryException.getMessage(), t,
+ ResponseMessage.build(gremlinContext.getRequestMessage())
+ .code(ResponseStatusCode.SERVER_ERROR_TEMPORARY)
+ .statusMessage(temporaryException.getMessage())
+
.statusAttributeException(temporaryException).create());
+ }
+
+ final Throwable root = ExceptionUtils.getRootCause(t);
+
+ if (root instanceof TimedInterruptTimeoutException) {
+ // occurs when the TimedInterruptCustomizerProvider is in play
+ final String msg = String.format("A timeout occurred within the
script during evaluation of [%s] - consider increasing the limit given to
TimedInterruptCustomizerProvider",
+ gremlinContext.getRequestMessage().getRequestId());
+ throw new RexsterException(msg, root,
ResponseMessage.build(gremlinContext.getRequestMessage())
+ .code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
+ .statusMessage("Timeout during script evaluation triggered
by TimedInterruptCustomizerProvider")
+ .create());
+ }
+
+ if (root instanceof TimeoutException) {
+ final String errorMessage = String.format("Script evaluation
exceeded the configured threshold for request [%s]",
+ gremlinContext.getRequestMessage().getRequestId());
+ throw new RexsterException(errorMessage, root,
ResponseMessage.build(gremlinContext.getRequestMessage())
+ .code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
+ .statusMessage(t.getMessage())
+ .create());
+ }
+
+ if (root instanceof InterruptedException ||
+ root instanceof TraversalInterruptedException ||
+ root instanceof InterruptedIOException) {
+ String msg = "Processing interrupted but the reason why was not
known";
+ switch (closeReason.get()) {
+ case CHANNEL_CLOSED:
+ msg = "Processing interrupted because the channel was
closed";
+ break;
+ case SESSION_TIMEOUT:
+ msg = String.format("Session closed - %s -
sessionLifetimeTimeout of %s ms exceeded", sessionId,
actualTimeoutLengthWhenClosed);
+ break;
+ case REQUEST_TIMEOUT:
+ msg = String.format("Evaluation exceeded timeout threshold
of %s ms", actualTimeoutLengthWhenClosed);
+ break;
+ }
+ final ResponseStatusCode code = closeReason.get() ==
CloseReason.SESSION_TIMEOUT || closeReason.get() == CloseReason.REQUEST_TIMEOUT
?
+ ResponseStatusCode.SERVER_ERROR_TIMEOUT :
ResponseStatusCode.SERVER_ERROR;
+ throw new RexsterException(msg, root,
ResponseMessage.build(gremlinContext.getRequestMessage())
+ .code(code)
+ .statusMessage(msg).create());
+ }
+
+ if (root instanceof MultipleCompilationErrorsException &&
root.getMessage().contains("Method too large") &&
+ ((MultipleCompilationErrorsException)
root).getErrorCollector().getErrorCount() == 1) {
+ final String errorMessage = String.format("The Gremlin statement
that was submitted exceeds the maximum compilation size allowed by the JVM,
please split it into multiple smaller statements - %s",
trimMessage(gremlinContext.getRequestMessage()));
+ logger.warn(errorMessage);
+ throw new RexsterException(errorMessage, root,
ResponseMessage.build(gremlinContext.getRequestMessage())
+ .code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
+ .statusMessage(errorMessage)
+ .statusAttributeException(root).create());
+ }
+
+ // GroovyRuntimeException will hit a pretty wide range of eval type
errors, like MissingPropertyException,
+ // CompilationFailedException, MissingMethodException, etc. If more
specific handling is required then
+ // try to catch it earlier above.
+ if (root instanceof GroovyRuntimeException ||
+ root instanceof VerificationException ||
+ root instanceof ScriptException) {
+ throw new RexsterException(root.getMessage(), root,
ResponseMessage.build(gremlinContext.getRequestMessage())
+ .code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
+ .statusMessage(root.getMessage())
+ .statusAttributeException(root).create());
+ }
+
+ throw new RexsterException(root.getClass().getSimpleName() + ": " +
root.getMessage(), root,
+ ResponseMessage.build(gremlinContext.getRequestMessage())
+ .code(ResponseStatusCode.SERVER_ERROR)
+ .statusAttributeException(root)
+ .statusMessage(root.getMessage()).create());
+ }
+
+ /**
+ * Used to decrease the size of a Gremlin script that triggered a "method
too large" exception so that it
+ * doesn't log a massive text string nor return a large error message.
+ */
+ private RequestMessage trimMessage(final RequestMessage msg) {
+ final RequestMessage trimmedMsg = RequestMessage.from(msg).create();
+ if (trimmedMsg.getArgs().containsKey(Tokens.ARGS_GREMLIN))
+ trimmedMsg.getArgs().put(Tokens.ARGS_GREMLIN,
trimmedMsg.getArgs().get(Tokens.ARGS_GREMLIN).toString().substring(0, 1021) +
"...");
+
+ return trimmedMsg;
+ }
+
+ /**
+ * Check if any exception in the chain is TemporaryException then we
should respond with the right error code so
+ * that the client knows to retry.
+ */
+ protected Optional<Throwable> determineIfTemporaryException(final
Throwable ex) {
+ return Stream.of(ExceptionUtils.getThrowables(ex)).
+ filter(i -> i instanceof TemporaryException).findFirst();
+ }
+
+ @Override
+ public synchronized void close() {
+ // already closing/closed
+ if (!sessions.containsKey(sessionId)) return;
+
+ sessions.remove(sessionId);
+
+ if (sessionCancelFuture.get() != null) {
+ final ScheduledFuture<?> f = sessionCancelFuture.get();
+ if (!f.isDone()) f.cancel(true);
+ }
+ }
+
+ protected Iterator<?> fromScript(final Context gremlinContext, final
String script) throws Exception {
+ final RequestMessage msg = gremlinContext.getRequestMessage();
+ final Map<String, Object> args = msg.getArgs();
+ final String language = args.containsKey(Tokens.ARGS_LANGUAGE) ?
(String) args.get(Tokens.ARGS_LANGUAGE) : "gremlin-groovy";
+ return IteratorUtils.asIterator(getScriptEngine(gremlinContext,
language).eval(
+ script, mergeBindingsFromRequest(gremlinContext,
getWorkerBindings())));
+ }
+
+ protected Optional<Iterator<?>> fromBytecode(final Context gremlinContext,
final Bytecode bytecode) throws Exception {
+ final RequestMessage msg = gremlinContext.getRequestMessage();
+
+ final Traversal.Admin<?, ?> traversal;
+ final Map<String, String> aliases = (Map<String, String>)
msg.optionalArgs(Tokens.ARGS_ALIASES).get();
+ final GraphManager graphManager = gremlinContext.getGraphManager();
+ final String traversalSourceName =
aliases.entrySet().iterator().next().getValue();
+ final TraversalSource g =
graphManager.getTraversalSource(traversalSourceName);
+
+ // handle bytecode based graph operations like commit/rollback commands
+ if (BytecodeHelper.isGraphOperation(bytecode)) {
+ handleGraphOperation(gremlinContext, bytecode, g.getGraph());
+ return Optional.empty();
+ } else {
+
+ final Optional<String> lambdaLanguage =
BytecodeHelper.getLambdaLanguage(bytecode);
+ if (!lambdaLanguage.isPresent())
+ traversal = JavaTranslator.of(g).translate(bytecode);
+ else {
+ final SimpleBindings bindings = new SimpleBindings();
+ bindings.put(traversalSourceName, g);
+ traversal =
gremlinContext.getGremlinExecutor().getScriptEngineManager().getEngineByName(lambdaLanguage.get()).eval(bytecode,
bindings, traversalSourceName);
+ }
+
+ // compile the traversal - without it getEndStep() has nothing in
it
+ traversal.applyStrategies();
+
+ return Optional.of(new TraverserIterator(traversal));
+ }
+ }
+
+ protected Bindings getWorkerBindings() throws RexsterException {
+ return new SimpleBindings(graphManager.getAsBindings());
+ }
+
+ protected Bindings mergeBindingsFromRequest(final Context gremlinContext,
final Bindings bindings) throws RexsterException {
Review comment:
we've always had bindings/parameters for anything that needed scripts -
definitely not removed or that would cause script performance to tank as the
scriptengine cache would have no effect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Consolidate the executor for bytecode & string based client
> -----------------------------------------------------------
>
> Key: TINKERPOP-2245
> URL: https://issues.apache.org/jira/browse/TINKERPOP-2245
> Project: TinkerPop
> Issue Type: Improvement
> Components: server
> Affects Versions: 3.4.2
> Reporter: Divij Vaidya
> Assignee: Stephen Mallette
> Priority: Minor
>
> We have two code paths in the server which perform (more or less) the same
> functions. One is the executor for string based queries and other is the
> executor for bytecode. This code can be refactored together so that the logic
> to handle timeout, handle exception during execution, handle exception before
> execution and others can consolidated.
> [https://github.com/apache/tinkerpop/blob/master/gremlin-groovy/src/main/java/org/apache/tinkerpop/gremlin/groovy/engine/GremlinExecutor.java#L246]
> and
> [https://github.com/apache/tinkerpop/blob/master/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/op/traversal/TraversalOpProcessor.java#L333]
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
