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 b092fa938af5e4e727d9e67a47865dd2397c3b9f Author: jftang <[email protected]> AuthorDate: Wed Jun 3 22:53:40 2020 +0800 add cs-server module --- contextservice/cs-server/bin/start-cs-server.sh | 38 ++++ contextservice/cs-server/bin/stop--cs-server.sh | 47 +++++ contextservice/cs-server/conf/application.yml | 28 +++ contextservice/cs-server/conf/linkis.properties | 22 +++ contextservice/cs-server/conf/log4j.properties | 26 +++ contextservice/cs-server/conf/log4j2.xml | 35 ++++ contextservice/cs-server/pom.xml | 183 +++++++++++++++++ .../cs-server/src/main/assembly/distribution.xml | 71 +++++++ .../linkis/cs/server/conf/ContextServerConf.java | 12 ++ .../cs/server/enumeration/ServiceMethod.java | 11 ++ .../linkis/cs/server/enumeration/ServiceType.java | 35 ++++ .../cs/server/parser/DefaultKeywordParser.java | 124 ++++++++++++ .../cs/server/parser/KeywordMethodEntity.java | 55 ++++++ .../linkis/cs/server/parser/KeywordParser.java | 11 ++ .../protocol/AbstractHttpRequestProtocol.java | 45 +++++ .../cs/server/protocol/ContextIDProtocol.java | 13 ++ .../server/protocol/ContextListenerProtocol.java | 14 ++ .../linkis/cs/server/protocol/ContextProtocol.java | 14 ++ .../linkis/cs/server/protocol/HttpProtocol.java | 9 + .../cs/server/protocol/HttpRequestProtocol.java | 24 +++ .../cs/server/protocol/HttpResponseProtocol.java | 21 ++ .../cs/server/protocol/RestResponseProtocol.java | 86 ++++++++ .../cs/server/restful/ContextIDRestfulApi.java | 99 ++++++++++ .../server/restful/ContextListenerRestfulApi.java | 83 ++++++++ .../cs/server/restful/ContextRestfulApi.java | 156 +++++++++++++++ .../linkis/cs/server/restful/CsRestfulParent.java | 105 ++++++++++ .../linkis/cs/server/scheduler/CsScheduler.java | 15 ++ .../cs/server/scheduler/DefaultCsScheduler.java | 83 ++++++++ .../linkis/cs/server/scheduler/HttpAnswerJob.java | 14 ++ .../cs/server/scheduler/HttpAnswerJobBuilder.java | 50 +++++ .../linkis/cs/server/scheduler/HttpJob.java | 14 ++ .../linkis/cs/server/scheduler/HttpJobBuilder.java | 21 ++ .../cs/server/scheduler/HttpPriorityJob.java | 10 + .../linkis/cs/server/scheduler/RestJobBuilder.java | 17 ++ .../scheduler/linkisImpl/CsExecuteRequest.java | 43 ++++ .../cs/server/scheduler/linkisImpl/CsExecutor.java | 49 +++++ .../scheduler/linkisImpl/CsExecutorManager.java | 55 ++++++ .../server/scheduler/linkisImpl/CsJobListener.java | 51 +++++ .../scheduler/linkisImpl/CsSchedulerBean.java | 36 ++++ .../scheduler/linkisImpl/CsSchedulerJob.java | 67 +++++++ .../linkisImpl/JobToExecuteRequestConsumer.java | 17 ++ .../linkis/cs/server/service/AbstractService.java | 82 ++++++++ .../linkis/cs/server/service/ContextIDService.java | 20 ++ .../cs/server/service/ContextListenerService.java | 23 +++ .../linkis/cs/server/service/ContextService.java | 41 ++++ .../linkis/cs/server/service/Service.java | 22 +++ .../server/service/impl/ContextIDServiceImpl.java | 64 ++++++ .../service/impl/ContextListenerServiceImpl.java | 81 ++++++++ .../cs/server/service/impl/ContextServiceImpl.java | 218 +++++++++++++++++++++ .../linkis/cs/server/util/CsUtils.java | 21 ++ .../cs-server/src/main/resources/application.yml | 22 +++ .../cs-server/src/main/resources/linkis.properties | 32 +++ .../cs-server/src/main/resources/log4j.properties | 37 ++++ .../cs-server/src/main/resources/log4j2.xml | 39 ++++ 54 files changed, 2611 insertions(+) diff --git a/contextservice/cs-server/bin/start-cs-server.sh b/contextservice/cs-server/bin/start-cs-server.sh new file mode 100644 index 0000000000..80cc775a4a --- /dev/null +++ b/contextservice/cs-server/bin/start-cs-server.sh @@ -0,0 +1,38 @@ +#!/bin/bash + +cd `dirname $0` +cd .. +HOME=`pwd` + +export SERVER_PID=$HOME/bin/linkis.pid +export SERVER_LOG_PATH=$HOME/logs +export SERVER_CLASS=com.webank.wedatasphere.linkis.DataWorkCloudApplication + +if test -z "$SERVER_HEAP_SIZE" +then + export SERVER_HEAP_SIZE="512M" +fi + +if test -z "$SERVER_JAVA_OPTS" +then + export SERVER_JAVA_OPTS=" -Xmx$SERVER_HEAP_SIZE -XX:+UseG1GC -Xloggc:$HOME/logs/linkis-gc.log" +fi + +if [[ -f "${SERVER_PID}" ]]; then + pid=$(cat ${SERVER_PID}) + if kill -0 ${pid} >/dev/null 2>&1; then + echo "Server is already running." + exit 1 + fi +fi + +nohup java $SERVER_JAVA_OPTS -cp ../module/lib/*:$HOME/conf:$HOME/lib/* $SERVER_CLASS 2>&1 > $SERVER_LOG_PATH/linkis.out & +pid=$! +if [[ -z "${pid}" ]]; then + echo "server $SERVER_NAME start failed!" + exit 1 +else + echo "server $SERVER_NAME start succeeded!" + echo $pid > $SERVER_PID + sleep 1 +fi \ No newline at end of file diff --git a/contextservice/cs-server/bin/stop--cs-server.sh b/contextservice/cs-server/bin/stop--cs-server.sh new file mode 100644 index 0000000000..f032887111 --- /dev/null +++ b/contextservice/cs-server/bin/stop--cs-server.sh @@ -0,0 +1,47 @@ +#!/bin/bash + +cd `dirname $0` +cd .. +HOME=`pwd` + +export SERVER_PID=$HOME/bin/linkis.pid + +function wait_for_server_to_die() { + local pid + local count + pid=$1 + timeout=$2 + count=0 + timeoutTime=$(date "+%s") + let "timeoutTime+=$timeout" + currentTime=$(date "+%s") + forceKill=1 + + while [[ $currentTime -lt $timeoutTime ]]; do + $(kill ${pid} > /dev/null 2> /dev/null) + if kill -0 ${pid} > /dev/null 2>&1; then + sleep 3 + else + forceKill=0 + break + fi + currentTime=$(date "+%s") + done + + if [[ forceKill -ne 0 ]]; then + $(kill -9 ${pid} > /dev/null 2> /dev/null) + fi +} + +if [[ ! -f "${SERVER_PID}" ]]; then + echo "server $SERVER_NAME is not running" +else + pid=$(cat ${SERVER_PID}) + if [[ -z "${pid}" ]]; then + echo "server $SERVER_NAME is not running" + else + wait_for_server_to_die $pid 40 + $(rm -f ${SERVER_PID}) + echo "server $SERVER_NAME is stopped." + fi +fi \ No newline at end of file diff --git a/contextservice/cs-server/conf/application.yml b/contextservice/cs-server/conf/application.yml new file mode 100644 index 0000000000..0f00605b84 --- /dev/null +++ b/contextservice/cs-server/conf/application.yml @@ -0,0 +1,28 @@ +server: + port: 9042 +spring: + application: + name: cloud-contextservice + + +eureka: + client: + serviceUrl: + defaultZone: locahost + instance: + metadata-map: + test: wedatasphere + +management: + endpoints: + web: + exposure: + include: refresh,info +logging: + config: classpath:log4j2.xml + +pagehelper: + helper-dialect: mysql + reasonable: true + support-methods-arguments: true + params: countSql \ No newline at end of file diff --git a/contextservice/cs-server/conf/linkis.properties b/contextservice/cs-server/conf/linkis.properties new file mode 100644 index 0000000000..78e88064ab --- /dev/null +++ b/contextservice/cs-server/conf/linkis.properties @@ -0,0 +1,22 @@ +# +# 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.server.mybatis.datasource.url=jdbc:mysql://127.0.0.1:3306/dss_dev_center?characterEncoding=UTF-8 +wds.linkis.server.mybatis.datasource.username= +wds.linkis.server.mybatis.datasource.password= +wds.linkis.server.version=v1 +##restful +wds.linkis.server.restful.scan.packages=com.webank.wedatasphere.linkis.cs.server.restful +##mybatis +wds.linkis.server.mybatis.mapperLocations=classpath*:com/webank/wedatasphere/linkis/cs/persistence/dao/impl/*.xml +wds.linkis.server.mybatis.typeAliasesPackage=com.webank.wedatasphere.linkis.cs.persistence.entity +wds.linkis.server.mybatis.BasePackage=com.webank.wedatasphere.linkis.cs.persistence.dao diff --git a/contextservice/cs-server/conf/log4j.properties b/contextservice/cs-server/conf/log4j.properties new file mode 100644 index 0000000000..de6691ad21 --- /dev/null +++ b/contextservice/cs-server/conf/log4j.properties @@ -0,0 +1,26 @@ +# +# 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. +# +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 \ No newline at end of file diff --git a/contextservice/cs-server/conf/log4j2.xml b/contextservice/cs-server/conf/log4j2.xml new file mode 100644 index 0000000000..1c68190669 --- /dev/null +++ b/contextservice/cs-server/conf/log4j2.xml @@ -0,0 +1,35 @@ +<?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/contextservice/cs-server/pom.xml b/contextservice/cs-server/pom.xml new file mode 100644 index 0000000000..eb2c3350b2 --- /dev/null +++ b/contextservice/cs-server/pom.xml @@ -0,0 +1,183 @@ +<?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-server</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-cs-cache</artifactId> + <version>${linkis.version}</version> + <exclusions> + <exclusion> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-module</artifactId> + </exclusion> + </exclusions> + </dependency> + + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cs-listener</artifactId> + <version>${linkis.version}</version> + </dependency> + + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cs-persistence</artifactId> + <version>${linkis.version}</version> + <exclusions> + <exclusion> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-module</artifactId> + </exclusion> + </exclusions> + </dependency> + + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cs-highavailable</artifactId> + <version>${linkis.version}</version> + <exclusions> + <exclusion> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-module</artifactId> + </exclusion> + </exclusions> + </dependency> + + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cs-search</artifactId> + <version>${linkis.version}</version> + <exclusions> + <exclusion> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-module</artifactId> + </exclusion> + </exclusions> + </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-scheduler</artifactId> + <version>${linkis.version}</version> + </dependency> + + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <version>4.12</version> + <scope>test</scope> + </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> + <executions> + <execution> + <id>scala-compile-first</id> + <phase>process-resources</phase> + <goals> + <goal>add-source</goal> + <goal>compile</goal> + </goals> + </execution> + </executions> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-jar-plugin</artifactId> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-assembly-plugin</artifactId> + <version>2.3</version> + <inherited>false</inherited> + <executions> + <execution> + <id>make-assembly</id> + <phase>package</phase> + <goals> + <goal>single</goal> + </goals> + <configuration> + <descriptors> + <descriptor>src/main/assembly/distribution.xml</descriptor> + </descriptors> + </configuration> + </execution> + </executions> + <configuration> + <skipAssembly>false</skipAssembly> + <finalName>linkis-cs-server</finalName> + <appendAssemblyId>false</appendAssemblyId> + <attach>false</attach> + <descriptors> + <descriptor>src/main/assembly/distribution.xml</descriptor> + </descriptors> + </configuration> + </plugin> + </plugins> + <resources> + <resource> + <directory>src/main/java</directory> + <includes> + <include>**/*.xml</include> + </includes> + </resource> + <resource> + <directory>src/main/resources</directory> + <excludes> + <exclude>**/*.properties</exclude> + <exclude>**/application.yml</exclude> + <exclude>**/bootstrap.yml</exclude> + <exclude>**/log4j2.xml</exclude> + </excludes> + </resource> + </resources> + </build> +</project> \ No newline at end of file diff --git a/contextservice/cs-server/src/main/assembly/distribution.xml b/contextservice/cs-server/src/main/assembly/distribution.xml new file mode 100644 index 0000000000..ac1d394d86 --- /dev/null +++ b/contextservice/cs-server/src/main/assembly/distribution.xml @@ -0,0 +1,71 @@ +<!-- + ~ 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. + --> + +<assembly + xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/2.3"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/2.3 http://maven.apache.org/xsd/assembly-1.1.2.xsd";> + <id>linkis-cs-server</id> + <formats> + <format>zip</format> + </formats> + <includeBaseDirectory>true</includeBaseDirectory> + <baseDirectory>linkis-cs-server</baseDirectory> + + <dependencySets> + <dependencySet> + <!-- Enable access to all projects in the current multimodule build! <useAllReactorProjects>true</useAllReactorProjects> --> + <!-- Now, select which projects to include in this module-set. --> + <outputDirectory>lib</outputDirectory> + <useProjectArtifact>true</useProjectArtifact> + <useTransitiveDependencies>true</useTransitiveDependencies> + <unpack>false</unpack> + <useStrictFiltering>false</useStrictFiltering> + <useTransitiveFiltering>true</useTransitiveFiltering> + + </dependencySet> + </dependencySets> + + <fileSets> + <fileSet> + <directory>${basedir}/conf</directory> + <includes> + <include>*</include> + </includes> + <fileMode>0777</fileMode> + <outputDirectory>conf</outputDirectory> + <lineEnding>unix</lineEnding> + </fileSet> + <fileSet> + <directory>${basedir}/bin</directory> + <includes> + <include>*</include> + </includes> + <fileMode>0777</fileMode> + <outputDirectory>bin</outputDirectory> + <lineEnding>unix</lineEnding> + </fileSet> + <fileSet> + <directory>.</directory> + <excludes> + <exclude>*/**</exclude> + </excludes> + <outputDirectory>logs</outputDirectory> + </fileSet> + </fileSets> + +</assembly> + diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/conf/ContextServerConf.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/conf/ContextServerConf.java new file mode 100644 index 0000000000..442a1d8545 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/conf/ContextServerConf.java @@ -0,0 +1,12 @@ +package com.webank.wedatasphere.linkis.cs.server.conf; + +import com.webank.wedatasphere.linkis.common.conf.CommonVars; + +/** + * @author peacewong + * @date 2020/2/9 17:04 + */ +public class ContextServerConf { + + public final static String KEYWORD_SCAN_PACKAGE = CommonVars.apply("wds.linkis.cs.keyword.scan.package","com.webank.wedatasphere.linkis.cs").getValue(); +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/enumeration/ServiceMethod.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/enumeration/ServiceMethod.java new file mode 100644 index 0000000000..5870cd9a5f --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/enumeration/ServiceMethod.java @@ -0,0 +1,11 @@ +package com.webank.wedatasphere.linkis.cs.server.enumeration; + +/** + * Created by patinousward on 2020/2/22. + */ +public enum ServiceMethod { + /** + * + */ + CREATE, GET, SEARCH, REMOVE, REMOVEALL, UPDATE, RESET, SET, BIND, HEARTBEAT +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/enumeration/ServiceType.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/enumeration/ServiceType.java new file mode 100644 index 0000000000..93649c3e9d --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/enumeration/ServiceType.java @@ -0,0 +1,35 @@ +package com.webank.wedatasphere.linkis.cs.server.enumeration; + +import com.webank.wedatasphere.linkis.cs.server.protocol.*; + +/** + * Created by patinousward on 2020/2/19. + */ +public enum ServiceType { + /** + * + */ + CONTEXT_ID { + @Override + public HttpRequestProtocol getRequestProtocol() { + return new ContextIDProtocol(); + } + }, + CONTEXT { + @Override + public HttpRequestProtocol getRequestProtocol() { + return new ContextProtocol(); + } + }, + + CONTEXT_LISTENER { + @Override + public HttpRequestProtocol getRequestProtocol() { + return new ContextListenerProtocol(); + } + }; + + public HttpRequestProtocol getRequestProtocol() { + return null; + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/DefaultKeywordParser.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/DefaultKeywordParser.java new file mode 100644 index 0000000000..67ffc184d8 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/DefaultKeywordParser.java @@ -0,0 +1,124 @@ +package com.webank.wedatasphere.linkis.cs.server.parser; + +import com.webank.wedatasphere.linkis.cs.common.annotation.KeywordMethod; +import com.webank.wedatasphere.linkis.cs.server.conf.ContextServerConf; +import org.apache.commons.lang.StringUtils; +import org.reflections.ReflectionUtils; +import org.reflections.Reflections; +import org.reflections.scanners.MethodAnnotationsScanner; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import java.lang.reflect.Method; +import java.util.*; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * @author peacewong + * @date 2020/2/9 16:19 + */ +@Component +public class DefaultKeywordParser implements KeywordParser { + + private static final Logger logger = LoggerFactory.getLogger(DefaultKeywordParser.class); + + + Map<String, Set<KeywordMethodEntity>> keywordMethods = new HashMap<>(); + + Map<String, Set<Class<?>>> classRecord = new HashMap<>(); + + + @PostConstruct + private void init() { + logger.info("init keyValueParser"); + scanKeywordMethods(); + } + + private void scanKeywordMethods() { + Reflections reflections = new Reflections(ContextServerConf.KEYWORD_SCAN_PACKAGE, new MethodAnnotationsScanner()); + Set<Method> methods = reflections.getMethodsAnnotatedWith(KeywordMethod.class); + Iterator<Method> iterator = methods.iterator(); + while (iterator.hasNext()) { + Method method = iterator.next(); + method.setAccessible(true); + + KeywordMethod annotation = method.getAnnotation(KeywordMethod.class); + KeywordMethodEntity keywordMethodEntity = new KeywordMethodEntity(); + keywordMethodEntity.setMethod(method); + keywordMethodEntity.setRegex(annotation.regex()); + keywordMethodEntity.setSplitter(annotation.splitter()); + + String className = method.getDeclaringClass().getName(); + if (!keywordMethods.containsKey(className)) { + keywordMethods.put(className, new HashSet<>()); + } + keywordMethods.get(className).add(keywordMethodEntity); + } + } + + /** + * + * @param obj + * @return + * @throws Exception + */ + private Set<String> parseKeywords(Object obj) throws Exception { + Objects.requireNonNull(obj); + Set<String> keywords = new HashSet<>(); + String className = obj.getClass().getName(); + if (!classRecord.containsKey(className)) { + classRecord.put(className, ReflectionUtils.getAllSuperTypes(obj.getClass())); + } + + Iterator<Class<?>> classIterator = classRecord.get(className).iterator(); + while (classIterator.hasNext()) { + Class<?> clazz = classIterator.next(); + if (! keywordMethods.containsKey(clazz.getName())) { + continue; + } + Iterator<KeywordMethodEntity> keywordMethodEntityIterator = keywordMethods.get(clazz.getName()).iterator(); + while (keywordMethodEntityIterator.hasNext()) { + KeywordMethodEntity methodEntity = keywordMethodEntityIterator.next(); + Object methodReturn = methodEntity.getMethod().invoke(obj); + if (null == methodReturn || StringUtils.isBlank(methodReturn.toString())) { + continue; + } + if (StringUtils.isNotBlank(methodEntity.getSplitter())) { + Collections.addAll(keywords, methodReturn.toString().split(methodEntity.getSplitter())); + } else if (StringUtils.isNotBlank(methodEntity.getRegex())) { + keywords.addAll(getString(methodReturn.toString(), methodEntity.getRegex())); + } else { + keywords.add(methodReturn.toString()); + } + } + } + return keywords; + } + + @Override + public Set<String> parse(Object obj) { + //先解析key + Set<String> keywords = new HashSet<>(); + try { + keywords = parseKeywords(obj); + } catch (Exception e){ + logger.error("Failed to parse keywords ", e); + } + return keywords; + } + + private Set<String> getString(String s, String regex) { + + Set<String> keywords = new HashSet<>(); + Pattern p = Pattern.compile(regex); + Matcher m = p.matcher(s); + while(m.find()) { + keywords.add(m.group()); + + } + return keywords; + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/KeywordMethodEntity.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/KeywordMethodEntity.java new file mode 100644 index 0000000000..2ab6a19ec0 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/KeywordMethodEntity.java @@ -0,0 +1,55 @@ +package com.webank.wedatasphere.linkis.cs.server.parser; + +import java.lang.reflect.Method; + +/** + * @author peacewong + * @date 2020/2/9 16:41 + */ +public class KeywordMethodEntity { + + private Method method; + + private String splitter; + + private String regex; + + public Method getMethod() { + return method; + } + + public void setMethod(Method method) { + this.method = method; + } + + public String getSplitter() { + return splitter; + } + + public void setSplitter(String splitter) { + this.splitter = splitter; + } + + public String getRegex() { + return regex; + } + + public void setRegex(String regex) { + this.regex = regex; + } + + @Override + public int hashCode() { + return method.hashCode(); + } + + @Override + public boolean equals(Object obj) { + return method.equals(obj); + } + + @Override + public String toString() { + return super.toString(); + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/KeywordParser.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/KeywordParser.java new file mode 100644 index 0000000000..5302fd1ec3 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/parser/KeywordParser.java @@ -0,0 +1,11 @@ +package com.webank.wedatasphere.linkis.cs.server.parser; + +import java.util.Set; + +/** + * @author peacewong + * @date 2020/2/18 19:01 + */ +public interface KeywordParser { + Set<String> parse(Object obj); +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/AbstractHttpRequestProtocol.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/AbstractHttpRequestProtocol.java new file mode 100644 index 0000000000..291396af8e --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/AbstractHttpRequestProtocol.java @@ -0,0 +1,45 @@ +package com.webank.wedatasphere.linkis.cs.server.protocol; + +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceMethod; + +/** + * Created by patinousward on 2020/2/18. + */ +public abstract class AbstractHttpRequestProtocol implements HttpRequestProtocol { + + private Object[] requestObjects; + + private String username; + + private ServiceMethod serviceMethod; + + @Override + public Object[] getRequestObjects() { + return requestObjects; + } + + @Override + public void setRequestObjects(Object[] requestObjects) { + this.requestObjects = requestObjects; + } + + @Override + public String getUsername() { + return username; + } + + @Override + public ServiceMethod getServiceMethod() { + return serviceMethod; + } + + @Override + public void setServiceMethod(ServiceMethod serviceMethod) { + this.serviceMethod = serviceMethod; + } + + @Override + public void setUsername(String username) { + this.username = username; + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextIDProtocol.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextIDProtocol.java new file mode 100644 index 0000000000..329b1824d6 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextIDProtocol.java @@ -0,0 +1,13 @@ +package com.webank.wedatasphere.linkis.cs.server.protocol; + +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; + +/** + * Created by patinousward on 2020/2/18. + */ +public class ContextIDProtocol extends AbstractHttpRequestProtocol { + @Override + public String getServiceName() { + return ServiceType.CONTEXT_ID.name(); + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextListenerProtocol.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextListenerProtocol.java new file mode 100644 index 0000000000..f2feedf8a8 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextListenerProtocol.java @@ -0,0 +1,14 @@ +package com.webank.wedatasphere.linkis.cs.server.protocol; + +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; + +/** + * Created by patinousward on 2020/2/22. + */ +public class ContextListenerProtocol extends AbstractHttpRequestProtocol { + + @Override + public String getServiceName() { + return ServiceType.CONTEXT_LISTENER.name(); + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextProtocol.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextProtocol.java new file mode 100644 index 0000000000..eedf6f90de --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/ContextProtocol.java @@ -0,0 +1,14 @@ +package com.webank.wedatasphere.linkis.cs.server.protocol; + +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; + +/** + * Created by patinousward on 2020/2/22. + */ +public class ContextProtocol extends AbstractHttpRequestProtocol { + + @Override + public String getServiceName() { + return ServiceType.CONTEXT.name(); + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpProtocol.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpProtocol.java new file mode 100644 index 0000000000..57d59db888 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpProtocol.java @@ -0,0 +1,9 @@ +package com.webank.wedatasphere.linkis.cs.server.protocol; + +import com.webank.wedatasphere.linkis.protocol.Protocol; + +/** + * Created by patinousward on 2020/2/18. + */ +public interface HttpProtocol extends Protocol { +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpRequestProtocol.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpRequestProtocol.java new file mode 100644 index 0000000000..1c2cc110a9 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpRequestProtocol.java @@ -0,0 +1,24 @@ +package com.webank.wedatasphere.linkis.cs.server.protocol; + +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceMethod; + +/** + * Created by patinousward on 2020/2/18. + */ +public interface HttpRequestProtocol extends HttpProtocol { + + Object[] getRequestObjects(); + + void setRequestObjects(Object[] objs); + + String getServiceName(); + + void setUsername(String username); + + String getUsername(); + + ServiceMethod getServiceMethod(); + + void setServiceMethod(ServiceMethod method); + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpResponseProtocol.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpResponseProtocol.java new file mode 100644 index 0000000000..098b14123c --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/HttpResponseProtocol.java @@ -0,0 +1,21 @@ +package com.webank.wedatasphere.linkis.cs.server.protocol; + +/** + * Created by patinousward on 2020/2/18. + */ +public interface HttpResponseProtocol<T> extends HttpProtocol { + + void waitForComplete() throws InterruptedException; + + void waitTimeEnd(long mills) throws InterruptedException; + + void notifyJob(); + + T get(); + + void set(T t); + + Object getResponseData(); + + void setResponseData(Object responseData); +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/RestResponseProtocol.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/RestResponseProtocol.java new file mode 100644 index 0000000000..319a6046fc --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/protocol/RestResponseProtocol.java @@ -0,0 +1,86 @@ +package com.webank.wedatasphere.linkis.cs.server.protocol; + +import com.webank.wedatasphere.linkis.server.Message; +import org.apache.commons.lang.exception.ExceptionUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.util.StringUtils; + +/** + * Created by patinousward on 2020/2/18. + */ +public class RestResponseProtocol implements HttpResponseProtocol<Message> { + + private Logger logger = LoggerFactory.getLogger(getClass()); + + private final Object lock = new Object(); + + private Message message; + + private Object responseData; + + @Override + public void waitForComplete() throws InterruptedException { + synchronized (lock) { + lock.wait(); + } + } + + @Override + public void waitTimeEnd(long mills) throws InterruptedException { + logger.info(String.format("start to wait %smills until job complete", mills)); + synchronized (lock) { + lock.wait(mills); + } + } + + @Override + public void notifyJob() { + logger.info("notify the job"); + synchronized (lock) { + lock.notify(); + } + } + + @Override + public Message get() { + return this.message; + } + + @Override + public void set(Message message) { + this.message = message; + } + + @Override + public Object getResponseData() { + return this.responseData; + } + + @Override + public void setResponseData(Object responseData) { + this.responseData = responseData; + } + + public void ok(String msg) { + if (message == null) { + message = new Message(); + } + if (StringUtils.isEmpty(msg)) { + message.setMessage("OK"); + } else { + message.setMessage(msg); + } + } + + public void error(String msg, Throwable t) { + if (message == null) { + message = new Message(); + message.setStatus(1); + } + message.setMessage(msg); + if (t != null) { + message.$less$less("stack", ExceptionUtils.getFullStackTrace(t)); + } + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextIDRestfulApi.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextIDRestfulApi.java new file mode 100644 index 0000000000..149a769d9d --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextIDRestfulApi.java @@ -0,0 +1,99 @@ +package com.webank.wedatasphere.linkis.cs.server.restful; + +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.common.protocol.ContextHTTPConstant; +import com.webank.wedatasphere.linkis.cs.common.utils.CSCommonUtils; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceMethod; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; +import com.webank.wedatasphere.linkis.cs.server.scheduler.CsScheduler; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpAnswerJob; +import com.webank.wedatasphere.linkis.server.Message; +import org.codehaus.jackson.JsonNode; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.*; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import java.io.IOException; + +/** + * Created by patinousward on 2020/2/18. + */ +@Component +@Path("/contextservice") +@Produces(MediaType.APPLICATION_JSON) +@Consumes(MediaType.APPLICATION_JSON) +public class ContextIDRestfulApi implements CsRestfulParent { + + @Autowired + private CsScheduler csScheduler; + + @POST + @Path("createContextID") + public Response createContextID(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, ClassNotFoundException, IOException, CSErrorException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.CREATE, contextID); + return Message.messageToResponse(generateResponse(answerJob, "contextId")); + } + + @GET + @Path("getContextID") + public Response getContextID(@Context HttpServletRequest req, @QueryParam("contextId") String id) throws InterruptedException, CSErrorException { + if (StringUtils.isEmpty(id)) { + throw new CSErrorException(97000, "contxtId cannot be empty"); + } + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.GET, id); + Message message = generateResponse(answerJob, "contextID"); + return Message.messageToResponse(message); + } + + @POST + @Path("updateContextID") + public Response updateContextID(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + if (StringUtils.isEmpty(contextID.getContextId())) { + throw new CSErrorException(97000, "contxtId cannot be empty"); + } + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.UPDATE, contextID); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("resetContextID") + public Response resetContextID(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException { + if (!jsonNode.has(ContextHTTPConstant.CONTEXT_ID_STR)) { + throw new CSErrorException(97000, ContextHTTPConstant.CONTEXT_ID_STR + " cannot be empty"); + } + String id = jsonNode.get(ContextHTTPConstant.CONTEXT_ID_STR).getTextValue(); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.RESET, id); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + + @POST + @Path("removeContextID") + public Response removeContextID(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException { + String id = jsonNode.get("contextId").getTextValue(); + if (StringUtils.isEmpty(id)) { + throw new CSErrorException(97000, "contxtId cannot be empty"); + } + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.REMOVE, id); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @Override + public ServiceType getServiceType() { + return ServiceType.CONTEXT_ID; + } + + @Override + public CsScheduler getScheduler() { + return this.csScheduler; + } + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextListenerRestfulApi.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextListenerRestfulApi.java new file mode 100644 index 0000000000..18e3deb65c --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextListenerRestfulApi.java @@ -0,0 +1,83 @@ +package com.webank.wedatasphere.linkis.cs.server.restful; + +import com.webank.wedatasphere.linkis.cs.common.entity.listener.CommonContextIDListenerDomain; +import com.webank.wedatasphere.linkis.cs.common.entity.listener.CommonContextKeyListenerDomain; +import com.webank.wedatasphere.linkis.cs.common.entity.listener.ContextIDListenerDomain; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKey; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceMethod; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; +import com.webank.wedatasphere.linkis.cs.server.scheduler.CsScheduler; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpAnswerJob; +import com.webank.wedatasphere.linkis.server.Message; +import org.codehaus.jackson.JsonNode; +import org.codehaus.jackson.map.ObjectMapper; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.Consumes; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.Produces; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import java.io.IOException; + +/** + * Created by patinousward on 2020/2/18. + */ +@Component +@Path("/contextservice") +@Produces(MediaType.APPLICATION_JSON) +@Consumes(MediaType.APPLICATION_JSON) +public class ContextListenerRestfulApi implements CsRestfulParent { + + @Autowired + private CsScheduler csScheduler; + + private ObjectMapper objectMapper = new ObjectMapper(); + + @POST + @Path("onBindIDListener") + public Response onBindIDListener(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + String source = jsonNode.get("source").getTextValue(); + ContextID contextID = getContextIDFromJsonNode(jsonNode); + ContextIDListenerDomain listener = new CommonContextIDListenerDomain(); + listener.setSource(source); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.BIND, contextID, listener); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("onBindKeyListener") + public Response onBindKeyListener(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + String source = jsonNode.get("source").getTextValue(); + ContextID contextID = getContextIDFromJsonNode(jsonNode); + ContextKey contextKey = getContextKeyFromJsonNode(jsonNode); + CommonContextKeyListenerDomain listener = new CommonContextKeyListenerDomain(); + listener.setSource(source); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.BIND, contextID, contextKey, listener); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("heartbeat") + public Response heartbeat(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, IOException, CSErrorException { + String source = jsonNode.get("source").getTextValue(); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.HEARTBEAT, source); + return Message.messageToResponse(generateResponse(answerJob, "ContextKeyValueBean")); + } + + @Override + public ServiceType getServiceType() { + return ServiceType.CONTEXT_LISTENER; + } + + @Override + public CsScheduler getScheduler() { + return this.csScheduler; + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextRestfulApi.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextRestfulApi.java new file mode 100644 index 0000000000..dfee6770b6 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/ContextRestfulApi.java @@ -0,0 +1,156 @@ +package com.webank.wedatasphere.linkis.cs.server.restful; + +import com.webank.wedatasphere.linkis.cs.common.entity.enumeration.ContextType; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKey; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKeyValue; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextValue; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.common.protocol.ContextHTTPConstant; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceMethod; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; +import com.webank.wedatasphere.linkis.cs.server.scheduler.CsScheduler; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpAnswerJob; +import com.webank.wedatasphere.linkis.server.Message; +import org.codehaus.jackson.JsonNode; +import org.codehaus.jackson.map.ObjectMapper; +import org.codehaus.jackson.type.TypeReference; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import javax.servlet.http.HttpServletRequest; +import javax.ws.rs.Consumes; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.Produces; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import java.io.IOException; +import java.util.Map; + + +/** + * Created by patinousward on 2020/2/18. + */ +@Component +@Path("contextservice") +@Produces(MediaType.APPLICATION_JSON) +@Consumes(MediaType.APPLICATION_JSON) +public class ContextRestfulApi implements CsRestfulParent { + + private static final Logger LOGGER = LoggerFactory.getLogger(ContextRestfulApi.class); + + @Autowired + private CsScheduler csScheduler; + + private ObjectMapper objectMapper = new ObjectMapper(); + + @POST + @Path("getContextValue") + public Response getContextValue(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + ContextKey contextKey = getContextKeyFromJsonNode(jsonNode); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.GET, contextID, contextKey); + Message message = generateResponse(answerJob, "contextValue"); + return Message.messageToResponse(message); + } + + + @POST + @Path("searchContextValue") + public Response searchContextValue(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + JsonNode condition = jsonNode.get("condition"); + Map<Object, Object> conditionMap = objectMapper.convertValue(condition, new TypeReference<Map<Object, Object>>() { + }); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.SEARCH, contextID, conditionMap); + Message message = generateResponse(answerJob, "contextKeyValue"); + return Message.messageToResponse(message); + } + +/* @GET + @Path("searchContextValueByCondition") + public Response searchContextValueByCondition(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException { + Condition condition = null; + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.SEARCH, condition); + return generateResponse(answerJob,""); + }*/ + + + @POST + @Path("setValueByKey") + public Response setValueByKey(@Context HttpServletRequest req, JsonNode jsonNode) throws CSErrorException, IOException, ClassNotFoundException, InterruptedException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + ContextKey contextKey = getContextKeyFromJsonNode(jsonNode); + ContextValue contextValue = getContextValueFromJsonNode(jsonNode); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.SET, contextID, contextKey, contextValue); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("setValue") + public Response setValue(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + ContextKeyValue contextKeyValue = getContextKeyValueFromJsonNode(jsonNode); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.SET, contextID, contextKeyValue); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("resetValue") + public Response resetValue(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + ContextKey contextKey = getContextKeyFromJsonNode(jsonNode); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.RESET, contextID, contextKey); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("removeValue") + public Response removeValue(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + ContextKey contextKey = getContextKeyFromJsonNode(jsonNode); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.REMOVE, contextID, contextKey); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("removeAllValue") + public Response removeAllValue(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.REMOVEALL, contextID); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("removeAllValueByKeyPrefixAndContextType") + public Response removeAllValueByKeyPrefixAndContextType(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + String contextType = jsonNode.get(ContextHTTPConstant.CONTEXT_KEY_TYPE_STR).getTextValue(); + String keyPrefix = jsonNode.get(ContextHTTPConstant.CONTEXT_KEY_PREFIX_STR).getTextValue(); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.REMOVEALL, contextID, ContextType.valueOf(contextType),keyPrefix); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @POST + @Path("removeAllValueByKeyPrefix") + public Response removeAllValueByKeyPrefix(@Context HttpServletRequest req, JsonNode jsonNode) throws InterruptedException, CSErrorException, IOException, ClassNotFoundException { + ContextID contextID = getContextIDFromJsonNode(jsonNode); + String keyPrefix = jsonNode.get(ContextHTTPConstant.CONTEXT_KEY_PREFIX_STR).getTextValue(); + HttpAnswerJob answerJob = submitRestJob(req, ServiceMethod.REMOVEALL, contextID,keyPrefix); + return Message.messageToResponse(generateResponse(answerJob, "")); + } + + @Override + public ServiceType getServiceType() { + return ServiceType.CONTEXT; + } + + @Override + public CsScheduler getScheduler() { + return this.csScheduler; + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/CsRestfulParent.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/CsRestfulParent.java new file mode 100644 index 0000000000..db52d7b5e1 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/restful/CsRestfulParent.java @@ -0,0 +1,105 @@ +package com.webank.wedatasphere.linkis.cs.server.restful; + +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKey; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKeyValue; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextValue; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceMethod; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpRequestProtocol; +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpResponseProtocol; +import com.webank.wedatasphere.linkis.cs.server.protocol.RestResponseProtocol; +import com.webank.wedatasphere.linkis.cs.server.scheduler.CsScheduler; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpAnswerJob; +import com.webank.wedatasphere.linkis.cs.server.scheduler.RestJobBuilder; +import com.webank.wedatasphere.linkis.cs.server.util.CsUtils; +import com.webank.wedatasphere.linkis.server.Message; +import com.webank.wedatasphere.linkis.server.security.SecurityFilter; +import org.codehaus.jackson.JsonNode; + +import javax.servlet.http.HttpServletRequest; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +/** + * Created by patinousward on 2020/2/22. + */ +public interface CsRestfulParent { + + default HttpAnswerJob submitRestJob(HttpServletRequest req, + ServiceMethod method, + Object... objects) throws InterruptedException { + // TODO: 2020/3/3 单例 + HttpAnswerJob job = (HttpAnswerJob) new RestJobBuilder().build(getServiceType()); + HttpRequestProtocol protocol = job.getRequestProtocol(); + protocol.setUsername(SecurityFilter.getLoginUsername(req)); + protocol.setServiceMethod(method); + protocol.setRequestObjects(objects); + getScheduler().sumbit(job); + return job; + } + + default Message generateResponse(HttpAnswerJob job, String responseKey) throws CSErrorException { + HttpResponseProtocol responseProtocol = job.getResponseProtocol(); + if (responseProtocol instanceof RestResponseProtocol) { + Message message = ((RestResponseProtocol) responseProtocol).get(); + if (message == null) { + return Message.error("job execute timeout"); + } + int status = ((RestResponseProtocol) responseProtocol).get().getStatus(); + if (status == 1) { + //failed + return ((RestResponseProtocol) responseProtocol).get(); + } else if (status == 0) { + Object data = job.getResponseProtocol().getResponseData(); + if (data == null) { + return Message.ok().data(responseKey, null); + } else if (data instanceof List && ((List) data).isEmpty()) { + return Message.ok().data(responseKey, new String[]{}); + } else if (data instanceof List) { + ArrayList<String> strings = new ArrayList<>(); + for (Object d : (List) data) { + strings.add(CsUtils.serialize(d)); + } + return Message.ok().data(responseKey, strings); + } else { + String dataStr = CsUtils.serialize(data); + return Message.ok().data(responseKey, dataStr); + } + } else { + + } + } + return Message.ok(); + } + + ServiceType getServiceType(); + + CsScheduler getScheduler(); + + default ContextID getContextIDFromJsonNode(JsonNode jsonNode) throws CSErrorException, IOException, ClassNotFoundException { + return deserialize(jsonNode, "contextID"); + } + + default <T> T deserialize(JsonNode jsonNode, String key) throws CSErrorException { + String str = jsonNode.get(key).getTextValue(); + return (T) CsUtils.SERIALIZE.deserialize(str); + } + + default ContextKey getContextKeyFromJsonNode(JsonNode jsonNode) throws CSErrorException, IOException, ClassNotFoundException { + return deserialize(jsonNode, "contextKey"); + } + + default ContextValue getContextValueFromJsonNode(JsonNode jsonNode) throws CSErrorException, IOException, ClassNotFoundException { + return deserialize(jsonNode, "contextValue"); + } + + default ContextKeyValue getContextKeyValueFromJsonNode(JsonNode jsonNode) throws CSErrorException, IOException, ClassNotFoundException { + return deserialize(jsonNode, "contextKeyValue"); + } + + + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/CsScheduler.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/CsScheduler.java new file mode 100644 index 0000000000..502e478a7c --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/CsScheduler.java @@ -0,0 +1,15 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler; + +import com.webank.wedatasphere.linkis.cs.server.service.Service; + +/** + * Created by patinousward on 2020/2/21. + */ +public interface CsScheduler { + + void addService(Service service); + + Service[] getServices(); + + void sumbit(HttpJob job) throws InterruptedException; +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/DefaultCsScheduler.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/DefaultCsScheduler.java new file mode 100644 index 0000000000..2e549a013e --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/DefaultCsScheduler.java @@ -0,0 +1,83 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler; + +import com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl.CsJobListener; +import com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl.CsSchedulerJob; +import com.webank.wedatasphere.linkis.cs.server.service.Service; +import com.webank.wedatasphere.linkis.scheduler.Scheduler; +import com.webank.wedatasphere.linkis.scheduler.queue.Group; +import com.webank.wedatasphere.linkis.scheduler.queue.GroupFactory; +import com.webank.wedatasphere.linkis.scheduler.queue.Job; +import com.webank.wedatasphere.linkis.scheduler.queue.parallelqueue.ParallelGroup; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.util.Arrays; +import java.util.List; +import java.util.Optional; + +/** + * Created by patinousward on 2020/2/21. + */ +@Component +public class DefaultCsScheduler implements CsScheduler { + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Autowired + private Scheduler scheduler; + + @Autowired + private List<Service> services; + + @Override + public void addService(Service service) { + services.add(service); + } + + @Override + public Service[] getServices() { + return this.services.toArray(new Service[]{}); + } + + @Override + public void sumbit(HttpJob job) throws InterruptedException { + // TODO: 2020/3/3 参数配置化 + GroupFactory groupFactory = scheduler.getSchedulerContext().getOrCreateGroupFactory(); + Group group = groupFactory.getOrCreateGroup(job.getRequestProtocol().getUsername()); + if (group instanceof ParallelGroup) { + ParallelGroup parallelGroup = (ParallelGroup) group; + if (parallelGroup.getMaxRunningJobs() == 0) { + parallelGroup.setMaxRunningJobs(100); + } + if (parallelGroup.getMaxAskExecutorTimes() == 0) { + parallelGroup.setMaxAskExecutorTimes(1000); + } + } + //create csJob + Job csJob = buildJob(job); + //注册listener + csJob.setJobListener(new CsJobListener()); + scheduler.submit(csJob); + if (job instanceof HttpAnswerJob) { + HttpAnswerJob answerJob = (HttpAnswerJob) job; + answerJob.getResponseProtocol().waitTimeEnd(5000); + } + } + + private Job buildJob(HttpJob job) { + CsSchedulerJob csJob = new CsSchedulerJob(); + //暂时将groupName给jobid + csJob.setId(job.getRequestProtocol().getUsername()); + csJob.set(job); + //从多个serveice中找出一个合适执行的service + Optional<Service> service = Arrays.stream(getServices()).filter(s -> s.ifAccept(job)).findFirst(); + if (service.isPresent()) { + logger.info(String.format("find %s service to execute job",service.get().getName())); + csJob.setConsuemr(service.get()::accept); + } + return csJob; + } + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpAnswerJob.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpAnswerJob.java new file mode 100644 index 0000000000..59cc2b9612 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpAnswerJob.java @@ -0,0 +1,14 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler; + +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpResponseProtocol; + +/** + * Created by patinousward on 2020/2/18. + */ +public interface HttpAnswerJob extends HttpJob { + + HttpResponseProtocol getResponseProtocol(); + + void setResponseProtocol(HttpResponseProtocol protocol); + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpAnswerJobBuilder.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpAnswerJobBuilder.java new file mode 100644 index 0000000000..d605200da7 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpAnswerJobBuilder.java @@ -0,0 +1,50 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler; + +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpRequestProtocol; +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpResponseProtocol; + +/** + * Created by patinousward on 2020/2/22. + */ +public abstract class HttpAnswerJobBuilder extends HttpJobBuilder { + + @Override + public final HttpJob build(ServiceType serviceType) { + return buildResponseProtocol(super.build(serviceType)); + } + + @Override + protected final HttpJob createHttpJob() { + return new DefaultHttpAnswerJob(); + } + + protected abstract HttpJob buildResponseProtocol(HttpJob job); + + private static class DefaultHttpAnswerJob implements HttpAnswerJob { + private HttpRequestProtocol httpRequestProtocol; + + private HttpResponseProtocol httpResponseProtocol; + + @Override + public HttpRequestProtocol getRequestProtocol() { + return this.httpRequestProtocol; + } + + @Override + public void setRequestProtocol(HttpRequestProtocol protocol) { + this.httpRequestProtocol = protocol; + } + + @Override + public HttpResponseProtocol getResponseProtocol() { + return this.httpResponseProtocol; + } + + @Override + public void setResponseProtocol(HttpResponseProtocol protocol) { + this.httpResponseProtocol = protocol; + } + } + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpJob.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpJob.java new file mode 100644 index 0000000000..680dc4132d --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpJob.java @@ -0,0 +1,14 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler; + +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpRequestProtocol; + +/** + * Created by patinousward on 2020/2/18. + */ +public interface HttpJob { + + HttpRequestProtocol getRequestProtocol(); + + void setRequestProtocol(HttpRequestProtocol protocol); + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpJobBuilder.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpJobBuilder.java new file mode 100644 index 0000000000..7bef2df9c6 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpJobBuilder.java @@ -0,0 +1,21 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler; + +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; + +/** + * Created by patinousward on 2020/2/22. + */ +public abstract class HttpJobBuilder { + + public HttpJob build(ServiceType serviceType) { + return buildRequestProtocol(createHttpJob(), serviceType); + } + + private HttpJob buildRequestProtocol(HttpJob job, ServiceType serviceType) { + job.setRequestProtocol(serviceType.getRequestProtocol()); + return job; + } + + protected abstract HttpJob createHttpJob(); + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpPriorityJob.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpPriorityJob.java new file mode 100644 index 0000000000..926596578d --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/HttpPriorityJob.java @@ -0,0 +1,10 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler; + +/** + * Created by patinousward on 2020/2/18. + */ +public interface HttpPriorityJob extends HttpJob { + + int getPriority(); + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/RestJobBuilder.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/RestJobBuilder.java new file mode 100644 index 0000000000..98c885fa4c --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/RestJobBuilder.java @@ -0,0 +1,17 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler; + +import com.webank.wedatasphere.linkis.cs.server.protocol.RestResponseProtocol; + +/** + * Created by patinousward on 2020/2/22. + */ +public class RestJobBuilder extends HttpAnswerJobBuilder { + + @Override + protected final HttpJob buildResponseProtocol(HttpJob job) { + RestResponseProtocol protocol = new RestResponseProtocol(); + ((HttpAnswerJob) job).setResponseProtocol(protocol); + return job; + } + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecuteRequest.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecuteRequest.java new file mode 100644 index 0000000000..21119c0f84 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecuteRequest.java @@ -0,0 +1,43 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl; + +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpJob; +import com.webank.wedatasphere.linkis.scheduler.executer.ExecuteRequest; + +import java.util.function.Consumer; + +/** + * Created by patinousward on 2020/2/18. + */ +public class CsExecuteRequest implements ExecuteRequest, JobToExecuteRequestConsumer<HttpJob> { + + private HttpJob httpJob; + + // TODO: 2020/3/3 变量名修改 + private Consumer<HttpJob> jobConsumer; + + @Override + public String code() { + return null; + } + + + @Override + public HttpJob get() { + return this.httpJob; + } + + @Override + public void set(HttpJob httpJob) { + this.httpJob = httpJob; + } + + @Override + public Consumer<HttpJob> getConsumer() { + return this.jobConsumer; + } + + @Override + public void setConsuemr(Consumer<HttpJob> jobConsumer) { + this.jobConsumer = jobConsumer; + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecutor.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecutor.java new file mode 100644 index 0000000000..d4be541959 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecutor.java @@ -0,0 +1,49 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl; + +import com.webank.wedatasphere.linkis.protocol.engine.EngineState$; +import com.webank.wedatasphere.linkis.scheduler.executer.*; +import scala.Enumeration; + +import java.io.IOException; + +/** + * Created by patinousward on 2020/2/22. + */ +public class CsExecutor implements Executor { + + private long id; + private Enumeration.Value state; + + @Override + public long getId() { + return this.id; + } + + @Override + public ExecuteResponse execute(ExecuteRequest executeRequest) { + //httpjob执行的地方 + try { + if (executeRequest instanceof CsExecuteRequest) { + CsExecuteRequest request = (CsExecuteRequest) executeRequest; + request.getConsumer().accept(request.get()); + } + return new SuccessExecuteResponse(); + } catch (Exception e) { + return new ErrorExecuteResponse(e.getMessage(), e); + } + } + + @Override + public EngineState$.Value state() { + return this.state; + } + + @Override + public ExecutorInfo getExecutorInfo() { + return new ExecutorInfo(id, state); + } + + @Override + public void close() throws IOException { + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecutorManager.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecutorManager.java new file mode 100644 index 0000000000..015fae3d6c --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsExecutorManager.java @@ -0,0 +1,55 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl; + +import com.webank.wedatasphere.linkis.scheduler.executer.Executor; +import com.webank.wedatasphere.linkis.scheduler.executer.ExecutorManager; +import com.webank.wedatasphere.linkis.scheduler.listener.ExecutorListener; +import com.webank.wedatasphere.linkis.scheduler.queue.SchedulerEvent; +import scala.Option; +import scala.Some; +import scala.concurrent.duration.Duration; + +/** + * Created by patinousward on 2020/2/18. + */ +public class CsExecutorManager extends ExecutorManager { + + @Override + public void setExecutorListener(ExecutorListener executorListener) { + + } + + @Override + public Executor createExecutor(SchedulerEvent event) { + return new CsExecutor(); + } + + @Override + public Option<Executor> askExecutor(SchedulerEvent event) { + return new Some<>(createExecutor(event)); + } + + @Override + public Option<Executor> askExecutor(SchedulerEvent event, Duration wait) { + return askExecutor(event); + } + + @Override + public Option<Executor> getById(long id) { + return new Some<>(null); + } + + @Override + public Executor[] getByGroup(String groupName) { + return new Executor[0]; + } + + @Override + public void delete(Executor executor) { + + } + + @Override + public void shutdown() { + + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsJobListener.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsJobListener.java new file mode 100644 index 0000000000..80a8fdb3db --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsJobListener.java @@ -0,0 +1,51 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl; + +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpResponseProtocol; +import com.webank.wedatasphere.linkis.cs.server.protocol.RestResponseProtocol; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpAnswerJob; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpJob; +import com.webank.wedatasphere.linkis.scheduler.executer.ErrorExecuteResponse; +import com.webank.wedatasphere.linkis.scheduler.listener.JobListener; +import com.webank.wedatasphere.linkis.scheduler.queue.Job; + +/** + * Created by patinousward on 2020/2/22. + */ +public class CsJobListener implements JobListener { + @Override + public void onJobScheduled(Job job) { + //nothing to do + } + + @Override + public void onJobInited(Job job) { + //nothing to do + } + + @Override + public void onJobWaitForRetry(Job job) { + //nothing to do + } + + @Override + public void onJobRunning(Job job) { + //nothing to do + } + + @Override + public void onJobCompleted(Job job) { + // TODO: 2020/2/22 正常 /异常 + if (job instanceof CsSchedulerJob) { + HttpJob httpJob = ((CsSchedulerJob) job).get(); + if (httpJob instanceof HttpAnswerJob) { + HttpAnswerJob answerJob = (HttpAnswerJob) httpJob; + HttpResponseProtocol responseProtocol = answerJob.getResponseProtocol(); + if (!job.isSucceed() && responseProtocol instanceof RestResponseProtocol) { + ErrorExecuteResponse errorResponse = job.getErrorResponse(); + ((RestResponseProtocol) responseProtocol).error(errorResponse.message(), errorResponse.t()); + } + answerJob.getResponseProtocol().notifyJob(); + } + } + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsSchedulerBean.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsSchedulerBean.java new file mode 100644 index 0000000000..9944daa8ae --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsSchedulerBean.java @@ -0,0 +1,36 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl; + +import com.webank.wedatasphere.linkis.scheduler.Scheduler; +import com.webank.wedatasphere.linkis.scheduler.SchedulerContext; +import com.webank.wedatasphere.linkis.scheduler.executer.ExecutorManager; +import com.webank.wedatasphere.linkis.scheduler.queue.parallelqueue.ParallelScheduler; +import com.webank.wedatasphere.linkis.scheduler.queue.parallelqueue.ParallelSchedulerContextImpl; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; + +/** + * Created by patinousward on 2020/2/18. + */ +@Configuration +@Import(CsExecutorManager.class) +public class CsSchedulerBean { + + @Bean + public SchedulerContext getSchedulerContext(CsExecutorManager csExecutorManager) { + return new ParallelSchedulerContextImpl(3000) { + @Override + public ExecutorManager getOrCreateExecutorManager() { + return csExecutorManager; + } + }; + } + + @Bean + public Scheduler getScheduler(SchedulerContext context) { + ParallelScheduler parallelScheduler = new ParallelScheduler(context); + parallelScheduler.init(); + return parallelScheduler; + } + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsSchedulerJob.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsSchedulerJob.java new file mode 100644 index 0000000000..5130bdbd17 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/CsSchedulerJob.java @@ -0,0 +1,67 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl; + +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpJob; +import com.webank.wedatasphere.linkis.scheduler.executer.ExecuteRequest; +import com.webank.wedatasphere.linkis.scheduler.queue.Job; +import com.webank.wedatasphere.linkis.scheduler.queue.JobInfo; + +import java.io.IOException; +import java.util.function.Consumer; + +/** + * Created by patinousward on 2020/2/18. + */ +public class CsSchedulerJob extends Job implements JobToExecuteRequestConsumer<HttpJob> { + + private HttpJob httpJob; + + private Consumer<HttpJob> jobConsumer; + + @Override + public HttpJob get() { + return this.httpJob; + } + + @Override + public void set(HttpJob httpJob) { + this.httpJob = httpJob; + } + + @Override + public Consumer<HttpJob> getConsumer() { + return this.jobConsumer; + } + + @Override + public void setConsuemr(Consumer<HttpJob> jobConsumer) { + this.jobConsumer = jobConsumer; + } + + + @Override + public void init() { + // TODO: 2020/2/18 + } + + @Override + public ExecuteRequest jobToExecuteRequest() { + CsExecuteRequest request = new CsExecuteRequest(); + request.set(httpJob); + request.setConsuemr(jobConsumer); + return request; + } + + @Override + public String getName() { + return getId(); + } + + @Override + public JobInfo getJobInfo() { + return null; + } + + @Override + public void close() throws IOException { + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/JobToExecuteRequestConsumer.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/JobToExecuteRequestConsumer.java new file mode 100644 index 0000000000..c859d89932 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/scheduler/linkisImpl/JobToExecuteRequestConsumer.java @@ -0,0 +1,17 @@ +package com.webank.wedatasphere.linkis.cs.server.scheduler.linkisImpl; + +import java.util.function.Consumer; + +/** + * Created by patinousward on 2020/2/22. + */ +public interface JobToExecuteRequestConsumer<T> { + + T get(); + + void set(T t); + + Consumer<T> getConsumer(); + + void setConsuemr(Consumer<T> tConsumer); +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/AbstractService.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/AbstractService.java new file mode 100644 index 0000000000..ad104911a8 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/AbstractService.java @@ -0,0 +1,82 @@ +package com.webank.wedatasphere.linkis.cs.server.service; + +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.common.exception.CSWarnException; +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpRequestProtocol; +import com.webank.wedatasphere.linkis.cs.server.protocol.HttpResponseProtocol; +import com.webank.wedatasphere.linkis.cs.server.protocol.RestResponseProtocol; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpAnswerJob; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpJob; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.Optional; + +/** + * Created by patinousward on 2020/2/18. + */ +public abstract class AbstractService implements Service { + + protected final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Override + public boolean ifAccept(HttpJob job) { + return getName().equals(job.getRequestProtocol().getServiceName()); + } + + @Override + public void accept(HttpJob job) throws CSWarnException { + try { + //根据参数类型和方法名,选出方法进行调用 + HttpRequestProtocol protocol = job.getRequestProtocol(); + Object[] params = protocol.getRequestObjects(); + String method = protocol.getServiceMethod().name().toUpperCase(); + Method[] methods = this.getClass().getMethods(); + Optional<Method> first = Arrays.stream(methods).filter(f -> f.getName().toUpperCase().contains(method)) + .filter(f -> f.getParameterTypes().length == params.length) + .filter(f -> judgeMethod(f, params)).findFirst(); + Object response = first.orElseThrow(() -> new CSErrorException(97000, "can not find a method to invoke")).invoke(this, params); + if (job instanceof HttpAnswerJob) { + HttpResponseProtocol responseProtocol = ((HttpAnswerJob) job).getResponseProtocol(); + if (responseProtocol instanceof RestResponseProtocol) { + ((RestResponseProtocol) responseProtocol).ok(null); + } + responseProtocol.setResponseData(response); + } + } catch (Exception e) { + logger.error(String.format("execute %s service failed:", getName()), e); + throw new CSWarnException(97000, e.getMessage()); + } + } + + private boolean judgeMethod(Method method, Object... objects) { + boolean flag = true; + //传入参数类型是否是方法参数的子类 + Class<?>[] parameterTypes = method.getParameterTypes(); + for (int i = 0; i < parameterTypes.length; i++) { + if (!parameterTypes[i].isAssignableFrom(objects[i].getClass())) { + flag = false; + break; + } + } + return flag; + } + + @Override + public void init() { + + } + + @Override + public void start() { + + } + + @Override + public void close() throws IOException { + + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextIDService.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextIDService.java new file mode 100644 index 0000000000..d00d9a68d8 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextIDService.java @@ -0,0 +1,20 @@ +package com.webank.wedatasphere.linkis.cs.server.service; + +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; + +/** + * Created by patinousward on 2020/2/18. + */ +public abstract class ContextIDService extends AbstractService { + + public abstract String createContextID(ContextID contextID) throws CSErrorException; + + public abstract ContextID getContextID(String id) throws CSErrorException; + + public abstract void updateContextID(ContextID contextID) throws CSErrorException; + + public abstract void resetContextID(String id) throws CSErrorException; + + public abstract void removeContextID(String id) throws CSErrorException; +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextListenerService.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextListenerService.java new file mode 100644 index 0000000000..fd9a9bc370 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextListenerService.java @@ -0,0 +1,23 @@ +package com.webank.wedatasphere.linkis.cs.server.service; + +import com.webank.wedatasphere.linkis.cs.common.entity.listener.ContextIDListenerDomain; +import com.webank.wedatasphere.linkis.cs.common.entity.listener.ContextKeyListenerDomain; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKey; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.listener.callback.imp.ContextKeyValueBean; + +import java.util.List; + +/** + * Created by patinousward on 2020/2/18. + */ +public abstract class ContextListenerService extends AbstractService { + + public abstract void onBind(ContextID contextID, ContextIDListenerDomain listener) throws CSErrorException; + + public abstract void onBind(ContextID contextID, ContextKey contextKey, ContextKeyListenerDomain listener) throws CSErrorException; + + public abstract List<ContextKeyValueBean> heartbeat(String source); + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextService.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextService.java new file mode 100644 index 0000000000..9d83aa7d7e --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/ContextService.java @@ -0,0 +1,41 @@ +package com.webank.wedatasphere.linkis.cs.server.service; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.webank.wedatasphere.linkis.cs.common.entity.enumeration.ContextType; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKey; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKeyValue; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextValue; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.condition.Condition; +import com.webank.wedatasphere.linkis.cs.exception.ContextSearchFailedException; + +import java.util.List; +import java.util.Map; + +/** + * Created by patinousward on 2020/2/18. + */ +public abstract class ContextService extends AbstractService { + + public abstract ContextValue getContextValue(ContextID contextID, ContextKey contextKey); + + public abstract List<ContextKeyValue> searchContextValue(ContextID contextID, Map<Object, Object> conditionMap) throws ContextSearchFailedException; + + //public abstract List<ContextKeyValue> searchContextValueByCondition(Condition condition) throws ContextSearchFailedException; + + public abstract void setValueByKey(ContextID contextID, ContextKey contextKey, ContextValue contextValue) throws CSErrorException, ClassNotFoundException, JsonProcessingException; + + public abstract void setValue(ContextID contextID, ContextKeyValue contextKeyValuee) throws CSErrorException, ClassNotFoundException, JsonProcessingException; + + public abstract void resetValue(ContextID contextID, ContextKey contextKey) throws CSErrorException; + + public abstract void removeValue(ContextID contextID, ContextKey contextKey) throws CSErrorException; + + public abstract void removeAllValue(ContextID contextID) throws CSErrorException; + + public abstract void removeAllValueByKeyPrefixAndContextType(ContextID contextID, ContextType contextType,String keyPrefix) throws CSErrorException; + + public abstract void removeAllValueByKeyPrefix(ContextID contextID,String keyPrefix) throws CSErrorException; + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/Service.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/Service.java new file mode 100644 index 0000000000..187fce5905 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/Service.java @@ -0,0 +1,22 @@ +package com.webank.wedatasphere.linkis.cs.server.service; + +import com.webank.wedatasphere.linkis.cs.common.exception.CSWarnException; +import com.webank.wedatasphere.linkis.cs.server.scheduler.HttpJob; + +import java.io.Closeable; + +/** + * Created by patinousward on 2020/2/18. + */ +public interface Service extends Closeable { + + void init(); + + void start(); + + String getName(); + + boolean ifAccept(HttpJob job); + + void accept(HttpJob job) throws CSWarnException; +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextIDServiceImpl.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextIDServiceImpl.java new file mode 100644 index 0000000000..eeef6cd124 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextIDServiceImpl.java @@ -0,0 +1,64 @@ +package com.webank.wedatasphere.linkis.cs.server.service.impl; + +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.persistence.ContextPersistenceManager; +import com.webank.wedatasphere.linkis.cs.persistence.persistence.ContextIDPersistence; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; +import com.webank.wedatasphere.linkis.cs.server.service.ContextIDService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +/** + * Created by patinousward on 2020/2/21. + */ +@Component +public class ContextIDServiceImpl extends ContextIDService { + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + @Autowired + private ContextPersistenceManager persistenceManager; + + private ContextIDPersistence getPersistence() throws CSErrorException { + return persistenceManager.getContextIDPersistence(); + } + + @Override + public String getName() { + return ServiceType.CONTEXT_ID.name(); + } + + + @Override + public String createContextID(ContextID contextID) throws CSErrorException { + getPersistence().createContextID(contextID); + logger.info(String.format("createContextID,csId:%s", contextID.getContextId())); + return contextID.getContextId(); + } + + @Override + public ContextID getContextID(String id) throws CSErrorException { + logger.info(String.format("getContextID,csId:%s", id)); + return getPersistence().getContextID(id); + } + + @Override + public void updateContextID(ContextID contextID) throws CSErrorException { + logger.info(String.format("getContextID,csId:%s", contextID.getContextId())); + getPersistence().updateContextID(contextID); + } + + @Override + public void resetContextID(String id) throws CSErrorException { + // TODO: 2020/2/23 reset 方法 + } + + @Override + public void removeContextID(String id) throws CSErrorException { + logger.info(String.format("removeContextID,csId:%s", id)); + getPersistence().deleteContextID(id); + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextListenerServiceImpl.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextListenerServiceImpl.java new file mode 100644 index 0000000000..9e23eb74c8 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextListenerServiceImpl.java @@ -0,0 +1,81 @@ +package com.webank.wedatasphere.linkis.cs.server.service.impl; + +import com.webank.wedatasphere.linkis.cs.common.entity.listener.CommonContextKeyListenerDomain; +import com.webank.wedatasphere.linkis.cs.common.entity.listener.ContextIDListenerDomain; +import com.webank.wedatasphere.linkis.cs.common.entity.listener.ContextKeyListenerDomain; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKey; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.listener.callback.imp.ContextKeyValueBean; +import com.webank.wedatasphere.linkis.cs.listener.manager.imp.DefaultContextListenerManager; +import com.webank.wedatasphere.linkis.cs.persistence.ContextPersistenceManager; +import com.webank.wedatasphere.linkis.cs.persistence.persistence.ContextIDListenerPersistence; +import com.webank.wedatasphere.linkis.cs.persistence.persistence.ContextKeyListenerPersistence; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; +import com.webank.wedatasphere.linkis.cs.server.service.ContextListenerService; +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; + +/** + * Created by patinousward on 2020/2/21. + */ +@Component +public class ContextListenerServiceImpl extends ContextListenerService { + + @Autowired + private ContextPersistenceManager persistenceManager; + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + + private ContextIDListenerPersistence getIDListenerPersistence() throws CSErrorException { + return persistenceManager.getContextIDListenerPersistence(); + } + + private ContextKeyListenerPersistence getKeyListenerPersistence() throws CSErrorException { + return persistenceManager.getContextKeyListenerPersistence(); + } + + @Override + public String getName() { + return ServiceType.CONTEXT_LISTENER.name(); + } + + + @Override + public void onBind(ContextID contextID, ContextIDListenerDomain domain) throws CSErrorException { + logger.info(String.format("onBind,csId:%s", contextID.getContextId())); + domain.setContextID(contextID); + getIDListenerPersistence().create(contextID, domain); + DefaultContextListenerManager instance = DefaultContextListenerManager.getInstance(); + instance.getContextIDCallbackEngine().registerClient(domain); + } + + @Override + public void onBind(ContextID contextID, ContextKey contextKey, ContextKeyListenerDomain domain) throws CSErrorException { + logger.info(String.format("onBind,csId:%s,key:%s", contextID.getContextId(), contextKey.getKey())); + domain.setContextKey(contextKey); + // TODO: 2020/2/28 + if (domain instanceof CommonContextKeyListenerDomain) { + ((CommonContextKeyListenerDomain) domain).setContextID(contextID); + } + getKeyListenerPersistence().create(contextID, domain); + DefaultContextListenerManager instance = DefaultContextListenerManager.getInstance(); + instance.getContextKeyCallbackEngine().registerClient(domain); + } + + @Override + public List<ContextKeyValueBean> heartbeat(String clientSource) { + logger.info(String.format("heartbeat,clientSource:%s", clientSource)); + DefaultContextListenerManager instance = DefaultContextListenerManager.getInstance(); + ArrayList<ContextKeyValueBean> idCallback = instance.getContextIDCallbackEngine().getListenerCallback(clientSource); + ArrayList<ContextKeyValueBean> keyCallback = instance.getContextKeyCallbackEngine().getListenerCallback(clientSource); + idCallback.addAll(keyCallback); + return idCallback; + } +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextServiceImpl.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextServiceImpl.java new file mode 100644 index 0000000000..e2b678c38c --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/service/impl/ContextServiceImpl.java @@ -0,0 +1,218 @@ +package com.webank.wedatasphere.linkis.cs.server.service.impl; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.webank.wedatasphere.linkis.cs.ContextSearch; +import com.webank.wedatasphere.linkis.cs.DefaultContextSearch; +import com.webank.wedatasphere.linkis.cs.common.entity.enumeration.ContextType; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextID; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKey; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextKeyValue; +import com.webank.wedatasphere.linkis.cs.common.entity.source.ContextValue; +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.contextcache.ContextCacheService; +import com.webank.wedatasphere.linkis.cs.exception.ContextSearchFailedException; +import com.webank.wedatasphere.linkis.cs.persistence.ContextPersistenceManager; +import com.webank.wedatasphere.linkis.cs.persistence.entity.PersistenceContextKeyValue; +import com.webank.wedatasphere.linkis.cs.persistence.persistence.ContextMapPersistence; +import com.webank.wedatasphere.linkis.cs.server.enumeration.ServiceType; +import com.webank.wedatasphere.linkis.cs.server.parser.KeywordParser; +import com.webank.wedatasphere.linkis.cs.server.service.ContextService; +import com.webank.wedatasphere.linkis.server.BDPJettyServerHelper; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Created by patinousward on 2020/2/21. + */ +@Component +public class ContextServiceImpl extends ContextService { + + //对接search + + private ContextSearch contextSearch = new DefaultContextSearch(); + + @Autowired + private ContextCacheService contextCacheService; + + @Autowired + private ContextPersistenceManager persistenceManager; + + @Autowired + private KeywordParser keywordParser; + + + private ObjectMapper jackson = BDPJettyServerHelper.jacksonJson(); + + + private ContextMapPersistence getPersistence() throws CSErrorException { + return persistenceManager.getContextMapPersistence(); + } + + @Override + public String getName() { + return ServiceType.CONTEXT.name(); + } + + @Override + public ContextValue getContextValue(ContextID contextID, ContextKey contextKey) { + //从缓存取即可,缓存中无会去数据库中拉取 + ContextKeyValue keyValue = contextCacheService.get(contextID, contextKey); + if (keyValue == null) { + return null; + } + logger.info(String.format("getContextValue,csId:%s,key:%s,csType:%s,csScope:%s", + contextID.getContextId(), contextKey.getKey(), contextKey.getContextType(), contextKey.getContextScope())); + return keyValue.getContextValue(); + } + + @Override + public List<ContextKeyValue> searchContextValue(ContextID contextID, Map<Object, Object> conditionMap) throws ContextSearchFailedException { + //return List<ContextKeyValue> + logger.info(String.format("searchContextValue,csId:%s", contextID.getContextId())); + return contextSearch.search(contextCacheService, contextID, conditionMap); + } + +/* @Override + public List<ContextKeyValue> searchContextValueByCondition(Condition condition) throws ContextSearchFailedException { + //return List<ContextKeyValue> + return contextSearch.search(contextCacheService, null, condition); + }*/ + + @Override + public void setValueByKey(ContextID contextID, ContextKey contextKey, ContextValue contextValue) throws CSErrorException, ClassNotFoundException, JsonProcessingException { + //1.获取value + Object value = contextValue.getValue(); + //2.解析keywords,放入contextKey的keys中 + Set<String> keys = keywordParser.parse(value); + keys.add(contextKey.getKey()); + contextValue.setKeywords(jackson.writeValueAsString(keys)); + //3.缓存中是否有这个key,没有创建,有就更新 + ContextKeyValue keyValue = contextCacheService.get(contextID, contextKey); + if (keyValue == null) { + //创建校验scope和tye + if (contextKey.getContextScope() == null || contextKey.getContextType() == null) { + throw new CSErrorException(97000, "try to create context ,type or scope cannot be empty"); + } + //这里没有给出contextKeyvalue的具体实现,给个默认的persistence + logger.warn(String.format("setValueByKey, keyValue is not exist, csId:%s,key:%s", contextID.getContextId(), contextKey.getKey())); + keyValue = new PersistenceContextKeyValue(); + keyValue.setContextKey(contextKey); + keyValue.setContextValue(contextValue); + getPersistence().create(contextID, keyValue); + } else { + //update的话,如果scope和type 是空的,要用数据库中的值,因为update缓存要用到 + if (contextKey.getContextScope() == null) { + contextKey.setContextScope(keyValue.getContextKey().getContextScope()); + } + if (contextKey.getContextType() == null) { + contextKey.setContextType(keyValue.getContextKey().getContextType()); + } + keyValue.setContextKey(contextKey); + keyValue.setContextValue(contextValue); + getPersistence().update(contextID, keyValue); + } + //4.更新缓存 + contextCacheService.put(contextID, keyValue); + logger.info(String.format("setValueByKey, csId:%s,key:%s,keywords:%s", contextID.getContextId(), contextKey.getKey(), contextValue.getKeywords())); + } + + @Override + public void setValue(ContextID contextID, ContextKeyValue contextKeyValue) throws CSErrorException, ClassNotFoundException, JsonProcessingException { + //1.解析keywords + Object value = contextKeyValue.getContextValue().getValue(); + Set<String> keys = keywordParser.parse(value); + keys.add(contextKeyValue.getContextKey().getKey()); + contextKeyValue.getContextValue().setKeywords(jackson.writeValueAsString(keys)); + //2.数据库中是否有这个key,没有就创建,有就更新 + ContextKeyValue keyValue = contextCacheService.get(contextID, contextKeyValue.getContextKey()); + if (keyValue == null) { + logger.warn("cache can not find contextId:{},key:{},now try to load from MySQL", contextID.getContextId(), contextKeyValue.getContextKey().getKey()); + keyValue = getPersistence().get(contextID, contextKeyValue.getContextKey()); + if (keyValue != null) { + logger.warn("MySQL find the key,now reset the cache and get it"); + contextCacheService.rest(contextID, contextKeyValue.getContextKey()); + keyValue = contextCacheService.get(contextID, contextKeyValue.getContextKey()); + } + } + if (keyValue == null) { + if (contextKeyValue.getContextKey().getContextScope() == null || contextKeyValue.getContextKey().getContextType() == null) { + throw new CSErrorException(97000, "try to create context ,type or scope cannot be empty"); + } + getPersistence().create(contextID, contextKeyValue); + } else { + //update的话,如果scope和type 是空的,要用数据库中的值,因为update缓存要用到 + if (contextKeyValue.getContextKey().getContextScope() == null) { + contextKeyValue.getContextKey().setContextScope(keyValue.getContextKey().getContextScope()); + } + if (contextKeyValue.getContextKey().getContextType() == null) { + contextKeyValue.getContextKey().setContextType(keyValue.getContextKey().getContextType()); + } + getPersistence().update(contextID, contextKeyValue); + } + //3.更新缓存 + contextCacheService.put(contextID, contextKeyValue); + logger.info(String.format("setValue, csId:%s,key:%s", contextID.getContextId(), contextKeyValue.getContextKey().getKey())); + } + + @Override + public void resetValue(ContextID contextID, ContextKey contextKey) throws CSErrorException { + //1.reset 数据库 + getPersistence().reset(contextID, contextKey); + //2.reset 缓存 + contextCacheService.rest(contextID, contextKey); + logger.info(String.format("resetValue, csId:%s,key:%s", contextID.getContextId(), contextKey.getKey())); + } + + @Override + public void removeValue(ContextID contextID, ContextKey contextKey) throws CSErrorException { + ContextKeyValue contextKeyValue = contextCacheService.get(contextID, contextKey); + if (contextKeyValue == null) { + return; + } + //如果scope和type 不为空,则从原数据中进行赋值 + if (contextKey.getContextScope() == null) { + contextKey.setContextScope(contextKeyValue.getContextKey().getContextScope()); + } + if (contextKey.getContextType() == null) { + contextKey.setContextType(contextKeyValue.getContextKey().getContextType()); + } + //1.remove 数据库 + getPersistence().remove(contextID, contextKey); + //2.remove 缓存 + contextCacheService.remove(contextID, contextKey); + logger.info(String.format("removeValue, csId:%s,key:%s", contextID.getContextId(), contextKey.getKey())); + } + + @Override + public void removeAllValue(ContextID contextID) throws CSErrorException { + //移除数据库 + getPersistence().removeAll(contextID); + //移除缓存 + contextCacheService.removeAll(contextID); + + logger.info(String.format("removeAllValue, csId:%s", contextID.getContextId())); + } + + @Override + public void removeAllValueByKeyPrefixAndContextType(ContextID contextID, ContextType contextType, String keyPrefix) throws CSErrorException { + contextCacheService.removeByKeyPrefix(contextID, keyPrefix, contextType); + getPersistence().removeByKeyPrefix(contextID, contextType, keyPrefix); + logger.info(String.format("removeAllValueByKeyPrefixAndContextType, csId:%s,csType:%s,keyPrefix:%s", + contextID.getContextId(), contextType, keyPrefix)); + } + + @Override + public void removeAllValueByKeyPrefix(ContextID contextID, String keyPrefix) throws CSErrorException { + contextCacheService.removeByKeyPrefix(contextID, keyPrefix); + getPersistence().removeByKeyPrefix(contextID, keyPrefix); + logger.info(String.format("removeAllValueByKeyPrefix, csId:%s,keyPrefix:%s", + contextID.getContextId(), keyPrefix)); + } + + +} diff --git a/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/util/CsUtils.java b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/util/CsUtils.java new file mode 100644 index 0000000000..c71727e151 --- /dev/null +++ b/contextservice/cs-server/src/main/java/com/webank/wedatasphere/linkis/cs/server/util/CsUtils.java @@ -0,0 +1,21 @@ +package com.webank.wedatasphere.linkis.cs.server.util; + +import com.webank.wedatasphere.linkis.cs.common.exception.CSErrorException; +import com.webank.wedatasphere.linkis.cs.common.serialize.helper.ContextSerializationHelper; +import com.webank.wedatasphere.linkis.cs.common.serialize.helper.SerializationHelper; + +/** + * Created by patinousward on 2020/2/22. + */ +public class CsUtils { + + public static final SerializationHelper SERIALIZE = ContextSerializationHelper.getInstance(); + + public static String serialize(Object o) throws CSErrorException { + if (o instanceof String){ + return (String)o; + } + return SERIALIZE.serialize(o); + } + +} diff --git a/contextservice/cs-server/src/main/resources/application.yml b/contextservice/cs-server/src/main/resources/application.yml new file mode 100644 index 0000000000..b4d9145d8e --- /dev/null +++ b/contextservice/cs-server/src/main/resources/application.yml @@ -0,0 +1,22 @@ +server: + port: 9004 +spring: + application: + name: cloud-contextservice + + +eureka: + client: + serviceUrl: + defaultZone: http://127.0.0.1:20303/eureka/ + instance: + metadata-map: + test: wedatasphere + +management: + endpoints: + web: + exposure: + include: refresh,info +logging: + config: classpath:log4j2.xml diff --git a/contextservice/cs-server/src/main/resources/linkis.properties b/contextservice/cs-server/src/main/resources/linkis.properties new file mode 100644 index 0000000000..9abef042eb --- /dev/null +++ b/contextservice/cs-server/src/main/resources/linkis.properties @@ -0,0 +1,32 @@ +# +# 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.mybatis.datasource.url=jdbc:mysql://127.0.0.1:3306/ide_gz_bdap_sit_01?characterEncoding=UTF-8 +wds.linkis.server.mybatis.datasource.username= +wds.linkis.server.mybatis.datasource.password= + + +wds.linkis.log.clear=true +wds.linkis.server.version=v1 + +##restful +wds.linkis.server.restful.scan.packages=com.webank.wedatasphere.linkis.cs.server.restful + +##mybatis +wds.linkis.server.mybatis.mapperLocations=classpath*:com\\webank\\wedatasphere\\linkis\\cs\\persistence\\dao\\impl\\*.xml + +wds.linkis.server.mybatis.typeAliasesPackage=com.webank.wedatasphere.linkis.cs.persistence.entity + +wds.linkis.server.mybatis.BasePackage=com.webank.wedatasphere.linkis.cs.persistence.dao diff --git a/contextservice/cs-server/src/main/resources/log4j.properties b/contextservice/cs-server/src/main/resources/log4j.properties new file mode 100644 index 0000000000..0807e60877 --- /dev/null +++ b/contextservice/cs-server/src/main/resources/log4j.properties @@ -0,0 +1,37 @@ +# +# 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-server/src/main/resources/log4j2.xml b/contextservice/cs-server/src/main/resources/log4j2.xml new file mode 100644 index 0000000000..3923cd9f39 --- /dev/null +++ b/contextservice/cs-server/src/main/resources/log4j2.xml @@ -0,0 +1,39 @@ +<?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> + --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
