bitflicker64 commented on code in PR #836:
URL:
https://github.com/apache/hugegraph-toolchain/pull/836#discussion_r4353371491
##########
hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/controller/space/GraphSpaceController.java:
##########
@@ -272,6 +276,18 @@ public void delete(@PathVariable("graphspace") String
graphspace) {
E.checkArgument(StringUtils.isNotEmpty(graphspace), "graphspace " +
"must not null");
+ if (this.vermeerService.directConfigured()) {
+ try {
+ List<String> graphs = this.authClient(graphspace,
null).graphs().listGraph();
+ for (String graph : graphs) {
+
this.vermeerService.cleanupDirectGraph(this.authClient(graphspace, graph));
Review Comment:
Minor: The queue check and the unload run per graph in one loop, so a space
delete can stop half way after it has already unloaded some copies.
With graphs `first` and `second`, if `second` has an accepted load or
reload, `deleteGraphIfPresent()` unloads `shop-first` and then throws on
`shop-second`. The space and its admins stay, which is right, but `first` has
lost its Vermeer copy. If the admin then keeps the space, `first` needs a full
reload. `testCleanupFailurePreservesSpaceAndAdministratorBindings` reaches the
same partial state for a failed DELETE but does not assert it.
Run the checks that can refuse first: read `tasks?type=todo` once, reject if
a load or reload is pending for any member graph's exact `shop-<graph>` name (a
`shop-` prefix would also match a space such as `shop-eu`), then unload. That
also turns one queue read per graph into one per space.
##########
hugegraph-hubble/hubble-dist/assembly/travis/vermeer_dev.py:
##########
@@ -0,0 +1,402 @@
+#!/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.
+
+"""Retained cloud development environment: Hubble + Vermeer + HugeGraph
hstore."""
+
+import argparse
+import json
+import math
+import os
+from pathlib import Path
+import secrets
+import shutil
+import signal
+import socket
+import stat
+import subprocess
+import time
+import urllib.request
+
+REPO = Path(__file__).resolve().parents[4]
+SETUP = Path(os.environ.get('VERMEER_DEV_HOME', '/workspace/.onboarding'))
Review Comment:
Minor: This helper (402 lines), `test_vermeer_dev.py` (344),
`vermeer_browser_smoke.cjs` (144) and the 244-line guide add 1,134 lines, plus
six PNGs of about 740 KB. They describe one prepared workspace, not Hubble CI,
and 5ac6ed78 adds `process_args()` and full command-line PID matching on top.
By default `prepare` expects `/workspace/.onboarding` (overridable with
`VERMEER_DEV_HOME`) with `tools/jdk17` and `tools/go1.23.7`, plus a Computer
checkout whose pinned commit is named but not checked. `start` then runs PD,
Store and Server containers. The new hubble-launcher-ci step only tests the
helper's own file and PID guards.
Drop the helper, its test, the CI step, the smoke script, the screenshots
with their README image and `.gitignore` entries, and keep a short manual setup
section in the guide. If they are still wanted, they can come in a separate PR.
##########
hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/space/DirectVermeerService.java:
##########
@@ -0,0 +1,353 @@
+/*
+ *
+ * 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.service.space;
+
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.hash.Hashing;
+import org.apache.hugegraph.config.HugeConfig;
+import org.apache.hugegraph.driver.HugeClient;
+import org.apache.hugegraph.exception.ForbiddenException;
+import org.apache.hugegraph.exception.ExternalException;
+import org.apache.hugegraph.options.HubbleOptions;
+import org.apache.hugegraph.service.auth.AuthModeService;
+import org.apache.hugegraph.structure.Task;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.HubbleUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Service
+public class DirectVermeerService {
+
+ @Autowired
+ private VermeerClient backend;
+ @Autowired
+ private HugeConfig config;
+
+ @Autowired
+ private AuthModeService authMode;
+
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ public boolean configured() {
+ return this.backend.configured();
+ }
+
+ public boolean supported() {
+ // Native HugeGraph loading reads partitions from PD and Store.
+ return this.config.get(HubbleOptions.PD_ENABLED);
+ }
+
+ public boolean permitted(HugeClient client) {
+ return this.authMode.anonymous() || client.auth().isSuperAdmin();
+ }
+
+ public boolean available() {
+ if (!this.configured()) {
+ return false;
+ }
+ try {
+ // Readiness is scoped by the authenticated service credential.
+ Map<String, Object> workers = this.backend.request("GET",
"workers/availability", null);
+ return Boolean.TRUE.equals(workers.get("available"));
+ } catch (VermeerClient.ConfigurationException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ return false;
+ }
+ }
+
+ public Map<String, Object> graphs() {
+ return this.backend.request("GET", "graphs", null);
+ }
+
+ public Map<String, Object> graph(String graphspace, String graph) {
+ String name = name(graphspace, graph);
+ List<Map<String, Object>> graphs = HubbleUtil.uncheckedCast(
+ this.graphs().getOrDefault("graphs", List.of()));
+ for (Map<String, Object> item : graphs) {
+ if (name.equals(item.get("name"))) {
+ return Map.of("graph", item);
+ }
+ }
+ return Map.of();
+ }
+
+ public Map<String, Object> load(HugeClient client, boolean force,
+ Map<String, Object> options) {
+ authorize(client);
+ Map<String, Object> graph = HubbleUtil.uncheckedCast(
+ this.graph(client.getGraphSpaceName(),
client.getGraphName()).get("graph"));
+ if (graph != null && "error".equals(graph.get("status"))) {
+ // A failed initial load leaves a graph that native load/reload
both
+ // reject. Unload that failed graph so the user can retry loading.
+ this.deleteGraph(client);
+ graph = null;
+ }
+ String taskType = graph == null ? "load" : force ? "reload" : "load";
+ Map<String, String> params = new HashMap<>();
+ // Only loader tuning is user-controlled. Targets and credentials are
not.
+ for (String key : List.of("load.parallel", "load.use_outedge",
+ "load.use_out_degree", "load.use_property"))
{
+ if (options != null && options.get(key) != null) {
+ params.put(key, options.get(key).toString());
+ }
+ }
+ params.putIfAbsent("load.parallel", "4");
+ params.putIfAbsent("load.use_outedge", "1");
+ params.putIfAbsent("load.use_out_degree", "1");
+ params.put("load.type", "hugegraph");
+ params.put("load.hg_pd_peers", json(List.of(
+ this.config.get(HubbleOptions.PD_PEERS).split(","))));
+ params.put("load.hugegraph_name", client.getGraphSpaceName() + "/" +
+ client.getGraphName() + "/g");
+ return this.submit(client, taskType, params);
+ }
+
+ public Map<String, Object> compute(HugeClient client, Map<String, Object>
options) {
+ authorize(client);
+ E.checkArgument(options != null && options.get("compute.algorithm") !=
null,
+ "compute.algorithm is required");
+ String algorithm = options.get("compute.algorithm").toString();
+ E.checkArgument(Set.of("pagerank", "wcc", "lpa", "degree",
"closeness_centrality",
+ "betweenness_centrality", "triangle_count",
"kcore", "sssp")
+ .contains(algorithm), "Unsupported Vermeer
algorithm: %s", algorithm);
+ Map<String, String> params = new HashMap<>();
+ options.forEach((key, value) -> {
+ E.checkArgument(key.startsWith("compute.") ||
+ key.startsWith(algorithm + ".") ||
+ "output.need_statistics".equals(key),
+ "Unsupported Vermeer parameter: %s", key);
+ if (value != null) {
+ params.put(key, value.toString());
+ }
+ });
+ // Query results directly. Never persist a HugeGraph password or bearer
+ // token in Vermeer task parameters for asynchronous output.
+ params.put("output.type", "local");
+ params.put("output.need_query", "1");
+ // Native compute is exclusive and LocalWriter truncates its part
files.
+ // Reuse one safe basename per graph instead of retaining files per
run.
+ String output = Hashing.sha256().hashString(json(List.of(
+ client.getGraphSpaceName(), client.getGraphName())),
StandardCharsets.UTF_8).toString();
+ params.put("output.file_path", "./data/hubble/" + output);
+ return this.submit(client, "compute", params);
+ }
+
+ private Map<String, Object> submit(HugeClient client, String type,
+ Map<String, String> params) {
+ Map<String, Object> result = this.backend.request("POST",
"tasks/create",
+ Map.of("graph", name(client), "task_type", type, "params",
params));
+ Map<String, Object> task =
HubbleUtil.uncheckedCast(result.get("task"));
+ return Map.of("task_id", publicId(task));
+ }
+
+ public Task task(HugeClient client, int id, boolean includeResult) {
+ authorize(client);
+ Map<String, Object> raw = this.scopedTask(client, id);
+ Task task = this.convert(raw);
+ if (includeResult && task.success() &&
"compute".equals(raw.get("task_type"))) {
+ Map<String, Object> values;
+ try {
+ values = this.resultValues(id, 0, 1000);
+ } catch (VermeerClient.ValuesNotFoundException e) {
+ // Only confirmed value-cache absence preserves metadata alone.
+ return task;
+ }
+ Map<String, Object> data = new HashMap<>(task.asMap());
+ data.put(Task.P.RESULT, json(values));
+ return this.mapper.convertValue(data, Task.class);
+ }
+ return task;
+ }
+
+ public Map<String, Object> results(HugeClient client, int id, int cursor,
int limit) {
+ authorize(client);
+ this.scopedTask(client, id);
+ E.checkArgument(cursor >= 0 && limit > 0 && limit <= 10000,
+ "Invalid result cursor or limit");
+ return this.resultValues(id, cursor, limit);
+ }
+
+ private Map<String, Object> resultValues(int id, int cursor, int limit) {
+ Map<String, Object> result = new HashMap<>(this.backend.request(
+ "GET", "tasks/value/" + nativeId(id) +
+ "?cursor=" + cursor + "&limit=" + (limit + 1), null));
+ List<?> vertices =
HubbleUtil.uncheckedCast(result.getOrDefault("vertices", List.of()));
+ int count = Math.min(vertices.size(), limit);
+ // Native pages have no total or continuation flag. Peek one value
ahead,
+ // then leave it at the next cursor so an exact final page has no
phantom successor.
+ result.put("vertices", new ArrayList<>(vertices.subList(0, count)));
+ result.put("cursor", cursor + count);
+ result.put("has_more", vertices.size() > limit);
+ return result;
+ }
+
+ public List<Task> tasks(HugeClient client) {
+ if (!this.supported() || !this.permitted(client)) {
+ return List.of();
+ }
+ authorize(client);
Review Comment:
Minor: `tasks()` already checked `permitted()` at line 212, and
`authorize()` checks it again. In authenticated mode each check is an
`isSuperAdmin()` request to Server, and `authorize()` adds `listVertices(1)` as
well. Each 5 s task-list refresh therefore makes three Server calls inside
`tasks()` before Vermeer is asked. For a super admin the vertex read cannot
deny anything the first check allowed, and `AsyncTaskService.list()` has just
read this graph's Server tasks anyway.
The pre-check has to stay, because it returns an empty list rather than an
error for non-admin users. Drop this `authorize(client)` call so the permission
check runs once.
##########
hugegraph-hubble/hubble-be/src/main/java/org/apache/hugegraph/service/space/VermeerClient.java:
##########
@@ -0,0 +1,149 @@
+/*
+ *
+ * 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.service.space;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Map;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.hugegraph.config.HugeConfig;
+import org.apache.hugegraph.exception.ExternalException;
+import org.apache.hugegraph.options.HubbleOptions;
+import org.apache.hugegraph.util.E;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Service
+public class VermeerClient {
+
+ @Autowired
+ private HugeConfig config;
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final HttpClient http = HttpClient.newBuilder()
+ .connectTimeout(Duration.ofSeconds(5)).build();
+
+ public boolean configured() {
+ String url = this.config.get(HubbleOptions.VERMEER_URL);
+ return url != null && !url.isBlank();
+ }
+
+ public Map<String, Object> request(String method, String path,
+ Map<String, ?> body) {
+ try {
+ E.checkState(this.configured(), "Configure vermeer.url first");
+ URI base = URI.create(this.config.get(HubbleOptions.VERMEER_URL));
+ E.checkArgument(("http".equals(base.getScheme()) ||
+ "https".equals(base.getScheme())) &&
+ base.getHost() != null && base.getUserInfo() ==
null &&
+ base.getQuery() == null && base.getFragment() ==
null,
+ "Invalid vermeer.url");
+ HttpRequest.Builder request = HttpRequest.newBuilder(
+ URI.create(base.toString().replaceAll("/+$", "") +
+ "/api/v1/" + path))
+ .timeout(Duration.ofSeconds(30))
+ .header("Content-Type", "application/json");
+ String tokenFile =
this.config.get(HubbleOptions.VERMEER_TOKEN_FILE);
+ if (tokenFile != null && !tokenFile.isBlank()) {
+ String token;
+ try {
+ token = Files.readString(Path.of(tokenFile)).trim();
+ } catch (IOException e) {
+ throw new ConfigurationException("Unable to read Vermeer
token file");
+ }
+ E.checkState(!token.isEmpty(), "Vermeer token file is empty");
+ request.header("Authorization", "Bearer " + token);
+ }
+ request.method(method, body == null ?
+ HttpRequest.BodyPublishers.noBody() :
+ HttpRequest.BodyPublishers.ofString(
+ this.mapper.writeValueAsString(body)));
+ HttpResponse<String> response = this.http.send(request.build(),
+ HttpResponse.BodyHandlers.ofString());
+ if (response.statusCode() < 200 || response.statusCode() >= 300) {
+ if (response.statusCode() == 401 || response.statusCode() ==
403) {
+ throw new ConfigurationException("Vermeer service
authentication failed");
+ }
+ if (this.valuesNotFound(method, path, response)) {
+ throw new ValuesNotFoundException();
+ }
+ throw new ExternalException("Vermeer request failed (HTTP " +
Review Comment:
Minor: A 404 from Vermeer is treated as a temporary outage. Hubble then
retries forever instead of reporting a configuration error.
`workers/availability` only exists in apache/hugegraph-computer#366
(b5c1693f). A Master built from apache/hugegraph-computer master (413f95fa) has
no such route, so it returns 404, and a `vermeer.url` with a wrong path or port
that reaches another HTTP service does the same. This line throws a plain
`ExternalException`. `DirectVermeerService.available()` turns that into
`false`, and `VermeerService.getVermeer()` returns `retryable=true`.
GraphAnalysis then polls every 5 s and never shows the error banner, and
nothing is logged. The readiness tests cover 503, 401, 403 and 200, but not 404.
Throw `ConfigurationException` for a 404 here, so the operator sees that the
Vermeer version or URL is wrong. No Vermeer handler returns 404 itself, so only
unknown routes are affected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
