[inlong] branch master updated: [INLONG-6497][Manager] Support Elasticsearch cluster (#6501)

[he...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

healchow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git


The following commit(s) were added to refs/heads/master by this push:
     new c1bc689f3 [INLONG-6497][Manager] Support Elasticsearch cluster (#6501)
c1bc689f3 is described below

commit c1bc689f320a394ddc308b80d2971c9867e7744c
Author: vernedeng <de...@pku.edu.cn>
AuthorDate: Sun Nov 20 16:34:39 2022 +0800

    [INLONG-6497][Manager] Support Elasticsearch cluster (#6501)
---
 .../inlong/manager/common/enums/ClusterType.java   |  2 +
 .../pojo/cluster/es/ElasticsearchClusterDTO.java   | 57 ++++++++++++++++
 .../pojo/cluster/es/ElasticsearchClusterInfo.java  | 50 ++++++++++++++
 .../cluster/es/ElasticsearchClusterRequest.java    | 41 +++++++++++
 .../cluster/ElasticsearchClusterOperator.java      | 79 ++++++++++++++++++++++
 5 files changed, 229 insertions(+)

diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ClusterType.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ClusterType.java
index d2934fc6c..c9fa9db82 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ClusterType.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ClusterType.java
@@ -32,6 +32,7 @@ public class ClusterType {
     public static final String PULSAR = "PULSAR";
     public static final String DATAPROXY = "DATAPROXY";
     public static final String KAFKA = "KAFKA";
+    public static final String ELASTICSEARCH = "ELASTICSEARCH";
 
     private static final Set<String> TYPE_SET = new HashSet<String>() {
         {
@@ -40,6 +41,7 @@ public class ClusterType {
             add(ClusterType.PULSAR);
             add(ClusterType.DATAPROXY);
             add(ClusterType.KAFKA);
+            add(ClusterType.ELASTICSEARCH);
         }
     };
 
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterDTO.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterDTO.java
new file mode 100644
index 000000000..5298137ce
--- /dev/null
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterDTO.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.inlong.manager.pojo.cluster.es;
+
+import io.swagger.annotations.ApiModel;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+import org.apache.inlong.manager.common.util.JsonUtils;
+
+import javax.validation.constraints.NotNull;
+
+/**
+ * Elasticsearch cluster info
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@ApiModel("Elasticsearch cluster info")
+public class ElasticsearchClusterDTO {
+
+    /**
+     * Get the dto instance from the request
+     */
+    public static ElasticsearchClusterDTO getFromRequest(ElasticsearchClusterRequest request) {
+        return ElasticsearchClusterDTO.builder()
+                .build();
+    }
+
+    /**
+     * Get the dto instance from the JSON string.
+     */
+    public static ElasticsearchClusterDTO getFromJson(@NotNull String extParams) {
+        try {
+            return JsonUtils.parseObject(extParams, ElasticsearchClusterDTO.class);
+        } catch (Exception e) {
+            throw new BusinessException(ErrorCodeEnum.CLUSTER_INFO_INCORRECT.getMessage() + ": " + e.getMessage());
+        }
+    }
+}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterInfo.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterInfo.java
new file mode 100644
index 000000000..c15de7451
--- /dev/null
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterInfo.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.inlong.manager.pojo.cluster.es;
+
+import io.swagger.annotations.ApiModel;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+import org.apache.inlong.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.util.CommonBeanUtils;
+import org.apache.inlong.manager.common.util.JsonTypeDefine;
+import org.apache.inlong.manager.pojo.cluster.ClusterInfo;
+import org.apache.inlong.manager.pojo.cluster.ClusterRequest;
+
+/**
+ * Elasticsearch cluster info
+ */
+@Data
+@SuperBuilder
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeDefine(value = ClusterType.ELASTICSEARCH)
+@ApiModel("Inlong cluster info for Elasticsearch")
+public class ElasticsearchClusterInfo extends ClusterInfo {
+
+    public ElasticsearchClusterInfo() {
+        this.setType(ClusterType.ELASTICSEARCH);
+    }
+
+    @Override
+    public ClusterRequest genRequest() {
+        return CommonBeanUtils.copyProperties(this, ElasticsearchClusterRequest::new);
+    }
+}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterRequest.java
new file mode 100644
index 000000000..569e9b5d1
--- /dev/null
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/es/ElasticsearchClusterRequest.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.inlong.manager.pojo.cluster.es;
+
+import io.swagger.annotations.ApiModel;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+import org.apache.inlong.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.util.JsonTypeDefine;
+import org.apache.inlong.manager.pojo.cluster.ClusterRequest;
+
+/**
+ * Inlong cluster request for Elasticsearch
+ */
+@Data
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeDefine(value = ClusterType.ELASTICSEARCH)
+@ApiModel("Inlong cluster request for Elasticsearch")
+public class ElasticsearchClusterRequest extends ClusterRequest {
+
+    public ElasticsearchClusterRequest() {
+        this.setType(ClusterType.ELASTICSEARCH);
+    }
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/ElasticsearchClusterOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/ElasticsearchClusterOperator.java
new file mode 100644
index 000000000..448629bd9
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/ElasticsearchClusterOperator.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.inlong.manager.service.cluster;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+import org.apache.inlong.manager.common.util.CommonBeanUtils;
+import org.apache.inlong.manager.dao.entity.InlongClusterEntity;
+import org.apache.inlong.manager.pojo.cluster.ClusterInfo;
+import org.apache.inlong.manager.pojo.cluster.ClusterRequest;
+import org.apache.inlong.manager.pojo.cluster.es.ElasticsearchClusterInfo;
+import org.apache.inlong.manager.pojo.cluster.es.ElasticsearchClusterRequest;
+import org.apache.inlong.manager.pojo.cluster.es.ElasticsearchClusterDTO;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * Elasticsearch cluster operator.
+ */
+@Service
+public class ElasticsearchClusterOperator extends AbstractClusterOperator {
+
+    @Autowired
+    private ObjectMapper mapper;
+
+    @Override
+    public Boolean accept(String clusterType) {
+        return getClusterType().equals(clusterType);
+    }
+
+    @Override
+    public String getClusterType() {
+        return ClusterType.ELASTICSEARCH;
+    }
+
+    @Override
+    protected void setTargetEntity(ClusterRequest request, InlongClusterEntity targetEntity) {
+        ElasticsearchClusterRequest esRequest = (ElasticsearchClusterRequest) request;
+        CommonBeanUtils.copyProperties(esRequest, targetEntity, true);
+        try {
+            ElasticsearchClusterDTO dto = ElasticsearchClusterDTO.getFromRequest(esRequest);
+            targetEntity.setExtParams(mapper.writeValueAsString(dto));
+        } catch (Exception e) {
+            throw new BusinessException(ErrorCodeEnum.SOURCE_INFO_INCORRECT.getMessage() + ": " + e.getMessage());
+        }
+    }
+
+    @Override
+    public ClusterInfo getFromEntity(InlongClusterEntity entity) {
+        if (entity == null) {
+            throw new BusinessException(ErrorCodeEnum.CLUSTER_NOT_FOUND);
+        }
+        ElasticsearchClusterInfo info = new ElasticsearchClusterInfo();
+        CommonBeanUtils.copyProperties(entity, info);
+        if (StringUtils.isNotBlank(entity.getExtParams())) {
+            ElasticsearchClusterDTO dto = ElasticsearchClusterDTO.getFromJson(entity.getExtParams());
+            CommonBeanUtils.copyProperties(dto, info);
+        }
+        return info;
+    }
+}