You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by va...@apache.org on 2023/03/08 15:20:46 UTC
[incubator-hugegraph] branch master updated: feat: support task auto manage by server role state machine (#2130)
This is an automated email from the ASF dual-hosted git repository.
vaughn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-hugegraph.git
The following commit(s) were added to refs/heads/master by this push:
new 31c6c6df1 feat: support task auto manage by server role state machine (#2130)
31c6c6df1 is described below
commit 31c6c6df134ff14c2f46e9a00e78535c5f8dc453
Author: vaughn <va...@apache.org>
AuthorDate: Wed Mar 8 23:20:36 2023 +0800
feat: support task auto manage by server role state machine (#2130)
---
.../hugegraph/api/filter/RedirectFilter.java | 156 +++++++++++++++
.../api/filter/RedirectFilterDynamicFeature.java | 20 +-
.../apache/hugegraph/api/gremlin/GremlinAPI.java | 2 +
.../org/apache/hugegraph/api/job/AlgorithmAPI.java | 10 +-
.../org/apache/hugegraph/api/job/ComputerAPI.java | 28 +--
.../org/apache/hugegraph/api/job/GremlinAPI.java | 32 +--
.../org/apache/hugegraph/api/job/RebuildAPI.java | 26 +--
.../java/org/apache/hugegraph/api/job/TaskAPI.java | 35 ++--
.../org/apache/hugegraph/api/raft/RaftAPI.java | 3 +
.../apache/hugegraph/api/schema/EdgeLabelAPI.java | 57 +++---
.../apache/hugegraph/api/schema/IndexLabelAPI.java | 43 ++--
.../hugegraph/api/schema/PropertyKeyAPI.java | 46 +++--
.../org/apache/hugegraph/api/schema/SchemaAPI.java | 18 +-
.../hugegraph/api/schema/VertexLabelAPI.java | 51 ++---
.../apache/hugegraph/auth/ConfigAuthenticator.java | 25 +--
.../apache/hugegraph/auth/HugeAuthenticator.java | 14 +-
.../hugegraph/auth/HugeFactoryAuthProxy.java | 14 +-
.../apache/hugegraph/auth/HugeGraphAuthProxy.java | 59 +++---
.../hugegraph/auth/StandardAuthenticator.java | 35 +++-
.../org/apache/hugegraph/core/GraphManager.java | 122 +++++++++---
.../apache/hugegraph/server/ApplicationConfig.java | 44 ++---
.../main/java/org/apache/hugegraph/HugeGraph.java | 36 ++--
.../org/apache/hugegraph/StandardHugeGraph.java | 96 +++++----
.../org/apache/hugegraph/config/CoreOptions.java | 1 -
.../ClusterRole.java} | 30 +--
.../ClusterRoleStore.java} | 8 +-
.../{election => masterelection}/Config.java | 6 +-
.../GlobalMasterInfo.java} | 34 +++-
.../masterelection/RoleElectionConfig.java | 76 ++++++++
.../masterelection/RoleElectionOptions.java | 95 +++++++++
.../RoleElectionStateMachine.java | 2 +-
.../masterelection/StandardClusterRoleStore.java | 216 +++++++++++++++++++++
.../StandardRoleElectionStateMachine.java} | 143 +++++++++-----
.../StandardStateMachineCallback.java | 105 ++++++++++
.../StateMachineCallback.java | 10 +-
.../StateMachineContext.java | 15 +-
.../apache/hugegraph/task/ServerInfoManager.java | 41 ++--
.../hugegraph/task/StandardTaskScheduler.java | 36 ++--
.../org/apache/hugegraph/task/TaskManager.java | 39 +++-
.../org/apache/hugegraph/dist/RegisterUtil.java | 32 ++-
.../core/RoleElectionStateMachineTest.java | 77 ++++----
41 files changed, 1448 insertions(+), 490 deletions(-)
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilter.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilter.java
new file mode 100644
index 000000000..23e5f0735
--- /dev/null
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilter.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hugegraph.api.filter;
+
+import java.io.IOException;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.masterelection.GlobalMasterInfo;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Log;
+import org.glassfish.hk2.api.IterableProvider;
+import org.glassfish.hk2.api.ServiceHandle;
+import org.glassfish.jersey.message.internal.HeaderUtils;
+import org.slf4j.Logger;
+
+import jakarta.ws.rs.NameBinding;
+import jakarta.ws.rs.client.Client;
+import jakarta.ws.rs.client.ClientBuilder;
+import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.client.Invocation;
+import jakarta.ws.rs.container.ContainerRequestContext;
+import jakarta.ws.rs.container.ContainerRequestFilter;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.MultivaluedMap;
+import jakarta.ws.rs.core.Response;
+
+public class RedirectFilter implements ContainerRequestFilter {
+
+ private static final Logger LOG = Log.logger(RedirectFilter.class);
+
+ private static final String X_HG_REDIRECT = "x-hg-redirect";
+
+ private static volatile Client client = null;
+
+ @Context
+ private IterableProvider<GraphManager> managerProvider;
+
+ private static final Set<String> MUST_BE_NULL = new HashSet<>();
+
+ static {
+ MUST_BE_NULL.add("DELETE");
+ MUST_BE_NULL.add("GET");
+ MUST_BE_NULL.add("HEAD");
+ MUST_BE_NULL.add("TRACE");
+ }
+
+ @Override
+ public void filter(ContainerRequestContext context) throws IOException {
+ ServiceHandle<GraphManager> handle = this.managerProvider.getHandle();
+ E.checkState(handle != null, "Context GraphManager is absent");
+ GraphManager manager = handle.getService();
+ E.checkState(manager != null, "Context GraphManager is absent");
+ GlobalMasterInfo globalMasterInfo = manager.globalMasterInfo();
+ if (globalMasterInfo == null || !globalMasterInfo.isFeatureSupport()) {
+ return;
+ }
+
+ String redirectTag = context.getHeaderString(X_HG_REDIRECT);
+ if (StringUtils.isNotEmpty(redirectTag)) {
+ return;
+ }
+
+ String url;
+ synchronized (globalMasterInfo) {
+ if (globalMasterInfo.isMaster() || StringUtils.isEmpty(globalMasterInfo.url())) {
+ return;
+ }
+ url = globalMasterInfo.url();
+ }
+
+ URI redirectUri;
+ try {
+ URIBuilder redirectURIBuilder = new URIBuilder(context.getUriInfo().getRequestUri());
+ URI masterURI = URI.create(url);
+ redirectURIBuilder.setHost(masterURI.getHost());
+ redirectURIBuilder.setPort(masterURI.getPort());
+ redirectURIBuilder.setScheme(masterURI.getScheme());
+
+ redirectUri = redirectURIBuilder.build();
+ } catch (URISyntaxException e) {
+ LOG.error("Redirect request exception occurred", e);
+ return;
+ }
+ this.initClientIfNeeded();
+ Response response = this.forwardRequest(context, redirectUri);
+ context.abortWith(response);
+ }
+
+ private Response forwardRequest(ContainerRequestContext requestContext, URI redirectUri) {
+ MultivaluedMap<String, String> headers = requestContext.getHeaders();
+ MultivaluedMap<String, Object> newHeaders = HeaderUtils.createOutbound();
+ if (headers != null) {
+ for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
+ for (String value : entry.getValue()) {
+ newHeaders.add(entry.getKey(), value);
+ }
+ }
+ }
+ newHeaders.add(X_HG_REDIRECT, new Date().getTime());
+ Invocation.Builder builder = client.target(redirectUri)
+ .request()
+ .headers(newHeaders);
+ Response response;
+ if (MUST_BE_NULL.contains(requestContext.getMethod())) {
+ response = builder.method(requestContext.getMethod());
+ } else {
+ response = builder.method(requestContext.getMethod(),
+ Entity.json(requestContext.getEntityStream()));
+ }
+ return response;
+ }
+
+ private void initClientIfNeeded() {
+ if (client != null) {
+ return;
+ }
+
+ synchronized (RedirectFilter.class) {
+ if (client != null) {
+ return;
+ }
+ client = ClientBuilder.newClient();
+ }
+ }
+
+ @NameBinding
+ @Retention(RetentionPolicy.RUNTIME)
+ public @interface RedirectMasterRole {
+ }
+}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeDataAdapter.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilterDynamicFeature.java
similarity index 57%
copy from hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeDataAdapter.java
copy to hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilterDynamicFeature.java
index 7151256fe..547661d5d 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeDataAdapter.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/filter/RedirectFilterDynamicFeature.java
@@ -15,13 +15,21 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.api.filter;
-import java.util.Optional;
+import jakarta.ws.rs.container.DynamicFeature;
+import jakarta.ws.rs.container.ResourceInfo;
+import jakarta.ws.rs.core.FeatureContext;
+import jakarta.ws.rs.ext.Provider;
-public interface RoleTypeDataAdapter {
+@Provider
+public class RedirectFilterDynamicFeature implements DynamicFeature {
- boolean updateIfNodePresent(RoleTypeData stateData);
-
- Optional<RoleTypeData> query();
+ @Override
+ public void configure(ResourceInfo resourceInfo, FeatureContext context) {
+ if (resourceInfo.getResourceMethod()
+ .isAnnotationPresent(RedirectFilter.RedirectMasterRole.class)) {
+ context.register(RedirectFilter.class);
+ }
+ }
}
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinAPI.java
index b9c6f018e..50df7e93d 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/gremlin/GremlinAPI.java
@@ -30,6 +30,7 @@ import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.core.UriInfo;
import org.apache.hugegraph.api.filter.CompressInterceptor.Compress;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.metrics.MetricsUtil;
import com.codahale.metrics.Histogram;
@@ -51,6 +52,7 @@ public class GremlinAPI extends GremlinQueryAPI {
@Compress
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
+ @RedirectFilter.RedirectMasterRole
public Response post(@Context HugeConfig conf,
@Context HttpHeaders headers,
String request) {
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/AlgorithmAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/AlgorithmAPI.java
index c9b8dbf9b..8e0e7d10c 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/AlgorithmAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/AlgorithmAPI.java
@@ -19,19 +19,20 @@ package org.apache.hugegraph.api.job;
import java.util.Map;
-import org.apache.hugegraph.core.GraphManager;
-import org.apache.hugegraph.server.RestServer;
-import org.slf4j.Logger;
-
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.api.filter.StatusFilter.Status;
import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.core.GraphManager;
import org.apache.hugegraph.job.AlgorithmJob;
import org.apache.hugegraph.job.JobBuilder;
+import org.apache.hugegraph.server.RestServer;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.JsonUtil;
import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
import com.codahale.metrics.annotation.Timed;
import com.google.common.collect.ImmutableMap;
@@ -56,6 +57,7 @@ public class AlgorithmAPI extends API {
@Status(Status.CREATED)
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> post(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String algorithm,
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/ComputerAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/ComputerAPI.java
index 42b79a67a..351eec303 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/ComputerAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/ComputerAPI.java
@@ -19,32 +19,33 @@ package org.apache.hugegraph.api.job;
import java.util.Map;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import jakarta.inject.Singleton;
-import jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.NotFoundException;
-import jakarta.ws.rs.POST;
-import jakarta.ws.rs.Path;
-import jakarta.ws.rs.PathParam;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.core.Context;
-
-import org.apache.hugegraph.core.GraphManager;
-import org.slf4j.Logger;
-
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.api.filter.StatusFilter.Status;
import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.core.GraphManager;
import org.apache.hugegraph.job.ComputerJob;
import org.apache.hugegraph.job.JobBuilder;
import org.apache.hugegraph.task.HugeTask;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.JsonUtil;
import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
import com.codahale.metrics.annotation.Timed;
import com.google.common.collect.ImmutableMap;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.inject.Singleton;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.NotFoundException;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.Context;
+
@Path("graphs/{graph}/jobs/computer")
@Singleton
@Tag(name = "ComputerAPI")
@@ -58,6 +59,7 @@ public class ComputerAPI extends API {
@Status(Status.CREATED)
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> post(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String computer,
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/GremlinAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/GremlinAPI.java
index 6a47ec5ad..8e45cbf11 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/GremlinAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/GremlinAPI.java
@@ -25,36 +25,37 @@ import java.nio.charset.CodingErrorAction;
import java.util.HashMap;
import java.util.Map;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import jakarta.annotation.security.RolesAllowed;
-import jakarta.inject.Singleton;
-import jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.POST;
-import jakarta.ws.rs.Path;
-import jakarta.ws.rs.PathParam;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.core.Context;
-
-import org.apache.hugegraph.core.GraphManager;
-import org.apache.hugegraph.define.Checkable;
-import org.apache.hugegraph.metrics.MetricsUtil;
-import org.slf4j.Logger;
-
import org.apache.hugegraph.HugeException;
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.api.filter.StatusFilter.Status;
import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.define.Checkable;
import org.apache.hugegraph.job.GremlinJob;
import org.apache.hugegraph.job.JobBuilder;
+import org.apache.hugegraph.metrics.MetricsUtil;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.JsonUtil;
import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
import com.codahale.metrics.Histogram;
import com.codahale.metrics.annotation.Timed;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableMap;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.annotation.security.RolesAllowed;
+import jakarta.inject.Singleton;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.Context;
+
@Path("graphs/{graph}/jobs/gremlin")
@Singleton
@Tag(name = "GremlinAPI")
@@ -73,6 +74,7 @@ public class GremlinAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=gremlin_execute"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> post(@Context GraphManager manager,
@PathParam("graph") String graph,
GremlinRequest request) {
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/RebuildAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/RebuildAPI.java
index 354a72945..f3a94e76b 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/RebuildAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/RebuildAPI.java
@@ -19,6 +19,18 @@ package org.apache.hugegraph.api.job;
import java.util.Map;
+import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
+import org.apache.hugegraph.api.filter.StatusFilter.Status;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import com.codahale.metrics.annotation.Timed;
+import com.google.common.collect.ImmutableMap;
+
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.annotation.security.RolesAllowed;
import jakarta.inject.Singleton;
@@ -28,17 +40,6 @@ import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.Context;
-import org.apache.hugegraph.api.filter.StatusFilter.Status;
-import org.apache.hugegraph.core.GraphManager;
-import org.slf4j.Logger;
-
-import org.apache.hugegraph.HugeGraph;
-import org.apache.hugegraph.api.API;
-import org.apache.hugegraph.backend.id.Id;
-import org.apache.hugegraph.util.Log;
-import com.codahale.metrics.annotation.Timed;
-import com.google.common.collect.ImmutableMap;
-
@Path("graphs/{graph}/jobs/rebuild")
@Singleton
@Tag(name = "RebuildAPI")
@@ -52,6 +53,7 @@ public class RebuildAPI extends API {
@Status(Status.ACCEPTED)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=index_write"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> vertexLabelRebuild(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name) {
@@ -68,6 +70,7 @@ public class RebuildAPI extends API {
@Status(Status.ACCEPTED)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=index_write"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> edgeLabelRebuild(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name) {
@@ -84,6 +87,7 @@ public class RebuildAPI extends API {
@Status(Status.ACCEPTED)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=index_write"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> indexLabelRebuild(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name) {
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java
index 67a445e2c..c2dedf32a 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/job/TaskAPI.java
@@ -24,6 +24,23 @@ import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
+import org.apache.groovy.util.Maps;
+import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
+import org.apache.hugegraph.api.filter.StatusFilter.Status;
+import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.backend.page.PageInfo;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.task.HugeTask;
+import org.apache.hugegraph.task.TaskScheduler;
+import org.apache.hugegraph.task.TaskStatus;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import com.codahale.metrics.annotation.Timed;
+
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.inject.Singleton;
import jakarta.ws.rs.BadRequestException;
@@ -38,22 +55,6 @@ import jakarta.ws.rs.Produces;
import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.Context;
-import org.apache.groovy.util.Maps;
-import org.apache.hugegraph.api.filter.StatusFilter.Status;
-import org.apache.hugegraph.core.GraphManager;
-import org.slf4j.Logger;
-
-import org.apache.hugegraph.api.API;
-import org.apache.hugegraph.backend.id.Id;
-import org.apache.hugegraph.backend.id.IdGenerator;
-import org.apache.hugegraph.backend.page.PageInfo;
-import org.apache.hugegraph.task.HugeTask;
-import org.apache.hugegraph.task.TaskScheduler;
-import org.apache.hugegraph.task.TaskStatus;
-import org.apache.hugegraph.util.E;
-import org.apache.hugegraph.util.Log;
-import com.codahale.metrics.annotation.Timed;
-
@Path("graphs/{graph}/tasks")
@Singleton
@Tag(name = "TaskAPI")
@@ -132,6 +133,7 @@ public class TaskAPI extends API {
@DELETE
@Timed
@Path("{id}")
+ @RedirectFilter.RedirectMasterRole
public void delete(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("id") long id) {
@@ -147,6 +149,7 @@ public class TaskAPI extends API {
@Path("{id}")
@Status(Status.ACCEPTED)
@Produces(APPLICATION_JSON_WITH_CHARSET)
+ @RedirectFilter.RedirectMasterRole
public Map<String, Object> update(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("id") long id,
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/raft/RaftAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/raft/RaftAPI.java
index 44d5b6587..dcf6cbb0d 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/raft/RaftAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/raft/RaftAPI.java
@@ -34,6 +34,7 @@ import jakarta.ws.rs.Produces;
import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.Context;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.core.GraphManager;
import org.slf4j.Logger;
@@ -151,6 +152,7 @@ public class RaftAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> addPeer(@Context GraphManager manager,
@PathParam("graph") String graph,
@QueryParam("group") @DefaultValue("default")
@@ -180,6 +182,7 @@ public class RaftAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> removePeer(@Context GraphManager manager,
@PathParam("graph") String graph,
@QueryParam("group")
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/EdgeLabelAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/EdgeLabelAPI.java
index 02ce5a6f1..ebf9a49a8 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/EdgeLabelAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/EdgeLabelAPI.java
@@ -18,43 +18,45 @@
package org.apache.hugegraph.api.schema;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import jakarta.annotation.security.RolesAllowed;
-import jakarta.inject.Singleton;
-import jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.DELETE;
-import jakarta.ws.rs.GET;
-import jakarta.ws.rs.POST;
-import jakarta.ws.rs.PUT;
-import jakarta.ws.rs.Path;
-import jakarta.ws.rs.PathParam;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.QueryParam;
-import jakarta.ws.rs.core.Context;
-
import org.apache.commons.collections.CollectionUtils;
-import org.apache.hugegraph.core.GraphManager;
-import org.apache.hugegraph.define.Checkable;
-import org.slf4j.Logger;
-
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.api.filter.StatusFilter.Status;
import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.define.Checkable;
import org.apache.hugegraph.schema.EdgeLabel;
import org.apache.hugegraph.schema.Userdata;
import org.apache.hugegraph.type.define.Frequency;
import org.apache.hugegraph.type.define.GraphMode;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
import com.codahale.metrics.annotation.Timed;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableMap;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.annotation.security.RolesAllowed;
+import jakarta.inject.Singleton;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.DELETE;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.PUT;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.Context;
+
@Path("graphs/{graph}/schema/edgelabels")
@Singleton
@Tag(name = "EdgeLabelAPI")
@@ -68,6 +70,7 @@ public class EdgeLabelAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=edge_label_write"})
+ @RedirectFilter.RedirectMasterRole
public String create(@Context GraphManager manager,
@PathParam("graph") String graph,
JsonEdgeLabel jsonEdgeLabel) {
@@ -86,6 +89,7 @@ public class EdgeLabelAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=edge_label_write"})
+ @RedirectFilter.RedirectMasterRole
public String update(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name,
@@ -156,6 +160,7 @@ public class EdgeLabelAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=edge_label_delete"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> delete(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name) {
@@ -203,8 +208,7 @@ public class EdgeLabelAPI extends API {
@Override
public void checkCreate(boolean isBatch) {
- E.checkArgumentNotNull(this.name,
- "The name of edge label can't be null");
+ E.checkArgumentNotNull(this.name, "The name of edge label can't be null");
}
private EdgeLabel.Builder convert2Builder(HugeGraph g) {
@@ -261,12 +265,13 @@ public class EdgeLabelAPI extends API {
@Override
public String toString() {
return String.format("JsonEdgeLabel{" +
- "name=%s, sourceLabel=%s, targetLabel=%s, frequency=%s, " +
- "sortKeys=%s, nullableKeys=%s, properties=%s, ttl=%s, " +
- "ttlStartTime=%s}",
- this.name, this.sourceLabel, this.targetLabel,
- this.frequency, this.sortKeys, this.nullableKeys,
- this.properties, this.ttl, this.ttlStartTime);
+ "name=%s, sourceLabel=%s, targetLabel=%s, frequency=%s, " +
+ "sortKeys=%s, nullableKeys=%s, properties=%s, ttl=%s, " +
+ "ttlStartTime=%s}",
+ this.name, this.sourceLabel, this.targetLabel,
+ this.frequency, Arrays.toString(this.sortKeys),
+ Arrays.toString(this.nullableKeys),
+ Arrays.toString(this.properties), this.ttl, this.ttlStartTime);
}
}
}
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/IndexLabelAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/IndexLabelAPI.java
index 33710bd4c..aa192fc1a 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/IndexLabelAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/IndexLabelAPI.java
@@ -18,32 +18,18 @@
package org.apache.hugegraph.api.schema;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import jakarta.annotation.security.RolesAllowed;
-import jakarta.inject.Singleton;
-import jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.DELETE;
-import jakarta.ws.rs.GET;
-import jakarta.ws.rs.POST;
-import jakarta.ws.rs.PUT;
-import jakarta.ws.rs.Path;
-import jakarta.ws.rs.PathParam;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.QueryParam;
-import jakarta.ws.rs.core.Context;
-
import org.apache.commons.collections.CollectionUtils;
-import org.apache.hugegraph.core.GraphManager;
-import org.apache.hugegraph.define.Checkable;
-import org.slf4j.Logger;
-
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.api.filter.StatusFilter.Status;
import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.define.Checkable;
import org.apache.hugegraph.schema.IndexLabel;
import org.apache.hugegraph.schema.SchemaElement;
import org.apache.hugegraph.schema.Userdata;
@@ -52,11 +38,27 @@ import org.apache.hugegraph.type.define.GraphMode;
import org.apache.hugegraph.type.define.IndexType;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
import com.codahale.metrics.annotation.Timed;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableMap;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.annotation.security.RolesAllowed;
+import jakarta.inject.Singleton;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.DELETE;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.PUT;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.Context;
+
@Path("graphs/{graph}/schema/indexlabels")
@Singleton
@Tag(name = "IndexLabelAPI")
@@ -70,6 +72,7 @@ public class IndexLabelAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=index_label_write"})
+ @RedirectFilter.RedirectMasterRole
public String create(@Context GraphManager manager,
@PathParam("graph") String graph,
JsonIndexLabel jsonIndexLabel) {
@@ -88,6 +91,7 @@ public class IndexLabelAPI extends API {
@Path("{name}")
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
+ @RedirectFilter.RedirectMasterRole
public String update(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name,
@@ -157,6 +161,7 @@ public class IndexLabelAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=index_label_delete"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> delete(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name) {
@@ -286,7 +291,7 @@ public class IndexLabelAPI extends API {
return String.format("JsonIndexLabel{name=%s, baseType=%s," +
"baseValue=%s, indexType=%s, fields=%s}",
this.name, this.baseType, this.baseValue,
- this.indexType, this.fields);
+ this.indexType, Arrays.toString(this.fields));
}
}
}
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/PropertyKeyAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/PropertyKeyAPI.java
index c7ddd6665..8bfbcc599 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/PropertyKeyAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/PropertyKeyAPI.java
@@ -18,33 +18,19 @@
package org.apache.hugegraph.api.schema;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import jakarta.annotation.security.RolesAllowed;
-import jakarta.inject.Singleton;
-import jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.DELETE;
-import jakarta.ws.rs.GET;
-import jakarta.ws.rs.POST;
-import jakarta.ws.rs.PUT;
-import jakarta.ws.rs.Path;
-import jakarta.ws.rs.PathParam;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.QueryParam;
-import jakarta.ws.rs.core.Context;
-
import org.apache.commons.collections.CollectionUtils;
-import org.apache.hugegraph.core.GraphManager;
-import org.apache.hugegraph.define.Checkable;
-import org.slf4j.Logger;
-
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.api.filter.StatusFilter.Status;
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.id.IdGenerator;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.define.Checkable;
import org.apache.hugegraph.schema.PropertyKey;
import org.apache.hugegraph.schema.SchemaElement;
import org.apache.hugegraph.schema.Userdata;
@@ -55,11 +41,27 @@ import org.apache.hugegraph.type.define.GraphMode;
import org.apache.hugegraph.type.define.WriteType;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
import com.codahale.metrics.annotation.Timed;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableMap;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.annotation.security.RolesAllowed;
+import jakarta.inject.Singleton;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.DELETE;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.PUT;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.Context;
+
@Path("graphs/{graph}/schema/propertykeys")
@Singleton
@Tag(name = "PropertyKeyAPI")
@@ -73,11 +75,11 @@ public class PropertyKeyAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=property_key_write"})
+ @RedirectFilter.RedirectMasterRole
public String create(@Context GraphManager manager,
@PathParam("graph") String graph,
JsonPropertyKey jsonPropertyKey) {
- LOG.debug("Graph [{}] create property key: {}",
- graph, jsonPropertyKey);
+ LOG.debug("Graph [{}] create property key: {}", graph, jsonPropertyKey);
checkCreatingBody(jsonPropertyKey);
HugeGraph g = graph(manager, graph);
@@ -93,6 +95,7 @@ public class PropertyKeyAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=property_key_write"})
+ @RedirectFilter.RedirectMasterRole
public String update(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name,
@@ -178,6 +181,7 @@ public class PropertyKeyAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=property_key_delete"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> delete(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name) {
@@ -266,7 +270,7 @@ public class PropertyKeyAPI extends API {
"writeType=%s, properties=%s}",
this.name, this.cardinality,
this.dataType, this.aggregateType,
- this.writeType, this.properties);
+ this.writeType, Arrays.toString(this.properties));
}
}
}
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/SchemaAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/SchemaAPI.java
index 7ebb32b5b..203ed2cbb 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/SchemaAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/SchemaAPI.java
@@ -21,6 +21,15 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.schema.SchemaManager;
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+import com.codahale.metrics.annotation.Timed;
+
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.annotation.security.RolesAllowed;
import jakarta.inject.Singleton;
@@ -30,15 +39,6 @@ import jakarta.ws.rs.PathParam;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.Context;
-import org.apache.hugegraph.core.GraphManager;
-import org.slf4j.Logger;
-
-import org.apache.hugegraph.HugeGraph;
-import org.apache.hugegraph.api.API;
-import org.apache.hugegraph.schema.SchemaManager;
-import org.apache.hugegraph.util.Log;
-import com.codahale.metrics.annotation.Timed;
-
@Path("graphs/{graph}/schema")
@Singleton
@Tag(name = "SchemaAPI")
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/VertexLabelAPI.java b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/VertexLabelAPI.java
index 8c62490ab..154290352 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/VertexLabelAPI.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/api/schema/VertexLabelAPI.java
@@ -18,43 +18,45 @@
package org.apache.hugegraph.api.schema;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import jakarta.annotation.security.RolesAllowed;
-import jakarta.inject.Singleton;
-import jakarta.ws.rs.Consumes;
-import jakarta.ws.rs.DELETE;
-import jakarta.ws.rs.GET;
-import jakarta.ws.rs.POST;
-import jakarta.ws.rs.PUT;
-import jakarta.ws.rs.Path;
-import jakarta.ws.rs.PathParam;
-import jakarta.ws.rs.Produces;
-import jakarta.ws.rs.QueryParam;
-import jakarta.ws.rs.core.Context;
-
import org.apache.commons.collections.CollectionUtils;
-import org.apache.hugegraph.core.GraphManager;
-import org.apache.hugegraph.define.Checkable;
-import org.slf4j.Logger;
-
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.api.API;
+import org.apache.hugegraph.api.filter.RedirectFilter;
import org.apache.hugegraph.api.filter.StatusFilter.Status;
import org.apache.hugegraph.backend.id.Id;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.define.Checkable;
import org.apache.hugegraph.schema.Userdata;
import org.apache.hugegraph.schema.VertexLabel;
import org.apache.hugegraph.type.define.GraphMode;
import org.apache.hugegraph.type.define.IdStrategy;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
import com.codahale.metrics.annotation.Timed;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableMap;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import jakarta.annotation.security.RolesAllowed;
+import jakarta.inject.Singleton;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.DELETE;
+import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.PUT;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.PathParam;
+import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.core.Context;
+
@Path("graphs/{graph}/schema/vertexlabels")
@Singleton
@Tag(name = "VertexLabelAPI")
@@ -68,6 +70,7 @@ public class VertexLabelAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=vertex_label_write"})
+ @RedirectFilter.RedirectMasterRole
public String create(@Context GraphManager manager,
@PathParam("graph") String graph,
JsonVertexLabel jsonVertexLabel) {
@@ -87,6 +90,7 @@ public class VertexLabelAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=vertex_label_write"})
+ @RedirectFilter.RedirectMasterRole
public String update(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name,
@@ -159,6 +163,7 @@ public class VertexLabelAPI extends API {
@Consumes(APPLICATION_JSON)
@Produces(APPLICATION_JSON_WITH_CHARSET)
@RolesAllowed({"admin", "$owner=$graph $action=vertex_label_delete"})
+ @RedirectFilter.RedirectMasterRole
public Map<String, Id> delete(@Context GraphManager manager,
@PathParam("graph") String graph,
@PathParam("name") String name) {
@@ -254,11 +259,11 @@ public class VertexLabelAPI extends API {
@Override
public String toString() {
return String.format("JsonVertexLabel{" +
- "name=%s, idStrategy=%s, primaryKeys=%s, nullableKeys=%s, " +
- "properties=%s, ttl=%s, ttlStartTime=%s}",
- this.name, this.idStrategy, this.primaryKeys,
- this.nullableKeys, this.properties, this.ttl,
- this.ttlStartTime);
+ "name=%s, idStrategy=%s, primaryKeys=%s, nullableKeys=%s, " +
+ "properties=%s, ttl=%s, ttlStartTime=%s}",
+ this.name, this.idStrategy, Arrays.toString(this.primaryKeys),
+ Arrays.toString(this.nullableKeys),
+ Arrays.toString(this.properties), this.ttl, this.ttlStartTime);
}
}
}
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ConfigAuthenticator.java b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ConfigAuthenticator.java
index 0f914cff4..95d048ede 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ConfigAuthenticator.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/ConfigAuthenticator.java
@@ -23,19 +23,17 @@ import java.util.Map;
import java.util.Objects;
import org.apache.commons.lang.NotImplementedException;
-import org.apache.tinkerpop.gremlin.groovy.jsr223.dsl.credential.CredentialGraphTokens;
-
+import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.backend.id.IdGenerator;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.ServerOptions;
import org.apache.hugegraph.util.E;
+import org.apache.tinkerpop.gremlin.groovy.jsr223.dsl.credential.CredentialGraphTokens;
public class ConfigAuthenticator implements HugeAuthenticator {
- public static final String KEY_USERNAME =
- CredentialGraphTokens.PROPERTY_USERNAME;
- public static final String KEY_PASSWORD =
- CredentialGraphTokens.PROPERTY_PASSWORD;
+ public static final String KEY_USERNAME = CredentialGraphTokens.PROPERTY_USERNAME;
+ public static final String KEY_PASSWORD = CredentialGraphTokens.PROPERTY_PASSWORD;
private final Map<String, String> tokens;
@@ -71,7 +69,7 @@ public class ConfigAuthenticator implements HugeAuthenticator {
if (username.equals(USER_ADMIN)) {
role = ROLE_ADMIN;
} else {
- // Return role with all permission, set user name as owner graph
+ // Return role with all permission, set username as owner graph
role = RolePermission.all(username);
}
} else {
@@ -83,12 +81,16 @@ public class ConfigAuthenticator implements HugeAuthenticator {
@Override
public AuthManager authManager() {
- throw new NotImplementedException(
- "AuthManager is unsupported by ConfigAuthenticator");
+ throw new NotImplementedException("AuthManager is unsupported by ConfigAuthenticator");
+ }
+
+ @Override
+ public HugeGraph graph() {
+ throw new NotImplementedException("graph() is unsupported by ConfigAuthenticator");
}
@Override
- public void initAdminUser(String password) throws Exception {
+ public void initAdminUser(String password) {
String adminToken = this.tokens.get(USER_ADMIN);
E.checkArgument(Objects.equals(adminToken, password),
"The password can't be changed for " +
@@ -97,7 +99,6 @@ public class ConfigAuthenticator implements HugeAuthenticator {
@Override
public SaslNegotiator newSaslNegotiator(InetAddress remoteAddress) {
- throw new NotImplementedException(
- "SaslNegotiator is unsupported by ConfigAuthenticator");
+ throw new NotImplementedException("SaslNegotiator is unsupported by ConfigAuthenticator");
}
}
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeAuthenticator.java b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeAuthenticator.java
index 82ac59d59..b22229e9b 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeAuthenticator.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeAuthenticator.java
@@ -21,13 +21,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.apache.tinkerpop.gremlin.groovy.jsr223.dsl.credential.CredentialGraphTokens;
-import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
-import org.apache.tinkerpop.gremlin.server.auth.AuthenticationException;
-import org.apache.tinkerpop.gremlin.server.auth.Authenticator;
-import org.apache.tinkerpop.shaded.jackson.annotation.JsonProperty;
-
import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.auth.HugeGraphAuthProxy.Context;
import org.apache.hugegraph.auth.SchemaDefine.AuthElement;
import org.apache.hugegraph.backend.id.Id;
@@ -38,6 +33,11 @@ import org.apache.hugegraph.config.ServerOptions;
import org.apache.hugegraph.type.Nameable;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.JsonUtil;
+import org.apache.tinkerpop.gremlin.groovy.jsr223.dsl.credential.CredentialGraphTokens;
+import org.apache.tinkerpop.gremlin.server.auth.AuthenticatedUser;
+import org.apache.tinkerpop.gremlin.server.auth.AuthenticationException;
+import org.apache.tinkerpop.gremlin.server.auth.Authenticator;
+import org.apache.tinkerpop.shaded.jackson.annotation.JsonProperty;
public interface HugeAuthenticator extends Authenticator {
@@ -66,6 +66,8 @@ public interface HugeAuthenticator extends Authenticator {
AuthManager authManager();
+ HugeGraph graph();
+
@Override
default void setup(final Map<String, Object> config) {
E.checkState(config != null,
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java
index e2b9ccfcc..e9291fe52 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeFactoryAuthProxy.java
@@ -26,24 +26,27 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.commons.configuration2.Configuration;
import org.apache.hugegraph.HugeException;
import org.apache.hugegraph.HugeFactory;
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.StandardHugeGraph;
import org.apache.hugegraph.backend.cache.CacheManager;
import org.apache.hugegraph.backend.tx.AbstractTransaction;
+import org.apache.hugegraph.backend.tx.GraphTransaction;
+import org.apache.hugegraph.backend.tx.IndexableTransaction;
import org.apache.hugegraph.concurrent.LockManager;
import org.apache.hugegraph.metrics.ServerReporter;
import org.apache.hugegraph.schema.SchemaElement;
import org.apache.hugegraph.schema.SchemaManager;
-import org.apache.hugegraph.backend.tx.GraphTransaction;
-import org.apache.hugegraph.backend.tx.IndexableTransaction;
+import org.apache.hugegraph.schema.builder.AbstractBuilder;
import org.apache.hugegraph.schema.builder.EdgeLabelBuilder;
import org.apache.hugegraph.schema.builder.IndexLabelBuilder;
import org.apache.hugegraph.schema.builder.PropertyKeyBuilder;
import org.apache.hugegraph.schema.builder.VertexLabelBuilder;
import org.apache.hugegraph.serializer.JsonSerializer;
import org.apache.hugegraph.structure.HugeEdge;
+import org.apache.hugegraph.structure.HugeElement;
import org.apache.hugegraph.structure.HugeProperty;
import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.task.HugeTask;
@@ -51,23 +54,20 @@ import org.apache.hugegraph.task.StandardTaskScheduler;
import org.apache.hugegraph.task.TaskCallable;
import org.apache.hugegraph.task.TaskCallable.SysTaskCallable;
import org.apache.hugegraph.task.TaskManager;
-import org.apache.hugegraph.schema.builder.AbstractBuilder;
-import org.apache.hugegraph.structure.HugeElement;
import org.apache.hugegraph.traversal.optimize.HugeCountStepStrategy;
import org.apache.hugegraph.traversal.optimize.HugeGraphStepStrategy;
import org.apache.hugegraph.traversal.optimize.HugeVertexStepStrategy;
import org.apache.hugegraph.util.Reflection;
import org.apache.hugegraph.variables.HugeVariables;
+
import com.google.common.collect.ImmutableSet;
-import org.apache.commons.configuration2.Configuration;
public final class HugeFactoryAuthProxy {
public static final String GRAPH_FACTORY =
"gremlin.graph=org.apache.hugegraph.auth.HugeFactoryAuthProxy";
- private static final Set<String> PROTECT_METHODS = ImmutableSet.of(
- "instance");
+ private static final Set<String> PROTECT_METHODS = ImmutableSet.of("instance");
private static final Map<HugeGraph, HugeGraph> GRAPHS = new HashMap<>();
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
index 51d6effc4..04cfac30d 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/HugeGraphAuthProxy.java
@@ -37,26 +37,6 @@ import java.util.function.Supplier;
import javax.security.sasl.AuthenticationException;
import org.apache.commons.configuration2.Configuration;
-import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
-import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
-import org.apache.tinkerpop.gremlin.process.traversal.Bytecode.Instruction;
-import org.apache.tinkerpop.gremlin.process.traversal.Script;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
-import org.apache.tinkerpop.gremlin.process.traversal.translator.GroovyTranslator;
-import org.apache.tinkerpop.gremlin.structure.Edge;
-import org.apache.tinkerpop.gremlin.structure.Element;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Property;
-import org.apache.tinkerpop.gremlin.structure.Transaction;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.structure.io.Io;
-import org.slf4j.Logger;
-
-import com.alipay.remoting.rpc.RpcServer;
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.auth.HugeAuthenticator.RolePerm;
import org.apache.hugegraph.auth.HugeAuthenticator.User;
@@ -75,6 +55,7 @@ import org.apache.hugegraph.config.TypedOption;
import org.apache.hugegraph.exception.NotSupportException;
import org.apache.hugegraph.iterator.FilterIterator;
import org.apache.hugegraph.iterator.MapperIterator;
+import org.apache.hugegraph.masterelection.RoleElectionStateMachine;
import org.apache.hugegraph.rpc.RpcServiceConfig4Client;
import org.apache.hugegraph.rpc.RpcServiceConfig4Server;
import org.apache.hugegraph.schema.EdgeLabel;
@@ -101,6 +82,26 @@ import org.apache.hugegraph.type.define.NodeRole;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
import org.apache.hugegraph.util.RateLimiter;
+import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
+import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
+import org.apache.tinkerpop.gremlin.process.traversal.Bytecode.Instruction;
+import org.apache.tinkerpop.gremlin.process.traversal.Script;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.translator.GroovyTranslator;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Element;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Property;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+import org.apache.tinkerpop.gremlin.structure.io.Io;
+import org.slf4j.Logger;
+
+import com.alipay.remoting.rpc.RpcServer;
import jakarta.ws.rs.ForbiddenException;
import jakarta.ws.rs.NotAuthorizedException;
@@ -704,6 +705,12 @@ public final class HugeGraphAuthProxy implements HugeGraph {
return this.authManager;
}
+ @Override
+ public RoleElectionStateMachine roleElectionStateMachine() {
+ this.verifyAdminPermission();
+ return this.hugegraph.roleElectionStateMachine();
+ }
+
@Override
public void switchAuthManager(AuthManager authManager) {
this.verifyAdminPermission();
@@ -1657,10 +1664,7 @@ public final class HugeGraphAuthProxy implements HugeGraph {
return Collections.emptyIterator();
}
- return new MapperIterator<TraversalStrategy<?>, TraversalStrategy<?>>(
- this.strategies.iterator(), (strategy) -> {
- return new TraversalStrategyProxy<>(strategy);
- });
+ return new MapperIterator<>(this.strategies.iterator(), TraversalStrategyProxy::new);
}
@Override
@@ -1668,7 +1672,7 @@ public final class HugeGraphAuthProxy implements HugeGraph {
return this.strategies.addStrategies(strategies);
}
- @SuppressWarnings({ "rawtypes", "unchecked" })
+ @SuppressWarnings({"unchecked"})
@Override
public TraversalStrategies removeStrategies(
Class<? extends TraversalStrategy>... strategyClasses) {
@@ -1766,8 +1770,7 @@ public final class HugeGraphAuthProxy implements HugeGraph {
}
@Override
- public int compareTo(@SuppressWarnings("rawtypes")
- Class<? extends TraversalStrategy> other) {
+ public int compareTo(Class<? extends TraversalStrategy> other) {
return this.origin.compareTo(other);
}
@@ -1869,7 +1872,7 @@ public final class HugeGraphAuthProxy implements HugeGraph {
public ContextThreadPoolExecutor(int corePoolSize, int maxPoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, maxPoolSize, 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(), threadFactory);
+ new LinkedBlockingQueue<>(), threadFactory);
}
@Override
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/StandardAuthenticator.java b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/StandardAuthenticator.java
index 2cda2333e..244bfd848 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/auth/StandardAuthenticator.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/auth/StandardAuthenticator.java
@@ -24,16 +24,16 @@ import java.util.Scanner;
import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang.StringUtils;
-import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
-
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.masterelection.RoleElectionOptions;
import org.apache.hugegraph.rpc.RpcClientProviderWithAuth;
import org.apache.hugegraph.util.ConfigUtil;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.StringEncoding;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
public class StandardAuthenticator implements HugeAuthenticator {
@@ -41,11 +41,6 @@ public class StandardAuthenticator implements HugeAuthenticator {
private HugeGraph graph = null;
- private HugeGraph graph() {
- E.checkState(this.graph != null, "Must setup Authenticator first");
- return this.graph;
- }
-
private void initAdminUser() throws Exception {
if (this.requireInitAdminUser()) {
this.initAdminUser(this.inputPassword());
@@ -53,9 +48,15 @@ public class StandardAuthenticator implements HugeAuthenticator {
this.graph.close();
}
+ @Override
+ public HugeGraph graph() {
+ E.checkState(this.graph != null, "Must setup Authenticator first");
+ return this.graph;
+ }
+
@Override
public void initAdminUser(String password) {
- // Not allowed to call by non main thread
+ // Not allowed to call by non-main thread
String caller = Thread.currentThread().getName();
E.checkState("main".equals(caller), "Invalid caller '%s'", caller);
@@ -88,7 +89,7 @@ public class StandardAuthenticator implements HugeAuthenticator {
// CHECKSTYLE:OFF
System.out.println(inputPrompt);
// CHECKSTYLE:ON
- @SuppressWarnings("resource") // just wrapper of System.in
+ // just wrapper of System.in
Scanner scanner = new Scanner(System.in);
password = scanner.nextLine();
}
@@ -125,6 +126,7 @@ public class StandardAuthenticator implements HugeAuthenticator {
String raftGroupPeers = config.get(ServerOptions.RAFT_GROUP_PEERS);
graphConfig.addProperty(ServerOptions.RAFT_GROUP_PEERS.name(),
raftGroupPeers);
+ this.transferRoleWorkerConfig(graphConfig, config);
this.graph = (HugeGraph) GraphFactory.open(graphConfig);
@@ -136,6 +138,21 @@ public class StandardAuthenticator implements HugeAuthenticator {
}
}
+ private void transferRoleWorkerConfig(HugeConfig graphConfig, HugeConfig config) {
+ graphConfig.addProperty(RoleElectionOptions.NODE_EXTERNAL_URL.name(),
+ config.get(ServerOptions.REST_SERVER_URL));
+ graphConfig.addProperty(RoleElectionOptions.BASE_TIMEOUT_MILLISECOND.name(),
+ config.get(RoleElectionOptions.BASE_TIMEOUT_MILLISECOND));
+ graphConfig.addProperty(RoleElectionOptions.EXCEEDS_FAIL_COUNT.name(),
+ config.get(RoleElectionOptions.EXCEEDS_FAIL_COUNT));
+ graphConfig.addProperty(RoleElectionOptions.RANDOM_TIMEOUT_MILLISECOND.name(),
+ config.get(RoleElectionOptions.RANDOM_TIMEOUT_MILLISECOND));
+ graphConfig.addProperty(RoleElectionOptions.HEARTBEAT_INTERVAL_SECOND.name(),
+ config.get(RoleElectionOptions.HEARTBEAT_INTERVAL_SECOND));
+ graphConfig.addProperty(RoleElectionOptions.MASTER_DEAD_TIMES.name(),
+ config.get(RoleElectionOptions.MASTER_DEAD_TIMES));
+ }
+
/**
* Verify if a user is legal
* @param username the username for authentication
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java b/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java
index 57a99b732..b203c1047 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/core/GraphManager.java
@@ -20,34 +20,24 @@ package org.apache.hugegraph.core;
import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.lang3.StringUtils;
-import org.apache.tinkerpop.gremlin.server.auth.AuthenticationException;
-import org.apache.tinkerpop.gremlin.server.util.MetricManager;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Transaction;
-import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
-import org.apache.hugegraph.auth.HugeFactoryAuthProxy;
-import org.apache.hugegraph.auth.HugeGraphAuthProxy;
-import org.apache.hugegraph.config.ServerOptions;
-import org.apache.hugegraph.metrics.MetricsUtil;
-import org.apache.hugegraph.metrics.ServerReporter;
-import org.apache.hugegraph.serializer.JsonSerializer;
-import org.apache.hugegraph.serializer.Serializer;
-import org.apache.hugegraph.server.RestServer;
-import org.slf4j.Logger;
-
-import com.alipay.sofa.rpc.config.ServerConfig;
import org.apache.hugegraph.HugeFactory;
import org.apache.hugegraph.HugeGraph;
import org.apache.hugegraph.auth.AuthManager;
import org.apache.hugegraph.auth.HugeAuthenticator;
+import org.apache.hugegraph.auth.HugeFactoryAuthProxy;
+import org.apache.hugegraph.auth.HugeGraphAuthProxy;
+import org.apache.hugegraph.auth.StandardAuthenticator;
import org.apache.hugegraph.backend.BackendException;
import org.apache.hugegraph.backend.cache.Cache;
import org.apache.hugegraph.backend.cache.CacheManager;
@@ -56,13 +46,23 @@ import org.apache.hugegraph.backend.id.IdGenerator;
import org.apache.hugegraph.backend.store.BackendStoreInfo;
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.EventHub;
import org.apache.hugegraph.exception.NotSupportException;
+import org.apache.hugegraph.masterelection.GlobalMasterInfo;
+import org.apache.hugegraph.masterelection.RoleElectionOptions;
+import org.apache.hugegraph.masterelection.RoleElectionStateMachine;
+import org.apache.hugegraph.masterelection.StandardStateMachineCallback;
+import org.apache.hugegraph.metrics.MetricsUtil;
+import org.apache.hugegraph.metrics.ServerReporter;
import org.apache.hugegraph.rpc.RpcClientProvider;
import org.apache.hugegraph.rpc.RpcConsumerConfig;
import org.apache.hugegraph.rpc.RpcProviderConfig;
import org.apache.hugegraph.rpc.RpcServer;
+import org.apache.hugegraph.serializer.JsonSerializer;
+import org.apache.hugegraph.serializer.Serializer;
+import org.apache.hugegraph.server.RestServer;
import org.apache.hugegraph.task.TaskManager;
import org.apache.hugegraph.testutil.Whitebox;
import org.apache.hugegraph.type.define.NodeRole;
@@ -70,6 +70,14 @@ import org.apache.hugegraph.util.ConfigUtil;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Events;
import org.apache.hugegraph.util.Log;
+import org.apache.tinkerpop.gremlin.server.auth.AuthenticationException;
+import org.apache.tinkerpop.gremlin.server.util.MetricManager;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Transaction;
+import org.apache.tinkerpop.gremlin.structure.util.GraphFactory;
+import org.slf4j.Logger;
+
+import com.alipay.sofa.rpc.config.ServerConfig;
public final class GraphManager {
@@ -82,6 +90,9 @@ public final class GraphManager {
private final RpcClientProvider rpcClient;
private final HugeConfig conf;
+ private RoleElectionStateMachine roleStateWorker;
+ private GlobalMasterInfo globalMasterInfo;
+
private Id server;
private NodeRole role;
@@ -265,6 +276,9 @@ public final class GraphManager {
}
this.destroyRpcServer();
this.unlistenChanges();
+ if (this.roleStateWorker != null) {
+ this.roleStateWorker.shutdown();
+ }
}
private void startRpcServer() {
@@ -358,12 +372,13 @@ public final class GraphManager {
String raftGroupPeers = this.conf.get(ServerOptions.RAFT_GROUP_PEERS);
config.addProperty(ServerOptions.RAFT_GROUP_PEERS.name(),
raftGroupPeers);
+ this.transferRoleWorkerConfig(config);
Graph graph = GraphFactory.open(config);
this.graphs.put(name, graph);
HugeConfig graphConfig = (HugeConfig) graph.configuration();
- assert graphConfPath.equals(graphConfig.file().getPath());
+ assert graphConfPath.equals(Objects.requireNonNull(graphConfig.file()).getPath());
LOG.info("Graph '{}' was successfully configured via '{}'",
name, graphConfPath);
@@ -375,6 +390,21 @@ public final class GraphManager {
}
}
+ private void transferRoleWorkerConfig(HugeConfig config) {
+ config.addProperty(RoleElectionOptions.NODE_EXTERNAL_URL.name(),
+ this.conf.get(ServerOptions.REST_SERVER_URL));
+ config.addProperty(RoleElectionOptions.BASE_TIMEOUT_MILLISECOND.name(),
+ this.conf.get(RoleElectionOptions.BASE_TIMEOUT_MILLISECOND));
+ config.addProperty(RoleElectionOptions.EXCEEDS_FAIL_COUNT.name(),
+ this.conf.get(RoleElectionOptions.EXCEEDS_FAIL_COUNT));
+ config.addProperty(RoleElectionOptions.RANDOM_TIMEOUT_MILLISECOND.name(),
+ this.conf.get(RoleElectionOptions.RANDOM_TIMEOUT_MILLISECOND));
+ config.addProperty(RoleElectionOptions.HEARTBEAT_INTERVAL_SECOND.name(),
+ this.conf.get(RoleElectionOptions.HEARTBEAT_INTERVAL_SECOND));
+ config.addProperty(RoleElectionOptions.MASTER_DEAD_TIMES.name(),
+ this.conf.get(RoleElectionOptions.MASTER_DEAD_TIMES));
+ }
+
private void waitGraphsReady() {
if (!this.rpcServer.enabled()) {
LOG.info("RpcServer is not enabled, skip wait graphs ready");
@@ -393,6 +423,7 @@ public final class GraphManager {
for (String graph : this.graphs()) {
// TODO: close tx from main thread
HugeGraph hugegraph = this.graph(graph);
+ assert hugegraph != null;
if (!hugegraph.backendStoreFeatures().supportsPersistence()) {
hugegraph.initBackend();
if (this.requireAuthentication()) {
@@ -428,11 +459,57 @@ public final class GraphManager {
"The server role can't be null or empty");
this.server = IdGenerator.of(server);
this.role = NodeRole.valueOf(role.toUpperCase());
+
+ boolean supportRoleStateWorker = this.supportRoleStateWorker();
+ if (supportRoleStateWorker) {
+ this.role = NodeRole.WORKER;
+ }
+
for (String graph : this.graphs()) {
HugeGraph hugegraph = this.graph(graph);
assert hugegraph != null;
hugegraph.serverStarted(this.server, this.role);
}
+
+ if (supportRoleStateWorker) {
+ this.initRoleStateWorker();
+ }
+ }
+
+ private void initRoleStateWorker() {
+ E.checkArgument(this.roleStateWorker == null, "Repetition init");
+ Executor applyThread = Executors.newSingleThreadExecutor();
+ this.roleStateWorker = this.authenticator().graph().roleElectionStateMachine();
+ this.globalMasterInfo = new GlobalMasterInfo();
+ StandardStateMachineCallback stateMachineCallback = new StandardStateMachineCallback(
+ TaskManager.instance(),
+ this.globalMasterInfo);
+ applyThread.execute(() -> {
+ this.roleStateWorker.apply(stateMachineCallback);
+ });
+ }
+
+ public GlobalMasterInfo globalMasterInfo() {
+ return this.globalMasterInfo;
+ }
+
+ private boolean supportRoleStateWorker() {
+ if (this.role.computer()) {
+ return false;
+ }
+
+ try {
+ if (!(this.authenticator() instanceof StandardAuthenticator)) {
+ LOG.info("{} authenticator does not support role election currently",
+ this.authenticator().getClass().getSimpleName());
+ return false;
+ }
+ } catch (IllegalStateException e) {
+ LOG.info("Unconfigured StandardAuthenticator, not support role election currently");
+ return false;
+ }
+
+ return true;
}
private void addMetrics(HugeConfig config) {
@@ -551,6 +628,7 @@ public final class GraphManager {
Object incomingValue = config.get(option);
for (String graphName : this.graphs.keySet()) {
HugeGraph graph = this.graph(graphName);
+ assert graph != null;
Object existedValue = graph.option(option);
E.checkArgument(!incomingValue.equals(existedValue),
"The value '%s' of option '%s' conflicts with " +
@@ -575,11 +653,11 @@ public final class GraphManager {
continue;
}
- MetricsUtil.registerGauge(Cache.class, hits, () -> cache.hits());
- MetricsUtil.registerGauge(Cache.class, miss, () -> cache.miss());
- MetricsUtil.registerGauge(Cache.class, exp, () -> cache.expire());
- MetricsUtil.registerGauge(Cache.class, size, () -> cache.size());
- MetricsUtil.registerGauge(Cache.class, cap, () -> cache.capacity());
+ MetricsUtil.registerGauge(Cache.class, hits, cache::hits);
+ MetricsUtil.registerGauge(Cache.class, miss, cache::miss);
+ MetricsUtil.registerGauge(Cache.class, exp, cache::expire);
+ MetricsUtil.registerGauge(Cache.class, size, cache::size);
+ MetricsUtil.registerGauge(Cache.class, cap, cache::capacity);
}
}
}
diff --git a/hugegraph-api/src/main/java/org/apache/hugegraph/server/ApplicationConfig.java b/hugegraph-api/src/main/java/org/apache/hugegraph/server/ApplicationConfig.java
index 3bbe3048c..d70e6e0b2 100644
--- a/hugegraph-api/src/main/java/org/apache/hugegraph/server/ApplicationConfig.java
+++ b/hugegraph-api/src/main/java/org/apache/hugegraph/server/ApplicationConfig.java
@@ -17,6 +17,14 @@
package org.apache.hugegraph.server;
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.api.filter.RedirectFilterDynamicFeature;
+import org.apache.hugegraph.config.HugeConfig;
+import org.apache.hugegraph.core.GraphManager;
+import org.apache.hugegraph.define.WorkLoad;
+import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.version.CoreVersion;
import org.apache.tinkerpop.gremlin.server.util.MetricManager;
import org.glassfish.hk2.api.Factory;
import org.glassfish.hk2.api.MultiException;
@@ -29,13 +37,6 @@ import org.glassfish.jersey.server.monitoring.ApplicationEventListener;
import org.glassfish.jersey.server.monitoring.RequestEvent;
import org.glassfish.jersey.server.monitoring.RequestEventListener;
-import org.apache.hugegraph.HugeException;
-import org.apache.hugegraph.config.HugeConfig;
-import org.apache.hugegraph.core.GraphManager;
-import org.apache.hugegraph.define.WorkLoad;
-import org.apache.hugegraph.event.EventHub;
-import org.apache.hugegraph.util.E;
-import org.apache.hugegraph.version.CoreVersion;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.jersey3.InstrumentedResourceMethodApplicationListener;
@@ -47,15 +48,13 @@ import jakarta.ws.rs.ApplicationPath;
@ApplicationPath("/")
@OpenAPIDefinition(
- info =
- @Info(
- title = "HugeGraph RESTful API",
- version = CoreVersion.DEFAULT_VERSION,
- description = "All management API for HugeGraph",
- contact =
- @Contact(
- url = "https://github.com/apache/hugegraph",
- name = "HugeGraph")))
+ info = @Info(
+ title = "HugeGraph RESTful API",
+ version = CoreVersion.DEFAULT_VERSION,
+ description = "All management API for HugeGraph",
+ contact = @Contact(url = "https://github.com/apache/hugegraph", name = "HugeGraph")
+ )
+)
public class ApplicationConfig extends ResourceConfig {
public ApplicationConfig(HugeConfig conf, EventHub hub) {
@@ -67,6 +66,8 @@ public class ApplicationConfig extends ResourceConfig {
// Register to use the jsr250 annotations @RolesAllowed
register(RolesAllowedDynamicFeature.class);
+ register(RedirectFilterDynamicFeature.class);
+
// Register HugeConfig to context
register(new ConfFactory(conf));
@@ -84,10 +85,9 @@ public class ApplicationConfig extends ResourceConfig {
register(OpenApiResource.class);
}
- private class ConfFactory extends AbstractBinder
- implements Factory<HugeConfig> {
+ private class ConfFactory extends AbstractBinder implements Factory<HugeConfig> {
- private HugeConfig conf = null;
+ private final HugeConfig conf;
public ConfFactory(HugeConfig conf) {
E.checkNotNull(conf, "configuration");
@@ -110,8 +110,7 @@ public class ApplicationConfig extends ResourceConfig {
}
}
- private class GraphManagerFactory extends AbstractBinder
- implements Factory<GraphManager> {
+ private class GraphManagerFactory extends AbstractBinder implements Factory<GraphManager> {
private GraphManager manager = null;
@@ -167,8 +166,7 @@ public class ApplicationConfig extends ResourceConfig {
}
}
- private class WorkLoadFactory extends AbstractBinder
- implements Factory<WorkLoad> {
+ private class WorkLoadFactory extends AbstractBinder implements Factory<WorkLoad> {
private final WorkLoad load;
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java b/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java
index e2c2000ba..4a6270864 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/HugeGraph.java
@@ -28,9 +28,23 @@ import org.apache.hugegraph.backend.query.Query;
import org.apache.hugegraph.backend.store.BackendFeatures;
import org.apache.hugegraph.backend.store.BackendStoreInfo;
import org.apache.hugegraph.backend.store.raft.RaftGroupManager;
+import org.apache.hugegraph.config.HugeConfig;
+import org.apache.hugegraph.config.TypedOption;
+import org.apache.hugegraph.masterelection.RoleElectionStateMachine;
import org.apache.hugegraph.rpc.RpcServiceConfig4Client;
import org.apache.hugegraph.rpc.RpcServiceConfig4Server;
+import org.apache.hugegraph.schema.EdgeLabel;
+import org.apache.hugegraph.schema.IndexLabel;
+import org.apache.hugegraph.schema.PropertyKey;
+import org.apache.hugegraph.schema.SchemaElement;
+import org.apache.hugegraph.schema.SchemaLabel;
+import org.apache.hugegraph.schema.SchemaManager;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.structure.HugeFeatures;
import org.apache.hugegraph.task.TaskScheduler;
+import org.apache.hugegraph.traversal.optimize.HugeCountStepStrategy;
+import org.apache.hugegraph.traversal.optimize.HugeGraphStepStrategy;
+import org.apache.hugegraph.traversal.optimize.HugeVertexStepStrategy;
import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.type.define.GraphMode;
import org.apache.hugegraph.type.define.GraphReadMode;
@@ -43,19 +57,6 @@ import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import com.alipay.remoting.rpc.RpcServer;
-import org.apache.hugegraph.config.HugeConfig;
-import org.apache.hugegraph.config.TypedOption;
-import org.apache.hugegraph.schema.EdgeLabel;
-import org.apache.hugegraph.schema.IndexLabel;
-import org.apache.hugegraph.schema.PropertyKey;
-import org.apache.hugegraph.schema.SchemaElement;
-import org.apache.hugegraph.schema.SchemaLabel;
-import org.apache.hugegraph.schema.SchemaManager;
-import org.apache.hugegraph.schema.VertexLabel;
-import org.apache.hugegraph.structure.HugeFeatures;
-import org.apache.hugegraph.traversal.optimize.HugeCountStepStrategy;
-import org.apache.hugegraph.traversal.optimize.HugeGraphStepStrategy;
-import org.apache.hugegraph.traversal.optimize.HugeVertexStepStrategy;
/**
* Graph interface for Gremlin operations
@@ -230,6 +231,8 @@ public interface HugeGraph extends Graph {
AuthManager authManager();
+ RoleElectionStateMachine roleElectionStateMachine();
+
void switchAuthManager(AuthManager authManager);
TaskScheduler taskScheduler();
@@ -311,10 +314,9 @@ public interface HugeGraph extends Graph {
}
static void registerTraversalStrategies(Class<?> clazz) {
- TraversalStrategies strategies = null;
- strategies = TraversalStrategies.GlobalCache
- .getStrategies(Graph.class)
- .clone();
+ TraversalStrategies strategies = TraversalStrategies.GlobalCache
+ .getStrategies(Graph.class)
+ .clone();
strategies.addStrategies(HugeVertexStepStrategy.instance(),
HugeGraphStepStrategy.instance(),
HugeCountStepStrategy.instance());
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java b/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
index 4542fb88c..c1d0108f7 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/StandardHugeGraph.java
@@ -25,6 +25,8 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hugegraph.analyzer.Analyzer;
+import org.apache.hugegraph.analyzer.AnalyzerFactory;
import org.apache.hugegraph.auth.AuthManager;
import org.apache.hugegraph.auth.StandardAuthManager;
import org.apache.hugegraph.backend.BackendException;
@@ -39,15 +41,47 @@ import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.id.IdGenerator;
import org.apache.hugegraph.backend.id.SnowflakeIdGenerator;
import org.apache.hugegraph.backend.query.Query;
+import org.apache.hugegraph.backend.serializer.AbstractSerializer;
+import org.apache.hugegraph.backend.serializer.SerializerFactory;
+import org.apache.hugegraph.backend.store.BackendFeatures;
+import org.apache.hugegraph.backend.store.BackendProviderFactory;
+import org.apache.hugegraph.backend.store.BackendStore;
+import org.apache.hugegraph.backend.store.BackendStoreInfo;
+import org.apache.hugegraph.backend.store.BackendStoreProvider;
import org.apache.hugegraph.backend.store.raft.RaftBackendStoreProvider;
import org.apache.hugegraph.backend.store.raft.RaftGroupManager;
import org.apache.hugegraph.backend.store.ram.RamTable;
import org.apache.hugegraph.backend.tx.GraphTransaction;
import org.apache.hugegraph.backend.tx.SchemaTransaction;
import org.apache.hugegraph.config.CoreOptions;
+import org.apache.hugegraph.config.HugeConfig;
+import org.apache.hugegraph.config.TypedOption;
+import org.apache.hugegraph.event.EventHub;
+import org.apache.hugegraph.event.EventListener;
+import org.apache.hugegraph.exception.NotAllowException;
import org.apache.hugegraph.io.HugeGraphIoRegistry;
+import org.apache.hugegraph.masterelection.ClusterRoleStore;
+import org.apache.hugegraph.masterelection.Config;
+import org.apache.hugegraph.masterelection.RoleElectionConfig;
+import org.apache.hugegraph.masterelection.RoleElectionOptions;
+import org.apache.hugegraph.masterelection.RoleElectionStateMachine;
+import org.apache.hugegraph.masterelection.StandardClusterRoleStore;
+import org.apache.hugegraph.masterelection.StandardRoleElectionStateMachine;
+import org.apache.hugegraph.perf.PerfUtil.Watched;
import org.apache.hugegraph.rpc.RpcServiceConfig4Client;
import org.apache.hugegraph.rpc.RpcServiceConfig4Server;
+import org.apache.hugegraph.schema.EdgeLabel;
+import org.apache.hugegraph.schema.IndexLabel;
+import org.apache.hugegraph.schema.PropertyKey;
+import org.apache.hugegraph.schema.SchemaElement;
+import org.apache.hugegraph.schema.SchemaLabel;
+import org.apache.hugegraph.schema.SchemaManager;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.structure.HugeEdge;
+import org.apache.hugegraph.structure.HugeEdgeProperty;
+import org.apache.hugegraph.structure.HugeFeatures;
+import org.apache.hugegraph.structure.HugeVertex;
+import org.apache.hugegraph.structure.HugeVertexProperty;
import org.apache.hugegraph.task.ServerInfoManager;
import org.apache.hugegraph.task.TaskManager;
import org.apache.hugegraph.task.TaskScheduler;
@@ -55,7 +89,13 @@ import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.type.define.GraphMode;
import org.apache.hugegraph.type.define.GraphReadMode;
import org.apache.hugegraph.type.define.NodeRole;
-import org.apache.hugegraph.util.*;
+import org.apache.hugegraph.util.ConfigUtil;
+import org.apache.hugegraph.util.DateUtil;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Events;
+import org.apache.hugegraph.util.LockUtil;
+import org.apache.hugegraph.util.Log;
+import org.apache.hugegraph.variables.HugeVariables;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.structure.Edge;
import org.apache.tinkerpop.gremlin.structure.Graph;
@@ -66,40 +106,9 @@ import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.io.Io;
import org.apache.tinkerpop.gremlin.structure.util.AbstractThreadLocalTransaction;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
-import org.apache.hugegraph.util.ConfigUtil;
-import org.apache.hugegraph.util.Events;
-import org.apache.hugegraph.util.LockUtil;
import org.slf4j.Logger;
import com.alipay.remoting.rpc.RpcServer;
-import org.apache.hugegraph.analyzer.Analyzer;
-import org.apache.hugegraph.analyzer.AnalyzerFactory;
-import org.apache.hugegraph.backend.serializer.AbstractSerializer;
-import org.apache.hugegraph.backend.serializer.SerializerFactory;
-import org.apache.hugegraph.backend.store.BackendFeatures;
-import org.apache.hugegraph.backend.store.BackendProviderFactory;
-import org.apache.hugegraph.backend.store.BackendStore;
-import org.apache.hugegraph.backend.store.BackendStoreInfo;
-import org.apache.hugegraph.backend.store.BackendStoreProvider;
-import org.apache.hugegraph.config.HugeConfig;
-import org.apache.hugegraph.config.TypedOption;
-import org.apache.hugegraph.event.EventHub;
-import org.apache.hugegraph.event.EventListener;
-import org.apache.hugegraph.exception.NotAllowException;
-import org.apache.hugegraph.perf.PerfUtil.Watched;
-import org.apache.hugegraph.schema.EdgeLabel;
-import org.apache.hugegraph.schema.IndexLabel;
-import org.apache.hugegraph.schema.PropertyKey;
-import org.apache.hugegraph.schema.SchemaElement;
-import org.apache.hugegraph.schema.SchemaLabel;
-import org.apache.hugegraph.schema.SchemaManager;
-import org.apache.hugegraph.schema.VertexLabel;
-import org.apache.hugegraph.structure.HugeEdge;
-import org.apache.hugegraph.structure.HugeEdgeProperty;
-import org.apache.hugegraph.structure.HugeFeatures;
-import org.apache.hugegraph.structure.HugeVertex;
-import org.apache.hugegraph.structure.HugeVertexProperty;
-import org.apache.hugegraph.variables.HugeVariables;
import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.RateLimiter;
@@ -156,6 +165,8 @@ public class StandardHugeGraph implements HugeGraph {
private final TaskManager taskManager;
private AuthManager authManager;
+ private RoleElectionStateMachine roleElectionStateMachine;
+
private final HugeFeatures features;
private final BackendStoreProvider storeProvider;
@@ -262,6 +273,8 @@ public class StandardHugeGraph implements HugeGraph {
serverId, serverRole, this.name);
this.serverInfoManager().initServerInfo(serverId, serverRole);
+ this.initRoleStateWorker(serverId);
+
// TODO: check necessary?
LOG.info("Check olap property-key tables for graph '{}'", this.name);
for (PropertyKey pk : this.schemaTransaction().getPropertyKeys()) {
@@ -276,6 +289,18 @@ public class StandardHugeGraph implements HugeGraph {
this.started = true;
}
+ private void initRoleStateWorker(Id serverId) {
+ Config roleStateMachineConfig = new RoleElectionConfig(serverId.toString(),
+ this.configuration.get(RoleElectionOptions.NODE_EXTERNAL_URL),
+ this.configuration.get(RoleElectionOptions.EXCEEDS_FAIL_COUNT),
+ this.configuration.get(RoleElectionOptions.RANDOM_TIMEOUT_MILLISECOND),
+ this.configuration.get(RoleElectionOptions.HEARTBEAT_INTERVAL_SECOND),
+ this.configuration.get(RoleElectionOptions.MASTER_DEAD_TIMES),
+ this.configuration.get(RoleElectionOptions.BASE_TIMEOUT_MILLISECOND));
+ ClusterRoleStore clusterRoleStore = new StandardClusterRoleStore(this.params);
+ this.roleElectionStateMachine = new StandardRoleElectionStateMachine(roleStateMachineConfig, clusterRoleStore);
+ }
+
@Override
public boolean started() {
return this.started;
@@ -371,7 +396,7 @@ public class StandardHugeGraph implements HugeGraph {
LockUtil.lock(this.name, LockUtil.GRAPH_LOCK);
try {
this.storeProvider.truncate();
- // TOOD: remove this after serverinfo saved in etcd
+ // TODO: remove this after serverinfo saved in etcd
this.serverStarted(this.serverInfoManager().selfServerId(),
this.serverInfoManager().selfServerRole());
} finally {
@@ -1038,6 +1063,11 @@ public class StandardHugeGraph implements HugeGraph {
return this.authManager;
}
+ @Override
+ public RoleElectionStateMachine roleElectionStateMachine() {
+ return this.roleElectionStateMachine;
+ }
+
@Override
public void switchAuthManager(AuthManager authManager) {
this.authManager = authManager;
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java b/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java
index 6a472b6ea..47d3b2bfa 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/config/CoreOptions.java
@@ -24,7 +24,6 @@ import static org.apache.hugegraph.config.OptionChecker.positiveInt;
import static org.apache.hugegraph.config.OptionChecker.rangeInt;
import org.apache.hugegraph.backend.query.Query;
-import org.apache.hugegraph.backend.tx.GraphTransaction;
import org.apache.hugegraph.type.define.CollectionType;
import org.apache.hugegraph.util.Bytes;
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeData.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/ClusterRole.java
similarity index 73%
rename from hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeData.java
rename to hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/ClusterRole.java
index 66f47e792..d9b95a9b9 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeData.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/ClusterRole.java
@@ -15,22 +15,24 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.masterelection;
import java.util.Objects;
-public class RoleTypeData {
+public class ClusterRole {
- private String node;
+ private final String node;
private long clock;
- private int epoch;
+ private final int epoch;
+ private final String url;
- public RoleTypeData(String node, int epoch) {
- this(node, epoch, 1);
+ public ClusterRole(String node, String url, int epoch) {
+ this(node, url, epoch, 1);
}
- public RoleTypeData(String node, int epoch, long clock) {
+ public ClusterRole(String node, String url, int epoch, long clock) {
this.node = node;
+ this.url = url;
this.epoch = epoch;
this.clock = clock;
}
@@ -59,18 +61,22 @@ public class RoleTypeData {
return this.node;
}
+ public String url() {
+ return this.url;
+ }
+
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
- if (!(obj instanceof RoleTypeData)) {
+ if (!(obj instanceof ClusterRole)) {
return false;
}
- RoleTypeData metaData = (RoleTypeData) obj;
- return clock == metaData.clock &&
- epoch == metaData.epoch &&
- Objects.equals(node, metaData.node);
+ ClusterRole clusterRole = (ClusterRole) obj;
+ return clock == clusterRole.clock &&
+ epoch == clusterRole.epoch &&
+ Objects.equals(node, clusterRole.node);
}
@Override
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeDataAdapter.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/ClusterRoleStore.java
similarity index 82%
rename from hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeDataAdapter.java
rename to hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/ClusterRoleStore.java
index 7151256fe..638dbf49f 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleTypeDataAdapter.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/ClusterRoleStore.java
@@ -15,13 +15,13 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.masterelection;
import java.util.Optional;
-public interface RoleTypeDataAdapter {
+public interface ClusterRoleStore {
- boolean updateIfNodePresent(RoleTypeData stateData);
+ boolean updateIfNodePresent(ClusterRole clusterRole);
- Optional<RoleTypeData> query();
+ Optional<ClusterRole> query();
}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/Config.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/Config.java
similarity index 91%
rename from hugegraph-core/src/main/java/org/apache/hugegraph/election/Config.java
rename to hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/Config.java
index 8afa3084b..ceb3200b7 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/Config.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/Config.java
@@ -15,19 +15,21 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.masterelection;
public interface Config {
String node();
+ String url();
+
int exceedsFailCount();
long randomTimeoutMillisecond();
long heartBeatIntervalSecond();
- int exceedsWorkerCount();
+ int masterDeadTimes();
long baseTimeoutMillisecond();
}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineContext.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/GlobalMasterInfo.java
similarity index 52%
copy from hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineContext.java
copy to hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/GlobalMasterInfo.java
index e314a352c..9d8f85a9b 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineContext.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/GlobalMasterInfo.java
@@ -15,21 +15,37 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.masterelection;
-public interface StateMachineContext {
+public class GlobalMasterInfo {
- Integer epoch();
+ private boolean isMaster;
+ private String url;
- String node();
+ private volatile boolean featureSupport;
- RoleElectionStateMachine stateMachine();
+ public GlobalMasterInfo() {
+ this.featureSupport = false;
+ }
- void epoch(Integer epoch);
+ public synchronized void set(boolean isMaster, String url) {
+ this.isMaster = isMaster;
+ this.url = url;
+ }
- Config config();
+ public synchronized boolean isMaster() {
+ return this.isMaster;
+ }
- RoleTypeDataAdapter adapter();
+ public synchronized String url() {
+ return this.url;
+ }
- void reset();
+ public void isFeatureSupport(boolean featureSupport) {
+ this.featureSupport = featureSupport;
+ }
+
+ public boolean isFeatureSupport() {
+ return this.featureSupport;
+ }
}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionConfig.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionConfig.java
new file mode 100644
index 000000000..c810b28e0
--- /dev/null
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionConfig.java
@@ -0,0 +1,76 @@
+/*
+ * 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.masterelection;
+
+public class RoleElectionConfig implements Config {
+
+ private final String node;
+ private final String url;
+ private final int exceedsFailCount;
+ private final long randomTimeoutMillisecond;
+ private final long heartBeatIntervalSecond;
+ private final int masterDeadTimes;
+ private final long baseTimeoutMillisecond;
+
+ public RoleElectionConfig(String node, String url, int exceedsFailCount,
+ long randomTimeoutMillisecond, long heartBeatIntervalSecond,
+ int masterDeadTimes, long baseTimeoutMillisecond) {
+ this.node = node;
+ this.url = url;
+ this.exceedsFailCount = exceedsFailCount;
+ this.randomTimeoutMillisecond = randomTimeoutMillisecond;
+ this.heartBeatIntervalSecond = heartBeatIntervalSecond;
+ this.masterDeadTimes = masterDeadTimes;
+ this.baseTimeoutMillisecond = baseTimeoutMillisecond;
+ }
+
+ @Override
+ public String node() {
+ return this.node;
+ }
+
+ @Override
+ public String url() {
+ return this.url;
+ }
+
+ @Override
+ public int exceedsFailCount() {
+ return this.exceedsFailCount;
+ }
+
+ @Override
+ public long randomTimeoutMillisecond() {
+ return this.randomTimeoutMillisecond;
+ }
+
+ @Override
+ public long heartBeatIntervalSecond() {
+ return this.heartBeatIntervalSecond;
+ }
+
+ @Override
+ public int masterDeadTimes() {
+ return this.masterDeadTimes;
+ }
+
+ @Override
+ public long baseTimeoutMillisecond() {
+ return this.baseTimeoutMillisecond;
+ }
+}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionOptions.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionOptions.java
new file mode 100644
index 000000000..1c8498c08
--- /dev/null
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionOptions.java
@@ -0,0 +1,95 @@
+/*
+ * 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.masterelection;
+
+import static org.apache.hugegraph.config.OptionChecker.disallowEmpty;
+import static org.apache.hugegraph.config.OptionChecker.rangeInt;
+
+import org.apache.hugegraph.config.ConfigOption;
+import org.apache.hugegraph.config.OptionHolder;
+
+public class RoleElectionOptions extends OptionHolder {
+
+ private RoleElectionOptions() {
+ super();
+ }
+
+ private static volatile RoleElectionOptions instance;
+
+ public static synchronized RoleElectionOptions instance() {
+ if (instance == null) {
+ instance = new RoleElectionOptions();
+ // Should initialize all static members first, then register.
+ instance.registerOptions();
+ }
+ return instance;
+ }
+
+ public static final ConfigOption<Integer> EXCEEDS_FAIL_COUNT =
+ new ConfigOption<>(
+ "server.role.fail_count",
+ "When the node failed count of update or query heartbeat " +
+ "is reaches this threshold, the node will become abdication state to guard" +
+ "safe property.",
+ rangeInt(0, Integer.MAX_VALUE),
+ 5
+ );
+
+ public static final ConfigOption<String> NODE_EXTERNAL_URL =
+ new ConfigOption<>(
+ "server.role.node_external_url",
+ "The url of external accessibility.",
+ disallowEmpty(),
+ "http://127.0.0.1:8080"
+ );
+
+ public static final ConfigOption<Integer> RANDOM_TIMEOUT_MILLISECOND =
+ new ConfigOption<>(
+ "server.role.random_timeout",
+ "The random timeout in ms that be used when candidate node request " +
+ "to become master state to reduce competitive voting.",
+ rangeInt(0, Integer.MAX_VALUE),
+ 1000
+ );
+
+ public static final ConfigOption<Integer> HEARTBEAT_INTERVAL_SECOND =
+ new ConfigOption<>(
+ "server.role.heartbeat_interval",
+ "The role state machine heartbeat interval second time.",
+ rangeInt(0, Integer.MAX_VALUE),
+ 2
+ );
+
+ public static final ConfigOption<Integer> MASTER_DEAD_TIMES =
+ new ConfigOption<>(
+ "server.role.master_dead_times",
+ "When the worker node detects that the number of times " +
+ "the master node fails to update heartbeat reaches this threshold, " +
+ "the worker node will become to a candidate node.",
+ rangeInt(0, Integer.MAX_VALUE),
+ 10
+ );
+
+ public static final ConfigOption<Integer> BASE_TIMEOUT_MILLISECOND =
+ new ConfigOption<>(
+ "server.role.base_timeout",
+ "The role state machine candidate state base timeout time.",
+ rangeInt(0, Integer.MAX_VALUE),
+ 500
+ );
+}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleElectionStateMachine.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionStateMachine.java
similarity index 95%
rename from hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleElectionStateMachine.java
rename to hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionStateMachine.java
index 70ea67f97..920bc104f 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleElectionStateMachine.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/RoleElectionStateMachine.java
@@ -15,7 +15,7 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.masterelection;
public interface RoleElectionStateMachine {
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardClusterRoleStore.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardClusterRoleStore.java
new file mode 100644
index 000000000..f0f69319d
--- /dev/null
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardClusterRoleStore.java
@@ -0,0 +1,216 @@
+/*
+ * 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.masterelection;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+import org.apache.hugegraph.HugeGraphParams;
+import org.apache.hugegraph.auth.SchemaDefine;
+import org.apache.hugegraph.backend.query.Condition;
+import org.apache.hugegraph.backend.query.ConditionQuery;
+import org.apache.hugegraph.backend.store.BackendEntry;
+import org.apache.hugegraph.backend.tx.GraphTransaction;
+import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.structure.HugeVertex;
+import org.apache.hugegraph.type.HugeType;
+import org.apache.hugegraph.type.define.DataType;
+import org.apache.hugegraph.type.define.HugeKeys;
+import org.apache.hugegraph.util.Log;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.T;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.slf4j.Logger;
+
+public class StandardClusterRoleStore implements ClusterRoleStore {
+
+ private static final Logger LOG = Log.logger(StandardClusterRoleStore.class);
+ private static final int RETRY_QUERY_TIMEOUT = 200;
+
+ private final HugeGraphParams graph;
+
+ private boolean firstTime;
+
+ public StandardClusterRoleStore(HugeGraphParams graph) {
+ this.graph = graph;
+ Schema schema = new Schema(graph);
+ schema.initSchemaIfNeeded();
+ this.firstTime = true;
+ }
+
+ @Override
+ public boolean updateIfNodePresent(ClusterRole clusterRole) {
+ // if epoch increase, update and return true
+ // if epoch equal, ignore different node, return false
+ Optional<Vertex> oldClusterRoleOpt = this.queryVertex();
+ if (oldClusterRoleOpt.isPresent()) {
+ ClusterRole oldClusterRole = this.from(oldClusterRoleOpt.get());
+ if (clusterRole.epoch() < oldClusterRole.epoch()) {
+ return false;
+ }
+
+ if (clusterRole.epoch() == oldClusterRole.epoch() &&
+ !Objects.equals(clusterRole.node(), oldClusterRole.node())) {
+ return false;
+ }
+ LOG.trace("Server {} epoch {} begin remove data old epoch {}, ",
+ clusterRole.node(), clusterRole.epoch(), oldClusterRole.epoch());
+ this.graph.systemTransaction().removeVertex((HugeVertex) oldClusterRoleOpt.get());
+ this.graph.systemTransaction().commitOrRollback();
+ LOG.trace("Server {} epoch {} success remove data old epoch {}, ",
+ clusterRole.node(), clusterRole.epoch(), oldClusterRole.epoch());
+ }
+ try {
+ GraphTransaction tx = this.graph.systemTransaction();
+ tx.doUpdateIfAbsent(this.constructEntry(clusterRole));
+ tx.commitOrRollback();
+ LOG.trace("Server {} epoch {} success update data",
+ clusterRole.node(), clusterRole.epoch());
+ } catch (Throwable ignore) {
+ LOG.trace("Server {} epoch {} fail update data",
+ clusterRole.node(), clusterRole.epoch());
+ return false;
+ }
+
+ return true;
+ }
+
+ private BackendEntry constructEntry(ClusterRole clusterRole) {
+ List<Object> list = new ArrayList<>(8);
+ list.add(T.label);
+ list.add(P.ROLE_DATA);
+
+ list.add(P.NODE);
+ list.add(clusterRole.node());
+
+ list.add(P.URL);
+ list.add(clusterRole.url());
+
+ list.add(P.CLOCK);
+ list.add(clusterRole.clock());
+
+ list.add(P.EPOCH);
+ list.add(clusterRole.epoch());
+
+ list.add(P.TYPE);
+ list.add("default");
+
+ HugeVertex vertex = this.graph.systemTransaction()
+ .constructVertex(false, list.toArray());
+
+ return this.graph.serializer().writeVertex(vertex);
+ }
+
+ @Override
+ public Optional<ClusterRole> query() {
+ Optional<Vertex> vertex = this.queryVertex();
+ if (!vertex.isPresent() && !this.firstTime) {
+ // If query nothing, retry once
+ try {
+ Thread.sleep(RETRY_QUERY_TIMEOUT);
+ } catch (InterruptedException ignored) {
+ }
+
+ vertex = this.queryVertex();
+ }
+ this.firstTime = false;
+ return vertex.map(this::from);
+ }
+
+ private ClusterRole from(Vertex vertex) {
+ String node = (String) vertex.property(P.NODE).value();
+ String url = (String) vertex.property(P.URL).value();
+ Long clock = (Long) vertex.property(P.CLOCK).value();
+ Integer epoch = (Integer) vertex.property(P.EPOCH).value();
+
+ return new ClusterRole(node, url, epoch, clock);
+ }
+
+ private Optional<Vertex> queryVertex() {
+ GraphTransaction tx = this.graph.systemTransaction();
+ ConditionQuery query = new ConditionQuery(HugeType.VERTEX);
+ VertexLabel vl = this.graph.graph().vertexLabel(P.ROLE_DATA);
+ query.eq(HugeKeys.LABEL, vl.id());
+ query.query(Condition.eq(vl.primaryKeys().get(0), "default"));
+ query.showHidden(true);
+ Iterator<Vertex> vertexIterator = tx.queryVertices(query);
+ if (vertexIterator.hasNext()) {
+ return Optional.of(vertexIterator.next());
+ }
+
+ return Optional.empty();
+ }
+
+ public static final class P {
+
+ public static final String ROLE_DATA = Graph.Hidden.hide("role_data");
+
+ public static final String LABEL = T.label.getAccessor();
+
+ public static final String NODE = Graph.Hidden.hide("role_node");
+
+ public static final String CLOCK = Graph.Hidden.hide("role_clock");
+
+ public static final String EPOCH = Graph.Hidden.hide("role_epoch");
+
+ public static final String URL = Graph.Hidden.hide("role_url");
+
+ public static final String TYPE = Graph.Hidden.hide("role_type");
+ }
+
+
+ public static final class Schema extends SchemaDefine {
+
+ public Schema(HugeGraphParams graph) {
+ super(graph, P.ROLE_DATA);
+ }
+
+ @Override
+ public void initSchemaIfNeeded() {
+ if (this.existVertexLabel(this.label)) {
+ return;
+ }
+
+ String[] properties = this.initProperties();
+
+ VertexLabel label = this.schema()
+ .vertexLabel(this.label)
+ .enableLabelIndex(true)
+ .usePrimaryKeyId()
+ .primaryKeys(P.TYPE)
+ .properties(properties)
+ .build();
+ this.graph.schemaTransaction().addVertexLabel(label);
+ }
+
+ private String[] initProperties() {
+ List<String> props = new ArrayList<>();
+
+ props.add(createPropertyKey(P.NODE, DataType.TEXT));
+ props.add(createPropertyKey(P.URL, DataType.TEXT));
+ props.add(createPropertyKey(P.CLOCK, DataType.LONG));
+ props.add(createPropertyKey(P.EPOCH, DataType.INT));
+ props.add(createPropertyKey(P.TYPE, DataType.TEXT));
+
+ return super.initProperties(props);
+ }
+ }
+}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleElectionStateMachineImpl.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleElectionStateMachine.java
similarity index 62%
rename from hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleElectionStateMachineImpl.java
rename to hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleElectionStateMachine.java
index 949d8648a..aa284def6 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/RoleElectionStateMachineImpl.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardRoleElectionStateMachine.java
@@ -15,24 +15,28 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.masterelection;
import java.security.SecureRandom;
import java.util.Optional;
import java.util.concurrent.locks.LockSupport;
import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
-public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
+public class StandardRoleElectionStateMachine implements RoleElectionStateMachine {
+
+ private static final Logger LOG = Log.logger(StandardRoleElectionStateMachine.class);
private volatile boolean shutdown;
private final Config config;
private volatile RoleState state;
- private final RoleTypeDataAdapter roleTypeDataAdapter;
+ private final ClusterRoleStore clusterRoleStore;
- public RoleElectionStateMachineImpl(Config config, RoleTypeDataAdapter adapter) {
+ public StandardRoleElectionStateMachine(Config config, ClusterRoleStore clusterRoleStore) {
this.config = config;
- this.roleTypeDataAdapter = adapter;
+ this.clusterRoleStore = clusterRoleStore;
this.state = new UnknownState(null);
this.shutdown = false;
}
@@ -49,7 +53,11 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
while (!this.shutdown) {
E.checkArgumentNotNull(this.state, "State don't be null");
try {
+ RoleState pre = this.state;
this.state = state.transform(context);
+ LOG.trace("server {} epoch {} role state change {} to {}",
+ context.node(), context.epoch(), pre.getClass().getSimpleName(),
+ this.state.getClass().getSimpleName());
Callback runnable = this.state.callback(stateMachineCallback);
runnable.call(context);
failCount = 0;
@@ -102,28 +110,29 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
@Override
public RoleState transform(StateMachineContext context) {
- RoleTypeDataAdapter adapter = context.adapter();
- Optional<RoleTypeData> roleTypeDataOpt = adapter.query();
- if (!roleTypeDataOpt.isPresent()) {
+ ClusterRoleStore adapter = context.adapter();
+ Optional<ClusterRole> clusterRoleOpt = adapter.query();
+ if (!clusterRoleOpt.isPresent()) {
context.reset();
Integer nextEpoch = this.epoch == null ? 1 : this.epoch + 1;
context.epoch(nextEpoch);
return new CandidateState(nextEpoch);
}
- RoleTypeData roleTypeData = roleTypeDataOpt.get();
- if (this.epoch != null && roleTypeData.epoch() < this.epoch) {
+ ClusterRole clusterRole = clusterRoleOpt.get();
+ if (this.epoch != null && clusterRole.epoch() < this.epoch) {
context.reset();
Integer nextEpoch = this.epoch + 1;
context.epoch(nextEpoch);
return new CandidateState(nextEpoch);
}
- context.epoch(roleTypeData.epoch());
- if (roleTypeData.isMaster(context.node())) {
- return new MasterState(roleTypeData);
+ context.epoch(clusterRole.epoch());
+ context.master(new MasterServerInfoImpl(clusterRole.node(), clusterRole.url()));
+ if (clusterRole.isMaster(context.node())) {
+ return new MasterState(clusterRole);
} else {
- return new WorkerState(roleTypeData);
+ return new WorkerState(clusterRole);
}
}
@@ -143,60 +152,61 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
@Override
public RoleState transform(StateMachineContext context) {
+ context.master(null);
RoleState.heartBeatPark(context);
return new UnknownState(this.epoch).transform(context);
}
@Override
public Callback callback(StateMachineCallback callback) {
- return callback::abdication;
+ return callback::onAsRoleAbdication;
}
}
private static class MasterState implements RoleState {
- private final RoleTypeData roleTypeData;
+ private final ClusterRole clusterRole;
- public MasterState(RoleTypeData roleTypeData) {
- this.roleTypeData = roleTypeData;
+ public MasterState(ClusterRole clusterRole) {
+ this.clusterRole = clusterRole;
}
@Override
public RoleState transform(StateMachineContext context) {
- this.roleTypeData.increaseClock();
+ this.clusterRole.increaseClock();
RoleState.heartBeatPark(context);
- if (context.adapter().updateIfNodePresent(this.roleTypeData)) {
+ if (context.adapter().updateIfNodePresent(this.clusterRole)) {
return this;
}
context.reset();
- context.epoch(this.roleTypeData.epoch());
- return new UnknownState(this.roleTypeData.epoch()).transform(context);
+ context.epoch(this.clusterRole.epoch());
+ return new UnknownState(this.clusterRole.epoch()).transform(context);
}
@Override
public Callback callback(StateMachineCallback callback) {
- return callback::master;
+ return callback::onAsRoleMaster;
}
}
private static class WorkerState implements RoleState {
- private RoleTypeData roleTypeData;
+ private ClusterRole clusterRole;
private int clock;
- public WorkerState(RoleTypeData roleTypeData) {
- this.roleTypeData = roleTypeData;
+ public WorkerState(ClusterRole clusterRole) {
+ this.clusterRole = clusterRole;
this.clock = 0;
}
@Override
public RoleState transform(StateMachineContext context) {
RoleState.heartBeatPark(context);
- RoleState nextState = new UnknownState(this.roleTypeData.epoch()).transform(context);
+ RoleState nextState = new UnknownState(this.clusterRole.epoch()).transform(context);
if (nextState instanceof WorkerState) {
this.merge((WorkerState) nextState);
- if (this.clock > context.config().exceedsWorkerCount()) {
- return new CandidateState(this.roleTypeData.epoch() + 1);
+ if (this.clock > context.config().masterDeadTimes()) {
+ return new CandidateState(this.clusterRole.epoch() + 1);
} else {
return this;
}
@@ -207,22 +217,22 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
@Override
public Callback callback(StateMachineCallback callback) {
- return callback::worker;
+ return callback::onAsRoleWorker;
}
public void merge(WorkerState state) {
- if (state.roleTypeData.epoch() > this.roleTypeData.epoch()) {
+ if (state.clusterRole.epoch() > this.clusterRole.epoch()) {
this.clock = 0;
- this.roleTypeData = state.roleTypeData;
- } else if (state.roleTypeData.epoch() < this.roleTypeData.epoch()) {
+ this.clusterRole = state.clusterRole;
+ } else if (state.clusterRole.epoch() < this.clusterRole.epoch()) {
throw new IllegalStateException("Epoch must increase");
- } else if (state.roleTypeData.epoch() == this.roleTypeData.epoch() &&
- state.roleTypeData.clock() < this.roleTypeData.clock()) {
+ } else if (state.clusterRole.epoch() == this.clusterRole.epoch() &&
+ state.clusterRole.clock() < this.clusterRole.clock()) {
throw new IllegalStateException("Clock must increase");
- } else if (state.roleTypeData.epoch() == this.roleTypeData.epoch() &&
- state.roleTypeData.clock() > this.roleTypeData.clock()) {
+ } else if (state.clusterRole.epoch() == this.clusterRole.epoch() &&
+ state.clusterRole.clock() > this.clusterRole.clock()) {
this.clock = 0;
- this.roleTypeData = state.roleTypeData;
+ this.clusterRole = state.clusterRole;
} else {
this.clock++;
}
@@ -241,11 +251,13 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
public RoleState transform(StateMachineContext context) {
RoleState.randomPark(context);
int epoch = this.epoch == null ? 1 : this.epoch;
- RoleTypeData roleTypeData = new RoleTypeData(context.config().node(), epoch);
- //failover to master success
- context.epoch(roleTypeData.epoch());
- if (context.adapter().updateIfNodePresent(roleTypeData)) {
- return new MasterState(roleTypeData);
+ ClusterRole clusterRole = new ClusterRole(context.config().node(),
+ context.config().url(), epoch);
+ // The master failover completed
+ context.epoch(clusterRole.epoch());
+ if (context.adapter().updateIfNodePresent(clusterRole)) {
+ context.master(new MasterServerInfoImpl(clusterRole.node(), clusterRole.url()));
+ return new MasterState(clusterRole);
} else {
return new UnknownState(epoch).transform(context);
}
@@ -253,7 +265,7 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
@Override
public Callback callback(StateMachineCallback callback) {
- return callback::candidate;
+ return callback::onAsRoleCandidate;
}
}
@@ -261,13 +273,20 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
private Integer epoch;
private final String node;
- private final RoleElectionStateMachineImpl machine;
+ private final StandardRoleElectionStateMachine machine;
+
+ private MasterServerInfo masterServerInfo;
- public StateMachineContextImpl(RoleElectionStateMachineImpl machine) {
+ public StateMachineContextImpl(StandardRoleElectionStateMachine machine) {
this.node = machine.config.node();
this.machine = machine;
}
+ @Override
+ public void master(MasterServerInfo info) {
+ this.masterServerInfo = info;
+ }
+
@Override
public Integer epoch() {
return this.epoch;
@@ -284,7 +303,7 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
}
@Override
- public RoleTypeDataAdapter adapter() {
+ public ClusterRoleStore adapter() {
return this.machine.adapter();
}
@@ -293,6 +312,11 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
return this.machine.config;
}
+ @Override
+ public MasterServerInfo master() {
+ return this.masterServerInfo;
+ }
+
@Override
public RoleElectionStateMachine stateMachine() {
return this.machine;
@@ -304,7 +328,28 @@ public class RoleElectionStateMachineImpl implements RoleElectionStateMachine {
}
}
- protected RoleTypeDataAdapter adapter() {
- return this.roleTypeDataAdapter;
+ private static class MasterServerInfoImpl implements StateMachineContext.MasterServerInfo {
+
+ private final String node;
+ private final String url;
+
+ public MasterServerInfoImpl(String node, String url) {
+ this.node = node;
+ this.url = url;
+ }
+
+ @Override
+ public String url() {
+ return this.url;
+ }
+
+ @Override
+ public String node() {
+ return this.node;
+ }
+ }
+
+ protected ClusterRoleStore adapter() {
+ return this.clusterRoleStore;
}
}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardStateMachineCallback.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardStateMachineCallback.java
new file mode 100644
index 000000000..11a989ecd
--- /dev/null
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StandardStateMachineCallback.java
@@ -0,0 +1,105 @@
+/*
+ * 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.masterelection;
+
+import java.util.Objects;
+
+import org.apache.hugegraph.task.TaskManager;
+import org.apache.hugegraph.util.Log;
+import org.slf4j.Logger;
+
+public class StandardStateMachineCallback implements StateMachineCallback {
+
+ private static final Logger LOG = Log.logger(StandardStateMachineCallback.class);
+
+ private final TaskManager taskManager;
+
+ private final GlobalMasterInfo globalMasterInfo;
+
+ private boolean isMaster = false;
+
+ public StandardStateMachineCallback(TaskManager taskManager, GlobalMasterInfo globalMasterInfo) {
+ this.taskManager = taskManager;
+ this.taskManager.enableRoleElected(true);
+ this.globalMasterInfo = globalMasterInfo;
+ this.globalMasterInfo.isFeatureSupport(true);
+ }
+
+ @Override
+ public void onAsRoleMaster(StateMachineContext context) {
+ if (!isMaster) {
+ this.taskManager.onAsRoleMaster();
+ LOG.info("Server {} change to master role", context.config().node());
+ }
+ this.initGlobalMasterInfo(context);
+ this.isMaster = true;
+ }
+
+ @Override
+ public void onAsRoleWorker(StateMachineContext context) {
+ if (isMaster) {
+ this.taskManager.onAsRoleWorker();
+ LOG.info("Server {} change to worker role", context.config().node());
+ }
+ this.initGlobalMasterInfo(context);
+ this.isMaster = false;
+ }
+
+ @Override
+ public void onAsRoleCandidate(StateMachineContext context) {
+ }
+
+ @Override
+ public void unknown(StateMachineContext context) {
+ if (isMaster) {
+ this.taskManager.onAsRoleWorker();
+ LOG.info("Server {} change to worker role", context.config().node());
+ }
+ this.initGlobalMasterInfo(context);
+
+ isMaster = false;
+ }
+
+ @Override
+ public void onAsRoleAbdication(StateMachineContext context) {
+ if (isMaster) {
+ this.taskManager.onAsRoleWorker();
+ LOG.info("Server {} change to worker role", context.config().node());
+ }
+ this.initGlobalMasterInfo(context);
+
+ isMaster = false;
+ }
+
+ @Override
+ public void error(StateMachineContext context, Throwable e) {
+ LOG.error("Server {} exception occurred", context.config().node(), e);
+ }
+
+ public void initGlobalMasterInfo(StateMachineContext context) {
+ StateMachineContext.MasterServerInfo master = context.master();
+ if (master == null) {
+ this.globalMasterInfo.set(false, null);
+ return;
+ }
+
+ boolean isMaster = Objects.equals(context.node(), master.node());
+ String url = master.url();
+ this.globalMasterInfo.set(isMaster, url);
+ }
+}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineCallback.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineCallback.java
similarity index 78%
rename from hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineCallback.java
rename to hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineCallback.java
index 02c58fb62..35abbe340 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineCallback.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineCallback.java
@@ -15,19 +15,19 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.masterelection;
public interface StateMachineCallback {
- void master(StateMachineContext context);
+ void onAsRoleMaster(StateMachineContext context);
- void worker(StateMachineContext context);
+ void onAsRoleWorker(StateMachineContext context);
- void candidate(StateMachineContext context);
+ void onAsRoleCandidate(StateMachineContext context);
void unknown(StateMachineContext context);
- void abdication(StateMachineContext context);
+ void onAsRoleAbdication(StateMachineContext context);
void error(StateMachineContext context, Throwable e);
}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineContext.java b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineContext.java
similarity index 80%
rename from hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineContext.java
rename to hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineContext.java
index e314a352c..a3eaf8962 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/election/StateMachineContext.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/masterelection/StateMachineContext.java
@@ -15,7 +15,7 @@
* under the License.
*/
-package org.apache.hugegraph.election;
+package org.apache.hugegraph.masterelection;
public interface StateMachineContext {
@@ -29,7 +29,18 @@ public interface StateMachineContext {
Config config();
- RoleTypeDataAdapter adapter();
+ MasterServerInfo master();
+
+ void master(MasterServerInfo info);
+
+ ClusterRoleStore adapter();
void reset();
+
+ interface MasterServerInfo {
+
+ String url();
+
+ String node();
+ }
}
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/task/ServerInfoManager.java b/hugegraph-core/src/main/java/org/apache/hugegraph/task/ServerInfoManager.java
index 830b6260c..e8cccf88e 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/task/ServerInfoManager.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/task/ServerInfoManager.java
@@ -25,31 +25,30 @@ import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.HugeGraphParams;
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.page.PageInfo;
import org.apache.hugegraph.backend.query.Condition;
import org.apache.hugegraph.backend.query.ConditionQuery;
-import org.apache.hugegraph.backend.query.Query;
import org.apache.hugegraph.backend.query.QueryResults;
import org.apache.hugegraph.backend.tx.GraphTransaction;
+import org.apache.hugegraph.exception.ConnectionException;
+import org.apache.hugegraph.iterator.ListIterator;
+import org.apache.hugegraph.iterator.MapperIterator;
import org.apache.hugegraph.schema.PropertyKey;
import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.type.define.HugeKeys;
import org.apache.hugegraph.type.define.NodeRole;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.slf4j.Logger;
-
-import org.apache.hugegraph.HugeException;
-import org.apache.hugegraph.HugeGraph;
-import org.apache.hugegraph.HugeGraphParams;
-import org.apache.hugegraph.exception.ConnectionException;
-import org.apache.hugegraph.iterator.ListIterator;
-import org.apache.hugegraph.iterator.MapperIterator;
-import org.apache.hugegraph.structure.HugeVertex;
import org.apache.hugegraph.util.DateUtil;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.slf4j.Logger;
+
import com.google.common.collect.ImmutableMap;
public class ServerInfoManager {
@@ -104,6 +103,19 @@ public class ServerInfoManager {
return true;
}
+ public synchronized void forceInitServerInfo(Id server, NodeRole role) {
+ if (this.closed) {
+ return;
+ }
+
+ E.checkArgument(server != null && role != null,
+ "The server id or role can't be null");
+ this.selfServerId = server;
+ this.selfServerRole = role;
+
+ this.saveServerInfo(this.selfServerId, this.selfServerRole);
+ }
+
public synchronized void initServerInfo(Id server, NodeRole role) {
E.checkArgument(server != null && role != null,
"The server id or role can't be null");
@@ -180,9 +192,8 @@ public class ServerInfoManager {
return !this.closed && this.graph.started() && this.graph.initialized();
}
- protected synchronized HugeServerInfo pickWorkerNode(
- Collection<HugeServerInfo> servers,
- HugeTask<?> task) {
+ protected synchronized HugeServerInfo pickWorkerNode(Collection<HugeServerInfo> servers,
+ HugeTask<?> task) {
HugeServerInfo master = null;
HugeServerInfo serverWithMinLoad = null;
int minLoad = Integer.MAX_VALUE;
@@ -259,7 +270,7 @@ public class ServerInfoManager {
private int save(Collection<HugeServerInfo> serverInfos) {
return this.call(() -> {
if (serverInfos.isEmpty()) {
- return serverInfos.size();
+ return 0;
}
HugeServerInfo.Schema schema = HugeServerInfo.schema(this.graph);
if (!schema.existVertexLabel(HugeServerInfo.P.SERVER)) {
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java b/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java
index b5b95256f..9eda3f6b0 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/task/StandardTaskScheduler.java
@@ -29,6 +29,9 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
+import org.apache.hugegraph.HugeException;
+import org.apache.hugegraph.HugeGraph;
+import org.apache.hugegraph.HugeGraphParams;
import org.apache.hugegraph.backend.id.Id;
import org.apache.hugegraph.backend.page.PageInfo;
import org.apache.hugegraph.backend.query.Condition;
@@ -37,32 +40,29 @@ import org.apache.hugegraph.backend.query.QueryResults;
import org.apache.hugegraph.backend.store.BackendStore;
import org.apache.hugegraph.backend.tx.GraphTransaction;
import org.apache.hugegraph.config.CoreOptions;
+import org.apache.hugegraph.exception.ConnectionException;
+import org.apache.hugegraph.exception.NotFoundException;
+import org.apache.hugegraph.iterator.ExtendableIterator;
+import org.apache.hugegraph.iterator.MapperIterator;
+import org.apache.hugegraph.job.EphemeralJob;
import org.apache.hugegraph.schema.IndexLabel;
import org.apache.hugegraph.schema.PropertyKey;
import org.apache.hugegraph.schema.SchemaManager;
import org.apache.hugegraph.schema.VertexLabel;
+import org.apache.hugegraph.structure.HugeVertex;
+import org.apache.hugegraph.task.HugeTask.P;
+import org.apache.hugegraph.task.TaskCallable.SysTaskCallable;
+import org.apache.hugegraph.task.TaskManager.ContextCallable;
import org.apache.hugegraph.type.HugeType;
import org.apache.hugegraph.type.define.Cardinality;
import org.apache.hugegraph.type.define.DataType;
import org.apache.hugegraph.type.define.HugeKeys;
+import org.apache.hugegraph.util.E;
+import org.apache.hugegraph.util.Log;
import org.apache.tinkerpop.gremlin.structure.Graph.Hidden;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.slf4j.Logger;
-import org.apache.hugegraph.HugeException;
-import org.apache.hugegraph.HugeGraph;
-import org.apache.hugegraph.HugeGraphParams;
-import org.apache.hugegraph.exception.ConnectionException;
-import org.apache.hugegraph.exception.NotFoundException;
-import org.apache.hugegraph.iterator.ExtendableIterator;
-import org.apache.hugegraph.iterator.MapperIterator;
-import org.apache.hugegraph.job.EphemeralJob;
-import org.apache.hugegraph.structure.HugeVertex;
-import org.apache.hugegraph.task.HugeTask.P;
-import org.apache.hugegraph.task.TaskCallable.SysTaskCallable;
-import org.apache.hugegraph.task.TaskManager.ContextCallable;
-import org.apache.hugegraph.util.E;
-import org.apache.hugegraph.util.Log;
import com.google.common.collect.ImmutableMap;
public class StandardTaskScheduler implements TaskScheduler {
@@ -123,7 +123,7 @@ public class StandardTaskScheduler implements TaskScheduler {
* NOTE: don't synchronized(this) due to scheduler thread hold
* this lock through scheduleTasks(), then query tasks and wait
* for db-worker thread after call(), the tx may not be initialized
- * but can't catch this lock, then cause dead lock.
+ * but can't catch this lock, then cause deadlock.
* We just use this.serverManager as a monitor here
*/
synchronized (this.serverManager) {
@@ -318,6 +318,10 @@ public class StandardTaskScheduler implements TaskScheduler {
continue;
}
+ if (!this.serverManager.master()) {
+ return;
+ }
+
HugeServerInfo server = this.serverManager().pickWorkerNode(
scheduleInfos, task);
if (server == null) {
@@ -632,7 +636,7 @@ public class StandardTaskScheduler implements TaskScheduler {
public void waitUntilAllTasksCompleted(long seconds)
throws TimeoutException {
long passes = seconds * 1000 / QUERY_INTERVAL;
- int taskSize = 0;
+ int taskSize;
for (long pass = 0;; pass++) {
taskSize = this.pendingTasks();
if (taskSize == 0) {
diff --git a/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java b/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java
index b01318154..524a1f759 100644
--- a/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java
+++ b/hugegraph-core/src/main/java/org/apache/hugegraph/task/TaskManager.java
@@ -26,6 +26,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import org.apache.hugegraph.type.define.NodeRole;
import org.apache.hugegraph.util.*;
import org.apache.hugegraph.util.Consumers;
import org.apache.hugegraph.util.LockUtil;
@@ -58,6 +59,8 @@ public final class TaskManager {
private final ExecutorService serverInfoDbExecutor;
private final PausableScheduledThreadPool schedulerExecutor;
+ private boolean enableRoleElected = false;
+
public static TaskManager instance() {
return MANAGER;
}
@@ -254,6 +257,36 @@ public final class TaskManager {
}
}
+ public void onAsRoleMaster() {
+ try {
+ for (TaskScheduler entry : this.schedulers.values()) {
+ StandardTaskScheduler scheduler = (StandardTaskScheduler) entry;
+ ServerInfoManager serverInfoManager = scheduler.serverManager();
+ serverInfoManager.forceInitServerInfo(serverInfoManager.selfServerId(), NodeRole.MASTER);
+ }
+ } catch (Throwable e) {
+ LOG.error("Exception occurred when change to master role", e);
+ throw e;
+ }
+ }
+
+ public void onAsRoleWorker() {
+ try {
+ for (TaskScheduler entry : this.schedulers.values()) {
+ StandardTaskScheduler scheduler = (StandardTaskScheduler) entry;
+ ServerInfoManager serverInfoManager = scheduler.serverManager();
+ serverInfoManager.forceInitServerInfo(serverInfoManager.selfServerId(), NodeRole.WORKER);
+ }
+ } catch (Throwable e) {
+ LOG.error("Exception occurred when change to worker role", e);
+ throw e;
+ }
+ }
+
+ public void enableRoleElected(boolean enableRoleElected) {
+ this.enableRoleElected = enableRoleElected;
+ }
+
private void scheduleOrExecuteJob() {
// Called by scheduler timer
try {
@@ -299,11 +332,15 @@ public final class TaskManager {
/*
* Master schedule tasks to suitable servers.
+ * Worker maybe become Master, so Master also need perform tasks assigned by
+ * previous Master when enableRoleElected is true.
+ * However, the master only needs to take the assignment,
+ * because the master stays the same when enableRoleElected is false.
* There is no suitable server when these tasks are created
*/
if (serverManager.master()) {
scheduler.scheduleTasks();
- if (!serverManager.onlySingleNode()) {
+ if (!this.enableRoleElected && !serverManager.onlySingleNode()) {
return;
}
}
diff --git a/hugegraph-dist/src/main/java/org/apache/hugegraph/dist/RegisterUtil.java b/hugegraph-dist/src/main/java/org/apache/hugegraph/dist/RegisterUtil.java
index 5627d55b6..34e5ea3cf 100644
--- a/hugegraph-dist/src/main/java/org/apache/hugegraph/dist/RegisterUtil.java
+++ b/hugegraph-dist/src/main/java/org/apache/hugegraph/dist/RegisterUtil.java
@@ -24,19 +24,19 @@ import java.util.ServiceLoader;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.configuration2.builder.fluent.Configurations;
import org.apache.commons.configuration2.ex.ConfigurationException;
-import org.slf4j.Logger;
-
import org.apache.hugegraph.HugeException;
import org.apache.hugegraph.backend.serializer.SerializerFactory;
import org.apache.hugegraph.backend.store.BackendProviderFactory;
import org.apache.hugegraph.config.CoreOptions;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.OptionSpace;
+import org.apache.hugegraph.masterelection.RoleElectionOptions;
import org.apache.hugegraph.plugin.HugeGraphPlugin;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
import org.apache.hugegraph.util.VersionUtil;
import org.apache.hugegraph.version.CoreVersion;
+import org.slf4j.Logger;
public class RegisterUtil {
@@ -45,6 +45,7 @@ public class RegisterUtil {
static {
OptionSpace.register("core", CoreOptions.instance());
OptionSpace.register("dist", DistOptions.instance());
+ OptionSpace.register("masterElection", RoleElectionOptions.instance());
}
public static void registerBackends() {
@@ -53,7 +54,7 @@ public class RegisterUtil {
E.checkState(input != null,
"Can't read file '%s' as stream", confFile);
- PropertiesConfiguration props = null;
+ PropertiesConfiguration props;
try {
props = new Configurations().properties(input);
} catch (ConfigurationException e) {
@@ -91,15 +92,14 @@ public class RegisterUtil {
registerPostgresql();
break;
default:
- throw new HugeException("Unsupported backend type '%s'",
- backend);
+ throw new HugeException("Unsupported backend type '%s'", backend);
}
}
public static void registerCassandra() {
// Register config
OptionSpace.register("cassandra",
- "org.apache.hugegraph.backend.store.cassandra.CassandraOptions");
+ "org.apache.hugegraph.backend.store.cassandra.CassandraOptions");
// Register serializer
SerializerFactory.register("cassandra",
"org.apache.hugegraph.backend.store.cassandra.CassandraSerializer");
@@ -111,7 +111,7 @@ public class RegisterUtil {
public static void registerScyllaDB() {
// Register config
OptionSpace.register("scylladb",
- "org.apache.hugegraph.backend.store.cassandra.CassandraOptions");
+ "org.apache.hugegraph.backend.store.cassandra.CassandraOptions");
// Register serializer
SerializerFactory.register("scylladb",
"org.apache.hugegraph.backend.store.cassandra.CassandraSerializer");
@@ -123,10 +123,10 @@ public class RegisterUtil {
public static void registerHBase() {
// Register config
OptionSpace.register("hbase",
- "org.apache.hugegraph.backend.store.hbase.HbaseOptions");
+ "org.apache.hugegraph.backend.store.hbase.HbaseOptions");
// Register serializer
SerializerFactory.register("hbase",
- "org.apache.hugegraph.backend.store.hbase.HbaseSerializer");
+ "org.apache.hugegraph.backend.store.hbase.HbaseSerializer");
// Register backend
BackendProviderFactory.register("hbase",
"org.apache.hugegraph.backend.store.hbase.HbaseStoreProvider");
@@ -135,7 +135,7 @@ public class RegisterUtil {
public static void registerRocksDB() {
// Register config
OptionSpace.register("rocksdb",
- "org.apache.hugegraph.backend.store.rocksdb.RocksDBOptions");
+ "org.apache.hugegraph.backend.store.rocksdb.RocksDBOptions");
// Register backend
BackendProviderFactory.register("rocksdb",
"org.apache.hugegraph.backend.store.rocksdb.RocksDBStoreProvider");
@@ -146,10 +146,10 @@ public class RegisterUtil {
public static void registerMysql() {
// Register config
OptionSpace.register("mysql",
- "org.apache.hugegraph.backend.store.mysql.MysqlOptions");
+ "org.apache.hugegraph.backend.store.mysql.MysqlOptions");
// Register serializer
SerializerFactory.register("mysql",
- "org.apache.hugegraph.backend.store.mysql.MysqlSerializer");
+ "org.apache.hugegraph.backend.store.mysql.MysqlSerializer");
// Register backend
BackendProviderFactory.register("mysql",
"org.apache.hugegraph.backend.store.mysql.MysqlStoreProvider");
@@ -158,7 +158,7 @@ public class RegisterUtil {
public static void registerPalo() {
// Register config
OptionSpace.register("palo",
- "org.apache.hugegraph.backend.store.palo.PaloOptions");
+ "org.apache.hugegraph.backend.store.palo.PaloOptions");
// Register serializer
SerializerFactory.register("palo",
"org.apache.hugegraph.backend.store.palo.PaloSerializer");
@@ -192,8 +192,7 @@ public class RegisterUtil {
* Scan the jars in plugins directory and load them
*/
public static void registerPlugins() {
- ServiceLoader<HugeGraphPlugin> plugins = ServiceLoader.load(
- HugeGraphPlugin.class);
+ ServiceLoader<HugeGraphPlugin> plugins = ServiceLoader.load(HugeGraphPlugin.class);
for (HugeGraphPlugin plugin : plugins) {
LOG.info("Loading plugin {}({})",
plugin.name(), plugin.getClass().getCanonicalName());
@@ -212,8 +211,7 @@ public class RegisterUtil {
plugin.register();
LOG.info("Loaded plugin '{}'", plugin.name());
} catch (Exception e) {
- throw new HugeException("Failed to load plugin '%s'",
- plugin.name(), e);
+ throw new HugeException("Failed to load plugin '%s'", plugin.name(), e);
}
}
}
diff --git a/hugegraph-test/src/main/java/org/apache/hugegraph/core/RoleElectionStateMachineTest.java b/hugegraph-test/src/main/java/org/apache/hugegraph/core/RoleElectionStateMachineTest.java
index a9972a469..9fbbb5e62 100644
--- a/hugegraph-test/src/main/java/org/apache/hugegraph/core/RoleElectionStateMachineTest.java
+++ b/hugegraph-test/src/main/java/org/apache/hugegraph/core/RoleElectionStateMachineTest.java
@@ -30,13 +30,13 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.LockSupport;
-import org.apache.hugegraph.election.Config;
-import org.apache.hugegraph.election.RoleElectionStateMachine;
-import org.apache.hugegraph.election.RoleElectionStateMachineImpl;
-import org.apache.hugegraph.election.RoleTypeData;
-import org.apache.hugegraph.election.RoleTypeDataAdapter;
-import org.apache.hugegraph.election.StateMachineCallback;
-import org.apache.hugegraph.election.StateMachineContext;
+import org.apache.hugegraph.masterelection.ClusterRole;
+import org.apache.hugegraph.masterelection.ClusterRoleStore;
+import org.apache.hugegraph.masterelection.Config;
+import org.apache.hugegraph.masterelection.RoleElectionStateMachine;
+import org.apache.hugegraph.masterelection.StandardRoleElectionStateMachine;
+import org.apache.hugegraph.masterelection.StateMachineCallback;
+import org.apache.hugegraph.masterelection.StateMachineContext;
import org.apache.hugegraph.testutil.Assert;
import org.apache.hugegraph.testutil.Utils;
import org.junit.Test;
@@ -86,10 +86,10 @@ public class RoleElectionStateMachineTest {
@Override
public String toString() {
return "LogEntry{" +
- "epoch=" + epoch +
- ", node='" + node + '\'' +
- ", role=" + role +
- '}';
+ "epoch=" + epoch +
+ ", node='" + node + '\'' +
+ ", role=" + role +
+ '}';
}
}
@@ -106,6 +106,11 @@ public class RoleElectionStateMachineTest {
return this.node;
}
+ @Override
+ public String url() {
+ return "http://127.0.0.1:8080";
+ }
+
@Override
public int exceedsFailCount() {
return 2;
@@ -122,7 +127,7 @@ public class RoleElectionStateMachineTest {
}
@Override
- public int exceedsWorkerCount() {
+ public int masterDeadTimes() {
return 5;
}
@@ -141,7 +146,7 @@ public class RoleElectionStateMachineTest {
final StateMachineCallback callback = new StateMachineCallback() {
@Override
- public void master(StateMachineContext context) {
+ public void onAsRoleMaster(StateMachineContext context) {
Integer epochId = context.epoch();
String node = context.node();
logRecords.add(new LogEntry(epochId, node, LogEntry.Role.master));
@@ -153,7 +158,7 @@ public class RoleElectionStateMachineTest {
}
@Override
- public void worker(StateMachineContext context) {
+ public void onAsRoleWorker(StateMachineContext context) {
Integer epochId = context.epoch();
String node = context.node();
logRecords.add(new LogEntry(epochId, node, LogEntry.Role.worker));
@@ -163,7 +168,7 @@ public class RoleElectionStateMachineTest {
}
@Override
- public void candidate(StateMachineContext context) {
+ public void onAsRoleCandidate(StateMachineContext context) {
Integer epochId = context.epoch();
String node = context.node();
logRecords.add(new LogEntry(epochId, node, LogEntry.Role.candidate));
@@ -183,7 +188,7 @@ public class RoleElectionStateMachineTest {
}
@Override
- public void abdication(StateMachineContext context) {
+ public void onAsRoleAbdication(StateMachineContext context) {
Integer epochId = context.epoch();
String node = context.node();
logRecords.add(new LogEntry(epochId, node, LogEntry.Role.abdication));
@@ -200,32 +205,33 @@ public class RoleElectionStateMachineTest {
}
};
- final List<RoleTypeData> metaDataLogs = Collections.synchronizedList(new ArrayList<>(100));
- final RoleTypeDataAdapter adapter = new RoleTypeDataAdapter() {
+ final List<ClusterRole> clusterRoleLogs = Collections.synchronizedList(new ArrayList<>(100));
+ final ClusterRoleStore clusterRoleStore = new ClusterRoleStore() {
volatile int epoch = 0;
- final Map<Integer, RoleTypeData> data = new ConcurrentHashMap<>();
+ final Map<Integer, ClusterRole> data = new ConcurrentHashMap<>();
- RoleTypeData copy(RoleTypeData stateData) {
- if (stateData == null) {
+ ClusterRole copy(ClusterRole clusterRole) {
+ if (clusterRole == null) {
return null;
}
- return new RoleTypeData(stateData.node(), stateData.epoch(), stateData.clock());
+ return new ClusterRole(clusterRole.node(), clusterRole.url(),
+ clusterRole.epoch(), clusterRole.clock());
}
@Override
- public boolean updateIfNodePresent(RoleTypeData stateData) {
- if (stateData.epoch() < this.epoch) {
+ public boolean updateIfNodePresent(ClusterRole clusterRole) {
+ if (clusterRole.epoch() < this.epoch) {
return false;
}
- RoleTypeData copy = this.copy(stateData);
- RoleTypeData newData = data.compute(copy.epoch(), (key, value) -> {
+ ClusterRole copy = this.copy(clusterRole);
+ ClusterRole newClusterRole = data.compute(copy.epoch(), (key, value) -> {
if (copy.epoch() > this.epoch) {
this.epoch = copy.epoch();
Assert.assertNull(value);
- metaDataLogs.add(copy);
+ clusterRoleLogs.add(copy);
Utils.println("The node " + copy + " become new master:");
return copy;
}
@@ -234,7 +240,7 @@ public class RoleElectionStateMachineTest {
if (Objects.equals(value.node(), copy.node()) &&
value.clock() <= copy.clock()) {
Utils.println("The master node " + copy + " keep heartbeat");
- metaDataLogs.add(copy);
+ clusterRoleLogs.add(copy);
if (value.clock() == copy.clock()) {
Assert.fail("Clock must increase when same epoch and node id");
}
@@ -243,13 +249,12 @@ public class RoleElectionStateMachineTest {
return value;
});
- return Objects.equals(newData, copy);
+ return Objects.equals(newClusterRole, copy);
}
@Override
- public Optional<RoleTypeData> query() {
- return Optional.ofNullable(
- this.copy(this.data.get(this.epoch)));
+ public Optional<ClusterRole> query() {
+ return Optional.ofNullable(this.copy(this.data.get(this.epoch)));
}
};
@@ -257,7 +262,7 @@ public class RoleElectionStateMachineTest {
Thread node1 = new Thread(() -> {
Config config = new TestConfig("1");
RoleElectionStateMachine stateMachine =
- new RoleElectionStateMachineImpl(config, adapter);
+ new StandardRoleElectionStateMachine(config, clusterRoleStore);
machines[1] = stateMachine;
stateMachine.apply(callback);
stop.countDown();
@@ -266,7 +271,7 @@ public class RoleElectionStateMachineTest {
Thread node2 = new Thread(() -> {
Config config = new TestConfig("2");
RoleElectionStateMachine stateMachine =
- new RoleElectionStateMachineImpl(config, adapter);
+ new StandardRoleElectionStateMachine(config, clusterRoleStore);
machines[2] = stateMachine;
stateMachine.apply(callback);
stop.countDown();
@@ -275,7 +280,7 @@ public class RoleElectionStateMachineTest {
Thread node3 = new Thread(() -> {
Config config = new TestConfig("3");
RoleElectionStateMachine stateMachine =
- new RoleElectionStateMachineImpl(config, adapter);
+ new StandardRoleElectionStateMachine(config, clusterRoleStore);
machines[3] = stateMachine;
stateMachine.apply(callback);
stop.countDown();
@@ -309,7 +314,7 @@ public class RoleElectionStateMachineTest {
Assert.assertGt(0, logRecords.size());
Map<Integer, String> masters = new HashMap<>();
- for (LogEntry entry: logRecords) {
+ for (LogEntry entry : logRecords) {
if (entry.role == LogEntry.Role.master) {
String lastNode = masters.putIfAbsent(entry.epoch, entry.node);
if (lastNode != null) {