This is an automated email from the ASF dual-hosted git repository.
dockerzhang 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 1e25c547e2 [INLONG-8392][Manager] Optimize the HeartbeatMsg class
(#9512)
1e25c547e2 is described below
commit 1e25c547e2897bfa651e635dd3f03153b35274f5
Author: fuweng11 <[email protected]>
AuthorDate: Fri Dec 22 16:38:44 2023 +0800
[INLONG-8392][Manager] Optimize the HeartbeatMsg class (#9512)
---
.../AddressInfo.java} | 38 +++++-------
.../inlong/common/heartbeat/HeartbeatMsg.java | 7 +++
.../ReportResourceType.java} | 37 ++----------
.../pojo/dataproxy/DataProxyNodeResponse.java | 2 +
.../cluster/dataproxy/DataProxyClusterNodeDTO.java | 67 ++++++++++++++++++++++
.../dataproxy/DataProxyClusterNodeRequest.java | 49 ++++++----------
.../dataproxy/DataProxyClusterNodeResponse.java | 49 ++++++----------
.../service/cluster/InlongClusterService.java | 10 ++++
.../service/cluster/InlongClusterServiceImpl.java | 56 ++++++++++++++++++
.../service/heartbeat/HeartbeatManager.java | 46 +++++++++++++--
.../controller/openapi/DataProxyController.java | 8 +++
11 files changed, 246 insertions(+), 123 deletions(-)
diff --git
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
b/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/AddressInfo.java
similarity index 58%
copy from
inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
copy to
inlong-common/src/main/java/org/apache/inlong/common/heartbeat/AddressInfo.java
index cccf61d428..c07c7faf69 100644
---
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
+++
b/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/AddressInfo.java
@@ -15,42 +15,34 @@
* limitations under the License.
*/
-package org.apache.inlong.common.pojo.dataproxy;
+package org.apache.inlong.common.heartbeat;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
import lombok.Data;
+import lombok.NoArgsConstructor;
-import java.util.List;
-
-/**
- * DataProxy node response, used for DataProxy SDK.
- */
@Data
-public class DataProxyNodeResponse {
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class AddressInfo {
/**
- * DataProxy cluster id
+ * Ip of component
*/
- @Deprecated
- private Integer clusterId;
+ private String ip;
/**
- * Is the DataProxy cluster an intranet? 0: no, 1: yes
+ * Port of component
*/
- private Integer isIntranet;
+ private String port;
/**
- * Is the DataProxy cluster in a switch status? 0: no, 1: yes
+ * Report source type of component
*/
- private Integer isSwitch;
+ private String reportSourceType;
- /**
- * Load of the DataProxy cluster, default is 20
- */
- private Integer load = 20;
-
- /**
- * List of the cluster node
- */
- private List<DataProxyNodeInfo> nodeList;
+ private String protocolType;
}
diff --git
a/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/HeartbeatMsg.java
b/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/HeartbeatMsg.java
index 3a54b714f7..43ede0823b 100644
---
a/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/HeartbeatMsg.java
+++
b/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/HeartbeatMsg.java
@@ -50,6 +50,8 @@ public class HeartbeatMsg {
*/
private String port;
+ private List<AddressInfo> addressInfos;
+
/**
* ProtocolType of component
*/
@@ -60,6 +62,11 @@ public class HeartbeatMsg {
*/
private String componentType;
+ /**
+ * Type of report source
+ */
+ private String reportSourceType;
+
/**
* Report time millis of component
*/
diff --git
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
b/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/ReportResourceType.java
similarity index 53%
copy from
inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
copy to
inlong-common/src/main/java/org/apache/inlong/common/heartbeat/ReportResourceType.java
index cccf61d428..3001622df6 100644
---
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
+++
b/inlong-common/src/main/java/org/apache/inlong/common/heartbeat/ReportResourceType.java
@@ -15,42 +15,13 @@
* limitations under the License.
*/
-package org.apache.inlong.common.pojo.dataproxy;
-
-import lombok.Data;
-
-import java.util.List;
+package org.apache.inlong.common.heartbeat;
/**
- * DataProxy node response, used for DataProxy SDK.
+ * Constants of reportResource
*/
-@Data
-public class DataProxyNodeResponse {
-
- /**
- * DataProxy cluster id
- */
- @Deprecated
- private Integer clusterId;
-
- /**
- * Is the DataProxy cluster an intranet? 0: no, 1: yes
- */
- private Integer isIntranet;
-
- /**
- * Is the DataProxy cluster in a switch status? 0: no, 1: yes
- */
- private Integer isSwitch;
-
- /**
- * Load of the DataProxy cluster, default is 20
- */
- private Integer load = 20;
+public class ReportResourceType {
- /**
- * List of the cluster node
- */
- private List<DataProxyNodeInfo> nodeList;
+ public static final String INLONG = "INLONG";
}
diff --git
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
b/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
index cccf61d428..7b53ef96db 100644
---
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
+++
b/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
@@ -33,6 +33,8 @@ public class DataProxyNodeResponse {
@Deprecated
private Integer clusterId;
+ private String reportSourceType;
+
/**
* Is the DataProxy cluster an intranet? 0: no, 1: yes
*/
diff --git
a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeDTO.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeDTO.java
new file mode 100644
index 0000000000..af44f1c7b5
--- /dev/null
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeDTO.java
@@ -0,0 +1,67 @@
+/*
+ * 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.dataproxy;
+
+import org.apache.inlong.common.heartbeat.ReportResourceType;
+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 io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.NotNull;
+
+/**
+ * DataProxy cluster node info
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel("DataProxy cluster node info")
+public class DataProxyClusterNodeDTO {
+
+ @ApiModelProperty("Report source type")
+ private String reportSourceType = ReportResourceType.INLONG;
+
+ /**
+ * Get the dto instance from the request
+ */
+ public static DataProxyClusterNodeDTO
getFromRequest(DataProxyClusterNodeRequest request) {
+ return DataProxyClusterNodeDTO.builder()
+ .reportSourceType(request.getReportSourceType())
+ .build();
+ }
+
+ /**
+ * Get the dto instance from the JSON string.
+ */
+ public static DataProxyClusterNodeDTO getFromJson(@NotNull String
extParams) {
+ try {
+ return JsonUtils.parseObject(extParams,
DataProxyClusterNodeDTO.class);
+ } catch (Exception e) {
+ throw new
BusinessException(ErrorCodeEnum.CLUSTER_INFO_INCORRECT.getMessage() + ": " +
e.getMessage());
+ }
+ }
+
+}
diff --git
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeRequest.java
similarity index 52%
copy from
inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
copy to
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeRequest.java
index cccf61d428..2d9c7007b2 100644
---
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeRequest.java
@@ -15,42 +15,29 @@
* limitations under the License.
*/
-package org.apache.inlong.common.pojo.dataproxy;
+package org.apache.inlong.manager.pojo.cluster.dataproxy;
-import lombok.Data;
+import org.apache.inlong.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.util.JsonTypeDefine;
+import org.apache.inlong.manager.pojo.cluster.ClusterNodeRequest;
-import java.util.List;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
/**
- * DataProxy node response, used for DataProxy SDK.
+ * Inlong cluster node request for Agent
*/
@Data
-public class DataProxyNodeResponse {
-
- /**
- * DataProxy cluster id
- */
- @Deprecated
- private Integer clusterId;
-
- /**
- * Is the DataProxy cluster an intranet? 0: no, 1: yes
- */
- private Integer isIntranet;
-
- /**
- * Is the DataProxy cluster in a switch status? 0: no, 1: yes
- */
- private Integer isSwitch;
-
- /**
- * Load of the DataProxy cluster, default is 20
- */
- private Integer load = 20;
-
- /**
- * List of the cluster node
- */
- private List<DataProxyNodeInfo> nodeList;
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeDefine(value = ClusterType.DATAPROXY)
+@ApiModel("Inlong cluster node request for dataproxy")
+public class DataProxyClusterNodeRequest extends ClusterNodeRequest {
+
+ @ApiModelProperty("Report source type")
+ private String reportSourceType;
}
diff --git
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeResponse.java
similarity index 52%
copy from
inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
copy to
inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeResponse.java
index cccf61d428..56469c6681 100644
---
a/inlong-common/src/main/java/org/apache/inlong/common/pojo/dataproxy/DataProxyNodeResponse.java
+++
b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeResponse.java
@@ -15,42 +15,29 @@
* limitations under the License.
*/
-package org.apache.inlong.common.pojo.dataproxy;
+package org.apache.inlong.manager.pojo.cluster.dataproxy;
-import lombok.Data;
+import org.apache.inlong.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.util.JsonTypeDefine;
+import org.apache.inlong.manager.pojo.cluster.ClusterNodeResponse;
-import java.util.List;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
/**
- * DataProxy node response, used for DataProxy SDK.
+ * Dataproxy cluster node response
*/
@Data
-public class DataProxyNodeResponse {
-
- /**
- * DataProxy cluster id
- */
- @Deprecated
- private Integer clusterId;
-
- /**
- * Is the DataProxy cluster an intranet? 0: no, 1: yes
- */
- private Integer isIntranet;
-
- /**
- * Is the DataProxy cluster in a switch status? 0: no, 1: yes
- */
- private Integer isSwitch;
-
- /**
- * Load of the DataProxy cluster, default is 20
- */
- private Integer load = 20;
-
- /**
- * List of the cluster node
- */
- private List<DataProxyNodeInfo> nodeList;
+@ToString(callSuper = true)
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeDefine(value = ClusterType.DATAPROXY)
+@ApiModel("Inlong cluster node response for dataproxy")
+public class DataProxyClusterNodeResponse extends ClusterNodeResponse {
+
+ @ApiModelProperty("Report source type")
+ private String reportSourceType;
}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterService.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterService.java
index dd35ec8e2d..5062e4e730 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterService.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterService.java
@@ -407,6 +407,16 @@ public interface InlongClusterService {
*/
DataProxyNodeResponse getDataProxyNodes(String inlongGroupId, String
protocolType);
+ /**
+ * Query data proxy nodes by the given inlong group id and protocol type
+ *
+ * @param clusterName inlong cluster name
+ * @param protocolType protocol type
+ * @param reportSourceType report source type
+ * @return data proxy node response
+ */
+ DataProxyNodeResponse getDataProxyNodesByCluster(String clusterName,
String protocolType, String reportSourceType);
+
/**
* Get the configuration of DataProxy through the cluster name to which
DataProxy belongs.
*
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterServiceImpl.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterServiceImpl.java
index 123221ff93..f1bcfa3ab1 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterServiceImpl.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterServiceImpl.java
@@ -19,6 +19,7 @@ package org.apache.inlong.manager.service.cluster;
import org.apache.inlong.common.constant.Constants;
import org.apache.inlong.common.constant.MQType;
+import org.apache.inlong.common.heartbeat.ReportResourceType;
import org.apache.inlong.common.pojo.audit.AuditConfig;
import org.apache.inlong.common.pojo.audit.MQInfo;
import org.apache.inlong.common.pojo.dataproxy.DataProxyCluster;
@@ -60,6 +61,7 @@ import
org.apache.inlong.manager.pojo.cluster.ClusterTagResponse;
import org.apache.inlong.manager.pojo.cluster.TenantClusterTagInfo;
import org.apache.inlong.manager.pojo.cluster.TenantClusterTagPageRequest;
import org.apache.inlong.manager.pojo.cluster.TenantClusterTagRequest;
+import
org.apache.inlong.manager.pojo.cluster.dataproxy.DataProxyClusterNodeDTO;
import org.apache.inlong.manager.pojo.cluster.pulsar.PulsarClusterDTO;
import org.apache.inlong.manager.pojo.common.PageResult;
import org.apache.inlong.manager.pojo.common.UpdateResult;
@@ -1178,6 +1180,60 @@ public class InlongClusterServiceImpl implements
InlongClusterService {
return response;
}
+ @Override
+ public DataProxyNodeResponse getDataProxyNodesByCluster(String
clusterName, String protocolType,
+ String reportSourceType) {
+ LOGGER.debug("begin to get data proxy nodes for clusterName={},
protocol={}", clusterName, protocolType);
+ InlongClusterEntity clusterEntity =
clusterMapper.selectByNameAndType(clusterName, ClusterType.DATAPROXY);
+ DataProxyNodeResponse response = new DataProxyNodeResponse();
+ if (clusterEntity == null) {
+ LOGGER.debug("not any dataproxy cluster for clusterName={},
protocol={}", clusterName, protocolType);
+ return response;
+ }
+ List<InlongClusterNodeEntity> nodeEntities =
+ clusterNodeMapper.selectByParentId(clusterEntity.getId(),
protocolType);
+ if (CollectionUtils.isEmpty(nodeEntities)) {
+ LOGGER.debug("not any data proxy node for clusterName={},
protocol={}", clusterName, protocolType);
+ return response;
+ }
+ // all cluster nodes belong to the same clusterId
+ response.setClusterId(clusterEntity.getId());
+ // TODO consider the data proxy load and re-balance
+ List<DataProxyNodeInfo> nodeList = new ArrayList<>();
+ for (InlongClusterNodeEntity nodeEntity : nodeEntities) {
+ if (Objects.equals(nodeEntity.getStatus(),
NodeStatus.HEARTBEAT_TIMEOUT.getStatus())) {
+ LOGGER.debug("dataproxy node was timeout, parentId={} ip={}
port={}", nodeEntity.getParentId(),
+ nodeEntity.getIp(), nodeEntity.getPort());
+ continue;
+ }
+ if (StringUtils.isNotBlank(nodeEntity.getExtParams())) {
+ DataProxyClusterNodeDTO dataProxyClusterNodeDTO =
DataProxyClusterNodeDTO.getFromJson(
+ nodeEntity.getExtParams());
+ if
(StringUtils.isBlank(dataProxyClusterNodeDTO.getReportSourceType())) {
+
dataProxyClusterNodeDTO.setReportSourceType(ReportResourceType.INLONG);
+ }
+ if (StringUtils.isNotBlank(reportSourceType) &&
!Objects.equals(
+ dataProxyClusterNodeDTO.getReportSourceType(),
reportSourceType)) {
+ continue;
+ }
+ }
+ DataProxyNodeInfo nodeInfo = new DataProxyNodeInfo();
+ nodeInfo.setId(nodeEntity.getId());
+ nodeInfo.setIp(nodeEntity.getIp());
+ nodeInfo.setPort(nodeEntity.getPort());
+ nodeInfo.setProtocolType(nodeEntity.getProtocolType());
+ nodeInfo.setNodeLoad(nodeEntity.getNodeLoad());
+ nodeList.add(nodeInfo);
+ }
+ response.setNodeList(nodeList);
+
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("success to get dp nodes for clusterName={},
protocol={}, result={}",
+ clusterName, protocolType, response);
+ }
+ return response;
+ }
+
private List<InlongClusterNodeEntity> getClusterNodes(String groupId,
String clusterType, String protocolType) {
InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
if (groupEntity == null) {
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
index 291218ecd5..53466f13cc 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/heartbeat/HeartbeatManager.java
@@ -19,6 +19,7 @@ package org.apache.inlong.manager.service.heartbeat;
import org.apache.inlong.common.enums.NodeSrvStatus;
import org.apache.inlong.common.heartbeat.AbstractHeartbeatManager;
+import org.apache.inlong.common.heartbeat.AddressInfo;
import org.apache.inlong.common.heartbeat.ComponentHeartbeat;
import org.apache.inlong.common.heartbeat.HeartbeatMsg;
import org.apache.inlong.manager.common.consts.InlongConstants;
@@ -38,6 +39,7 @@ import
org.apache.inlong.manager.dao.mapper.StreamSourceEntityMapper;
import org.apache.inlong.manager.pojo.cluster.ClusterInfo;
import org.apache.inlong.manager.pojo.cluster.ClusterNodeRequest;
import org.apache.inlong.manager.pojo.cluster.agent.AgentClusterNodeDTO;
+import
org.apache.inlong.manager.pojo.cluster.dataproxy.DataProxyClusterNodeDTO;
import org.apache.inlong.manager.service.cluster.InlongClusterOperator;
import org.apache.inlong.manager.service.cluster.InlongClusterOperatorFactory;
@@ -153,10 +155,25 @@ public class HeartbeatManager implements
AbstractHeartbeatManager {
// if the heartbeat was not in the cache, insert or update the node by
the heartbeat info
HeartbeatMsg lastHeartbeat =
heartbeatCache.getIfPresent(componentHeartbeat);
+ if (heartbeat.getAddressInfos() != null) {
+ heartbeat.setPort(Joiner.on(InlongConstants.COMMA)
+
.join(heartbeat.getAddressInfos().stream().map(AddressInfo::getPort).collect(Collectors.toList())));
+ heartbeat.setIp(Joiner.on(InlongConstants.COMMA)
+
.join(heartbeat.getAddressInfos().stream().map(AddressInfo::getIp).collect(Collectors.toList())));
+
heartbeat.setReportSourceType(Joiner.on(InlongConstants.COMMA).join(heartbeat.getAddressInfos().stream()
+
.map(AddressInfo::getReportSourceType).collect(Collectors.toList())));
+ }
// protocolType may be null, and the protocolTypes' length may be less
than ports' length
String[] ports = heartbeat.getPort().split(InlongConstants.COMMA);
String[] ips = heartbeat.getIp().split(InlongConstants.COMMA);
+ String[] reportSourceTypes = null;
+ if (StringUtils.isNotBlank(heartbeat.getReportSourceType()) &&
ports.length > 1) {
+ reportSourceTypes =
heartbeat.getReportSourceType().split(InlongConstants.COMMA);
+ if (reportSourceTypes.length < ports.length) {
+ reportSourceTypes = null;
+ }
+ }
String protocolType = heartbeat.getProtocolType();
String[] protocolTypes = null;
if (StringUtils.isNotBlank(protocolType) && ports.length > 1) {
@@ -173,6 +190,11 @@ public class HeartbeatManager implements
AbstractHeartbeatManager {
assert heartbeatMsg != null;
heartbeatMsg.setPort(ports[i].trim());
heartbeatMsg.setIp(ips[i].trim());
+ if (reportSourceTypes != null) {
+ heartbeatMsg.setReportSourceType(reportSourceTypes[i].trim());
+ } else {
+
heartbeatMsg.setReportSourceType(heartbeat.getReportSourceType());
+ }
if (protocolTypes != null) {
heartbeatMsg.setProtocolType(protocolTypes[i]);
} else {
@@ -301,12 +323,26 @@ public class HeartbeatManager implements
AbstractHeartbeatManager {
private void insertOrUpdateNodeGroup(InlongClusterNodeEntity clusterNode,
HeartbeatMsg heartbeat) {
Set<String> groupSet = StringUtils.isBlank(heartbeat.getNodeGroup()) ?
new HashSet<>()
:
Arrays.stream(heartbeat.getNodeGroup().split(InlongConstants.COMMA)).collect(Collectors.toSet());
- AgentClusterNodeDTO agentClusterNodeDTO = new AgentClusterNodeDTO();
- if (StringUtils.isNotBlank(clusterNode.getExtParams())) {
- agentClusterNodeDTO =
AgentClusterNodeDTO.getFromJson(clusterNode.getExtParams());
-
agentClusterNodeDTO.setAgentGroup(Joiner.on(InlongConstants.COMMA).join(groupSet));
+ String extParams = null;
+ switch (clusterNode.getType()) {
+ case ClusterType.AGENT:
+ AgentClusterNodeDTO agentClusterNodeDTO = new
AgentClusterNodeDTO();
+ if (StringUtils.isNotBlank(clusterNode.getExtParams())) {
+ agentClusterNodeDTO =
AgentClusterNodeDTO.getFromJson(clusterNode.getExtParams());
+
agentClusterNodeDTO.setAgentGroup(Joiner.on(InlongConstants.COMMA).join(groupSet));
+ }
+ extParams = GSON.toJson(agentClusterNodeDTO);
+ break;
+ case ClusterType.DATAPROXY:
+ DataProxyClusterNodeDTO dataProxyClusterNodeDTO = new
DataProxyClusterNodeDTO();
+ if (StringUtils.isNotBlank(clusterNode.getExtParams())) {
+ dataProxyClusterNodeDTO =
DataProxyClusterNodeDTO.getFromJson(clusterNode.getExtParams());
+
dataProxyClusterNodeDTO.setReportSourceType(heartbeat.getReportSourceType());
+ }
+ extParams = GSON.toJson(dataProxyClusterNodeDTO);
+ break;
}
- clusterNode.setExtParams(GSON.toJson(agentClusterNodeDTO));
+ clusterNode.setExtParams(extParams);
}
private int deleteClusterNode(InlongClusterNodeEntity clusterNode) {
diff --git
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/DataProxyController.java
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/DataProxyController.java
index cea54c8c82..3c38276000 100644
---
a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/DataProxyController.java
+++
b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/openapi/DataProxyController.java
@@ -59,6 +59,14 @@ public class DataProxyController {
return
Response.success(clusterService.getDataProxyNodes(inlongGroupId, protocolType));
}
+ @PostMapping(value = "/dataproxy/getIpListByClusterName/{clusterName}")
+ @ApiOperation(value = "Get data proxy IP list by clusterName")
+ public Response<DataProxyNodeResponse>
getIpListByClusterName(@PathVariable String clusterName,
+ @RequestParam(required = false) String protocolType,
+ @RequestParam(required = false) String reportSourceType) {
+ return
Response.success(clusterService.getDataProxyNodesByCluster(clusterName,
protocolType, reportSourceType));
+ }
+
@PostMapping("/dataproxy/getConfig")
@ApiOperation(value = "Get data proxy topic list")
public Response<DataProxyConfig> getConfig(@RequestBody
DataProxyConfigRequest request) {
