You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kylin.apache.org by sh...@apache.org on 2015/12/02 03:57:39 UTC
[2/2] kylin git commit: KYLIN-1188 Kylin node self-discover
KYLIN-1188 Kylin node self-discover
Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/6bf9053f
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/6bf9053f
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/6bf9053f
Branch: refs/heads/helix-latest
Commit: 6bf9053f22ddf25faf23082d94c91120fef03814
Parents: b3680dd
Author: shaofengshi <sh...@apache.org>
Authored: Tue Dec 1 17:39:09 2015 +0800
Committer: shaofengshi <sh...@apache.org>
Committed: Tue Dec 1 17:39:09 2015 +0800
----------------------------------------------------------------------
.../org/apache/kylin/common/KylinConfig.java | 4 +-
.../kylin/rest/controller/JobController.java | 60 +++++++++++++-------
.../kylin/rest/helix/HelixJobEngineAdmin.java | 50 ++++++++++++++--
.../apache/kylin/rest/service/CubeService.java | 12 ++--
.../rest/helix/JobControllerConnectorTest.java | 16 +++---
5 files changed, 97 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kylin/blob/6bf9053f/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java
index f2ec347..9f2f1a9 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfig.java
@@ -759,8 +759,8 @@ public class KylinConfig implements Serializable {
return getMetadataUrl();
}
- public String getHelixClusterName() {
- return getMetadataUrlPrefix();
+ public String getClusterName() {
+ return this.getOptional("kylin.cluster.name", getMetadataUrlPrefix());
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/6bf9053f/server/src/main/java/org/apache/kylin/rest/controller/JobController.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/controller/JobController.java b/server/src/main/java/org/apache/kylin/rest/controller/JobController.java
index c0ebd8e..af8b1e6 100644
--- a/server/src/main/java/org/apache/kylin/rest/controller/JobController.java
+++ b/server/src/main/java/org/apache/kylin/rest/controller/JobController.java
@@ -21,15 +21,18 @@ package org.apache.kylin.rest.controller;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import joptsimple.internal.Strings;
+import org.apache.helix.ExternalViewChangeListener;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.model.ExternalView;
import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.restclient.Broadcaster;
import org.apache.kylin.job.JobInstance;
import org.apache.kylin.job.constant.JobStatusEnum;
-import org.apache.kylin.job.lock.JobLock;
import org.apache.kylin.rest.exception.InternalErrorException;
import org.apache.kylin.rest.helix.HelixJobEngineAdmin;
import org.apache.kylin.rest.helix.JobControllerConnector;
-import org.apache.kylin.rest.helix.JobControllerConstants;
import org.apache.kylin.rest.helix.v1.DefaultStateModelFactory;
+import org.apache.kylin.rest.helix.v1.JobEngineSMDV1;
import org.apache.kylin.rest.request.JobListRequest;
import org.apache.kylin.rest.service.JobService;
import org.slf4j.Logger;
@@ -45,8 +48,6 @@ import org.springframework.web.bind.annotation.ResponseBody;
import java.io.IOException;
import java.util.*;
-import static org.apache.kylin.rest.helix.JobControllerConstants.RESOURCE_NAME;
-
/**
* @author ysong1
* @author Jack
@@ -60,9 +61,6 @@ public class JobController extends BasicController implements InitializingBean {
@Autowired
private JobService jobService;
- @Autowired
- private JobLock jobLock;
-
/*
* (non-Javadoc)
*
@@ -76,30 +74,47 @@ public class JobController extends BasicController implements InitializingBean {
TimeZone tzone = TimeZone.getTimeZone(timeZone);
TimeZone.setDefault(tzone);
- if (System.getProperty("kylin.rest.address") == null) {
- throw new RuntimeException("There is no -Dkylin.rest.address set; Please check bin/kylin.sh");
- }
-
- final String restAddress = System.getProperty("kylin.rest.address");
- final String hostname = Preconditions.checkNotNull(restAddress.substring(0, restAddress.lastIndexOf(":")));
- final String port = Preconditions.checkNotNull(restAddress.substring(restAddress.lastIndexOf(":") + 1));
- final String instanceName = hostname + "_" + port;
+ final String instanceName = HelixJobEngineAdmin.getCurrentInstanceName();
final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
final String zkAddress = Preconditions.checkNotNull(kylinConfig.getZookeeperAddress());
- HelixJobEngineAdmin helixJobEngineAdmin = HelixJobEngineAdmin.getInstance(zkAddress);
- helixJobEngineAdmin.initV1(kylinConfig.getHelixClusterName(), JobControllerConstants.RESOURCE_NAME);
- helixJobEngineAdmin.startControllers(kylinConfig.getHelixClusterName());
- final JobControllerConnector jcc = new JobControllerConnector(hostname, port, zkAddress, kylinConfig.getHelixClusterName(), new DefaultStateModelFactory(instanceName, kylinConfig));
+ final HelixJobEngineAdmin helixJobEngineAdmin = HelixJobEngineAdmin.getInstance(zkAddress);
+ helixJobEngineAdmin.initV1(kylinConfig.getClusterName(), HelixJobEngineAdmin.RESOURCE_NAME);
+ helixJobEngineAdmin.startController(kylinConfig.getClusterName(), new ExternalViewChangeListener() {
+ @Override
+ public void onExternalViewChange(List<ExternalView> list, NotificationContext notificationContext) {
+ for (ExternalView view : list) {
+ if (view.getResourceName().equals(HelixJobEngineAdmin.RESOURCE_NAME)) {
+ final Set<String> partitionSet = view.getPartitionSet();
+ Preconditions.checkArgument(partitionSet.size() == 1);
+ final Map<String, String> stateMap = view.getStateMap(partitionSet.iterator().next());
+
+ List<String> liveInstances = Lists.newArrayList();
+ for (String instance : stateMap.keySet()) {
+ String state = stateMap.get(instance);
+ if (JobEngineSMDV1.States.LEADER.toString().equalsIgnoreCase(state)
+ || JobEngineSMDV1.States.STANDBY.toString().equalsIgnoreCase(state)) {
+ liveInstances.add(instance);
+ }
+ }
+
+ updateKylinCluster(liveInstances);
+ }
+ }
+ }
+ });
+ final String hostname = instanceName.substring(0, instanceName.lastIndexOf("_"));
+ final String port = instanceName.substring(instanceName.lastIndexOf("_") + 1);
+ final JobControllerConnector jcc = new JobControllerConnector(hostname, port, zkAddress, kylinConfig.getClusterName(), new DefaultStateModelFactory(instanceName, kylinConfig));
jcc.register();
- helixJobEngineAdmin.rebalance(kylinConfig.getHelixClusterName(), RESOURCE_NAME);
+ helixJobEngineAdmin.rebalance(kylinConfig.getClusterName(), HelixJobEngineAdmin.RESOURCE_NAME);
jcc.start();
- updateKylinConfig(helixJobEngineAdmin.getInstancesInCluster(kylinConfig.getHelixClusterName()));
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
public void run() {
jcc.stop();
+ helixJobEngineAdmin.stopController(kylinConfig.getClusterName());
}
}));
@@ -212,7 +227,7 @@ public class JobController extends BasicController implements InitializingBean {
this.jobService = jobService;
}
- private void updateKylinConfig(List<String> instances) {
+ private void updateKylinCluster(List<String> instances) {
List<String> instanceRestAddresses = Lists.newArrayList();
for (String instanceName : instances) {
int indexOfUnderscore = instanceName.lastIndexOf("_");
@@ -221,6 +236,7 @@ public class JobController extends BasicController implements InitializingBean {
String restServersInCluster = Strings.join(instanceRestAddresses, ",");
KylinConfig.getInstanceFromEnv().setProperty(KylinConfig.KYLIN_REST_SERVERS, restServersInCluster);
System.setProperty(KylinConfig.KYLIN_REST_SERVERS, restServersInCluster);
+ Broadcaster.clearCache();
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/6bf9053f/server/src/main/java/org/apache/kylin/rest/helix/HelixJobEngineAdmin.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/helix/HelixJobEngineAdmin.java b/server/src/main/java/org/apache/kylin/rest/helix/HelixJobEngineAdmin.java
index 7ab2b23..5fed0ce 100644
--- a/server/src/main/java/org/apache/kylin/rest/helix/HelixJobEngineAdmin.java
+++ b/server/src/main/java/org/apache/kylin/rest/helix/HelixJobEngineAdmin.java
@@ -18,7 +18,11 @@
package org.apache.kylin.rest.helix;
import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.helix.ExternalViewChangeListener;
+import org.apache.helix.HelixManager;
import org.apache.helix.controller.HelixControllerMain;
import org.apache.helix.manager.zk.ZKHelixAdmin;
import org.apache.helix.model.ExternalView;
@@ -34,8 +38,12 @@ import java.util.concurrent.ConcurrentMap;
/**
*/
public class HelixJobEngineAdmin {
+
+ public static final String RESOURCE_NAME = "job_engine_v1";
+
private static ConcurrentMap<String, HelixJobEngineAdmin> instanceMaps = Maps.newConcurrentMap();
+ private static ConcurrentMap<String, HelixManager> controllerMaps = Maps.newConcurrentMap();
public static HelixJobEngineAdmin getInstance(String zkAddress) {
Preconditions.checkNotNull(zkAddress);
@@ -43,6 +51,17 @@ public class HelixJobEngineAdmin {
return instanceMaps.get(zkAddress);
}
+ public static String getCurrentInstanceName() {
+ final String restAddress = System.getProperty("kylin.rest.address");
+ if (StringUtils.isEmpty(restAddress)) {
+ throw new RuntimeException("There is no -Dkylin.rest.address set; Please check bin/kylin.sh");
+ }
+
+ final String hostname = Preconditions.checkNotNull(restAddress.substring(0, restAddress.lastIndexOf(":")));
+ final String port = Preconditions.checkNotNull(restAddress.substring(restAddress.lastIndexOf(":") + 1));
+ return hostname + "_" + port;
+ }
+
private static final Logger logger = LoggerFactory.getLogger(HelixJobEngineAdmin.class);
private final String zkAddress;
private final ZKHelixAdmin admin;
@@ -51,7 +70,6 @@ public class HelixJobEngineAdmin {
this.zkAddress = zkAddress;
this.admin = new ZKHelixAdmin(zkAddress);
}
-
public void initV1(String clusterName, String resourceName) {
final StateModelDefinition jobEngineSMDV1 = JobEngineSMDV1.getJobEngineStateModelDefinitionV1();
admin.addCluster(clusterName, false);
@@ -64,9 +82,22 @@ public class HelixJobEngineAdmin {
}
- public void startControllers(String clusterName) {
- HelixControllerMain.startHelixController(zkAddress, clusterName, "localhost_" + JobControllerConstants.CONTROLLER_PORT,
+ public void startController(String clusterName, ExternalViewChangeListener externalViewChangeListener) throws Exception {
+ HelixManager manager = HelixControllerMain.startHelixController(zkAddress, clusterName, getCurrentInstanceName(),
HelixControllerMain.STANDALONE);
+ controllerMaps.put(clusterName, manager);
+ manager.connect();
+ if (externalViewChangeListener != null) {
+ manager.addExternalViewChangeListener(externalViewChangeListener);
+ }
+ }
+
+ public void stopController(String clusterName) {
+ if (controllerMaps.containsKey(clusterName)) {
+ controllerMaps.remove(clusterName).disconnect();
+ } else {
+ throw new IllegalArgumentException("Didn't find HelixManager with clusterName '" + clusterName + "'.");
+ }
}
public String getInstanceState(String clusterName, String resourceName, String instanceName) {
@@ -126,7 +157,16 @@ public class HelixJobEngineAdmin {
logger.info("cluster:" + clusterName + " instances:" + instancesInCluster);
}
- public List<String> getInstancesInCluster(String clusterName) {
- return admin.getInstancesInCluster(clusterName);
+ public boolean isLeaderRole(String clusterName, String instanceName) {
+ final String instanceState =
+ getInstanceState(clusterName,
+ HelixJobEngineAdmin.RESOURCE_NAME,
+ instanceName);
+ logger.debug("instance state: " + instanceState);
+ if (JobEngineSMDV1.States.LEADER.toString().equalsIgnoreCase(instanceState)) {
+ return true;
+ }
+
+ return false;
}
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/6bf9053f/server/src/main/java/org/apache/kylin/rest/service/CubeService.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/kylin/rest/service/CubeService.java b/server/src/main/java/org/apache/kylin/rest/service/CubeService.java
index 0c50228..c846f48 100644
--- a/server/src/main/java/org/apache/kylin/rest/service/CubeService.java
+++ b/server/src/main/java/org/apache/kylin/rest/service/CubeService.java
@@ -56,8 +56,6 @@ import org.apache.kylin.metadata.realization.RealizationType;
import org.apache.kylin.rest.constant.Constant;
import org.apache.kylin.rest.exception.InternalErrorException;
import org.apache.kylin.rest.helix.HelixJobEngineAdmin;
-import org.apache.kylin.rest.helix.JobControllerConstants;
-import org.apache.kylin.rest.helix.v1.JobEngineSMDV1;
import org.apache.kylin.rest.request.MetricsRequest;
import org.apache.kylin.rest.response.HBaseResponse;
import org.apache.kylin.rest.response.MetricsResponse;
@@ -577,12 +575,10 @@ public class CubeService extends BasicService {
public void updateOnNewSegmentReady(String cubeName) {
logger.debug("on updateOnNewSegmentReady: " + cubeName);
final KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
- final String instanceState = HelixJobEngineAdmin.getInstance(kylinConfig.getZookeeperAddress()).
- getInstanceState(kylinConfig.getHelixClusterName(),
- JobControllerConstants.RESOURCE_NAME,
- JobControllerConstants.INSTANCE_NAME);
- logger.debug("server state: " + instanceState);
- if (JobEngineSMDV1.States.LEADER.toString().equalsIgnoreCase(instanceState)) {
+ HelixJobEngineAdmin jobEngineAdmin = HelixJobEngineAdmin.getInstance(kylinConfig.getZookeeperAddress());
+ boolean isLeaderRole = jobEngineAdmin.isLeaderRole(kylinConfig.getClusterName(), HelixJobEngineAdmin.getCurrentInstanceName());
+ logger.debug("server is leader role ? " + isLeaderRole);
+ if (isLeaderRole == true) {
keepCubeRetention(cubeName);
mergeCubeSegment(cubeName);
}
http://git-wip-us.apache.org/repos/asf/kylin/blob/6bf9053f/server/src/test/java/org/apache/kylin/rest/helix/JobControllerConnectorTest.java
----------------------------------------------------------------------
diff --git a/server/src/test/java/org/apache/kylin/rest/helix/JobControllerConnectorTest.java b/server/src/test/java/org/apache/kylin/rest/helix/JobControllerConnectorTest.java
index 8b11728..a2aaddd 100644
--- a/server/src/test/java/org/apache/kylin/rest/helix/JobControllerConnectorTest.java
+++ b/server/src/test/java/org/apache/kylin/rest/helix/JobControllerConnectorTest.java
@@ -24,7 +24,6 @@ import org.I0Itec.zkclient.ZkServer;
import org.apache.hadoop.fs.FileUtil;
import org.apache.helix.manager.zk.ZKHelixAdmin;
-import static org.apache.kylin.rest.helix.JobControllerConstants.*;
import static org.junit.Assert.assertEquals;
import org.apache.kylin.rest.helix.v1.TestJobEngineStateModelV1;
@@ -51,7 +50,8 @@ public class JobControllerConnectorTest {
@Before
- public void setup() {
+ public void setup() throws Exception {
+ System.setProperty("kylin.rest.address", "localhost:7070");
// start zookeeper on localhost
final File tmpDir = new File("/tmp/helix-quickstart");
FileUtil.fullyDelete(tmpDir);
@@ -68,8 +68,8 @@ public class JobControllerConnectorTest {
zkHelixAdmin.dropCluster(CLUSTER_NAME);
connectors = Lists.newArrayList();
helixJobEngineAdmin = HelixJobEngineAdmin.getInstance(zkAddress);
- helixJobEngineAdmin.initV1(CLUSTER_NAME, RESOURCE_NAME);
- helixJobEngineAdmin.startControllers(CLUSTER_NAME);
+ helixJobEngineAdmin.initV1(CLUSTER_NAME, HelixJobEngineAdmin.RESOURCE_NAME);
+ helixJobEngineAdmin.startController(CLUSTER_NAME, null);
}
@@ -77,22 +77,22 @@ public class JobControllerConnectorTest {
public void test() throws Exception {
JobControllerConnector connector = new JobControllerConnector("localhost", "10000", zkAddress, CLUSTER_NAME, new TestStateModelFactory("localhost", "10000"));
connector.register();
- helixJobEngineAdmin.rebalance(CLUSTER_NAME, RESOURCE_NAME);
+ helixJobEngineAdmin.rebalance(CLUSTER_NAME, HelixJobEngineAdmin.RESOURCE_NAME);
connector.start();
connectors.add(connector);
Thread.sleep(1000);
- System.out.println(helixJobEngineAdmin.collectStateInfo("add 1 nodes", CLUSTER_NAME, RESOURCE_NAME));
+ System.out.println(helixJobEngineAdmin.collectStateInfo("add 1 nodes", CLUSTER_NAME, HelixJobEngineAdmin.RESOURCE_NAME));
assertEquals(1, TestJobEngineStateModelV1.getLeaderCount().get());
assertEquals(0, TestJobEngineStateModelV1.getStandbyCount().get());
assertEquals(0, TestJobEngineStateModelV1.getOfflineCount().get());
connector = new JobControllerConnector("localhost", "10001", zkAddress, CLUSTER_NAME, new TestStateModelFactory("localhost", "10001"));
connector.register();
- helixJobEngineAdmin.rebalance(CLUSTER_NAME, RESOURCE_NAME);
+ helixJobEngineAdmin.rebalance(CLUSTER_NAME, HelixJobEngineAdmin.RESOURCE_NAME);
connector.start();
connectors.add(connector);
Thread.sleep(1000);
- System.out.println(helixJobEngineAdmin.collectStateInfo("add 2 nodes", CLUSTER_NAME, RESOURCE_NAME));
+ System.out.println(helixJobEngineAdmin.collectStateInfo("add 2 nodes", CLUSTER_NAME, HelixJobEngineAdmin.RESOURCE_NAME));
assertEquals(1, TestJobEngineStateModelV1.getLeaderCount().get());
assertEquals(1, TestJobEngineStateModelV1.getStandbyCount().get());
assertEquals(0, TestJobEngineStateModelV1.getOfflineCount().get());