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 ab994be708203d1948ae0b3064bbf7ed4513044d Author: kantlin <[email protected]> AuthorDate: Fri Feb 28 22:51:58 2020 +0800 Fixed #301 --- datasource/metadatamanager/elasticsearch/pom.xml | 78 ++++++++++++ .../metadatamanager/service/ElasticConnection.java | 140 +++++++++++++++++++++ .../service/ElasticMetaService.java | 87 +++++++++++++ .../service/ElasticParamsMapper.java | 32 +++++ .../src/main/resources/application.yml | 23 ++++ .../src/main/resources/linkis.properties | 21 ++++ .../src/main/resources/log4j.properties | 33 +++++ .../elasticsearch/src/main/resources/log4j2.xml | 35 ++++++ .../service/receiver/ElasticReceiver.scala | 29 +++++ 9 files changed, 478 insertions(+) diff --git a/datasource/metadatamanager/elasticsearch/pom.xml b/datasource/metadatamanager/elasticsearch/pom.xml new file mode 100644 index 0000000000..4727619725 --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/pom.xml @@ -0,0 +1,78 @@ +<?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.3</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>linkis-metadatamanager-service-es</artifactId> + <properties> + <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> + <es.version>6.7.1</es.version> + </properties> + + <dependencies> + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-metadatamanager-common</artifactId> + <version>${linkis.version}</version> + </dependency> + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-module</artifactId> + <version>${linkis.version}</version> + <exclusions> + <exclusion> + <artifactId>asm</artifactId> + <groupId>org.ow2.asm</groupId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.elasticsearch.client</groupId> + <artifactId>elasticsearch-rest-client</artifactId> + <version>${es.version}</version> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-deploy-plugin</artifactId> + </plugin> + + <plugin> + <groupId>net.alchim31.maven</groupId> + <artifactId>scala-maven-plugin</artifactId> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-jar-plugin</artifactId> + </plugin> + </plugins> + <resources> + <resource> + <directory>src/main/resources</directory> + </resource> + </resources> + <finalName>${project.artifactId}-${project.version}</finalName> + </build> +</project> diff --git a/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticConnection.java b/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticConnection.java new file mode 100644 index 0000000000..b485c69a3a --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticConnection.java @@ -0,0 +1,140 @@ +/* + * 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. + */ + +package com.webank.wedatasphere.linkis.metadatamanager.service; + +import com.webank.wedatasphere.linkis.metadatamanager.common.Json; +import org.apache.commons.lang.StringUtils; +import org.apache.http.HttpHost; +import org.apache.http.auth.AuthScope; +import org.apache.http.auth.UsernamePasswordCredentials; +import org.apache.http.client.CredentialsProvider; +import org.apache.http.impl.client.BasicCredentialsProvider; +import org.apache.http.impl.nio.reactor.IOReactorConfig; +import org.elasticsearch.client.Request; +import org.elasticsearch.client.Response; +import org.elasticsearch.client.RestClient; +import org.elasticsearch.client.RestClientBuilder; + +import java.io.Closeable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * @author davidhua + * 2020/02/14 + */ +public class ElasticConnection implements Closeable { + + public static final String DEFAULT_TYPE_NAME = "type"; + + private static final String DEFAULT_MAPPING_NAME = "mappings"; + private static final String DEFAULT_INDEX_NAME = "index"; + private static final String FIELD_PROPS = "properties"; + + private RestClient restClient; + + public ElasticConnection(String[] endPoints, String username, String password) throws IOException { + HttpHost[] httpHosts = new HttpHost[endPoints.length]; + for(int i = 0; i < endPoints.length; i++){ + httpHosts[i] = HttpHost.create(endPoints[i]); + } + RestClientBuilder restClientBuilder = RestClient.builder(httpHosts); + CredentialsProvider credentialsProvider = null; + if(StringUtils.isNotBlank(username) && StringUtils.isNotBlank(password)){ + credentialsProvider = new BasicCredentialsProvider(); + credentialsProvider.setCredentials(AuthScope.ANY, + new UsernamePasswordCredentials(username, password)); + } + //set only one thread + CredentialsProvider finalCredentialsProvider = credentialsProvider; + restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> { + if(null != finalCredentialsProvider){ + httpClientBuilder + .setDefaultCredentialsProvider(finalCredentialsProvider); + } + return httpClientBuilder.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build()); + }); + this.restClient = restClientBuilder.build(); + //Try to test connection + ping(); + } + + public List<String> getAllIndices() throws IOException { + List<String> indices = new ArrayList<>(); + Request request = new Request("GET", "_cat/indices"); + request.addParameter("format", "JSON"); + Response response = restClient.performRequest(request); + List<Map<String, Object>> list = Json.fromJson(response.getEntity().getContent(), Map.class); + list.forEach( v ->{ + String index = String.valueOf(v.getOrDefault(DEFAULT_INDEX_NAME, "")); + if(StringUtils.isNotBlank(index) && !index.startsWith(".")){ + indices.add(index); + } + }); + return indices; + } + + public List<String> getTypes(String index) throws IOException{ + List<String> types = new ArrayList<>(); + Request request = new Request("GET", index +"/_mappings"); + Response response = restClient.performRequest(request); + Map<String, Map<String, Object>> result = + Json.fromJson(response.getEntity().getContent(), Map.class); + Map<String, Object> indexMap = result.get(index); + Object props = indexMap.get(DEFAULT_MAPPING_NAME); + if(props instanceof Map){ + Set keySet = ((Map)props).keySet(); + for(Object v : keySet){ + types.add(String.valueOf(v)); + } + } + return types; + } + + public Map<Object, Object> getProps(String index, String type) throws IOException{ + Request request = new Request("GET", index + "/_mappings/" + type); + Response response = restClient.performRequest(request); + Map<String, Map<String, Object>> result = + Json.fromJson(response.getEntity().getContent(), Map.class); + Map mappings = (Map)result.get(index).get("mappings"); + Map propsMap = mappings; + if(mappings.containsKey(type)){ + Object typeMap = mappings.get(type); + if(typeMap instanceof Map){ + propsMap = (Map)typeMap; + } + } + Object props = propsMap.get(FIELD_PROPS); + if(props instanceof Map){ + return (Map)props; + } + return null; + } + public void ping() throws IOException{ + Response response = restClient.performRequest(new Request("GET", "/")); + int code = response.getStatusLine().getStatusCode(); + int successCode = 200; + if(code != successCode){ + throw new RuntimeException("Ping to ElasticSearch ERROR, response code: " + code); + } + } + + @Override + public void close() throws IOException { + this.restClient.close(); + } +} diff --git a/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticMetaService.java b/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticMetaService.java new file mode 100644 index 0000000000..8778ba42eb --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticMetaService.java @@ -0,0 +1,87 @@ +/* + * 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. + */ + +package com.webank.wedatasphere.linkis.metadatamanager.service; + +import com.webank.wedatasphere.linkis.metadatamanager.common.Json; +import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaColumnInfo; +import com.webank.wedatasphere.linkis.metadatamanager.common.service.AbstractMetaService; +import com.webank.wedatasphere.linkis.metadatamanager.common.service.MetadataConnection; +import org.springframework.stereotype.Component; + +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** + * @author davidhua + * 2020/02/14 + */ +@Component +public class ElasticMetaService extends AbstractMetaService<ElasticConnection> { + @Override + public MetadataConnection<ElasticConnection> getConnection(String operator, Map<String, Object> params) + throws Exception{ + String[] endPoints = new String[]{}; + Object urls = params.get(ElasticParamsMapper.PARAM_ES_URLS.getValue()); + if(!(urls instanceof List)){ + List<String> urlList = Json.fromJson(String.valueOf(urls), List.class, String.class); + assert urlList != null; + endPoints = urlList.toArray(endPoints); + }else{ + endPoints = ((List<String>)urls).toArray(endPoints); + } + ElasticConnection conn = new ElasticConnection(endPoints, String.valueOf(params.getOrDefault(ElasticParamsMapper.PARAM_ES_USERNAME.getValue(), "")), + String.valueOf(params.getOrDefault(ElasticParamsMapper.PARAM_ES_PASSWORD.getValue(), ""))); + return new MetadataConnection<>(conn, false); + } + + @Override + public List<String> queryDatabases(ElasticConnection connection) { + //Get indices + try{ + return connection.getAllIndices(); + }catch (Exception e){ + throw new RuntimeException("Fail to get ElasticSearch indices(获取索引列表失败)", e); + } + } + + @Override + public List<String> queryTables(ElasticConnection connection, String database) { + //Get types + try{ + return connection.getTypes(database); + }catch (Exception e){ + throw new RuntimeException("Fail to get ElasticSearch types(获取索引类型失败)", e); + } + } + + @Override + public List<MetaColumnInfo> queryColumns(ElasticConnection connection, String database, String table) { + try { + Map<Object, Object> props = connection.getProps(database, table); + return props.entrySet().stream().map(entry -> { + MetaColumnInfo info = new MetaColumnInfo(); + info.setName(String.valueOf(entry.getKey())); + Object value = entry.getValue(); + if(value instanceof Map){ + info.setType(String.valueOf(((Map)value) + .getOrDefault(ElasticConnection.DEFAULT_TYPE_NAME, ""))); + } + return info; + }).collect(Collectors.toList()); + } catch (Exception e) { + throw new RuntimeException("Fail to get ElasticSearch columns(获取索引字段失败)", e); + } + } +} diff --git a/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticParamsMapper.java b/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticParamsMapper.java new file mode 100644 index 0000000000..3af7e8ad86 --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/service/ElasticParamsMapper.java @@ -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. + */ + +package com.webank.wedatasphere.linkis.metadatamanager.service; + +import com.webank.wedatasphere.linkis.common.conf.CommonVars; + +/** + * Configuration + * @author davidhua + * 2020/02/14 + */ +public class ElasticParamsMapper { + public static final CommonVars<String> PARAM_ES_URLS = + CommonVars.apply("wds.linkis.server.mdm.service.es.urls", "elasticUrls"); + + public static final CommonVars<String> PARAM_ES_USERNAME = + CommonVars.apply("wds.linkis.server.mdm.service.es.username", "username"); + + public static final CommonVars<String> PARAM_ES_PASSWORD = + CommonVars.apply("wds.linkis.server.mdm.service.es.password", "password"); +} diff --git a/datasource/metadatamanager/elasticsearch/src/main/resources/application.yml b/datasource/metadatamanager/elasticsearch/src/main/resources/application.yml new file mode 100644 index 0000000000..8b3f9ce44a --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/src/main/resources/application.yml @@ -0,0 +1,23 @@ +server: + port: 8295 +spring: + application: + name: mdm-service-elasticsearch + +eureka: + client: + serviceUrl: + defaultZone: http://${ip}:${port}/eureka/ + instance: + metadata-map: + test: wedatasphere + +management: + endpoints: + web: + exposure: + include: refresh,info +logging: + config: classpath:log4j2.xml + + diff --git a/datasource/metadatamanager/elasticsearch/src/main/resources/linkis.properties b/datasource/metadatamanager/elasticsearch/src/main/resources/linkis.properties new file mode 100644 index 0000000000..f83c10d281 --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/src/main/resources/linkis.properties @@ -0,0 +1,21 @@ +# +# 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.mapperLocations= +wds.linkis.server.mybatis.typeAliasesPackage= +wds.linkis.server.mybatis.BasePackage= +wds.linkis.server.restful.scan.packages= + +#sit +wds.linkis.server.version=v1 + diff --git a/datasource/metadatamanager/elasticsearch/src/main/resources/log4j.properties b/datasource/metadatamanager/elasticsearch/src/main/resources/log4j.properties new file mode 100644 index 0000000000..d5ee44b86b --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/src/main/resources/log4j.properties @@ -0,0 +1,33 @@ +# +# 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 \ No newline at end of file diff --git a/datasource/metadatamanager/elasticsearch/src/main/resources/log4j2.xml b/datasource/metadatamanager/elasticsearch/src/main/resources/log4j2.xml new file mode 100644 index 0000000000..1c68190669 --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/src/main/resources/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/datasource/metadatamanager/elasticsearch/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/service/receiver/ElasticReceiver.scala b/datasource/metadatamanager/elasticsearch/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/service/receiver/ElasticReceiver.scala new file mode 100644 index 0000000000..dd3049f1de --- /dev/null +++ b/datasource/metadatamanager/elasticsearch/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/service/receiver/ElasticReceiver.scala @@ -0,0 +1,29 @@ +/* + * 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. + */ + +package com.webank.wedatasphere.linkis.metadatamanager.service.receiver + +import com.webank.wedatasphere.linkis.DataWorkCloudApplication +import com.webank.wedatasphere.linkis.metadatamanager.common.service.MetadataService +import com.webank.wedatasphere.linkis.metadatamanager.common.receiver.BaseMetaReceiver +import com.webank.wedatasphere.linkis.metadatamanager.common.service.MetadataService +import javax.annotation.PostConstruct +import org.springframework.stereotype.Component + +@Component +class ElasticReceiver extends BaseMetaReceiver{ + @PostConstruct + def init(): Unit = { + metadataService = DataWorkCloudApplication.getApplicationContext.getBean(classOf[MetadataService]) + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
