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 <deng...@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;
+ }
+}
