[improve] delete MetricsSource.java

[improve] add license header
2026-09-17 18:19:02 +00:00 · 2025-09-06 16:16:59 +08:00 · 2025-09-06 16:16:07 +08:00 · 2025-09-06 16:14:59 +08:00 · 2025-09-06 16:13:11 +08:00 · 2025-09-06 16:12:20 +08:00
32 changed files with 1803 additions and 815 deletions
@@ -1,308 +0,0 @@
-/*
- * 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.prometheus;
-
-import lombok.extern.slf4j.Slf4j;
-import org.apache.hertzbeat.collector.collect.common.http.CommonHttpClient;
-import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
-import org.apache.hertzbeat.collector.util.CollectUtil;
-import org.apache.hertzbeat.common.constants.CommonConstants;
-import org.apache.hertzbeat.common.constants.NetworkConstants;
-import org.apache.hertzbeat.common.constants.SignConstants;
-import org.apache.hertzbeat.common.entity.job.Metrics;
-import org.apache.hertzbeat.common.entity.job.protocol.PrometheusProtocol;
-import org.apache.hertzbeat.common.entity.message.CollectRep;
-import org.apache.hertzbeat.common.util.Base64Util;
-import org.apache.hertzbeat.common.util.CommonUtil;
-import org.apache.hertzbeat.common.util.IpDomainUtil;
-import org.apache.http.HttpHeaders;
-import org.apache.http.HttpHost;
-import org.apache.http.HttpStatus;
-import org.apache.http.auth.AuthScope;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.client.AuthCache;
-import org.apache.http.client.ClientProtocolException;
-import org.apache.http.client.CredentialsProvider;
-import org.apache.http.client.config.RequestConfig;
-import org.apache.http.client.methods.CloseableHttpResponse;
-import org.apache.http.client.methods.HttpUriRequest;
-import org.apache.http.client.methods.RequestBuilder;
-import org.apache.http.client.protocol.HttpClientContext;
-import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.auth.DigestScheme;
-import org.apache.http.impl.client.BasicAuthCache;
-import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.apache.http.protocol.HttpContext;
-import org.apache.http.util.EntityUtils;
-import org.springframework.http.MediaType;
-import org.springframework.util.StringUtils;
-
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.net.ConnectException;
-import java.net.UnknownHostException;
-import java.nio.charset.StandardCharsets;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import javax.net.ssl.SSLException;
-
-import static org.apache.hertzbeat.common.constants.SignConstants.RIGHT_DASH;
-
-
-@Slf4j
-public class PrometheusProxyCollectImpl implements PrometheusCollect {
-
-    private final Set<Integer> defaultSuccessStatusCodes = Stream.of(HttpStatus.SC_OK, HttpStatus.SC_CREATED,
-            HttpStatus.SC_ACCEPTED, HttpStatus.SC_MULTIPLE_CHOICES, HttpStatus.SC_MOVED_PERMANENTLY,
-            HttpStatus.SC_MOVED_TEMPORARILY).collect(Collectors.toSet());
-
-    public static final String RAW_TEXT_CONTENT_FIELD_NAME = "raw_text_content";
-
-    @Override
-    public List<CollectRep.MetricsData> collect(CollectRep.MetricsData.Builder builder, Metrics metrics) {
-        PrometheusProtocol prometheusProtocol = metrics.getPrometheus();
-        HttpUriRequest request;
-        try {
-            validateParams(metrics);
-        } catch (Exception e) {
-            builder.setCode(CollectRep.Code.FAIL);
-            builder.setMsg(e.getMessage());
-            return Collections.singletonList(builder.build());
-        }
-
-        HttpContext httpContext = createHttpContext(prometheusProtocol);
-        request = createHttpRequest(prometheusProtocol);
-
-        try (CloseableHttpResponse response = CommonHttpClient.getHttpClient().execute(request, httpContext)) {
-            int statusCode = response.getStatusLine().getStatusCode();
-            log.debug("Prometheus proxy collect, response status: {}", statusCode);
-
-            if (!defaultSuccessStatusCodes.contains(statusCode)) {
-                builder.setCode(CollectRep.Code.FAIL);
-                builder.setMsg(NetworkConstants.STATUS_CODE + SignConstants.BLANK + statusCode);
-                return Collections.singletonList(builder.build());
-            }
-
-            String rawTextContent = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
-
-            builder.clearFields();
-            builder.clearValues();
-
-            CollectRep.Field rawDataField = CollectRep.Field.newBuilder()
-                    .setName(RAW_TEXT_CONTENT_FIELD_NAME)
-                    .setType(CommonConstants.TYPE_STRING)
-                    .build();
-            builder.addField(rawDataField);
-
-            CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
-            valueRowBuilder.addColumn(rawTextContent);
-            builder.addValueRow(valueRowBuilder.build());
-
-            builder.setCode(CollectRep.Code.SUCCESS);
-        } catch (ClientProtocolException e1) {
-            String errorMsg = CommonUtil.getMessageFromThrowable(e1);
-            log.error("Prometheus proxy collect error: {}. Host: {}, Port: {}", errorMsg, prometheusProtocol.getHost(), prometheusProtocol.getPort(), e1);
-            builder.setCode(CollectRep.Code.UN_CONNECTABLE);
-            builder.setMsg(errorMsg);
-        } catch (UnknownHostException e2) {
-            String errorMsg = CommonUtil.getMessageFromThrowable(e2);
-            log.info("Prometheus proxy collect unknown host: {}. Host: {}", errorMsg, prometheusProtocol.getHost(), e2);
-            builder.setCode(CollectRep.Code.UN_REACHABLE);
-            builder.setMsg("unknown host:" + errorMsg);
-        } catch (InterruptedIOException | ConnectException | SSLException e3) {
-            String errorMsg = CommonUtil.getMessageFromThrowable(e3);
-            log.info("Prometheus proxy collect connect error: {}. Host: {}, Port: {}", errorMsg, prometheusProtocol.getHost(), prometheusProtocol.getPort(), e3);
-            builder.setCode(CollectRep.Code.UN_CONNECTABLE);
-            builder.setMsg(errorMsg);
-        } catch (IOException e4) {
-            String errorMsg = CommonUtil.getMessageFromThrowable(e4);
-            log.info("Prometheus proxy collect IO error: {}. Host: {}, Port: {}", errorMsg, prometheusProtocol.getHost(), prometheusProtocol.getPort(), e4);
-            builder.setCode(CollectRep.Code.FAIL);
-            builder.setMsg(errorMsg);
-        } catch (Exception e) {
-            String errorMsg = CommonUtil.getMessageFromThrowable(e);
-            log.error("Prometheus proxy collect unknown error: {}. Host: {}, Port: {}", errorMsg, prometheusProtocol.getHost(), prometheusProtocol.getPort(), e);
-            builder.setCode(CollectRep.Code.FAIL);
-            builder.setMsg(errorMsg);
-        } finally {
-            if (request != null) {
-                request.abort();
-            }
-        }
-        return Collections.singletonList(builder.build());
-    }
-
-    @Override
-    public String supportProtocol() {
-        return DispatchConstants.PROTOCOL_PROMETHEUS;
-    }
-
-    private void validateParams(Metrics metrics) throws Exception {
-        if (metrics == null || metrics.getPrometheus() == null) {
-            throw new Exception("Prometheus collect must has prometheus params");
-        }
-        PrometheusProtocol protocol = metrics.getPrometheus();
-        if (!StringUtils.hasText(protocol.getHost())
-            || !StringUtils.hasText(protocol.getPort())) {
-            throw new Exception("Prometheus collect must has host and port params");
-        }
-        if (protocol.getPath() == null
-                || !StringUtils.hasText(protocol.getPath())
-                || !protocol.getPath().startsWith(RIGHT_DASH)) {
-            protocol.setPath(protocol.getPath() == null ? RIGHT_DASH : RIGHT_DASH + protocol.getPath().trim());
-        }
-    }
-    
-    /**
-     * create httpContext
-     * This method is adapted from PrometheusAutoCollectImpl
-     * @param protocol prometheus protocol
-     * @return context
-     */
-    public HttpContext createHttpContext(PrometheusProtocol protocol) {
-        PrometheusProtocol.Authorization auth = protocol.getAuthorization();
-        if (auth != null && DispatchConstants.DIGEST_AUTH.equals(auth.getType())) {
-            HttpClientContext clientContext = new HttpClientContext();
-            if (StringUtils.hasText(auth.getDigestAuthUsername())
-                        && StringUtils.hasText(auth.getDigestAuthPassword())) {
-                CredentialsProvider provider = new BasicCredentialsProvider();
-                UsernamePasswordCredentials credentials =
-                        new UsernamePasswordCredentials(auth.getDigestAuthUsername(), auth.getDigestAuthPassword());
-                provider.setCredentials(AuthScope.ANY, credentials);
-                AuthCache authCache = new BasicAuthCache();
-                HttpHost targetHost = new HttpHost(protocol.getHost(), Integer.parseInt(protocol.getPort()));
-                authCache.put(targetHost, new DigestScheme());
-                clientContext.setCredentialsProvider(provider);
-                clientContext.setAuthCache(authCache);
-                return clientContext;
-            }
-        }
-        return null;
-    }
-
-    /**
-     * create http request
-     * This method is adapted from PrometheusAutoCollectImpl
-     * @param protocol http params
-     * @return http uri request
-     */
-    public HttpUriRequest createHttpRequest(PrometheusProtocol protocol) {
-        RequestBuilder requestBuilder = RequestBuilder.get();
-        // params
-        Map<String, String> params = protocol.getParams();
-        if (params != null && !params.isEmpty()) {
-            for (Map.Entry<String, String> param : params.entrySet()) {
-                if (StringUtils.hasText(param.getValue())) {
-                    requestBuilder.addParameter(param.getKey(), param.getValue());
-                }
-            }
-        }
-        requestBuilder.addHeader(HttpHeaders.CONNECTION, NetworkConstants.KEEP_ALIVE);
-        requestBuilder.addHeader(HttpHeaders.USER_AGENT, NetworkConstants.USER_AGENT);
-        // headers  The custom request header is overwritten here
-        Map<String, String> headers = protocol.getHeaders();
-        if (headers != null && !headers.isEmpty()) {
-            for (Map.Entry<String, String> header : headers.entrySet()) {
-                if (StringUtils.hasText(header.getValue())) {
-                    requestBuilder.addHeader(CollectUtil.replaceUriSpecialChar(header.getKey()),
-                            CollectUtil.replaceUriSpecialChar(header.getValue()));
-                }
-            }
-        }
-        if (headers == null || headers.keySet().stream().noneMatch(HttpHeaders.ACCEPT::equalsIgnoreCase)) {
-            requestBuilder.addHeader(HttpHeaders.ACCEPT, MediaType.TEXT_PLAIN_VALUE + ";version=0.0.4,*/*;q=0.1");
-        }
-        
-        if (protocol.getAuthorization() != null) {
-            PrometheusProtocol.Authorization authorization = protocol.getAuthorization();
-            if (DispatchConstants.BEARER_TOKEN.equalsIgnoreCase(authorization.getType())) {
-                if (StringUtils.hasText(authorization.getBearerTokenToken())) {
-                    String value = DispatchConstants.BEARER + " " + authorization.getBearerTokenToken();
-                    requestBuilder.addHeader(HttpHeaders.AUTHORIZATION, value);
-                }
-            } else if (DispatchConstants.BASIC_AUTH.equals(authorization.getType())) {
-                if (StringUtils.hasText(authorization.getBasicAuthUsername())
-                            && StringUtils.hasText(authorization.getBasicAuthPassword())) {
-                    String authStr = authorization.getBasicAuthUsername() + ":" + authorization.getBasicAuthPassword();
-                    String encodedAuth = Base64Util.encode(authStr);
-                    requestBuilder.addHeader(HttpHeaders.AUTHORIZATION, DispatchConstants.BASIC + " " + encodedAuth);
-                }
-            }
-        }
-
-        if (StringUtils.hasLength(protocol.getPayload())) {
-            requestBuilder.setEntity(new StringEntity(protocol.getPayload(), StandardCharsets.UTF_8));
-            if (headers == null || headers.keySet().stream().noneMatch(HttpHeaders.CONTENT_TYPE::equalsIgnoreCase)) {
-                requestBuilder.setHeader(HttpHeaders.CONTENT_TYPE, MediaType.TEXT_PLAIN_VALUE);
-            }
-        }
-        
-        String uriPath = CollectUtil.replaceUriSpecialChar(protocol.getPath());
-        if (IpDomainUtil.isHasSchema(protocol.getHost())) {
-            requestBuilder.setUri(protocol.getHost() + SignConstants.DOUBLE_MARK + protocol.getPort() + uriPath);
-        } else {
-            String ipAddressType = IpDomainUtil.checkIpAddressType(protocol.getHost());
-            String baseUri = NetworkConstants.IPV6.equals(ipAddressType)
-                                     ? String.format("[%s]:%s%s", protocol.getHost(), protocol.getPort(), uriPath)
-                                     : String.format("%s:%s%s", protocol.getHost(), protocol.getPort(), uriPath);
-            boolean ssl = Boolean.parseBoolean(protocol.getSsl());
-            if (ssl) {
-                requestBuilder.setUri(NetworkConstants.HTTPS_HEADER + baseUri);
-            } else {
-                requestBuilder.setUri(NetworkConstants.HTTP_HEADER + baseUri);
-            }
-        }
-        
-        // custom timeout
-        int timeout = CollectUtil.getTimeout(protocol.getTimeout());
-        if (timeout > 0) {
-            RequestConfig requestConfig = RequestConfig.custom()
-                                                  .setConnectTimeout(timeout)
-                                                  .setSocketTimeout(timeout)
-                                                  .setConnectionRequestTimeout(timeout)
-                                                  .setRedirectsEnabled(true)
-                                                  .build();
-            requestBuilder.setConfig(requestConfig);
-        } else {
-            RequestConfig requestConfig = RequestConfig.custom()
-                                                  .setRedirectsEnabled(true)
-                                                  .build();
-            requestBuilder.setConfig(requestConfig);
-        }
-        return requestBuilder.build();
-    }
-
-    /**
-     * get collect instance
-     * @return instance
-     */
-    public static PrometheusProxyCollectImpl getInstance() {
-        return PrometheusProxyCollectImpl.SingleInstance.INSTANCE;
-    }
-
-    /**
-     * static instance
-     */
-    private static class SingleInstance {
-        private static final PrometheusProxyCollectImpl INSTANCE = new PrometheusProxyCollectImpl();
-    }
-}
@@ -0,0 +1,133 @@
+/*
+ * 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;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.metrics.HertzBeatMetricsCollector;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+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.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+/**
+ *
+ */
+@Slf4j
+@Component
+public class CollectTaskTimeoutMonitor {
+    /**
+     * Collection task timeout value
+     */
+    private static final long DURATION_TIME = 240_000L;
+    /**
+     * Metrics task and start time mapping map
+     */
+    private final Map<String, MetricsTime> metricsTimeoutMonitorMap = new ConcurrentHashMap<>(16);
+
+    @Autowired
+    private HertzBeatMetricsCollector metricsCollector;
+    private CommonDispatcher commonDispatcher;
+
+    public void start(CommonDispatcher commonDispatcher) {
+        this.commonDispatcher = commonDispatcher;
+
+        // monitoring metrics collection task execution timeout
+        ThreadFactory threadFactory = new ThreadFactoryBuilder()
+                .setNameFormat("metrics-task-timeout-monitor-%d")
+                .setDaemon(true)
+                .build();
+        ScheduledThreadPoolExecutor scheduledExecutor = new ScheduledThreadPoolExecutor(1, threadFactory);
+        scheduledExecutor.scheduleWithFixedDelay(this::monitorCollectTaskTimeout, 2, 20, TimeUnit.SECONDS);
+    }
+
+    public void putMetrics(String key, MetricsTime value) {
+        this.metricsTimeoutMonitorMap.put(key, value);
+    }
+
+    public MetricsTime removeMetrics(String key) {
+        return this.metricsTimeoutMonitorMap.remove(key);
+    }
+
+    private void monitorCollectTaskTimeout() {
+        try {
+            // Detect whether the collection unit of each metrics has timed out for 4 minutes,
+            // and if it times out, it will be discarded and an exception will be returned.
+            long deadline = System.currentTimeMillis() - DURATION_TIME;
+            for (Map.Entry<String, MetricsTime> entry : metricsTimeoutMonitorMap.entrySet()) {
+                MetricsTime metricsTime = entry.getValue();
+                if (metricsTime.getStartTime() < deadline) {
+                    // Metrics collection timeout
+                    MetricsTime removedMetricsTime = metricsTimeoutMonitorMap.remove(entry.getKey());
+                    if (removedMetricsTime == null) {
+                        continue;
+                    }
+                    WheelTimerTask timerJob = (WheelTimerTask) metricsTime.getTimeout().task();
+                    Job job = timerJob.getJob();
+                    // timeout metrics
+                    if (metricsCollector != null) {
+                        long duration = System.currentTimeMillis() - removedMetricsTime.getStartTime();
+                        metricsCollector.recordCollectMetrics(job, duration, "timeout");
+                    }
+
+                    CollectRep.MetricsData metricsData = CollectRep.MetricsData.newBuilder()
+                            .setId(job.getMonitorId())
+                            .setTenantId(job.getTenantId())
+                            .setApp(job.getApp())
+                            .setMetrics(metricsTime.getMetrics().getName())
+                            .setPriority(metricsTime.getMetrics().getPriority())
+                            .setTime(System.currentTimeMillis())
+                            .setCode(CollectRep.Code.TIMEOUT)
+                            .setMsg("collect timeout")
+                            .build();
+                    log.error("[Collect Timeout]: \n{}", metricsData);
+                    if (metricsData.getPriority() == CommonConstants.AVAILABLE_METRICS) {
+                        //todo 使用chain bootstrap
+//                        commonDispatcher.dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData);
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.error("[Task Timeout Monitor]-{}.", e.getMessage(), e);
+        }
+    }
+
+    /**
+     * Metrics times.
+     */
+    @Data
+    @AllArgsConstructor
+    public static class MetricsTime {
+        private long startTime;
+        private Metrics metrics;
+        private Timeout timeout;
+    }
+}
@@ -17,13 +17,23 @@

 package org.apache.hertzbeat.collector.dispatch;

-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.gson.Gson;
-import lombok.AllArgsConstructor;
-import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.context.impl.DefaultContext;
 import org.apache.hertzbeat.collector.dispatch.entrance.internal.CollectJobService;
+import org.apache.hertzbeat.collector.handler.ChainBootstrap;
+import org.apache.hertzbeat.collector.handler.CollectMetricsDataDataStream;
+import org.apache.hertzbeat.collector.listener.CalculateFieldsListener;
+import org.apache.hertzbeat.collector.listener.MetricsDataDeliveryListener;
+import org.apache.hertzbeat.collector.listener.RemoveTimeoutMonitorListener;
+import org.apache.hertzbeat.collector.listener.RerunDataStream;
+import org.apache.hertzbeat.collector.listener.ResponseJobDataListener;
+import org.apache.hertzbeat.collector.listener.ValidateResponseListener;
 import org.apache.hertzbeat.collector.metrics.HertzBeatMetricsCollector;
+import org.apache.hertzbeat.collector.handler.impl.BatchExecuteTaskChain;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.collector.CollectorMetaData;
 import org.apache.hertzbeat.common.timer.Timeout;
 import org.apache.hertzbeat.collector.timer.TimerDispatch;
 import org.apache.hertzbeat.collector.timer.WheelTimerTask;
@@ -37,17 +47,18 @@ import org.apache.hertzbeat.common.queue.CommonDataQueue;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;

+import java.util.Collections;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;

 /**
 * Collection task and response data scheduler
@@ -56,10 +67,6 @@ import java.util.concurrent.atomic.AtomicReference;
@Slf4j
 public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatch {

-    /**
-     * Collection task timeout value
-     */
-    private static final long DURATION_TIME = 240_000L;
    /**
     * Trigger sub task max num
     */
@@ -81,16 +88,14 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
     * collection data exporter
     */
    private final CommonDataQueue commonDataQueue;
-    /**
-     * Metrics task and start time mapping map
-     */
-    private final Map<String, MetricsTime> metricsTimeoutMonitorMap;

    private final List<UnitConvert> unitConvertList;

    private final WorkerPool workerPool;

-    private final String collectorIdentity;
+    private final CollectorMetaData metaData;
+
+    private final CollectTaskTimeoutMonitor collectTaskTimeoutMonitor;

    @Autowired
    private HertzBeatMetricsCollector metricsCollector;
@@ -100,17 +105,24 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
                            CommonDataQueue commonDataQueue,
                            WorkerPool workerPool,
                            CollectJobService collectJobService,
-                            List<UnitConvert> unitConvertList) {
+                            List<UnitConvert> unitConvertList,
+                            CollectTaskTimeoutMonitor collectTaskTimeoutMonitor) {
        this.commonDataQueue = commonDataQueue;
        this.jobRequestQueue = jobRequestQueue;
        this.timerDispatch = timerDispatch;
        this.unitConvertList = unitConvertList;
        this.workerPool = workerPool;
-        this.collectorIdentity = collectJobService.getCollectorIdentity();
-        this.metricsTimeoutMonitorMap = new ConcurrentHashMap<>(16);
-        this.start();
+        this.metaData = CollectorMetaData.builder()
+                .identity(collectJobService.getCollectorIdentity())
+                .mode(collectJobService.getCollectorMode())
+                .startTime(new Date())
+                .build();
+        this.collectTaskTimeoutMonitor = collectTaskTimeoutMonitor;
+//        this.start();
+        this.collectTaskTimeoutMonitor.start(this);
    }

+    @Deprecated
    public void start() {
        try {
            // Pull the collection task from the task queue and put it into the thread pool for execution
@@ -138,91 +150,104 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
                }
                log.info("Thread Interrupted, Shutdown the [metrics-task-dispatcher]");
            });
-            // monitoring metrics collection task execution timeout
-            ThreadFactory threadFactory = new ThreadFactoryBuilder()
-                    .setNameFormat("metrics-task-timeout-monitor-%d")
-                    .setDaemon(true)
-                    .build();
-            ScheduledThreadPoolExecutor scheduledExecutor = new ScheduledThreadPoolExecutor(1, threadFactory);
-            scheduledExecutor.scheduleWithFixedDelay(this::monitorCollectTaskTimeout, 2, 20, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("Common Dispatcher error: {}.", e.getMessage(), e);
        }
    }

