hugegraph
/
hugegraph-sync
mirror of https://github.com/apache/hugegraph
0
0
Fork 2
Code Issues Packages Projects Releases Wiki Activity

Merge 9faf6d572a into de62d97f34

This commit is contained in:
KAI 2026-08-04 14:01:34 +00:00 committed by GitHub
parent de62d97f34 9faf6d572a
commit e28f69be66
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 2608 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/api/profile/GraphsAPI.java
View File

@ -25,6 +25,7 @@ import java.util.Collections;
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.HugeGraphAuthProxy;
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<String, Object> convConfig(Map<String, Object> config) {
Map<String, Object> result = new HashMap<>(config.size());
@ -247,6 +258,95 @@ public class GraphsAPI extends API {
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<String, GraphStatusEntry> status = manager.graphStatus(
graphSpace, name);
Set<String> 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<String, Object> 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")

25
hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ContextGremlinServer.java
View File

@ -130,14 +130,27 @@ public class ContextGremlinServer extends GremlinServer {
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);
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;
}
// The graph can serve requests only after all bindings are done
this.eventHub.notify(Events.GRAPH_BOUND, graph);
}
private void removeGraph(String name) {

487
hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java
View File

@ -23,6 +23,8 @@ import static org.apache.hugegraph.space.GraphSpace.DEFAULT_GRAPH_SPACE_SERVICE_
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.Set;
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.CoreOptions;
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.GlobalMasterInfo;
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.traversal.optimize.HugeScriptTraversal;
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<String> serverUrlsToPd;
private final Boolean serverDeployInK8s;
private final HugeConfig config;
@ -198,6 +209,12 @@ public final class GraphManager {
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 @@ public final class GraphManager {
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 final class GraphManager {
}
}
/*
* 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 @@ public final class GraphManager {
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.<String, String>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 final class GraphManager {
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 final class GraphManager {
}
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<String, Object> configs, boolean init,
boolean grpcThread,
AtomicBoolean registered) {
String nickname;
if (configs.get("nickname") != null) {
nickname = configs.get("nickname").toString();
@ -1348,28 +1472,286 @@ public final class GraphManager {
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<String, GraphStatusEntry> 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.
* <p>
* 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.
* <p>
* 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<String> 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<String> 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<String> servers = new HashSet<>();
for (NodeInfo nodeInfo : nodeInfos.getInfoList()) {
Map<String, String> 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<String> graphs() {
return Collections.unmodifiableSet(this.graphs.keySet());
}
@ -1565,6 +1947,16 @@ public final class GraphManager {
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 @@ public final class GraphManager {
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 final class GraphManager {
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 final class GraphManager {
} 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 @@ public final class GraphManager {
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 @@ public final class GraphManager {
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 @@ public final class GraphManager {
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;
}

263
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/GraphStatusAggregate.java Normal file
View File

@ -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.
* <p>
* 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<GraphStatusEntry> 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<GraphStatusEntry> entries;
private GraphStatusAggregate(GraphStatus status, int readyCount,
int expected,
List<GraphStatusEntry> 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.
* <p>
* 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<GraphStatusEntry> entries,
Collection<String> 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<String> live = new HashSet<>(liveServers);
List<GraphStatusEntry> reported = new ArrayList<>();
Set<String> 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<GraphStatusEntry> entries,
int expected) {
List<GraphStatusEntry> 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<GraphStatusEntry> entries() {
return this.entries;
}
public Map<String, Object> asMap() {
Map<String, Object> 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<Map<String, Object>> servers =
new ArrayList<>(this.entries.size());
for (GraphStatusEntry entry : this.entries) {
servers.add(entry.asMap());
}
map.put(SERVERS_KEY, servers);
return map;
}
}

113
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/GraphStatusEntry.java Normal file
View File

@ -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.
* <p>
* 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<String, Object> asMap() {
Map<String, Object> 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;
}
}

24
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/MetaManager.java
View File

@ -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 class MetaManager {
this.graphMetaManager.removeSysGraphConfig();
}
public void updateGraphStatus(String graphSpace, String graph,
GraphStatusEntry entry) {
this.graphMetaManager.updateGraphStatus(graphSpace, graph, entry);
}
public Map<String, GraphStatusEntry> 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);
}

70
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManager.java
View File

@ -26,6 +26,7 @@ import static org.apache.hugegraph.meta.MetaManager.META_PATH_EVENT;
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.Map;
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 class GraphMetaManager extends AbstractMetaManager {
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<String, GraphStatusEntry> getGraphStatus(String graphSpace,
String graph) {
Map<String, GraphStatusEntry> status =
CollectionFactory.newMap(CollectionType.EC);
Map<String, String> keyValues = this.metaDriver.scanWithPrefix(
this.graphStatusPrefix(graphSpace, graph));
for (Map.Entry<String, String> 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 <T> void listenGraphAdd(Consumer<T> consumer) {
this.listen(this.graphAddKey(), consumer);
}
@ -222,6 +264,34 @@ public class GraphMetaManager extends AbstractMetaManager {
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,

45
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/type/define/GraphStatus.java Normal file
View File

@ -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
}

2
hugegraph-server/hugegraph-core/src/main/java/org/apache/hugegraph/util/Events.java
View File

@ -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";
}

416
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/GraphStatusAggregateTest.java Normal file
View File

@ -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<String, Object> 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<GraphStatusEntry> 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<String, Object> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<String, Object> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> entries =
Collections.singletonList(entry("server-1",
GraphStatus.READY));
Map<String, Object> 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<String, Object> zero = GraphStatusAggregate.of(entries, 0).asMap();
Assert.assertTrue(zero.containsKey(EXPECTED_COUNT_KEY));
Assert.assertNull(zero.get(EXPECTED_COUNT_KEY));
Map<String, Object> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<Map<String, Object>> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> entries = Collections.singletonList(
entry("server-a", GraphStatus.READY));
for (Collection<String> unknown : Arrays.asList(
(Collection<String>) null, Collections.<String>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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<Map<String, Object>> servers(Map<String, Object> map) {
Object servers = map.get(SERVERS_KEY);
Assert.assertInstanceOf(List.class, servers);
return (List<Map<String, Object>>) servers;
}
private static GraphStatusEntry entry(String server, GraphStatus status) {
return new GraphStatusEntry(server, status, null, 1L);
}
}

140
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/GraphStatusEntryTest.java Normal file
View File

@ -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<String, Object> 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<String, Object> 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<String, Object> 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());
}
}

155
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/meta/managers/GraphMetaManagerStatusTest.java Normal file
View File

@ -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<String> key = ArgumentCaptor.forClass(String.class);
ArgumentCaptor<String> 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<String, String> 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<String, GraphStatusEntry> 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<String, String> 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<String, GraphStatusEntry> 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<String, GraphStatusEntry> 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<String, String> 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);
}
}

12
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/UnitTestSuite.java
View File

@ -23,13 +23,17 @@ import org.apache.hugegraph.auth.StandardAuthManagerV2Test;
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.DataTypeTest;
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 @@ import org.junit.runners.Suite;
/* api space */
GraphSpaceAPITest.class,
/* api profile */
GraphStatusAPITest.class,
/* cache */
CacheTest.RamCacheTest.class,
CacheTest.OffheapCacheTest.class,
@ -116,6 +124,9 @@ import org.junit.runners.Suite;
CachedSchemaTransactionTest.class,
MetaManagerSchemaCacheClearEventTest.class,
EtcdMetaDriverTest.class,
GraphMetaManagerStatusTest.class,
GraphStatusEntryTest.class,
GraphStatusAggregateTest.class,
CachedGraphTransactionTest.class,
CacheManagerTest.class,
RamTableTest.class,
@ -144,6 +155,7 @@ import org.junit.runners.Suite;
RolePermissionTest.class,
ExceptionTest.class,
GraphManagerConfigTest.class,
GraphManagerStatusTest.class,
BackendStoreInfoTest.class,
TraversalUtilTest.class,
TraversalUtilOptimizeTest.class,

447
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/api/profile/GraphStatusAPITest.java Normal file
View File

@ -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<String, Object> 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<Map<String, Object>> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, GraphSpace>());
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<String, Object> 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<Map<String, Object>> 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<String, Object> status(GraphManager manager) {
return status(manager, securityContext(true));
}
@SuppressWarnings("unchecked")
private static Map<String, Object> status(GraphManager manager,
SecurityContext sc) {
Object result = new GraphsAPI().status(manager, GRAPHSPACE, GRAPH, sc);
Assert.assertInstanceOf(Map.class, result);
return (Map<String, Object>) result;
}
@SuppressWarnings("unchecked")
private static List<Map<String, Object>> servers(Map<String, Object> map) {
Object servers = map.get(SERVERS_KEY);
Assert.assertInstanceOf(List.class, servers);
return (List<Map<String, Object>>) 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<String, Object> 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<String, GraphStatusEntry> reported(
GraphStatusEntry... entries) {
Map<String, GraphStatusEntry> status = new LinkedHashMap<>();
for (GraphStatusEntry entry : entries) {
status.put(entry.server(), entry);
}
return status;
}
private static GraphManager pdManager(Map<String, GraphStatusEntry> status,
int replicas, boolean configExists) {
return pdManager(status, discoveryClient(replicas), configExists);
}
private static GraphManager pdManager(Map<String, GraphStatusEntry> 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<String, GraphSpace> spaces = new ConcurrentHashMap<>();
spaces.put(GRAPHSPACE, new GraphSpace(GRAPHSPACE));
Whitebox.setInternalState(manager, "graphSpaces", spaces);
Map<String, Map<String, Object>> 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<String, Graph>());
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<String, Graph> 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);
}
}
}

333
hugegraph-server/hugegraph-test/src/main/java/org/apache/hugegraph/unit/core/GraphManagerStatusTest.java Normal file
View File

@ -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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<GraphStatusEntry> 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<String, Object> graphConfig() {
Map<String, Object> 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<GraphStatusEntry> reported,
RuntimeException updateFailure) {
GraphManager manager = newManager(new PropertiesConfiguration());
Whitebox.setInternalState(manager, "PDExist", true);
Whitebox.setInternalState(manager, "serverId", SERVER);
Map<String, GraphSpace> 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();
}
}