in server/src/main/java/com/jetbrains/youtrackdb/internal/server/plugin/gremlin/YTDBAbstractOpProcessor.java [582:737]
protected void evalOpInternal(final Context ctx,
final Supplier<GremlinExecutor> gremlinExecutorSupplier,
final YTDBGraphTraversalSource traversalSource) {
final var timerContext = evalOpTimer.time();
final var msg = ctx.getRequestMessage();
final var gremlinExecutor = gremlinExecutorSupplier.get();
final var settings = ctx.getSettings();
final var args = msg.getArgs();
final var script = (String) args.get(Tokens.ARGS_GREMLIN);
final var language = "gremlin-lang";
final Bindings bindings = new SimpleBindings();
//noinspection unchecked
Optional.ofNullable((Map<String, Object>) msg.getArgs().get(Tokens.ARGS_BINDINGS))
.ifPresent(bindings::putAll);
bindings.put(Tokens.VAL_TRAVERSAL_SOURCE_ALIAS, traversalSource);
// timeout override - handle both deprecated and newly named configuration. earlier logic should prevent
// both configurations from being submitted at the same time
final var seto = args.containsKey(Tokens.ARGS_EVAL_TIMEOUT) ?
((Number) args.get(Tokens.ARGS_EVAL_TIMEOUT)).longValue() : settings.getEvaluationTimeout();
final var lifeCycle = GremlinExecutor.LifeCycle.build()
.evaluationTimeoutOverride(seto)
.afterFailure((b, t) -> {
var tx = traversalSource.tx();
if (tx.isOpen()) {
tx.rollback();
}
})
.afterTimeout((b, t) -> {
var tx = traversalSource.tx();
if (tx.isOpen()) {
tx.rollback();
}
})
.beforeEval(b -> {
b.putAll(bindings);
})
.withResult(o -> {
final var itty = IteratorUtils.asIterator(o);
logger.debug("Preparing to iterate results from - {} - in thread [{}]", msg,
Thread.currentThread().getName());
if (settings.enableAuditLog) {
var user = ctx.getChannelHandlerContext().channel().attr(StateKey.AUTHENTICATED_USER)
.get();
if (null == user) { // This is expected when using the AllowAllAuthenticator
user = AuthenticatedUser.ANONYMOUS_USER;
}
var address = ctx.getChannelHandlerContext().channel().remoteAddress().toString();
if (!address.isEmpty() && address.charAt(0) == '/' && address.length() > 1) {
address = address.substring(1);
}
auditLogger.info("User {} with address {} requested: {}", user.getName(), address,
script);
}
try {
handleIterator(ctx, itty, traversalSource);
} catch (Exception ex) {
var tx = traversalSource.tx();
if (tx.isOpen()) {
tx.rollback();
}
//noinspection unchecked
CloseableIterator.closeIterator(itty);
// wrap up the exception and rethrow. the error will be written to the client by the evalFuture
// as it will completeExceptionally in the GremlinExecutor
throw new RuntimeException(ex);
}
}).create();
try {
final var evalFuture = gremlinExecutor.eval(script, language, bindings, lifeCycle);
evalFuture.handle((v, t) -> {
timerContext.stop();
if (t != null) {
// 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 var possibleSpecialException = determineIfSpecialException(t);
if (possibleSpecialException.isPresent()) {
final var special = possibleSpecialException.get();
final var specialResponseMsg = ResponseMessage.build(msg).
statusMessage(special.getMessage()).
statusAttributeException(special);
if (special instanceof TemporaryException) {
specialResponseMsg.code(ResponseStatusCode.SERVER_ERROR_TEMPORARY);
} else if (special instanceof Failure failure) {
specialResponseMsg.code(ResponseStatusCode.SERVER_ERROR_FAIL_STEP).
statusAttribute(Tokens.STATUS_ATTRIBUTE_FAIL_STEP_MESSAGE, failure.format());
}
ctx.writeAndFlush(specialResponseMsg.create());
} else if (t instanceof OpProcessorException) {
ctx.writeAndFlush(((OpProcessorException) t).getResponseMessage());
} else if (t instanceof TimedInterruptTimeoutException) {
// occurs when the TimedInterruptCustomizerProvider is in play
final var errorMessage = String.format(
"A timeout occurred within the script during evaluation of [%s] - consider increasing the limit given to TimedInterruptCustomizerProvider",
msg);
logger.warn(errorMessage);
ctx.writeAndFlush(
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
.statusMessage(
"Timeout during script evaluation triggered by TimedInterruptCustomizerProvider")
.statusAttributeException(t).create());
} else if (t instanceof TimeoutException) {
final var errorMessage = String.format(
"Script evaluation exceeded the configured threshold for request [%s]", msg);
logger.warn(errorMessage, t);
ctx.writeAndFlush(
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_TIMEOUT)
.statusMessage(t.getMessage())
.statusAttributeException(t).create());
} else {
// try to trap the specific jvm error of "Method code too large!" to re-write it as something nicer,
// but only re-write if it's the only error because otherwise we might lose some other important
// information related to the failure. at this point, there hasn't been a scenario that has
// presented itself where the "Method code too large!" comes with other compilation errors so
// it seems that this message trumps other compilation errors to some reasonable degree that ends
// up being favorable for this problem
if (t instanceof MultipleCompilationErrorsException && t.getMessage()
.contains("Method too large") &&
((MultipleCompilationErrorsException) t).getErrorCollector().getErrorCount() == 1) {
final var 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(msg));
logger.warn(errorMessage);
ctx.writeAndFlush(
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
.statusMessage(errorMessage)
.statusAttributeException(t).create());
} else {
final var errorMessage = (t.getMessage() == null) ? t.toString() : t.getMessage();
logger.warn(String.format("Exception processing a script on request [%s].", msg), t);
ctx.writeAndFlush(
ResponseMessage.build(msg).code(ResponseStatusCode.SERVER_ERROR_EVALUATION)
.statusMessage(errorMessage)
.statusAttributeException(t).create());
}
}
}
//noinspection ReturnOfNull
return null;
});
} catch (RejectedExecutionException ree) {
ctx.writeAndFlush(ResponseMessage.build(msg).code(ResponseStatusCode.TOO_MANY_REQUESTS)
.statusMessage("Rate limiting").create());
}
}