From d1239ef88d2d9b4b9b23fb920bcf7f6a387053d7 Mon Sep 17 00:00:00 2001 From: contrueCT Date: Wed, 21 Oct 2026 21:31:13 +0800 Subject: [PATCH 1/4] feat(server): add embedded REST Gremlin execution - Add selectable forward and embedded REST executors with forward as default. - Reuse the existing Gremlin workers, serializer, auth context and transactions. - Integrate executor startup, shutdown and supported-topology validation. - Add differential contract tests, usage docs and reproducible benchmarks. --- docs/gremlin-rest-execution.md | 141 +++ hugegraph-server/README.md | 2 + .../gremlin/EmbeddedGremlinRestExecutor.java | 406 +++++++++ .../gremlin/ForwardGremlinRestExecutor.java | 71 ++ .../hugegraph/api/gremlin/GremlinAPI.java | 19 +- .../api/gremlin/GremlinQueryAPI.java | 14 +- .../api/gremlin/GremlinRestExecutor.java | 36 + .../gremlin/GremlinRestExecutorFactory.java | 96 ++ .../hugegraph/config/ServerOptions.java | 8 + .../hugegraph/server/ApplicationConfig.java | 26 + .../server/HugeGraphWsAndHttpChannelizer.java | 12 + .../apache/hugegraph/server/RestServer.java | 20 +- .../static/conf/rest-server.properties | 3 + .../hugegraph/dist/HugeGraphServer.java | 27 +- .../scripts/gremlin-rest-benchmark.py | 82 ++ .../scripts/gremlin-rest-rocksdb-benchmark.py | 363 ++++++++ .../auth/EmbeddedGremlinRestExecutorTest.java | 848 ++++++++++++++++++ .../apache/hugegraph/unit/UnitTestSuite.java | 2 + 18 files changed, 2164 insertions(+), 12 deletions(-) create mode 100644 docs/gremlin-rest-execution.md create mode 100644 hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java create mode 100644 hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/ForwardGremlinRestExecutor.java create mode 100644 hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutor.java create mode 100644 hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java create mode 100644 hugegraph-server/hugegraph-test/scripts/gremlin-rest-benchmark.py create mode 100644 hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py create mode 100644 hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java diff --git a/docs/gremlin-rest-execution.md b/docs/gremlin-rest-execution.md new file mode 100644 index 0000000000..47c49ffdd4 --- /dev/null +++ b/docs/gremlin-rest-execution.md @@ -0,0 +1,141 @@ +# REST Gremlin execution modes + +`gremlinserver.rest_execution_mode` selects the execution path for JSON requests to +REST `GET /gremlin` and `POST /gremlin`. The default is `forward`. + +```text +forward: REST -> Jersey HTTP client -> Gremlin HTTP listener -> Gremlin workers +embedded: REST ---------------------------------------------> Gremlin workers +``` + +Both paths use the same Gremlin executor, graph manager, script engines and configured +JSON serializer. The Gremlin listener remains available to HTTP/WebSocket clients, +Driver and Console. Cypher and Gremlin jobs keep their existing execution paths. + +## Configuration + +In `conf/rest-server.properties`: + +```properties +gremlinserver.url=http://127.0.0.1:8182 +gremlinserver.rest_execution_mode=embedded +``` + +Restart the server to change modes. To revert, set the mode to `forward` and restart. +There is no automatic forwarding retry after an embedded execution failure: a query +may already have committed even when the caller receives an error. + +The initial embedded implementation supports the standard HugeGraph +`WsAndHttpChannelizer` configuration, adapted internally to +`HugeGraphWsAndHttpChannelizer`. It requires the URL to identify the same process's +plain HTTP listener on a loopback address, with matching host/bind address and port. +Remote targets, TLS Gremlin listeners, custom channelizers and custom authorizers +continue to use `forward`. + +Authentication must agree between REST and Gremlin: either both are anonymous, or +both use `StandardAuthenticator`, with `WsAndHttpBasicAuthHandler` on Gremlin and +`authentication.config.tokens` resolving to the same REST configuration file. +Unsupported combinations fail startup; they do not silently select another mode. + +## Preserved behavior + +- The REST endpoint remains JSON-only. It does not expose typed GraphSON or GraphBinary + request formats, or introduce new content negotiation. +- GET keeps Jersey URI construction and the existing single-value rule for every query + parameter. POST uses the TinkerPop 3.8.1 JSON request parser. +- Alias resolution uses the live graph manager. Request bindings override alias bindings. +- Evaluation, iterator materialization, commit and serialization stay on the same Gremlin + worker, preserving thread-bound graph transactions and authentication context. +- Successful HTTP requests commit before serialization. A serialization failure does not + imply rollback of an already committed write. +- Existing rollback hooks, evaluation timeout, error mapping and Gremlin metrics remain + in use. REST input/output histograms remain at the same call sites; output length + reflects the response entity metadata of the selected transport. +- The serializer instance is obtained from the initialized Channelizer. No second serializer + registry, graph registry or script engine is created. + +Embedded requests authenticate once at the REST entry point. Workers inherit the existing +HugeGraph user context without rechecking credentials. Requests accepted before token expiry +or a permission change retain the existing context semantics; subsequent requests follow the +existing authentication and cache behavior. This feature adds no cross-node revocation or +immutable authorization snapshot guarantee. Audit records use the original REST peer address; +forward audit records continue to use the internal HTTP client's address. + +## Timeouts, load and lifecycle + +`gremlinserver.max_route` is retained as blocking admission for embedded requests. Pending +callers wait rather than receive a new overload rejection. `gremlinserver.timeout` bounds +waiting for the result after admission, while the Gremlin executor retains its separate +execution timeout. The outer REST server's request timeout is unchanged. + +A REST timeout does not cancel or retry the script. The script may continue and commit; +TinkerPop's existing execution timeout and rollback hooks remain responsible for execution. +Admission is released when the REST execution wait ends. This preserves the existing +possibility of work accumulating after client timeouts; it introduces no new hard task limit. + +The adapter is unavailable until Gremlin startup has initialized its serializer. During +normal server shutdown it rejects new submissions before the existing Gremlin stop sequence. +The Gremlin server continues to own worker shutdown and graph closure. This feature adds no +new readiness endpoint, deployment probe, global drain deadline or stronger termination guarantee. + +Protocol limits are checked against the internal request shape. This does not provide an +additional early limit before Jersey reads the outer REST request body. Existing transport +edge behavior is pinned by differential tests, including URI/header failures; it is not a new +listener-wide protection policy. + +## Verification + +Use Java 17 from the repository root: + +```bash +mvn test -pl hugegraph-server/hugegraph-test -am -P unit-test \ + -Dtest=EmbeddedGremlinRestExecutorTest,HttpGremlinRequestHandlerTest,ContextGremlinServerHttpTest,ContextTaskTest,WsAndHttpBasicAuthHandlerTest,GremlinQueryAPITest \ + -Dsurefire.failIfNoSpecifiedTests=false +``` + +The focused tests compare real forward HTTP responses with embedded execution, exercise +both REST HTTP endpoints, and cover parsing/encoding, protocol limits, error mapping, +request identity, context isolation, blocking admission, timeouts, shared serializers, +dynamic aliases and graph transaction side effects on Memory and RocksDB. Response stack traces and newly generated +request IDs require semantic comparison; deterministic responses also use byte comparisons. + +Run backend and distribution checks appropriate to the intended deployment before enabling +embedded. Focused tests do not establish HStore compatibility or production performance. +Default-mode changes require separate representative performance evidence. + +Readiness, overload policy, authentication-cache consistency and global shutdown redesign +remain separate tasks from this execution-path change. + +For a manual script-only HTTP transport comparison, run the focused tests above, then: + +```bash +python3 hugegraph-server/hugegraph-test/scripts/gremlin-rest-benchmark.py \ + --output /tmp/gremlin-rest-benchmark.json +``` + +Set `JAVA_HOME` to Java 17. The runner uses the tested runtime classpath, starts two +local REST endpoints sharing one Gremlin worker pool, checks every measured response, +and alternates mode order across four rounds. It saves latency samples and environment +metadata. These measurements include client/REST overhead and exclude real backend I/O; +100 warmup requests do not establish steady-state JIT behavior. + +For a Linux single-node RocksDB comparison, build the distribution and use an eight-CPU +host with Java 17. The following runner requires a new output directory and binds both +listeners to loopback. It generates its own graph and preserves all evidence under that +directory: + +```bash +python3 hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py \ + --distribution /path/to/apache-hugegraph-server-1.8.0 \ + --java-home /path/to/jdk-17 \ + --output /path/to/new-results-directory +``` + +It checks GET/POST, bindings, JSON values, aliases, rollback, committed write counts and +restart persistence. The default dataset has 10,000 vertices and 50,000 edges. Four server +starts use forward/embedded/embedded/forward order, identical JVM settings and persistent +data. Each workload has a five-second warmup and three-second measurements at concurrency +1, 8 and 32. Samples use a closed-loop client on the same host; CPU affinity separates +the client and server, while unrelated host workloads can still affect both. The results +include raw latency samples, response checks, CPU/load snapshots, GC and server logs. +These short measurements do not establish production saturation or steady-state behavior. diff --git a/hugegraph-server/README.md b/hugegraph-server/README.md index f559004bb7..edf06801dc 100644 --- a/hugegraph-server/README.md +++ b/hugegraph-server/README.md @@ -8,6 +8,8 @@ HugeGraph Server consists of two layers of functionality: the graph engine layer - REST Server: Provides a RESTful API for querying graph/schema information, supports the [Gremlin](https://tinkerpop.apache.org/gremlin.html) and [Cypher](https://en.wikipedia.org/wiki/Cypher) query languages, and offers APIs for service monitoring and operations. - See [Cypher HTTP contracts](../docs/server-runtime.md#cypher-http-contracts) for request forms, verified behavior, and unverified scope. + - See [REST Gremlin execution modes](../docs/gremlin-rest-execution.md) for optional in-process + execution, supported configurations and compatibility boundaries. - Graph Engine: Supports both OLTP and OLAP graph computation types, with OLTP implementing the [Apache TinkerPop3](https://tinkerpop.apache.org) framework. - Backend Interface: Implements the storage of graph data to the backend. diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java new file mode 100644 index 0000000000..424fa328f6 --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java @@ -0,0 +1,406 @@ +/* + * 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.hugegraph.api.gremlin; + +import java.net.SocketTimeoutException; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import javax.script.Bindings; +import javax.script.SimpleBindings; + +import org.apache.commons.lang3.exception.ExceptionUtils; +import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.http.util.VersionInfo; +import org.apache.hugegraph.auth.HugeGraphAuthProxy; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.ServerOptions; +import org.apache.hugegraph.util.E; +import org.apache.hugegraph.util.JsonUtil; +import org.apache.hugegraph.util.Log; +import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor; +import org.apache.tinkerpop.gremlin.jsr223.GremlinScriptChecker; +import org.apache.tinkerpop.gremlin.process.traversal.TraversalSource; +import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.AbstractTraverser; +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.AllowAllAuthenticator; +import org.apache.tinkerpop.gremlin.server.handler.HttpHandlerUtil; +import org.apache.tinkerpop.gremlin.server.util.MetricManager; +import org.apache.tinkerpop.gremlin.server.util.ServerGremlinExecutor; +import org.apache.tinkerpop.gremlin.structure.Element; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceFactory; +import org.apache.tinkerpop.gremlin.util.Tokens; +import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils; +import org.apache.tinkerpop.gremlin.util.message.RequestMessage; +import org.apache.tinkerpop.gremlin.util.message.ResponseMessage; +import org.apache.tinkerpop.gremlin.util.message.ResponseStatusCode; +import org.apache.tinkerpop.gremlin.util.ser.MessageTextSerializer; +import org.apache.tinkerpop.gremlin.util.ser.SerializationException; +import org.glassfish.jersey.internal.Version; +import org.slf4j.Logger; + +import com.codahale.metrics.MetricRegistry; +import com.codahale.metrics.Timer; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.buffer.UnpooledByteBufAllocator; +import io.netty.handler.codec.http.DefaultFullHttpRequest; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpVersion; +import jakarta.ws.rs.ProcessingException; +import jakarta.ws.rs.ServiceUnavailableException; +import jakarta.ws.rs.core.MultivaluedMap; +import jakarta.ws.rs.core.Response; +import jakarta.ws.rs.core.UriBuilder; + +/** + * In-process adapter for the TinkerPop 3.8.1 JSON HTTP contract. Materialization, + * commit and serialization must remain inside GremlinExecutor's transform so + * that graph transactions and the authenticated user stay on the worker. + */ +public final class EmbeddedGremlinRestExecutor implements GremlinRestExecutor { + + private static final Logger LOG = Log.logger(EmbeddedGremlinRestExecutor.class); + private static final Logger AUDIT = Log.logger(GremlinServer.AUDIT_LOGGER_NAME); + private static final Timer EVAL_TIMER = MetricManager.INSTANCE.getTimer( + MetricRegistry.name(GremlinServer.class, "op", "eval")); + + private static final String USER_AGENT = userAgent(); + + private final Settings settings; + private final GraphManager graphs; + private final GremlinExecutor executor; + private final Semaphore admission; + private final long waitMillis; + private final URI target; + private volatile MessageTextSerializer serializer; + private volatile boolean stopping; + + public EmbeddedGremlinRestExecutor(HugeConfig config, ServerGremlinExecutor server) { + this.settings = server.getSettings(); + this.graphs = server.getGraphManager(); + this.executor = server.getGremlinExecutor(); + this.admission = new Semaphore(config.get(ServerOptions.GREMLIN_SERVER_MAX_ROUTE), true); + this.waitMillis = TimeUnit.SECONDS.toMillis(config.get(ServerOptions.GREMLIN_SERVER_TIMEOUT)); + this.target = URI.create(config.get(ServerOptions.GREMLIN_SERVER_URL)); + } + + public void initialize(MessageTextSerializer serializer) { + E.checkNotNull(serializer, "application/json serializer"); + E.checkState(this.serializer == null && !this.stopping, + "Gremlin REST executor is already initialized or stopping"); + this.serializer = serializer; + } + + @Override + public Response get(String authorization, MultivaluedMap parameters, + String remoteAddress) { + UriBuilder builder = UriBuilder.fromUri(this.target); + for (Map.Entry> entry : parameters.entrySet()) { + E.checkArgument(entry.getValue().size() == 1, + "Invalid query param '%s', can only accept one value, but got %s", + entry.getKey(), entry.getValue()); + builder.queryParam(entry.getKey(), entry.getValue().get(0)); + } + final URI canonical; + try { + canonical = builder.build(); + } catch (IllegalArgumentException e) { + // JerseyWebTarget.getUri() reports unresolved templates this way. + throw new IllegalStateException(e.getMessage(), e); + } + String uri = canonical.getRawPath(); + if (uri == null || uri.isEmpty()) { + uri = "/"; + } + if (canonical.getRawQuery() != null) { + uri += "?" + canonical.getRawQuery(); + } + return this.execute(HttpMethod.GET, uri, new byte[0], authorization, remoteAddress); + } + + private String path() { + String path = this.target.getRawPath(); + return path == null || path.isEmpty() ? "/" : path; + } + + @Override + public Response post(String authorization, String body, String remoteAddress) { + String uri = this.path(); + if (this.target.getRawQuery() != null) { + uri += "?" + this.target.getRawQuery(); + } + return this.execute(HttpMethod.POST, uri, body.getBytes(StandardCharsets.UTF_8), + authorization, remoteAddress); + } + + private void checkAvailable() { + if (this.stopping || this.serializer == null) { + throw new ServiceUnavailableException("Gremlin REST is not accepting requests"); + } + } + + private Response execute(HttpMethod method, String uri, byte[] body, + String authorization, String remoteAddress) { + this.checkAvailable(); + boolean acquired = false; + try { + // Like the existing connection pool, wait for a lease without a new + // fail-fast policy. The read timeout starts after admission. + this.admission.acquire(); + acquired = true; + this.checkAvailable(); + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(this.waitMillis); + if ((method + " " + uri + " HTTP/1.1").length() > + this.settings.maxInitialLineLength) { + return this.protocolError(); + } + if (this.headerSize(method, authorization) > this.settings.maxHeaderSize) { + return this.protocolError(); + } + if (body.length > this.settings.maxContentLength) { + return Response.status(413).build(); + } + if ("/favicon.ico".equals(uri)) { + return error(404, null, "Gremlin Server doesn't have a favicon.ico", null); + } + final RequestMessage request; + FullHttpRequest http = new DefaultFullHttpRequest( + HttpVersion.HTTP_1_1, method, uri, Unpooled.wrappedBuffer(body)); + try { + request = HttpHandlerUtil.getRequestMessageFromHttpRequest(http); + } catch (IllegalArgumentException e) { + return error(400, null, e.getMessage(), null); + } catch (RuntimeException e) { + return error(500, null, e.getMessage(), null); + } finally { + http.release(); + } + if (this.settings.enableAuditLog) { + HugeGraphAuthProxy.Context context = HugeGraphAuthProxy.getContext(); + boolean anonymous = AllowAllAuthenticator.class.getName().equals( + this.settings.authentication.authenticator); + String user = anonymous || context == null ? "anonymous" : context.user().username(); + AUDIT.info("User {} with address {} requested: {}", user, remoteAddress, + request.getArgOrDefault(Tokens.ARGS_GREMLIN, "")); + } + Timer.Context timer = EVAL_TIMER.time(); + final Bindings bindings; + try { + bindings = this.bindings(request); + } catch (IllegalStateException e) { + return error(400, request.getRequestId(), e.getMessage(), null); + } + CompletableFuture evaluation; + try { + evaluation = this.executor.eval( + request.getArg(Tokens.ARGS_GREMLIN), + request.getArg(Tokens.ARGS_LANGUAGE), bindings, + request.getArgOrDefault(Tokens.ARGS_EVAL_TIMEOUT, null), value -> { + timer.stop(); + return this.serialize(request, value); + }); + } catch (Exception e) { + Throwable root = ExceptionUtils.getRootCause(e); + return error(500, request.getRequestId(), + root == null ? e.getMessage() : root.getMessage(), null); + } + CompletableFuture response = evaluation.handle((value, failure) -> { + if (failure != null) { + String message = failure.getMessage() != null ? failure.getMessage() : + "Error encountered evaluating script: " + + request.getArg(Tokens.ARGS_GREMLIN); + return error(500, request.getRequestId(), message, failure); + } + byte[] bytes = (byte[]) value; + return Response.ok(bytes, "application/json") + .header("Content-Length", bytes.length).build(); + }); + // Never cancel this CompletableFuture on a REST read timeout: + // TinkerPop owns the execution task and its timeout/rollback hooks. + // The completion handler also preserves error metrics after a client timeout. + try { + long remaining = Math.max(0L, deadline - System.nanoTime()); + return response.get(remaining, TimeUnit.NANOSECONDS); + } catch (ExecutionException e) { + throw new ProcessingException(e.getCause()); + } catch (TimeoutException e) { + throw new ProcessingException(new SocketTimeoutException("Read timed out")); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new ProcessingException(e); + } finally { + if (acquired) { + this.admission.release(); + } + } + } + + private Response protocolError() { + // The decoder replaces an invalid request with a headerless GET. The + // standard auth handler rejects that request before the JSON handler. + if (!AllowAllAuthenticator.class.getName().equals(this.settings.authentication.authenticator)) { + return Response.status(401).build(); + } + return error(400, null, "no gremlin script supplied", null); + } + + private static String userAgent() { + VersionInfo version = VersionInfo.loadVersionInfo( + "org.apache.http.client", HttpClientBuilder.class + .getClassLoader()); + String release = version == null ? "UNAVAILABLE" : version.getRelease(); + return "Jersey/" + Version.getVersion() + " (Apache HttpClient " + release + ")"; + } + + private int headerSize(HttpMethod method, String authorization) { + // HttpObjectDecoder counts header lines without CRLF. Match the headers + // produced by the pinned Jersey/Apache connector, not the outer REST headers. + String host = this.target.getHost(); + if (this.target.getPort() != 80) { + host += ":" + this.target.getPort(); + } + int length = ("Accept: application/json" + + "User-Agent: " + USER_AGENT + + "Accept-Encoding: gzip" + "Host: " + host + + "Connection: Keep-Alive").length(); + if (authorization != null) { + length += ("Authorization: " + authorization).getBytes(StandardCharsets.ISO_8859_1).length; + } + if (HttpMethod.POST.equals(method)) { + length += "Content-Type: application/jsonTransfer-Encoding: chunked".length(); + } + return length; + } + + private Bindings bindings(RequestMessage request) { + Bindings bindings = new SimpleBindings(); + Map aliases = request.getArgOrDefault(Tokens.ARGS_ALIASES, + Collections.emptyMap()); + for (Map.Entry alias : aliases.entrySet()) { + Graph graph = this.graphs.getGraph(alias.getValue()); + if (graph != null) { + bindings.put(alias.getKey(), graph); + } else { + TraversalSource source = this.graphs.getTraversalSource(alias.getValue()); + if (source == null) { + throw new IllegalStateException(String.format( + "Could not rebind [%s] to [%s] as [%s] not in the Graph or " + + "TraversalSource global bindings", alias.getKey(), alias.getValue(), + alias.getValue())); + } + bindings.put(alias.getKey(), source); + } + } + bindings.putAll(request.getArgOrDefault(Tokens.ARGS_BINDINGS, Collections.emptyMap())); + return bindings; + } + + private byte[] serialize(RequestMessage request, Object value) { + List results = IteratorUtils.asList(value); + Optional materialize = GremlinScriptChecker.parse( + (String) request.getArg(Tokens.ARGS_GREMLIN)).getMaterializeProperties(); + if (materialize.isPresent() && + Tokens.MATERIALIZE_PROPERTIES_TOKENS.equals(materialize.get())) { + // Keep the pinned HTTP handler's behavior, including empty-result errors. + Object first = results.get(0); + if (first instanceof Element) { + for (int i = 0; i < results.size(); i++) { + results.set(i, ReferenceFactory.detach((Element) results.get(i))); + } + } else if (first instanceof AbstractTraverser) { + for (Object item : results) { + ((AbstractTraverser) item).detach(); + } + } + } + ResponseMessage response = ResponseMessage.build(request.getRequestId()) + .code(ResponseStatusCode.SUCCESS) + .result(results).create(); + if (this.settings.strictTransactionManagement) { + Map aliases = request.getArg(Tokens.ARGS_ALIASES); + this.graphs.commit(new HashSet<>(aliases.values())); + } else { + this.graphs.commitAll(); + } + ByteBuf buffer = null; + try { + buffer = this.serializer.serializeResponseAsBinary(response, + UnpooledByteBufAllocator.DEFAULT); + byte[] bytes = new byte[buffer.readableBytes()]; + buffer.readBytes(bytes); + return bytes; + } catch (SerializationException e) { + // Preserve TP's cause-free serialization failure after commit. + throw new CompletionException(new SerializationException( + "Could not serialize the result with application/json - " + e.getMessage())); + } finally { + if (buffer != null) { + buffer.release(); + } + } + } + + private static Response error(int status, UUID requestId, String message, Throwable cause) { + LOG.warn("Invalid request - responding with {} and {}", status, message, cause); + MetricManager.INSTANCE.getMeter(MetricRegistry.name(GremlinServer.class, "errors")).mark(); + Map body = new LinkedHashMap<>(); + body.put("message", message); + if (cause != null) { + body.put("Exception-Class", cause.getClass().getName()); + body.put("exceptions", ExceptionUtils.getThrowableList(cause).stream() + .map(t -> t.getClass().getName()).toList()); + body.put("stackTrace", ExceptionUtils.getStackTrace(cause)); + } + if (requestId != null) { + body.put("requestId", requestId.toString()); + } + byte[] bytes = JsonUtil.toJson(body).getBytes(StandardCharsets.UTF_8); + return Response.status(status).type("application/json").entity(bytes) + .header("Content-Length", bytes.length).build(); + } + + @Override + public void beginShutdown() { + this.stopping = true; + } + + @Override + public void close() { + // Shared workers, graphs and serializers are owned by GremlinServer. + this.beginShutdown(); + } +} diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/ForwardGremlinRestExecutor.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/ForwardGremlinRestExecutor.java new file mode 100644 index 0000000000..0ce3e0d35c --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/ForwardGremlinRestExecutor.java @@ -0,0 +1,71 @@ +/* + * 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.hugegraph.api.gremlin; + +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.ServerOptions; + +import jakarta.ws.rs.ServiceUnavailableException; +import jakarta.ws.rs.core.MultivaluedMap; +import jakarta.ws.rs.core.Response; + +public final class ForwardGremlinRestExecutor implements GremlinRestExecutor { + + private final HugeConfig config; + private GremlinClient client; + private boolean closed; + + public ForwardGremlinRestExecutor(HugeConfig config) { + this.config = config; + } + + private synchronized GremlinClient client() { + if (this.closed) { + throw new ServiceUnavailableException("Gremlin REST is stopping"); + } + if (this.client == null) { + int routes = this.config.get(ServerOptions.GREMLIN_SERVER_MAX_ROUTE); + this.client = new GremlinClient( + this.config.get(ServerOptions.GREMLIN_SERVER_URL), + this.config.get(ServerOptions.GREMLIN_SERVER_TIMEOUT) * 1000, + routes, routes); + } + return this.client; + } + + @Override + public Response get(String authorization, MultivaluedMap parameters, + String remoteAddress) { + return this.client().doGetRequest(authorization, parameters); + } + + @Override + public Response post(String authorization, String body, String remoteAddress) { + return this.client().doPostRequest(authorization, body); + } + + @Override + public synchronized void close() { + if (!this.closed) { + this.closed = true; + if (this.client != null) { + this.client.close(); + } + } + } +} diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinAPI.java index c1701ea943..299a5ce285 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinAPI.java @@ -21,12 +21,15 @@ import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.metrics.MetricsUtil; import org.apache.hugegraph.util.E; +import org.glassfish.grizzly.http.server.Request; import com.codahale.metrics.Histogram; import com.codahale.metrics.annotation.Timed; import io.swagger.v3.oas.annotations.Parameter; import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.inject.Inject; +import jakarta.inject.Provider; import jakarta.inject.Singleton; import jakarta.ws.rs.Consumes; import jakarta.ws.rs.GET; @@ -44,6 +47,18 @@ @Tag(name = "GremlinAPI") public class GremlinAPI extends GremlinQueryAPI { + @Inject + private GremlinRestExecutor executor; + + @Context + private Provider peer; + + private String remoteAddress() { + Request request = this.peer.get(); + return request == null ? "unknown" : + request.getRemoteAddr() + ":" + request.getRemotePort(); + } + private static final Histogram GREMLIN_INPUT_HISTOGRAM = MetricsUtil.registerHistogram(GremlinAPI.class, "gremlin-input"); private static final Histogram GREMLIN_OUTPUT_HISTOGRAM = @@ -65,7 +80,7 @@ public Response post(@Context HugeConfig conf, // Response.temporaryRedirect(UriBuilder.fromUri(location).build()) // .build(); String auth = headers.getHeaderString(HttpHeaders.AUTHORIZATION); - Response response = this.client().doPostRequest(auth, request); + Response response = this.executor.post(auth, request, this.remoteAddress()); GREMLIN_INPUT_HISTOGRAM.update(request.length()); GREMLIN_OUTPUT_HISTOGRAM.update(response.getLength()); return transformResponseIfNeeded(response); @@ -82,7 +97,7 @@ public Response get(@Context HugeConfig conf, String query = uriInfo.getRequestUri().getRawQuery(); E.checkArgumentNotNull(query, "The request query can't be empty"); MultivaluedMap params = uriInfo.getQueryParameters(); - Response response = this.client().doGetRequest(auth, params); + Response response = this.executor.get(auth, params, this.remoteAddress()); GREMLIN_INPUT_HISTOGRAM.update(query.length()); GREMLIN_OUTPUT_HISTOGRAM.update(response.getLength()); return transformResponseIfNeeded(response); diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinQueryAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinQueryAPI.java index 2ab55f584a..d321c90420 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinQueryAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinQueryAPI.java @@ -17,6 +17,7 @@ package org.apache.hugegraph.api.gremlin; +import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.Set; @@ -24,6 +25,7 @@ import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.config.ServerOptions; import org.apache.hugegraph.exception.HugeGremlinException; +import org.apache.hugegraph.util.JsonUtil; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; @@ -81,13 +83,21 @@ protected static Response transformResponseIfNeeded(Response response) { } if (mediaType == null || !JSON.equals(mediaType.getSubtype())) { - String message = response.readEntity(String.class); + String message = response.getEntity() == null ? "" : + response.getEntity() instanceof byte[] ? + new String((byte[]) response.getEntity(), + StandardCharsets.UTF_8) : + response.readEntity(String.class); throw new HugeGremlinException(status.getStatusCode(), ImmutableMap.of("message", message)); } @SuppressWarnings("unchecked") - Map map = response.readEntity(Map.class); + Map map = response.getEntity() instanceof byte[] ? + JsonUtil.fromJson( + new String((byte[]) response.getEntity(), + StandardCharsets.UTF_8), + Map.class) : response.readEntity(Map.class); String exClassName = (String) map.get("Exception-Class"); if (FORBIDDEN_REQUEST_EXCEPTIONS.contains(exClassName)) { status = Response.Status.FORBIDDEN; diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutor.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutor.java new file mode 100644 index 0000000000..95e7b0e0e0 --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutor.java @@ -0,0 +1,36 @@ +/* + * 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.hugegraph.api.gremlin; + +import jakarta.ws.rs.core.MultivaluedMap; +import jakarta.ws.rs.core.Response; + +/** The execution path for the JSON-only REST Gremlin endpoint. */ +public interface GremlinRestExecutor extends AutoCloseable { + + Response get(String authorization, MultivaluedMap parameters, + String remoteAddress); + + Response post(String authorization, String body, String remoteAddress); + + default void beginShutdown() { + } + + @Override + void close(); +} diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java new file mode 100644 index 0000000000..add4713020 --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java @@ -0,0 +1,96 @@ +/* + * 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.hugegraph.api.gremlin; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.URI; +import java.nio.file.Path; + +import org.apache.hugegraph.auth.StandardAuthenticator; +import org.apache.hugegraph.auth.WsAndHttpBasicAuthHandler; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.ServerOptions; +import org.apache.hugegraph.server.HugeGraphWsAndHttpChannelizer; +import org.apache.hugegraph.util.E; +import org.apache.tinkerpop.gremlin.server.GremlinServer; +import org.apache.tinkerpop.gremlin.server.Settings; +import org.apache.tinkerpop.gremlin.server.auth.AllowAllAuthenticator; + +public final class GremlinRestExecutorFactory { + + private GremlinRestExecutorFactory() { + } + + public static GremlinRestExecutor standalone(HugeConfig config) { + E.checkArgument("forward".equals(config.get(ServerOptions.GREMLIN_REST_EXECUTION_MODE)), + "embedded requires startup through HugeGraphServer"); + return new ForwardGremlinRestExecutor(config); + } + + public static GremlinRestExecutor create(HugeConfig config, GremlinServer server, + String restConfigFile) throws IOException { + if ("forward".equals(config.get(ServerOptions.GREMLIN_REST_EXECUTION_MODE))) { + return new ForwardGremlinRestExecutor(config); + } + Settings settings = server.getServerGremlinExecutor().getSettings(); + URI target = URI.create(config.get(ServerOptions.GREMLIN_SERVER_URL)); + E.checkArgument("http".equals(target.getScheme()) && target.getHost() != null && + target.getUserInfo() == null && target.getFragment() == null && + InetAddress.getByName(target.getHost()).isLoopbackAddress() && + target.getPort() == settings.port && + (InetAddress.getByName(settings.host).isAnyLocalAddress() || + InetAddress.getByName(settings.host).equals( + InetAddress.getByName(target.getHost()))), + "embedded requires gremlinserver.url to target this local Gremlin Server"); + E.checkArgument(settings.ssl == null || !settings.ssl.enabled, + "embedded does not support a TLS Gremlin listener"); + E.checkArgument(HugeGraphWsAndHttpChannelizer.class.getName().equals(settings.channelizer), + "embedded requires HugeGraphWsAndHttpChannelizer"); + E.checkArgument(settings.authorization == null || + settings.authorization.authorizer == null, + "embedded does not support a custom Gremlin authorizer"); + String restAuth = config.get(ServerOptions.AUTHENTICATOR); + String gremlinAuth = settings.authentication.authenticator; + if (restAuth.isEmpty()) { + E.checkArgument(AllowAllAuthenticator.class.getName().equals(gremlinAuth), + "REST and Gremlin authentication must agree for embedded execution"); + } else { + E.checkArgument(StandardAuthenticator.class.getName().equals(restAuth) && + restAuth.equals(gremlinAuth) && + WsAndHttpBasicAuthHandler.class.getName().equals( + settings.authentication.authenticationHandler), + "embedded only supports matching standard HugeGraph authentication"); + Object tokens = settings.authentication.config == null ? null : + settings.authentication.config.get("tokens"); + E.checkArgument(tokens instanceof String && + Path.of(restConfigFile).toRealPath().equals( + Path.of((String) tokens).toRealPath()), + "embedded requires the same REST and Gremlin authentication config"); + } + return new EmbeddedGremlinRestExecutor(config, server.getServerGremlinExecutor()); + } + + public static void initialize(GremlinRestExecutor executor, GremlinServer server) { + if (executor instanceof EmbeddedGremlinRestExecutor) { + HugeGraphWsAndHttpChannelizer channelizer = + (HugeGraphWsAndHttpChannelizer) server.getChannelizer(); + ((EmbeddedGremlinRestExecutor) executor).initialize(channelizer.jsonSerializer()); + } + } +} diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java index 3441291e4d..1343dce758 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java @@ -111,6 +111,14 @@ public class ServerOptions extends OptionHolder { "http://127.0.0.1:8182" ).withUrlNormalization("http://"); + public static final ConfigOption GREMLIN_REST_EXECUTION_MODE = + new ConfigOption<>( + "gremlinserver.rest_execution_mode", + "Execution path for REST /gremlin: forward or same-process embedded.", + allowValues("forward", "embedded"), + "forward" + ); + public static final ConfigOption GREMLIN_SERVER_TIMEOUT = new ConfigOption<>( "gremlinserver.timeout", diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/ApplicationConfig.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/ApplicationConfig.java index 5cc3e0eb33..8628fa5ca1 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/ApplicationConfig.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/ApplicationConfig.java @@ -18,6 +18,8 @@ package org.apache.hugegraph.server; import org.apache.hugegraph.HugeFactory; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutor; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutorFactory; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.core.GraphManager; import org.apache.hugegraph.define.WorkLoad; @@ -74,6 +76,30 @@ public class ApplicationConfig extends ResourceConfig { private ServletConfig servletConfig; public ApplicationConfig(HugeConfig conf, EventHub hub) { + this(conf, hub, GremlinRestExecutorFactory.standalone(conf)); + } + + public ApplicationConfig(HugeConfig conf, EventHub hub, + GremlinRestExecutor executor) { + register(new AbstractBinder() { + @Override + protected void configure() { + bind(executor).to(GremlinRestExecutor.class); + } + }); + register(new ApplicationEventListener() { + @Override + public void onEvent(ApplicationEvent event) { + if (event.getType() == ApplicationEvent.Type.DESTROY_FINISHED) { + executor.close(); + } + } + + @Override + public RequestEventListener onRequest(RequestEvent event) { + return null; + } + }); packages("org.apache.hugegraph.api"); // Register Jackson to support json diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HugeGraphWsAndHttpChannelizer.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HugeGraphWsAndHttpChannelizer.java index 94f358425f..124c1cc0da 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HugeGraphWsAndHttpChannelizer.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/HugeGraphWsAndHttpChannelizer.java @@ -17,9 +17,12 @@ package org.apache.hugegraph.server; +import org.apache.hugegraph.util.E; import org.apache.tinkerpop.gremlin.server.AbstractChannelizer; import org.apache.tinkerpop.gremlin.server.handler.WsAndHttpChannelizerHandler; import org.apache.tinkerpop.gremlin.server.util.ServerGremlinExecutor; +import org.apache.tinkerpop.gremlin.util.MessageSerializer; +import org.apache.tinkerpop.gremlin.util.ser.MessageTextSerializer; import io.netty.channel.ChannelPipeline; @@ -36,6 +39,15 @@ public void init(ServerGremlinExecutor executor) { this.graphManager, this.settings)); } + public MessageTextSerializer jsonSerializer() { + MessageSerializer serializer = + this.serializers.get("application/json"); + E.checkState( + serializer instanceof MessageTextSerializer, + "embedded requires an application/json MessageTextSerializer"); + return (MessageTextSerializer) serializer; + } + @Override public void configure(ChannelPipeline pipeline) { this.handler.configure(pipeline); diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/RestServer.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/RestServer.java index 403c788632..8265dea3d1 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/RestServer.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/server/RestServer.java @@ -23,6 +23,8 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.Future; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutor; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutorFactory; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.config.ServerOptions; import org.apache.hugegraph.event.EventHub; @@ -49,17 +51,24 @@ public class RestServer { private final HugeConfig conf; private final EventHub eventHub; private HttpServer httpServer = null; + private final GremlinRestExecutor gremlinRestExecutor; public RestServer(HugeConfig conf, EventHub hub) { + this(conf, hub, GremlinRestExecutorFactory.standalone(conf)); + } + + public RestServer(HugeConfig conf, EventHub hub, + GremlinRestExecutor executor) { this.conf = conf; this.eventHub = hub; + this.gremlinRestExecutor = executor; } public void start() throws IOException { String url = this.conf.get(ServerOptions.REST_SERVER_URL); URI uri = UriBuilder.fromUri(url).build(); - ResourceConfig rc = new ApplicationConfig(this.conf, this.eventHub); + ResourceConfig rc = new ApplicationConfig(this.conf, this.eventHub, this.gremlinRestExecutor); this.httpServer = this.configHttpServer(uri, rc); try { @@ -122,6 +131,7 @@ private HttpServer configHttpServer(URI uri, ResourceConfig rc) { } public Future shutdown() { + this.gremlinRestExecutor.beginShutdown(); E.checkNotNull(this.httpServer, "http server"); /* * Since 2.3.x shutdown() won't call shutdownNow(), so the event @@ -169,11 +179,17 @@ public void shutdownNow() { } public static RestServer start(String conf, EventHub hub) throws Exception { + return start(conf, hub, null); + } + + public static RestServer start(String conf, EventHub hub, + GremlinRestExecutor executor) throws Exception { LOG.info("RestServer starting..."); ApiVersion.check(); HugeConfig config = new HugeConfig(conf); - RestServer server = new RestServer(config, hub); + RestServer server = executor == null ? new RestServer(config, hub) : + new RestServer(config, hub, executor); server.start(); LOG.info("RestServer started"); diff --git a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties index be7b82a3b0..b89abd77cc 100644 --- a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties +++ b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties @@ -4,6 +4,9 @@ restserver.url=http://127.0.0.1:8080 #restserver.enable_graphspaces_filter=false # gremlin server url, need to be consistent with host and port in gremlin-server.yaml #gremlinserver.url=127.0.0.1:8182 +# REST /gremlin: forward (default) or embedded for this process only. +# See repository docs/gremlin-rest-execution.md before enabling embedded. +gremlinserver.rest_execution_mode=forward graphs=./conf/graphs diff --git a/hugegraph-server/hugegraph-dist/src/main/java/org/apache/hugegraph/dist/HugeGraphServer.java b/hugegraph-server/hugegraph-dist/src/main/java/org/apache/hugegraph/dist/HugeGraphServer.java index 3cbce0ff31..ec000a3e73 100644 --- a/hugegraph-server/hugegraph-dist/src/main/java/org/apache/hugegraph/dist/HugeGraphServer.java +++ b/hugegraph-server/hugegraph-dist/src/main/java/org/apache/hugegraph/dist/HugeGraphServer.java @@ -17,30 +17,33 @@ package org.apache.hugegraph.dist; -import org.apache.hugegraph.exception.HugeException; +import java.util.concurrent.CompletableFuture; + import org.apache.hugegraph.HugeFactory; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutor; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutorFactory; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.config.ServerOptions; import org.apache.hugegraph.constant.ServiceConstant; import org.apache.hugegraph.core.GraphManager; -import org.apache.hugegraph.query.ConditionQuery; -import org.apache.hugegraph.type.HugeType; import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.exception.HugeException; import org.apache.hugegraph.meta.MetaManager; import org.apache.hugegraph.meta.PdMetaDriver; +import org.apache.hugegraph.query.ConditionQuery; import org.apache.hugegraph.server.RestServer; +import org.apache.hugegraph.type.HugeType; import org.apache.hugegraph.util.ConfigUtil; import org.apache.hugegraph.util.Log; import org.apache.logging.log4j.LogManager; import org.apache.tinkerpop.gremlin.server.GremlinServer; import org.slf4j.Logger; -import java.util.concurrent.CompletableFuture; - public class HugeGraphServer { private static final Logger LOG = Log.logger(HugeGraphServer.class); + private final GremlinRestExecutor gremlinRestExecutor; private final RestServer restServer; private final GremlinServer gremlinServer; private final MemoryMonitor memoryMonitor; @@ -61,6 +64,7 @@ public HugeGraphServer(String gremlinServerConf, String restServerConf) GremlinServer preparedGremlinServer = null; GremlinServer gremlinServer = null; MemoryMonitor memoryMonitor = null; + GremlinRestExecutor restExecutor = null; try { initializeQueryClasses(); ConfigUtil.checkGremlinConfig(gremlinServerConf); @@ -99,9 +103,14 @@ public HugeGraphServer(String gremlinServerConf, String restServerConf) try { // Start HugeRestServer (loads graphs from PD; events go to // ContextGremlinServer which is already listening) - restServer = HugeRestServer.start(restServerConf, hub); + restExecutor = GremlinRestExecutorFactory.create( + restServerConfig, preparedGremlinServer, restServerConf); + restServer = RestServer.start(restServerConf, hub, restExecutor); } catch (Throwable e) { LOG.error("HugeRestServer start error: ", e); + if (restExecutor != null) { + restExecutor.close(); + } stopPreparedGremlinServer(preparedGremlinServer); throw e; } @@ -110,8 +119,11 @@ public HugeGraphServer(String gremlinServerConf, String restServerConf) // Now start the pre-prepared GremlinServer gremlinServer = HugeGremlinServer.startPrepared( preparedGremlinServer); + GremlinRestExecutorFactory.initialize( + restExecutor, gremlinServer); } catch (Throwable e) { LOG.error("HugeGremlinServer start error: ", e); + stopPreparedGremlinServer(preparedGremlinServer); try { restServer.shutdown().get(); } catch (Throwable t) { @@ -126,12 +138,14 @@ public HugeGraphServer(String gremlinServerConf, String restServerConf) memoryMonitor.start(); } catch (Throwable e) { LOG.error("MemoryMonitor start error: ", e); + restExecutor.beginShutdown(); rollbackStartup(memoryMonitor, gremlinServer, restServer); throw e; } } finally { System.setSecurityManager(securityManager); } + this.gremlinRestExecutor = restExecutor; this.restServer = restServer; this.gremlinServer = gremlinServer; this.memoryMonitor = memoryMonitor; @@ -177,6 +191,7 @@ static void rollbackStartup(MemoryMonitor monitor, } public void stop() { + this.gremlinRestExecutor.beginShutdown(); this.memoryMonitor.stop(); try { diff --git a/hugegraph-server/hugegraph-test/scripts/gremlin-rest-benchmark.py b/hugegraph-server/hugegraph-test/scripts/gremlin-rest-benchmark.py new file mode 100644 index 0000000000..37720a95b6 --- /dev/null +++ b/hugegraph-server/hugegraph-test/scripts/gremlin-rest-benchmark.py @@ -0,0 +1,82 @@ +#!/usr/bin/env python3 +# +# 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. +#/ + +"""Run the script-only REST transport comparison after the focused Maven tests.""" + +import argparse +import datetime +import json +import os +from pathlib import Path +import platform +import shutil +import subprocess +import xml.etree.ElementTree as ET + + +def main(): + module = Path(__file__).resolve().parents[1] + repo = module.parents[1] + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, + default=module / "target/gremlin-rest-benchmark.json") + args = parser.parse_args() + output = args.output.resolve() + output.parent.mkdir(parents=True, exist_ok=True) + report = module / "target/surefire-reports/TEST-org.apache.hugegraph.auth.EmbeddedGremlinRestExecutorTest.xml" + if not report.exists(): + parser.error("Run EmbeddedGremlinRestExecutorTest with Maven first") + properties = {entry.attrib["name"]: entry.attrib["value"] + for entry in ET.parse(report).getroot().find("properties")} + java = str(Path(os.environ["JAVA_HOME"]) / "bin/java") if "JAVA_HOME" in os.environ else shutil.which("java") + if not java: + parser.error("Java 17 is required; set JAVA_HOME") + version = subprocess.check_output([java, "-version"], stderr=subprocess.STDOUT, text=True) + if 'version "17.' not in version: + parser.error("Use Java 17 to match the supported Server runtime") + started = datetime.datetime.now(datetime.timezone.utc).isoformat() + subprocess.run([ + java, + "@" + str(repo / "hugegraph-server/hugegraph-dist/src/assembly/static/bin/jvm-module.options"), + "@" + str(module / "conf/jvm-test-module.options"), + "-cp", properties["java.class.path"], + "org.apache.hugegraph.auth.EmbeddedGremlinRestExecutorTest", str(output), + ], cwd=module, check=True) + data = json.loads(output.read_text()) + data["environment"] = { + "started_at_utc": started, + "base_commit": subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=repo, text=True).strip(), + "working_tree_dirty": bool(subprocess.check_output(["git", "status", "--porcelain"], cwd=repo)), + "architecture": platform.machine(), + "platform": platform.platform(), + "authentication": "anonymous", + "gremlin_workers": 1, + "rest_workers_per_mode": 2, + "clients": 1, + "warmup_requests_per_mode_per_script": 100, + "rounds": 4, + "requests_per_mode_per_round": 200, + "ordering": "alternate forward/embedded order across rounds", + "limitations": "same-host single-JVM script-only comparison; no backend I/O, CPU isolation or production load", + } + output.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n") + print(f"Saved correctness-checked samples to {output}") + + +if __name__ == "__main__": + main() diff --git a/hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py b/hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py new file mode 100644 index 0000000000..21da107d91 --- /dev/null +++ b/hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py @@ -0,0 +1,363 @@ +#!/usr/bin/env python3 +# 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. + +"""Linux standalone RocksDB comparison; owns only the supplied NEW output directory. + +Uses Python's standard library. Closed-loop clients validate every response. +ABBA restarts use the same distribution, persistent dataset and JVM settings. +Server CPU affinity: 0-5; load generator: 6-7. Requires at least eight CPUs. +""" + +import argparse +import concurrent.futures +import gzip +import hashlib +import http.client +import json +import math +import os +from pathlib import Path +import platform +import resource +import shutil +import socket +import subprocess +import threading +import time +import urllib.parse + + +def save(path, value): + path.write_text(json.dumps(value, indent=2, ensure_ascii=False) + "\n") + + +def free_port(): + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +class Client: + def __init__(self, port): + self.connection = http.client.HTTPConnection("127.0.0.1", port, timeout=40) + + def request(self, script, bindings=None, aliases=None, get=False): + payload = {"gremlin": script, "language": "gremlin-groovy", + "bindings": bindings or {}, "aliases": aliases or {"g": "DEFAULT-hugegraph"}} + if get: + query = {"gremlin": script, "language": "gremlin-groovy"} + query.update({"bindings." + key: value for key, value in payload["bindings"].items()}) + query.update({"aliases." + key: value for key, value in payload["aliases"].items()}) + self.connection.request("GET", "/gremlin?" + urllib.parse.urlencode(query)) + else: + self.connection.request("POST", "/gremlin", json.dumps(payload), + {"Content-Type": "application/json"}) + response = self.connection.getresponse() + raw = response.read() + return response.status, json.loads(raw), len(raw), response.getheader("Content-Type") + + def success(self, script, bindings=None, get=False): + status, body, _, _ = self.request(script, bindings, get=get) + assert status == 200, (status, body) + assert body["status"]["code"] == 200, body + return body["result"]["data"] + + def close(self): + self.connection.close() + + +def percentile(values, quantile): + return values[max(0, math.ceil(len(values) * quantile) - 1)] + + +def clear_records(client): + deleted = 0 + while True: + batch = client.success("ids=g.traversal().V().hasLabel('record').not(hasId(999998L))" + ".limit(1000).id().toList();" + "if(!ids.isEmpty()){g.traversal().V(ids).drop().iterate()};ids.size()")[0] + deleted += batch + if batch == 0: + return deleted + + +def proc_sample(pid): + fields = Path(f"/proc/{pid}/stat").read_text().split() + return {"cpu_ticks": int(fields[13]) + int(fields[14]), + "rss_bytes": int(fields[23]) * os.sysconf("SC_PAGE_SIZE"), + "threads": int(fields[19]), "host_load": os.getloadavg(), + "host_cpu": Path("/proc/stat").read_text().splitlines()[0], + "pressure": Path("/proc/pressure/cpu").read_text().strip()} + + +def workload(name, n): + cases = { + "constant": ("1+2", lambda i: {}, lambda data, i: data == [3]), + "point": ("g.traversal().V(vid).values('name')", + lambda i: {"vid": i % n}, lambda data, i: data == [f"p{i % n}"]), + "one_hop": ("g.traversal().V(vid).out('knows').id().order()", + lambda i: {"vid": i % n}, + lambda data, i: data == sorted([(i % n + j) % n for j in range(1, 6)])), + "two_hop": ("g.traversal().V(vid).out('knows').out('knows').id().order()", + lambda i: {"vid": i % n}, + lambda data, i: data == sorted([(i % n + j + k) % n + for j in range(1, 6) for k in range(1, 6)])), + "result_1000": ("g.traversal().V(ids).values('name').order()", + lambda i: {"ids": list(range(1000))}, + lambda data, i: data == sorted([f"p{j}" for j in range(1000)])), + "write": ("g.addVertex(org.apache.tinkerpop.gremlin.structure.T.label,'record'," + "org.apache.tinkerpop.gremlin.structure.T.id,vid,'payload',payload);vid", + lambda i: {"vid": 1000000 + i, "payload": "x" * 64}, + lambda data, i: data == [1000000 + i]), + } + return cases[name] + + +def measure(port, pid, name, n, concurrency, duration, sequence, warmup=False): + script, binding, validate = workload(name, n) + barrier = threading.Barrier(concurrency + 1) + stop = threading.Event() + + def worker(index): + client = Client(port) + samples, sizes, failures = [], [], [] + counter = sequence + index + barrier.wait() + try: + while not stop.is_set(): + started = time.perf_counter_ns() + try: + status, body, length, media = client.request(script, binding(counter)) + data = body.get("result", {}).get("data") + assert status == 200 and body["status"]["code"] == 200, (status, body) + assert media.startswith("application/json"), media + assert validate(data, counter), (name, counter, data) + samples.append(time.perf_counter_ns() - started) + sizes.append(length) + except Exception as error: + failures.append(repr(error)[:1000]) + stop.set() + counter += concurrency + finally: + client.close() + return samples, sizes, failures + + before = proc_sample(pid) + usage = resource.getrusage(resource.RUSAGE_SELF) + with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = [pool.submit(worker, i) for i in range(concurrency)] + started = time.monotonic() + barrier.wait() + stop.wait(duration) + stop.set() + results = [future.result() for future in futures] + elapsed = time.monotonic() - started + after = proc_sample(pid) + usage_after = resource.getrusage(resource.RUSAGE_SELF) + samples = sorted(sample for result in results for sample in result[0]) + failures = [error for result in results for error in result[2]] + assert samples and not failures, (name, failures) + return {"workload": name, "concurrency": concurrency, "warmup": warmup, + "seconds": elapsed, "completed": len(samples), "errors": failures, + "throughput": len(samples) / elapsed, + "p50_ms": percentile(samples, .50) / 1000000, + "p95_ms": percentile(samples, .95) / 1000000, + "p99_ms": percentile(samples, .99) / 1000000, + "response_bytes_mean": sum(sum(result[1]) for result in results) / len(samples), + "server_cpu_cores": (after["cpu_ticks"] - before["cpu_ticks"]) / + os.sysconf("SC_CLK_TCK") / elapsed, + "client_cpu_cores": (usage_after.ru_utime + usage_after.ru_stime - + usage.ru_utime - usage.ru_stime) / elapsed, + "before": before, "after": after, "latency_ns": samples} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--distribution", type=Path, required=True) + parser.add_argument("--java-home", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--vertices", type=int, default=10000) + parser.add_argument("--seconds", type=float, default=3) + parser.add_argument("--warmup-seconds", type=float, default=5) + args = parser.parse_args() + assert args.vertices >= 1000 + assert set(range(8)).issubset(os.sched_getaffinity(0)) + args.distribution = args.distribution.resolve() + args.java_home = args.java_home.resolve() + args.output = args.output.resolve() + args.output.mkdir(parents=True, exist_ok=False) + os.sched_setaffinity(0, {6, 7}) + dist, output = args.distribution, args.output + java = str(args.java_home / "bin/java") + assert (args.java_home / "conf/security/java.security").is_file(), "Incomplete Java 17 configuration" + version = subprocess.check_output([java, "-version"], stderr=subprocess.STDOUT, text=True) + assert 'version "17.' in version, version + cp = ":".join(str(p) for p in sorted((dist / "lib").glob("hugegraph*.jar"))) + cp += ":" + ":".join(str(p) for p in sorted((dist / "lib").glob("*.jar")) + if not p.name.startswith("hugegraph")) + (output / "graphs").mkdir() + (output / "graphs/hugegraph.properties").write_text( + "gremlin.graph=org.apache.hugegraph.HugeFactory\nbackend=rocksdb\nserializer=binary\n" + f"store=hugegraph\nrocksdb.data_path={output}/data\nrocksdb.wal_path={output}/data\n" + "vertex.cache_capacity=10000\nedge.cache_capacity=50000\n") + shutil.copytree(dist / "scripts", output / "scripts") + rp, gp = free_port(), free_port() + template = (dist / "conf/gremlin-server.yaml").read_text() + (output / "gremlin.yaml").write_text( + f"host: 127.0.0.1\nport: {gp}\ngremlinPool: 4\nthreadPoolBoss: 1\nthreadPoolWorker: 4\n" + + template) + common = (f"graphs={output}/graphs\nrestserver.url=http://127.0.0.1:{rp}\n" + f"gremlinserver.url=http://127.0.0.1:{gp}\nrestserver.max_worker_threads=16\n" + "gremlinserver.max_route=16\ngremlinserver.timeout=30\nmemory_monitor.threshold=1\n") + for mode in ("forward", "embedded"): + (output / f"{mode}.properties").write_text(common + f"gremlinserver.rest_execution_mode={mode}\n") + jvm = [java, "-Xms1g", "-Xmx2g", "-XX:ActiveProcessorCount=6", + "@" + str(dist / "bin/jvm-module.options"), + "-Dlog4j.configurationFile=" + str(dist / "conf/log4j2.xml")] + env = {**os.environ, "JAVA_HOME": str(args.java_home)} + affinity = lambda: os.sched_setaffinity(0, set(range(6))) + with (output / "init.log").open("w") as log: + subprocess.run(jvm + ["-cp", cp, "org.apache.hugegraph.cmd.InitStore", + str(output / "forward.properties")], cwd=output, env=env, + stdout=log, stderr=subprocess.STDOUT, check=True, timeout=300, + preexec_fn=affinity) + result = {"environment": {"host": platform.node(), "platform": platform.platform(), + "java": version, "cpus": os.cpu_count(), "server_cpus": list(range(6)), + "client_cpus": [6, 7], "load_model": "closed-loop, same host", + "cache": "warm application/RocksDB/OS; no cache dropping", + "vertices": args.vertices, "edges": args.vertices * 5, + "duration_seconds": args.seconds, "mode_order": ["forward", "embedded", + "embedded", "forward"], + "warmup_seconds": args.warmup_seconds, + "distribution_sha256": { + p.name: hashlib.sha256(p.read_bytes()).hexdigest() + for p in (dist / "lib").glob("hugegraph*.jar")}}, + "runs": []} + n, sequence = args.vertices, 0 + for run_id, mode in enumerate(result["environment"]["mode_order"]): + run_dir = output / f"run-{run_id}-{mode}" + run_dir.mkdir() + with (run_dir / "server.log").open("w") as log: + process = subprocess.Popen( + jvm + [f"-Xlog:gc*:file={run_dir}/gc.log:time,uptime,level,tags", + "-cp", cp, "org.apache.hugegraph.dist.HugeGraphServer", + str(output / "gremlin.yaml"), str(output / f"{mode}.properties")], + cwd=output, env=env, stdout=log, stderr=subprocess.STDOUT, preexec_fn=affinity) + run = {"id": run_id, "mode": mode, "checks": {}, "measurements": []} + try: + deadline = time.monotonic() + 300 + while True: + assert process.poll() is None, f"Server exited; see {run_dir}/server.log" + try: + client = Client(rp) + assert client.success("1+2") == [3] + break + except (OSError, AssertionError, json.JSONDecodeError): + client.close() + if time.monotonic() > deadline: + raise + time.sleep(.5) + if run_id == 0: + client.success("s=g.schema();s.propertyKey('name').asText().ifNotExist().create();" + "s.propertyKey('payload').asText().ifNotExist().create();" + "s.vertexLabel('person').useCustomizeNumberId().properties('name')" + ".ifNotExist().create();" + "s.vertexLabel('record').useCustomizeNumberId().properties('payload')" + ".ifNotExist().create();" + "s.edgeLabel('knows').sourceLabel('person').targetLabel('person')" + ".ifNotExist().create();'schema-ok'") + for begin in range(0, n, 200): + client.success("(begin..g.addVertex(" + "org.apache.tinkerpop.gremlin.structure.T.label,'person'," + "org.apache.tinkerpop.gremlin.structure.T.id,i.longValue()," + "'name','p'+i)};end", + {"begin": begin, "end": min(begin + 200, n)}) + for begin in range(0, n, 100): + client.success("(begin..v=g.traversal().V(i).next();" + "(1..5).each{j->v.addEdge('knows'," + "g.traversal().V((i+j)%n).next())}};end", + {"begin": begin, "end": min(begin + 100, n), "n": n}) + print("dataset loaded", flush=True) + assert client.success("g.traversal().V().hasLabel('person').count()") == [n] + assert client.success("g.traversal().E().count()") == [n * 5] + clear_records(client) + assert client.success("x+y", {"x": 2, "y": 5}) == [7] + assert client.success("1+2", get=True) == [3] + nested = client.success("[['text':'中文','items':[null,1,true]]]") + assert nested == [{"text": "中文", "items": [None, 1, True]}], nested + status, body, _, _ = client.request("1", aliases={"g": "missing_graph"}) + assert status == 400, (status, body) + run["checks"]["unknown_alias"] = {"status": status, "body": body} + status, body, _, _ = client.request( + "g.addVertex(org.apache.tinkerpop.gremlin.structure.T.label,'record'," + "org.apache.tinkerpop.gremlin.structure.T.id,999999L,'payload','rollback');" + "throw new IllegalArgumentException('rollback')") + assert status == 400, (status, body) + assert client.success("g.traversal().V(999999L).count()") == [0] + run["checks"]["rollback"] = {"status": status, "exception": body.get("exception")} + if run_id > 0: + assert client.success("g.traversal().V(999998L).values('payload')") == ["persisted"] + else: + client.success("g.addVertex(org.apache.tinkerpop.gremlin.structure.T.label,'record'," + "org.apache.tinkerpop.gremlin.structure.T.id,999998L," + "'payload','persisted');true") + run["checks"]["get_post_bindings_json_alias_rollback_persistence"] = "passed" + client.close() + for name in ("constant", "point", "one_hop", "two_hop", "result_1000", "write"): + warm = measure(rp, process.pid, name, n, 4, args.warmup_seconds, sequence, warmup=True) + sequence += 1000000 + written = warm["completed"] if name == "write" else 0 + for concurrency in (1, 8, 32): + measurement = measure(rp, process.pid, name, n, concurrency, + args.seconds, sequence) + sequence += 1000000 + if name == "write": + written += measurement["completed"] + raw = measurement.pop("latency_ns") + sample_file = f"{name}-c{concurrency}-latency-ns.json.gz" + with gzip.open(run_dir / sample_file, "wt") as sample: + json.dump(raw, sample) + measurement["samples"] = f"{run_dir.name}/{sample_file}" + run["measurements"].append(measurement) + save(output / "progress.json", {"run": run_id, "mode": mode, + "last": measurement}) + print(json.dumps({"run": run_id, "mode": mode, "workload": name, + "concurrency": concurrency, "n": measurement["completed"], + "rps": round(measurement["throughput"], 1), + "p95_ms": round(measurement["p95_ms"], 3)}), flush=True) + if name == "write": + client = Client(rp) + assert client.success("g.traversal().V().hasLabel('record').count()") == [written + 1] + run["checks"]["committed_writes"] = written + assert clear_records(client) == written + client.close() + finally: + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=90) + run["shutdown_exit"] = process.returncode + except subprocess.TimeoutExpired: + process.kill() + process.wait() + raise RuntimeError(f"Shutdown timed out: {run_dir}") + result["runs"].append(run) + save(output / "results.json", result) + print("all runs completed: " + str(output / "results.json"), flush=True) + + +if __name__ == "__main__": + main() diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java new file mode 100644 index 0000000000..1e44ce31be --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java @@ -0,0 +1,848 @@ +/* + * 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.hugegraph.auth; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.hugegraph.HugeFactory; +import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.api.filter.AuthenticationFilter; +import org.apache.hugegraph.api.filter.CompressInterceptor; +import org.apache.hugegraph.api.filter.ExceptionFilter; +import org.apache.hugegraph.api.gremlin.EmbeddedGremlinRestExecutor; +import org.apache.hugegraph.api.gremlin.ForwardGremlinRestExecutor; +import org.apache.hugegraph.api.gremlin.GremlinAPI; +import org.apache.hugegraph.api.gremlin.GremlinClient; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutor; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutorFactory; +import org.apache.hugegraph.config.HugeConfig; +import org.apache.hugegraph.config.OptionSpace; +import org.apache.hugegraph.config.ServerOptions; +import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.dist.RegisterUtil; +import org.apache.hugegraph.event.EventHub; +import org.apache.hugegraph.server.HugeGraphWsAndHttpChannelizer; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.util.Events; +import org.apache.hugegraph.util.JsonUtil; +import org.apache.tinkerpop.gremlin.server.Settings; +import org.apache.tinkerpop.gremlin.server.channel.WsAndHttpChannelizer; +import org.apache.tinkerpop.gremlin.structure.util.empty.EmptyGraph; +import org.apache.tinkerpop.gremlin.util.message.ResponseMessage; +import org.apache.tinkerpop.gremlin.util.ser.GraphSONUntypedMessageSerializerV1; +import org.apache.tinkerpop.gremlin.util.ser.MessageTextSerializer; +import org.apache.tinkerpop.gremlin.util.ser.SerializationException; +import org.glassfish.grizzly.http.server.HttpServer; +import org.glassfish.grizzly.http.server.NetworkListener; +import org.glassfish.grizzly.threadpool.ThreadPoolConfig; +import org.glassfish.hk2.utilities.binding.AbstractBinder; +import org.glassfish.jersey.grizzly2.httpserver.GrizzlyHttpServerFactory; +import org.glassfish.jersey.server.ResourceConfig; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.mockito.Mockito; + +import io.netty.buffer.ByteBufAllocator; +import io.netty.channel.Channel; +import jakarta.ws.rs.ProcessingException; +import jakarta.ws.rs.ServiceUnavailableException; +import jakarta.ws.rs.core.MultivaluedHashMap; +import jakarta.ws.rs.core.MultivaluedMap; +import jakarta.ws.rs.core.Response; + +public class EmbeddedGremlinRestExecutorTest { + + private static final String ID = "6a46b56c-5511-491d-8908-aeb1d8185a81"; + private static ContextGremlinServer server; + private static HugeConfig config; + private static GremlinClient forward; + private static EmbeddedGremlinRestExecutor embedded; + private static Settings settings; + private static EventHub hub; + + @BeforeClass + public static void setup() throws Exception { + OptionSpace.register("server", ServerOptions.instance()); + settings = new Settings(); + settings.host = "127.0.0.1"; + settings.port = 0; + settings.gremlinPool = 1; + settings.threadPoolBoss = 1; + settings.threadPoolWorker = 1; + settings.evaluationTimeout = 5000; + settings.channelizer = WsAndHttpChannelizer.class.getName(); + Settings.SerializerSettings json = new Settings.SerializerSettings(); + json.className = GraphSONUntypedMessageSerializerV1.class.getName(); + json.config = Map.of("ioRegistries", Arrays.asList( + "org.apache.hugegraph.io.HugeGraphIoRegistry", + "org.apache.hugegraph.io.HugeGraphSONV1MessageIoRegistry")); + settings.serializers = Collections.singletonList(json); + hub = new EventHub("embedded-http-test", 1); + server = new ContextGremlinServer(settings, hub); + server.getServerGremlinExecutor().getGraphManager().putGraph("known", EmptyGraph.instance()); + server.start().get(30, TimeUnit.SECONDS); + Channel listening = Whitebox.getInternalState(server, "serverSocketChannel"); + settings.port = ((InetSocketAddress) listening.localAddress()).getPort(); + config = new HugeConfig(Map.of( + "gremlinserver.url", "http://127.0.0.1:" + settings.port, + "gremlinserver.rest_execution_mode", "embedded", + "gremlinserver.timeout", "2", "gremlinserver.max_route", "2")); + embedded = (EmbeddedGremlinRestExecutor) GremlinRestExecutorFactory.create(config, server, ""); + GremlinRestExecutorFactory.initialize(embedded, server); + forward = new GremlinClient(config.get(ServerOptions.GREMLIN_SERVER_URL), 2000, 2, 2); + HugeGraphAuthProxy.resetContext(); + } + + @AfterClass + public static void teardown() throws Exception { + if (embedded != null) { + embedded.close(); + } + if (forward != null) { + forward.close(); + } + if (server != null) { + server.stop().get(30, TimeUnit.SECONDS); + } + HugeGraphAuthProxy.resetContext(); + } + + @Test + public void testPostWireContract() { + for (String script : Arrays.asList("1+2", "[]", "[null,true,1,2.5,'中文']", + "[a:1,b:[2,3]]", "'hello'", "null")) { + comparePost(body(script)); + } + comparePost("{\"requestId\":\"" + ID + "\",\"gremlin\":\"x\",\"bindings\":" + + "{\"x\":{\"a\":[null,true,123,1.25]}}}"); + } + + @Test + public void testGetEncodingAndStringBindings() { + for (String value : Arrays.asList("a+b c%20d", "中文&=?#", "%7Bx%7D/[y]", "a\\b\"c")) { + MultivaluedMap params = new MultivaluedHashMap<>(); + params.add("gremlin", "x"); + params.add("bindings.x", value); + params.add("requestId", ID); + compare(forward.doGetRequest(null, params), embedded.get(null, params, "test")); + } + } + + @Test + public void testUnresolvedGetTemplateMatchesForward() { + MultivaluedMap params = new MultivaluedHashMap<>(); + params.add("gremlin", "x"); + params.add("bindings.x", "{x}"); + Assert.assertThrows(IllegalStateException.class, + () -> forward.doGetRequest(null, params)); + Assert.assertThrows(IllegalStateException.class, + () -> embedded.get(null, params, "test")); + } + + @Test + public void testErrorsAndAliases() { + for (String request : Arrays.asList("{}", "{", "null", "", "[]", + "{\"gremlin\":1}", "{\"gremlin\":null}", + "{\"gremlin\":\"1\",\"bindings\":[]}", + "{\"gremlin\":\"1\",\"requestId\":\"bad\"}", + "{\"requestId\":\"" + ID + "\",\"gremlin\":\"1\",\"aliases\":{\"g\":\"missing\"}}", + body("throw new IllegalArgumentException('bad')"), + body("throw new SecurityException('denied')"), + body("doesNotExist()"))) { + comparePost(request); + } + comparePost(body("42")); + } + + @Test + public void testRequestIdGeneratedWhenAbsent() { + try (Response response = embedded.post(null, "{\"gremlin\":\"1\"}", "test")) { + Map result = json(response); + Assert.assertNotNull(UUID.fromString((String) result.get("requestId"))); + } + } + + @Test + public void testDuplicateGetParameter() { + MultivaluedMap params = new MultivaluedHashMap<>(); + params.add("gremlin", "1"); + params.add("unrecognized", "a"); + params.add("unrecognized", "b"); + Assert.assertThrows(IllegalArgumentException.class, + () -> embedded.get(null, params, "test")); + Assert.assertThrows(IllegalArgumentException.class, + () -> forward.doGetRequest(null, params)); + } + + @Test + public void testSharedSerializerAndStartupStopGuards() { + EmbeddedGremlinRestExecutor pending = new EmbeddedGremlinRestExecutor( + config, server.getServerGremlinExecutor()); + Assert.assertThrows(ServiceUnavailableException.class, + () -> pending.post(null, body("1"), "test")); + GremlinRestExecutorFactory.initialize(pending, server); + Assert.assertSame(((HugeGraphWsAndHttpChannelizer) server.getChannelizer()).jsonSerializer(), + Whitebox.getInternalState(pending, "serializer")); + pending.beginShutdown(); + Assert.assertThrows(ServiceUnavailableException.class, + () -> pending.post(null, body("1"), "test")); + pending.close(); + pending.close(); + comparePost(body("1")); + } + + @Test + public void testFactoryRejectsRemoteAndMismatchedAuthentication() { + HugeConfig remote = new HugeConfig(Map.of( + "gremlinserver.url", "http://192.0.2.1:8182", + "gremlinserver.rest_execution_mode", "embedded")); + Assert.assertThrows(IllegalArgumentException.class, + () -> GremlinRestExecutorFactory.create(remote, server, "")); + HugeConfig auth = new HugeConfig(Map.of( + "gremlinserver.url", config.get(ServerOptions.GREMLIN_SERVER_URL), + "gremlinserver.rest_execution_mode", "embedded", + "auth.authenticator", StandardAuthenticator.class.getName())); + Assert.assertThrows(IllegalArgumentException.class, + () -> GremlinRestExecutorFactory.create(auth, server, "")); + } + + @Test + public void testReadTimeoutDoesNotCancelExecution() throws Exception { + AtomicInteger writes = new AtomicInteger(); + Whitebox.invoke(server.getServerGremlinExecutor().getGremlinExecutor(), "globalBindings", + new Class[]{String.class, Object.class}, "put", "writes", writes); + HugeConfig shortWait = new HugeConfig(Map.of( + "gremlinserver.url", config.get(ServerOptions.GREMLIN_SERVER_URL), + "gremlinserver.timeout", "1", "gremlinserver.max_route", "1")); + EmbeddedGremlinRestExecutor timed = new EmbeddedGremlinRestExecutor( + shortWait, server.getServerGremlinExecutor()); + GremlinRestExecutorFactory.initialize(timed, server); + try { + Assert.assertThrows(ProcessingException.class, () -> timed.post(null, + body("Thread.sleep(1300); writes.incrementAndGet()"), "test")); + // Queued on the same worker: completion proves the timed-out script ran once. + comparePost(body("writes.get()")); + Assert.assertEquals(1, writes.get()); + } finally { + timed.close(); + } + } + + @Test + public void testWorkerUsesSubmittingUserAndClearsItAfterFailure() { + try { + for (String name : Arrays.asList("alice", "bob", "alice")) { + HugeGraphAuthProxy.setContext(new HugeGraphAuthProxy.Context( + new HugeAuthenticator.User(name, RolePermission.admin()))); + try (Response response = embedded.post(null, body( + "org.apache.hugegraph.auth.HugeGraphAuthProxy.getContext().user().username()"), + "test")) { + Assert.assertEquals(Collections.singletonList(name), + ((Map) json(response).get("result")).get("data")); + } + try (Response ignored = embedded.post(null, body("throw new RuntimeException('x')"), + "test")) { + Assert.assertEquals(500, ignored.getStatus()); + } + } + HugeGraphAuthProxy.resetContext(); + try (Response response = embedded.post(null, body( + "org.apache.hugegraph.auth.HugeGraphAuthProxy.getContext() == null"), "test")) { + Assert.assertEquals(Collections.singletonList(true), + ((Map) json(response).get("result")).get("data")); + } + } finally { + HugeGraphAuthProxy.resetContext(); + } + } + + @Test + public void testProtocolSizeLimits() { + for (int size : new int[]{settings.maxContentLength - 256, + settings.maxContentLength + 256}) { + String request = JsonUtil.toJson(Map.of("gremlin", "1", "padding", "x".repeat(size))); + try (Response left = forward.doPostRequest(null, request); + Response right = embedded.post(null, request, "test")) { + Assert.assertEquals(left.getStatus(), right.getStatus()); + } + } + MultivaluedMap params = new MultivaluedHashMap<>(); + params.add("gremlin", "'" + "x".repeat(settings.maxInitialLineLength + 10) + "'"); + try (Response left = forward.doGetRequest(null, params); + Response right = embedded.get(null, params, "test")) { + Assert.assertEquals(left.getStatus(), right.getStatus()); + } + } + + @Test + public void testHeaderLimitBoundary() throws Exception { + int low = 0; + int high = settings.maxHeaderSize; + while (low + 1 < high) { + int middle = (low + high) / 2; + // A decoder-rejected request can leave extra bytes on the old + // connection. Use a fresh connection to measure the actual limit. + GremlinClient client = new GremlinClient(config.get(ServerOptions.GREMLIN_SERVER_URL), + 2000, 1, 1); + try (Response response = client.doPostRequest("x".repeat(middle), body("1"))) { + if (response.getStatus() == 200) { + low = middle; + } else { + high = middle; + } + } finally { + client.close(); + } + } + for (int size : new int[]{low, high}) { + GremlinClient client = new GremlinClient(config.get(ServerOptions.GREMLIN_SERVER_URL), + 2000, 1, 1); + try { + compare(client.doPostRequest("x".repeat(size), body("1")), + embedded.post("x".repeat(size), body("1"), "test")); + } finally { + client.close(); + } + } + } + + @Test + public void testOuterRestResponseContract() throws Exception { + ForwardGremlinRestExecutor proxy = new ForwardGremlinRestExecutor(config); + HttpServer left = rest(proxy); + HttpServer right = rest(embedded); + try { + HttpClient client = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build(); + for (String request : Arrays.asList(body("1+2"), "{}", + body("throw new IllegalArgumentException('bad')"), + body("throw new SecurityException('denied')"))) { + HttpResponse expected = client.send(post(left, request), + HttpResponse.BodyHandlers.ofString()); + HttpResponse actual = client.send(post(right, request), + HttpResponse.BodyHandlers.ofString()); + Assert.assertEquals(expected.statusCode(), actual.statusCode()); + Assert.assertEquals(expected.headers().firstValue("content-type"), + actual.headers().firstValue("content-type")); + Assert.assertEquals(expected.body(), actual.body()); + } + } finally { + left.shutdownNow(); + right.shutdownNow(); + proxy.close(); + } + } + + @Test + public void testRestAuthenticatesBasicAndBearerOnce() throws Exception { + GraphManager manager = + Mockito.mock(GraphManager.class); + Mockito.when(manager.requireAuthentication()).thenReturn(true); + AtomicInteger authentications = new AtomicInteger(); + Mockito.when(manager.authenticate(Mockito.anyMap())).thenAnswer(call -> { + authentications.incrementAndGet(); + Map credentials = call.getArgument(0); + String name = credentials.containsKey(HugeAuthenticator.KEY_TOKEN) ? "bearer" : "basic"; + HugeAuthenticator.User user = new HugeAuthenticator.User(name, RolePermission.admin()); + HugeGraphAuthProxy.setContext(new HugeGraphAuthProxy.Context(user)); + return user; + }); + HttpServer rest = rest(embedded, manager); + try { + HttpClient client = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build(); + int port = rest.getListeners().iterator().next().getPort(); + for (String credential : Arrays.asList("Basic dXNlcjpwYXNz", "Bearer test-token")) { + HttpRequest request = HttpRequest.newBuilder(URI.create( + "http://127.0.0.1:" + port + "/gremlin")) + .timeout(Duration.ofSeconds(10)) + .header("Content-Type", "application/json") + .header("Authorization", credential) + .POST(HttpRequest.BodyPublishers.ofString(body( + "org.apache.hugegraph.auth.HugeGraphAuthProxy.getContext().user().username()"))) + .build(); + HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); + Assert.assertEquals(response.body(), 200, response.statusCode()); + Map result = JsonUtil.fromJson(response.body(), Map.class); + Assert.assertEquals(Collections.singletonList( + credential.startsWith("Basic") ? "basic" : "bearer"), + ((Map) result.get("result")).get("data")); + } + Assert.assertEquals(2, authentications.get()); + } finally { + rest.shutdownNow(); + } + } + + private static HttpRequest post(HttpServer server, String body) { + int port = server.getListeners().iterator().next().getPort(); + return HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + port + "/gremlin")) + .timeout(Duration.ofSeconds(10)) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(body)).build(); + } + + private static HttpServer rest(GremlinRestExecutor executor) throws Exception { + return rest(executor, null); + } + + private static HttpServer rest(GremlinRestExecutor executor, + GraphManager manager) throws Exception { + ResourceConfig resources = new ResourceConfig(GremlinAPI.class, + ExceptionFilter.HugeGremlinExceptionMapper.class, + ExceptionFilter.UnknownExceptionMapper.class, + ExceptionFilter.IllegalArgumentExceptionMapper.class, + CompressInterceptor.class); + if (manager != null) { + resources.register(AuthenticationFilter.class); + } + resources.register(new AbstractBinder() { + @Override + protected void configure() { + bind(executor).to(GremlinRestExecutor.class); + bind(config).to(HugeConfig.class); + if (manager != null) { + bind(manager).to(GraphManager.class); + } + } + }); + HttpServer server = GrizzlyHttpServerFactory.createHttpServer( + URI.create("http://127.0.0.1:0"), resources, false); + NetworkListener listener = server.getListeners().iterator().next(); + listener.getTransport().setSelectorRunnersCount(1); + listener.getTransport().setWorkerThreadPoolConfig( + ThreadPoolConfig.defaultConfig().setCorePoolSize(2).setMaxPoolSize(2)); + server.start(); + return server; + } + + @Test + public void testForwardHeaderShape() throws Exception { + try (ServerSocket listener = new ServerSocket(0)) { + CompletableFuture captured = CompletableFuture.supplyAsync(() -> { + try (Socket connection = listener.accept()) { + BufferedReader reader = new BufferedReader(new InputStreamReader( + connection.getInputStream(), StandardCharsets.ISO_8859_1)); + StringBuilder headers = new StringBuilder(); + String line; + while ((line = reader.readLine()) != null && !line.isEmpty()) { + headers.append(line).append("\r\n"); + } + connection.getOutputStream().write(("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n" + + "Content-Type: application/json\r\nConnection: close\r\n\r\n{}") + .getBytes(StandardCharsets.US_ASCII)); + return headers.toString(); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + GremlinClient client = new GremlinClient("http://127.0.0.1:" + listener.getLocalPort(), + 2000, 1, 1); + try (Response response = client.doPostRequest("Bearer example", body("1"))) { + Assert.assertEquals(200, response.getStatus()); + String headers = captured.get(5, TimeUnit.SECONDS); + Assert.assertTrue(headers.contains("Transfer-Encoding: chunked")); + Assert.assertTrue(headers.contains("Content-Type: application/json")); + Assert.assertTrue(headers.contains("Accept-Encoding: gzip")); + Assert.assertTrue(headers.contains("User-Agent: Jersey/")); + } finally { + client.close(); + } + } + } + + @Test + public void testHugeGraphTransactionsSerializationAndDynamicAliases() throws Exception { + checkHugeGraphContract("memory"); + } + + @Test + public void testRocksDBTransactionsSerializationAndDynamicAliases() throws Exception { + checkHugeGraphContract("rocksdb"); + } + + private static void checkHugeGraphContract(String backend) throws Exception { + RegisterUtil.registerBackends(); + Path data = Files.createTempDirectory("gremlin-rest-" + backend); + HugeGraph graph = HugeFactory.open(new HugeConfig(Map.of( + "backend", backend, "serializer", "rocksdb".equals(backend) ? "binary" : "text", + "store", "embedded_contract_" + backend, + "rocksdb.data_path", data.toString(), "rocksdb.wal_path", data.toString()))); + try { + graph.initBackend(); + graph.schema().propertyKey("name").asText().ifNotExist().create(); + graph.schema().vertexLabel("person").useCustomizeNumberId().properties("name") + .ifNotExist().create(); + graph.tx().commit(); + hub.notify(Events.GRAPH_CREATE, graph).get(5, TimeUnit.SECONDS); + String name = graph.spaceGraphName(); + String add = "g.addVertex(org.apache.tinkerpop.gremlin.structure.T.label,'person'," + + "org.apache.tinkerpop.gremlin.structure.T.id,1L,'name','alice')"; + String request = graphBody(add, name); + byte[] expected; + try (Response response = forward.doPostRequest(null, request)) { + Assert.assertEquals(200, response.getStatus()); + expected = response.readEntity(byte[].class); + } + try (Response response = embedded.post(null, graphBody( + "g.traversal().V().drop().iterate()", name), "test")) { + Assert.assertEquals(200, response.getStatus()); + } + try (Response response = embedded.post(null, request, "test")) { + Assert.assertEquals(200, response.getStatus()); + Assert.assertArrayEquals(expected, (byte[]) response.getEntity()); + } + for (GremlinRestExecutor path : Arrays.asList( + new ForwardGremlinRestExecutor(config), embedded)) { + try (Response response = path.post(null, graphBody( + "g.addVertex(org.apache.tinkerpop.gremlin.structure.T.label,'person'," + + "org.apache.tinkerpop.gremlin.structure.T.id,2L,'name','bob');" + + "throw new IllegalArgumentException('rollback')", name), "test")) { + Assert.assertEquals(500, response.getStatus()); + } + try (Response response = path.post(null, graphBody( + "g.traversal().V().count()", name), "test")) { + Assert.assertEquals(Collections.singletonList(1), + ((Map) json(response).get("result")).get("data")); + } + if (path != embedded) { + path.close(); + } + } + EmbeddedGremlinRestExecutor brokenSerializer = new EmbeddedGremlinRestExecutor( + config, server.getServerGremlinExecutor()); + MessageTextSerializer failing = + Mockito.mock(MessageTextSerializer.class); + Mockito.when(failing.serializeResponseAsBinary( + Mockito.any(ResponseMessage.class), + Mockito.any(ByteBufAllocator.class))) + .thenThrow(new SerializationException("intentional failure")); + brokenSerializer.initialize(failing); + try (Response response = brokenSerializer.post(null, + graphBody(add.replace("1L", "3L"), name), "test")) { + Assert.assertEquals(500, response.getStatus()); + Assert.assertEquals(SerializationException.class.getName(), + json(response).get("Exception-Class")); + } finally { + brokenSerializer.close(); + } + // The HTTP contract commits before serialization, even if serialization fails. + try (Response response = embedded.post(null, graphBody( + "g.traversal().V().count()", name), "test")) { + Assert.assertEquals(Collections.singletonList(2), + ((Map) json(response).get("result")).get("data")); + } + hub.notify(Events.GRAPH_DROP, graph).get(5, TimeUnit.SECONDS); + comparePost(graphBody("1", name)); + } finally { + hub.notify(Events.GRAPH_DROP, graph).get(5, TimeUnit.SECONDS); + graph.close(); + org.apache.commons.io.FileUtils.deleteDirectory(data.toFile()); + } + } + + @Test + public void testEvalTimeoutUsesGremlinEnvelope() { + String script = "org.apache.tinkerpop.gremlin.process.traversal.AnonymousTraversalSource" + + ".traversal().withEmbedded(org.apache.tinkerpop.gremlin.tinkergraph.structure." + + "TinkerGraph.open()).with('evaluationTimeout',50)" + + ".inject(1).map{Thread.sleep(300);1}"; + comparePost(body(script)); + try (Response response = embedded.post(null, body(script), "test")) { + Assert.assertEquals(500, response.getStatus()); + Assert.assertEquals("java.util.concurrent.TimeoutException", json(response).get("Exception-Class")); + } + } + + @Test + public void testConcurrentUserContexts() throws Exception { + ExecutorService callers = Executors.newFixedThreadPool(2); + try { + List> results = new ArrayList<>(); + for (String name : Arrays.asList("alice", "bob")) { + results.add(callers.submit(() -> { + try { + HugeGraphAuthProxy.setContext(new HugeGraphAuthProxy.Context( + new HugeAuthenticator.User(name, RolePermission.admin()))); + for (int i = 0; i < 5; i++) { + try (Response response = embedded.post(null, body( + "org.apache.hugegraph.auth.HugeGraphAuthProxy.getContext().user().username()"), + "test")) { + Assert.assertEquals(Collections.singletonList(name), + ((Map) json(response).get("result")).get("data")); + } + } + } finally { + HugeGraphAuthProxy.resetContext(); + } + })); + } + for (Future result : results) { + result.get(10, TimeUnit.SECONDS); + } + } finally { + callers.shutdownNow(); + } + } + + @Test + public void testAdmissionWaitsAndDoesNotRejectThirdRequest() throws Exception { + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Object engine = server.getServerGremlinExecutor().getGremlinExecutor(); + Whitebox.invoke(engine, "globalBindings", new Class[]{String.class, Object.class}, + "put", "entered", entered); + Whitebox.invoke(engine, "globalBindings", new Class[]{String.class, Object.class}, + "put", "release", release); + ExecutorService callers = Executors.newFixedThreadPool(3); + try { + List> results = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + results.add(callers.submit(() -> { + try (Response response = embedded.post(null, body( + "entered.countDown(); release.await(); 1"), "test")) { + return response.getStatus(); + } + })); + } + Assert.assertTrue(entered.await(1, TimeUnit.SECONDS)); + Semaphore admission = Whitebox.getInternalState(embedded, "admission"); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(1); + while (admission.getQueueLength() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + Assert.assertEquals(0, admission.availablePermits()); + Assert.assertEquals(1, admission.getQueueLength()); + release.countDown(); + for (Future result : results) { + Assert.assertEquals(Integer.valueOf(200), result.get(5, TimeUnit.SECONDS)); + } + } finally { + release.countDown(); + callers.shutdownNow(); + } + } + + /** Manual transport microbenchmark; not part of the correctness suite. */ + public static void main(String[] args) throws Exception { + setup(); + ForwardGremlinRestExecutor proxy = new ForwardGremlinRestExecutor(config); + HttpServer forwardRest = rest(proxy); + HttpServer embeddedRest = rest(embedded); + List> measurements = new ArrayList<>(); + try { + HttpClient client = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1).build(); + for (String script : Arrays.asList("1+2", "(1..100).toList()", "Thread.sleep(10);1")) { + String request = body(script); + String expected = client.send(post(forwardRest, request), + HttpResponse.BodyHandlers.ofString()).body(); + for (int i = 0; i < 100; i++) { + client.send(post(forwardRest, request), HttpResponse.BodyHandlers.ofString()); + client.send(post(embeddedRest, request), HttpResponse.BodyHandlers.ofString()); + } + for (int round = 0; round < 4; round++) { + for (int turn = 0; turn < 2; turn++) { + boolean local = (round + turn) % 2 == 0; + long[] elapsed = new long[200]; + for (int i = 0; i < elapsed.length; i++) { + long start = System.nanoTime(); + HttpResponse response = client.send( + post(local ? embeddedRest : forwardRest, request), + HttpResponse.BodyHandlers.ofString()); + elapsed[i] = System.nanoTime() - start; + Assert.assertEquals(response.body(), 200, response.statusCode()); + Assert.assertEquals(expected, response.body()); + } + long total = Arrays.stream(elapsed).sum(); + Arrays.sort(elapsed); + Map sample = new LinkedHashMap<>(); + sample.put("script", script); + sample.put("mode", local ? "embedded" : "forward"); + sample.put("round", round); + sample.put("samples", elapsed.length); + sample.put("p50_us", elapsed[elapsed.length / 2] / 1000.0); + sample.put("p95_us", elapsed[(int) (elapsed.length * 0.95)] / 1000.0); + sample.put("throughput_rps", elapsed.length * 1e9 / total); + sample.put("sorted_latency_ns", elapsed); + measurements.add(sample); + } + } + } + Files.writeString(Path.of(args[0]), JsonUtil.toJson(Map.of( + "java", System.getProperty("java.version"), "os", System.getProperty("os.name"), + "scope", "single-client script-only HTTP transport microbenchmark", + "measurements", measurements))); + } finally { + forwardRest.shutdownNow(); + embeddedRest.shutdownNow(); + proxy.close(); + teardown(); + } + } + + public static class ProtocolAuthenticator extends StandardAuthenticator { + + @Override + public void setup(Map ignored) { + } + + @Override + public HugeAuthenticator.User authenticate(Map credentials) { + return HugeAuthenticator.User.ADMIN; + } + } + + @Test + public void testAuthenticatedProtocolLimits() throws Exception { + Settings limited = new Settings(); + limited.host = "127.0.0.1"; + limited.port = 0; + limited.gremlinPool = 1; + limited.threadPoolBoss = 1; + limited.threadPoolWorker = 1; + limited.maxInitialLineLength = 128; + limited.maxHeaderSize = 512; + limited.channelizer = WsAndHttpChannelizer.class.getName(); + limited.serializers = settings.serializers; + limited.authentication.authenticator = ProtocolAuthenticator.class.getName(); + limited.authentication.authenticationHandler = WsAndHttpBasicAuthHandler.class.getName(); + ContextGremlinServer authenticated = new ContextGremlinServer(limited, + new EventHub("embedded-auth-protocol-test", 1)); + try { + authenticated.start().get(30, TimeUnit.SECONDS); + Channel socket = Whitebox.getInternalState(authenticated, "serverSocketChannel"); + String url = "http://127.0.0.1:" + ((InetSocketAddress) socket.localAddress()).getPort(); + HugeConfig config = new HugeConfig(Map.of("gremlinserver.url", url)); + EmbeddedGremlinRestExecutor adapter = new EmbeddedGremlinRestExecutor( + config, authenticated.getServerGremlinExecutor()); + GremlinRestExecutorFactory.initialize(adapter, authenticated); + GremlinClient client = new GremlinClient(url, 2000, 1, 1); + try { + MultivaluedMap params = new MultivaluedHashMap<>(); + params.add("gremlin", "1/*" + "x".repeat(200) + "*/"); + try (Response left = client.doGetRequest("Basic dXNlcjpwYXNz", params); + Response right = adapter.get("Basic dXNlcjpwYXNz", params, "test")) { + Assert.assertEquals(left.getStatus(), right.getStatus()); + } + } finally { + client.close(); + adapter.close(); + } + } finally { + authenticated.stop().get(30, TimeUnit.SECONDS); + HugeGraphAuthProxy.resetContext(); + } + } + + @Test + public void testAnonymousAuditUsesRestPeerWithoutCredentials() { + org.apache.logging.log4j.core.Logger logger = (org.apache.logging.log4j.core.Logger) + org.apache.logging.log4j.LogManager.getLogger( + org.apache.tinkerpop.gremlin.server.GremlinServer.AUDIT_LOGGER_NAME); + org.apache.logging.log4j.Level oldLevel = logger.getLevel(); + org.apache.logging.log4j.core.Appender appender = + Mockito.mock(org.apache.logging.log4j.core.Appender.class); + List messages = new ArrayList<>(); + Mockito.when(appender.getName()).thenReturn("embedded-audit-test"); + Mockito.when(appender.isStarted()).thenReturn(true); + Mockito.doAnswer(call -> { + org.apache.logging.log4j.core.LogEvent event = call.getArgument(0); + messages.add(event.getMessage().getFormattedMessage()); + return null; + }).when(appender).append(Mockito.any(org.apache.logging.log4j.core.LogEvent.class)); + logger.addAppender(appender); + logger.setLevel(org.apache.logging.log4j.Level.INFO); + boolean oldAudit = settings.enableAuditLog; + try { + settings.enableAuditLog = true; + HugeGraphAuthProxy.setContext(HugeGraphAuthProxy.Context.admin()); + try (Response response = embedded.post("Bearer audit-secret-sentinel", body("1"), "peer:123")) { + Assert.assertEquals(200, response.getStatus()); + } + Assert.assertEquals(Collections.singletonList( + "User anonymous with address peer:123 requested: 1"), messages); + } finally { + settings.enableAuditLog = oldAudit; + HugeGraphAuthProxy.resetContext(); + logger.removeAppender(appender); + logger.setLevel(oldLevel); + } + } + + private static String graphBody(String script, String graph) { + return JsonUtil.toJson(Map.of("gremlin", script, "requestId", ID, + "aliases", Map.of("g", graph))); + } + + private static String body(String script) { + return JsonUtil.toJson(Map.of("gremlin", script, "requestId", ID)); + } + + private static void comparePost(String body) { + compare(forward.doPostRequest(null, body), embedded.post(null, body, "test")); + } + + private static Map json(Response response) { + String text = response.getEntity() instanceof byte[] ? + new String((byte[]) response.getEntity(), StandardCharsets.UTF_8) : + response.readEntity(String.class); + return JsonUtil.fromJson(text, Map.class); + } + + private static void compare(Response expected, Response actual) { + try (expected; actual) { + Assert.assertEquals(expected.getStatus(), actual.getStatus()); + Map left = json(expected); + Map right = json(actual); + if (!ID.equals(left.get("requestId"))) { + left.remove("requestId"); + right.remove("requestId"); + } + left.remove("stackTrace"); + right.remove("stackTrace"); + Assert.assertEquals(left, right); + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java index 448669b95a..956cb9af90 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java @@ -28,6 +28,7 @@ import org.apache.hugegraph.api.cypher.CypherClientTest; import org.apache.hugegraph.auth.ContextGremlinServerHttpTest; import org.apache.hugegraph.auth.HttpGremlinRequestHandlerTest; +import org.apache.hugegraph.auth.EmbeddedGremlinRestExecutorTest; import org.apache.hugegraph.auth.ContextTaskTest; import org.apache.hugegraph.auth.StandardAuthManagerV2Test; import org.apache.hugegraph.auth.WsAndHttpBasicAuthHandlerTest; @@ -168,6 +169,7 @@ CypherOpProcessorTest.class, WsAndHttpBasicAuthHandlerTest.class, HttpGremlinRequestHandlerTest.class, + EmbeddedGremlinRestExecutorTest.class, ContextGremlinServerHttpTest.class, GraphSpaceGroupAPITest.class, GraphSpaceAuthPayloadTest.class, From d1fc963761c12ade9d1520141fd57a7cffcb937b Mon Sep 17 00:00:00 2001 From: contrueCT Date: Wed, 28 Oct 2026 00:39:20 +0800 Subject: [PATCH 2/4] fix(test): initialize REST executor in stop fixture - Initialize the executor mock used by the isolated server stop fixture. - Assert admission shutdown precedes Gremlin and REST shutdown. - Validate the timeout regression with 48 Java 17 regression tests. --- .../main/java/org/apache/hugegraph/HugeFactoryTest.java | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java index 702fd28289..7ab5ffeb16 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/HugeFactoryTest.java @@ -38,6 +38,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutor; import org.apache.hugegraph.backend.store.BackendStoreProvider; import org.apache.hugegraph.backend.tx.GraphTransaction; import org.apache.hugegraph.config.HugeConfig; @@ -63,6 +64,7 @@ import org.apache.hugegraph.util.Events; import org.junit.Test; import org.mockito.AdditionalAnswers; +import org.mockito.InOrder; import org.mockito.Mockito; import org.apache.tinkerpop.gremlin.server.GremlinServer; @@ -489,12 +491,18 @@ protected void done() { }).when(provider).close(); Whitebox.setInternalState(graph, "storeProvider", provider); HugeGraphServer server = Mockito.mock(HugeGraphServer.class, Mockito.CALLS_REAL_METHODS); + GremlinRestExecutor restExecutor = Mockito.mock(GremlinRestExecutor.class); + Whitebox.setInternalState(server, "gremlinRestExecutor", restExecutor); Whitebox.setInternalState(server, "gremlinServer", gremlin); Whitebox.setInternalState(server, "restServer", rest); Whitebox.setInternalState(server, "memoryMonitor", Mockito.mock(MemoryMonitor.class)); server.stop(); + InOrder shutdown = Mockito.inOrder(restExecutor, gremlin, rest); + shutdown.verify(restExecutor).beginShutdown(); + shutdown.verify(gremlin).stop(); + shutdown.verify(rest).shutdown(); Assert.assertEquals(2, earlyCloses.get()); Assert.assertEquals(TaskStatus.SUCCESS, task.status()); Assert.assertTrue("Pending task must persist completion before owner close", completionPersisted.get()); From 427b47d492629547a21777fb755c02554e30fd5b Mon Sep 17 00:00:00 2001 From: imbajin Date: Wed, 28 Oct 2026 13:28:15 +0800 Subject: [PATCH 3/4] fix(server): validate embedded Gremlin target addresses - Resolve every configured Gremlin target address before enabling embedded mode. - Reject mixed loopback and non-loopback DNS answers. - Match the selected listener bind independent of target answer order. - Add deterministic IPv4, IPv6, wildcard and mixed-address regression tests. --- .../gremlin/GremlinRestExecutorFactory.java | 28 ++++++++--- .../unit/api/gremlin/GremlinQueryAPITest.java | 48 +++++++++++++++++++ 2 files changed, 70 insertions(+), 6 deletions(-) diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java index add4713020..82271c1b17 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java @@ -21,6 +21,7 @@ import java.net.InetAddress; import java.net.URI; import java.nio.file.Path; +import java.util.Arrays; import org.apache.hugegraph.auth.StandardAuthenticator; import org.apache.hugegraph.auth.WsAndHttpBasicAuthHandler; @@ -50,14 +51,16 @@ public static GremlinRestExecutor create(HugeConfig config, GremlinServer server } Settings settings = server.getServerGremlinExecutor().getSettings(); URI target = URI.create(config.get(ServerOptions.GREMLIN_SERVER_URL)); + String localTargetError = + "embedded requires gremlinserver.url to target this local Gremlin Server"; E.checkArgument("http".equals(target.getScheme()) && target.getHost() != null && target.getUserInfo() == null && target.getFragment() == null && - InetAddress.getByName(target.getHost()).isLoopbackAddress() && - target.getPort() == settings.port && - (InetAddress.getByName(settings.host).isAnyLocalAddress() || - InetAddress.getByName(settings.host).equals( - InetAddress.getByName(target.getHost()))), - "embedded requires gremlinserver.url to target this local Gremlin Server"); + target.getPort() == settings.port, + localTargetError); + InetAddress[] targetAddresses = InetAddress.getAllByName(target.getHost()); + InetAddress bindAddress = InetAddress.getByName(settings.host); + E.checkArgument(isSameProcessTarget(targetAddresses, bindAddress), + localTargetError); E.checkArgument(settings.ssl == null || !settings.ssl.enabled, "embedded does not support a TLS Gremlin listener"); E.checkArgument(HugeGraphWsAndHttpChannelizer.class.getName().equals(settings.channelizer), @@ -86,6 +89,19 @@ public static GremlinRestExecutor create(HugeConfig config, GremlinServer server return new EmbeddedGremlinRestExecutor(config, server.getServerGremlinExecutor()); } + private static boolean isSameProcessTarget(InetAddress[] targetAddresses, + InetAddress bindAddress) { + if (targetAddresses.length == 0 || + Arrays.stream(targetAddresses).anyMatch(address -> + !address.isLoopbackAddress())) { + return false; + } + if (bindAddress.isAnyLocalAddress()) { + return true; + } + return Arrays.stream(targetAddresses).anyMatch(bindAddress::equals); + } + public static void initialize(GremlinRestExecutor executor, GremlinServer server) { if (executor instanceof EmbeddedGremlinRestExecutor) { HugeGraphWsAndHttpChannelizer channelizer = diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/gremlin/GremlinQueryAPITest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/gremlin/GremlinQueryAPITest.java index 94deaf921d..14177b16dd 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/gremlin/GremlinQueryAPITest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/gremlin/GremlinQueryAPITest.java @@ -18,8 +18,10 @@ package org.apache.hugegraph.unit.api.gremlin; import java.lang.reflect.Method; +import java.net.InetAddress; import org.apache.hugegraph.api.gremlin.GremlinQueryAPI; +import org.apache.hugegraph.api.gremlin.GremlinRestExecutorFactory; import org.apache.hugegraph.testutil.Assert; import org.apache.hugegraph.unit.BaseUnitTest; import org.junit.Test; @@ -33,6 +35,15 @@ private static boolean matchBadRequest(String exClass) throws Exception { return (boolean) m.invoke(null, exClass); } + private static boolean isSameProcessTarget( + InetAddress[] targetAddresses, + InetAddress bindAddress) throws Exception { + Method m = GremlinRestExecutorFactory.class.getDeclaredMethod( + "isSameProcessTarget", InetAddress[].class, InetAddress.class); + m.setAccessible(true); + return (boolean) m.invoke(null, (Object) targetAddresses, bindAddress); + } + @Test public void testMatchBadRequestExceptionWithTinkerpop() throws Exception { Assert.assertTrue(matchBadRequest( @@ -60,4 +71,41 @@ public void testMatchBadRequestExceptionWithOther() throws Exception { Assert.assertFalse(matchBadRequest("java.lang.NullPointerException")); Assert.assertFalse(matchBadRequest("java.io.IOException")); } + + @Test + public void testSameProcessTargetMatchesAnyLoopbackAddress() throws Exception { + InetAddress ipv4 = InetAddress.getByName("127.0.0.1"); + InetAddress ipv6 = InetAddress.getByName("::1"); + + Assert.assertTrue(isSameProcessTarget( + new InetAddress[]{ipv6, ipv4}, ipv4)); + Assert.assertTrue(isSameProcessTarget( + new InetAddress[]{ipv4, ipv6}, ipv4)); + } + + @Test + public void testSameProcessTargetRejectsMixedOrDifferentAddresses() + throws Exception { + InetAddress ipv4 = InetAddress.getByName("127.0.0.1"); + InetAddress ipv6 = InetAddress.getByName("::1"); + InetAddress remote = InetAddress.getByName("192.0.2.1"); + + Assert.assertFalse(isSameProcessTarget( + new InetAddress[]{ipv4, remote}, ipv4)); + Assert.assertFalse(isSameProcessTarget( + new InetAddress[]{ipv6}, ipv4)); + } + + @Test + public void testSameProcessTargetWithWildcardBind() throws Exception { + InetAddress ipv4 = InetAddress.getByName("127.0.0.1"); + InetAddress ipv6 = InetAddress.getByName("::1"); + InetAddress remote = InetAddress.getByName("192.0.2.1"); + InetAddress wildcard = InetAddress.getByName("0.0.0.0"); + + Assert.assertTrue(isSameProcessTarget( + new InetAddress[]{ipv4, ipv6}, wildcard)); + Assert.assertFalse(isSameProcessTarget( + new InetAddress[]{ipv4, remote}, wildcard)); + } } From a2f98d525da4c6fd5c68c0550f488a0dd1af0761 Mon Sep 17 00:00:00 2001 From: contrueCT Date: Wed, 28 Oct 2026 23:27:04 +0800 Subject: [PATCH 4/4] feat(server): auto-select local Gremlin execution - Remove the REST execution-mode option and select embedded execution for local listener targets. - Support local interfaces, host aliases and TLS listeners while preserving remote forwarding. - Cover automatic selection, DNS address checks and a live TLS listener with regression tests. - Update configuration docs and the baseline/candidate benchmark runner. --- docs/gremlin-rest-execution.md | 74 +++++---- hugegraph-server/README.md | 2 +- .../gremlin/EmbeddedGremlinRestExecutor.java | 3 +- .../gremlin/GremlinRestExecutorFactory.java | 63 +++++--- .../hugegraph/config/ServerOptions.java | 8 - .../static/conf/rest-server.properties | 5 +- .../scripts/gremlin-rest-rocksdb-benchmark.py | 30 ++-- .../auth/EmbeddedGremlinRestExecutorTest.java | 145 +++++++++++++++++- .../unit/api/gremlin/GremlinQueryAPITest.java | 6 +- 9 files changed, 251 insertions(+), 85 deletions(-) diff --git a/docs/gremlin-rest-execution.md b/docs/gremlin-rest-execution.md index 47c49ffdd4..4617c54b1c 100644 --- a/docs/gremlin-rest-execution.md +++ b/docs/gremlin-rest-execution.md @@ -1,41 +1,54 @@ -# REST Gremlin execution modes +# REST Gremlin execution -`gremlinserver.rest_execution_mode` selects the execution path for JSON requests to -REST `GET /gremlin` and `POST /gremlin`. The default is `forward`. +JSON requests to REST `GET /gremlin` and `POST /gremlin` automatically execute +in-process when `gremlinserver.url` identifies the local Gremlin Server started by +`HugeGraphServer`. Targets on other hosts or ports use HTTP forwarding. There is no +execution-mode configuration option. ```text -forward: REST -> Jersey HTTP client -> Gremlin HTTP listener -> Gremlin workers -embedded: REST ---------------------------------------------> Gremlin workers +local Gremlin target: REST -------------------------------------> Gremlin workers +remote Gremlin target: REST -> Jersey HTTP client -> Gremlin HTTP -> Gremlin workers ``` -Both paths use the same Gremlin executor, graph manager, script engines and configured -JSON serializer. The Gremlin listener remains available to HTTP/WebSocket clients, -Driver and Console. Cypher and Gremlin jobs keep their existing execution paths. +Both paths use the configured graph manager, script engines and JSON serializer. +For local targets these are the same instances owned by the co-started Gremlin Server. +The Gremlin listener remains available to HTTP/WebSocket clients, Driver and Console. +Cypher and Gremlin jobs keep their existing execution paths. -## Configuration +## Automatic selection -In `conf/rest-server.properties`: +In `conf/rest-server.properties`, keep the existing Gremlin URL consistent with the +host and port in `conf/gremlin-server.yaml`: ```properties gremlinserver.url=http://127.0.0.1:8182 -gremlinserver.rest_execution_mode=embedded ``` -Restart the server to change modes. To revert, set the mode to `forward` and restart. -There is no automatic forwarding retry after an embedded execution failure: a query -may already have committed even when the caller receives an error. +Loopback addresses, local interface addresses, local hostnames and wildcard bind +addresses identify the same host. All addresses resolved for a hostname must be local. +The URL's port must match the co-started listener's port, so a URL for another Gremlin +process on the same machine continues to use forwarding. Omitted HTTP/HTTPS ports are +interpreted as 80/443. Unresolved hostnames remain on the forwarding path. +A standalone `RestServer` without a co-started Gremlin Server always forwards. -The initial embedded implementation supports the standard HugeGraph -`WsAndHttpChannelizer` configuration, adapted internally to -`HugeGraphWsAndHttpChannelizer`. It requires the URL to identify the same process's -plain HTTP listener on a loopback address, with matching host/bind address and port. -Remote targets, TLS Gremlin listeners, custom channelizers and custom authorizers -continue to use `forward`. +TLS does not affect the choice: local HTTP and HTTPS targets both use in-process +execution, including when the Gremlin listener enables TLS. This does not change the +TLS settings of either listener. TLS handshakes and client certificates on the Gremlin +listener apply to direct network clients; embedded requests use the REST entry point's +transport and authentication, without an internal TLS connection. + +Restart after changing the target URL. There is no automatic forwarding retry after +an embedded execution failure: a query may already have committed even when the caller +receives an error. + +Local execution supports the standard HugeGraph `WsAndHttpChannelizer` configuration, +adapted internally to `HugeGraphWsAndHttpChannelizer`. Custom channelizers and custom +Gremlin authorizers fail startup for local targets. Authentication must agree between REST and Gremlin: either both are anonymous, or both use `StandardAuthenticator`, with `WsAndHttpBasicAuthHandler` on Gremlin and `authentication.config.tokens` resolving to the same REST configuration file. -Unsupported combinations fail startup; they do not silently select another mode. +Unsupported local combinations fail startup; they do not silently select another path. ## Preserved behavior @@ -99,9 +112,9 @@ request identity, context isolation, blocking admission, timeouts, shared serial dynamic aliases and graph transaction side effects on Memory and RocksDB. Response stack traces and newly generated request IDs require semantic comparison; deterministic responses also use byte comparisons. -Run backend and distribution checks appropriate to the intended deployment before enabling -embedded. Focused tests do not establish HStore compatibility or production performance. -Default-mode changes require separate representative performance evidence. +Run backend and distribution checks appropriate to the intended deployment before +rolling out automatic local execution. Focused tests do not establish HStore compatibility +or production performance. Readiness, overload policy, authentication-cache consistency and global shutdown redesign remain separate tasks from this execution-path change. @@ -119,7 +132,8 @@ and alternates mode order across four rounds. It saves latency samples and envir metadata. These measurements include client/REST overhead and exclude real backend I/O; 100 warmup requests do not establish steady-state JIT behavior. -For a Linux single-node RocksDB comparison, build the distribution and use an eight-CPU +For a Linux single-node RocksDB comparison, build this distribution and retain a +pre-automatic-selection distribution whose default path is forwarding. Use an eight-CPU host with Java 17. The following runner requires a new output directory and binds both listeners to loopback. It generates its own graph and preserves all evidence under that directory: @@ -127,15 +141,19 @@ directory: ```bash python3 hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py \ --distribution /path/to/apache-hugegraph-server-1.8.0 \ + --forward-distribution /path/to/previous-forward-default-build \ --java-home /path/to/jdk-17 \ --output /path/to/new-results-directory ``` It checks GET/POST, bindings, JSON values, aliases, rollback, committed write counts and restart persistence. The default dataset has 10,000 vertices and 50,000 edges. Four server -starts use forward/embedded/embedded/forward order, identical JVM settings and persistent -data. Each workload has a five-second warmup and three-second measurements at concurrency -1, 8 and 32. Samples use a closed-loop client on the same host; CPU affinity separates +starts use the baseline/candidate/candidate/baseline builds in forward/embedded/embedded/forward +order, identical JVM settings and persistent data. The runner records each build's jar hashes. +This compares builds, so other changes between their revisions may affect results. The +script-only runner above explicitly constructs both adapters in one JVM and remains +available for a same-build transport comparison. Each workload has a five-second warmup +and three-second measurements at concurrency 1, 8 and 32. Samples use a closed-loop client on the same host; CPU affinity separates the client and server, while unrelated host workloads can still affect both. The results include raw latency samples, response checks, CPU/load snapshots, GC and server logs. These short measurements do not establish production saturation or steady-state behavior. diff --git a/hugegraph-server/README.md b/hugegraph-server/README.md index edf06801dc..bdeee5a8c8 100644 --- a/hugegraph-server/README.md +++ b/hugegraph-server/README.md @@ -8,7 +8,7 @@ HugeGraph Server consists of two layers of functionality: the graph engine layer - REST Server: Provides a RESTful API for querying graph/schema information, supports the [Gremlin](https://tinkerpop.apache.org/gremlin.html) and [Cypher](https://en.wikipedia.org/wiki/Cypher) query languages, and offers APIs for service monitoring and operations. - See [Cypher HTTP contracts](../docs/server-runtime.md#cypher-http-contracts) for request forms, verified behavior, and unverified scope. - - See [REST Gremlin execution modes](../docs/gremlin-rest-execution.md) for optional in-process + - See [REST Gremlin execution](../docs/gremlin-rest-execution.md) for automatic local in-process execution, supported configurations and compatibility boundaries. - Graph Engine: Supports both OLTP and OLAP graph computation types, with OLTP implementing the [Apache TinkerPop3](https://tinkerpop.apache.org) framework. - Backend Interface: Implements the storage of graph data to the backend. diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java index 424fa328f6..ff276cf83f 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java @@ -290,7 +290,8 @@ private int headerSize(HttpMethod method, String authorization) { // HttpObjectDecoder counts header lines without CRLF. Match the headers // produced by the pinned Jersey/Apache connector, not the outer REST headers. String host = this.target.getHost(); - if (this.target.getPort() != 80) { + int defaultPort = "https".equalsIgnoreCase(this.target.getScheme()) ? 443 : 80; + if (this.target.getPort() != -1 && this.target.getPort() != defaultPort) { host += ":" + this.target.getPort(); } int length = ("Accept: application/json" + diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java index 82271c1b17..794b13d156 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java @@ -19,9 +19,10 @@ import java.io.IOException; import java.net.InetAddress; +import java.net.NetworkInterface; import java.net.URI; +import java.net.UnknownHostException; import java.nio.file.Path; -import java.util.Arrays; import org.apache.hugegraph.auth.StandardAuthenticator; import org.apache.hugegraph.auth.WsAndHttpBasicAuthHandler; @@ -39,30 +40,16 @@ private GremlinRestExecutorFactory() { } public static GremlinRestExecutor standalone(HugeConfig config) { - E.checkArgument("forward".equals(config.get(ServerOptions.GREMLIN_REST_EXECUTION_MODE)), - "embedded requires startup through HugeGraphServer"); return new ForwardGremlinRestExecutor(config); } public static GremlinRestExecutor create(HugeConfig config, GremlinServer server, String restConfigFile) throws IOException { - if ("forward".equals(config.get(ServerOptions.GREMLIN_REST_EXECUTION_MODE))) { - return new ForwardGremlinRestExecutor(config); - } Settings settings = server.getServerGremlinExecutor().getSettings(); URI target = URI.create(config.get(ServerOptions.GREMLIN_SERVER_URL)); - String localTargetError = - "embedded requires gremlinserver.url to target this local Gremlin Server"; - E.checkArgument("http".equals(target.getScheme()) && target.getHost() != null && - target.getUserInfo() == null && target.getFragment() == null && - target.getPort() == settings.port, - localTargetError); - InetAddress[] targetAddresses = InetAddress.getAllByName(target.getHost()); - InetAddress bindAddress = InetAddress.getByName(settings.host); - E.checkArgument(isSameProcessTarget(targetAddresses, bindAddress), - localTargetError); - E.checkArgument(settings.ssl == null || !settings.ssl.enabled, - "embedded does not support a TLS Gremlin listener"); + if (!targetsLocalServer(target, settings)) { + return new ForwardGremlinRestExecutor(config); + } E.checkArgument(HugeGraphWsAndHttpChannelizer.class.getName().equals(settings.channelizer), "embedded requires HugeGraphWsAndHttpChannelizer"); E.checkArgument(settings.authorization == null || @@ -89,17 +76,43 @@ public static GremlinRestExecutor create(HugeConfig config, GremlinServer server return new EmbeddedGremlinRestExecutor(config, server.getServerGremlinExecutor()); } + private static boolean targetsLocalServer(URI target, Settings settings) throws IOException { + String scheme = target.getScheme(); + if (!("http".equalsIgnoreCase(scheme) || "https".equalsIgnoreCase(scheme)) || + target.getHost() == null || target.getUserInfo() != null || target.getFragment() != null) { + return false; + } + int port = target.getPort() == -1 ? + ("https".equalsIgnoreCase(scheme) ? 443 : 80) : target.getPort(); + // A different local port may identify another Gremlin Server process. + if (port != settings.port || settings.host == null || settings.host.isEmpty()) { + return false; + } + try { + return isSameProcessTarget(InetAddress.getAllByName(target.getHost()), + InetAddress.getByName(settings.host)); + } catch (UnknownHostException ignored) { + // Keep unresolved forwarding targets on the existing HTTP client path. + return false; + } + } + private static boolean isSameProcessTarget(InetAddress[] targetAddresses, - InetAddress bindAddress) { - if (targetAddresses.length == 0 || - Arrays.stream(targetAddresses).anyMatch(address -> - !address.isLoopbackAddress())) { + InetAddress bindAddress) throws IOException { + if (targetAddresses.length == 0 || !isLocalAddress(bindAddress)) { return false; } - if (bindAddress.isAnyLocalAddress()) { - return true; + for (InetAddress address : targetAddresses) { + if (!isLocalAddress(address)) { + return false; + } } - return Arrays.stream(targetAddresses).anyMatch(bindAddress::equals); + return true; + } + + private static boolean isLocalAddress(InetAddress address) throws IOException { + return address.isAnyLocalAddress() || address.isLoopbackAddress() || + NetworkInterface.getByInetAddress(address) != null; } public static void initialize(GremlinRestExecutor executor, GremlinServer server) { diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java index 1343dce758..3441291e4d 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/config/ServerOptions.java @@ -111,14 +111,6 @@ public class ServerOptions extends OptionHolder { "http://127.0.0.1:8182" ).withUrlNormalization("http://"); - public static final ConfigOption GREMLIN_REST_EXECUTION_MODE = - new ConfigOption<>( - "gremlinserver.rest_execution_mode", - "Execution path for REST /gremlin: forward or same-process embedded.", - allowValues("forward", "embedded"), - "forward" - ); - public static final ConfigOption GREMLIN_SERVER_TIMEOUT = new ConfigOption<>( "gremlinserver.timeout", diff --git a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties index b89abd77cc..fc8ad20d27 100644 --- a/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties +++ b/hugegraph-server/hugegraph-dist/src/assembly/static/conf/rest-server.properties @@ -4,9 +4,8 @@ restserver.url=http://127.0.0.1:8080 #restserver.enable_graphspaces_filter=false # gremlin server url, need to be consistent with host and port in gremlin-server.yaml #gremlinserver.url=127.0.0.1:8182 -# REST /gremlin: forward (default) or embedded for this process only. -# See repository docs/gremlin-rest-execution.md before enabling embedded. -gremlinserver.rest_execution_mode=forward +# REST /gremlin executes in-process when the URL targets this local Gremlin listener. +# Remote targets continue to use HTTP forwarding. See docs/gremlin-rest-execution.md. graphs=./conf/graphs diff --git a/hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py b/hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py index 21da107d91..3f02e0cf4d 100644 --- a/hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py +++ b/hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py @@ -17,7 +17,8 @@ """Linux standalone RocksDB comparison; owns only the supplied NEW output directory. Uses Python's standard library. Closed-loop clients validate every response. -ABBA restarts use the same distribution, persistent dataset and JVM settings. +ABBA restarts compare a forward-default baseline with the automatic-selection build. +Both builds share a persistent dataset and JVM settings. Server CPU affinity: 0-5; load generator: 6-7. Requires at least eight CPUs. """ @@ -187,6 +188,8 @@ def worker(index): def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--distribution", type=Path, required=True) + parser.add_argument("--forward-distribution", type=Path, required=True, + help="Pre-automatic-selection build whose default REST path is forwarding") parser.add_argument("--java-home", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--vertices", type=int, default=10000) @@ -196,6 +199,8 @@ def main(): assert args.vertices >= 1000 assert set(range(8)).issubset(os.sched_getaffinity(0)) args.distribution = args.distribution.resolve() + args.forward_distribution = args.forward_distribution.resolve() + assert args.forward_distribution != args.distribution, "Use a separate forward-default baseline build" args.java_home = args.java_home.resolve() args.output = args.output.resolve() args.output.mkdir(parents=True, exist_ok=False) @@ -205,9 +210,14 @@ def main(): assert (args.java_home / "conf/security/java.security").is_file(), "Incomplete Java 17 configuration" version = subprocess.check_output([java, "-version"], stderr=subprocess.STDOUT, text=True) assert 'version "17.' in version, version - cp = ":".join(str(p) for p in sorted((dist / "lib").glob("hugegraph*.jar"))) - cp += ":" + ":".join(str(p) for p in sorted((dist / "lib").glob("*.jar")) - if not p.name.startswith("hugegraph")) + distributions = {"forward": args.forward_distribution, "embedded": dist} + classpaths = {} + for mode, distribution in distributions.items(): + assert (distribution / "lib").is_dir(), distribution + jars = sorted((distribution / "lib").glob("hugegraph*.jar")) + jars += sorted(p for p in (distribution / "lib").glob("*.jar") + if not p.name.startswith("hugegraph")) + classpaths[mode] = ":".join(str(p) for p in jars) (output / "graphs").mkdir() (output / "graphs/hugegraph.properties").write_text( "gremlin.graph=org.apache.hugegraph.HugeFactory\nbackend=rocksdb\nserializer=binary\n" @@ -223,14 +233,14 @@ def main(): f"gremlinserver.url=http://127.0.0.1:{gp}\nrestserver.max_worker_threads=16\n" "gremlinserver.max_route=16\ngremlinserver.timeout=30\nmemory_monitor.threshold=1\n") for mode in ("forward", "embedded"): - (output / f"{mode}.properties").write_text(common + f"gremlinserver.rest_execution_mode={mode}\n") + (output / f"{mode}.properties").write_text(common) jvm = [java, "-Xms1g", "-Xmx2g", "-XX:ActiveProcessorCount=6", "@" + str(dist / "bin/jvm-module.options"), "-Dlog4j.configurationFile=" + str(dist / "conf/log4j2.xml")] env = {**os.environ, "JAVA_HOME": str(args.java_home)} affinity = lambda: os.sched_setaffinity(0, set(range(6))) with (output / "init.log").open("w") as log: - subprocess.run(jvm + ["-cp", cp, "org.apache.hugegraph.cmd.InitStore", + subprocess.run(jvm + ["-cp", classpaths["embedded"], "org.apache.hugegraph.cmd.InitStore", str(output / "forward.properties")], cwd=output, env=env, stdout=log, stderr=subprocess.STDOUT, check=True, timeout=300, preexec_fn=affinity) @@ -242,9 +252,11 @@ def main(): "duration_seconds": args.seconds, "mode_order": ["forward", "embedded", "embedded", "forward"], "warmup_seconds": args.warmup_seconds, + "comparison": "previous forward-default build vs automatic local execution", "distribution_sha256": { - p.name: hashlib.sha256(p.read_bytes()).hexdigest() - for p in (dist / "lib").glob("hugegraph*.jar")}}, + mode: {p.name: hashlib.sha256(p.read_bytes()).hexdigest() + for p in (distribution / "lib").glob("hugegraph*.jar")} + for mode, distribution in distributions.items()}}, "runs": []} n, sequence = args.vertices, 0 for run_id, mode in enumerate(result["environment"]["mode_order"]): @@ -253,7 +265,7 @@ def main(): with (run_dir / "server.log").open("w") as log: process = subprocess.Popen( jvm + [f"-Xlog:gc*:file={run_dir}/gc.log:time,uptime,level,tags", - "-cp", cp, "org.apache.hugegraph.dist.HugeGraphServer", + "-cp", classpaths[mode], "org.apache.hugegraph.dist.HugeGraphServer", str(output / "gremlin.yaml"), str(output / f"{mode}.properties")], cwd=output, env=env, stdout=log, stderr=subprocess.STDOUT, preexec_fn=affinity) run = {"id": run_id, "mode": mode, "checks": {}, "measurements": []} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java index 1e44ce31be..dd2658298c 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java @@ -19,7 +19,9 @@ import java.io.BufferedReader; import java.io.InputStreamReader; +import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.NetworkInterface; import java.net.ServerSocket; import java.net.Socket; import java.net.URI; @@ -29,6 +31,7 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; +import java.security.KeyStore; import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; @@ -46,6 +49,9 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManagerFactory; + import org.apache.hugegraph.HugeFactory; import org.apache.hugegraph.HugeGraph; import org.apache.hugegraph.api.filter.AuthenticationFilter; @@ -129,7 +135,6 @@ public static void setup() throws Exception { settings.port = ((InetSocketAddress) listening.localAddress()).getPort(); config = new HugeConfig(Map.of( "gremlinserver.url", "http://127.0.0.1:" + settings.port, - "gremlinserver.rest_execution_mode", "embedded", "gremlinserver.timeout", "2", "gremlinserver.max_route", "2")); embedded = (EmbeddedGremlinRestExecutor) GremlinRestExecutorFactory.create(config, server, ""); GremlinRestExecutorFactory.initialize(embedded, server); @@ -236,15 +241,139 @@ public void testSharedSerializerAndStartupStopGuards() { } @Test - public void testFactoryRejectsRemoteAndMismatchedAuthentication() { - HugeConfig remote = new HugeConfig(Map.of( - "gremlinserver.url", "http://192.0.2.1:8182", - "gremlinserver.rest_execution_mode", "embedded")); - Assert.assertThrows(IllegalArgumentException.class, - () -> GremlinRestExecutorFactory.create(remote, server, "")); + public void testFactorySelectsLocalExecutorAutomatically() throws Exception { + assertEmbedded(config); + assertEmbedded(new HugeConfig(Map.of( + "gremlinserver.url", "http://localhost:" + settings.port))); + String previousHost = settings.host; + try { + settings.host = "0.0.0.0"; + assertEmbedded(config); + for (NetworkInterface network : Collections.list(NetworkInterface.getNetworkInterfaces())) { + for (InetAddress address : Collections.list(network.getInetAddresses())) { + String host = address.getHostAddress(); + URI url = new URI("http", null, host, settings.port, null, null, null); + assertEmbedded(new HugeConfig(Map.of("gremlinserver.url", url.toString()))); + } + } + } finally { + settings.host = previousHost; + } + } + + @Test + public void testFactoryForwardsOtherHostsPortsAndStandaloneRest() throws Exception { + for (String url : Arrays.asList("http://192.0.2.1:" + settings.port, + "http://127.0.0.1:1", + "http://gremlin.invalid:" + settings.port)) { + try (GremlinRestExecutor executor = GremlinRestExecutorFactory.create( + new HugeConfig(Map.of("gremlinserver.url", url)), server, "")) { + Assert.assertTrue(executor instanceof ForwardGremlinRestExecutor); + } + } + try (GremlinRestExecutor executor = GremlinRestExecutorFactory.standalone(config)) { + Assert.assertTrue(executor instanceof ForwardGremlinRestExecutor); + } + } + + @Test + public void testFactorySupportsTlsListenerAndHttpsTarget() throws Exception { + Settings.SslSettings previousSsl = settings.ssl; + int previousPort = settings.port; + try { + settings.ssl = new Settings.SslSettings(); + settings.ssl.enabled = true; + assertEmbedded(new HugeConfig(Map.of( + "gremlinserver.url", "https://localhost:" + settings.port))); + settings.port = 443; + assertEmbedded(new HugeConfig(Map.of("gremlinserver.url", "https://localhost"))); + settings.port = 80; + assertEmbedded(new HugeConfig(Map.of("gremlinserver.url", "http://localhost"))); + } finally { + settings.ssl = previousSsl; + settings.port = previousPort; + } + } + + @Test + public void testEmbeddedExecutionWithRunningTlsListener() throws Exception { + Path directory = Files.createTempDirectory("embedded-tls"); + Path store = directory.resolve("server.p12"); + Process keytool = new ProcessBuilder( + Path.of(System.getProperty("java.home"), "bin", "keytool").toString(), + "-genkeypair", "-alias", "gremlin", "-keyalg", "RSA", "-keysize", "2048", + "-dname", "CN=localhost", "-ext", "SAN=dns:localhost,ip:127.0.0.1", "-validity", "1", + "-storetype", "PKCS12", "-keystore", store.toString(), "-storepass", "changeit", + "-noprompt").redirectErrorStream(true).redirectOutput(directory.resolve("keytool.log").toFile()) + .start(); + ContextGremlinServer secured = null; + try { + Assert.assertTrue(keytool.waitFor(30, TimeUnit.SECONDS)); + Assert.assertEquals(0, keytool.exitValue()); + Settings tls = new Settings(); + tls.host = "127.0.0.1"; + tls.port = 0; + tls.gremlinPool = 1; + tls.threadPoolBoss = 1; + tls.threadPoolWorker = 1; + tls.channelizer = WsAndHttpChannelizer.class.getName(); + tls.serializers = settings.serializers; + tls.ssl = new Settings.SslSettings(); + tls.ssl.enabled = true; + tls.ssl.keyStore = store.toString(); + tls.ssl.keyStoreType = "PKCS12"; + tls.ssl.keyStorePassword = "changeit"; + secured = new ContextGremlinServer(tls, new EventHub("embedded-tls-test", 1)); + secured.start().get(30, TimeUnit.SECONDS); + Channel socket = Whitebox.getInternalState(secured, "serverSocketChannel"); + tls.port = ((InetSocketAddress) socket.localAddress()).getPort(); + String url = "https://localhost:" + tls.port; + HugeConfig local = new HugeConfig(Map.of("gremlinserver.url", url)); + TrustManagerFactory trust = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + trust.init(KeyStore.getInstance(store.toFile(), "changeit".toCharArray())); + SSLContext ssl = SSLContext.getInstance("TLS"); + ssl.init(null, trust.getTrustManagers(), null); + HttpClient client = HttpClient.newBuilder().sslContext(ssl) + .connectTimeout(Duration.ofSeconds(5)).build(); + HttpResponse forwarded = client.send(HttpRequest.newBuilder(URI.create(url)) + .timeout(Duration.ofSeconds(5)).header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(body("1+2"))).build(), + HttpResponse.BodyHandlers.ofString()); + Assert.assertEquals(200, forwarded.statusCode()); + try (GremlinRestExecutor executor = GremlinRestExecutorFactory.create(local, secured, "")) { + Assert.assertTrue(executor instanceof EmbeddedGremlinRestExecutor); + GremlinRestExecutorFactory.initialize(executor, secured); + try (Response response = executor.post(null, body("1+2"), "test")) { + Assert.assertEquals(200, response.getStatus()); + Assert.assertEquals(JsonUtil.fromJson(forwarded.body(), Map.class), json(response)); + } + } + } finally { + keytool.destroyForcibly(); + if (secured != null) { + secured.stop().get(30, TimeUnit.SECONDS); + } + org.apache.commons.io.FileUtils.deleteDirectory(directory.toFile()); + HugeGraphAuthProxy.resetContext(); + } + } + + private static void assertEmbedded(HugeConfig local) throws Exception { + try (GremlinRestExecutor executor = GremlinRestExecutorFactory.create(local, server, "")) { + Assert.assertTrue(executor instanceof EmbeddedGremlinRestExecutor); + GremlinRestExecutorFactory.initialize(executor, server); + try (Response response = executor.post(null, body("1+2"), "test")) { + Assert.assertEquals(200, response.getStatus()); + Assert.assertEquals(Collections.singletonList(3), + ((Map) json(response).get("result")).get("data")); + } + } + } + + @Test + public void testFactoryRejectsMismatchedAuthentication() { HugeConfig auth = new HugeConfig(Map.of( "gremlinserver.url", config.get(ServerOptions.GREMLIN_SERVER_URL), - "gremlinserver.rest_execution_mode", "embedded", "auth.authenticator", StandardAuthenticator.class.getName())); Assert.assertThrows(IllegalArgumentException.class, () -> GremlinRestExecutorFactory.create(auth, server, "")); diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/gremlin/GremlinQueryAPITest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/gremlin/GremlinQueryAPITest.java index 14177b16dd..f955a4521f 100644 --- a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/gremlin/GremlinQueryAPITest.java +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/gremlin/GremlinQueryAPITest.java @@ -84,7 +84,7 @@ public void testSameProcessTargetMatchesAnyLoopbackAddress() throws Exception { } @Test - public void testSameProcessTargetRejectsMixedOrDifferentAddresses() + public void testSameProcessTargetRejectsMixedOrRemoteAddresses() throws Exception { InetAddress ipv4 = InetAddress.getByName("127.0.0.1"); InetAddress ipv6 = InetAddress.getByName("::1"); @@ -92,8 +92,10 @@ public void testSameProcessTargetRejectsMixedOrDifferentAddresses() Assert.assertFalse(isSameProcessTarget( new InetAddress[]{ipv4, remote}, ipv4)); - Assert.assertFalse(isSameProcessTarget( + Assert.assertTrue(isSameProcessTarget( new InetAddress[]{ipv6}, ipv4)); + Assert.assertFalse(isSameProcessTarget(new InetAddress[0], ipv4)); + Assert.assertFalse(isSameProcessTarget(new InetAddress[]{ipv4}, remote)); } @Test