diff --git a/docs/gremlin-rest-execution.md b/docs/gremlin-rest-execution.md new file mode 100644 index 0000000000..4617c54b1c --- /dev/null +++ b/docs/gremlin-rest-execution.md @@ -0,0 +1,159 @@ +# REST Gremlin execution + +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 +local Gremlin target: REST -------------------------------------> Gremlin workers +remote Gremlin target: REST -> Jersey HTTP client -> Gremlin HTTP -> Gremlin workers +``` + +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. + +## Automatic selection + +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 +``` + +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. + +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 local combinations fail startup; they do not silently select another path. + +## 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 +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. + +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 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: + +```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 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 f559004bb7..bdeee5a8c8 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](../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 new file mode 100644 index 0000000000..ff276cf83f --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/EmbeddedGremlinRestExecutor.java @@ -0,0 +1,407 @@ +/* + * 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(); + 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" + + "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..794b13d156 --- /dev/null +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinRestExecutorFactory.java @@ -0,0 +1,125 @@ +/* + * 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.NetworkInterface; +import java.net.URI; +import java.net.UnknownHostException; +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) { + return new ForwardGremlinRestExecutor(config); + } + + public static GremlinRestExecutor create(HugeConfig config, GremlinServer server, + String restConfigFile) throws IOException { + Settings settings = server.getServerGremlinExecutor().getSettings(); + URI target = URI.create(config.get(ServerOptions.GREMLIN_SERVER_URL)); + if (!targetsLocalServer(target, settings)) { + return new ForwardGremlinRestExecutor(config); + } + 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()); + } + + 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) throws IOException { + if (targetAddresses.length == 0 || !isLocalAddress(bindAddress)) { + return false; + } + for (InetAddress address : targetAddresses) { + if (!isLocalAddress(address)) { + return false; + } + } + 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) { + 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/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..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,6 +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 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-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..3f02e0cf4d --- /dev/null +++ b/hugegraph-server/hugegraph-test/scripts/gremlin-rest-rocksdb-benchmark.py @@ -0,0 +1,375 @@ +#!/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 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. +""" + +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("--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) + 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.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) + 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 + 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" + 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) + 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", 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) + 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, + "comparison": "previous forward-default build vs automatic local execution", + "distribution_sha256": { + 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"]): + 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", 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": []} + 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/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()); 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..dd2658298c --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/auth/EmbeddedGremlinRestExecutorTest.java @@ -0,0 +1,977 @@ +/* + * 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.InetAddress; +import java.net.InetSocketAddress; +import java.net.NetworkInterface; +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.security.KeyStore; +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 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; +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.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 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), + "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, 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..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 @@ -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,43 @@ 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 testSameProcessTargetRejectsMixedOrRemoteAddresses() + 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.assertTrue(isSameProcessTarget( + new InetAddress[]{ipv6}, ipv4)); + Assert.assertFalse(isSameProcessTarget(new InetAddress[0], ipv4)); + Assert.assertFalse(isSameProcessTarget(new InetAddress[]{ipv4}, remote)); + } + + @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)); + } }