-    private void monitorCollectTaskTimeout() {
-        try {
-            // Detect whether the collection unit of each metrics has timed out for 4 minutes,
-            // and if it times out, it will be discarded and an exception will be returned.
-            long deadline = System.currentTimeMillis() - DURATION_TIME;
-            for (Map.Entry<String, MetricsTime> entry : metricsTimeoutMonitorMap.entrySet()) {
-                MetricsTime metricsTime = entry.getValue();
-                if (metricsTime.getStartTime() < deadline) {
-                    // Metrics collection timeout
-                    MetricsTime removedMetricsTime = metricsTimeoutMonitorMap.remove(entry.getKey());
-                    if (removedMetricsTime == null) {
-                        continue;
-                    }
-                    WheelTimerTask timerJob = (WheelTimerTask) metricsTime.getTimeout().task();
-                    Job job = timerJob.getJob();
-                    // timeout metrics
-                    if (metricsCollector != null) {
-                        long duration = System.currentTimeMillis() - removedMetricsTime.getStartTime();
-                        metricsCollector.recordCollectMetrics(job, duration, "timeout");
-                    }
-
-                    CollectRep.MetricsData metricsData = CollectRep.MetricsData.newBuilder()
-                            .setId(job.getMonitorId())
-                            .setTenantId(job.getTenantId())
-                            .setApp(job.getApp())
-                            .setMetrics(metricsTime.getMetrics().getName())
-                            .setPriority(metricsTime.getMetrics().getPriority())
-                            .setTime(System.currentTimeMillis())
-                            .setCode(CollectRep.Code.TIMEOUT).setMsg("collect timeout").build();
-                    log.error("[Collect Timeout]: \n{}", metricsData);
-                    if (metricsData.getPriority() == 0) {
-                        dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData);
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("[Task Timeout Monitor]-{}.", e.getMessage(), e);
-        }
-    }
-
    @Override
    public void dispatchMetricsTask(Timeout timeout) {
        // Divide the collection task of a single application into corresponding collection tasks of the metrics under it.
        // Put each collect task into the thread pool for scheduling
        WheelTimerTask timerTask = (WheelTimerTask) timeout.task();
        Job job = timerTask.getJob();
-        job.constructPriorMetrics();
-        Set<Metrics> metricsSet = job.getNextCollectMetrics(null, true);
-        metricsSet.forEach(metrics -> {
-            MetricsCollect metricsCollect = new MetricsCollect(metrics, timeout, this,
-                    collectorIdentity, unitConvertList);
-            jobRequestQueue.addJob(metricsCollect);
-            if (metrics.getPrometheus() != null) {
-                metricsTimeoutMonitorMap.put(String.valueOf(job.getId()),
-                        new MetricsTime(System.currentTimeMillis(), metrics, timeout));
-            } else {
-                metricsTimeoutMonitorMap.put(job.getId() + "-" + metrics.getName(),
-                        new MetricsTime(System.currentTimeMillis(), metrics, timeout));
-            }
-        });
+
+        ChainBootstrap bootstrap = constructMetricsCollectTaskChain(job);
+        //todo context需要划分作用域
+        bootstrap.addContext(ContextKey.META_DATA, metaData)
+                .addContext(ContextKey.JOB, job)
+                .addContext(ContextKey.TIMEOUT, timeout)
+                .addListener(new CalculateFieldsListener(unitConvertList))
+                .addListener(new ValidateResponseListener())
+                .onEachDataStreamComplete(new RemoveTimeoutMonitorListener(collectTaskTimeoutMonitor));
+
+        if (job.isCyclic()) {
+            bootstrap.withWorkerPool(workerPool)
+                    .addListener(new MetricsDataDeliveryListener(commonDataQueue))
+                    .onComplete(new RerunDataStream(timerDispatch));
+        } else {
+            bootstrap.addListener(new ResponseJobDataListener(timerDispatch));
+        }
+
+
+        bootstrap.start();
    }

+    private ChainBootstrap constructMetricsCollectTaskChain(Job job) {
+        long now = System.currentTimeMillis();
+        Map<Byte, List<Metrics>> currentCollectMetrics = job.getMetrics().stream()
+                .filter(metrics -> (now >= metrics.getCollectTime() + metrics.getInterval() * 1000L))
+                .peek(metric -> {
+                    metric.setCollectTime(now);
+                    // Determine whether to configure aliasFields If not, configure the default
+                    if ((metric.getAliasFields() == null || metric.getAliasFields().isEmpty()) && metric.getFields() != null) {
+                        metric.setAliasFields(metric.getFields().stream().map(Metrics.Field::getField).collect(Collectors.toList()));
+                    }
+                    // Set the default metrics execution priority, if not filled, the default last priority
+                    if (metric.getPriority() == null) {
+                        metric.setPriority(Byte.MAX_VALUE);
+                    }
+                })
+                .collect(Collectors.groupingBy(Metrics::getPriority));
+
+        // the current collect metrics can not empty, if empty, add a default availability metrics
+        // due the metric collect is trigger by the previous metric collect
+        if (currentCollectMetrics.isEmpty()) {
+            Optional<Metrics> defaultMetricOption = job.getMetrics().stream()
+                    .filter(metric -> metric.getPriority() == CommonConstants.AVAILABLE_METRICS).findFirst();
+            if (defaultMetricOption.isPresent()) {
+                Metrics defaultMetric = defaultMetricOption.get();
+                defaultMetric.setCollectTime(now);
+                currentCollectMetrics.put(CommonConstants.AVAILABLE_METRICS, Collections.singletonList(defaultMetric));
+            } else {
+                log.error("metrics must has one priority 0 metrics at least.");
+            }
+        }
+
+        ChainBootstrap chainBootstrap = ChainBootstrap.withContext(DefaultContext.newInstance())
+                .withChain(new BatchExecuteTaskChain<Metrics>());
+        // order by priority
+        currentCollectMetrics.keySet().stream()
+                .sorted()
+                .forEach(priority -> {
+                    if (job.isCyclic() || isOneTimeJobAndIsAvailableMetrics(job, priority)) {
+                        List<Metrics> metricsList = currentCollectMetrics.get(priority);
+                        CollectMetricsDataDataStream collectHandler = CollectMetricsDataDataStream.builder()
+                                .collectTaskTimeoutMonitor(collectTaskTimeoutMonitor)
+                                .build();
+                        collectHandler.setSourceDataList(metricsList);
+
+                        chainBootstrap.addDataStream(collectHandler);
+                    }
+                });
+
+        return chainBootstrap;
+    }
+
+    private boolean isOneTimeJobAndIsAvailableMetrics(Job job, byte priority) {
+        return (!job.isCyclic()) && priority == CommonConstants.AVAILABLE_METRICS;
+    }
+
+    @Deprecated
    @Override
    public void dispatchCollectData(Timeout timeout, Metrics metrics, CollectRep.MetricsData metricsData) {
        WheelTimerTask timerJob = (WheelTimerTask) timeout.task();
        Job job = timerJob.getJob();
-        String monitorKey;
+        String monitorKey = job.getId() + "-" + metrics.getName();
        if (metrics.isHasSubTask()) {
-            monitorKey = job.getId() + "-" + metrics.getName() + "-sub-" + metrics.getSubTaskId();
-        } else {
-            monitorKey = job.getId() + "-" + metrics.getName();
+            monitorKey = monitorKey + "-sub-" + metrics.getSubTaskId();
        }
-        MetricsTime metricsTime = metricsTimeoutMonitorMap.remove(monitorKey);
+        CollectTaskTimeoutMonitor.MetricsTime metricsTime = this.collectTaskTimeoutMonitor.removeMetrics(monitorKey);

        // job completed metrics
        if (metricsTime != null && metricsCollector != null) {
@@ -238,31 +263,28 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
                return;
            }
        }
+
+
        Set<Metrics> metricsSet = job.getNextCollectMetrics(metrics, false);
        if (job.isCyclic()) {
-            if (log.isDebugEnabled()) {
-                log.debug("Cyclic Job: {} - {} - {}", job.getMonitorId(), job.getApp(), metricsData.getMetrics());
-                for (CollectRep.ValueRow valueRow : metricsData.getValues()) {
-                    for (CollectRep.Field field : metricsData.getFields()) {
-                        log.debug("Field-->{},Value-->{}", field.getName(), valueRow.getColumns(metricsData.getFields().indexOf(field)));
-                    }
-                }
-            }
+            cyclicJobDebugLog(job, metricsData);
+
            // If metricsSet is null, it means that the execution is completed or whether the priority of the collection metrics is 0, that is, the availability collection metrics.
            // If the availability collection fails, the next metrics scheduling will be cancelled and the next round of scheduling will be entered directly.
            boolean isAvailableCollectFailed = metricsSet != null && !metricsSet.isEmpty()
-                    && metrics.getPriority() == (byte) 0 && metricsData.getCode() != CollectRep.Code.SUCCESS;
+                    && metrics.getPriority() == CommonConstants.AVAILABLE_METRICS && metricsData.getCode() != CollectRep.Code.SUCCESS;
            if (metricsSet == null || isAvailableCollectFailed || job.isSd()) {
                // The collection and execution task of this job are completed.
                // The periodic task pushes the task to the time wheel again.
                // First, determine the execution time of the task and the task collection interval.
                if (!timeout.isCancelled()) {
                    long spendTime = System.currentTimeMillis() - job.getDispatchTime();
-                    long interval = job.getInterval() - spendTime / 1000;
+                    long interval = job.getInterval() - spendTime / 1000L;
                    interval = interval <= 0 ? 0 : interval;
                    timerDispatch.cyclicJob(timerJob, interval, TimeUnit.SECONDS);
                }
-            } else if (!metricsSet.isEmpty()) {
+            }
+            else if (!metricsSet.isEmpty()) {
                // The execution of the current level metrics is completed, and the execution of the next level metrics starts
                // use pre collect metrics data to replace next metrics config params
                List<Map<String, Configmap>> configmapList = CollectUtil.getConfigmapFromPreCollectData(metricsData);
@@ -273,12 +295,15 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
                    Set<String> cryPlaceholderFields = CollectUtil.matchCryPlaceholderField(GSON.toJsonTree(metricItem));
                    if (cryPlaceholderFields.isEmpty()) {
                        MetricsCollect metricsCollect = new MetricsCollect(metricItem, timeout, this,
-                                collectorIdentity, unitConvertList);
+                                metaData.getIdentity(), unitConvertList);
                        jobRequestQueue.addJob(metricsCollect);
-                        metricsTimeoutMonitorMap.put(job.getId() + "-" + metricItem.getName(),
-                                new MetricsTime(System.currentTimeMillis(), metricItem, timeout));
+
+                        this.collectTaskTimeoutMonitor.putMetrics(job.getId() + "-" + metricItem.getName(),
+                                new CollectTaskTimeoutMonitor.MetricsTime(System.currentTimeMillis(), metricItem, timeout));
                        continue;
                    }
+
+
                    boolean isSubTask = configmapList.stream().anyMatch(map -> map.keySet().stream().anyMatch(cryPlaceholderFields::contains));
                    int subTaskNum = isSubTask ? Math.min(configmapList.size(), MAX_SUB_TASK_NUM) : 1;
                    AtomicInteger subTaskNumAtomic = new AtomicInteger(subTaskNum);
@@ -294,95 +319,30 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc
                        metric.setSubTaskId(index);
                        metric.setSubTaskDataRef(metricsDataReference);
                        MetricsCollect metricsCollect = new MetricsCollect(metric, timeout, this,
-                                collectorIdentity, unitConvertList);
+                                metaData.getIdentity(), unitConvertList);
                        jobRequestQueue.addJob(metricsCollect);
-                        metricsTimeoutMonitorMap.put(job.getId() + "-" + metric.getName() + "-sub-" + index,
-                                new MetricsTime(System.currentTimeMillis(), metric, timeout));
+
+                        this.collectTaskTimeoutMonitor.putMetrics(job.getId() + "-" + metric.getName() + "-sub-" + index,
+                                new CollectTaskTimeoutMonitor.MetricsTime(System.currentTimeMillis(), metric, timeout));
                    }

                }
