diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java index 94169e4a90..484e60ed3b 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java @@ -25,6 +25,7 @@ import java.util.Date; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -39,9 +40,13 @@ import org.apache.hugegraph.auth.HugePermission; import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.exception.NotFoundException; +import org.apache.hugegraph.meta.GraphStatusAggregate; +import org.apache.hugegraph.meta.GraphStatusEntry; import org.apache.hugegraph.space.GraphSpace; import org.apache.hugegraph.type.define.GraphMode; import org.apache.hugegraph.type.define.GraphReadMode; +import org.apache.hugegraph.type.define.GraphStatus; import org.apache.hugegraph.util.ConfigUtil; import org.apache.hugegraph.util.E; import org.apache.hugegraph.util.JsonUtil; @@ -82,6 +87,12 @@ public class GraphsAPI extends API { private static final String GRAPH_ACTION = "action"; private static final String UPDATE = "update"; private static final String GRAPH_ACTION_RELOAD = "reload"; + /** + * How long the status of a server that is no longer registered is kept, + * long enough to outlast a registration that lapsed while its server was + * merely slow + */ + private static final long STATUS_STALE_AFTER = 10 * 60 * 1000L; private static Map convConfig(Map config) { Map result = new HashMap<>(config.size()); @@ -247,6 +258,95 @@ public Object get(@Context GraphManager manager, return ImmutableMap.of("name", g.name(), "backend", g.backend()); } + @GET + @Timed + @Path("{name}/status") + @Produces(APPLICATION_JSON_WITH_CHARSET) + @RolesAllowed({"space_member", "$owner=$name"}) + public Object status(@Context GraphManager manager, + @Parameter(description = "The graph space name") + @PathParam("graphspace") String graphSpace, + @Parameter(description = "The graph name") + @PathParam("name") String name, + @Context SecurityContext sc) { + LOG.debug("Get status of graph '{}' in graph space '{}'", + name, graphSpace); + + /* + * A graph that is still loading is not bound to this server yet, so + * graph(manager, graphSpace, name) would answer 404 for exactly the + * case this API is meant to report. It's also what verifies the + * permission on the graph itself for the other reads of this + * resource, so the role of the caller is checked here instead + */ + space(manager, graphSpace); + String role = RequiredPerm.roleFor(graphSpace, name, + HugePermission.READ); + if (!sc.isUserInRole(role) && !isAdminManager(manager)) { + throw new ForbiddenException(String.format( + "The user is not allowed to read graph '%s'", name)); + } + + GraphStatusAggregate aggregate; + if (manager.isPDEnabled()) { + /* + * The status of a dropped graph is removed on every drop path, + * but a removal that couldn't be done leaves an entry behind, and + * a graph that no longer exists has to answer 404 either way. The + * config of a graph is written before its creation returns, so a + * graph that is still loading is found here + */ + if (missing(manager, graphSpace, name)) { + throw new NotFoundException(String.format( + "Graph '%s' does not exist", name)); + } + Map status = manager.graphStatus( + graphSpace, name); + Set servers = manager.serviceServers(graphSpace); + aggregate = GraphStatusAggregate.of(status.values(), servers, + STATUS_STALE_AFTER, + System.currentTimeMillis()); + } else { + // Without PD nothing is reported, this server is the whole cluster + if (localGraph(manager, graphSpace, name) == null) { + throw new NotFoundException(String.format( + "Graph '%s' does not exist", name)); + } + GraphStatusEntry entry = new GraphStatusEntry( + manager.serverId(), GraphStatus.READY, null, + System.currentTimeMillis()); + aggregate = GraphStatusAggregate.of( + Collections.singletonList(entry), 1); + } + + Map result = new LinkedHashMap<>(); + result.put("graphspace", graphSpace); + result.put("graph", name); + result.putAll(aggregate.asMap()); + return result; + } + + /** + * Whether the graph is known to be gone. Only a metadata read that + * answered is taken as an answer: a read that failed leaves the graph + * reported as it was last seen rather than as dropped + */ + private static boolean missing(GraphManager manager, String graphSpace, + String name) { + // The local map first, it answers without asking the cluster metadata + if (localGraph(manager, graphSpace, name) != null) { + return false; + } + Boolean configured = manager.graphConfigExists(graphSpace, name); + return configured != null && !configured; + } + + private static HugeGraph localGraph(GraphManager manager, + String graphSpace, String name) { + return manager.graph(String.join(GraphManager.DELIMITER, graphSpace, + name)); + } + @POST @Timed @Path("{name}/default") diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java index 0f5881b1a5..20d2bd8e8c 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java @@ -130,14 +130,27 @@ private void injectGraph(HugeGraph graph) { GremlinExecutor executor = this.getServerGremlinExecutor() .getGremlinExecutor(); - manager.putGraph(name, graph); + try { + manager.putGraph(name, graph); + + GraphTraversalSource g = manager.getGraph(name).traversal(); + manager.putTraversalSource(G_PREFIX + name, g); - GraphTraversalSource g = manager.getGraph(name).traversal(); - manager.putTraversalSource(G_PREFIX + name, g); + Whitebox.invoke(executor, "globalBindings", + new Class[]{String.class, Object.class}, + "put", name, graph); + } catch (Throwable e) { + /* + * The graph create event is notified without waiting for it, so a + * failure here reaches nobody: tell the graph manager, otherwise + * the graph is left reported as loading for good + */ + this.eventHub.notify(Events.GRAPH_BIND_FAILED, graph); + throw e; + } - Whitebox.invoke(executor, "globalBindings", - new Class[]{String.class, Object.class}, - "put", name, graph); + // The graph can serve requests only after all bindings are done + this.eventHub.notify(Events.GRAPH_BOUND, graph); } private void removeGraph(String name) { diff --git a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java index f716285c67..9e6fec24a4 100644 --- a/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java +++ b/hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java @@ -23,6 +23,8 @@ import java.io.IOException; import java.io.StringWriter; +import java.net.InetAddress; +import java.net.URI; import java.text.ParseException; import java.util.Arrays; import java.util.Collections; @@ -36,6 +38,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -65,6 +68,7 @@ import org.apache.hugegraph.config.HugeConfig; import org.apache.hugegraph.config.ServerOptions; import org.apache.hugegraph.config.TypedOption; +import org.apache.hugegraph.event.Event; import org.apache.hugegraph.event.EventHub; import org.apache.hugegraph.exception.ExistedException; import org.apache.hugegraph.exception.NotFoundException; @@ -80,6 +84,7 @@ import org.apache.hugegraph.masterelection.RoleElectionOptions; import org.apache.hugegraph.masterelection.RoleElectionStateMachine; import org.apache.hugegraph.masterelection.StandardRoleListener; +import org.apache.hugegraph.meta.GraphStatusEntry; import org.apache.hugegraph.meta.MetaDriver; import org.apache.hugegraph.meta.MetaManager; import org.apache.hugegraph.meta.PdMetaDriver; @@ -115,6 +120,7 @@ import org.apache.hugegraph.type.define.CollectionType; import org.apache.hugegraph.type.define.GraphMode; import org.apache.hugegraph.type.define.GraphReadMode; +import org.apache.hugegraph.type.define.GraphStatus; import org.apache.hugegraph.type.define.NodeRole; import org.apache.hugegraph.util.ConfigUtil; import org.apache.hugegraph.util.E; @@ -150,6 +156,9 @@ public final class GraphManager { public static final int NICKNAME_MAX_LENGTH = 48; public static final String DELIMITER = "-"; public static final String NAMESPACE_CREATE = "namespace_create"; + private static final String UNKNOWN_SERVER_ID = "unknown-server"; + private static final String BIND_FAILED_MESSAGE = "GremlinBindingFailure"; + private static final int SERVER_ID_MAX_LENGTH = 128; private static final Logger LOG = Log.logger(GraphManager.class); private KvStore kvStore; @@ -179,6 +188,8 @@ public final class GraphManager { private final HugeConfig conf; private final EventHub eventHub; private final String url; + private final String serverId; + private final boolean serverIdDistinct; private final Set serverUrlsToPd; private final Boolean serverDeployInK8s; private final HugeConfig config; @@ -198,6 +209,12 @@ public GraphManager(HugeConfig conf, EventHub hub) { this.config = conf; this.url = conf.get(ServerOptions.REST_SERVER_URL); + String derivedServerId = initServerId(conf); + this.serverIdDistinct = derivedServerId != null; + this.serverId = this.serverIdDistinct ? derivedServerId : + sanitizeServerId( + conf.get(ServerOptions.REST_SERVER_URL)); + LOG.info("The id of this server is '{}'", this.serverId); this.serverUrlsToPd = new HashSet<>(Arrays.asList( conf.get(ServerOptions.SERVER_URLS_TO_PD).split(","))); this.serverDeployInK8s = @@ -362,12 +379,44 @@ private void loadMetaFromPD() { this.loadGraphSpaces(); this.kvStore = this.kvStoreInit(); + + /* + * Drop what this server reported before it restarted, while it is not + * registered yet. Its id is stable across a restart, so the status it + * left behind reads as the status of the server that is starting, and + * a graph it reported ready would be counted ready again before it + * has opened it + */ + this.clearOwnGraphStatus(); + this.loadServices(); this.loadGraphsFromMeta(this.graphConfigs()); this.listenMetaChanges(); } + /** + * Removes the status this server reported for every graph it may serve. + * Best effort: a status that couldn't be removed makes a graph look ready + * before it is, which is worth a warning but not a failed startup + */ + private void clearOwnGraphStatus() { + try { + for (String graphSpace : this.graphSpaces.keySet()) { + if (!servesGraphSpace(this.serviceGraphSpace, graphSpace)) { + continue; + } + for (String graph : this.graphs(graphSpace)) { + this.removeGraphStatus(graphSpace, graph); + } + } + } catch (Throwable e) { + LOG.warn("Failed to clear the graph status of this server: {}", + e.getMessage()); + LOG.debug("Failed to clear the graph status of this server", e); + } + } + public void initAdminUserIfNeeded(String password) { HugeUser user = new HugeUser("admin"); user.nickname("超级管理员"); @@ -864,6 +913,13 @@ public void clearGraphSpace(String name) { } } + /* + * The loop above walks the graphs this server has open, the graphs of + * the space this server never opened leave their status behind, so + * the whole subtree of the space is dropped + */ + this.clearGraphSpaceStatus(name); + // Clear all services for (String key : this.services.keySet()) { if (key.startsWith(name)) { @@ -916,15 +972,20 @@ private void registerServiceToPd(String graphSpace, Service service) { licenseValid = true; } }) - .setLabelMap(ImmutableMap.of( - PdRegisterLabel.REGISTER_TYPE.name(), - PdRegisterType.NODE_PORT.name(), - PdRegisterLabel.GRAPHSPACE.name(), graphSpace, - PdRegisterLabel.SERVICE_NAME.name(), service.name(), - PdRegisterLabel.SERVICE_ID.name(), service.serviceId(), - PdRegisterLabel.cores.name(), - String.valueOf(Runtime.getRuntime().availableProcessors()) - )).setVersion(CoreVersion.VERSION.toString()); + .setLabelMap(ImmutableMap.builder() + .put(PdRegisterLabel.REGISTER_TYPE.name(), + PdRegisterType.NODE_PORT.name()) + .put(PdRegisterLabel.GRAPHSPACE.name(), graphSpace) + .put(PdRegisterLabel.SERVICE_NAME.name(), + service.name()) + .put(PdRegisterLabel.SERVICE_ID.name(), + service.serviceId()) + .put(PdRegisterLabel.SERVER_ID.name(), + this.serverId()) + .put(PdRegisterLabel.cores.name(), + String.valueOf(Runtime.getRuntime() + .availableProcessors())) + .build()).setVersion(CoreVersion.VERSION.toString()); String pdServiceId = register.registerService(config); service.pdServiceId(pdServiceId); @@ -963,7 +1024,8 @@ public void registerK8StoPd(Service service) throws Exception { PdRegisterLabel.REGISTER_TYPE.name(), PdRegisterType.NODE_PORT.name(), PdRegisterLabel.GRAPHSPACE.name(), this.serviceGraphSpace, PdRegisterLabel.SERVICE_NAME.name(), service.name(), - PdRegisterLabel.SERVICE_ID.name(), service.serviceId() + PdRegisterLabel.SERVICE_ID.name(), service.serviceId(), + PdRegisterLabel.SERVER_ID.name(), this.serverId() )); String ddsHost = this.metaManager.getDDSHost(); @@ -1276,6 +1338,68 @@ public HugeGraph createGraph(String graphSpace, String name, String creator, } E.checkArgumentNotNull(name, "The graph name can't be null"); checkGraphName(name); + + /* + * From here on this server is opening the graph, so every way out has + * to leave a terminal status behind: an entry that stays LOADING is + * indistinguishable from a server that is still working on the graph + */ + this.reportGraphStatus(graphSpace, name, GraphStatus.LOADING, null); + AtomicBoolean registered = new AtomicBoolean(); + try { + return this.createPdGraph(graphSpace, name, creator, configs, + init, grpcThread, registered); + } catch (Throwable e) { + LOG.error("Graph '{}' of graph space '{}' can't be created", + name, graphSpace, e); + /* + * An attempt that registered the graph and failed afterwards + * leaves it half open: this server would answer for it while it + * reports it failed, and it would never open it again, so the + * graph would stay reported failed for good. Take it back out so + * that opening it can be tried again + */ + if (registered.get()) { + this.unregisterFailedGraph(graphSpace, name); + this.reportGraphStatus(graphSpace, name, GraphStatus.FAILED, + statusMessage(e)); + } else if (!this.graphs.containsKey( + spaceGraphName(graphSpace, name))) { + this.reportGraphStatus(graphSpace, name, GraphStatus.FAILED, + statusMessage(e)); + } + /* + * The graph is registered by an attempt that isn't this one, it + * is served and its status belongs to that attempt + */ + throw e; + } + } + + /** + * Takes a graph this server registered and then failed to finish opening + * back out, so that it is neither served nor kept from being opened + * again. Best effort: a graph that can't be closed is still removed + */ + private void unregisterFailedGraph(String graphSpace, String name) { + Graph graph = this.graphs.remove(spaceGraphName(graphSpace, name)); + if (!(graph instanceof HugeGraph)) { + return; + } + try { + ((HugeGraph) graph).close(); + } catch (Throwable e) { + LOG.warn("Failed to close graph '{}' of graph space '{}': {}", + name, graphSpace, e.getMessage()); + LOG.debug("Failed to close the graph that can't be created", e); + } + } + + private HugeGraph createPdGraph(String graphSpace, String name, + String creator, + Map configs, boolean init, + boolean grpcThread, + AtomicBoolean registered) { String nickname; if (configs.get("nickname") != null) { nickname = configs.get("nickname").toString(); @@ -1348,28 +1472,286 @@ public HugeGraph createGraph(String graphSpace, String name, String creator, this.metaManager.notifyGraphAdd(graphSpace, name); } this.graphs.put(graphName, graph); + registered.set(true); if (!grpcThread) { this.metaManager.updateGraphSpaceConfig(graphSpace, gs); } - // Let gremlin server and rest server context add graph - this.eventHub.notify(Events.GRAPH_CREATE, graph); - if (init) { String schema = propConfig.getString( CoreOptions.SCHEMA_INIT_TEMPLATE.name()); - if (schema == null || schema.isEmpty()) { - return graph; + if (schema != null && !schema.isEmpty()) { + String schemas = this.schemaTemplate(graphSpace, + schema).schema(); + prepareSchema(graph, schemas); } - String schemas = this.schemaTemplate(graphSpace, schema).schema(); - prepareSchema(graph, schemas); } + + /* + * Let gremlin server and rest server context add graph. Notified last: + * the event is handled without waiting for it and its handler reports + * the graph ready, so anything that can still fail has to run before + * it, otherwise a failed creation could be reported ready + */ + this.eventHub.notify(Events.GRAPH_CREATE, graph); + if (grpcThread) { HugeGraphAuthProxy.resetContext(); } return graph; } + /** + * The id this server reports the status of its graphs with. It has to be + * stable across a restart of the same server and different from the id of + * every other server of the cluster. + */ + public String serverId() { + return this.serverId; + } + + private static String initServerId(HugeConfig conf) { + String configured = conf.get(ServerOptions.SERVER_ID); + if (StringUtils.isNotBlank(configured)) { + return sanitizeServerId(configured); + } + /* + * The rest server url is a bind address rather than an identity, it's + * the same for every replica of a container image. The host name is + * what tells containers and pods apart, the port is appended for the + * servers that share a host + */ + String host = localHostName(); + if (host != null) { + int port = restServerPort(conf); + return sanitizeServerId(port > 0 ? host + "_" + port : host); + } + /* + * Nothing here tells this server apart from another one started from + * the same image: the rest server url is a bind address and is the + * same for all of them. The caller still needs an id to key the + * status by, it just can't be trusted to be unique + */ + LOG.error("The id of this server can't be derived, the host name " + + "can't be resolved and '{}' is not set. The status of the " + + "graphs of this server is reported under an id that is " + + "shared with every server started the same way, so this " + + "server is left out of the readiness of its graphs. Set " + + "'{}' to a value that is unique in the cluster", + ServerOptions.SERVER_ID.name(), + ServerOptions.SERVER_ID.name()); + return null; + } + + private static String localHostName() { + try { + String host = InetAddress.getLocalHost().getHostName(); + return StringUtils.isBlank(host) ? null : host; + } catch (Exception e) { + LOG.warn("Failed to resolve the local host name", e); + return null; + } + } + + private static int restServerPort(HugeConfig conf) { + try { + return URI.create(conf.get(ServerOptions.REST_SERVER_URL)) + .getPort(); + } catch (Exception e) { + return -1; + } + } + + private static String sanitizeServerId(String id) { + // The status meta key is joined by '/', keep the id to safe chars + String sanitized = StringUtils.trimToEmpty(id) + .replaceAll("[^A-Za-z0-9._-]", "_"); + if (sanitized.length() > SERVER_ID_MAX_LENGTH) { + sanitized = sanitized.substring(0, SERVER_ID_MAX_LENGTH); + } + return sanitized.isEmpty() ? UNKNOWN_SERVER_ID : sanitized; + } + + /** + * The status message is stored in the cluster metadata and served to + * every member of the graph space, while opening a graph is allowed to + * the owner of the graph space only. Exception messages of the backend + * carry data paths and connection strings, so only the type of the + * failure is published; the exception itself is left to the server log. + */ + private static String statusMessage(Throwable e) { + return e.getClass().getSimpleName(); + } + + private void reportGraphStatus(String graphSpace, String name, + GraphStatus status, String message) { + if (!this.isPDEnabled()) { + return; + } + try { + GraphStatusEntry entry = new GraphStatusEntry( + this.serverId(), status, message, + System.currentTimeMillis()); + this.metaManager.updateGraphStatus(graphSpace, name, entry); + } catch (Throwable e) { + // Reporting is best-effort, it must never break graph creation + LOG.warn("Failed to report status {} of graph '{}-{}'", + status, graphSpace, name, e); + } + } + + private void clearGraphStatus(String graphSpace, String name) { + if (!this.isPDEnabled()) { + return; + } + try { + this.metaManager.clearGraphStatus(graphSpace, name); + } catch (Throwable e) { + // Clearing is best-effort, it must never break dropping a graph + LOG.warn("Failed to clear status of graph '{}-{}'", + graphSpace, name, e); + } + } + + /** + * Remove the status this server reported for a graph, called where this + * server stops serving the graph but the cluster keeps it. + */ + private void removeGraphStatus(String graphSpace, String name) { + if (!this.isPDEnabled()) { + return; + } + try { + this.metaManager.removeGraphStatus(graphSpace, name, + this.serverId()); + } catch (Throwable e) { + LOG.warn("Failed to remove status of graph '{}-{}' reported by " + + "server '{}'", graphSpace, name, this.serverId(), e); + } + } + + private void clearGraphSpaceStatus(String graphSpace) { + if (!this.isPDEnabled()) { + return; + } + try { + this.metaManager.clearGraphSpaceStatus(graphSpace); + } catch (Throwable e) { + LOG.warn("Failed to clear the graph status of graph space '{}'", + graphSpace, e); + } + } + + public Map graphStatus(String graphSpace, + String graph) { + if (!this.isPDEnabled()) { + return Collections.emptyMap(); + } + try { + return this.metaManager.getGraphStatus(graphSpace, graph); + } catch (Throwable e) { + // Clients poll the status, keep the stack trace out of the log + LOG.warn("Failed to get status of graph '{}-{}': {}", + graphSpace, graph, e.getMessage()); + LOG.debug("Failed to get status of graph", e); + return Collections.emptyMap(); + } + } + + /** + * Whether a graph is registered in the cluster metadata, or null when the + * metadata can't be read. Telling the two apart matters: a graph that + * can't be looked up is not a graph that doesn't exist, and the caller of + * an API clients poll would take the second for a final answer. + *

+ * Unlike listing the graphs of a graph space this reads a single key, the + * caller is on a polled path. + */ + public Boolean graphConfigExists(String graphSpace, String name) { + if (!this.isPDEnabled()) { + return Boolean.FALSE; + } + try { + return this.metaManager.getGraphConfig(graphSpace, name) != null; + } catch (Throwable e) { + LOG.warn("Failed to get the config of graph '{}-{}': {}", + graphSpace, name, e.getMessage()); + LOG.debug("Failed to get the config of graph", e); + return null; + } + } + + /** + * The ids of the servers registered to serve the graphs of a graph space, + * or an empty set when they can't be listed. They are the ids the servers + * report their graph status with, so the reported status can be matched + * against the servers that are actually part of the cluster. + *

+ * An id that isn't known to be unique can't be matched against anything: + * the servers that share it are one id to the cluster and one of them + * reporting ready would answer for all of them. The servers are reported + * as unknown instead, which keeps the graphs of this graph space below + * ready rather than answering ready for a server that isn't. + */ + public Set serviceServers(String graphSpace) { + if (!this.isPDEnabled() || !this.serverIdDistinct) { + return Collections.emptySet(); + } + try { + return this.serviceServerIds(graphSpace); + } catch (Throwable e) { + // Clients poll the status, keep the stack trace out of the log + LOG.warn("Failed to list the servers of graph space '{}': {}", + graphSpace, e.getMessage()); + LOG.debug("Failed to list the servers of graph space", e); + return Collections.emptySet(); + } + } + + /** + * The ids of the servers of this service that serve the given graph + * space. A server registers one node per url it listens on, so the nodes + * are grouped by the id each server reports its graph status with. + */ + private Set serviceServerIds(String graphSpace) { + Query query = Query.newBuilder() + .setAppName(this.cluster) + .putLabels(PdRegisterLabel.REGISTER_TYPE.name(), + PdRegisterType.NODE_PORT.name()) + .putLabels(PdRegisterLabel.SERVICE_NAME.name(), + this.serviceID) + .build(); + NodeInfos nodeInfos = this.pdClient.getNodeInfos(query); + Set servers = new HashSet<>(); + for (NodeInfo nodeInfo : nodeInfos.getInfoList()) { + Map labels = nodeInfo.getLabelsMap(); + String registered = labels.get(PdRegisterLabel.GRAPHSPACE.name()); + if (!servesGraphSpace(registered, graphSpace)) { + continue; + } + String server = labels.get(PdRegisterLabel.SERVER_ID.name()); + /* + * A server of an older version registers no id, count it by + * address instead: that can only make the number larger, never + * let a graph look ready before every server reported + */ + servers.add(StringUtils.isNotBlank(server) ? server : + nodeInfo.getAddress()); + } + return servers; + } + + private static boolean servesGraphSpace(String registered, + String graphSpace) { + if (StringUtils.isBlank(registered)) { + // A server of unknown graph space is counted rather than skipped + return true; + } + // A server registered to DEFAULT serves the graphs of every space + return DEFAULT_GRAPH_SPACE_SERVICE_NAME.equals(registered) || + registered.equals(graphSpace); + } + public Set graphs() { return Collections.unmodifiableSet(this.graphs.keySet()); } @@ -1565,6 +1947,16 @@ private void loadGraph(String name, String graphConfPath) { LOG.info("Graph '{}' was successfully configured via '{}'", name, graphConfPath); + /* + * The status of a graph of the local config directory is deliberately + * not reported. Its name is taken from a file name and the config + * directory is read whether or not this server takes its graphs from + * it, so a local file named after a graph of the cluster would report + * over the graph of the cluster. A local graph is known to this + * server only, and the status API answers for it from the local + * instance rather than from the cluster metadata + */ + if (this.requireAuthentication() && !(graph instanceof HugeGraphAuthProxy)) { LOG.warn("You may need to support access control for '{}' with {}", @@ -1757,11 +2149,47 @@ private void listenChanges() { this.graphs.remove(graph.spaceGraphName()); return null; }); + /* + * listenChanges is called both by the constructor and by init, and a + * listener is appended every time, so a graph would be reported ready + * twice. Register the listener only when it isn't registered yet + */ + if (this.eventHub.containsListener(Events.GRAPH_BOUND)) { + return; + } + this.eventHub.listen(Events.GRAPH_BOUND, event -> { + return this.reportBindResult(event, GraphStatus.READY, null); + }); + this.eventHub.listen(Events.GRAPH_BIND_FAILED, event -> { + return this.reportBindResult(event, GraphStatus.FAILED, + BIND_FAILED_MESSAGE); + }); + } + + private Object reportBindResult(Event event, GraphStatus status, + String message) { + LOG.debug("RestServer accepts event '{}'", event.name()); + event.checkArgs(HugeGraph.class); + HugeGraph graph = (HugeGraph) event.args()[0]; + /* + * Can't call graph.name() here, it verifies permission while the + * event thread carries no authentication context + */ + String[] parts = graph.spaceGraphName().split(DELIMITER); + if (parts.length < 2) { + LOG.warn("The graph name format is incorrect: {}", + graph.spaceGraphName()); + return null; + } + this.reportGraphStatus(parts[0], parts[1], status, message); + return null; } private void unlistenChanges() { this.eventHub.unlisten(Events.GRAPH_CREATE); this.eventHub.unlisten(Events.GRAPH_DROP); + this.eventHub.unlisten(Events.GRAPH_BOUND); + this.eventHub.unlisten(Events.GRAPH_BIND_FAILED); } private void listenMetaChanges() { @@ -2028,6 +2456,8 @@ public void dropGraph(String graphSpace, String name, boolean clear) { throw new HugeException( "Failed to remove graph config of '%s'", name, e); } + // The graph leaves the cluster, no server reports it any more + this.clearGraphStatus(graphSpace, name); /** * close task scheduler before clear data, @@ -2049,6 +2479,12 @@ public void dropGraph(String graphSpace, String name, boolean clear) { } catch (Exception e) { LOG.warn("Failed to close graph", e); } + } else { + /* + * The graph stays in the cluster and only this server stops + * serving it, so only the status of this server goes away + */ + this.removeGraphStatus(graphSpace, name); } GraphSpace gs = this.graphSpace(graphSpace); if (!grpcThread) { @@ -2333,6 +2769,13 @@ private enum PdRegisterLabel { GRAPHSPACE, SERVICE_NAME, SERVICE_ID, + /* + * The id of the server behind the node, one server registers one node + * per url it listens on. It's the id the server reports the status of + * its graphs with, so the nodes of a service can be counted in the + * same unit the reported status is counted in + */ + SERVER_ID, cores } @@ -2401,6 +2844,8 @@ private void graphAddHandler(T response) { this.metaManager.getGraphConfig(parts[0], parts[1]); if (config == null) { LOG.error("The graph config not exist: {}", graphName); + this.reportGraphStatus(parts[0], parts[1], GraphStatus.FAILED, + "The graph config not exist"); continue; } Object objc = config.get("creator"); @@ -2423,7 +2868,13 @@ private void graphAddHandler(T response) { if (graph.tx().isOpen()) { graph.tx().close(); } - } catch (HugeException e) { + } catch (Throwable e) { + /* + * Not only HugeException: opening a graph also fails with an + * IllegalArgumentException of a checked argument, and such a + * failure has to be handled like any other one. The status of + * the graph is reported by createGraph itself + */ if (!this.startIgnoreSingleGraphError) { throw e; } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/GraphStatusAggregate.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/GraphStatusAggregate.java new file mode 100644 index 0000000000..28a0109ed4 --- /dev/null +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/GraphStatusAggregate.java @@ -0,0 +1,263 @@ +/* + * 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.meta; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.hugegraph.type.define.GraphStatus; + +/** + * The status of one graph across all the servers that reported it. + *

+ * The aggregate is a readiness gate, so it fails closed: it answers READY + * only when every server that is expected to serve the graph reported READY. + * When the number of expected servers is unknown the aggregate can't tell a + * complete cluster from a partial one, and answers LOADING. + */ +public final class GraphStatusAggregate { + + public static final String STATUS_KEY = "status"; + public static final String READY_COUNT_KEY = "ready_count"; + public static final String TOTAL_COUNT_KEY = "total_count"; + public static final String EXPECTED_COUNT_KEY = "expected_count"; + public static final String SERVERS_KEY = "servers"; + + /** + * Rendered instead of a status name when no server reported yet, it's not + * a state a server can be in so it's kept out of {@link GraphStatus}. + */ + private static final String STATUS_UNKNOWN = "UNKNOWN"; + + /** + * The number of servers expected to report is unknown + */ + private static final int UNKNOWN_EXPECTED = 0; + + private static final Comparator BY_SERVER = + Comparator.comparing(GraphStatusEntry::server, + Comparator.nullsLast( + Comparator.naturalOrder())); + + private final GraphStatus status; + private final int readyCount; + private final int expected; + private final List entries; + + private GraphStatusAggregate(GraphStatus status, int readyCount, + int expected, + List entries) { + this.status = status; + this.readyCount = readyCount; + this.expected = expected; + this.entries = entries; + } + + /** + * Aggregates the reported status against the servers that are currently + * registered in the cluster. A server keeps no state of its own, so the + * status it reported outlives it: an entry left behind by a server that + * is gone would otherwise hold a healthy graph down forever when it reads + * FAILED, or make up a quorum the running servers never reached when it + * reads READY. + *

+ * A server missing from the registration isn't taken as gone right away: + * a registration is refreshed periodically and lapses for a while when a + * server is merely slow, and dropping the entry of a server that is in + * fact still loading would answer READY too early. Only an entry that is + * both unregistered and older than {@code staleAfter} is dropped, so the + * reading errs on the side of holding the graph back. + * + * @param entries the status reported by each server, may be null + * @param liveServers the ids of the servers registered for the graph + * space, empty or null when they can't be listed, in + * which case nothing is dropped and the aggregate + * can't reach READY + * @param staleAfter how long an unregistered server's status is kept, in + * milliseconds + * @param now the current time, in milliseconds + */ + public static GraphStatusAggregate of(Collection entries, + Collection liveServers, + long staleAfter, long now) { + if (liveServers == null || liveServers.isEmpty()) { + /* + * The registered servers are the only thing that tells a server + * that is gone from one that is merely slow, so without them + * nothing can be dropped and the status stays below ready. The + * entries are still rendered: a graph of a cluster whose servers + * can't be listed is exactly when the caller wants to see who + * reported what + */ + return of(entries, UNKNOWN_EXPECTED); + } + Set live = new HashSet<>(liveServers); + List reported = new ArrayList<>(); + Set readyServers = new HashSet<>(); + boolean failed = false; + boolean loading = false; + if (entries != null) { + for (GraphStatusEntry entry : entries) { + if (entry == null) { + continue; + } + boolean registered = live.contains(entry.server()); + boolean stale = now - entry.updateTime() > staleAfter; + if (!registered && stale) { + continue; + } + reported.add(entry); + GraphStatus entryStatus = entry.status(); + if (entryStatus == GraphStatus.FAILED) { + failed = true; + } else if (entryStatus == GraphStatus.READY) { + readyServers.add(entry.server()); + } else { + // An unset status is not a guarantee, treat it as loading + loading = true; + } + } + } + reported.sort(BY_SERVER); + + /* + * Which servers reported ready, not how many: the status left behind + * by a server that is gone must never stand in for a registered + * server that never reported, otherwise the graph reads ready while + * one of the servers of the cluster can't serve it + */ + GraphStatus status; + if (reported.isEmpty()) { + status = null; + } else if (failed) { + status = GraphStatus.FAILED; + } else if (loading || !readyServers.containsAll(live)) { + status = GraphStatus.LOADING; + } else { + status = GraphStatus.READY; + } + return new GraphStatusAggregate(status, readyServers.size(), + live.size(), + Collections.unmodifiableList(reported)); + } + + /** + * @param entries the status reported by each server, may be null or empty + * @param expected the number of servers expected to report, a value less + * than or equal to zero means the number is unknown and + * the aggregate can't reach READY + */ + public static GraphStatusAggregate of(Collection entries, + int expected) { + List sorted = new ArrayList<>(); + if (entries != null) { + for (GraphStatusEntry entry : entries) { + if (entry != null) { + sorted.add(entry); + } + } + } + sorted.sort(BY_SERVER); + + int ready = 0; + boolean failed = false; + boolean loading = false; + for (GraphStatusEntry entry : sorted) { + GraphStatus entryStatus = entry.status(); + if (entryStatus == GraphStatus.FAILED) { + failed = true; + } else if (entryStatus == GraphStatus.READY) { + ready++; + } else { + // An unset status is not a guarantee, treat it as loading + loading = true; + } + } + + GraphStatus status; + if (sorted.isEmpty()) { + status = null; + } else if (failed) { + status = GraphStatus.FAILED; + } else if (loading) { + status = GraphStatus.LOADING; + } else if (expected > 0 && ready >= expected) { + status = GraphStatus.READY; + } else { + /* + * Either a server has not reported yet, or the number of servers + * that should report is unknown. Both mean the graph can't be + * guaranteed to answer on every server of the cluster + */ + status = GraphStatus.LOADING; + } + + return new GraphStatusAggregate(status, ready, expected, + Collections.unmodifiableList(sorted)); + } + + public GraphStatus status() { + return this.status; + } + + public int readyCount() { + return this.readyCount; + } + + public int totalCount() { + return this.entries.size(); + } + + public int expected() { + return this.expected; + } + + public boolean expectedKnown() { + return this.expected > 0; + } + + public List entries() { + return this.entries; + } + + public Map asMap() { + Map map = new LinkedHashMap<>(); + map.put(STATUS_KEY, this.status == null ? STATUS_UNKNOWN : + this.status.name()); + map.put(READY_COUNT_KEY, this.readyCount); + map.put(TOTAL_COUNT_KEY, this.totalCount()); + // Always rendered, null tells the client the number can't be known + Integer expectedCount = this.expectedKnown() ? this.expected : null; + map.put(EXPECTED_COUNT_KEY, expectedCount); + + List> servers = + new ArrayList<>(this.entries.size()); + for (GraphStatusEntry entry : this.entries) { + servers.add(entry.asMap()); + } + map.put(SERVERS_KEY, servers); + return map; + } +} diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/GraphStatusEntry.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/GraphStatusEntry.java new file mode 100644 index 0000000000..a1d090a04e --- /dev/null +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/GraphStatusEntry.java @@ -0,0 +1,113 @@ +/* + * 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.meta; + +import java.util.LinkedHashMap; +import java.util.Map; + +import org.apache.commons.lang3.StringUtils; +import org.apache.hugegraph.type.define.GraphStatus; +import org.apache.hugegraph.util.JsonUtil; +import org.apache.hugegraph.util.Log; +import org.apache.tinkerpop.shaded.jackson.annotation.JsonIgnoreProperties; +import org.apache.tinkerpop.shaded.jackson.annotation.JsonProperty; +import org.slf4j.Logger; + +/** + * The status of one graph on one server, stored as a json value under the + * per-server graph status key. + *

+ * Servers of different versions read each other's status, so a value written + * with fields this version doesn't know must stay readable. An unknown status + * value is a different matter: it can't be mapped to a state and the whole + * entry is dropped, which leaves the server that wrote it uncounted and can + * only hold the aggregate below READY. + */ +@JsonIgnoreProperties(ignoreUnknown = true) +public class GraphStatusEntry { + + public static final String SERVER_KEY = "server"; + public static final String STATUS_KEY = "status"; + public static final String MESSAGE_KEY = "message"; + public static final String UPDATE_TIME_KEY = "update_time"; + + private static final Logger LOG = Log.logger(GraphStatusEntry.class); + + @JsonProperty(SERVER_KEY) + private String server; + + @JsonProperty(STATUS_KEY) + private GraphStatus status; + + @JsonProperty(MESSAGE_KEY) + private String message; + + @JsonProperty(UPDATE_TIME_KEY) + private long updateTime; + + public GraphStatusEntry() { + // Pass + } + + public GraphStatusEntry(String server, GraphStatus status, String message, + long updateTime) { + this.server = server; + this.status = status; + this.message = message; + this.updateTime = updateTime; + } + + public static GraphStatusEntry fromValue(String value) { + if (StringUtils.isBlank(value)) { + return null; + } + try { + return JsonUtil.fromJson(value, GraphStatusEntry.class); + } catch (RuntimeException e) { + LOG.debug("Malformed graph status payload, ignoring: {}", value, e); + return null; + } + } + + public String server() { + return this.server; + } + + public GraphStatus status() { + return this.status; + } + + public String message() { + return this.message; + } + + public long updateTime() { + return this.updateTime; + } + + public Map asMap() { + Map map = new LinkedHashMap<>(); + map.put(SERVER_KEY, this.server); + map.put(STATUS_KEY, this.status == null ? null : this.status.name()); + if (this.message != null) { + map.put(MESSAGE_KEY, this.message); + } + map.put(UPDATE_TIME_KEY, this.updateTime); + return map; + } +} diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/MetaManager.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/MetaManager.java index 6637baf22c..f6b800cb40 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/MetaManager.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/MetaManager.java @@ -78,6 +78,7 @@ public class MetaManager { public static final String META_PATH_SERVICE = "SERVICE"; public static final String META_PATH_SERVICE_CONF = "SERVICE_CONF"; public static final String META_PATH_GRAPH_CONF = "GRAPH_CONF"; + public static final String META_PATH_GRAPH_STATUS = "GRAPH_STATUS"; public static final String META_PATH_CONF = "CONF"; public static final String META_PATH_GRAPH = "GRAPH"; public static final String META_PATH_SCHEMA = "SCHEMA"; @@ -450,6 +451,29 @@ public void removeSysGraphConfig() { this.graphMetaManager.removeSysGraphConfig(); } + public void updateGraphStatus(String graphSpace, String graph, + GraphStatusEntry entry) { + this.graphMetaManager.updateGraphStatus(graphSpace, graph, entry); + } + + public Map getGraphStatus(String graphSpace, + String graph) { + return this.graphMetaManager.getGraphStatus(graphSpace, graph); + } + + public void removeGraphStatus(String graphSpace, String graph, + String server) { + this.graphMetaManager.removeGraphStatus(graphSpace, graph, server); + } + + public void clearGraphStatus(String graphSpace, String graph) { + this.graphMetaManager.clearGraphStatus(graphSpace, graph); + } + + public void clearGraphSpaceStatus(String graphSpace) { + this.graphMetaManager.clearGraphSpaceStatus(graphSpace); + } + public GraphSpace graphSpace(String name) { return this.spaceMetaManager.graphSpace(name); } diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManager.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManager.java index 52b2c3946e..8c1be72629 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManager.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManager.java @@ -26,6 +26,7 @@ import static org.apache.hugegraph.meta.MetaManager.META_PATH_GRAPH; import static org.apache.hugegraph.meta.MetaManager.META_PATH_GRAPHSPACE; import static org.apache.hugegraph.meta.MetaManager.META_PATH_GRAPH_CONF; +import static org.apache.hugegraph.meta.MetaManager.META_PATH_GRAPH_STATUS; import static org.apache.hugegraph.meta.MetaManager.META_PATH_HUGEGRAPH; import static org.apache.hugegraph.meta.MetaManager.META_PATH_JOIN; import static org.apache.hugegraph.meta.MetaManager.META_PATH_REMOVE; @@ -39,6 +40,7 @@ import java.util.function.Consumer; import org.apache.commons.lang3.StringUtils; +import org.apache.hugegraph.meta.GraphStatusEntry; import org.apache.hugegraph.meta.MetaDriver; import org.apache.hugegraph.type.define.CollectionType; import org.apache.hugegraph.util.JsonUtil; @@ -165,6 +167,46 @@ public void removeSysGraphConfig() { this.metaDriver.delete(this.sysGraphConfKey()); } + public void updateGraphStatus(String graphSpace, String graph, + GraphStatusEntry entry) { + this.metaDriver.put(this.graphStatusKey(graphSpace, graph, + entry.server()), + JsonUtil.toJson(entry)); + } + + public Map getGraphStatus(String graphSpace, + String graph) { + Map status = + CollectionFactory.newMap(CollectionType.EC); + Map keyValues = this.metaDriver.scanWithPrefix( + this.graphStatusPrefix(graphSpace, graph)); + for (Map.Entry entry : keyValues.entrySet()) { + GraphStatusEntry value = GraphStatusEntry.fromValue( + entry.getValue()); + if (value == null) { + continue; + } + String[] parts = entry.getKey().split(META_PATH_DELIMITER); + status.put(parts[parts.length - 1], value); + } + return status; + } + + public void removeGraphStatus(String graphSpace, String graph, + String server) { + this.metaDriver.delete(this.graphStatusKey(graphSpace, graph, server)); + } + + public void clearGraphStatus(String graphSpace, String graph) { + this.metaDriver.deleteWithPrefix(this.graphStatusPrefix(graphSpace, + graph)); + } + + public void clearGraphSpaceStatus(String graphSpace) { + this.metaDriver.deleteWithPrefix( + this.graphSpaceStatusPrefix(graphSpace)); + } + public void listenGraphAdd(Consumer consumer) { this.listen(this.graphAddKey(), consumer); } @@ -222,6 +264,34 @@ private String sysGraphConfKey() { META_PATH_SYS_GRAPH_CONF); } + private String graphSpaceStatusPrefix(String graphSpace) { + // HUGEGRAPH/{cluster}/GRAPHSPACE/{graphspace}/GRAPH_STATUS/ + return String.join(META_PATH_DELIMITER, + META_PATH_HUGEGRAPH, + this.cluster, + META_PATH_GRAPHSPACE, + graphSpace, + META_PATH_GRAPH_STATUS, + Strings.EMPTY); + } + + private String graphStatusPrefix(String graphSpace, String graph) { + return this.graphStatusKey(graphSpace, graph, Strings.EMPTY); + } + + private String graphStatusKey(String graphSpace, String graph, + String server) { + // HUGEGRAPH/{cluster}/GRAPHSPACE/{graphspace}/GRAPH_STATUS/{graph}/{server} + return String.join(META_PATH_DELIMITER, + META_PATH_HUGEGRAPH, + this.cluster, + META_PATH_GRAPHSPACE, + graphSpace, + META_PATH_GRAPH_STATUS, + graph, + server); + } + private String graphAddKey() { // HUGEGRAPH/{cluster}/EVENT/GRAPH/ADD return String.join(META_PATH_DELIMITER, diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/type/define/GraphStatus.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/type/define/GraphStatus.java new file mode 100644 index 0000000000..f28ae0be57 --- /dev/null +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/type/define/GraphStatus.java @@ -0,0 +1,45 @@ +/* + * 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.type.define; + +/** + * The status of one graph on one server. It describes that single server + * only, the status of the graph across the cluster is the aggregate of the + * status reported by every server. + */ +public enum GraphStatus { + + /* + * The server started to open the graph and has not finished yet, it + * can't answer requests against the graph. + */ + LOADING, + + /* + * The server opened the graph and bound it to its gremlin server, so it + * can answer requests against the graph. It says nothing about the other + * servers, and nothing about the schema of the graph being initialized. + */ + READY, + + /* + * The server failed to open the graph, the cause is carried by the + * status message. + */ + FAILED +} diff --git a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Events.java b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Events.java index 9cff8e4415..0a565d9812 100644 --- a/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Events.java +++ b/hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Events.java @@ -30,4 +30,6 @@ public final class Events { public static final String GRAPH_CREATE = "graph.create"; public static final String GRAPH_DROP = "graph.drop"; + public static final String GRAPH_BOUND = "graph.bound"; + public static final String GRAPH_BIND_FAILED = "graph.bind_failed"; } diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/GraphStatusAggregateTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/GraphStatusAggregateTest.java new file mode 100644 index 0000000000..ce6388490a --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/GraphStatusAggregateTest.java @@ -0,0 +1,416 @@ +/* + * 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.meta; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.type.define.GraphStatus; +import org.junit.Test; + +public class GraphStatusAggregateTest { + + private static final String STATUS_KEY = "status"; + private static final String READY_COUNT_KEY = "ready_count"; + private static final String TOTAL_COUNT_KEY = "total_count"; + private static final String EXPECTED_COUNT_KEY = "expected_count"; + private static final String SERVERS_KEY = "servers"; + private static final String SERVER_KEY = "server"; + + private static final String STATUS_UNKNOWN = "UNKNOWN"; + + // The entries are stamped at 1L, so they are stale well before NOW + private static final long STALE_AFTER = 1000L; + private static final long NOW = 1_000_000L; + + @Test + public void testEmptyEntriesRenderUnknownStatus() { + GraphStatusAggregate aggregate = + GraphStatusAggregate.of(Collections.emptyList(), 0); + + Assert.assertNull(aggregate.status()); + Assert.assertEquals(0, aggregate.readyCount()); + Assert.assertEquals(0, aggregate.totalCount()); + Assert.assertTrue(aggregate.entries().isEmpty()); + + Map map = aggregate.asMap(); + Assert.assertEquals(STATUS_UNKNOWN, map.get(STATUS_KEY)); + Assert.assertEquals(0, map.get(READY_COUNT_KEY)); + Assert.assertEquals(0, map.get(TOTAL_COUNT_KEY)); + Assert.assertTrue(servers(map).isEmpty()); + } + + @Test + public void testUnknownExpectedIsNotReady() { + // The number of servers that should report is unknown, so a server + // that did not report yet can't be told from one that doesn't exist + List entries = + Arrays.asList(entry("server-1", GraphStatus.READY), + entry("server-2", GraphStatus.READY)); + + for (int expected : new int[]{-5, 0}) { + GraphStatusAggregate aggregate = + GraphStatusAggregate.of(entries, expected); + + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + Assert.assertFalse(aggregate.expectedKnown()); + Assert.assertEquals(2, aggregate.readyCount()); + Assert.assertEquals(2, aggregate.totalCount()); + + Map map = aggregate.asMap(); + Assert.assertEquals(GraphStatus.LOADING.name(), + map.get(STATUS_KEY)); + // The unknown number is rendered rather than dropped, a client + // that only reads the status must not have to notice a gap + Assert.assertTrue(map.containsKey(EXPECTED_COUNT_KEY)); + Assert.assertNull(map.get(EXPECTED_COUNT_KEY)); + } + } + + @Test + public void testNullEntriesRenderUnknownStatus() { + GraphStatusAggregate aggregate = GraphStatusAggregate.of(null, 3); + + Assert.assertNull(aggregate.status()); + Assert.assertEquals(0, aggregate.totalCount()); + Assert.assertEquals(STATUS_UNKNOWN, + aggregate.asMap().get(STATUS_KEY)); + } + + @Test + public void testNullEntryIsIgnored() { + List entries = new ArrayList<>(); + entries.add(entry("server-1", GraphStatus.READY)); + entries.add(null); + + GraphStatusAggregate aggregate = GraphStatusAggregate.of(entries, 1); + + Assert.assertEquals(GraphStatus.READY, aggregate.status()); + Assert.assertEquals(1, aggregate.readyCount()); + Assert.assertEquals(1, aggregate.totalCount()); + } + + @Test + public void testSingleReadyWithUnknownExpectedIsLoading() { + // A readiness gate has to fail closed: a single server that reported + // ready is not a ready cluster while the size of the cluster is + // unknown, and the caller only reads the status + List entries = + Collections.singletonList(entry("server-1", + GraphStatus.READY)); + + GraphStatusAggregate aggregate = GraphStatusAggregate.of(entries, 0); + + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + Assert.assertEquals(1, aggregate.readyCount()); + Assert.assertEquals(1, aggregate.totalCount()); + Assert.assertEquals(0, aggregate.expected()); + Assert.assertFalse(aggregate.expectedKnown()); + Assert.assertEquals(GraphStatus.LOADING.name(), + aggregate.asMap().get(STATUS_KEY)); + } + + @Test + public void testAllReadyBelowExpectedIsLoading() { + // The servers that reported are all ready, but one of the three + // servers of the cluster has not reported yet, so the graph is not + // ready cluster wide + List entries = + Arrays.asList(entry("server-1", GraphStatus.READY), + entry("server-2", GraphStatus.READY)); + + GraphStatusAggregate aggregate = GraphStatusAggregate.of(entries, 3); + + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + Assert.assertEquals(2, aggregate.readyCount()); + Assert.assertEquals(2, aggregate.totalCount()); + Assert.assertEquals(3, aggregate.expected()); + + Map map = aggregate.asMap(); + Assert.assertEquals(GraphStatus.LOADING.name(), map.get(STATUS_KEY)); + Assert.assertEquals(2, map.get(READY_COUNT_KEY)); + Assert.assertEquals(2, map.get(TOTAL_COUNT_KEY)); + Assert.assertEquals(3, map.get(EXPECTED_COUNT_KEY)); + } + + @Test + public void testReadyCountEqualToExpectedIsReady() { + List entries = + Arrays.asList(entry("server-1", GraphStatus.READY), + entry("server-2", GraphStatus.READY)); + + GraphStatusAggregate aggregate = GraphStatusAggregate.of(entries, 2); + + Assert.assertEquals(GraphStatus.READY, aggregate.status()); + Assert.assertEquals(2, aggregate.readyCount()); + Assert.assertEquals(2, aggregate.expected()); + } + + @Test + public void testReadyCountAboveExpectedIsReady() { + // A server that was scaled down may still be counted as expected + List entries = + Arrays.asList(entry("server-1", GraphStatus.READY), + entry("server-2", GraphStatus.READY), + entry("server-3", GraphStatus.READY)); + + GraphStatusAggregate aggregate = GraphStatusAggregate.of(entries, 2); + + Assert.assertEquals(GraphStatus.READY, aggregate.status()); + Assert.assertEquals(3, aggregate.readyCount()); + Assert.assertEquals(3, aggregate.totalCount()); + } + + @Test + public void testFailedOverridesLoadingAndReady() { + List entries = + Arrays.asList(entry("server-1", GraphStatus.READY), + entry("server-2", GraphStatus.LOADING), + entry("server-3", GraphStatus.FAILED)); + + for (int expected : new int[]{-1, 0, 1, 3, 9}) { + GraphStatusAggregate aggregate = + GraphStatusAggregate.of(entries, expected); + + Assert.assertEquals(GraphStatus.FAILED, aggregate.status()); + Assert.assertEquals(1, aggregate.readyCount()); + Assert.assertEquals(3, aggregate.totalCount()); + Assert.assertEquals(GraphStatus.FAILED.name(), + aggregate.asMap().get(STATUS_KEY)); + } + } + + @Test + public void testLoadingOverridesReady() { + List entries = + Arrays.asList(entry("server-1", GraphStatus.READY), + entry("server-2", GraphStatus.LOADING)); + + for (int expected : new int[]{-1, 0, 1, 2}) { + GraphStatusAggregate aggregate = + GraphStatusAggregate.of(entries, expected); + + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + Assert.assertEquals(1, aggregate.readyCount()); + Assert.assertEquals(2, aggregate.totalCount()); + } + } + + @Test + public void testEntryWithNullStatusIsNotReady() { + List entries = + Arrays.asList(entry("server-1", GraphStatus.READY), + entry("server-2", null)); + + GraphStatusAggregate aggregate = GraphStatusAggregate.of(entries, 2); + + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + Assert.assertEquals(1, aggregate.readyCount()); + Assert.assertEquals(2, aggregate.totalCount()); + + // An unset status stays unset in the rendered per server map + Assert.assertNull(servers(aggregate.asMap()).get(1).get(STATUS_KEY)); + } + + @Test + public void testExpectedCountAlwaysRendered() { + List entries = + Collections.singletonList(entry("server-1", + GraphStatus.READY)); + + Map known = + GraphStatusAggregate.of(entries, 2).asMap(); + Assert.assertTrue(known.containsKey(EXPECTED_COUNT_KEY)); + Assert.assertEquals(2, known.get(EXPECTED_COUNT_KEY)); + + // An unknown number is rendered as null rather than left out, so a + // client can tell it apart from a number it failed to read + Map zero = GraphStatusAggregate.of(entries, 0).asMap(); + Assert.assertTrue(zero.containsKey(EXPECTED_COUNT_KEY)); + Assert.assertNull(zero.get(EXPECTED_COUNT_KEY)); + + Map negative = + GraphStatusAggregate.of(entries, -5).asMap(); + Assert.assertTrue(negative.containsKey(EXPECTED_COUNT_KEY)); + Assert.assertNull(negative.get(EXPECTED_COUNT_KEY)); + } + + @Test + public void testServersOrderedByServerId() { + List entries = + Arrays.asList(entry("server-c", GraphStatus.READY), + entry(null, GraphStatus.READY), + entry("server-a", GraphStatus.READY), + entry("server-b", GraphStatus.READY)); + + GraphStatusAggregate aggregate = GraphStatusAggregate.of(entries, 0); + + List sorted = aggregate.entries(); + Assert.assertEquals(4, sorted.size()); + Assert.assertEquals("server-a", sorted.get(0).server()); + Assert.assertEquals("server-b", sorted.get(1).server()); + Assert.assertEquals("server-c", sorted.get(2).server()); + Assert.assertNull(sorted.get(3).server()); + + List> servers = servers(aggregate.asMap()); + Assert.assertEquals("server-a", servers.get(0).get(SERVER_KEY)); + Assert.assertEquals("server-b", servers.get(1).get(SERVER_KEY)); + Assert.assertEquals("server-c", servers.get(2).get(SERVER_KEY)); + Assert.assertNull(servers.get(3).get(SERVER_KEY)); + } + + @Test + public void testStatusOfAServerThatIsGoneIsDropped() { + // The server that reported failed is no longer part of the cluster, + // its status must not hold the graph down + List entries = Arrays.asList( + entry("server-a", GraphStatus.READY), + entry("server-b", GraphStatus.FAILED)); + GraphStatusAggregate aggregate = GraphStatusAggregate.of( + entries, Collections.singletonList("server-a"), + STALE_AFTER, NOW); + + Assert.assertEquals(GraphStatus.READY, aggregate.status()); + Assert.assertEquals(1, aggregate.readyCount()); + Assert.assertEquals(1, aggregate.totalCount()); + Assert.assertEquals(1, aggregate.expected()); + } + + @Test + public void testReadyOfAServerThatIsGoneDoesNotCount() { + // Two servers of an earlier deployment left a ready status behind, + // they must not make up the quorum the running servers didn't reach + List entries = Arrays.asList( + entry("old-a", GraphStatus.READY), + entry("old-b", GraphStatus.READY), + entry("server-a", GraphStatus.READY)); + GraphStatusAggregate aggregate = GraphStatusAggregate.of( + entries, Arrays.asList("server-a", "server-b"), + STALE_AFTER, NOW); + + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + Assert.assertEquals(1, aggregate.readyCount()); + Assert.assertEquals(1, aggregate.totalCount()); + Assert.assertEquals(2, aggregate.expected()); + } + + @Test + public void testAllServersGoneLeavesTheStatusUnknown() { + List entries = Collections.singletonList( + entry("old-a", GraphStatus.FAILED)); + GraphStatusAggregate aggregate = GraphStatusAggregate.of( + entries, Collections.singletonList("server-a"), + STALE_AFTER, NOW); + + Assert.assertNull(aggregate.status()); + Assert.assertEquals(0, aggregate.totalCount()); + Assert.assertEquals("UNKNOWN", aggregate.asMap().get(STATUS_KEY)); + } + + @Test + public void testUnknownServersNeverReachReady() { + // The servers of the graph space can't be listed, so the status that + // was reported can't be told apart from the status of servers that + // are gone and nothing is dropped + List entries = Collections.singletonList( + entry("server-a", GraphStatus.READY)); + + for (Collection unknown : Arrays.asList( + (Collection) null, Collections.emptyList())) { + GraphStatusAggregate aggregate = GraphStatusAggregate.of( + entries, unknown, STALE_AFTER, NOW); + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + Assert.assertEquals(1, aggregate.totalCount()); + Assert.assertNull(aggregate.asMap().get(EXPECTED_COUNT_KEY)); + } + } + + @Test + public void testRecentStatusOfAnUnregisteredServerIsKept() { + /* + * A registration is refreshed periodically and lapses for a while + * when a server is slow. Dropping the status of such a server would + * leave only the servers that are ready and answer ready while the + * missing one is still loading + */ + List entries = Arrays.asList( + new GraphStatusEntry("server-a", GraphStatus.READY, null, NOW), + new GraphStatusEntry("server-b", GraphStatus.LOADING, null, + NOW)); + GraphStatusAggregate aggregate = GraphStatusAggregate.of( + entries, Collections.singletonList("server-a"), + STALE_AFTER, NOW); + + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + // The status of the server that is loading was kept, so it still + // holds the graph back + Assert.assertEquals(2, aggregate.totalCount()); + Assert.assertEquals(1, aggregate.expected()); + } + + @Test + public void testAServerThatIsGoneDoesNotStandInForOneThatIsRegistered() { + /* + * A server was replaced and its status is still recent enough to be + * kept, while the server that took its place has not reported yet. + * Counting the reports would make the two cancel out and answer + * ready, so which servers reported has to be looked at + */ + List entries = Arrays.asList( + new GraphStatusEntry("old-a", GraphStatus.READY, null, NOW), + new GraphStatusEntry("server-b", GraphStatus.READY, null, + NOW)); + GraphStatusAggregate aggregate = GraphStatusAggregate.of( + entries, Arrays.asList("server-b", "server-c"), + STALE_AFTER, NOW); + + Assert.assertEquals(GraphStatus.LOADING, aggregate.status()); + Assert.assertEquals(2, aggregate.expected()); + } + + @Test + public void testReadyNeedsEveryRegisteredServer() { + List entries = Arrays.asList( + new GraphStatusEntry("server-a", GraphStatus.READY, null, NOW), + new GraphStatusEntry("server-b", GraphStatus.READY, null, + NOW)); + GraphStatusAggregate aggregate = GraphStatusAggregate.of( + entries, Arrays.asList("server-a", "server-b"), + STALE_AFTER, NOW); + + Assert.assertEquals(GraphStatus.READY, aggregate.status()); + Assert.assertEquals(2, aggregate.readyCount()); + Assert.assertEquals(2, aggregate.expected()); + } + + @SuppressWarnings("unchecked") + private static List> servers(Map map) { + Object servers = map.get(SERVERS_KEY); + Assert.assertInstanceOf(List.class, servers); + return (List>) servers; + } + + private static GraphStatusEntry entry(String server, GraphStatus status) { + return new GraphStatusEntry(server, status, null, 1L); + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/GraphStatusEntryTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/GraphStatusEntryTest.java new file mode 100644 index 0000000000..894cc64969 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/GraphStatusEntryTest.java @@ -0,0 +1,140 @@ +/* + * 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.meta; + +import java.util.Map; + +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.type.define.GraphStatus; +import org.apache.hugegraph.util.JsonUtil; +import org.junit.Test; + +public class GraphStatusEntryTest { + + private static final String SERVER_KEY = "server"; + private static final String STATUS_KEY = "status"; + private static final String MESSAGE_KEY = "message"; + private static final String UPDATE_TIME_KEY = "update_time"; + + @Test + public void testJsonRoundTrip() { + GraphStatusEntry entry = new GraphStatusEntry("server-1", + GraphStatus.FAILED, + "open failed", 1234L); + + GraphStatusEntry parsed = + GraphStatusEntry.fromValue(JsonUtil.toJson(entry)); + + assertEntry(parsed, "server-1", GraphStatus.FAILED, "open failed", + 1234L); + } + + @Test + public void testJsonRoundTripWithoutMessage() { + GraphStatusEntry entry = new GraphStatusEntry("server-1", + GraphStatus.READY, + null, 1234L); + + GraphStatusEntry parsed = + GraphStatusEntry.fromValue(JsonUtil.toJson(entry)); + + assertEntry(parsed, "server-1", GraphStatus.READY, null, 1234L); + } + + @Test + public void testFromValueReturnsNullForEmptyPayload() { + Assert.assertNull(GraphStatusEntry.fromValue(null)); + Assert.assertNull(GraphStatusEntry.fromValue("")); + Assert.assertNull(GraphStatusEntry.fromValue(" ")); + } + + @Test + public void testFromValueReturnsNullForMalformedJson() { + // A malformed payload written by another server must not break the + // reader, it is dropped instead + Assert.assertNull(GraphStatusEntry.fromValue("{not-json")); + Assert.assertNull(GraphStatusEntry.fromValue("not json at all")); + Assert.assertNull(GraphStatusEntry.fromValue("[]")); + } + + @Test + public void testFromValueIgnoresUnknownField() { + // A value written by a newer server carries fields this server does + // not know, the known fields must still be read + String value = "{\"server\":\"server-1\",\"status\":\"READY\"," + + "\"update_time\":7,\"unknown_field\":\"x\"}"; + + GraphStatusEntry parsed = GraphStatusEntry.fromValue(value); + + assertEntry(parsed, "server-1", GraphStatus.READY, null, 7L); + } + + @Test + public void testFromValueReturnsNullForUnknownStatus() { + String value = "{\"server\":\"server-1\",\"status\":\"BOGUS\"}"; + + Assert.assertNull(GraphStatusEntry.fromValue(value)); + } + + @Test + public void testAsMapOmitsNullMessage() { + GraphStatusEntry entry = new GraphStatusEntry("server-1", + GraphStatus.READY, + null, 42L); + + Map map = entry.asMap(); + + Assert.assertFalse(map.containsKey(MESSAGE_KEY)); + Assert.assertEquals("server-1", map.get(SERVER_KEY)); + Assert.assertEquals(GraphStatus.READY.name(), map.get(STATUS_KEY)); + Assert.assertEquals(42L, map.get(UPDATE_TIME_KEY)); + } + + @Test + public void testAsMapIncludesMessage() { + GraphStatusEntry entry = new GraphStatusEntry("server-1", + GraphStatus.FAILED, + "open failed", 42L); + + Map map = entry.asMap(); + + Assert.assertEquals("open failed", map.get(MESSAGE_KEY)); + Assert.assertEquals(GraphStatus.FAILED.name(), map.get(STATUS_KEY)); + } + + @Test + public void testAsMapOfEmptyEntry() { + Map map = new GraphStatusEntry().asMap(); + + Assert.assertTrue(map.containsKey(STATUS_KEY)); + Assert.assertNull(map.get(SERVER_KEY)); + Assert.assertNull(map.get(STATUS_KEY)); + Assert.assertFalse(map.containsKey(MESSAGE_KEY)); + Assert.assertEquals(0L, map.get(UPDATE_TIME_KEY)); + } + + private static void assertEntry(GraphStatusEntry entry, String server, + GraphStatus status, String message, + long updateTime) { + Assert.assertNotNull(entry); + Assert.assertEquals(server, entry.server()); + Assert.assertEquals(status, entry.status()); + Assert.assertEquals(message, entry.message()); + Assert.assertEquals(updateTime, entry.updateTime()); + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManagerStatusTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManagerStatusTest.java new file mode 100644 index 0000000000..d8d6a79c1d --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManagerStatusTest.java @@ -0,0 +1,155 @@ +/* + * 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.meta.managers; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.apache.hugegraph.meta.GraphStatusEntry; +import org.apache.hugegraph.meta.MetaDriver; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.type.define.GraphStatus; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +public class GraphMetaManagerStatusTest { + + private static final String CLUSTER = "cluster"; + private static final String GRAPH_SPACE = "gs1"; + private static final String GRAPH = "graph1"; + private static final String SERVER = "server-1"; + + private static final String STATUS_PREFIX = + "HUGEGRAPH/cluster/GRAPHSPACE/gs1/GRAPH_STATUS/graph1/"; + private static final String SERVER_STATUS_KEY = + "HUGEGRAPH/cluster/GRAPHSPACE/gs1/GRAPH_STATUS/graph1/server-1"; + private static final String OTHER_SERVER_STATUS_KEY = + "HUGEGRAPH/cluster/GRAPHSPACE/gs1/GRAPH_STATUS/graph1/server-2"; + + @Test + public void testUpdateGraphStatusPutsPerServerKey() { + MetaDriver driver = Mockito.mock(MetaDriver.class); + GraphMetaManager manager = new GraphMetaManager(driver, CLUSTER); + GraphStatusEntry entry = new GraphStatusEntry(SERVER, + GraphStatus.LOADING, + "opening", 1024L); + + manager.updateGraphStatus(GRAPH_SPACE, GRAPH, entry); + + ArgumentCaptor key = ArgumentCaptor.forClass(String.class); + ArgumentCaptor value = ArgumentCaptor.forClass(String.class); + Mockito.verify(driver).put(key.capture(), value.capture()); + + Assert.assertEquals(SERVER_STATUS_KEY, key.getValue()); + GraphStatusEntry stored = GraphStatusEntry.fromValue(value.getValue()); + Assert.assertNotNull(stored); + Assert.assertEquals(SERVER, stored.server()); + Assert.assertEquals(GraphStatus.LOADING, stored.status()); + Assert.assertEquals("opening", stored.message()); + Assert.assertEquals(1024L, stored.updateTime()); + } + + @Test + public void testGetGraphStatusKeyedByServerId() { + Map keyValues = new LinkedHashMap<>(); + keyValues.put(SERVER_STATUS_KEY, statusJson("server-1", "READY", 11)); + keyValues.put(OTHER_SERVER_STATUS_KEY, + statusJson("server-2", "LOADING", 12)); + MetaDriver driver = driverReturning(keyValues); + GraphMetaManager manager = new GraphMetaManager(driver, CLUSTER); + + Map status = + manager.getGraphStatus(GRAPH_SPACE, GRAPH); + + Mockito.verify(driver).scanWithPrefix(STATUS_PREFIX); + Assert.assertEquals(2, status.size()); + Assert.assertEquals(GraphStatus.READY, status.get("server-1").status()); + Assert.assertEquals(11L, status.get("server-1").updateTime()); + Assert.assertEquals(GraphStatus.LOADING, + status.get("server-2").status()); + } + + @Test + public void testGetGraphStatusSkipsUnparseableValue() { + Map keyValues = new LinkedHashMap<>(); + keyValues.put(SERVER_STATUS_KEY, statusJson("server-1", "READY", 11)); + keyValues.put(OTHER_SERVER_STATUS_KEY, "{not-json"); + MetaDriver driver = driverReturning(keyValues); + GraphMetaManager manager = new GraphMetaManager(driver, CLUSTER); + + Map status = + manager.getGraphStatus(GRAPH_SPACE, GRAPH); + + // The unreadable value is dropped, the readable one is kept + Assert.assertEquals(1, status.size()); + Assert.assertEquals(GraphStatus.READY, status.get("server-1").status()); + Assert.assertFalse(status.containsKey("server-2")); + } + + @Test + public void testGetGraphStatusReturnsEmptyWhenNothingReported() { + MetaDriver driver = driverReturning(Collections.emptyMap()); + GraphMetaManager manager = new GraphMetaManager(driver, CLUSTER); + + Map status = + manager.getGraphStatus(GRAPH_SPACE, GRAPH); + + Mockito.verify(driver).scanWithPrefix(STATUS_PREFIX); + Assert.assertNotNull(status); + Assert.assertTrue(status.isEmpty()); + } + + @Test + public void testRemoveGraphStatusDeletesPerServerKey() { + MetaDriver driver = Mockito.mock(MetaDriver.class); + GraphMetaManager manager = new GraphMetaManager(driver, CLUSTER); + + manager.removeGraphStatus(GRAPH_SPACE, GRAPH, SERVER); + + Mockito.verify(driver).delete(SERVER_STATUS_KEY); + Mockito.verify(driver, Mockito.never()) + .deleteWithPrefix(Mockito.anyString()); + } + + @Test + public void testClearGraphStatusDeletesWholeGraphPrefix() { + MetaDriver driver = Mockito.mock(MetaDriver.class); + GraphMetaManager manager = new GraphMetaManager(driver, CLUSTER); + + manager.clearGraphStatus(GRAPH_SPACE, GRAPH); + + Mockito.verify(driver).deleteWithPrefix(STATUS_PREFIX); + Mockito.verify(driver, Mockito.never()).delete(Mockito.anyString()); + } + + private static MetaDriver driverReturning(Map keyValues) { + MetaDriver driver = Mockito.mock(MetaDriver.class); + Mockito.when(driver.scanWithPrefix(Mockito.anyString())) + .thenReturn(keyValues); + return driver; + } + + private static String statusJson(String server, String status, + long updateTime) { + return String.format("{\"server\":\"%s\",\"status\":\"%s\"," + + "\"update_time\":%s}", + server, status, updateTime); + } +} 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 1d3dd58a83..4fa84a1920 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 @@ -23,13 +23,17 @@ import org.apache.hugegraph.auth.WsAndHttpBasicAuthHandlerTest; import org.apache.hugegraph.core.RoleElectionStateMachineTest; import org.apache.hugegraph.meta.EtcdMetaDriverTest; +import org.apache.hugegraph.meta.GraphStatusAggregateTest; +import org.apache.hugegraph.meta.GraphStatusEntryTest; import org.apache.hugegraph.meta.MetaManagerSchemaCacheClearEventTest; import org.apache.hugegraph.meta.managers.AuthMetaManagerTest; +import org.apache.hugegraph.meta.managers.GraphMetaManagerStatusTest; import org.apache.hugegraph.traversal.optimize.TraversalUtilOptimizeTest; import org.apache.hugegraph.unit.api.auth.LoginAPITest; import org.apache.hugegraph.unit.api.filter.LoadDetectFilterTest; import org.apache.hugegraph.unit.api.filter.PathFilterTest; import org.apache.hugegraph.unit.api.gremlin.GremlinQueryAPITest; +import org.apache.hugegraph.unit.api.profile.GraphStatusAPITest; import org.apache.hugegraph.unit.api.space.GraphSpaceAPITest; import org.apache.hugegraph.unit.auth.HugeGraphAuthProxyTest; import org.apache.hugegraph.unit.cache.CacheManagerTest; @@ -46,6 +50,7 @@ import org.apache.hugegraph.unit.core.DirectionsTest; import org.apache.hugegraph.unit.core.ExceptionTest; import org.apache.hugegraph.unit.core.GraphManagerConfigTest; +import org.apache.hugegraph.unit.core.GraphManagerStatusTest; import org.apache.hugegraph.unit.core.LocksTableTest; import org.apache.hugegraph.unit.core.PageStateTest; import org.apache.hugegraph.unit.core.QueryTest; @@ -109,6 +114,9 @@ /* api space */ GraphSpaceAPITest.class, + /* api profile */ + GraphStatusAPITest.class, + /* cache */ CacheTest.RamCacheTest.class, CacheTest.OffheapCacheTest.class, @@ -116,6 +124,9 @@ CachedSchemaTransactionTest.class, MetaManagerSchemaCacheClearEventTest.class, EtcdMetaDriverTest.class, + GraphMetaManagerStatusTest.class, + GraphStatusEntryTest.class, + GraphStatusAggregateTest.class, CachedGraphTransactionTest.class, CacheManagerTest.class, RamTableTest.class, @@ -144,6 +155,7 @@ RolePermissionTest.class, ExceptionTest.class, GraphManagerConfigTest.class, + GraphManagerStatusTest.class, BackendStoreInfoTest.class, TraversalUtilTest.class, TraversalUtilOptimizeTest.class, diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/profile/GraphStatusAPITest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/profile/GraphStatusAPITest.java new file mode 100644 index 0000000000..fb5c1664ab --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/profile/GraphStatusAPITest.java @@ -0,0 +1,447 @@ +/* + * 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.unit.api.profile; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.api.profile.GraphsAPI; +import org.apache.hugegraph.core.GraphManager; +import org.apache.hugegraph.exception.NotFoundException; +import org.apache.hugegraph.meta.GraphStatusEntry; +import org.apache.hugegraph.meta.MetaManager; +import org.apache.hugegraph.pd.client.DiscoveryClientImpl; +import org.apache.hugegraph.pd.grpc.discovery.NodeInfo; +import org.apache.hugegraph.pd.grpc.discovery.NodeInfos; +import org.apache.hugegraph.pd.grpc.discovery.Query; +import org.apache.hugegraph.space.GraphSpace; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.type.define.GraphStatus; +import org.apache.hugegraph.unit.BaseUnitTest; +import org.apache.tinkerpop.gremlin.structure.Graph; +import org.junit.Test; +import org.mockito.Mockito; + +import jakarta.ws.rs.ForbiddenException; +import jakarta.ws.rs.core.SecurityContext; +import sun.misc.Unsafe; + +public class GraphStatusAPITest extends BaseUnitTest { + + private static final String GRAPHSPACE = "DEFAULT"; + private static final String GRAPH = "hugegraph"; + private static final String SERVER_1 = "127.0.0.1:8080"; + private static final String SERVER_2 = "127.0.0.1:8081"; + private static final String SERVER_URL = "http://" + SERVER_1; + + private static final String GRAPHSPACE_KEY = "graphspace"; + private static final String GRAPH_KEY = "graph"; + private static final String STATUS_KEY = "status"; + private static final String READY_COUNT_KEY = "ready_count"; + private static final String TOTAL_COUNT_KEY = "total_count"; + private static final String EXPECTED_COUNT_KEY = "expected_count"; + private static final String SERVERS_KEY = "servers"; + private static final String SERVER_KEY = "server"; + private static final String MESSAGE_KEY = "message"; + + private static final String SERVER_ID_LABEL = "SERVER_ID"; + private static final String GRAPHSPACE_LABEL = "GRAPHSPACE"; + + @Test + public void testStatusIsReadyWhenEveryReplicaReported() { + GraphManager manager = pdManager( + reported(entry(SERVER_1, GraphStatus.READY, null), + entry(SERVER_2, GraphStatus.READY, null)), + 2, true); + + Map result = status(manager); + + Assert.assertEquals(GRAPHSPACE, result.get(GRAPHSPACE_KEY)); + Assert.assertEquals(GRAPH, result.get(GRAPH_KEY)); + Assert.assertEquals(GraphStatus.READY.name(), result.get(STATUS_KEY)); + Assert.assertEquals(2, result.get(READY_COUNT_KEY)); + Assert.assertEquals(2, result.get(TOTAL_COUNT_KEY)); + Assert.assertEquals(2, result.get(EXPECTED_COUNT_KEY)); + + List> servers = servers(result); + Assert.assertEquals(2, servers.size()); + Assert.assertEquals(SERVER_1, servers.get(0).get(SERVER_KEY)); + Assert.assertEquals(SERVER_2, servers.get(1).get(SERVER_KEY)); + } + + @Test + public void testStatusIsLoadingWhenAReplicaHasNotReported() { + GraphManager manager = pdManager( + reported(entry(SERVER_1, GraphStatus.READY, null)), 3, true); + + Map result = status(manager); + + Assert.assertEquals(GraphStatus.LOADING.name(), result.get(STATUS_KEY)); + Assert.assertEquals(1, result.get(READY_COUNT_KEY)); + Assert.assertEquals(1, result.get(TOTAL_COUNT_KEY)); + Assert.assertEquals(3, result.get(EXPECTED_COUNT_KEY)); + Assert.assertEquals(1, servers(result).size()); + } + + @Test + public void testStatusIsFailedWhenAReplicaFailed() { + GraphManager manager = pdManager( + reported(entry(SERVER_1, GraphStatus.READY, null), + entry(SERVER_2, GraphStatus.FAILED, "backend down")), + 2, true); + + Map result = status(manager); + + Assert.assertEquals(GraphStatus.FAILED.name(), result.get(STATUS_KEY)); + Assert.assertEquals(1, result.get(READY_COUNT_KEY)); + Assert.assertEquals(2, result.get(TOTAL_COUNT_KEY)); + + Map failed = servers(result).get(1); + Assert.assertEquals(GraphStatus.FAILED.name(), failed.get(STATUS_KEY)); + Assert.assertEquals("backend down", failed.get(MESSAGE_KEY)); + } + + @Test + public void testStatusIsUnknownBeforeAnyReplicaReports() { + // The graph config is already written, no server reported status yet + GraphManager manager = pdManager(Collections.emptyMap(), 2, true); + + Map result = status(manager); + + Assert.assertEquals("UNKNOWN", result.get(STATUS_KEY)); + Assert.assertEquals(0, result.get(READY_COUNT_KEY)); + Assert.assertEquals(0, result.get(TOTAL_COUNT_KEY)); + Assert.assertEquals(2, result.get(EXPECTED_COUNT_KEY)); + Assert.assertTrue(servers(result).isEmpty()); + } + + @Test + public void testStatusCountsServersRatherThanAddresses() { + // One server registers one node per url it listens on, the servers + // that should report are counted, not the urls they answer on + NodeInfos infos = NodeInfos.newBuilder() + .addInfo(node(SERVER_1, "server-a", + GRAPHSPACE)) + .addInfo(node("10.0.0.1:8080", "server-a", + GRAPHSPACE)) + .addInfo(node(SERVER_2, "server-b", + GRAPHSPACE)) + .build(); + GraphManager manager = pdManager( + reported(entry("server-a", GraphStatus.READY, null), + entry("server-b", GraphStatus.READY, null)), + discoveryClient(infos), true); + + Map result = status(manager); + + Assert.assertEquals(2, result.get(EXPECTED_COUNT_KEY)); + Assert.assertEquals(GraphStatus.READY.name(), result.get(STATUS_KEY)); + } + + @Test + public void testStatusSkipsServersOfAnotherGraphSpace() { + // A server registered to another graph space never serves this graph + NodeInfos infos = NodeInfos.newBuilder() + .addInfo(node(SERVER_1, "server-a", + GRAPHSPACE)) + .addInfo(node(SERVER_2, "server-b", + "other")) + .build(); + GraphManager manager = pdManager( + reported(entry("server-a", GraphStatus.READY, null)), + discoveryClient(infos), true); + + Map result = status(manager); + + Assert.assertEquals(1, result.get(EXPECTED_COUNT_KEY)); + Assert.assertEquals(GraphStatus.READY.name(), result.get(STATUS_KEY)); + } + + @Test + public void testStatusIsLoadingWhenTheExpectedCountIsUnknown() { + // Asking pd failed, so it can't be told whether the servers that + // reported ready are all the servers of the cluster + DiscoveryClientImpl pdClient = Mockito.mock(DiscoveryClientImpl.class); + Mockito.when(pdClient.getNodeInfos(Mockito.any(Query.class))) + .thenThrow(new IllegalStateException("pd is unreachable")); + GraphManager manager = pdManager( + reported(entry(SERVER_1, GraphStatus.READY, null)), + pdClient, true); + + Map result = status(manager); + + Assert.assertEquals(GraphStatus.LOADING.name(), result.get(STATUS_KEY)); + Assert.assertEquals(1, result.get(READY_COUNT_KEY)); + Assert.assertTrue(result.containsKey(EXPECTED_COUNT_KEY)); + Assert.assertNull(result.get(EXPECTED_COUNT_KEY)); + } + + @Test + public void testStatusNotFoundForUnknownGraph() { + GraphManager manager = pdManager(Collections.emptyMap(), 2, false); + + Assert.assertThrows(NotFoundException.class, () -> { + new GraphsAPI().status(manager, GRAPHSPACE, GRAPH, + securityContext(true)); + }, e -> { + Assert.assertContains("Graph 'hugegraph' does not exist", + e.getMessage()); + }); + } + + @Test + public void testStatusNotFoundForDroppedGraphWithLeftoverStatus() { + // The graph is gone but the removal of one status entry didn't make + // it, the answer is still that the graph does not exist + GraphManager manager = pdManager( + reported(entry(SERVER_1, GraphStatus.READY, null)), 1, false); + + Assert.assertThrows(NotFoundException.class, () -> { + new GraphsAPI().status(manager, GRAPHSPACE, GRAPH, + securityContext(true)); + }, e -> { + Assert.assertContains("Graph 'hugegraph' does not exist", + e.getMessage()); + }); + } + + @Test + public void testStatusNotFoundForUnknownGraphSpace() { + GraphManager manager = pdManager( + reported(entry(SERVER_1, GraphStatus.READY, null)), 1, true); + Whitebox.setInternalState(manager, "graphSpaces", + new ConcurrentHashMap()); + + Assert.assertThrows(NotFoundException.class, () -> { + new GraphsAPI().status(manager, GRAPHSPACE, GRAPH, + securityContext(true)); + }, e -> { + Assert.assertContains("Graph space", e.getMessage()); + }); + } + + @Test + public void testStatusIsReadyForLocalGraphWithoutPd() { + GraphManager manager = standaloneManager(true); + + Map result = status(manager); + + Assert.assertEquals(GRAPHSPACE, result.get(GRAPHSPACE_KEY)); + Assert.assertEquals(GRAPH, result.get(GRAPH_KEY)); + Assert.assertEquals(GraphStatus.READY.name(), result.get(STATUS_KEY)); + Assert.assertEquals(1, result.get(READY_COUNT_KEY)); + Assert.assertEquals(1, result.get(TOTAL_COUNT_KEY)); + Assert.assertEquals(1, result.get(EXPECTED_COUNT_KEY)); + + List> servers = servers(result); + Assert.assertEquals(1, servers.size()); + Assert.assertEquals(SERVER_1, servers.get(0).get(SERVER_KEY)); + Assert.assertEquals(GraphStatus.READY.name(), + servers.get(0).get(STATUS_KEY)); + } + + @Test + public void testStatusNotFoundForUnknownGraphWithoutPd() { + GraphManager manager = standaloneManager(false); + + Assert.assertThrows(NotFoundException.class, () -> { + new GraphsAPI().status(manager, GRAPHSPACE, GRAPH, + securityContext(true)); + }, e -> { + Assert.assertContains("Graph 'hugegraph' does not exist", + e.getMessage()); + }); + } + + @SuppressWarnings("unchecked") + private static Map status(GraphManager manager) { + return status(manager, securityContext(true)); + } + + @SuppressWarnings("unchecked") + private static Map status(GraphManager manager, + SecurityContext sc) { + Object result = new GraphsAPI().status(manager, GRAPHSPACE, GRAPH, sc); + Assert.assertInstanceOf(Map.class, result); + return (Map) result; + } + + @SuppressWarnings("unchecked") + private static List> servers(Map map) { + Object servers = map.get(SERVERS_KEY); + Assert.assertInstanceOf(List.class, servers); + return (List>) servers; + } + + @Test + public void testStatusIsForbiddenWithoutReadOnTheGraph() { + /* + * A member of the graph space may hold a permission on another graph + * of the space only. The other reads of this resource verify the + * permission while opening the graph, this one can't + */ + GraphManager manager = pdManager( + reported(entry(SERVER_1, GraphStatus.READY, null)), 1, true); + + Assert.assertThrows(ForbiddenException.class, () -> { + status(manager, securityContext(false)); + }, e -> { + Assert.assertContains("not allowed to read graph 'hugegraph'", + e.getMessage()); + }); + } + + @Test + public void testStatusIsNotFoundWhenTheMetadataCantBeRead() { + /* + * A graph config that can't be read is not a graph that was dropped, + * and the caller polls this API: answering not found would have it + * take a metadata failure for a final answer + */ + GraphManager manager = pdManager( + reported(entry(SERVER_1, GraphStatus.READY, null)), 1, true); + MetaManager metaManager = Whitebox.getInternalState(manager, + "metaManager"); + Mockito.when(metaManager.getGraphConfig(GRAPHSPACE, GRAPH)) + .thenThrow(new IllegalStateException("pd is unreachable")); + + Map result = status(manager); + + Assert.assertEquals(GRAPH, result.get(GRAPH_KEY)); + } + + private static SecurityContext securityContext(boolean allowed) { + SecurityContext sc = Mockito.mock(SecurityContext.class); + Mockito.when(sc.isUserInRole(Mockito.anyString())) + .thenReturn(allowed); + return sc; + } + + private static GraphStatusEntry entry(String server, GraphStatus status, + String message) { + return new GraphStatusEntry(server, status, message, 1L); + } + + private static Map reported( + GraphStatusEntry... entries) { + Map status = new LinkedHashMap<>(); + for (GraphStatusEntry entry : entries) { + status.put(entry.server(), entry); + } + return status; + } + + private static GraphManager pdManager(Map status, + int replicas, boolean configExists) { + return pdManager(status, discoveryClient(replicas), configExists); + } + + private static GraphManager pdManager(Map status, + DiscoveryClientImpl pdClient, + boolean configExists) { + GraphManager manager = allocateGraphManager(); + Whitebox.setInternalState(manager, "PDExist", true); + Whitebox.setInternalState(manager, "cluster", "hg"); + Whitebox.setInternalState(manager, "serviceID", "hugegraph-service"); + Whitebox.setInternalState(manager, "url", SERVER_URL); + Whitebox.setInternalState(manager, "serverId", SERVER_1); + Whitebox.setInternalState(manager, "serverIdDistinct", true); + Whitebox.setInternalState(manager, "pdClient", pdClient); + + Map spaces = new ConcurrentHashMap<>(); + spaces.put(GRAPHSPACE, new GraphSpace(GRAPHSPACE)); + Whitebox.setInternalState(manager, "graphSpaces", spaces); + + Map> configs = new LinkedHashMap<>(); + if (configExists) { + configs.put(GRAPHSPACE + "-" + GRAPH, Collections.emptyMap()); + } + MetaManager metaManager = Mockito.mock(MetaManager.class); + Mockito.when(metaManager.getGraphStatus(GRAPHSPACE, GRAPH)) + .thenReturn(status); + Mockito.when(metaManager.graphConfigs(GRAPHSPACE)).thenReturn(configs); + Mockito.when(metaManager.getGraphConfig(GRAPHSPACE, GRAPH)) + .thenReturn(configExists ? Collections.emptyMap() : null); + Whitebox.setInternalState(manager, "metaManager", metaManager); + + Whitebox.setInternalState(manager, "graphs", + new ConcurrentHashMap()); + return manager; + } + + private static GraphManager standaloneManager(boolean graphOpened) { + GraphManager manager = allocateGraphManager(); + Whitebox.setInternalState(manager, "PDExist", false); + Whitebox.setInternalState(manager, "url", SERVER_URL); + Whitebox.setInternalState(manager, "serverId", SERVER_1); + + Map graphs = new ConcurrentHashMap<>(); + if (graphOpened) { + graphs.put(GRAPHSPACE + "-" + GRAPH, + Mockito.mock(HugeGraph.class)); + } + Whitebox.setInternalState(manager, "graphs", graphs); + return manager; + } + + private static DiscoveryClientImpl discoveryClient(int replicas) { + NodeInfos.Builder infos = NodeInfos.newBuilder(); + for (int i = 0; i < replicas; i++) { + infos.addInfo(NodeInfo.newBuilder() + .setAddress("127.0.0.1:" + (8080 + i)) + .build()); + } + return discoveryClient(infos.build()); + } + + private static DiscoveryClientImpl discoveryClient(NodeInfos infos) { + DiscoveryClientImpl client = Mockito.mock(DiscoveryClientImpl.class); + Mockito.when(client.getNodeInfos(Mockito.any(Query.class))) + .thenReturn(infos); + return client; + } + + private static NodeInfo node(String address, String server, + String graphSpace) { + NodeInfo.Builder node = NodeInfo.newBuilder().setAddress(address); + if (server != null) { + node.putLabels(SERVER_ID_LABEL, server); + } + if (graphSpace != null) { + node.putLabels(GRAPHSPACE_LABEL, graphSpace); + } + return node.build(); + } + + private static GraphManager allocateGraphManager() { + try { + Field field = Unsafe.class.getDeclaredField("theUnsafe"); + field.setAccessible(true); + Unsafe unsafe = (Unsafe) field.get(null); + return (GraphManager) unsafe.allocateInstance(GraphManager.class); + } catch (Exception e) { + throw new AssertionError(e); + } + } +} diff --git a/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/GraphManagerStatusTest.java b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/GraphManagerStatusTest.java new file mode 100644 index 0000000000..cfccdb0d96 --- /dev/null +++ b/hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/GraphManagerStatusTest.java @@ -0,0 +1,333 @@ +/* + * 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.unit.core; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import org.apache.commons.configuration2.PropertiesConfiguration; +import org.apache.hugegraph.HugeGraph; +import org.apache.hugegraph.config.CoreOptions; +import org.apache.hugegraph.config.HugeConfig; +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.meta.GraphStatusEntry; +import org.apache.hugegraph.meta.MetaManager; +import org.apache.hugegraph.space.GraphSpace; +import org.apache.hugegraph.testutil.Assert; +import org.apache.hugegraph.testutil.Whitebox; +import org.apache.hugegraph.type.define.GraphStatus; +import org.apache.hugegraph.util.Events; +import org.junit.BeforeClass; +import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; + +/** + * The status a server reports for a graph it opens: a graph that starts to + * open is loading, a graph that reached the gremlin bindings is ready, and a + * graph that failed anywhere on the way is failed. + */ +public class GraphManagerStatusTest { + + private static final String GRAPH_SPACE = "DEFAULT"; + private static final String GRAPH = "status_graph"; + private static final String SPACE_GRAPH = GRAPH_SPACE + "-" + GRAPH; + private static final String SERVER = "server-1"; + + @BeforeClass + public static void setup() { + RegisterUtil.registerBackends(); + reviveEventExecutor(); + } + + /** + * The executor of EventHub is a static field shared by the whole suite and + * an earlier test class shuts it down when it closes the factory. The pool + * is built with a caller runs policy, which silently drops the task it is + * handed once it is shut down, so the future of an event is never done and + * every wait on it blocks forever. A new EventHub keeps the dead pool, its + * init returns early while the field is set, so hand the field a live pool + * before a graph is opened. The threads are daemons, they never hold the + * jvm of the suite up + */ + private static void reviveEventExecutor() { + ExecutorService executor = + Whitebox.getInternalState(EventHub.class, "executor"); + if (executor != null && !executor.isShutdown()) { + return; + } + ExecutorService revived = Executors.newCachedThreadPool(task -> { + Thread thread = new Thread(task, "status-test-event-worker"); + thread.setDaemon(true); + return thread; + }); + Whitebox.setInternalState(EventHub.class, "executor", revived); + } + + @Test + public void testServerIdIsConfiguredIdWhenSet() { + PropertiesConfiguration properties = new PropertiesConfiguration(); + properties.setProperty(ServerOptions.SERVER_ID.name(), "server/one 1"); + GraphManager manager = newManager(properties); + + try { + // The status key is joined by '/', it can't occur in a server id + Assert.assertEquals("server_one_1", manager.serverId()); + } finally { + manager.close(); + } + } + + @Test + public void testServerIdIsStableAndTellsColocatedServersApart() { + /* + * The rest server url is a bind address rather than an identity, the + * replicas of a container image all carry the same one. The id has to + * survive a restart of a server and separate the servers of a host + */ + GraphManager first = newManager(urlProperties("http://0.0.0.0:8080")); + GraphManager restarted = + newManager(urlProperties("http://0.0.0.0:8080")); + GraphManager other = newManager(urlProperties("http://0.0.0.0:8081")); + + try { + Assert.assertEquals(first.serverId(), restarted.serverId()); + Assert.assertNotEquals(first.serverId(), other.serverId()); + Assert.assertFalse(first.serverId().contains("/")); + } finally { + first.close(); + restarted.close(); + other.close(); + } + } + + @Test + public void testCreateReportsLoadingThenReadyOnBound() { + List reported = new ArrayList<>(); + GraphManager manager = newManager(reported, null); + try { + HugeGraph graph = manager.createGraph(GRAPH_SPACE, GRAPH, "admin", + graphConfig(), false); + Assert.assertNotNull(graph); + + // The graph is open but not bound to the gremlin server yet + Assert.assertEquals(1, reported.size()); + assertEntry(reported.get(0), GraphStatus.LOADING); + + notifyBound(manager, graph); + + Assert.assertEquals(2, reported.size()); + assertEntry(reported.get(1), GraphStatus.READY); + Assert.assertNull(reported.get(1).message()); + } finally { + close(manager); + } + } + + @Test + public void testBoundIsReportedOnceWhenListenersAreRegisteredTwice() { + /* + * The listeners are registered by the constructor and again by init, + * a graph that is bound once must not be reported twice + */ + List reported = new ArrayList<>(); + GraphManager manager = newManager(reported, null); + try { + HugeGraph graph = manager.createGraph(GRAPH_SPACE, GRAPH, "admin", + graphConfig(), false); + Whitebox.invoke(manager.getClass(), "listenChanges", manager); + reported.clear(); + + notifyBound(manager, graph); + + Assert.assertEquals(1, reported.size()); + assertEntry(reported.get(0), GraphStatus.READY); + } finally { + close(manager); + } + } + + @Test + public void testFailureAfterTheBackendIsOpenReportsFailed() { + /* + * The backend of the graph opens, the meta update that comes after it + * doesn't. The entry has to end up failed rather than stay loading + * forever, the client already got an error for its request + */ + List reported = new ArrayList<>(); + GraphManager manager = newManager( + reported, new IllegalStateException("meta is unreachable")); + try { + Assert.assertThrows(IllegalStateException.class, () -> { + manager.createGraph(GRAPH_SPACE, GRAPH, "admin", + graphConfig(), false); + }); + + Assert.assertEquals(2, reported.size()); + assertEntry(reported.get(0), GraphStatus.LOADING); + assertEntry(reported.get(1), GraphStatus.FAILED); + // The type of the failure is kept, its message is bounded + Assert.assertContains("IllegalStateException", + reported.get(1).message()); + } finally { + close(manager); + } + } + + @Test + public void testFailedBindingReportsFailed() { + /* + * The graph create event is notified without waiting for it, so a + * binding that throws reaches nobody: without a report of its own the + * graph would stay loading with nothing left to end it + */ + List reported = new ArrayList<>(); + GraphManager manager = newManager(reported, null); + try { + HugeGraph graph = manager.createGraph(GRAPH_SPACE, GRAPH, "admin", + graphConfig(), false); + reported.clear(); + + EventHub hub = Whitebox.getInternalState(manager, "eventHub"); + hub.call(Events.GRAPH_BIND_FAILED, graph); + + Assert.assertEquals(1, reported.size()); + assertEntry(reported.get(0), GraphStatus.FAILED); + Assert.assertNotNull(reported.get(0).message()); + } finally { + close(manager); + } + } + + @Test + public void testAGraphThatCantBeCreatedIsNotLeftRegistered() { + /* + * A graph that failed after it was registered would be answered for + * while it reads failed, and this server would never open it again, + * so the failure would stay for good. It has to be taken back out so + * that opening it can be tried again + */ + List reported = new ArrayList<>(); + GraphManager manager = newManager( + reported, new IllegalStateException("meta is unreachable")); + try { + Assert.assertThrows(IllegalStateException.class, () -> { + manager.createGraph(GRAPH_SPACE, GRAPH, "admin", + graphConfig(), false); + }); + + assertEntry(reported.get(reported.size() - 1), + GraphStatus.FAILED); + Assert.assertNull(manager.graph(GRAPH_SPACE + "-" + GRAPH)); + } finally { + close(manager); + } + } + + private static void assertEntry(GraphStatusEntry entry, + GraphStatus status) { + Assert.assertEquals(status, entry.status()); + Assert.assertEquals(SERVER, entry.server()); + Assert.assertTrue(entry.updateTime() > 0L); + } + + private static void notifyBound(GraphManager manager, HugeGraph graph) { + /* + * call() invokes the sole registered listener in this thread, unlike + * notify() which hands the event to the executor and answers a future + * the caller has to wait on. It also asserts that the graph manager + * holds exactly one listener for the event, it throws when the event + * has no listener at all and when it has more than one + */ + EventHub hub = Whitebox.getInternalState(manager, "eventHub"); + hub.call(Events.GRAPH_BOUND, graph); + } + + private static Map graphConfig() { + Map configs = new HashMap<>(); + configs.put(CoreOptions.BACKEND.name(), "memory"); + configs.put(CoreOptions.SERIALIZER.name(), "text"); + configs.put(CoreOptions.STORE.name(), GRAPH); + return configs; + } + + private static PropertiesConfiguration urlProperties(String url) { + PropertiesConfiguration properties = new PropertiesConfiguration(); + properties.setProperty(ServerOptions.REST_SERVER_URL.name(), url); + return properties; + } + + private static GraphManager newManager(PropertiesConfiguration props) { + return new GraphManager(new HugeConfig(props), + new EventHub("status-test")); + } + + /** + * @param reported collects every status this server writes + * @param updateFailure thrown by the meta update that follows the open of + * the backend, null to let the creation succeed + */ + private static GraphManager newManager(List reported, + RuntimeException updateFailure) { + GraphManager manager = newManager(new PropertiesConfiguration()); + Whitebox.setInternalState(manager, "PDExist", true); + Whitebox.setInternalState(manager, "serverId", SERVER); + + Map spaces = new ConcurrentHashMap<>(); + spaces.put(GRAPH_SPACE, new GraphSpace(GRAPH_SPACE)); + Whitebox.setInternalState(manager, "graphSpaces", spaces); + + MetaManager metaManager = Mockito.mock(MetaManager.class); + Mockito.when(metaManager.graphSpace(GRAPH_SPACE)) + .thenReturn(new GraphSpace(GRAPH_SPACE)); + Mockito.doAnswer((InvocationOnMock invocation) -> { + reported.add(invocation.getArgument(2)); + return null; + }).when(metaManager).updateGraphStatus(Mockito.anyString(), + Mockito.anyString(), + Mockito.any()); + if (updateFailure != null) { + Mockito.doThrow(updateFailure).when(metaManager) + .updateGraphSpaceConfig(Mockito.anyString(), + Mockito.any()); + } + Whitebox.setInternalState(manager, "metaManager", metaManager); + return manager; + } + + private static void close(GraphManager manager) { + try { + HugeGraph graph = manager.graph(SPACE_GRAPH); + if (graph != null) { + graph.clearBackend(); + graph.close(); + } + } catch (Exception ignored) { + // The graph may have failed to open, nothing to close then + } + manager.close(); + } +}