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 21b5f2223dfe273b884232250c6ea8b84c596df5 Author: xj2jx <[email protected]> AuthorDate: Sun Feb 23 16:04:15 2020 +0800 Provide a metadata management common module close #302 --- datasource/metadatamanager/common/pom.xml | 74 +++++++ .../linkis/metadatamanager/common/Json.java | 111 +++++++++++ .../metadatamanager/common/MdmConfiguration.java | 28 +++ .../common/cache/CacheConfiguration.java | 28 +++ .../metadatamanager/common/cache/CacheManager.java | 42 ++++ .../common/cache/ConnCacheManager.java | 56 ++++++ .../common/domain/MetaColumnInfo.java | 64 +++++++ .../common/domain/MetaPartitionInfo.java | 93 +++++++++ .../common/exception/MetaRuntimeException.java | 31 +++ .../common/service/AbstractMetaService.java | 213 +++++++++++++++++++++ .../common/service/BaseMetadataService.java | 32 ++++ .../common/service/MetadataConnection.java | 55 ++++++ .../common/service/MetadataDbService.java | 67 +++++++ .../common/service/MetadataService.java | 20 ++ .../common/protocol/MetadataOperateProtocol.scala | 27 +++ .../common/protocol/MetadataProtocol.scala | 20 ++ .../common/protocol/MetadataQueryProtocol.scala | 57 ++++++ .../common/receiver/BaseMetaReceiver.scala | 63 ++++++ 18 files changed, 1081 insertions(+) diff --git a/datasource/metadatamanager/common/pom.xml b/datasource/metadatamanager/common/pom.xml new file mode 100644 index 0000000000..7f0c5e981e --- /dev/null +++ b/datasource/metadatamanager/common/pom.xml @@ -0,0 +1,74 @@ +<?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-common</artifactId> + + <properties> + <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> + </properties> + + <dependencies> + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-common</artifactId> + <version>${linkis.version}</version> + <exclusions> + <exclusion> + <artifactId>asm</artifactId> + <groupId>org.ow2.asm</groupId> + </exclusion> + </exclusions> + </dependency> + <!--rpc--> + <dependency> + <groupId>com.webank.wedatasphere.linkis</groupId> + <artifactId>linkis-cloudRPC</artifactId> + <version>${linkis.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/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/Json.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/Json.java new file mode 100644 index 0000000000..203c4a1409 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/Json.java @@ -0,0 +1,111 @@ +/* + * 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.common; + +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.*; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.BufferedReader; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.util.ArrayList; + +/** + * Json utils + * Created by jackyxxie on 2018/9/3. + */ +public class Json { + private static final String PREFIX = "["; + private static final String SUFFIX = "]"; + private static final Logger logger = LoggerFactory.getLogger(Json.class); + + private static ObjectMapper mapper; + + static{ + mapper = new ObjectMapper(); + mapper.configure(JsonParser.Feature.ALLOW_COMMENTS, true); + mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true); + mapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true); + mapper.configure(DeserializationFeature.READ_ENUMS_USING_TO_STRING, true); + mapper.configure(SerializationFeature.WRITE_ENUMS_USING_TO_STRING, true); + mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true); + //empty beans allowed + mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false); + //ignore unknown properties + mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); + //cancel to scape non ascii + mapper.configure(JsonGenerator.Feature.ESCAPE_NON_ASCII, false); + } + private Json(){} + + @SuppressWarnings("unchecked") + public static <T> T fromJson(String json, Class<?> clazz, Class<?>... parameters){ + if(StringUtils.isNotBlank(json)){ + try{ + if(parameters.length > 0){ + return (T)mapper.readValue(json, mapper.getTypeFactory().constructParametricType(clazz, parameters)); + } + if(json.startsWith(PREFIX) + && json.endsWith(SUFFIX)){ + JavaType javaType = mapper.getTypeFactory() + .constructParametricType(ArrayList.class, clazz); + return mapper.readValue(json, javaType); + } + return (T)mapper.readValue(json, clazz); + } catch (Exception e) { + logger.info(e.getLocalizedMessage()); + throw new RuntimeException(e); + } + } + return null; + } + + public static <T> T fromJson(InputStream stream, Class<?> clazz, Class<?>... parameters){ + StringBuilder builder = new StringBuilder(); + String jsonStr = null; + try{ + BufferedReader reader = new BufferedReader(new InputStreamReader(stream, "UTF-8")); + while((jsonStr = reader.readLine()) != null){ + builder.append(jsonStr); + } + reader.close(); + }catch(Exception e){ + logger.info(e.getLocalizedMessage()); + throw new RuntimeException(e); + } + return fromJson(builder.toString(), clazz, parameters); + } + + public static String toJson(Object obj, Class<?> model){ + ObjectWriter writer = mapper.writer(); + if(null != obj){ + try{ + if(null != model){ + writer = writer.withView(model); + } + return writer.writeValueAsString(obj); + } catch (JsonProcessingException e) { + logger.info(e.getLocalizedMessage()); + throw new RuntimeException(e); + } + } + return null; + } + +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/MdmConfiguration.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/MdmConfiguration.java new file mode 100644 index 0000000000..480b36d42d --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/MdmConfiguration.java @@ -0,0 +1,28 @@ +/* + * 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.common; + +import com.webank.wedatasphere.linkis.common.conf.CommonVars; + +/** + * Created by jackyxxie on 2020/2/10. + */ +public class MdmConfiguration { + + public static CommonVars<String> METADATA_SERVICE_APPLICATION = + CommonVars.apply("wds.linkis.server.mdm.service.app.name", "mdm-service"); + + public static CommonVars<String> DATA_SOURCE_SERVICE_APPLICATION = + CommonVars.apply("wds.linkis.server.dsm.app.name", "dsm-server"); +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheConfiguration.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheConfiguration.java new file mode 100644 index 0000000000..01d083dc66 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheConfiguration.java @@ -0,0 +1,28 @@ +/* + * 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.common.cache; + +import com.webank.wedatasphere.linkis.common.conf.CommonVars; + +/** + * Created by jackyxxie on 2020/02/14. + */ +public class CacheConfiguration { + + public static CommonVars<Long> CACHE_MAX_SIZE = + CommonVars.apply("wds.linkis.server.mdm.service.cache.size", 1000L); + + public static CommonVars<Long> CACHE_EXPIRE_TIME = + CommonVars.apply("wds.linkis.server.mdm.service.cache.expire", 600L); +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheManager.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheManager.java new file mode 100644 index 0000000000..a1c608a931 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/CacheManager.java @@ -0,0 +1,42 @@ +/* + * 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.common.cache; + + +import com.google.common.cache.Cache; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; +import com.google.common.cache.RemovalListener; + +/** + * Created by jackyxxie on 2020/2/14. + */ +public interface CacheManager { + /** + * build simple cache + * @param cacheId + * @param removalListener + * @return + */ + <V> Cache<String, V> buildCache(String cacheId, RemovalListener<String, V> removalListener); + + /** + * build loading cache + * @param cacheId + * @param loader + * @param removalListener + * @return + */ + <V> LoadingCache<String, V> buildCache(String cacheId, CacheLoader<String, V> loader, RemovalListener<String, V> removalListener); +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/ConnCacheManager.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/ConnCacheManager.java new file mode 100644 index 0000000000..7d96236149 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/cache/ConnCacheManager.java @@ -0,0 +1,56 @@ +/* + * 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.common.cache; + +import com.google.common.cache.*; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; + +/** + * Created by jackyxxie on 2020/02/14. + */ +public class ConnCacheManager implements CacheManager { + private ConcurrentHashMap<String, Cache> cacheStore = new ConcurrentHashMap<>(); + private ConnCacheManager(){ + + } + + public static CacheManager custom(){ + return new ConnCacheManager(); + } + + @Override + public <V> Cache<String, V> buildCache(String cacheId, RemovalListener<String, V> removalListener) { + Cache<String, V> vCache = CacheBuilder.newBuilder() + .maximumSize(CacheConfiguration.CACHE_MAX_SIZE.getValue()) + .expireAfterWrite(CacheConfiguration.CACHE_EXPIRE_TIME.getValue(), TimeUnit.SECONDS) + .removalListener(removalListener) + .build(); + cacheStore.putIfAbsent(cacheId, vCache); + return vCache; + } + + @Override + public <V> LoadingCache<String, V> buildCache(String cacheId, CacheLoader<String, V> loader, + RemovalListener<String, V> removalListener) { + LoadingCache<String, V> vCache = CacheBuilder.newBuilder() + .maximumSize(CacheConfiguration.CACHE_MAX_SIZE.getValue()) + .expireAfterWrite(CacheConfiguration.CACHE_EXPIRE_TIME.getValue(), TimeUnit.SECONDS) + .removalListener(removalListener) + .build(loader); + cacheStore.putIfAbsent(cacheId, vCache); + return vCache; + } +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaColumnInfo.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaColumnInfo.java new file mode 100644 index 0000000000..60911a8b13 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaColumnInfo.java @@ -0,0 +1,64 @@ +/* + * 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.common.domain; + +import org.codehaus.jackson.annotate.JsonIgnoreProperties; +import org.codehaus.jackson.map.annotate.JsonSerialize; + +import java.io.Serializable; + +/** + * The meta information of field + * Created by jackyxxie on 2020/2/10. + */ +@JsonSerialize(include= JsonSerialize.Inclusion.NON_EMPTY) +@JsonIgnoreProperties(ignoreUnknown = true) +public class MetaColumnInfo implements Serializable { + private int index = -1; + private boolean isPrimaryKey; + private String name; + private String type; + + public int getIndex() { + return index; + } + + public void setIndex(int index) { + this.index = index; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public boolean isPrimaryKey() { + return isPrimaryKey; + } + + public void setPrimaryKey(boolean primaryKey) { + isPrimaryKey = primaryKey; + } +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaPartitionInfo.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaPartitionInfo.java new file mode 100644 index 0000000000..d4286c4c60 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/domain/MetaPartitionInfo.java @@ -0,0 +1,93 @@ +/* + * 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.common.domain; + +import org.codehaus.jackson.annotate.JsonIgnoreProperties; +import org.codehaus.jackson.map.annotate.JsonSerialize; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * The meta information of partition + * Created by jackyxxie on 2020/2/10. + */ +@JsonSerialize(include= JsonSerialize.Inclusion.NON_EMPTY) +@JsonIgnoreProperties(ignoreUnknown = true) +public class MetaPartitionInfo implements Serializable { + private List<String> partKeys = new ArrayList<>(); + + private String name; + /** + * Partition tree + */ + private PartitionNode root; + + @JsonSerialize(include= JsonSerialize.Inclusion.NON_EMPTY) + @JsonIgnoreProperties(ignoreUnknown = true) + public static class PartitionNode{ + /** + * Node name + */ + private String name; + /** + * Key: partition value + * Value: child partition node + */ + private Map<String, PartitionNode> partitions = new HashMap<>(); + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public Map<String, PartitionNode> getPartitions() { + return partitions; + } + + public void setPartitions(Map<String, PartitionNode> partitions) { + this.partitions = partitions; + } + } + + public List<String> getPartKeys() { + return partKeys; + } + + public void setPartKeys(List<String> partKeys) { + this.partKeys = partKeys; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public PartitionNode getRoot() { + return root; + } + + public void setRoot(PartitionNode root) { + this.root = root; + } +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/exception/MetaRuntimeException.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/exception/MetaRuntimeException.java new file mode 100644 index 0000000000..e3dcb0ef36 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/exception/MetaRuntimeException.java @@ -0,0 +1,31 @@ +/* + * 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.common.exception; + +import com.webank.wedatasphere.linkis.common.exception.WarnException; + +/** + * Created by jackyxxie on 2020/2/10. + */ +public class MetaRuntimeException extends WarnException { + private static final int ERROR_CODE = 99900; + public MetaRuntimeException(String desc) { + super(ERROR_CODE, desc); + } + + public MetaRuntimeException(String desc, String ip, int port, String serviceKind) { + super(ERROR_CODE, desc, ip, port, serviceKind); + } + +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/AbstractMetaService.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/AbstractMetaService.java new file mode 100644 index 0000000000..744440f32b --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/AbstractMetaService.java @@ -0,0 +1,213 @@ +/* + * 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.common.service; + +import com.google.common.cache.Cache; +import com.webank.wedatasphere.linkis.common.exception.WarnException; +import com.webank.wedatasphere.linkis.metadatamanager.common.Json; +import com.webank.wedatasphere.linkis.metadatamanager.common.cache.CacheManager; +import com.webank.wedatasphere.linkis.metadatamanager.common.cache.ConnCacheManager; +import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaColumnInfo; +import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaPartitionInfo; +import com.webank.wedatasphere.linkis.metadatamanager.common.exception.MetaRuntimeException; +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.PostConstruct; +import java.io.Closeable; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.util.List; +import java.util.Map; +import java.util.function.Function; + +/** + * Created by jackyxxie on 2020/2/10. + */ +public abstract class AbstractMetaService<C extends Closeable> implements MetadataService { + private static final Logger LOG = LoggerFactory.getLogger(AbstractMetaService.class); + private static final String CONN_CACHE_REQ = "_STORED"; + + + private CacheManager connCacheManager; + + /** + * Caching connections which built by connect parameters requested with + */ + protected Cache<String, MetadataConnection<C>> reqCache; + + @PostConstruct + public void init(){ + connCacheManager = ConnCacheManager.custom(); + initCache(connCacheManager); + } + /** + * If want to use cache component, you should invoke this in constructor method + * @param cacheManager + */ + protected void initCache(CacheManager cacheManager){ + String prefix = this.getClass().getSimpleName(); + reqCache = cacheManager.buildCache(prefix + CONN_CACHE_REQ, notification ->{ + assert notification.getValue() != null; + close(notification.getValue().getConnection()); + }); + } + + @Override + public abstract MetadataConnection<C> getConnection(String operator, Map<String, Object> params) throws Exception; + + @Override + public List<String> getDatabases(String operator, Map<String, Object> params) { + return this.getConnAndRun(operator, params, this::queryDatabases); + } + + @Override + public List<String> getTables(String operator, Map<String, Object> params, String database) { + return this.getConnAndRun(operator, params, conn -> this.queryTables(conn, database)); + } + + @Override + public Map<String, String> getTableProps(String operator, Map<String, Object> params, String database, String table) { + return this.getConnAndRun(operator, params, conn -> this.queryTableProps(conn, database, table)); + } + + @Override + public MetaPartitionInfo getPartitions(String operator, Map<String, Object> params, String database, String table) { + return this.getConnAndRun(operator, params, conn -> this.queryPartitions(conn, database, table)); + } + + @Override + public List<MetaColumnInfo> getColumns(String operator, Map<String, Object> params, String database, String table) { + return this.getConnAndRun(operator, params, conn -> this.queryColumns(conn, database, table)); + } + + /** + * Get database list by connection + * @param connection metadata connection + * @return + */ + public List<String> queryDatabases(C connection){ + throw new WarnException(-1, "This method is no supported"); + } + + /** + * Get table list by connection and database + * @param connection metadata connection + * @param database database + * @return + */ + public List<String> queryTables(C connection, String database){ + throw new WarnException(-1, "This method is no supported"); + } + + /** + * Get partitions by connection, database and table + * @param connection metadata connection + * @param database database + * @param table table + * @return + */ + public MetaPartitionInfo queryPartitions(C connection, String database, String table){ + throw new WarnException(-1, "This method is no supported"); + } + + /** + * Get columns by connection, database and table + * @param connection metadata connection + * @param database database + * @param table table + * @return + */ + public List<MetaColumnInfo> queryColumns(C connection, String database, String table){ + throw new WarnException(-1, "This method is no supported"); + } + + /** + * Get table properties + * @param connection metadata connection + * @param database database + * @param table table + * @return + */ + public Map<String, String> queryTableProps(C connection, String database, String table){ + throw new WarnException(-1, "This method is no supported"); + } + + protected void close(C connection){ + try { + connection.close(); + } catch (IOException e) { + throw new MetaRuntimeException("Fail to close connection[关闭连接失败], [" + e.getMessage() + "]"); + } + } + + protected <R>R getConnAndRun(String operator, Map<String, Object> params, Function<C, R> action) { + String cacheKey = ""; + try{ + cacheKey = md5String(Json.toJson(params, null), "", 2); + MetadataConnection<C> connection; + if(null != reqCache) { + connection = reqCache.get(cacheKey, () -> getConnection(operator, params)); + }else{ + connection = getConnection(operator, params); + } + return run(connection, action); + }catch(Exception e){ + LOG.error("Error to invoke meta service", e); + if(StringUtils.isNotBlank(cacheKey)){ + reqCache.invalidate(cacheKey); + } + throw new MetaRuntimeException(e.getMessage()); + } + } + private <R>R run(MetadataConnection<C> connection, Function<C, R> action){ + if(connection.isLock()){ + connection.getLock().lock(); + try{ + return action.apply(connection.getConnection()); + }finally{ + connection.getLock().unlock(); + } + }else{ + return action.apply(connection.getConnection()); + } + } + + private String md5String(String source, String salt, int iterator){ + StringBuilder token = new StringBuilder(); + try{ + MessageDigest digest = MessageDigest.getInstance("md5"); + if(StringUtils.isNotEmpty(salt)){ + digest.update(salt.getBytes(StandardCharsets.UTF_8)); + } + byte[] result = digest.digest(source.getBytes()); + for(int i = 0; i < iterator - 1; i++){ + digest.reset(); + result = digest.digest(result); + } + for (byte aResult : result) { + int temp = aResult & 0xFF; + if (temp <= 0xF) { + token.append("0"); + } + token.append(Integer.toHexString(temp)); + } + }catch(Exception e){ + throw new RuntimeException(e.getMessage()); + } + return token.toString(); + } +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/BaseMetadataService.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/BaseMetadataService.java new file mode 100644 index 0000000000..e9146d4059 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/BaseMetadataService.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.common.service; + + +import java.io.Closeable; +import java.util.Map; + +/** + * Created by jackyxxie on 2020/02/10. + */ +public interface BaseMetadataService { + + /** + * Get connection + * + * @param params connect params + * @return + */ + MetadataConnection<? extends Closeable> getConnection(String operator, Map<String, Object> params) throws Exception; +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataConnection.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataConnection.java new file mode 100644 index 0000000000..80a2d9d720 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataConnection.java @@ -0,0 +1,55 @@ +/* + * 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.common.service; + +import java.util.concurrent.locks.ReentrantLock; + +/** + * Connection for metadata + * Created by jackyxxie on 2020/2/10. + */ +public class MetadataConnection<C> { + /*** + * Avoid complication when different threads calling connection API + */ + private ReentrantLock lock = new ReentrantLock(); + + private boolean isLock = true; + + private C connection; + + public MetadataConnection(C connection){ + this.connection = connection; + } + + public MetadataConnection(C connection, boolean isLock){ + this.connection = connection; + this.isLock = isLock; + } + public ReentrantLock getLock() { + return lock; + } + + public C getConnection() { + return connection; + } + + public boolean isLock() { + return isLock; + } + + public void setLock(boolean lock) { + isLock = lock; + } +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataDbService.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataDbService.java new file mode 100644 index 0000000000..b4de097080 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataDbService.java @@ -0,0 +1,67 @@ +/* + * 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.common.service; + +import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaColumnInfo; +import com.webank.wedatasphere.linkis.metadatamanager.common.domain.MetaPartitionInfo; + +import java.util.List; +import java.util.Map; + +/** + * Created by jackyxxie on 2020/2/10. + */ +public interface MetadataDbService extends BaseMetadataService { + + /** + * Get all databases + * @param params connect params + * @return + */ + List<String> getDatabases(String operator, Map<String, Object> params); + + /** + * Get all tables from database specified + * @param params params + * @param database database name + * @return + */ + List<String> getTables(String operator, Map<String, Object> params, String database); + + /** + * Get table properties from database specified + * @param params params + * @param database database name + * @return + */ + Map<String, String> getTableProps(String operator, Map<String, Object> params, String database, String table); + /** + * Get all partitions from table specified + * @param params params + * @param database + * @param table + * @return + */ + MetaPartitionInfo getPartitions(String operator, Map<String, Object> params, String database, String table); + + /** + * Get all field information from table specified + * @param params + * @param database + * @param table + * @return + */ + List<MetaColumnInfo> getColumns(String operator, Map<String, Object> params, String database, String table); + +} diff --git a/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataService.java b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataService.java new file mode 100644 index 0000000000..3152182514 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/java/com/webank/wedatasphere/linkis/metadatamanager/common/service/MetadataService.java @@ -0,0 +1,20 @@ +/* + * 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.common.service; + +/** + * Created by jackyxxie on 2020/2/10. + */ +public interface MetadataService extends MetadataDbService { +} diff --git a/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataOperateProtocol.scala b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataOperateProtocol.scala new file mode 100644 index 0000000000..834dd268ba --- /dev/null +++ b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataOperateProtocol.scala @@ -0,0 +1,27 @@ +/* + * 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.common.protocol + +import java.util + +trait MetadataOperateProtocol { + +} + +/** + * Request to do connect + * @param version + * @param params + */ +case class MetadataConnect(operator: String, params: util.Map[String, Object], version: String) extends MetadataOperateProtocol diff --git a/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataProtocol.scala b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataProtocol.scala new file mode 100644 index 0000000000..38940e24a6 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataProtocol.scala @@ -0,0 +1,20 @@ +/* + * 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.common.protocol + +trait MetadataProtocol { + +} + +case class MetadataResponse(status: Boolean, data: String) extends MetadataProtocol diff --git a/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataQueryProtocol.scala b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataQueryProtocol.scala new file mode 100644 index 0000000000..2668989c90 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/protocol/MetadataQueryProtocol.scala @@ -0,0 +1,57 @@ +/* + * 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.common.protocol + +import java.util + +trait MetadataQueryProtocol { + +} + +/** + * Request to get database list + * @param params + */ +case class MetaGetDatabases(params: util.Map[String, Object], operator: String) extends MetadataQueryProtocol + +/** + * Request to get table list + * @param params + */ +case class MetaGetTables(params: util.Map[String, Object], database: String, operator: String) extends MetadataQueryProtocol + +/** + * Request to get table properties + * @param params + * @param database + * @param table + */ +case class MetaGetTableProps(params: util.Map[String, Object], database: String, table: String, operator: String) extends MetadataQueryProtocol + +/** + * Request to get partition list + * @param params + * @param database + * @param table + */ +case class MetaGetPartitions(params: util.Map[String, Object], database: String, table: String, operator: String) extends MetadataQueryProtocol + +/** + * Request to get column list + * @param params + * @param database + * @param table + */ +case class MetaGetColumns(params: util.Map[String, Object], database: String, table: String, operator: String) extends MetadataQueryProtocol + diff --git a/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/receiver/BaseMetaReceiver.scala b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/receiver/BaseMetaReceiver.scala new file mode 100644 index 0000000000..d4ecf7d899 --- /dev/null +++ b/datasource/metadatamanager/common/src/main/scala/com/webank/wedatasphere/linkis/metadatamanager/common/receiver/BaseMetaReceiver.scala @@ -0,0 +1,63 @@ +/* + * 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.common.receiver + +import com.webank.wedatasphere.linkis.common.exception.WarnException +import com.webank.wedatasphere.linkis.common.utils.{Logging, Utils} +import com.webank.wedatasphere.linkis.metadatamanager.common.protocol._ +import com.webank.wedatasphere.linkis.metadatamanager.common.service.MetadataService +import com.webank.wedatasphere.linkis.rpc.{Receiver, Sender} +import com.webank.wedatasphere.linkis.server.BDPJettyServerHelper + +import scala.concurrent.duration.Duration + +/** + * Created by jackyxxie on 2020/2/10. + */ +class BaseMetaReceiver extends Receiver with Logging{ + protected var metadataService: MetadataService = _ + + override def receive(message: Any, sender: Sender): Unit = ??? + + override def receiveAndReply(message: Any, sender: Sender): Any = invoke(metadataService, message) + + override def receiveAndReply(message: Any, duration: Duration, sender: Sender): Any = invoke(metadataService, message) + + + def invoke(service: MetadataService, message: Any): Any = Utils.tryCatch{ + val data = message match { + case MetaGetDatabases(params, operator) => service.getDatabases(operator, params) + case MetaGetTableProps(params, database, table, operator) => service.getTableProps(operator, params, database, table) + case MetaGetTables(params, database, operator) => service.getTables(operator, params, database) + case MetaGetPartitions(params, database, table, operator) => service.getPartitions(operator, params, database, table) + case MetaGetColumns(params, database, table, operator) => service.getColumns(operator, params, database, table) + case MetadataConnect(operator, params, version) => + service.getConnection(operator, params) + //MetadataConnection is not scala class + null + case _ => new Object() + } + MetadataResponse(status = true, BDPJettyServerHelper.gson.toJson(data)) + }{ + case e:WarnException => val errorMsg = e.getMessage + info(s"Fail to invoke meta service: [$message],[$errorMsg]") + MetadataResponse(status = false, errorMsg) + case t:Throwable => + val errorMsg = t.getMessage + if (message.isInstanceOf[MetadataConnect]) + info(s"Fail to invoke meta service: [$message], [$errorMsg]") + else error(s"Fail to invoke meta service", t) + MetadataResponse(status = false, t.getMessage) + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