-            } else {
-                // The list of metrics at the current execution level has not been fully executed.
-                // It needs to wait for the execution of other metrics task of the same level to complete the execution and enter the next level for execution.
-            }
-            // If it is an asynchronous periodic cyclic task, directly response the collected data
-            if (job.isSd()) {
-                CollectRep.MetricsData sdMetricsData = CollectRep.MetricsData.newBuilder(metricsData).build();
-                commonDataQueue.sendServiceDiscoveryData(sdMetricsData);
-            }
-            commonDataQueue.sendMetricsData(metricsData);
-        } else {
-            // If it is a temporary one-time task, you need to wait for the collected data of all metrics task to be packaged and returned.
-            // Insert the current metrics data into the job for unified assembly
-            job.addCollectMetricsData(metricsData);
-            if (log.isDebugEnabled()) {
-                log.debug("One-time Job: {}", metricsData.getMetrics());
-                for (CollectRep.ValueRow valueRow : metricsData.getValues()) {
-                    for (CollectRep.Field field : metricsData.getFields()) {
-                        log.debug("Field-->{},Value-->{}", field.getName(), valueRow.getColumns(metricsData.getFields().indexOf(field)));
-                    }
-                }
-            }
-
-            if (job.isSd() || metricsSet == null) {
-                // The collection and execution of all metrics of this job are completed
-                // and the result listener is notified of the combination of all metrics data
-                timerDispatch.responseSyncJobData(job.getId(), job.getResponseDataTemp());
-            } else if (!metricsSet.isEmpty()) {
-                // The execution of the current level metrics is completed, and the execution of the next level metrics starts
-                metricsSet.forEach(metricItem -> {
-                    MetricsCollect metricsCollect = new MetricsCollect(metricItem, timeout, this,
-                            collectorIdentity, unitConvertList);
-                    jobRequestQueue.addJob(metricsCollect);
-                    metricsTimeoutMonitorMap.put(job.getId() + "-" + metricItem.getName(),
-                            new MetricsTime(System.currentTimeMillis(), metricItem, timeout));
-                });
-            } else {
-                // The list of metrics task at the current execution level has not been fully executed.
-                // It needs to wait for the execution of other metrics task of the same level to complete the execution and enter the next level for execution.
            }
        }
    }

