Feature: Support discovery monitoring instance through http_sd

collector/src/main/java/org/apache/hertzbeat/collector/collect/AbstractCollect.java

@@ -40,4 +40,12 @@ public abstract class AbstractCollect {
      * @return protocol str
      */
     public abstract String supportProtocol();
+
+    /**
+     * get protocol class
+     * @return protocol class
+     */
+    public Class<?> getSupportProtocolClass() {
+        return null;
+    }
 }

collector/src/main/java/org/apache/hertzbeat/collector/collect/common/cache/sd/ConnectionConfig.java

@@ -0,0 +1,35 @@
+/*
+ * 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.hertzbeat.collector.collect.common.cache.sd;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Connection config
+ */
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class ConnectionConfig {
+    private String host;
+    private String port;
+}

collector/src/main/java/org/apache/hertzbeat/collector/collect/common/cache/sd/ServiceDiscoveryCache.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.hertzbeat.collector.collect.common.cache.sd;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.util.CollectionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * SD Cache
+ */
+public class ServiceDiscoveryCache {
+    private static final Map<Long, List<ConnectionConfig>> sd = Maps.newConcurrentMap();
+
+    public static List<ConnectionConfig> getConfig(Long jobId) {
+        return sd.getOrDefault(jobId, Lists.newArrayList());
+    }
+
+    public static void updateConfig(Long jobId, List<ConnectionConfig> configList) {
+        if (Objects.isNull(jobId) || CollectionUtils.isEmpty(configList)) {
+            return;
+        }
+
+        sd.put(jobId, configList.stream()
+                .filter(config -> StringUtils.isNoneBlank(config.getHost(), config.getPort()))
+                .collect(Collectors.toList()));
+    }
+
+    public static void removeConfig(Long jobId) {
+        sd.remove(jobId);
+    }
+
+    public static void clear() {
+        sd.clear();
+    }
+}

collector/src/main/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcher.java

@@ -17,13 +17,18 @@
 
 package org.apache.hertzbeat.collector.dispatch;
 
+import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.gson.Gson;
 import com.google.gson.JsonElement;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.net.telnet.TelnetClient;
+import org.apache.hertzbeat.collector.collect.common.cache.sd.ConnectionConfig;
+import org.apache.hertzbeat.collector.collect.common.cache.sd.ServiceDiscoveryCache;
 import org.apache.hertzbeat.collector.dispatch.entrance.internal.CollectJobService;
+import org.apache.hertzbeat.collector.dispatch.entrance.sd.ServiceDiscoveryFetcher;
 import org.apache.hertzbeat.collector.dispatch.timer.Timeout;
 import org.apache.hertzbeat.collector.dispatch.timer.TimerDispatch;
 import org.apache.hertzbeat.collector.dispatch.timer.WheelTimerTask;
@@ -35,7 +40,9 @@
 import org.apache.hertzbeat.common.entity.message.CollectRep;
 import org.apache.hertzbeat.common.queue.CommonDataQueue;
 import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
 
