[INLONG-8392][Manager] Optimize the HeartbeatMsg class

inlong-common/src/main/java/org/apache/inlong/common/heartbeat/AddressInfo.java

@@ -0,0 +1,48 @@
+/*
+ * 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.common.heartbeat;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class AddressInfo {
+
+    /**
+     * Ip of component
+     */
+    private String ip;
+
+    /**
+     * Port of component
+     */
+    private String port;
+
+    /**
+     * Report source type of component
+     */
+    private String reportSourceType;
+
+    private String protocolType;
+
+}

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
      */

inlong-common/src/main/java/org/apache/inlong/common/heartbeat/ReportResourceType.java

@@ -0,0 +1,27 @@
+/*
+ * 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.common.heartbeat;
+
+/**
+ * Constants of reportResource
+ */
+public class ReportResourceType {
+
+    public static final String INLONG = "INLONG";
+
+}

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
      */

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());
+        }
+    }
+
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeRequest.java

@@ -0,0 +1,43 @@
+/*
+ * 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.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.util.JsonTypeDefine;
+import org.apache.inlong.manager.pojo.cluster.ClusterNodeRequest;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+/**
+ * Inlong cluster node request for Agent
+ */
+@Data
+@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;
+
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/cluster/dataproxy/DataProxyClusterNodeResponse.java

@@ -0,0 +1,43 @@
+/*
+ * 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.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.util.JsonTypeDefine;
+import org.apache.inlong.manager.pojo.cluster.ClusterNodeResponse;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+/**
+ * Dataproxy cluster node response
+ */
+@Data
+@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;
+
+}

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterService.java

@@ -407,6 +407,16 @@ List<ClusterNodeResponse> listNodeByGroupId(
      */
     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.
      *

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterServiceImpl.java

@@ -19,6 +19,7 @@
 
 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.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 DataProxyNodeResponse getDataProxyNodes(String groupId, String protocolTy
         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) {