-    @Override
-    public void dispatchCollectData(Timeout timeout, Metrics metrics, List<CollectRep.MetricsData> metricsDataList) {
-        WheelTimerTask timerJob = (WheelTimerTask) timeout.task();
-        Job job = timerJob.getJob();
-        MetricsTime metricsTime = metricsTimeoutMonitorMap.remove(String.valueOf(job.getId()));
-        if (metricsTime != null && metricsCollector != null) {
-            long duration = System.currentTimeMillis() - metricsTime.getStartTime();
-            // For a list, we consider it a success if at least one item is successful.
-            boolean isSuccess = metricsDataList.stream().anyMatch(item -> item.getCode() == CollectRep.Code.SUCCESS);
-            metricsCollector.recordCollectMetrics(job, duration, isSuccess ? "success" : "fail");
-        }
-        if (job.isCyclic()) {
-            // The collection and execution of all task of this job are completed.
-            // The periodic task pushes the task to the time wheel again.
-            // First, determine the execution time of the task and the task collection interval.
-            if (!timeout.isCancelled()) {
-                long spendTime = System.currentTimeMillis() - job.getDispatchTime();
-                long interval = job.getInterval() - spendTime / 1000;
-                interval = interval <= 0 ? 0 : interval;
-                timerDispatch.cyclicJob(timerJob, interval, TimeUnit.SECONDS);
-            }
-            // it is an asynchronous periodic cyclic task, directly response the collected data
-            metricsDataList.forEach(commonDataQueue::sendMetricsData);
-        } else {
-            // The collection and execution of all metrics of this job are completed
-            // and the result listener is notified of the combination of all metrics data
-            timerDispatch.responseSyncJobData(job.getId(), metricsDataList);
+    private void cyclicJobDebugLog(Job job, CollectRep.MetricsData metricsData) {
+        if (log.isDebugEnabled()) {
+            log.debug("Cyclic Job: {} - {} - {}", job.getMonitorId(), job.getApp(), metricsData.getMetrics());
+            metricsDataDebugLog(metricsData);
        }
    }

-
-    /**
-     * Metrics times.
-     */
-    @Data
-    @AllArgsConstructor
-    protected static class MetricsTime {
-        private long startTime;
-        private Metrics metrics;
-        private Timeout timeout;
+    private void metricsDataDebugLog(CollectRep.MetricsData metricsData) {
+        for (CollectRep.ValueRow valueRow : metricsData.getValues()) {
+            for (CollectRep.Field field : metricsData.getFields()) {
+                log.debug("Field-->{},Value-->{}", field.getName(), valueRow.getColumns(metricsData.getFields().indexOf(field)));
+            }
+        }
    }
 }
@@ -129,7 +129,6 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {
        this.newTime = System.currentTimeMillis();
        this.timeout = timeout;
        this.metrics = metrics;
-        this.collectorIdentity = collectorIdentity;
        WheelTimerTask timerJob = (WheelTimerTask) timeout.task();
        Job job = timerJob.getJob();
        this.id = job.getMonitorId();
@@ -153,327 +152,9 @@ public class MetricsCollect implements Runnable, Comparable<MetricsCollect> {

    @Override
    public void run() {
-        this.startTime = System.currentTimeMillis();
-        setNewThreadName(id, app, startTime, metrics);
-        CollectRep.MetricsData.Builder response = CollectRep.MetricsData.newBuilder();
-        response.setApp(app).setId(id).setTenantId(tenantId)
-                .setLabels(labels).setAnnotations(annotations).addMetadataAll(metadata);
-        // for prometheus auto or proxy mode
-        if (DispatchConstants.PROTOCOL_PROMETHEUS.equalsIgnoreCase(metrics.getProtocol())) {
-            List<CollectRep.MetricsData> metricsData;
-
-            // TODO: Refactor Prometheus metrics collection logic.
-            // The current implementation for proxy mode and auto mode needs review and potential simplification.
-            // Consider a more unified approach or clarify the conditions for each mode.
-            /*
-            // TODO USE PROXY MODE
-            if (prometheusProxyMode) {
-                List<CollectRep.MetricsData> proxyData = PrometheusProxyCollectImpl.getInstance().collect(response, metrics);
-                List<CollectRep.MetricsData> autoData = PrometheusAutoCollectImpl.getInstance().collect(response, metrics);
-                metricsData = new LinkedList<>();
-                if (proxyData != null) {
-                    metricsData.addAll(proxyData);
-                }
-                if (autoData != null) {
-                    metricsData.addAll(autoData);
-                }
-            } else {
-                metricsData = PrometheusAutoCollectImpl.getInstance().collect(response, metrics);
-            }
-            */
-            metricsData = PrometheusAutoCollectImpl.getInstance().collect(response, metrics);
-            validateResponse(metricsData == null ? null : metricsData.stream().findFirst().orElse(null));
-            collectDataDispatch.dispatchCollectData(timeout, metrics, metricsData);
-            return;
-        }
-        response.setMetrics(metrics.getName());
-        // According to the metrics collection protocol, application type, etc.,
-        // dispatch to the real application metrics collection implementation class
-        AbstractCollect abstractCollect = CollectStrategyFactory.invoke(metrics.getProtocol());
-        if (abstractCollect == null) {
-            log.error("[Dispatcher] - not support this: app: {}, metrics: {}, protocol: {}.",
-                    app, metrics.getName(), metrics.getProtocol());
-            response.setCode(CollectRep.Code.FAIL);
-            response.setMsg("not support " + app + ", "
-                    + metrics.getName() + ", " + metrics.getProtocol());
-        } else {
-            try {
-                abstractCollect.preCheck(metrics);
-                abstractCollect.collect(response, metrics);
-            } catch (Exception e) {
-                String msg = e.getMessage();
-                if (msg == null && e.getCause() != null) {
-                    msg = e.getCause().getMessage();
-                }
-                if (e instanceof IllegalArgumentException) {
-                    log.error("[Metrics PreCheck]: {}.", msg, e);
-                } else {
-                    log.error("[Metrics Collect]: {}.", msg, e);
-                }
-                response.setCode(CollectRep.Code.FAIL);
-                if (msg != null) {
-                    response.setMsg(msg);
-                }
-            }
-        }
-        // Alias attribute expression replacement calculation
-        if (fastFailed()) {
-            return;
-        }
-        calculateFields(metrics, response);
-        CollectRep.MetricsData metricsData = validateResponse(response);
-        collectDataDispatch.dispatchCollectData(timeout, metrics, metricsData);
+//        collectDataDispatch.dispatchCollectData(timeout, metrics, metricsData);
    }

-    /**
-     * Calculate the real metrics value according to the calculates and aliasFields configuration
-     *
-     * @param metrics     Metrics configuration
-     * @param collectData Data collection
-     */
-    public void calculateFields(Metrics metrics, CollectRep.MetricsData.Builder collectData) {
-        collectData.setPriority(metrics.getPriority());
-        List<CollectRep.Field> fieldList = new LinkedList<>();
-        for (Metrics.Field field : metrics.getFields()) {
-            CollectRep.Field.Builder fieldBuilder = CollectRep.Field.newBuilder();
-            fieldBuilder.setName(field.getField()).setType(field.getType()).setLabel(field.isLabel());
-            if (field.getUnit() != null) {
-                fieldBuilder.setUnit(field.getUnit());
-            }
-            fieldList.add(fieldBuilder.build());
-        }
-        collectData.addAllFields(fieldList);
-        List<CollectRep.ValueRow> aliasRowList = collectData.getValuesList();
-        if (aliasRowList == null || aliasRowList.isEmpty()) {
-            return;
-        }
-        collectData.clearValues();
-        // Preprocess calculates first
-        if (metrics.getCalculates() == null) {
-            metrics.setCalculates(Collections.emptyList());
-        }
-        // eg: database_pages=Database pages unconventional mapping
-        Map<String, String> fieldAliasMap = new HashMap<>(8);
-        Map<String, JexlExpression> fieldExpressionMap = metrics.getCalculates()
-                .stream()
-                .map(cal -> transformCal(cal, fieldAliasMap))
-                .filter(Objects::nonNull)
-                .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (JexlExpression) arr[1], (oldValue, newValue) -> newValue));
-
-        if (metrics.getUnits() == null) {
-            metrics.setUnits(Collections.emptyList());
-        }
-        Map<String, Pair<String, String>> fieldUnitMap = metrics.getUnits()
-                .stream()
-                .map(this::transformUnit)
-                .filter(Objects::nonNull)
-                .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (Pair<String, String>) arr[1], (oldValue, newValue) -> newValue));
-
-        List<Metrics.Field> fields = metrics.getFields();
-        List<String> aliasFields = Optional.ofNullable(metrics.getAliasFields()).orElseGet(Collections::emptyList);
-        Map<String, String> aliasFieldValueMap = new HashMap<>(8);
-        Map<String, Object> fieldValueMap = new HashMap<>(8);
-        Map<String, Object> stringTypefieldValueMap = new HashMap<>(8);
-        Map<String, String> aliasFieldUnitMap = new HashMap<>(8);
-        CollectRep.ValueRow.Builder realValueRowBuilder = CollectRep.ValueRow.newBuilder();
-        for (CollectRep.ValueRow aliasRow : aliasRowList) {
-            for (int aliasIndex = 0; aliasIndex < aliasFields.size(); aliasIndex++) {
-                String aliasFieldValue = aliasRow.getColumns(aliasIndex);
-                String aliasField = aliasFields.get(aliasIndex);
-                if (!CommonConstants.NULL_VALUE.equals(aliasFieldValue)) {
-                    aliasFieldValueMap.put(aliasField, aliasFieldValue);
-                    // whether the alias field is a number
-                    CollectUtil.DoubleAndUnit doubleAndUnit = CollectUtil
-                            .extractDoubleAndUnitFromStr(aliasFieldValue);
-                    if (doubleAndUnit != null && doubleAndUnit.getValue() != null) {
-                        fieldValueMap.put(aliasField, doubleAndUnit.getValue());
-                        if (doubleAndUnit.getUnit() != null) {
-                            aliasFieldUnitMap.put(aliasField, doubleAndUnit.getUnit());
-                        }
-                    } else {
-                        fieldValueMap.put(aliasField, aliasFieldValue);
-                    }
-                    stringTypefieldValueMap.put(aliasField, aliasFieldValue);
-                } else {
-                    fieldValueMap.put(aliasField, null);
-                    stringTypefieldValueMap.put(aliasField, null);
-                }
-            }
-
-            for (Metrics.Field field : fields) {
-                String realField = field.getField();
-                JexlExpression expression = fieldExpressionMap.get(realField);
-                String value = null;
-                String aliasFieldUnit = null;
-                if (expression != null) {
-                    try {
-                        Map<String, Object> context;
-                        if (CommonConstants.TYPE_STRING == field.getType()) {
-                            context = stringTypefieldValueMap;
-                        } else {
-                            for (Map.Entry<String, String> unitEntry : aliasFieldUnitMap.entrySet()) {
-                                if (expression.getSourceText().contains(unitEntry.getKey())) {
-                                    aliasFieldUnit = unitEntry.getValue();
-                                    break;
-                                }
-                            }
-                            context = fieldValueMap;
-                        }
-
-                        // Also executed when valueList is empty, covering pure string assignment expressions
-                        Object objValue = JexlExpressionRunner.evaluate(expression, context);
-
-                        if (objValue != null) {
-                            value = String.valueOf(objValue);
-                        }
-                    } catch (Exception e) {
-                        log.warn("[calculates execute warning, use original value.] {}", e.getMessage());
-                        value = Optional.ofNullable(fieldValueMap.get(expression.getSourceText()))
-                                .map(String::valueOf)
-                                .orElse(null);
-                    }
-                } else {
-                    // does not exist then map the alias value
-                    String aliasField = fieldAliasMap.get(realField);
-                    if (aliasField != null) {
-                        value = aliasFieldValueMap.get(aliasField);
-                    } else {
-                        value = aliasFieldValueMap.get(realField);
-                    }
-
-                    if (value != null) {
-                        final byte fieldType = field.getType();
-                        if (fieldType == CommonConstants.TYPE_NUMBER) {
-                            CollectUtil.DoubleAndUnit doubleAndUnit = CollectUtil
-                                    .extractDoubleAndUnitFromStr(value);
-                            final Double tempValue = doubleAndUnit == null ? null : doubleAndUnit.getValue();
-                            value = tempValue == null ? null : String.valueOf(tempValue);
-                            aliasFieldUnit = doubleAndUnit == null ? null : doubleAndUnit.getUnit();
-                        } else if (fieldType == CommonConstants.TYPE_TIME) {
-                            final int tempValue;
-                            value = (tempValue = CommonUtil.parseTimeStrToSecond(value)) == -1 ? null : String.valueOf(tempValue);
-                        }
-                    }
-                }
-
-                Pair<String, String> unitPair = fieldUnitMap.get(realField);
-                if (aliasFieldUnit != null) {
-                    if (unitPair != null) {
-                        unitPair.setLeft(aliasFieldUnit);
-                    } else if (field.getUnit() != null && !aliasFieldUnit.equalsIgnoreCase(field.getUnit())) {
-                        unitPair = Pair.of(aliasFieldUnit, field.getUnit());
-                    }
-                }
-                if (value != null && unitPair != null) {
-                    for (UnitConvert unitConvert : unitConvertList) {
-                        if (unitConvert.checkUnit(unitPair.getLeft()) && unitConvert.checkUnit(unitPair.getRight())) {
-                            value = unitConvert.convert(value, unitPair.getLeft(), unitPair.getRight());
-                        }
-                    }
-                }
-                // Handle metrics values that may have units such as 34%, 34Mb, and limit values to 4 decimal places
-                if (CommonConstants.TYPE_NUMBER == field.getType()) {
-                    value = CommonUtil.parseDoubleStr(value, field.getUnit());
-                }
-                if (value == null) {
-                    value = CommonConstants.NULL_VALUE;
-                }
-                realValueRowBuilder.addColumn(value);
-            }
-            aliasFieldValueMap.clear();
-            fieldValueMap.clear();
-            aliasFieldUnitMap.clear();
-            stringTypefieldValueMap.clear();
-            CollectRep.ValueRow realValueRow = realValueRowBuilder.build();
-            realValueRowBuilder.clear();
-            // apply filter calculation to the real value row
-            if (!CollectionUtils.isEmpty(metrics.getFilters())) {
-                Map<String, Object> contextMap = new HashMap<>(8);
-                for (int i = 0; i < fields.size(); i++) {
-                    Metrics.Field field = fields.get(i);
-                    String value = realValueRow.getColumns(i);
-                    contextMap.put(field.getField(), value);
-                }
-                boolean isMatch = false;
-                for (String filterExpr : metrics.getFilters()) {
-                    try {
-                        JexlExpression expression = JexlExpressionRunner.compile(filterExpr);
-                        if ((Boolean) JexlExpressionRunner.evaluate(expression, contextMap)) {
-                            isMatch = true;
-                            break;
-                        }
-                    } catch (Exception e) {
-                        log.warn("[metrics data row filters execute warning] {}.", e.getMessage());
-                    }
-                }
-                if (!isMatch) {
-                    // ignore this data row
-                    continue;
-                }
-            }
-            collectData.addValueRow(realValueRow);
-        }
-    }
-
-    /**
-     * @param cal           cal
-     * @param fieldAliasMap field alias map
-     * @return expr
-     */
-    private Object[] transformCal(String cal, Map<String, String> fieldAliasMap) {
-        int splitIndex = cal.indexOf("=");
-        if (splitIndex < 0) {
-            return null;
-        }
-        String field = cal.substring(0, splitIndex).trim();
-        String expressionStr = cal.substring(splitIndex + 1).trim().replace("\\#", "#");
-        JexlExpression expression;
-        try {
-            expression = JexlExpressionRunner.compile(expressionStr);
-        } catch (Exception e) {
-            fieldAliasMap.put(field, expressionStr);
-            return null;
-        }
-        return new Object[]{field, expression};
-    }
-
-    /**
-     * transform unit
-     *
-     * @param unit unit
-     * @return units
-     */
-    private Object[] transformUnit(String unit) {
-        int equalIndex = unit.indexOf("=");
-        int arrowIndex = unit.indexOf("->");
-        if (equalIndex < 0 || arrowIndex < 0) {
-            return null;
-        }
-        String field = unit.substring(0, equalIndex).trim();
-        String originUnit = unit.substring(equalIndex + 1, arrowIndex).trim();
-        String newUnit = unit.substring(arrowIndex + 2).trim();
-        return new Object[]{field, Pair.of(originUnit, newUnit)};
-    }
-
-    private boolean fastFailed() {
-        return this.timeout == null || this.timeout.isCancelled();
-    }
-
-    private CollectRep.MetricsData validateResponse(CollectRep.MetricsData.Builder builder) {
-        long endTime = System.currentTimeMillis();
-        builder.setTime(endTime);
-        long runningTime = endTime - startTime;
-        long allTime = endTime - newTime;
-        if (startTime - newTime >= WARN_DISPATCH_TIME) {
-            log.warn("[Collector Dispatch Warn, Dispatch Use {}ms.", startTime - newTime);
-        }
-        if (builder.getCode() != CollectRep.Code.SUCCESS) {
-            log.info("[Collect Failed, Run {}ms, All {}ms] Reason: {}", runningTime, allTime, builder.getMsg());
-        } else {
-            log.info("[Collect Success, Run {}ms, All {}ms].", runningTime, allTime);
-        }
-        return builder.build();
-    }

    private void validateResponse(CollectRep.MetricsData metricsData) {
        if (metricsData == null) {
@@ -0,0 +1,127 @@
+/*
+ * 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.handler;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.collect.AbstractCollect;
+import org.apache.hertzbeat.collector.collect.strategy.CollectStrategyFactory;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.constants.ContextStatus;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.dispatch.CollectTaskTimeoutMonitor;
+import org.apache.hertzbeat.collector.handler.impl.AbstractBatchDataStream;
+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.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+
+
+@Data
+@Slf4j
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode(callSuper = true)
+public class CollectMetricsDataDataStream extends AbstractBatchDataStream<Metrics, CollectRep.MetricsData.Builder> {
+    private CollectTaskTimeoutMonitor collectTaskTimeoutMonitor;
+
+    @Override
+    public CollectRep.MetricsData.Builder executeWithResponse(Context context, Metrics data) {
+        // preset start info
+        context.put(ContextKey.METRICS, data);
+        Job job = context.get(ContextKey.JOB);
+        long startTime = context.get(ContextKey.METRICS_COLLECT_START_TIME);
+        setNewThreadName(job.getMonitorId(), job.getApp(), startTime, data);
+
+
+        Timeout timeout = context.get(ContextKey.TIMEOUT);
+        String key = data.getPrometheus() != null ? String.valueOf(job.getId()) : job.getId() + "-" + data.getName();
+        context.put(ContextKey.METRICS_KEY, key);
+        this.collectTaskTimeoutMonitor.putMetrics(key, new CollectTaskTimeoutMonitor.MetricsTime(startTime, data, timeout));
+
+
+        CollectRep.MetricsData.Builder fetchedData = this.fetchData(job, data);
+        if (fetchedData.getCode() != CollectRep.Code.SUCCESS && CommonConstants.AVAILABLE_METRICS == data.getPriority()) {
+            context.setStatus(ContextStatus.TRUNCATE_HANDLER);
+        }
+        return fetchedData;
+    }
+
+    private void setNewThreadName(long monitorId, String app, long startTime, Metrics metrics) {
+        String builder = monitorId + "-" + app + "-" + metrics.getName() + "-" + String.valueOf(startTime).substring(9);
+        Thread.currentThread().setName(builder);
+    }
+
+    private CollectRep.MetricsData.Builder fetchData(Job job, Metrics metrics) {
+        CollectRep.MetricsData.Builder response = CollectRep.MetricsData.newBuilder();
+        response.setApp(job.getApp())
+                .setId(job.getMonitorId())
+                .setTenantId(job.getTenantId())
+                .setLabels(job.getLabels())
+                .setAnnotations(job.getAnnotations())
+                .addMetadataAll(job.getMetadata());
+
+        //todo transcribe Prometheus to different chain
+        // for prometheus auto or proxy mode
+//        if (DispatchConstants.PROTOCOL_PROMETHEUS.equalsIgnoreCase(metrics.getProtocol())) {
+//            List<CollectRep.MetricsData> metricsData = PrometheusAutoCollectImpl.getInstance().collect(response, metrics);
+//            validateResponse(metricsData == null ? null : metricsData.stream().findFirst().orElse(null));
+//            collectDataDispatch.dispatchCollectData(timeout, metrics, metricsData);
+//            return null;
+//        }
+
+        response.setMetrics(metrics.getName());
+        // According to the metrics collection protocol, application type, etc.,
+        // dispatch to the real application metrics collection implementation class
+        AbstractCollect abstractCollect = CollectStrategyFactory.invoke(metrics.getProtocol());
+        if (abstractCollect == null) {
+            log.error("[Dispatcher] - not support this: app: {}, metrics: {}, protocol: {}.", job.getApp(), metrics.getName(), metrics.getProtocol());
+            response.setCode(CollectRep.Code.FAIL);
+            response.setMsg("not support " + job.getApp() + ", " + metrics.getName() + ", " + metrics.getProtocol());
+
+            return response;
+        }
+
+        try {
+            abstractCollect.preCheck(metrics);
+            abstractCollect.collect(response, metrics);
+        } catch (Exception e) {
+            String msg = e.getMessage();
+            if (msg == null && e.getCause() != null) {
+                msg = e.getCause().getMessage();
+            }
+            if (e instanceof IllegalArgumentException) {
+                log.error("[Metrics PreCheck]: {}.", msg, e);
+            } else {
+                log.error("[Metrics Collect]: {}.", msg, e);
+            }
+            response.setCode(CollectRep.Code.FAIL);
+            if (msg != null) {
+                response.setMsg(msg);
+            }
+        }
+
+        return response;
+    }
+}
@@ -0,0 +1,34 @@
+/*
+ * 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.handler;
+
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.handler.impl.AbstractBatchDataStream;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+
+/**
+ *
+ */
+public class DynamicSubTaskCollectMetricsDataDataStream extends AbstractBatchDataStream<Metrics, CollectRep.MetricsData.Builder> {
+    @Override
+    public CollectRep.MetricsData.Builder executeWithResponse(Context context, Metrics data) {
+        //todo 动态拆分
+        return null;
+    }
+}
@@ -0,0 +1,289 @@
+/*
+ * 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.listener;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.jexl3.JexlExpression;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.dispatch.unit.UnitConvert;
+import org.apache.hertzbeat.collector.handler.ContextBoundListener;
+import org.apache.hertzbeat.collector.util.CollectUtil;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.util.CommonUtil;
+import org.apache.hertzbeat.common.util.JexlExpressionRunner;
+import org.apache.hertzbeat.common.util.Pair;
+import org.springframework.util.CollectionUtils;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ *
+ */
+@Slf4j
+@AllArgsConstructor
+public class CalculateFieldsListener implements ContextBoundListener<CollectRep.MetricsData.Builder> {
+    private List<UnitConvert> unitConvertList;
+
+    @Override
+    public void execute(Context context, CollectRep.MetricsData.Builder data) {
+        Metrics metrics = context.get(ContextKey.METRICS);
+
+        this.calculateFields(metrics, unitConvertList, data);
+    }
+
+    /**
+     * Calculate the real metrics value according to the calculates and aliasFields configuration
+     *
+     * @param metrics     Metrics configuration
+     * @param collectData Data collection
+     */
+    private void calculateFields(Metrics metrics, List<UnitConvert> unitConvertList, CollectRep.MetricsData.Builder collectData) {
+        collectData.setPriority(metrics.getPriority());
+        List<CollectRep.Field> fieldList = new LinkedList<>();
+        for (Metrics.Field field : metrics.getFields()) {
+            CollectRep.Field.Builder fieldBuilder = CollectRep.Field.newBuilder();
+            fieldBuilder.setName(field.getField()).setType(field.getType()).setLabel(field.isLabel());
+            if (field.getUnit() != null) {
+                fieldBuilder.setUnit(field.getUnit());
+            }
+            fieldList.add(fieldBuilder.build());
+        }
+        collectData.addAllFields(fieldList);
+        List<CollectRep.ValueRow> aliasRowList = collectData.getValuesList();
+        if (aliasRowList == null || aliasRowList.isEmpty()) {
+            return;
+        }
+        collectData.clearValues();
+        // Preprocess calculates first
+        if (metrics.getCalculates() == null) {
+            metrics.setCalculates(Collections.emptyList());
+        }
+        // eg: database_pages=Database pages unconventional mapping
+        Map<String, String> fieldAliasMap = new HashMap<>(8);
+        Map<String, JexlExpression> fieldExpressionMap = metrics.getCalculates()
+                .stream()
+                .map(cal -> transformCal(cal, fieldAliasMap))
+                .filter(Objects::nonNull)
+                .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (JexlExpression) arr[1], (oldValue, newValue) -> newValue));
+
+        if (metrics.getUnits() == null) {
+            metrics.setUnits(Collections.emptyList());
+        }
+        Map<String, Pair<String, String>> fieldUnitMap = metrics.getUnits()
+                .stream()
+                .map(this::transformUnit)
+                .filter(Objects::nonNull)
+                .collect(Collectors.toMap(arr -> (String) arr[0], arr -> (Pair<String, String>) arr[1], (oldValue, newValue) -> newValue));
+
+        List<Metrics.Field> fields = metrics.getFields();
+        List<String> aliasFields = Optional.ofNullable(metrics.getAliasFields()).orElseGet(Collections::emptyList);
+        Map<String, String> aliasFieldValueMap = new HashMap<>(8);
+        Map<String, Object> fieldValueMap = new HashMap<>(8);
+        Map<String, Object> stringTypefieldValueMap = new HashMap<>(8);
+        Map<String, String> aliasFieldUnitMap = new HashMap<>(8);
+        CollectRep.ValueRow.Builder realValueRowBuilder = CollectRep.ValueRow.newBuilder();
+        for (CollectRep.ValueRow aliasRow : aliasRowList) {
+            for (int aliasIndex = 0; aliasIndex < aliasFields.size(); aliasIndex++) {
+                String aliasFieldValue = aliasRow.getColumns(aliasIndex);
+                String aliasField = aliasFields.get(aliasIndex);
+                if (!CommonConstants.NULL_VALUE.equals(aliasFieldValue)) {
+                    aliasFieldValueMap.put(aliasField, aliasFieldValue);
+                    // whether the alias field is a number
+                    CollectUtil.DoubleAndUnit doubleAndUnit = CollectUtil
+                            .extractDoubleAndUnitFromStr(aliasFieldValue);
+                    if (doubleAndUnit != null && doubleAndUnit.getValue() != null) {
+                        fieldValueMap.put(aliasField, doubleAndUnit.getValue());
+                        if (doubleAndUnit.getUnit() != null) {
+                            aliasFieldUnitMap.put(aliasField, doubleAndUnit.getUnit());
+                        }
+                    } else {
+                        fieldValueMap.put(aliasField, aliasFieldValue);
+                    }
+                    stringTypefieldValueMap.put(aliasField, aliasFieldValue);
+                } else {
+                    fieldValueMap.put(aliasField, null);
+                    stringTypefieldValueMap.put(aliasField, null);
+                }
+            }
+
+            for (Metrics.Field field : fields) {
+                String realField = field.getField();
+                JexlExpression expression = fieldExpressionMap.get(realField);
+                String value = null;
+                String aliasFieldUnit = null;
+                if (expression != null) {
+                    try {
+                        Map<String, Object> context;
+                        if (CommonConstants.TYPE_STRING == field.getType()) {
+                            context = stringTypefieldValueMap;
+                        } else {
+                            for (Map.Entry<String, String> unitEntry : aliasFieldUnitMap.entrySet()) {
+                                if (expression.getSourceText().contains(unitEntry.getKey())) {
+                                    aliasFieldUnit = unitEntry.getValue();
+                                    break;
+                                }
+                            }
+                            context = fieldValueMap;
+                        }
+
+                        // Also executed when valueList is empty, covering pure string assignment expressions
+                        Object objValue = JexlExpressionRunner.evaluate(expression, context);
+
+                        if (objValue != null) {
+                            value = String.valueOf(objValue);
+                        }
+                    } catch (Exception e) {
+                        log.warn("[calculates execute warning, use original value.] {}", e.getMessage());
+                        value = Optional.ofNullable(fieldValueMap.get(expression.getSourceText()))
+                                .map(String::valueOf)
+                                .orElse(null);
+                    }
+                } else {
+                    // does not exist then map the alias value
+                    String aliasField = fieldAliasMap.get(realField);
+                    if (aliasField != null) {
+                        value = aliasFieldValueMap.get(aliasField);
+                    } else {
+                        value = aliasFieldValueMap.get(realField);
+                    }
+
+                    if (value != null) {
+                        final byte fieldType = field.getType();
+                        if (fieldType == CommonConstants.TYPE_NUMBER) {
+                            CollectUtil.DoubleAndUnit doubleAndUnit = CollectUtil
+                                    .extractDoubleAndUnitFromStr(value);
+                            final Double tempValue = doubleAndUnit == null ? null : doubleAndUnit.getValue();
+                            value = tempValue == null ? null : String.valueOf(tempValue);
+                            aliasFieldUnit = doubleAndUnit == null ? null : doubleAndUnit.getUnit();
+                        } else if (fieldType == CommonConstants.TYPE_TIME) {
+                            final int tempValue;
+                            value = (tempValue = CommonUtil.parseTimeStrToSecond(value)) == -1 ? null : String.valueOf(tempValue);
+                        }
+                    }
+                }
+
+                Pair<String, String> unitPair = fieldUnitMap.get(realField);
+                if (aliasFieldUnit != null) {
+                    if (unitPair != null) {
+                        unitPair.setLeft(aliasFieldUnit);
+                    } else if (field.getUnit() != null && !aliasFieldUnit.equalsIgnoreCase(field.getUnit())) {
+                        unitPair = Pair.of(aliasFieldUnit, field.getUnit());
+                    }
+                }
+                if (value != null && unitPair != null) {
+                    for (UnitConvert unitConvert : unitConvertList) {
+                        if (unitConvert.checkUnit(unitPair.getLeft()) && unitConvert.checkUnit(unitPair.getRight())) {
+                            value = unitConvert.convert(value, unitPair.getLeft(), unitPair.getRight());
+                        }
+                    }
+                }
+                // Handle metrics values that may have units such as 34%, 34Mb, and limit values to 4 decimal places
+                if (CommonConstants.TYPE_NUMBER == field.getType()) {
+                    value = CommonUtil.parseDoubleStr(value, field.getUnit());
+                }
+                if (value == null) {
+                    value = CommonConstants.NULL_VALUE;
+                }
+                realValueRowBuilder.addColumn(value);
+            }
+            aliasFieldValueMap.clear();
+            fieldValueMap.clear();
+            aliasFieldUnitMap.clear();
+            stringTypefieldValueMap.clear();
+            CollectRep.ValueRow realValueRow = realValueRowBuilder.build();
+            realValueRowBuilder.clear();
+            // apply filter calculation to the real value row
+            if (!CollectionUtils.isEmpty(metrics.getFilters())) {
+                Map<String, Object> contextMap = new HashMap<>(8);
+                for (int i = 0; i < fields.size(); i++) {
+                    Metrics.Field field = fields.get(i);
+                    String value = realValueRow.getColumns(i);
+                    contextMap.put(field.getField(), value);
+                }
+                boolean isMatch = false;
+                for (String filterExpr : metrics.getFilters()) {
+                    try {
+                        JexlExpression expression = JexlExpressionRunner.compile(filterExpr);
+                        if ((Boolean) JexlExpressionRunner.evaluate(expression, contextMap)) {
+                            isMatch = true;
+                            break;
+                        }
+                    } catch (Exception e) {
+                        log.warn("[metrics data row filters execute warning] {}.", e.getMessage());
+                    }
+                }
+                if (!isMatch) {
+                    // ignore this data row
+                    continue;
+                }
+            }
+            collectData.addValueRow(realValueRow);
+        }
+    }
+
+    /**
+     * @param cal           cal
+     * @param fieldAliasMap field alias map
+     * @return expr
+     */
+    private Object[] transformCal(String cal, Map<String, String> fieldAliasMap) {
+        int splitIndex = cal.indexOf("=");
+        if (splitIndex < 0) {
+            return null;
+        }
+        String field = cal.substring(0, splitIndex).trim();
+        String expressionStr = cal.substring(splitIndex + 1).trim().replace("\\#", "#");
+        JexlExpression expression;
+        try {
+            expression = JexlExpressionRunner.compile(expressionStr);
+        } catch (Exception e) {
+            fieldAliasMap.put(field, expressionStr);
+            return null;
+        }
+        return new Object[]{field, expression};
+    }
+
+    /**
+     * transform unit
+     *
+     * @param unit unit
+     * @return units
+     */
+    private Object[] transformUnit(String unit) {
+        int equalIndex = unit.indexOf("=");
+        int arrowIndex = unit.indexOf("->");
+        if (equalIndex < 0 || arrowIndex < 0) {
+            return null;
+        }
+        String field = unit.substring(0, equalIndex).trim();
+        String originUnit = unit.substring(equalIndex + 1, arrowIndex).trim();
+        String newUnit = unit.substring(arrowIndex + 2).trim();
+        return new Object[]{field, Pair.of(originUnit, newUnit)};
+    }
+}
@@ -0,0 +1,38 @@
+/*
+ * 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.listener;
+
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.constants.ContextStatus;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.handler.ContextBoundListener;
+import org.apache.hertzbeat.common.timer.Timeout;
+
+/**
+ *
+ */
+public class CommonMetricsDataListener<T> implements ContextBoundListener<T> {
+    @Override
+    public void execute(Context context, T data) {
+        Timeout timeout = context.get(ContextKey.TIMEOUT);
+
+        if (timeout == null || timeout.isCancelled()) {
+            context.setStatus(ContextStatus.STOP);
+        }
+    }
+}
@@ -0,0 +1,69 @@
+/*
+ * 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.listener;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.handler.ContextBoundListener;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.queue.CommonDataQueue;
+
+/**
+ * 周期任务专用
+ */
+@Slf4j
+@AllArgsConstructor
+public class MetricsDataDeliveryListener implements ContextBoundListener<CollectRep.MetricsData.Builder> {
+    private CommonDataQueue commonDataQueue;
+
+    @Override
+    public void execute(Context context, CollectRep.MetricsData.Builder data) {
+        Job job = context.get(ContextKey.JOB);
+        CollectRep.MetricsData metricsData = data.build();
+
+        cyclicJobDebugLog(job, metricsData);
+
+        sendToQueue(job, metricsData);
+    }
+
+    private void sendToQueue(Job job, CollectRep.MetricsData metricsData) {
+        if (job.isSd()) {
+            CollectRep.MetricsData sdMetricsData = CollectRep.MetricsData.newBuilder(metricsData).build();
+            commonDataQueue.sendServiceDiscoveryData(sdMetricsData);
+        }
+        commonDataQueue.sendMetricsData(metricsData);
+    }
+
+    private void cyclicJobDebugLog(Job job, CollectRep.MetricsData metricsData) {
+        if (log.isDebugEnabled()) {
+            log.debug("Cyclic Job: {} - {} - {}", job.getMonitorId(), job.getApp(), metricsData.getMetrics());
+            metricsDataDebugLog(metricsData);
+        }
+    }
+
+    private void metricsDataDebugLog(CollectRep.MetricsData metricsData) {
+        for (CollectRep.ValueRow valueRow : metricsData.getValues()) {
+            for (CollectRep.Field field : metricsData.getFields()) {
+                log.debug("Field-->{},Value-->{}", field.getName(), valueRow.getColumns(metricsData.getFields().indexOf(field)));
+            }
+        }
+    }
+}
@@ -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.hertzbeat.collector.listener;
+
+import lombok.AllArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.dispatch.CollectTaskTimeoutMonitor;
+import org.apache.hertzbeat.collector.handler.ContextBoundListener;
+
+/**
+ *
+ */
+@AllArgsConstructor
+public class RemoveTimeoutMonitorListener implements ContextBoundListener<Object> {
+    private CollectTaskTimeoutMonitor collectTaskTimeoutMonitor;
+
+    @Override
+    public void execute(Context context, Object data) {
+        String metricsKey = context.get(ContextKey.METRICS_KEY);
+        if (StringUtils.isBlank(metricsKey)) {
+            return;
+        }
+
+        collectTaskTimeoutMonitor.removeMetrics(metricsKey);
+    }
+}
@@ -0,0 +1,55 @@
+/*
+ * 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.listener;
+
+import lombok.AllArgsConstructor;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.handler.ContextBoundDataStream;
+import org.apache.hertzbeat.collector.timer.TimerDispatch;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.timer.Timeout;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 周期任务专用
+ */
+@AllArgsConstructor
+public class RerunDataStream implements ContextBoundDataStream<Object> {
+    private TimerDispatch timerDispatch;
+
+    @Override
+    public void execute(Context context, Object data) {
+        Job job = context.get(ContextKey.JOB);
+        Timeout timeout = context.get(ContextKey.TIMEOUT);
+
+        if (!timeout.isCancelled()) {
+            long spendTime = System.currentTimeMillis() - job.getDispatchTime();
+            long interval = job.getInterval() - spendTime / 1000L;
+            interval = interval <= 0 ? 0 : interval;
+            timerDispatch.cyclicJob((WheelTimerTask) timeout.task(), interval, TimeUnit.SECONDS);
+        }
+    }
+
+    @Override
+    public void whenException(Context context, Object data, Throwable throwable) {
+
+    }
+}
@@ -0,0 +1,61 @@
+/*
+ * 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.listener;
+
+import com.google.common.collect.Lists;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.handler.ContextBoundListener;
+import org.apache.hertzbeat.collector.timer.TimerDispatch;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+
+/**
+ * 一次性任务专用
+ */
+@Slf4j
+@AllArgsConstructor
+public class ResponseJobDataListener implements ContextBoundListener<CollectRep.MetricsData.Builder> {
+    private TimerDispatch timerDispatch;
+
+    @Override
+    public void execute(Context context, CollectRep.MetricsData.Builder data) {
+        Job job = context.get(ContextKey.JOB);
+        CollectRep.MetricsData metricsData = data.build();
+
+        oneTimeJobDebugLog(metricsData);
+        timerDispatch.responseSyncJobData(job.getId(), Lists.newArrayList(metricsData));
+    }
+
+    private void oneTimeJobDebugLog(CollectRep.MetricsData metricsData) {
+        if (log.isDebugEnabled()) {
+            log.debug("One-time Job: {}", metricsData.getMetrics());
+            metricsDataDebugLog(metricsData);
+        }
+    }
+
+    private void metricsDataDebugLog(CollectRep.MetricsData metricsData) {
+        for (CollectRep.ValueRow valueRow : metricsData.getValues()) {
+            for (CollectRep.Field field : metricsData.getFields()) {
+                log.debug("Field-->{},Value-->{}", field.getName(), valueRow.getColumns(metricsData.getFields().indexOf(field)));
+            }
+        }
+    }
+}
@@ -0,0 +1,53 @@
+/*
+ * 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.listener;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.handler.ContextBoundListener;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+
+/**
+ *
+ * @author Calvin
+ * @date 9/1/2025
+ */
+@Slf4j
+public class ValidateResponseListener implements ContextBoundListener<CollectRep.MetricsData.Builder> {
+
+    @Override
+    public void execute(Context context, CollectRep.MetricsData.Builder data) {
+        long startTime = context.get(ContextKey.METRICS_COLLECT_START_TIME);
+        Metrics metrics = context.get(ContextKey.METRICS);
+
+        this.validateResponse(startTime, metrics, data);
+    }
+
+    private void validateResponse(long startTime, Metrics metrics, CollectRep.MetricsData.Builder builder) {
+        long endTime = System.currentTimeMillis();
+        builder.setTime(endTime);
+        long allTime = endTime - startTime;
+        if (builder.getCode() != CollectRep.Code.SUCCESS) {
+            log.info("[Metrics: {}][Collect Failed, Run {}ms] Reason: {}", metrics.getName(), allTime, builder.getMsg());
+        } else {
+            log.info("[Metrics: {}][Collect Success, Run {}ms].", metrics.getName(), allTime);
+        }
+    }
+}
@@ -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.hertzbeat.collector.constants;
+
+import lombok.Getter;
+import org.apache.hertzbeat.common.entity.collector.CollectorMetaData;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.timer.Timeout;
+
+public enum ContextKey {
+    META_DATA(CollectorMetaData.class),
+    JOB(Job.class),
+    TIMEOUT(Timeout.class),
+    METRICS_COLLECT_START_TIME(Long.class),
+    METRICS(Metrics.class),
+    METRICS_KEY(String.class),
+    ;
+
+    @Getter
+    private final Class<?> clazz;
+
+    ContextKey(Class<?> clazz) {
+        this.clazz = clazz;
+    }
+}
@@ -0,0 +1,25 @@
+/*
+ * 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.constants;
+
+/**
+ *
+ */
+public enum ContextStatus {
+    WAITING, RUNNING, STOP, TRUNCATE_HANDLER
+}
@@ -0,0 +1,25 @@
+/*
+ * 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.constants;
+
+/**
+ *
+ */
+public enum HandlerType {
+    NORMAL, ON_COMPLETE
+}
@@ -0,0 +1,73 @@
+/*
+ * 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.context;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hertzbeat.collector.constants.ContextStatus;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ *
+ */
+public abstract class AbstractInmemoryContext implements Context {
+    protected final AtomicReference<ContextStatus> contextStatus = new AtomicReference<>(ContextStatus.WAITING);
+    protected final Map<Object, Object> map = new ConcurrentHashMap<>();
+    @Getter
+    @Setter
+    private Throwable error;
+
+
+    @Override
+    public <T> void put(Object key, T value) {
+        this.map.put(key, value);
+    }
+
+    @Override
+    public <T> T remove(Object key) {
+        return (T) this.map.remove(key);
+    }
+
+    @Override
+    public <T> T get(Object key) {
+        return (T) this.map.get(key);
+    }
+
+    @Override
+    public <T> T getOrDefault(Object key, T defaultValue) {
+        return (T) this.map.getOrDefault(key, defaultValue);
+    }
+
+    @Override
+    public boolean hasKey(Object key) {
+        return this.map.containsKey(key);
+    }
+
+    @Override
+    public ContextStatus getStatus() {
+        return contextStatus.get();
+    }
+
+    @Override
+    public void setStatus(ContextStatus status) {
+        contextStatus.set(status);
+    }
+}
@@ -0,0 +1,29 @@
+/*
+ * 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.context;
+
+import org.apache.hertzbeat.collector.constants.ContextStatus;
+
+/**
+ * 只维护与上下文的元数据
+ */
+public interface Context extends ContextView, ContextOperation {
+    ContextStatus getStatus();
+
+    void setStatus(ContextStatus status);
+}
@@ -0,0 +1,34 @@
+/*
+ * 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.context;
+
+import io.micrometer.common.lang.Nullable;
+
+/**
+ *
+ */
+public interface ContextOperation {
+    <T> void put(Object key, T value);
+
+    <T> T remove(Object key);
+
+    @Nullable
+    Throwable getError();
+
+    void setError(Throwable error);
+}
@@ -0,0 +1,29 @@
+/*
+ * 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.context;
+
+/**
+ * 对上下文内容的查询操作
+ */
+public interface ContextView {
+    <T> T get(Object key);
+
+    <T> T getOrDefault(Object key, T defaultValue);
+
+    boolean hasKey(Object key);
+}
@@ -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.context.impl;
+
+import org.apache.hertzbeat.collector.context.AbstractInmemoryContext;
+
+/**
+ *
+ */
+public class DefaultContext extends AbstractInmemoryContext {
+    private DefaultContext() {
+    }
+
+    public static DefaultContext newInstance() {
+        return new DefaultContext();
+    }
+}
@@ -36,13 +36,4 @@ public interface CollectDataDispatch {
     */
    void dispatchCollectData(Timeout timeout, Metrics metrics, CollectRep.MetricsData metricsData);

-    /**
-     * Processing and distributing collection result data
-     *
-     * @param timeout     time wheel timeout        
-     * @param metrics     The following metrics collection tasks    
-     * @param metricsDataList Collect result data       
-     */
-    void dispatchCollectData(Timeout timeout, Metrics metrics, List<CollectRep.MetricsData> metricsDataList);
-
 }