+import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -175,7 +182,10 @@ public void dispatchMetricsTask(Timeout timeout) {
         // Put each collect task into the thread pool for scheduling
         WheelTimerTask timerTask = (WheelTimerTask) timeout.task();
         Job job = timerTask.getJob();
+        // Nothing happen if failed to update sd cache
+        updateSdCacheGracefully(job);
         job.constructPriorMetrics();
+
         Set<Metrics> metricsSet = job.getNextCollectMetrics(null, true);
         metricsSet.forEach(metrics -> {
             MetricsCollect metricsCollect = new MetricsCollect(metrics, timeout, this,
@@ -340,6 +350,38 @@ public void dispatchCollectData(Timeout timeout, Metrics metrics, List<CollectRe
 
     }
 
+    private void updateSdCacheGracefully(Job job) {
+        if (Objects.isNull(job.getSdProtocol())) {
+            return;
+        }
+
+        // fetch connection config
+        List<ConnectionConfig> configList = ServiceDiscoveryFetcher.doFetch(job.getSdProtocol());
+        List<ConnectionConfig> availableConfigList = Lists.newArrayListWithExpectedSize(configList.size());
+        // check if config is available
+        configList.forEach(config -> {
+            TelnetClient telnetClient = new TelnetClient("vt200");
+            telnetClient.setConnectTimeout(10_000);
+            try {
+                telnetClient.connect(config.getHost(), Integer.parseInt(config.getPort()));
+                if (telnetClient.isConnected()) {
+                    availableConfigList.add(config);
+                }
+            } catch (IOException ignore) {
+            } finally {
+                try {
+                    telnetClient.disconnect();
+                } catch (IOException ignore) {
+                }
+            }
+        });
+
+        if (!CollectionUtils.isEmpty(availableConfigList)) {
+            ServiceDiscoveryCache.removeConfig(job.getId());
+            ServiceDiscoveryCache.updateConfig(job.getId(), availableConfigList);
+        }
+    }
+
     private List<Map<String, Configmap>> getConfigmapFromPreCollectData(CollectRep.MetricsData metricsData) {
         if (metricsData.getValuesCount() <= 0 || metricsData.getFieldsCount() <= 0) {
             return new LinkedList<>();

collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java

@@ -21,6 +21,8 @@
 import com.googlecode.aviator.Expression;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.collect.common.cache.sd.ConnectionConfig;
+import org.apache.hertzbeat.collector.collect.common.cache.sd.ServiceDiscoveryCache;
 import org.apache.hertzbeat.collector.dispatch.timer.Timeout;
 import org.apache.hertzbeat.collector.dispatch.timer.WheelTimerTask;
 import org.apache.hertzbeat.collector.dispatch.unit.UnitConvert;
@@ -31,10 +33,13 @@
 import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.common.entity.job.Job;
 import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.job.protocol.CommonProtocol;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
 import org.apache.hertzbeat.common.util.CommonUtil;
 import org.apache.hertzbeat.common.util.Pair;
+import org.springframework.util.CollectionUtils;
 
+import java.lang.reflect.Field;
 import java.util.*;
 import java.util.stream.Collectors;
 
@@ -56,6 +61,7 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
      * Tenant ID
      */
     protected long tenantId;
+    protected long jobId;
     /**
      * Monitor ID
      */
@@ -105,6 +111,7 @@ public MetricsCollect(Metrics metrics, Timeout timeout,
         this.collectorIdentity = collectorIdentity;
         WheelTimerTask timerJob = (WheelTimerTask) timeout.task();
         Job job = timerJob.getJob();
+        this.jobId = job.getId();
         this.monitorId = job.getMonitorId();
         this.tenantId = job.getTenantId();
         this.app = job.getApp();
@@ -146,6 +153,9 @@ public void run() {
             response.setMsg("not support " + app + ", "
                     + metrics.getName() + ", " + metrics.getProtocol());
         } else {
+            // reset host and port if sd is available
+            resetHostAndPortBySd(abstractCollect);
+
             try {
                 abstractCollect.collect(response, monitorId, app, metrics);
             } catch (Exception e) {
@@ -169,6 +179,35 @@ public void run() {
         collectDataDispatch.dispatchCollectData(timeout, metrics, metricsData);
     }
 
+    /**
+     * reset host and port from {@link ServiceDiscoveryCache}
+     */
+    private void resetHostAndPortBySd(AbstractCollect abstractCollect) {
+        Class<?> supportProtocolClass = abstractCollect.getSupportProtocolClass();
+        final List<ConnectionConfig> configList = ServiceDiscoveryCache.getConfig(jobId);
+        if (Objects.isNull(supportProtocolClass) || CollectionUtils.isEmpty(configList)) {
+            return;
+        }
+        // selecting config strategy can be added here
+        ConnectionConfig connectionConfig = configList.get(0);
+
+        final List<Field> fieldList = Arrays.stream(metrics.getClass().getDeclaredFields())
+                .filter(field -> Objects.equals(supportProtocolClass, field.getType()))
+                .collect(Collectors.toList());
+        if (CollectionUtils.isEmpty(fieldList)) {
+            return;
+        }
+
+        for (Field protocolField : fieldList) {
+            try {
+                protocolField.setAccessible(true);
+                final CommonProtocol commonProtocol = (CommonProtocol) protocolField.get(metrics);
+                commonProtocol.setHost(connectionConfig.getHost());
+                commonProtocol.setPort(connectionConfig.getPort());
+            } catch (IllegalAccessException ignore) {
+            }
+        }
+    }
 
     /**
      * Calculate the real metrics value according to the calculates and aliasFields configuration

collector/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServer.java

@@ -96,6 +96,7 @@ private void init(final DispatchProperties properties, final CommonThreadPool th
         this.remotingClient.registerProcessor(ClusterMsg.MessageType.GO_OFFLINE, new GoOfflineProcessor());
         this.remotingClient.registerProcessor(ClusterMsg.MessageType.GO_ONLINE, new GoOnlineProcessor());
         this.remotingClient.registerProcessor(ClusterMsg.MessageType.GO_CLOSE, new GoCloseProcessor(this));
+        this.remotingClient.registerProcessor(ClusterMsg.MessageType.ISSUE_SD_UPDATE_TASK, new UpdateServiceDiscoveryProcessor(this));
     }
 
     public void shutdown() {

collector/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/internal/CollectJobService.java

@@ -17,6 +17,7 @@
 
 package org.apache.hertzbeat.collector.dispatch.entrance.internal;
 
+import org.apache.hertzbeat.collector.collect.common.cache.sd.ServiceDiscoveryCache;
 import org.apache.hertzbeat.collector.dispatch.DispatchProperties;
 import org.apache.hertzbeat.collector.dispatch.WorkerPool;
 import org.apache.hertzbeat.collector.dispatch.entrance.CollectServer;
@@ -25,6 +26,7 @@
 import org.apache.hertzbeat.common.entity.job.Job;
 import org.apache.hertzbeat.common.entity.message.ClusterMsg;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.entity.sd.ServiceDiscoveryProtocol;
 import org.apache.hertzbeat.common.util.IpDomainUtil;
 import org.apache.hertzbeat.common.util.JsonUtil;
 import lombok.extern.slf4j.Slf4j;
@@ -101,6 +103,8 @@ public void response(List<CollectRep.MetricsData> responseMetrics) {
         } catch (Exception e) {
             log.info("The sync task runs for 120 seconds with no response and returns");
         }
+
+        ServiceDiscoveryCache.removeConfig(job.getId());
         return metricsData;
     }
 
@@ -139,6 +143,13 @@ public void addAsyncCollectJob(Job job) {
         timerDispatch.addJob(job.clone(), null);
     }
 
+    /**
+     * Update Service Provider Cache for cyclic task
+     */
+    public void updateServiceProvider(ServiceDiscoveryProtocol serviceDiscoveryProtocol) {
+        timerDispatch.updateJobSdCache(serviceDiscoveryProtocol);
+    }
+
     /**
      * Cancel periodic asynchronous collection tasks
      * 取消周期性异步采集任务

collector/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/UpdateServiceDiscoveryProcessor.java

@@ -0,0 +1,46 @@
+/*
+ * 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.hertzbeat.collector.dispatch.entrance.processor;
+
+import io.netty.channel.ChannelHandlerContext;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.dispatch.entrance.CollectServer;
+import org.apache.hertzbeat.common.entity.message.ClusterMsg;
+import org.apache.hertzbeat.common.entity.sd.ServiceDiscoveryProtocol;
+import org.apache.hertzbeat.common.util.JsonUtil;
+import org.apache.hertzbeat.remoting.netty.NettyRemotingProcessor;
+
+import java.util.Objects;
+
+/**
+ * handle updating sd cache message
+ */
+@Slf4j
+public record UpdateServiceDiscoveryProcessor(CollectServer collectServer) implements NettyRemotingProcessor {
+    @Override
+    public ClusterMsg.Message handle(ChannelHandlerContext ctx, ClusterMsg.Message message) {
+        ServiceDiscoveryProtocol protocol = JsonUtil.fromJson(message.getMsg(), ServiceDiscoveryProtocol.class);
+        if (Objects.isNull(protocol)) {
+            log.error("collector receive sd update task message is null");
+            return null;
+        }
+
+        collectServer.getCollectJobService().updateServiceProvider(protocol);
+        return null;
+    }
+}

collector/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/sd/ServiceDiscoveryFetchStrategy.java

@@ -0,0 +1,32 @@
+/*
+ * 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.hertzbeat.collector.dispatch.entrance.sd;
+
+import org.apache.hertzbeat.collector.collect.common.cache.sd.ConnectionConfig;
+import org.apache.hertzbeat.common.entity.sd.ServiceDiscoveryProtocol;
+
+import java.util.List;
+
+/**
+ * Service Discovery Fetch Strategy.
+ */
+public interface ServiceDiscoveryFetchStrategy {
+    List<ConnectionConfig> fetch(String target);
+
+    ServiceDiscoveryProtocol.Type getType();
+}