This is an automated email from the ASF dual-hosted git repository. aiceflower pushed a commit to branch release-0.9.4 in repository https://gitbox.apache.org/repos/asf/linkis.git
commit 1a3a83e34e9479c44e9c3ee4f52cfb9780ee2d28 Author: alexkun <[email protected]> AuthorDate: Wed Jun 3 18:32:56 2020 +0800 Provide ContextService HighAvailable module. --- contextservice/cs-highavailable/pom.xml | 77 ++++++++ .../cs/highavailable/AbstractContextHAManager.java | 23 +++ .../linkis/cs/highavailable/ContextHAManager.java | 15 ++ .../cs/highavailable/DefaultContextHAManager.java | 114 ++++++++++++ .../conf/ContextHighAvailableConf.java | 6 + .../cs/highavailable/exception/ErrorCode.java | 22 +++ .../highavailable/ha/BackupInstanceGenerator.java | 11 ++ .../cs/highavailable/ha/ContextHAChecker.java | 15 ++ .../cs/highavailable/ha/ContextHAIDGenerator.java | 11 ++ .../ha/impl/BackupInstanceGeneratorImpl.java | 74 ++++++++ .../ha/impl/ContextHACheckerImpl.java | 135 ++++++++++++++ .../ha/impl/ContextHAIDGeneratorImpl.java | 54 ++++++ .../pluggable/HAContextPersistenceManagerImpl.java | 80 ++++++++ .../highavailable/proxy/MethodInterceptorImpl.java | 204 +++++++++++++++++++++ .../highavailable/test/TestContextHAManager.java | 190 +++++++++++++++++++ .../cs/highavailable/test/haid/TestHAID.java | 40 ++++ .../test/persist/TestPersistence.java | 30 +++ .../src/test/resources/application.yml | 20 ++ .../src/test/resources/linkis.properties | 34 ++++ .../src/test/resources/log4j.properties | 36 ++++ .../cs-highavailable/src/test/resources/log4j2.xml | 38 ++++ .../linkis/rpc/conf/RPCConfiguration.scala | 3 + .../rpc/instancealias/InstanceAliasConverter.scala | 14 ++ .../rpc/instancealias/InstanceAliasManager.scala | 24 +++ .../impl/DefaultInstanceAliasConverter.scala | 35 ++++ .../impl/InstanceAliasManagerImpl.scala | 92 ++++++++++ .../wedatasphere/linkis/rpc/utils/RPCUtils.scala | 9 + 27 files changed, 1406 insertions(+) diff --git a/contextservice/cs-highavailable/pom.xml b/contextservice/cs-highavailable/pom.xml new file mode 100644 index 0000000000..fe5cdcf8e4 --- /dev/null +++ b/contextservice/cs-highavailable/pom.xml @@ -0,0 +1,77 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ Copyright 2019 WeBank + ~ Licensed 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. + --> + +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>linkis</artifactId> + <groupId>com.webank.wedatasphere.linkis</groupId> + <version>0.9.4</version> + <relativePath>../../pom.xml</relativePath> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>linkis-cs-highavailable</artifactId> + + <dependencies> + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cs-common</artifactId> + <version>${linkis.version}</version> + </dependency> + + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cloudRPC</artifactId> + <version>${linkis.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cs-persistence</artifactId> + <version>${linkis.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.commons</groupId> + <artifactId>commons-math3</artifactId> + <version>3.1.1</version> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-deploy-plugin</artifactId> + </plugin> + + <plugin> + <groupId>net.alchim31.maven</groupId> + <artifactId>scala-maven-plugin</artifactId> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-jar-plugin</artifactId> + </plugin> + </plugins> + <resources> + <resource> + <directory>${basedir}/src/main/resources</directory> + </resource> + </resources> + <finalName>${project.artifactId}-${project.version}</finalName> + </build> +</project> \ No newline at end of file diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/AbstractContextHAManager.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/AbstractContextHAManager.java new file mode 100644 index 0000000000..f4be9f980a --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/AbstractContextHAManager.java @@ -0,0 +1,23 @@ +package com.webank.wedatasphere.linkis.cs.highavailable; + +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.BackupInstanceGenerator; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.ContextHAChecker; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.ContextHAIDGenerator; + +/** + * + * @Author alexyang + * @Date 2020/2/16 + */ +public abstract class AbstractContextHAManager implements ContextHAManager { + + public abstract ContextHAIDGenerator getContextHAIDGenerator(); + + public abstract ContextHAChecker getContextHAChecker(); + + public abstract BackupInstanceGenerator getBackupInstanceGenerator(); + + public abstract HAContextID convertProxyHAID(HAContextID contextID) throws CSErrorException; +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ContextHAManager.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ContextHAManager.java new file mode 100644 index 0000000000..bb440b254d --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ContextHAManager.java @@ -0,0 +1,15 @@ +package com.webank.wedatasphere.linkis.cs.highavailable; + +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; + +/** + * 动态代理实现持久层HAIDKey和contextID的动态转换 + * + * @Author alexyang + * @Date 2020/2/16 + */ +public interface ContextHAManager { + + <T> T getContextHAProxy(T persistence) throws CSErrorException; + +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/DefaultContextHAManager.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/DefaultContextHAManager.java new file mode 100644 index 0000000000..79acbd7001 --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/DefaultContextHAManager.java @@ -0,0 +1,114 @@ +package com.webank.wedatasphere.linkis.cs.highavailable; + +import com.google.gson.Gson; +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.highavailable.exception.ErrorCode; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.BackupInstanceGenerator; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.ContextHAChecker; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.ContextHAIDGenerator; +import com.webank.wedatasphere.linkis.cs.highavailable.proxy.MethodInterceptorImpl; +import net.sf.cglib.proxy.Callback; +import net.sf.cglib.proxy.Enhancer; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +/** + * ContextService高可用管理器默认实现 + * 采用CGLib动态代理，一般用于CS持久层存储转换，将HAContextID实例进行转换 + * @Author alexyang + * @Date 2020/2/19 + */ +@Component +public class DefaultContextHAManager extends AbstractContextHAManager { + + private static final Logger logger = LoggerFactory.getLogger(DefaultContextHAManager.class); + private static final Gson gson = new Gson(); + + @Autowired + private ContextHAIDGenerator contextHAIDGenerator; + @Autowired + private ContextHAChecker contextHAChecker; + @Autowired + private BackupInstanceGenerator backupInstanceGenerator; + + public DefaultContextHAManager() {} + + + @Override + public <T> T getContextHAProxy(T persistence) throws CSErrorException { + Callback callback = new MethodInterceptorImpl(this, persistence); + Enhancer enhancer = new Enhancer(); + enhancer.setSuperclass(persistence.getClass()); + Callback [] callbacks = new Callback[] {callback}; + enhancer.setCallbacks(callbacks); + return (T)enhancer.create(); + } + + @Override + public HAContextID convertProxyHAID(HAContextID haContextID) throws CSErrorException { + + if (null == haContextID) { + logger.error("HaContextID cannot be null."); + throw new CSErrorException(ErrorCode.INVALID_HAID, "HaContextID cannot be null."); + } + if (StringUtils.isBlank(haContextID.getContextId())) { + // generate new haid + HAContextID tmpHAID = contextHAIDGenerator.generateHAContextID(null); + haContextID.setContextId(tmpHAID.getContextId()); + haContextID.setInstance(tmpHAID.getInstance()); + haContextID.setBackupInstance(tmpHAID.getBackupInstance()); + return haContextID; + } else if (StringUtils.isNotBlank(haContextID.getInstance()) && StringUtils.isNotBlank(haContextID.getBackupInstance())) { + if (StringUtils.isNumeric(haContextID.getContextId())) { + // convert contextID to haID + String haIdKey = contextHAChecker.convertHAIDToHAKey(haContextID); + haContextID.setContextId(haIdKey); + } else if (contextHAChecker.isHAIDValid(haContextID.getContextId())) { + String contextID = contextHAChecker.parseHAIDFromKey(haContextID.getContextId()).getContextId(); + haContextID.setContextId(contextID); + } else { + throw new CSErrorException(ErrorCode.INVALID_HAID, "Invalid contextID in haContextID : " + gson.toJson(haContextID)); + } + return haContextID; + } else { + // complete ha property + if (StringUtils.isNumeric(haContextID.getContextId())) { + HAContextID tmpHAID = contextHAIDGenerator.generateHAContextID(haContextID); + haContextID.setInstance(tmpHAID.getInstance()); + haContextID.setBackupInstance(tmpHAID.getBackupInstance()); + } else if (contextHAChecker.isHAIDValid(haContextID.getContextId())) { + HAContextID tmpHAID = contextHAChecker.parseHAIDFromKey(haContextID.getContextId()); + haContextID.setContextId(tmpHAID.getContextId()); + haContextID.setInstance(tmpHAID.getInstance()); + haContextID.setBackupInstance(tmpHAID.getBackupInstance()); + } else { + throw new CSErrorException(ErrorCode.INVALID_HAID, "Invalid contextID in haContextID : " + gson.toJson(haContextID)); + } + // todo debug + if (contextHAChecker.isHAContextIDValid(haContextID)) { + logger.info("HAID : " + contextHAChecker.convertHAIDToHAKey(haContextID)); + } + return haContextID; + } + } + + @Override + public ContextHAIDGenerator getContextHAIDGenerator() { + return contextHAIDGenerator; + } + + @Override + public ContextHAChecker getContextHAChecker() { + return contextHAChecker; + } + + @Override + public BackupInstanceGenerator getBackupInstanceGenerator() { + return backupInstanceGenerator; + } + +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/conf/ContextHighAvailableConf.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/conf/ContextHighAvailableConf.java new file mode 100644 index 0000000000..82b16e0e72 --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/conf/ContextHighAvailableConf.java @@ -0,0 +1,6 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.conf; + +public class ContextHighAvailableConf { + + +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/exception/ErrorCode.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/exception/ErrorCode.java new file mode 100644 index 0000000000..2f16c6bbe9 --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/exception/ErrorCode.java @@ -0,0 +1,22 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.exception; + +/** + * @Author alexyang + * @Date 2020/2/18 + */ +public class ErrorCode { + + public static int INVALID_INSTANCE_ALIAS = 70010; + + public static int INVALID_HAID = 70011; + + public static int GENERATE_HAID_ERROR = 70012; + + public static int INVALID_CONTEXTID = 70013; + + public static int GENERATE_BACKUP_INSTANCE_ERROR = 70014; + + public static int INVALID_INSTANCE = 70015; + + public static int INVAID_HA_CONTEXTID = 70016; +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/BackupInstanceGenerator.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/BackupInstanceGenerator.java new file mode 100644 index 0000000000..926ef3918b --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/BackupInstanceGenerator.java @@ -0,0 +1,11 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.ha; + +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; + +public interface BackupInstanceGenerator { + + String getBackupInstance(String haID) throws CSErrorException; + + String chooseBackupInstance(String mainInstance) throws CSErrorException; + +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/ContextHAChecker.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/ContextHAChecker.java new file mode 100644 index 0000000000..fc94c6e29b --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/ContextHAChecker.java @@ -0,0 +1,15 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.ha; + +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; + +public interface ContextHAChecker { + + boolean isHAIDValid(String haIDKey); + + boolean isHAContextIDValid(HAContextID haContextID) throws CSErrorException; + + String convertHAIDToHAKey(HAContextID haContextID) throws CSErrorException; + + HAContextID parseHAIDFromKey(String haIDKey) throws CSErrorException; +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/ContextHAIDGenerator.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/ContextHAIDGenerator.java new file mode 100644 index 0000000000..9a47f88ad4 --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/ContextHAIDGenerator.java @@ -0,0 +1,11 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.ha; + +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; + +public interface ContextHAIDGenerator { + + HAContextID generateHAContextID(ContextID contextID) throws CSErrorException; + +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/BackupInstanceGeneratorImpl.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/BackupInstanceGeneratorImpl.java new file mode 100644 index 0000000000..10649b3c5a --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/BackupInstanceGeneratorImpl.java @@ -0,0 +1,74 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.ha.impl; + +import com.webank.wedatasphere.linkis.common.ServiceInstance; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.highavailable.exception.ErrorCode; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.BackupInstanceGenerator; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.ContextHAChecker; +import com.webank.wedatasphere.linkis.rpc.instancealias.InstanceAliasManager; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.util.ArrayList; +import java.util.List; +import java.util.Random; + +@Component +public class BackupInstanceGeneratorImpl implements BackupInstanceGenerator { + private final static Logger logger = LoggerFactory.getLogger(BackupInstanceGeneratorImpl.class); + + @Autowired + private InstanceAliasManager instanceAliasManager; + + @Autowired + private ContextHAChecker contextHAChecker; + + @Override + public String getBackupInstance(String haIDKey) throws CSErrorException { + + String alias = null; + if (StringUtils.isNotBlank(haIDKey) && contextHAChecker.isHAIDValid(haIDKey)) { + alias = contextHAChecker.parseHAIDFromKey(haIDKey).getBackupInstance(); + } else { + throw new CSErrorException(ErrorCode.INVALID_HAID, "Invalid HAID :" + haIDKey); + } + return alias; + } + + @Override + public String chooseBackupInstance(String mainInstanceAlias) throws CSErrorException { + ServiceInstance mainInstance = null; + try { + mainInstance = instanceAliasManager.getInstanceByAlias(mainInstanceAlias); + } catch (Exception e) { + logger.error("Get Instance error, alias : {}, message : {}", mainInstanceAlias, e.getMessage()); + throw new CSErrorException(ErrorCode.INVALID_INSTANCE_ALIAS, e.getMessage() + ", alias : " + mainInstanceAlias); + } + List<ServiceInstance> allInstanceList = instanceAliasManager.getAllInstanceList(); + List<ServiceInstance> remainInstanceList = new ArrayList<>(); + for (ServiceInstance instance : allInstanceList) { + if (instance.equals(mainInstance)) { + continue; + } + remainInstanceList.add(instance); + } + if (remainInstanceList.size() > 0) { + int index = getBackupInstanceIndex(remainInstanceList); + return instanceAliasManager.getAliasByServiceInstance(remainInstanceList.get(index)); + } else { + // only one service instance + logger.error("Only one instance, no remains."); + return instanceAliasManager.getAliasByServiceInstance(mainInstance); + } + } + + private int getBackupInstanceIndex(List<ServiceInstance> instanceList) { + + // todo refactor according to load-balance + return new Random().nextInt(instanceList.size()); + } + +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/ContextHACheckerImpl.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/ContextHACheckerImpl.java new file mode 100644 index 0000000000..431e128bd0 --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/ContextHACheckerImpl.java @@ -0,0 +1,135 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.ha.impl; + +import com.webank.wedatasphere.linkis.DataWorkCloudApplication; +import com.webank.wedatasphere.linkis.common.ServiceInstance; +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.CommonHAContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.common.utils.CSHighAvailableUtils; +import com.webank.wedatasphere.linkis.cs.highavailable.exception.ErrorCode; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.ContextHAChecker; +import com.webank.wedatasphere.linkis.rpc.instancealias.impl.InstanceAliasManagerImpl; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.util.ArrayList; +import java.util.List; + +/** + * @Author alexyang + * @Date 2020/2/19 + */ +@Component +public class ContextHACheckerImpl implements ContextHAChecker { + + private final static Logger logger = LoggerFactory.getLogger(ContextHACheckerImpl.class); + + @Autowired + private InstanceAliasManagerImpl instanceAliasManager; + /** + * ${第一个instance长度}${第二个instance长度}{instance别名1}{instance别名2}{实际ID} + * + * @param haIDKey + * @return + */ + @Override + public boolean isHAIDValid(String haIDKey) { + if (CSHighAvailableUtils.checkHAIDBasicFormat(haIDKey)) { + try { + return checkHAIDInstance(CSHighAvailableUtils.decodeHAID(haIDKey)); + } catch (CSErrorException e) { + return false; + } + } else { + return false; + } + } + + /** + * 主备实例同时有效，且id为有效的HAID或数字时才有效 + * @param haContextID + * @return + * @throws CSErrorException + */ + @Override + public boolean isHAContextIDValid(HAContextID haContextID) throws CSErrorException { + boolean valid = false; + if (null != haContextID && StringUtils.isNotBlank(haContextID.getInstance()) + && StringUtils.isNotBlank(haContextID.getBackupInstance())) { + if (StringUtils.isNotBlank(haContextID.getContextId())) { + if (StringUtils.isNumeric(haContextID.getContextId())) { + valid = checkHAIDInstance(haContextID); + } else { + valid = isHAIDValid(haContextID.getContextId()); + } + } else { + valid = false; + } + } else { + valid = false; + } + return valid; + } + + @Override + public String convertHAIDToHAKey(HAContextID haContextID) throws CSErrorException { + if (null == haContextID || StringUtils.isBlank(haContextID.getInstance()) + || StringUtils.isBlank(haContextID.getBackupInstance()) + || StringUtils.isBlank(haContextID.getContextId())) { + throw new CSErrorException(ErrorCode.INVALID_HAID, "Incomplete HAID Object cannot be encoded. mainInstance : " + + haContextID.getInstance() + ", backupInstance : " + haContextID.getBackupInstance() + ", contextID : " + haContextID.getContextId()); + } + if (StringUtils.isNumeric(haContextID.getContextId())) { + return encode(haContextID); + } else if (isHAIDValid(haContextID.getContextId())) { + return haContextID.getContextId(); + } else { + logger.error("ConvertHAIDToHAKey error, invald HAID : " + haContextID.getContextId()); + throw new CSErrorException(ErrorCode.INVALID_HAID, "ConvertHAIDToHAKey error, invald HAID : " + haContextID.getContextId()); + } + } + + /** + * Encode HAContextID to HAKey String. + * ${第一个instance长度}${第二个instance长度}{instance别名0}{instance别名2}{实际ID} + * @return + */ + private String encode(HAContextID haContextID) throws CSErrorException { + List<String> backupInstanceList = new ArrayList<>(); + if (StringUtils.isNotBlank(haContextID.getBackupInstance())) { + backupInstanceList.add(haContextID.getBackupInstance()); + } else { + backupInstanceList.add(haContextID.getInstance()); + } + return CSHighAvailableUtils.encodeHAIDKey(haContextID.getContextId(), haContextID.getInstance(), backupInstanceList); + } + + private boolean checkHAIDInstance(HAContextID haContextID) { + ServiceInstance serverMainInstance = DataWorkCloudApplication.getServiceInstance(); + String mainInstanceAlias = haContextID.getInstance(); + String backupInstanceAlias = haContextID.getBackupInstance(); + if (!instanceAliasManager.isInstanceAliasValid(mainInstanceAlias) && !instanceAliasManager.isInstanceAliasValid(backupInstanceAlias)) { + return false; + } + ServiceInstance mainInstance = instanceAliasManager.getInstanceByAlias(mainInstanceAlias); + ServiceInstance backupInstance = instanceAliasManager.getInstanceByAlias(backupInstanceAlias); + if (serverMainInstance.equals(mainInstance) || serverMainInstance.equals(backupInstance)) { + return true; + } else { + return false; + } + } + + @Override + public HAContextID parseHAIDFromKey(String haIDKey) throws CSErrorException { + HAContextID haContextID = null; + if (StringUtils.isBlank(haIDKey) || !CSHighAvailableUtils.checkHAIDBasicFormat(haIDKey)) { + logger.error("Invalid haIDKey : " + haIDKey); + throw new CSErrorException(ErrorCode.INVALID_HAID, "Invalid haIDKey : " + haIDKey); + } + return CSHighAvailableUtils.decodeHAID(haIDKey); + } +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/ContextHAIDGeneratorImpl.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/ContextHAIDGeneratorImpl.java new file mode 100644 index 0000000000..f563ef2627 --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/ha/impl/ContextHAIDGeneratorImpl.java @@ -0,0 +1,54 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.ha.impl; + +import com.google.gson.Gson; +import com.webank.wedatasphere.linkis.DataWorkCloudApplication; +import com.webank.wedatasphere.linkis.common.ServiceInstance; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.CommonHAContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.highavailable.exception.ErrorCode; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.BackupInstanceGenerator; +import com.webank.wedatasphere.linkis.cs.highavailable.ha.ContextHAIDGenerator; +import com.webank.wedatasphere.linkis.rpc.instancealias.InstanceAliasConverter; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class ContextHAIDGeneratorImpl implements ContextHAIDGenerator { + + private static final Logger logger = LoggerFactory.getLogger(ContextHAIDGeneratorImpl.class); + + @Autowired + private BackupInstanceGenerator backupInstanceGenerator; + + @Autowired + private InstanceAliasConverter instanceAliasConverter; + + @Override + public HAContextID generateHAContextID(ContextID contextID) throws CSErrorException { + String contextIDKey = null; + if (null != contextID && StringUtils.isNotBlank(contextID.getContextId())) { + contextIDKey = contextID.getContextId(); + } + + ServiceInstance mainInstance = DataWorkCloudApplication.getServiceInstance(); + if (null == mainInstance || StringUtils.isBlank(mainInstance.getInstance())) { + logger.error("MainInstance cannot be null."); + throw new CSErrorException(ErrorCode.INVALID_INSTANCE, "MainInstance backupInstance cannot be null."); + } + String mainInstanceAlias = instanceAliasConverter.instanceToAlias(mainInstance.getInstance()); + String backupInstance = backupInstanceGenerator.chooseBackupInstance(mainInstanceAlias); + if (StringUtils.isBlank(backupInstance)) { + logger.error("Generate backupInstance cannot be null."); + throw new CSErrorException(ErrorCode.GENERATE_BACKUP_INSTANCE_ERROR, "Generate backupInstance cannot be null."); + } + HAContextID haContextID = new CommonHAContextID(mainInstanceAlias, backupInstance, contextIDKey); + logger.info("Generate a haContextID : {}" + new Gson().toJson(haContextID)); + return haContextID; + } + +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/pluggable/HAContextPersistenceManagerImpl.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/pluggable/HAContextPersistenceManagerImpl.java new file mode 100644 index 0000000000..0a6f707f94 --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/pluggable/HAContextPersistenceManagerImpl.java @@ -0,0 +1,80 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.pluggable; + +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.highavailable.AbstractContextHAManager; +import com.webank.wedatasphere.linkis.cs.persistence.ContextPersistenceManager; +import com.webank.wedatasphere.linkis.cs.persistence.persistence.*; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; + +/** + * @Author alexyang + * @Date 2020/2/22 + */ +@Component +public class HAContextPersistenceManagerImpl implements ContextPersistenceManager { + + @Autowired + private ContextIDPersistence contextIDPersistence; + @Autowired + private ContextMapPersistence contextMapPersistence; + + @Autowired + private ContextMetricsPersistence contextMetricsPersistence; + @Autowired + private ContextIDListenerPersistence contextIDListenerPersistence; + @Autowired + private ContextKeyListenerPersistence contextKeyListenerPersistence; + @Autowired + private TransactionManager transactionManager; + + + @Autowired + private AbstractContextHAManager contextHAManager; + + @PostConstruct + void init() throws CSErrorException { + contextIDPersistence = contextHAManager.getContextHAProxy(contextIDPersistence); + contextMapPersistence = contextHAManager.getContextHAProxy(contextMapPersistence); + contextMetricsPersistence = contextHAManager.getContextHAProxy(contextMetricsPersistence); + contextIDListenerPersistence = contextHAManager.getContextHAProxy(contextIDListenerPersistence); + contextKeyListenerPersistence = contextHAManager.getContextHAProxy(contextKeyListenerPersistence); + } + + @Override + public ContextIDPersistence getContextIDPersistence() { + return this.contextIDPersistence; + } + + @Override + public ContextMapPersistence getContextMapPersistence() { + return this.contextMapPersistence; + } + + + + + + @Override + public ContextMetricsPersistence getContextMetricsPersistence() { + return this.contextMetricsPersistence; + } + + @Override + public ContextIDListenerPersistence getContextIDListenerPersistence() { + return this.contextIDListenerPersistence; + } + + @Override + public ContextKeyListenerPersistence getContextKeyListenerPersistence() { + return this.contextKeyListenerPersistence; + } + + @Override + public TransactionManager getTransactionManager() { + return this.transactionManager; + } + +} diff --git a/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/proxy/MethodInterceptorImpl.java b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/proxy/MethodInterceptorImpl.java new file mode 100644 index 0000000000..835189588f --- /dev/null +++ b/contextservice/cs-highavailable/src/main/java/com/webank/wedatasphere/linkis/cs/highavailable/proxy/MethodInterceptorImpl.java @@ -0,0 +1,204 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.proxy; + +import com.google.gson.Gson; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.highavailable.AbstractContextHAManager; +import com.webank.wedatasphere.linkis.cs.highavailable.exception.ErrorCode; +import net.sf.cglib.proxy.MethodInterceptor; +import net.sf.cglib.proxy.MethodProxy; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.reflect.Method; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * 基于CGLib库实现的动态代理拦截器，拦截被代理方法的参数，在被代理方法之前和之后进行增强 + * 被代理方法调用前增强场景较多，一般对参数为HAContextID实例、参数名包含contextid的方法参数进行转换 + * 被代理方法调用后，目前只用于将HAContextID的数字型contextID转换为HAIDKey + * @Author alexyang + * @Date 2020/2/25 + */ +public class MethodInterceptorImpl implements MethodInterceptor { + + private final static Logger logger = LoggerFactory.getLogger(MethodInterceptorImpl.class); + private final static Gson gson = new Gson(); + private AbstractContextHAManager contextHAManager; + private Object object; + private final Map<Integer, String> contextIDCacheMap = new HashMap<Integer, String>(); + + public MethodInterceptorImpl(AbstractContextHAManager contextHAManager, Object object) { + this.contextHAManager = contextHAManager; + this.object = object; + } + + @Override + public Object intercept(Object o, Method method, Object[] args, MethodProxy methodProxy) throws Throwable { + + this.contextIDCacheMap.clear(); + // 1，执行方法前转换 + for (int i = 0; i < args.length; i++) { + + // ①参数含有ContextID实例 + if (ContextID.class.isInstance(args[i])) { + ContextID contextID = (ContextID)args[i]; + convertContextIDBeforeInvoke(contextID, i); + } + + // ②参数含有getContextID方法的 + convertGetContextIDBeforeInvoke(args[i]); + } + + // ③方法名含有ContextID，并且有String类型参数，取第一个转换 + if (method.getName().toLowerCase().contains("contextid")) { + for (int j = 0; j < args.length; j++) { + if (String.class.isInstance(args[j])) { + String pStr = (String)args[j]; + if (StringUtils.isNotBlank(pStr) && !StringUtils.isNumeric(pStr)) { + if (this.contextHAManager.getContextHAChecker().isHAIDValid(pStr)) { + String contextID = this.contextHAManager.getContextHAChecker().parseHAIDFromKey(pStr).getContextId(); + args[j] = contextID; + } else { + logger.error("Invalid HAID : " + pStr + " in method : " + method.getName()); + throw new CSErrorException(ErrorCode.INVALID_HAID, "Invalid HAID : " + pStr + " in method : " + method.getName()); + } + } + break; + } + } + } + + // 2，执行原方法 + Object oriResult = method.invoke(this.object, args); + + // 3，执行方法后处理 + // （1）返回值处理 + if (null != oriResult) { + // ①判断集合中元素是否有getContextID方法 + if (List.class.isInstance(oriResult)) { + List objList = (List)oriResult; + for (Object oneParameter : objList) { + convertGetContextIDAfterInvoke(oneParameter); + } + // ②判断ContextID + } else if (HAContextID.class.isInstance(oriResult)) { + HAContextID haContextID = (HAContextID)oriResult; + if (StringUtils.isNumeric(haContextID.getContextId())) { + String haId = this.contextHAManager.getContextHAChecker().convertHAIDToHAKey(haContextID); + haContextID.setContextId(haId); + } + } else { + // ③ 返回值方法含有getContextID + convertGetContextIDAfterInvoke(oriResult); + } + } + + // （2）请求参数还原 + // ①参数有ContextID实例 + for (int k = 0; k < args.length; k++) { + if (ContextID.class.isInstance(args[k])) { + if (this.contextIDCacheMap.containsKey(k)) { + ContextID contextID = (ContextID)args[k]; + contextID.setContextId(this.contextIDCacheMap.get(k)); + } else { + if (HAContextID.class.isInstance(args[k])) { + HAContextID haContextID = (HAContextID)args[k]; + if (StringUtils.isNumeric(haContextID.getContextId())) { + if (StringUtils.isNotBlank(haContextID.getInstance()) && StringUtils.isNotBlank(haContextID.getBackupInstance())) { + String haId = this.contextHAManager.getContextHAChecker().convertHAIDToHAKey(haContextID); + haContextID.setContextId(haId); + } else { + logger.error("Invalid HAContextID : " + gson.toJson(haContextID)); + throw new CSErrorException(ErrorCode.INVAID_HA_CONTEXTID, "Invalid HAContextID : " + gson.toJson(haContextID)); + } + } + } + } + } else { + // ②参数含有getContextID方法 + convertGetContextIDAfterInvoke(args[k]); + } + } + // ③方法名含有ContextID，并且有String类型参数 引用不需要作处理 + + return oriResult; + } + + + private void convertContextIDBeforeInvoke(ContextID contextID, int index) throws CSErrorException { + if (null == contextID) { + return; + } + if (StringUtils.isNumeric(contextID.getContextId())) { + if (HAContextID.class.isInstance(contextID)) { + logger.error("ContextId of HAContextID instance cannot be numberic. contextId : " + gson.toJson(contextID)); + throw new CSErrorException(ErrorCode.INVALID_CONTEXTID, "ContextId of HAContextID instance cannot be numberic. contextId : " + gson.toJson(contextID)); + } + } else { + if (HAContextID.class.isInstance(contextID)) { + if (null == contextID.getContextId()) { + this.contextHAManager.convertProxyHAID((HAContextID) contextID); + } else if (this.contextHAManager.getContextHAChecker().isHAIDValid(contextID.getContextId())) { + if (index > 0) { + this.contextIDCacheMap.put(index, contextID.getContextId()); + } + this.contextHAManager.convertProxyHAID((HAContextID) contextID); + } else { + logger.error("Invalid haContextId. contextId : " + gson.toJson(contextID)); + throw new CSErrorException(ErrorCode.INVALID_HAID, "Invalid haContextId. contextId : " + gson.toJson(contextID)); + } + } + } + } + + private void convertGetContextIDBeforeInvoke(Object object) throws CSErrorException { + for (Method innerMethod : object.getClass().getMethods()) { + if (innerMethod.getName().toLowerCase().contains("getcontextid")) { + try { + Object result = innerMethod.invoke(object); + if (null != object && ContextID.class.isInstance(result)) { + convertContextIDBeforeInvoke((ContextID)result, -1); + } else { + logger.warn("Method {} returns non-contextid object : {}", innerMethod.getName(), gson.toJson(object)); + } + } catch (Exception e) { + logger.error("call method : {} error, ", innerMethod.getName(), e); + } + } + } + } + + private void convertGetContextIDAfterInvoke(Object object) throws CSErrorException { + for (Method innerMethod : object.getClass().getMethods()) { + convertGetContextIDAfterInvokeMethod(innerMethod); + } + } + + private void convertGetContextIDAfterInvokeMethod(Method method) throws CSErrorException { + if (method.getName().toLowerCase().contains("getcontextid")) { + Object result = null; + try { + result = method.invoke(object); + } catch (Exception e) { + logger.warn("Invoke method : {} error. ", method.getName(), e); + } + if (null != result && HAContextID.class.isInstance(result)) { + HAContextID haContextID = (HAContextID)result; + if (StringUtils.isNumeric(haContextID.getContextId()) + && StringUtils.isNotBlank(haContextID.getInstance()) + && StringUtils.isNotBlank(haContextID.getBackupInstance())) { + String haid = this.contextHAManager.getContextHAChecker().convertHAIDToHAKey(haContextID); + haContextID.setContextId(haid); + } else { + logger.error("GetContextID method : " + method.getName() + " returns invalid haContextID : " + gson.toJson(result)); + throw new CSErrorException(ErrorCode.INVALID_HAID, "GetContextID method : " + method.getName() + " returns invalid haContextID : " + gson.toJson(result)); + } + } + } + } +} diff --git a/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/TestContextHAManager.java b/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/TestContextHAManager.java new file mode 100644 index 0000000000..a6bbd16e8f --- /dev/null +++ b/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/TestContextHAManager.java @@ -0,0 +1,190 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.test; + +import com.google.gson.Gson; +import com.webank.wedatasphere.linkis.DataWorkCloudApplication; +import com.webank.wedatasphere.linkis.common.ServiceInstance; +import com.webank.wedatasphere.linkis.common.conf.BDPConfiguration; +import com.webank.wedatasphere.linkis.common.conf.Configuration; +import com.webank.wedatasphere.linkis.common.exception.DWCException; +import com.webank.wedatasphere.linkis.common.utils.Utils; +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.highavailable.AbstractContextHAManager; +import com.webank.wedatasphere.linkis.cs.highavailable.test.haid.TestHAID; +import com.webank.wedatasphere.linkis.cs.highavailable.test.persist.TestPersistence; +import com.webank.wedatasphere.linkis.server.BDPJettyServerHelper; +import com.webank.wedatasphere.linkis.server.conf.ServerConfiguration; +import org.apache.commons.lang.StringUtils; +import org.eclipse.jetty.server.Handler; +import org.eclipse.jetty.server.Server; +import org.eclipse.jetty.servlet.FilterHolder; +import org.eclipse.jetty.webapp.WebAppContext; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.boot.context.event.ApplicationPreparedEvent; +import org.springframework.boot.web.embedded.jetty.JettyServerCustomizer; +import org.springframework.boot.web.embedded.jetty.JettyServletWebServerFactory; +import org.springframework.boot.web.server.WebServerFactoryCustomizer; +import org.springframework.boot.web.servlet.support.SpringBootServletInitializer; +import org.springframework.cloud.client.discovery.EnableDiscoveryClient; +import org.springframework.cloud.context.scope.refresh.RefreshScopeRefreshedEvent; +import org.springframework.context.ApplicationListener; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.core.env.CompositePropertySource; +import org.springframework.core.env.Environment; +import org.springframework.core.env.PropertySource; +import org.springframework.core.env.StandardEnvironment; +import org.springframework.web.filter.CharacterEncodingFilter; + +import javax.servlet.DispatcherType; +import java.util.EnumSet; + +/** + * @Author alexyang + * @Date 2020/2/22 + */ +@SpringBootApplication +@EnableDiscoveryClient +@ComponentScan(basePackages = "com.webank.wedatasphere.linkis") +public class TestContextHAManager extends SpringBootServletInitializer { + + private static ConfigurableApplicationContext applicationContext; + private static ServiceInstance serviceInstance; + private static final Gson gson = new Gson(); + + public static void main(String [] args) throws ReflectiveOperationException { + + final SpringApplication application = new SpringApplication(TestContextHAManager.class); + application.addListeners(new ApplicationListener<ApplicationPreparedEvent>(){ + public void onApplicationEvent(ApplicationPreparedEvent applicationPreparedEvent) { + System.out.println("add config from config server..."); + if(applicationContext == null) { + applicationContext = applicationPreparedEvent.getApplicationContext(); + } + System.out.println("initialize DataWorkCloud spring application..."); + initDWCApplication(); + + } + }); + application.addListeners(new ApplicationListener<RefreshScopeRefreshedEvent>() { + public void onApplicationEvent(RefreshScopeRefreshedEvent applicationEvent) { + System.out.println("refresh config from config server..."); + updateRemoteConfig(); + } + }); + String listeners = ServerConfiguration.BDP_SERVER_SPRING_APPLICATION_LISTENERS().getValue(); + if(StringUtils.isNotBlank(listeners)) { + for (String listener : listeners.split(",")) { + application.addListeners((ApplicationListener<?>) Class.forName(listener).newInstance()); + } + } + applicationContext = application.run(args); + + try { +// Thread.sleep(3000l); + AbstractContextHAManager haManager = (AbstractContextHAManager) applicationContext.getBean(AbstractContextHAManager.class); + if (null == haManager) { + System.err.println("Null haManager!"); + return ; + } + testHAManager(haManager); + } catch (Exception e) { + e.printStackTrace(); + } + + } + + private static void initDWCApplication() { + serviceInstance = new ServiceInstance(); + serviceInstance.setApplicationName(applicationContext.getEnvironment().getProperty("spring.application.name")); + serviceInstance.setInstance(Utils.getComputerName() + ":" + applicationContext.getEnvironment().getProperty("server.port")); + DWCException.setApplicationName(serviceInstance.getApplicationName()); + DWCException.setHostname(Utils.getComputerName()); + DWCException.setHostPort(Integer.parseInt(applicationContext.getEnvironment().getProperty("server.port"))); + } + + public static void updateRemoteConfig() { + addOrUpdateRemoteConfig(applicationContext.getEnvironment(), true); + } + + public static void addRemoteConfig() { + addOrUpdateRemoteConfig(applicationContext.getEnvironment(), false); + } + + private static void addOrUpdateRemoteConfig(Environment env, boolean isUpdateOrNot) { + StandardEnvironment environment = (StandardEnvironment) env; + PropertySource propertySource = environment.getPropertySources().get("bootstrapProperties"); + if(propertySource == null) { + return; + } + CompositePropertySource source = (CompositePropertySource) propertySource; + for (String key: source.getPropertyNames()) { + Object val = source.getProperty(key); + if(val == null) { + continue; + } + if(isUpdateOrNot) { + System.out.println("update remote config => " + key + " = " + source.getProperty(key)); + BDPConfiguration.set(key, val.toString()); + } else { + System.out.println("add remote config => " + key + " = " + source.getProperty(key)); + BDPConfiguration.setIfNotExists(key, val.toString()); + } + } + } + + @Override + protected SpringApplicationBuilder configure(SpringApplicationBuilder builder) { + return builder.sources(DataWorkCloudApplication.class); + } + + @Bean + public WebServerFactoryCustomizer<JettyServletWebServerFactory> jettyFactoryCustomizer() { + return new WebServerFactoryCustomizer<JettyServletWebServerFactory>() { + public void customize(JettyServletWebServerFactory jettyServletWebServerFactory) { + jettyServletWebServerFactory.addServerCustomizers(new JettyServerCustomizer() { + public void customize(Server server) { + Handler[] childHandlersByClass = server.getChildHandlersByClass(WebAppContext.class); + final WebAppContext webApp = (WebAppContext) childHandlersByClass[0]; + FilterHolder filterHolder = new FilterHolder(CharacterEncodingFilter.class); + filterHolder.setInitParameter("encoding", Configuration.BDP_ENCODING().getValue()); + filterHolder.setInitParameter("forceEncoding", "true"); + webApp.addFilter(filterHolder, "/*", EnumSet.allOf(DispatcherType.class)); + BDPJettyServerHelper.setupRestApiContextHandler(webApp); + if(ServerConfiguration.BDP_SERVER_SOCKET_MODE().getValue()) { + BDPJettyServerHelper.setupControllerServer(webApp); + } + if(!ServerConfiguration.BDP_SERVER_DISTINCT_MODE().getValue()) { + BDPJettyServerHelper.setupWebAppContext(webApp); + } + } + }); + } + }; + } + + + + // test + private static void testHAManager(AbstractContextHAManager contextHAManager) { + // 1 test create + TestHAID haid = new TestHAID(); + try { + TestPersistence testPersistence = contextHAManager.getContextHAProxy(new TestPersistence()); + HAContextID haContextID = testPersistence.createHAID(haid); + testPersistence.passHAID(haContextID); + testPersistence.setContextId(haContextID.getContextId()); + } catch (CSErrorException e) { + e.printStackTrace(); + } + System.out.println("Test HaManager End."); + } + + public static ServiceInstance getServiceInstance() { + return serviceInstance; + } + +} diff --git a/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/haid/TestHAID.java b/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/haid/TestHAID.java new file mode 100644 index 0000000000..bf78bd5763 --- /dev/null +++ b/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/haid/TestHAID.java @@ -0,0 +1,40 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.test.haid; + +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; + +public class TestHAID implements HAContextID { + + private String contextId; + private String instance; + private String backupInstance; + + @Override + public String getContextId() { + return contextId; + } + + @Override + public void setContextId(String contextId) { + this.contextId = contextId; + } + + @Override + public String getInstance() { + return instance; + } + + @Override + public void setInstance(String instance) { + this.instance = instance; + } + + @Override + public String getBackupInstance() { + return backupInstance; + } + + @Override + public void setBackupInstance(String backupInstance) { + this.backupInstance = backupInstance; + } +} diff --git a/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/persist/TestPersistence.java b/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/persist/TestPersistence.java new file mode 100644 index 0000000000..29e11e2c2a --- /dev/null +++ b/contextservice/cs-highavailable/src/test/java/com/webank/wedatasphere/linkis/cs/highavailable/test/persist/TestPersistence.java @@ -0,0 +1,30 @@ +package com.webank.wedatasphere.linkis.cs.highavailable.test.persist; + +import com.google.gson.Gson; +import com.webank.wedatasphere.linkis.cs.common.entity.source.HAContextID; + +public class TestPersistence { + + private Gson gson = new Gson(); + + public void testPrint() { + System.out.println("TestPersistence: testPrint()"); + } + + public HAContextID createHAID(HAContextID haContextID) { + System.out.println("TestPersistence: createHAID(), params: haContextID : " + gson.toJson(haContextID)); + haContextID.setContextId("1"); + return haContextID; + } + + public HAContextID passHAID(HAContextID haContextID) { + System.out.println("TestPersistence: passHAID(), params: haContextID : " + gson.toJson(haContextID)); + return haContextID; + } + + public void setContextId(String haid) { + System.out.println("TestPersistence: setContextId(), : " + gson.toJson(haid)); + } + + +} diff --git a/contextservice/cs-highavailable/src/test/resources/application.yml b/contextservice/cs-highavailable/src/test/resources/application.yml new file mode 100644 index 0000000000..1062fb0d7a --- /dev/null +++ b/contextservice/cs-highavailable/src/test/resources/application.yml @@ -0,0 +1,20 @@ +server: + port: 9010 +spring: + application: + name: CLOUD-CONTEXTSERVICE + +eureka: + client: + serviceUrl: + defaultZone: http://127.0.0.1:20303/eureka/ + registry-fetch-interval-seconds: 5 + instance: + metadata-map: + test: wedatasphere + +management: + endpoints: + web: + exposure: + include: refresh,info diff --git a/contextservice/cs-highavailable/src/test/resources/linkis.properties b/contextservice/cs-highavailable/src/test/resources/linkis.properties new file mode 100644 index 0000000000..f4d3720d08 --- /dev/null +++ b/contextservice/cs-highavailable/src/test/resources/linkis.properties @@ -0,0 +1,34 @@ +# +# Copyright 2019 WeBank +# +# Licensed 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. +# + +#wds.linkis.test.mode=true + +wds.linkis.server.version=v1 + +wds.linkis.ldap.proxy.url= +wds.linkis.ldap.proxy.baseDN= + +wds.linkis.server.restful.uri=/ + +wds.linkis.server.web.session.timeout=1h + +wds.linkis.gateway.conf.enable.proxy.user=false + +wds.linkis.gateway.conf.url.pass.auth=/dss/ + +wds.linkis.gateway.admin.user=hadoop + +wds.linkis.gateway.conf.enable.token.auth=true diff --git a/contextservice/cs-highavailable/src/test/resources/log4j.properties b/contextservice/cs-highavailable/src/test/resources/log4j.properties new file mode 100644 index 0000000000..a7e6854c4d --- /dev/null +++ b/contextservice/cs-highavailable/src/test/resources/log4j.properties @@ -0,0 +1,36 @@ +# +# Copyright 2019 WeBank +# +# Licensed 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. +# + +### set log levels ### + +log4j.rootCategory=INFO,console + +log4j.appender.console=org.apache.log4j.ConsoleAppender +log4j.appender.console.Threshold=INFO +log4j.appender.console.layout=org.apache.log4j.PatternLayout +#log4j.appender.console.layout.ConversionPattern= %d{ISO8601} %-5p (%t) [%F:%M(%L)] - %m%n +log4j.appender.console.layout.ConversionPattern= %d{ISO8601} %-5p (%t) %p %c{1} - %m%n + + +log4j.appender.com.webank.bdp.ide.core=org.apache.log4j.DailyRollingFileAppender +log4j.appender.com.webank.bdp.ide.core.Threshold=INFO +log4j.additivity.com.webank.bdp.ide.core=false +log4j.appender.com.webank.bdp.ide.core.layout=org.apache.log4j.PatternLayout +log4j.appender.com.webank.bdp.ide.core.Append=true +log4j.appender.com.webank.bdp.ide.core.File=logs/linkis.log +log4j.appender.com.webank.bdp.ide.core.layout.ConversionPattern= %d{ISO8601} %-5p (%t) [%F:%M(%L)] - %m%n + +log4j.logger.org.springframework=INFO diff --git a/contextservice/cs-highavailable/src/test/resources/log4j2.xml b/contextservice/cs-highavailable/src/test/resources/log4j2.xml new file mode 100644 index 0000000000..ad88ea570e --- /dev/null +++ b/contextservice/cs-highavailable/src/test/resources/log4j2.xml @@ -0,0 +1,38 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ Copyright 2019 WeBank + ~ + ~ Licensed 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. + --> + +<configuration status="error" monitorInterval="30"> + <appenders> + <Console name="Console" target="SYSTEM_OUT"> + <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/> + <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level [%t] %logger{36} %L %M - %msg%xEx%n"/> + </Console> + <RollingFile name="RollingFile" fileName="logs/linkis.log" + filePattern="logs/$${date:yyyy-MM}/linkis-log-%d{yyyy-MM-dd}-%i.log"> + <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%-5level] [%-40t] %c{1.} (%L) [%M] - %msg%xEx%n"/> + <SizeBasedTriggeringPolicy size="100MB"/> + <DefaultRolloverStrategy max="20"/> + </RollingFile> + </appenders> + <loggers> + <root level="INFO"> + <appender-ref ref="RollingFile"/> + <appender-ref ref="Console"/> + </root> + </loggers> +</configuration> + diff --git a/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/conf/RPCConfiguration.scala b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/conf/RPCConfiguration.scala index 8315b3ec9b..e1d067002e 100644 --- a/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/conf/RPCConfiguration.scala +++ b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/conf/RPCConfiguration.scala @@ -39,4 +39,7 @@ object RPCConfiguration { val PUBLIC_SERVICE_APPLICATION_NAME = CommonVars("wds.linkis.gateway.conf.publicservice.name", "publicservice") val PUBLIC_SERVICE_LIST = CommonVars("wds.linkis.gateway.conf.publicservice.list", "query,jobhistory,application,configuration,filesystem,udf,variable").getValue.split(",") + val BDP_RPC_INSTANCE_ALIAS_SERVICE_REFRESH_INTERVAL = CommonVars("wds.linkis.rpc.instancealias.refresh.interval", new TimeType("3s")) + + val CONTEXT_SERVICE_APPLICATION_NAME = CommonVars("wds.linkis.gateway.conf.contextservice.name", "contextservice") } diff --git a/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/InstanceAliasConverter.scala b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/InstanceAliasConverter.scala new file mode 100644 index 0000000000..3b14808727 --- /dev/null +++ b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/InstanceAliasConverter.scala @@ -0,0 +1,14 @@ +package com.webank.wedatasphere.linkis.rpc.instancealias + +/** + * @Author alexyang + * @Date 2020/2/18 + */ +trait InstanceAliasConverter { + + def instanceToAlias(instance: String): String + + def aliasToInstance(alias: String): String + + def checkAliasFormatValid(alias: String): Boolean +} diff --git a/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/InstanceAliasManager.scala b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/InstanceAliasManager.scala new file mode 100644 index 0000000000..3a6b4a771c --- /dev/null +++ b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/InstanceAliasManager.scala @@ -0,0 +1,24 @@ +package com.webank.wedatasphere.linkis.rpc.instancealias + +import com.webank.wedatasphere.linkis.common.ServiceInstance +import javax.annotation.Nullable + +/** + * @Author alexyang + * @Date 2020/2/18 + */ +trait InstanceAliasManager { + + def getAliasByServiceInstance(instance: ServiceInstance): String + + def getAliasByInstance(instance: String): String + + @Nullable + def getInstanceByAlias(alias: String): ServiceInstance + + def refresh(): Unit + + def getAllInstanceList(): java.util.List[ServiceInstance] + + def isInstanceAliasValid(alias: String): Boolean +} diff --git a/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/impl/DefaultInstanceAliasConverter.scala b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/impl/DefaultInstanceAliasConverter.scala new file mode 100644 index 0000000000..2957cdfdda --- /dev/null +++ b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/impl/DefaultInstanceAliasConverter.scala @@ -0,0 +1,35 @@ +package com.webank.wedatasphere.linkis.rpc.instancealias.impl + +import java.util.Base64 +import java.util.regex.Pattern + +import com.webank.wedatasphere.linkis.rpc.instancealias.InstanceAliasConverter +import org.apache.commons.lang.StringUtils +import org.springframework.stereotype.Component + +/** + * @Author alexyang + * @Date 2020/2/18 + */ +@Component +class DefaultInstanceAliasConverter extends InstanceAliasConverter { + + val pattern = Pattern.compile("[a-zA-Z\\d=\\+/]+") + + // todo use base64 for the moment + override def instanceToAlias(instance: String): String = { + new String(Base64.getEncoder.encode(instance.getBytes())) + } + + override def aliasToInstance(alias: String): String = { + new String(Base64.getDecoder.decode(alias)) + } + + override def checkAliasFormatValid(alias: String): Boolean = { + if (StringUtils.isBlank(alias)) { + return false + } + val matcher = pattern.matcher(alias) + matcher.find() + } +} diff --git a/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/impl/InstanceAliasManagerImpl.scala b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/impl/InstanceAliasManagerImpl.scala new file mode 100644 index 0000000000..353f6a4c07 --- /dev/null +++ b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/instancealias/impl/InstanceAliasManagerImpl.scala @@ -0,0 +1,92 @@ +package com.webank.wedatasphere.linkis.rpc.instancealias.impl + +import java.util + +import com.webank.wedatasphere.linkis.DataWorkCloudApplication +import com.webank.wedatasphere.linkis.common.ServiceInstance +import com.webank.wedatasphere.linkis.common.utils.Logging +import com.webank.wedatasphere.linkis.rpc.conf.RPCConfiguration +import com.webank.wedatasphere.linkis.rpc.instancealias.{InstanceAliasConverter, InstanceAliasManager} +import com.webank.wedatasphere.linkis.rpc.sender.eureka.EurekaRPCServerLoader +import com.webank.wedatasphere.linkis.rpc.utils.RPCUtils +import org.springframework.beans.factory.annotation.Autowired +import org.springframework.stereotype.Component + +import scala.collection.JavaConversions._ + +/** + * @Author alexyang + * @Date 2020/2/18 + */ +@Component +class InstanceAliasManagerImpl extends InstanceAliasManager with Logging { + + private val serverLoader = new EurekaRPCServerLoader() + + private val mainInstance = DataWorkCloudApplication.getServiceInstance + + @Autowired + var instanceAliasConverter: InstanceAliasConverter = _ + + override def getAliasByInstance(instance: String): String = { + instanceAliasConverter.instanceToAlias(instance) + } + + override def getInstanceByAlias(alias: String): ServiceInstance = { + val serviceID = getContextServiceID() + if (null == serviceID) { + return null + } + val instances = serverLoader.getServiceInstances(serviceID) + if (null == instances || instances.isEmpty) { + error(s"None serviec instances for Context Service ID : " + serviceID) + return null + } + val targetInstance = instanceAliasConverter.aliasToInstance(alias) + instances.foreach(ins => { + if (ins.getInstance.equalsIgnoreCase(targetInstance)) { + return ins + } + }) + null + } + + def getContextServiceID(): String = { + RPCUtils.findService (RPCConfiguration.CONTEXT_SERVICE_APPLICATION_NAME.getValue, list => { + val services = list.filter (_.contains (RPCConfiguration.CONTEXT_SERVICE_APPLICATION_NAME.getValue) ) + services.headOption + }).getOrElse(null) + } + + @Deprecated + override def refresh(): Unit = { + + } + + override def getAllInstanceList(): util.List[ServiceInstance] = { + val serviceID = getContextServiceID() + if (null == serviceID) { + return new util.ArrayList[ServiceInstance](0) + } + serverLoader.getServiceInstances(serviceID).toList + } + + override def isInstanceAliasValid(alias: String): Boolean = { + if (!instanceAliasConverter.checkAliasFormatValid(alias)) { + return false + } + if (null != getInstanceByAlias(alias)) { + true + } else { + false + } + } + + override def getAliasByServiceInstance(instance: ServiceInstance): String = { + if (null == instance) { + return null + } + instanceAliasConverter.instanceToAlias(instance.getInstance) + } +} + diff --git a/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/utils/RPCUtils.scala b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/utils/RPCUtils.scala index cbaa838186..cd50461060 100644 --- a/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/utils/RPCUtils.scala +++ b/core/cloudRPC/src/main/scala/com/webank/wedatasphere/linkis/rpc/utils/RPCUtils.scala @@ -21,8 +21,10 @@ import java.net.ConnectException import com.netflix.client.ClientException import com.webank.wedatasphere.linkis.rpc.exception.NoInstanceExistsException +import com.webank.wedatasphere.linkis.rpc.sender.SpringCloudFeignConfigurationCache import feign.RetryableException import org.apache.commons.lang.StringUtils +import scala.collection.JavaConversions._ /** * Created by enjoyyin on 2019/2/22. @@ -50,4 +52,11 @@ object RPCUtils { case _ => false } + def findService(parsedServiceId: String, tooManyDeal: List[String] => Option[String]): Option[String] = { + val services = SpringCloudFeignConfigurationCache.getDiscoveryClient + .getServices.filter(_.toLowerCase.contains(parsedServiceId.toLowerCase)).toList + if(services.length == 1) Some(services.head) + else if(services.length > 1) tooManyDeal(services) + else None + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