@@ -0,0 +1,113 @@
+/*
+ * 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.handler;
+
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.hertzbeat.collector.constants.HandlerType;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.dispatch.WorkerPool;
+import org.apache.hertzbeat.collector.handler.impl.AbstractListenerBoundDataStream;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ *
+ */
+@Slf4j
+public class ChainBootstrap {
+    @Setter
+    private Context context;
+    private TaskChain<?> taskChain;
+    private WorkerPool workerPool;
+    private final List<ContextBoundDataStream> contextBoundDataStreamList = new ArrayList<>();
+    private final List<ContextBoundDataStream> onCompleteContextBoundDataStreamList = new ArrayList<>();
+    private final List<ContextBoundListener> dataListenerList = new ArrayList<>();
+    private final List<ContextBoundListener> onCompleteListenerList = new ArrayList<>();
+
+    public static ChainBootstrap withContext(Context context) {
+        ChainBootstrap bootstrap = new ChainBootstrap();
+        bootstrap.setContext(context);
+        return bootstrap;
+    }
+
+    public ChainBootstrap withChain(TaskChain<?> taskChain) {
+        this.taskChain = taskChain;
+        return this;
+    }
+
+    public ChainBootstrap withWorkerPool(WorkerPool workerPool) {
+        this.workerPool = workerPool;
+        return this;
+    }
+
+    public <T> ChainBootstrap addContext(Object key, T value) {
+        context.put(key, value);
+        return this;
+    }
+
+    public ChainBootstrap addDataStream(ContextBoundDataStream contextBoundDataStream) {
+        contextBoundDataStreamList.add(contextBoundDataStream);
+        return this;
+    }
+
+    public ChainBootstrap onComplete(ContextBoundDataStream contextBoundDataStream) {
+        onCompleteContextBoundDataStreamList.add(contextBoundDataStream);
+        return this;
+    }
+
+    public ChainBootstrap addListener(ContextBoundListener dataListener) {
+        dataListenerList.add(dataListener);
+        return this;
+    }
+
+    public ChainBootstrap onEachDataStreamComplete(ContextBoundListener dataListener) {
+        onCompleteListenerList.add(dataListener);
+        return this;
+    }
+
+    public void start() {
+        if (taskChain == null || context == null) {
+            log.error("Failed to start chain boostrap due to null value of Context or TaskChain");
+            return;
+        }
+
+        for (ContextBoundDataStream contextBoundDataStream : contextBoundDataStreamList) {
+            if (contextBoundDataStream instanceof AbstractListenerBoundDataStream listenerBoundHandler) {
+                if (CollectionUtils.isNotEmpty(dataListenerList)) {
+                    listenerBoundHandler.getDataListenerList().addAll(dataListenerList);
+                }
+                if (CollectionUtils.isNotEmpty(onCompleteListenerList)) {
+                    listenerBoundHandler.getOnCompleteListenerList().addAll(onCompleteListenerList);
+                }
+            }
+
+            taskChain.addLast(HandlerType.NORMAL, contextBoundDataStream);
+        }
+
+        onCompleteContextBoundDataStreamList.forEach(handler -> taskChain.addLast(HandlerType.ON_COMPLETE, handler));
+
+        if (workerPool != null) {
+            workerPool.executeJob(() -> taskChain.execute(context));
+        } else {
+            taskChain.execute(context);
+        }
+    }
+}
@@ -0,0 +1,29 @@
+/*
+ * 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.handler;
+
+import org.apache.hertzbeat.collector.context.Context;
+
+/**
+ *
+ */
+public interface ContextBoundDataStream<T> {
+    void execute(Context context, T data);
+
+    void whenException(Context context, T data, Throwable throwable);
+}
@@ -0,0 +1,26 @@
+/*
+ * 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.handler;
+
+import org.apache.hertzbeat.collector.context.Context;
+
+/**
+ */
+public interface ContextBoundListener<T> {
+    void execute(Context context, T data);
+}
@@ -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.handler;
+
+import org.apache.hertzbeat.collector.constants.HandlerType;
+import org.apache.hertzbeat.collector.context.Context;
+
+/**
+ *
+ */
+public interface TaskChain<T> {
+    void execute(Context context);
+
+    void execute(Context context, T data);
+
+    void addLast(HandlerType handlerType, ContextBoundDataStream<T> handler);
+}
@@ -0,0 +1,38 @@
+/*
+ * 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.handler.impl;
+
+import lombok.Setter;
+import org.apache.hertzbeat.collector.context.Context;
+
+import java.util.List;
+
+/**
+ *
+ */
+public abstract class AbstractBatchDataStream<T, R> extends AbstractListenerBoundDataStream<T, R> {
+    @Setter
+    protected List<T> sourceDataList;
+
+    @Override
+    public void execute(Context context, T data) {
+        for (T t : sourceDataList) {
+            super.execute(context, t);
+        }
+    }
+}
@@ -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.hertzbeat.collector.handler.impl;
+
+import org.apache.hertzbeat.collector.constants.HandlerType;
+import org.apache.hertzbeat.collector.handler.ContextBoundDataStream;
+import org.apache.hertzbeat.collector.handler.TaskChain;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ *
+ */
+public abstract class AbstractContextBoundTaskChain<T> implements TaskChain<T> {
+    protected final Map<HandlerType, List<ContextBoundDataStream<T>>> contextBoundHandlerMap = new HashMap<>();
+
+    @Override
+    public void addLast(HandlerType handlerType, ContextBoundDataStream<T> handler) {
+        if (!contextBoundHandlerMap.containsKey(handlerType)) {
+            contextBoundHandlerMap.put(handlerType, new ArrayList<>());
+        }
+
+        contextBoundHandlerMap.get(handlerType).add(handler);
+    }
+}
@@ -0,0 +1,90 @@
+/*
+ * 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.handler.impl;
+
+import lombok.Getter;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.hertzbeat.collector.constants.ContextKey;
+import org.apache.hertzbeat.collector.constants.ContextStatus;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.handler.ContextBoundDataStream;
+import org.apache.hertzbeat.collector.handler.ContextBoundListener;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ *
+ */
+public abstract class AbstractListenerBoundDataStream<T, R> implements ContextBoundDataStream<T> {
+    @Getter
+    private final List<? extends ContextBoundListener<R>> dataListenerList = new ArrayList<>();
+    @Getter
+    private final List<? extends ContextBoundListener<R>> onCompleteListenerList = new ArrayList<>();
+
+    @Override
+    public void execute(Context context, T data) {
+        long startTime = System.currentTimeMillis();
+        context.put(ContextKey.METRICS_COLLECT_START_TIME, startTime);
+
+        R executeResult = executeWithResponse(context, data);
+
+        runListener(context, executeResult);
+
+        runOnCompleteListener(context, executeResult);
+    }
+
+    public R executeWithResponse(Context context, T data) {
+        // no-op
+        return null;
+    }
+
+    @Override
+    public void whenException(Context context, T data, Throwable throwable) {
+        // no-op
+    }
+
+    private void runListener(Context context, R executeResult) {
+        if (CollectionUtils.isEmpty(dataListenerList)) {
+            return;
+        }
+
+        if (ContextStatus.STOP.equals(context.getStatus())) {
+            return;
+        }
+
+        //todo 异常处理
+        for (ContextBoundListener<R> listener : dataListenerList) {
+            listener.execute(context, executeResult);
+
+            if (ContextStatus.STOP.equals(context.getStatus())) {
+                break;
+            }
+        }
+    }
+
+    private void runOnCompleteListener(Context context, R executeResult) {
+        if (CollectionUtils.isEmpty(onCompleteListenerList)) {
+            return;
+        }
+
+        for (ContextBoundListener<R> listener : onCompleteListenerList) {
+            listener.execute(context, executeResult);
+        }
+    }
+}
@@ -0,0 +1,63 @@
+/*
+ * 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.handler.impl;
+
+import org.apache.hertzbeat.collector.constants.ContextStatus;
+import org.apache.hertzbeat.collector.constants.HandlerType;
+import org.apache.hertzbeat.collector.context.Context;
+import org.apache.hertzbeat.collector.handler.ContextBoundDataStream;
+
+import java.util.ArrayList;
+
+
+/**
+ *
+ */
+public class BatchExecuteTaskChain<T> extends AbstractContextBoundTaskChain<T> {
+    @Override
+    public void execute(Context context) {
+        this.execute(context, null);
+    }
+
+    @Override
+    public void execute(Context context, T data) {
+        context.setStatus(ContextStatus.RUNNING);
+
+        for (ContextBoundDataStream<T> contextBoundDataStream : contextBoundHandlerMap.getOrDefault(HandlerType.NORMAL, new ArrayList<>())) {
+            runHandler(context, data, contextBoundDataStream);
+
+            if (ContextStatus.TRUNCATE_HANDLER.equals(context.getStatus()) || ContextStatus.STOP.equals(context.getStatus())) {
+                break;
+            }
+
+            // in order to init error info for the next loop
+            context.setError(null);
+        }
+
+        contextBoundHandlerMap.getOrDefault(HandlerType.ON_COMPLETE, new ArrayList<>()).forEach(handler -> runHandler(context, data, handler));
+    }
+
+    private static <T> void runHandler(Context context, T data, ContextBoundDataStream<T> contextBoundDataStream) {
+        try {
+            contextBoundDataStream.execute(context, data);
+        } catch (Exception exception) {
+            context.setError(exception);
+            contextBoundDataStream.whenException(context, data, exception);
+        }
+    }
+}
@@ -382,4 +382,6 @@ public interface CommonConstants {
     * JEXL custom function `json`
     */
    String JEXL_CUSTOM_JSON_FUNCTION = "json";
+
+    byte AVAILABLE_METRICS = 0;
 }
@@ -0,0 +1,38 @@
+/*
+ * 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.common.entity.collector;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Date;
+
+/**
+ *
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class CollectorMetaData {
+    private String identity;
+    private String mode;
+    private Date startTime;
+}
Author	SHA1	Message	Date
Calvin	99ccf6cd48	[improve] delete MetricsSource.java	2025-09-06 16:16:59 +08:00
Calvin	986e833798	[improve] add license header	2025-09-06 16:16:07 +08:00
Calvin	13b87b6c51	[improve] add license header	2025-09-06 16:14:59 +08:00
Calvin	292f03045b	[improve] add license header	2025-09-06 16:13:11 +08:00
Calvin	2746615f63	[improve] add license header	2025-09-06 16:12:20 +08:00
Calvin	195e7e95fa	[refactor] rename handler to data stream and delete UnitConverter.java	2025-09-06 15:45:33 +08:00
Calvin	d0b3c2dba2	[refactor] optimize collector module	2025-09-06 14:55:23 +08:00