Merge branch 'master' into online-prometheus-parser

Signed-off-by: tomsun28 <tomsun28@outlook.com>

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/HttpCollectImpl.java

@@ -18,11 +18,14 @@
 package org.apache.hertzbeat.collector.collect.http;
 
 import static org.apache.hertzbeat.common.constants.SignConstants.RIGHT_DASH;
+
 import com.google.gson.JsonArray;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
+
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.InterruptedIOException;
 import java.io.StringReader;
 import java.net.ConnectException;
@@ -34,9 +37,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 import javax.net.ssl.SSLException;
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -44,13 +45,14 @@ import javax.xml.xpath.XPath;
 import javax.xml.xpath.XPathConstants;
 import javax.xml.xpath.XPathExpressionException;
 import javax.xml.xpath.XPathFactory;
+
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.collector.collect.AbstractCollect;
 import org.apache.hertzbeat.collector.collect.common.http.CommonHttpClient;
 import org.apache.hertzbeat.collector.collect.http.promethus.AbstractPrometheusParse;
 import org.apache.hertzbeat.collector.collect.http.promethus.PrometheusParseCreator;
-import org.apache.hertzbeat.collector.collect.http.promethus.exporter.ExporterParser;
-import org.apache.hertzbeat.collector.collect.http.promethus.exporter.MetricFamily;
+import org.apache.hertzbeat.collector.collect.prometheus.parser.MetricFamily;
+import org.apache.hertzbeat.collector.collect.prometheus.parser.OnlineParser;
 import org.apache.hertzbeat.collector.constants.CollectorConstants;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
 import org.apache.hertzbeat.collector.util.CollectUtil;
@@ -103,10 +105,13 @@ import java.util.Collections;
  */
 @Slf4j
 public class HttpCollectImpl extends AbstractCollect {
-    private static final Map<Long, ExporterParser> EXPORTER_PARSER_TABLE = new ConcurrentHashMap<>();
-    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());
+    private final Set<Integer> defaultSuccessStatusCodes = Set.of(
+            HttpStatus.SC_OK,
+            HttpStatus.SC_CREATED,
+            HttpStatus.SC_ACCEPTED,
+            HttpStatus.SC_MULTIPLE_CHOICES,
+            HttpStatus.SC_MOVED_PERMANENTLY,
+            HttpStatus.SC_MOVED_TEMPORARILY);
 
     @Override
     public void preCheck(Metrics metrics) throws IllegalArgumentException {
@@ -127,7 +132,7 @@ public class HttpCollectImpl extends AbstractCollect {
         if (CollectionUtils.isEmpty(httpProtocol.getSuccessCodes())) {
             httpProtocol.setSuccessCodes(List.of(HttpStatus.SC_OK + ""));
         }
-        
+
         HttpContext httpContext = createHttpContext(metrics.getHttp());
         HttpUriRequest request = createHttpRequest(metrics.getHttp());
         try (CloseableHttpResponse response = CommonHttpClient.getHttpClient().execute(request, httpContext)) {
@@ -139,10 +144,11 @@ public class HttpCollectImpl extends AbstractCollect {
                 builder.setMsg(NetworkConstants.STATUS_CODE + SignConstants.BLANK + statusCode);
                 return;
             }
-            // todo This code converts an InputStream directly to a String. For large data in Prometheus exporters,
-            // this could create large objects, potentially impacting JVM memory space significantly.
-            // Option 1: Parse using InputStream, but this requires significant code changes;
-            // Option 2: Manually trigger garbage collection, similar to how it's done in Dubbo for large inputs.
+            /*
+             this could create large objects, potentially impacting JVM memory space significantly.
+             Option 1: Parse using InputStream, but this requires significant code changes;
+             Option 2: Manually trigger garbage collection, similar to how it's done in Dubbo for large inputs.
+             */
             String resp = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
             if (!StringUtils.hasText(resp)) {
                 log.info("http response entity is empty, status: {}.", statusCode);
@@ -156,7 +162,7 @@ public class HttpCollectImpl extends AbstractCollect {
                     case DispatchConstants.PARSE_PROM_QL ->
                             parseResponseByPromQl(resp, metrics.getAliasFields(), metrics.getHttp(), builder);
                     case DispatchConstants.PARSE_PROMETHEUS ->
-                            parseResponseByPrometheusExporter(resp, metrics.getAliasFields(), builder);
+                            parseResponseByPrometheusExporter(response.getEntity().getContent(), metrics.getAliasFields(), builder);
                     case DispatchConstants.PARSE_XML_PATH ->
                             parseResponseByXmlPath(resp, metrics, builder, responseTime);
                     case DispatchConstants.PARSE_WEBSITE ->
@@ -594,36 +600,22 @@ public class HttpCollectImpl extends AbstractCollect {
         prometheusParser.handle(resp, aliasFields, http, builder);
     }
 
-    private void parseResponseByPrometheusExporter(String resp, List<String> aliasFields,
-                                                   CollectRep.MetricsData.Builder builder) {
-        if (!EXPORTER_PARSER_TABLE.containsKey(builder.getId())) {
-            EXPORTER_PARSER_TABLE.put(builder.getId(), new ExporterParser());
+    private void parseResponseByPrometheusExporter(InputStream content, List<String> aliasFields, CollectRep.MetricsData.Builder builder) throws IOException {
+        Map<String, MetricFamily> metricFamilyMap = OnlineParser.parseMetrics(content);
+        if (metricFamilyMap == null || metricFamilyMap.isEmpty()) {
+            return;
         }
-        ExporterParser parser = EXPORTER_PARSER_TABLE.get(builder.getId());
-        Map<String, MetricFamily> metricFamilyMap = parser.textToMetric(resp);
         String metrics = builder.getMetrics();
         if (metricFamilyMap.containsKey(metrics)) {
             MetricFamily metricFamily = metricFamilyMap.get(metrics);
             for (MetricFamily.Metric metric : metricFamily.getMetricList()) {
-                Map<String, String> labelMap = metric.getLabelPair()
+                Map<String, String> labelMap = metric.getLabels()
                         .stream()
                         .collect(Collectors.toMap(MetricFamily.Label::getName, MetricFamily.Label::getValue));
                 CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder();
                 for (String aliasField : aliasFields) {
                     if ("value".equals(aliasField)) {
-                        if (metric.getCounter() != null) {
-                            valueRowBuilder.addColumn(String.valueOf(metric.getCounter().getValue()));
-                        } else if (metric.getGauge() != null) {
-                            valueRowBuilder.addColumn(String.valueOf(metric.getGauge().getValue()));
-                        } else if (metric.getUntyped() != null) {
-                            valueRowBuilder.addColumn(String.valueOf(metric.getUntyped().getValue()));
-                        } else if (metric.getInfo() != null) {
-                            valueRowBuilder.addColumn(String.valueOf(metric.getInfo().getValue()));
-                        } else if (metric.getSummary() != null) {
-                            valueRowBuilder.addColumn(String.valueOf(metric.getSummary().getValue()));
-                        } else if (metric.getHistogram() != null) {
-                            valueRowBuilder.addColumn(String.valueOf(metric.getHistogram().getValue()));
-                        }
+                        valueRowBuilder.addColumn(String.valueOf(metric.getValue()));
                     } else {
                         String columnValue = labelMap.get(aliasField);
                         valueRowBuilder.addColumn(columnValue == null ? CommonConstants.NULL_VALUE : columnValue);

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/promethus/exporter/ExporterParser.java

@@ -1,432 +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.http.promethus.exporter;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.hertzbeat.collector.collect.http.promethus.ParseException;
-import org.apache.hertzbeat.common.util.StrBuffer;
-
-/**
- * Resolves the data passed by prometheus's exporter interface http:xxx/metrics
- * Reference: prometheus text_parse.go code, entry: TextToMetricFamilies
- */
-@Slf4j
-public class ExporterParser {
-    private static final String HELP = "HELP";
-    private static final String TYPE = "TYPE";
-    private static final String EOF = "EOF";
-    private static final String METRIC_NAME_LABEL = ".name";
-    private static final String QUANTILE_LABEL = "quantile";
-    private static final String BUCKET_LABEL = "le";
-    private static final String NAME_LABEL = "__name__";
-    private static final String SUM_SUFFIX = "_sum";
-    private static final String COUNT_SUFFIX = "_count";
-
-    private static final char LEFT_CURLY_BRACKET = '{';
-    private static final char RIGHT_CURLY_BRACKET = '}';
-    private static final char EQUALS = '=';
-    private static final char QUOTES = '"';
-    private static final char ENTER = '\n';
-    private static final char SPACE = ' ';
-    private static final char COMMA = ',';
-    private final Lock lock = new ReentrantLock();
-    private MetricFamily currentMetricFamily;
-    private String currentQuantile;
-    private String currentBucket;
-
-    public Map<String, MetricFamily> textToMetric(String resp) {
-        // key: metric name, value: metric family
-        Map<String, MetricFamily> metricMap = new ConcurrentHashMap<>(10);
-        lock.lock();
-        try {
-            String[] lines = resp.split("\n");
-            for (String line : lines) {
-                this.parseLine(metricMap, new StrBuffer(line));
-            }
-            return metricMap;
-        } catch (Exception e) {
-            log.error("parse prometheus exporter data error, msg: {}", e.getMessage());
-        } finally {
-            lock.unlock();
-        }
-        return metricMap;
-    }
-
-    private void parseLine(Map<String, MetricFamily> metricMap, StrBuffer buffer) {
-        buffer.skipBlankTabs();
-        if (buffer.isEmpty()) {
-            return;
-        }
-        switch (buffer.charAt(0)) {
-            case '#' -> {
-                buffer.read();
-                this.currentMetricFamily = null;
-                this.parseComment(metricMap, buffer);
-            }
-            case ENTER -> {
-            }
-            default -> {
-                this.currentBucket = null;
-                this.currentQuantile = null;
-                this.parseMetric(buffer);
-            }
-        }
-    }
-
-    private void parseComment(Map<String, MetricFamily> metricMap, StrBuffer buffer) {
-        buffer.skipBlankTabs();
-        if (buffer.isEmpty()) {
-            return;
-        }
-        String token = this.readTokenUnitWhitespace(buffer);
-        if (EOF.equals(token)) {
-            return;
-        }
-        if (!HELP.equals(token) && !TYPE.equals(token)) {
-            log.error("parse comment error {}, start without {} or {}", buffer.toStr(), HELP, TYPE);
-            return;
-        }
-        String metricName = this.readTokenAsMetricName(buffer);
-        this.currentMetricFamily = metricMap.computeIfAbsent(metricName, key -> new MetricFamily());
-        this.currentMetricFamily.setName(metricName);
-        switch (token) {
-            case HELP -> this.parseHelp(buffer);
-            case TYPE -> this.parseType(buffer);
-            default -> {
-            }
-        }
-    }
-
-    private void parseHelp(StrBuffer line) {
-        line.skipBlankTabs();
-        this.currentMetricFamily.setHelp(line.toStr());
-    }
-
-    private void parseType(StrBuffer line) {
-        line.skipBlankTabs();
-        String type = line.toStr().toLowerCase();
-        MetricType metricType = MetricType.getType(type);
-        if (metricType == null) {
-            throw new ParseException("pare type error");
-        }
-        this.currentMetricFamily.setMetricType(metricType);
-    }
-
-    private void parseMetric(StrBuffer buffer) {
-        String metricName = this.readTokenAsMetricName(buffer);
-        MetricFamily.Label label = new MetricFamily.Label();
-        label.setName(METRIC_NAME_LABEL);
-        label.setValue(metricName);
-
-        if (metricName.isEmpty()) {
-            log.error("error parse metric, metric name is null, line: {}", buffer.toStr());
-            return;
-        }
-
-        List<MetricFamily.Metric> metricList = this.currentMetricFamily.getMetricList();
-        if (metricList == null) {
-            metricList = new ArrayList<>();
-            this.currentMetricFamily.setMetricList(metricList);
-        }
-
-        // TODO For the time being, the data is displayed in the form of labels. If there is a better chart display method in the future, we will optimize it.
-        MetricFamily.Metric metric = new MetricFamily.Metric();
-        metricList.add(metric);
-
-        metric.setLabelPair(new ArrayList<>());
-        metric.getLabelPair().add(label);
-        this.readLabels(metric, buffer);
-    }
-
-    private void readLabels(MetricFamily.Metric metric, StrBuffer buffer) {
-        buffer.skipBlankTabs();
-        if (buffer.isEmpty()) {
-            return;
-        }
-        if (buffer.charAt(0) == LEFT_CURLY_BRACKET) {
-            buffer.read();
-            this.startReadLabelName(metric, buffer);
-        } else {
-            this.readLabelValue(metric, null, buffer);
-        }
-    }
-
-    private void startReadLabelName(MetricFamily.Metric metric, StrBuffer buffer) {
-        buffer.skipBlankTabs();
-        if (buffer.isEmpty()) {
-            return;
-        }
-        if (buffer.charAt(0) == RIGHT_CURLY_BRACKET) {
-            buffer.read();
-            buffer.skipBlankTabs();
-            if (buffer.isEmpty()) {
-                return;
-            }
-            this.readLabelValue(metric, new MetricFamily.Label(), buffer);
-            return;
-        }
-        String labelName = this.readTokenAsLabelName(buffer);
-        if (labelName.isEmpty() || labelName.equals(NAME_LABEL)) {
-            throw new ParseException("invalid label name" + labelName + ", label name size = 0 or label name equals " + NAME_LABEL);
-        }
-        MetricFamily.Label label = new MetricFamily.Label();
-        label.setName(labelName);
-        if (buffer.read() != EQUALS) {
-            throw new ParseException("parse error, not match the format of labelName=labelValue");
-        }
-        this.startReadLabelValue(metric, label, buffer);
-    }
-
-    private void startReadLabelValue(MetricFamily.Metric metric, MetricFamily.Label label, StrBuffer buffer) {
-        buffer.skipBlankTabs();
-        if (buffer.isEmpty()) {
-            return;
-        }
-        char c = buffer.read();
-        if (c != QUOTES) {
-            throw new ParseException("expected '\"' at start of label value, line: " + buffer.toStr());
-        }
-        String labelValue = this.readTokenAsLabelValue(buffer);
-        label.setValue(labelValue);
-        if (!this.isValidLabelValue(labelValue)) {
-            throw new ParseException("no valid label value: " + labelValue);
-        }
-        if (this.currentMetricFamily.getMetricType().equals(MetricType.SUMMARY) && label.getName().equals(QUANTILE_LABEL)) {
-            this.currentQuantile = labelValue;
-        } else if (this.currentMetricFamily.getMetricType().equals(MetricType.HISTOGRAM) && label.getName().equals(BUCKET_LABEL)) {
-            this.currentBucket = labelValue;
-        }
-        metric.getLabelPair().add(label);
-
-        if (buffer.isEmpty()) {
-            return;
-        }
-        c = buffer.read();
-        switch (c) {
-            case COMMA -> this.startReadLabelName(metric, buffer);
-            case RIGHT_CURLY_BRACKET -> this.readLabelValue(metric, label, buffer);
-            default -> throw new ParseException("expected '}' or ',' at end of label value, line: " + buffer.toStr());
-        }
-    }
-
-    private void readLabelValue(MetricFamily.Metric metric, MetricFamily.Label label, StrBuffer buffer) {
-        buffer.skipBlankTabs();
-        if (buffer.isEmpty()) {
-            return;
-        }
-        switch (this.currentMetricFamily.getMetricType()) {
-            case INFO -> {
-                MetricFamily.Info info = new MetricFamily.Info();
-                info.setValue(buffer.toDouble());
-                metric.setInfo(info);
-            }
-            case COUNTER -> {
-                MetricFamily.Counter counter = new MetricFamily.Counter();
-                counter.setValue(buffer.toDouble());
-                metric.setCounter(counter);
-            }
-            case GAUGE -> {
-                MetricFamily.Gauge gauge = new MetricFamily.Gauge();
-                gauge.setValue(buffer.toDouble());
-                metric.setGauge(gauge);
-            }
-            case UNTYPED -> {
-                MetricFamily.Untyped untyped = new MetricFamily.Untyped();
-                untyped.setValue(buffer.toDouble());
-                metric.setUntyped(untyped);
-            }
-            case SUMMARY -> {
-                // For the time being, the data is displayed in the form of labels. If there is a better chart display method in the future, we will optimize it.
-                MetricFamily.Summary summary = new MetricFamily.Summary();
-                summary.setValue(buffer.toDouble());
-                metric.setSummary(summary);
-            }
-            case HISTOGRAM -> {
-                // For the time being, the data is displayed in the form of labels. If there is a better chart display method in the future, we will optimize it.
-                MetricFamily.Histogram histogram = new MetricFamily.Histogram();
-                histogram.setValue(buffer.toDouble());
-                metric.setHistogram(histogram);
-            }
-            default -> throw new ParseException("no such type in metricFamily");
-        }
-    }
-
-    /**
-     * Reads the token before the first whitespace
-     *
-     * @param buffer A line data object
-     * @return token unit
-     */
-    private String readTokenUnitWhitespace(StrBuffer buffer) {
-        StringBuilder builder = new StringBuilder();
-        while (!buffer.isEmpty()) {
-            char c = buffer.read();
-            if (c == SPACE) {
-                break;
-            }
-            builder.append(c);
-        }
-        return builder.toString();
-    }
-
-    /**
-     * Gets the name of the metric
-     *
-     * @param buffer A line data object
-     * @return token name
-     */
-    private String readTokenAsMetricName(StrBuffer buffer) {
-        buffer.skipBlankTabs();
-        StringBuilder builder = new StringBuilder();
-        if (this.isValidMetricNameStart(buffer.charAt(0))) {
-            while (!buffer.isEmpty()) {
-                char c = buffer.read();
-                if (!this.isValidMetricNameContinuation(c)) {
-                    buffer.rollback();
-                    break;
-                }
-                builder.append(c);
-            }
-            return builder.toString();
-        }
-        throw new ParseException("parse metric name error");
-    }
-
-    /**
-     * Gets the name of the label
-     *
-     * @param buffer A line data object
-     * @return label name
-     */
-    private String readTokenAsLabelName(StrBuffer buffer) {
-        buffer.skipBlankTabs();
-        StringBuilder builder = new StringBuilder();
-        char c = buffer.read();
-        if (this.isValidLabelNameStart(c)) {
-            builder.append(c);
-            while (!buffer.isEmpty()) {
-                c = buffer.read();
-                if (!this.isValidLabelNameContinuation(c)) {
-                    buffer.rollback();
-                    break;
-                }
-                builder.append(c);
-            }
-            return builder.toString();
-        }
-        throw new ParseException("parse label name error");
-    }
-
-    /**
-     * Gets the value of the label
-     *
-     * @param buffer A line data object
-     * @return label value
-     */
-    private String readTokenAsLabelValue(StrBuffer buffer) {
-        StringBuilder builder = new StringBuilder();
-        boolean escaped = false;
-        while (!buffer.isEmpty()) {
-            char c = buffer.read();
-            // Handle '\\' escape sequences
-            if (escaped) {
-                switch (c) {
-                    case QUOTES, '\\' -> builder.append(c);
-                    case 'n' -> builder.append('\n');
-                    default -> throw new ParseException("parse label value error");
-                }
-                escaped = false;
-            } else {
-                switch (c) {
-                    case QUOTES -> {
-                        return builder.toString();
-                    }
-                    case ENTER -> throw new ParseException("parse label value error, next line");
-                    case '\\' -> escaped = true;
-                    default -> builder.append(c);
-                }
-            }
-        }
-        return builder.toString();
-    }
-
-    /**
-     * Checks whether a character conforms to the first character rule for metric names
-     *
-     * @param c metric character
-     * @return true/false
-     */
-    private boolean isValidMetricNameStart(char c) {
-        return isValidLabelNameStart(c) || c == ':';
-    }
-
-    /**
-     * Checks whether a character conforms to rules for metric name characters other than the first
-     *
-     * @param c metric character
-     * @return true/false
-     */
-    private boolean isValidMetricNameContinuation(char c) {
-        return isValidLabelNameContinuation(c) || c == ':';
-    }
-
-    /**
-     * Checks whether a character conforms to the first character rule for label names
-     *
-     * @param c metric character
-     * @return true/false
-     */
-    private boolean isValidLabelNameStart(char c) {
-        return (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || c == '_';
-    }
-
-    /**
-     * Checks whether a character conforms to rules for label name characters other than the first
-     *
-     * @param c metric character
-     * @return true/false
-     */
-    private boolean isValidLabelNameContinuation(char c) {
-        return isValidLabelNameStart(c) || (c >= '0' && c <= '9');
-    }
-
-    /**
-     * Checks if a string is a valid UTF-8 encoded string
-     *
-     * @param s label value
-     * @return true/false
-     */
-    private boolean isValidLabelValue(String s) {
-        return s != null;
-    }
-
-    private boolean isSum(String s) {
-        return s != null && s.endsWith(SUM_SUFFIX);
-    }
-
-    private boolean isCount(String s) {
-        return s != null && s.endsWith(COUNT_SUFFIX);
-    }
-
-}

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/promethus/exporter/MetricFamily.java

@@ -1,252 +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.http.promethus.exporter;
-
-import java.util.ArrayList;
-import java.util.List;
-import lombok.Data;
-import lombok.ToString;
-
-/**
- * MetricFamily.
- */
-@Data
-@ToString
-public class MetricFamily {
-    /**
-     * metric name
-     */
-    private String name;
-
-    /**
-     * metric help
-     */
-    private String help;
-
-    /**
-     * metric type
-     */
-    private MetricType metricType;
-
-    /**
-     * Specific metric
-     */
-    private List<Metric> metricList;
-
-    /**
-     * Metric
-     */
-    @Data
-    public static class Metric {
-
-        /**
-         * Label data, mainly corresponding to the content within {}
-         */
-        private List<Label> labelPair;
-
-        /**
-         * info
-         */
-        private Info info;
-
-        /**
-         * gauge
-         */
-        private Gauge gauge;
-
-        /**
-         * counter
-         */
-        private Counter counter;
-
-        /**
-         * summary
-         */
-        private Summary summary;
-
-        /**
-         * untyped
-         */
-        private Untyped untyped;
-
-        /**
-         * histogram
-         */
-        private Histogram histogram;
-
-        /**
-         * timestampMs
-         */
-        private Long timestampMs;
-    }
-
-    /**
-     * Label
-     */
-    @Data
-    public static class Label {
-
-        /**
-         * name
-         */
-        private String name;
-
-        /**
-         * value
-         */
-        private String value;
-    }
-
-    /**
-     * Info
-     */
-    @Data
-    public static class Info {
-
-        /**
-         * value
-         */
-        private double value;
-
-    }
-
-    /**
-     * Counter
-     */
-    @Data
-    public static class Counter {
-
-        /**
-         * value
-         */
-        private double value;
-
-        // Exemplar
-    }
-
-    /**
-     * Gauge
-     */
-    @Data
-    public static class Gauge {
-
-        /**
-         * value
-         */
-        private double value;
-    }
-
-    /**
-     * untyped
-     */
-    @Data
-    public static class Untyped {
-
-        /**
-         * value
-         */
-        private double value;
-    }
-
-    /**
-     * Summary
-     */
-    @Data
-    public static class Summary {
-
-        /**
-         * value
-         */
-        private double value;
-
-        /**
-         * count
-         */
-        private long count;
-
-        /**
-         * sum
-         */
-        private double sum;
-
-        /**
-         * quantileList
-         */
-        private List<Quantile> quantileList = new ArrayList<>();
-    }
-
-    /**
-     * Quantile
-     */
-    @Data
-    public static class Quantile {
-        /**
-         * Corresponding to the quantile field in Prometheus
-         */
-        private double xLabel;
-
-        /**
-         * value
-         */
-        private double value;
-    }
-
-    /**
-     * Histogram
-     */
-    @Data
-    public static class Histogram {
-
-        /**
-         * value
-         */
-        private double value;
-
-        /**
-         * count
-         */
-        private long count;
-
-        /**
-         * sum
-         */
-        private double sum;
-
-        /**
-         * bucketList
-         */
-        private List<Bucket> bucketList = new ArrayList<>();
-    }
-
-    /**
-     * Bucket
-     */
-    @Data
-    public static class Bucket {
-
-        /**
-         * cumulativeCount
-         */
-        private long cumulativeCount;
-
-        /**
-         * upperBound
-         */
-        private double upperBound;
-    }
-}

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/http/promethus/exporter/MetricType.java

@@ -1,52 +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.http.promethus.exporter;
-
-/**
- * prometheus metrics type
- */
-public enum MetricType {
-    // for string metric info
-    INFO("info"),
-    // Represents a monotonically increasing counter, e.g., counting occurrences
-    COUNTER("counter"),
-    // A metric type that can fluctuate up and down, e.g., CPU usage rate
-    GAUGE("gauge"),
-    SUMMARY("summary"),
-    UNTYPED("untyped"),
-    HISTOGRAM("histogram");
-
-    private final String value;
-
-    MetricType(String value) {
-        this.value = value;
-    }
-
-    public String getValue() {
-        return value;
-    }
-
-    public static MetricType getType(String value) {
-        for (MetricType metricType : values()) {
-            if (metricType.getValue().equals(value)) {
-                return metricType;
-            }
-        }
-        return null;
-    }
-}

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/PrometheusAutoCollectImpl.java

@@ -19,6 +19,7 @@ package org.apache.hertzbeat.collector.collect.prometheus;
 
 import static org.apache.hertzbeat.common.constants.SignConstants.RIGHT_DASH;
 import java.io.IOException;
+import java.io.InputStream;
 import java.io.InterruptedIOException;
 import java.net.ConnectException;
 import java.net.UnknownHostException;
@@ -33,8 +34,7 @@ import java.util.stream.Stream;
 import javax.net.ssl.SSLException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.collector.collect.common.http.CommonHttpClient;
-import org.apache.hertzbeat.common.entity.dto.MetricFamily;
-import org.apache.hertzbeat.collector.collect.prometheus.parser.TextParser;
+import org.apache.hertzbeat.collector.collect.prometheus.parser.MetricFamily;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
 import org.apache.hertzbeat.collector.util.CollectUtil;
 import org.apache.hertzbeat.common.constants.CommonConstants;
@@ -46,6 +46,7 @@ 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.hertzbeat.collector.collect.prometheus.parser.OnlineParser;
 import org.apache.http.HttpHeaders;
 import org.apache.http.HttpHost;
 import org.apache.http.HttpStatus;
@@ -64,7 +65,6 @@ 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;
 
@@ -99,25 +99,12 @@ public class PrometheusAutoCollectImpl {
                 builder.setMsg(NetworkConstants.STATUS_CODE + SignConstants.BLANK + statusCode);
                 return null;
             }
-            // todo: The InputStream is directly converted to a String here
-            //       For large data in the Prometheus exporter, this can generate large objects, which could severely impact JVM memory space
-            // todo: Option one: Use InputStream for parsing, but this requires significant code changes
-            //       Option two: Manually trigger garbage collection, which can be referenced from Dubbo for long i
-            String resp = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
-            long collectTime = System.currentTimeMillis();
-            builder.setTime(collectTime);
-            if (resp == null || !StringUtils.hasText(resp)) {
-                log.error("http response content is empty, status: {}.", statusCode);
+            try {
+                return parseResponseByPrometheusExporter(response.getEntity().getContent(), builder);
+            } catch (Exception e) {
+                log.info("parse error: {}.", e.getMessage(), e);
                 builder.setCode(CollectRep.Code.FAIL);
-                builder.setMsg("http response content is empty");
-            } else {
-                try {
-                    return parseResponseByPrometheusExporter(resp, metrics.getAliasFields(), builder);
-                } catch (Exception e) {
-                    log.info("parse error: {}.", e.getMessage(), e);
-                    builder.setCode(CollectRep.Code.FAIL);
-                    builder.setMsg("parse response data error:" + e.getMessage());
-                }   
+                builder.setMsg("parse response data error:" + e.getMessage());
             }
         } catch (ClientProtocolException e1) {
             String errorMsg = CommonUtil.getMessageFromThrowable(e1);
@@ -168,10 +155,12 @@ public class PrometheusAutoCollectImpl {
         }
     }
     
-    private List<CollectRep.MetricsData> parseResponseByPrometheusExporter(String resp, List<String> aliasFields,
-                                                                           CollectRep.MetricsData.Builder builder) {
-        Map<String, MetricFamily> metricFamilyMap = TextParser.textToMetricFamilies(resp);
+    private List<CollectRep.MetricsData> parseResponseByPrometheusExporter(InputStream inputStream, CollectRep.MetricsData.Builder builder) throws IOException {
+        Map<String, MetricFamily> metricFamilyMap = OnlineParser.parseMetrics(inputStream);
         List<CollectRep.MetricsData> metricsDataList = new LinkedList<>();
+        if (metricFamilyMap == null) {
+            return metricsDataList;
+        }
         for (Map.Entry<String, MetricFamily> entry : metricFamilyMap.entrySet()) {
             builder.clearFields();
             builder.clearValues();

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/MetricFamily.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.hertzbeat.common.entity.dto;
+package org.apache.hertzbeat.collector.collect.prometheus.parser;
 
 import java.util.List;
 import lombok.Data;

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.hertzbeat.common.util;
+package org.apache.hertzbeat.collector.collect.prometheus.parser;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -23,7 +23,6 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.hertzbeat.common.entity.dto.MetricFamily;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -38,7 +37,7 @@ import java.util.concurrent.ConcurrentHashMap;
 @Slf4j
 public class OnlineParser {
 
-    private static final Map<Integer, Integer> escapeMap = new HashMap<>();
+    private static final Map<Integer, Integer> escapeMap = new HashMap<>(8);
     
     static {
         escapeMap.put((int) 'n', (int) '\n');
@@ -51,6 +50,30 @@ public class OnlineParser {
         escapeMap.put((int) '\\', (int) '\\');
     }
 
+    private OnlineParser() {
+    }
+
+    public static Map<String, MetricFamily> parseMetrics(InputStream inputStream) throws IOException {
+        Map<String, MetricFamily> metricFamilyMap = new ConcurrentHashMap<>(10);
+        try {
+            int i = getChar(inputStream);
+            while (i != -1) {
+                if (i == '#' || i == '\n') {
+                    skipToLineEnd(inputStream).maybeEol().maybeEof().noElse();
+                } else {
+                    StringBuilder stringBuilder = new StringBuilder();
+                    stringBuilder.append((char) i);
+                    parseMetric(inputStream, metricFamilyMap, stringBuilder);
+                }
+                i = getChar(inputStream);
+            }
+        } catch (FormatException e) {
+            log.error("prometheus parser failed because of wrong input format. {}", e.getMessage());
+            return null;
+        }
+        return metricFamilyMap;
+    }
+
     private static class FormatException extends Exception {
 
         public FormatException() {
@@ -133,7 +156,7 @@ public class OnlineParser {
             return this.i;
         }
 
-        private int getInt() throws FormatException {
+        private int getInt() {
             return this.i;
         }
 
@@ -153,14 +176,9 @@ public class OnlineParser {
         }
     }
 
-    private static CharChecker parseOneChar(InputStream inputStream) throws IOException, FormatException {
-        int i = getChar(inputStream);
-        return new CharChecker(i);
-    }
-
     private static CharChecker parseOneDouble(InputStream inputStream, StringBuilder stringBuilder) throws IOException, FormatException {
         int i = getChar(inputStream);
-        while ((i >= '0' && i <= '9') || (i >= 'a' && i <= 'z') || (i >= 'A' && i <= 'Z') || i == '-' || i == '+' || i == 'e' || i == '.') {
+        while (i >= '0' && i <= '9' || i >= 'a' && i <= 'z' || i >= 'A' && i <= 'Z' || i == '-' || i == '+' || i == '.') {
             stringBuilder.append((char) i);
             i = getChar(inputStream);
         }
@@ -199,17 +217,10 @@ public class OnlineParser {
             if (i == '\\') {
                 i = getChar(inputStream);
                 switch (i) {
-                    case 'n':
-                        stringBuilder.append('\n');
-                        break;
-                    case '\\':
-                        stringBuilder.append('\\');
-                        break;
-                    case '\"':
-                        stringBuilder.append('\"');
-                        break;
-                    default:
-                        throw new FormatException();
+                    case 'n' -> stringBuilder.append('\n');
+                    case '\\' -> stringBuilder.append('\\');
+                    case '\"' -> stringBuilder.append('\"');
+                    default -> throw new FormatException();
                 }
             } else {
                 stringBuilder.append((char) i);
@@ -285,7 +296,7 @@ public class OnlineParser {
     }
 
     private static CharChecker parseMetric(InputStream inputStream, Map<String, MetricFamily> metricFamilyMap, StringBuilder stringBuilder) throws IOException, FormatException {
-        MetricFamily metricFamily = null;
+        MetricFamily metricFamily;
         MetricFamily.Metric metric = new MetricFamily.Metric();
         int i = parseMetricName(inputStream, stringBuilder).maybeSpace().maybeLeftBracket().noElse();
         String metricName = stringBuilder.toString();
@@ -333,26 +344,4 @@ public class OnlineParser {
         metricFamily.getMetricList().add(metric);
         return new CharChecker(i);
     }
-
-    public static Map<String, MetricFamily> parseMetrics(InputStream inputStream) throws IOException {
-        Map<String, MetricFamily> metricFamilyMap = new ConcurrentHashMap<>(10);
-        try {
-            int i = getChar(inputStream);
-            while (i != -1) {
-                if (i == '#' || i == '\n') {
-                    skipToLineEnd(inputStream).maybeEol().maybeEof().noElse();
-                } else {
-                    StringBuilder stringBuilder = new StringBuilder();
-                    stringBuilder.append((char) i);
-                    parseMetric(inputStream, metricFamilyMap, stringBuilder);
-                }
-                i = getChar(inputStream);
-            }
-        } catch (FormatException e) {
-            log.error("prometheus parser failed because of wrong input format. {}", e.getMessage());
-            return null;
-        }
-        return metricFamilyMap;
-    }
-
 }

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/TextParser.java

@@ -24,14 +24,15 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.collector.collect.http.promethus.ParseException;
-import org.apache.hertzbeat.common.entity.dto.MetricFamily;
 import org.apache.hertzbeat.common.util.StrBuffer;
 
 /**
  * Resolves the data passed by prometheus's exporter interface http:xxx/metrics
  * Reference: prometheus text_parse.go code, entry: TextToMetricFamilies
+ * recommend use OnlineParser
  */
 @Slf4j
+@Deprecated(since = "1.7.0")
 public class TextParser {
     private static final String NAME_LABEL = "__name__";
     private static final char LEFT_CURLY_BRACKET = '{';
@@ -45,7 +46,6 @@ public class TextParser {
 
     /**
      * parser prometheus exporter text metrics data
-     * todo use inputStream bytebuffer instead of resp string
      * @param resp txt data
      * @return metrics family
      */

hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/push/PushCollectImpl.java

@@ -58,13 +58,13 @@ public class PushCollectImpl extends AbstractCollect {
     private static final Map<Long, Long> timeMap = new ConcurrentHashMap<>();
 
     // ms
-    private static final Integer timeout = 3000;
+    private static final Integer DEFAULT_TIMEOUT = 3000;
 
     private static final Integer SUCCESS_CODE = 200;
 
     // It's hard to determine how long ago the first data collection was, because there's no way to know when the last collection occurred.
     // This makes it difficult to avoid re-collecting data after a restart. The default is 30 seconds
-    private static final Integer firstCollectInterval = 30000;
+    private static final Integer FIRST_COLLECT_INTERVAL = 30000;
 
     @Override
     public void preCheck(Metrics metrics) throws IllegalArgumentException {
@@ -80,7 +80,7 @@ public class PushCollectImpl extends AbstractCollect {
         long monitorId = builder.getId();
         PushProtocol pushProtocol = metrics.getPush();
 
-        Long time = timeMap.getOrDefault(monitorId, curTime - firstCollectInterval);
+        Long time = timeMap.getOrDefault(monitorId, curTime - FIRST_COLLECT_INTERVAL);
         timeMap.put(monitorId, curTime);
 
         HttpContext httpContext = createHttpContext(pushProtocol);
@@ -145,10 +145,10 @@ public class PushCollectImpl extends AbstractCollect {
 
         //requestBuilder.setUri(pushProtocol.getUri());
 
-        if (timeout > 0) {
+        if (DEFAULT_TIMEOUT > 0) {
             RequestConfig requestConfig = RequestConfig.custom()
-                    .setConnectTimeout(timeout)
-                    .setSocketTimeout(timeout)
+                    .setConnectTimeout(DEFAULT_TIMEOUT)
+                    .setSocketTimeout(DEFAULT_TIMEOUT)
                     .setRedirectsEnabled(true)
                     .build();
             requestBuilder.setConfig(requestConfig);

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/http/promethus/exporter/ExporterParserTest.java

@@ -1,79 +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.http.promethus.exporter;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import java.util.Map;
-import org.junit.jupiter.api.Test;
-
-/**
- * Test case for {@link ExporterParser}
- */
-class ExporterParserTest {
-
-    @Test
-    void textToMetric() {
-        String resp = """
-                # HELP disk_total_bytes Total space for path
-                # TYPE disk_total_bytes gauge
-                disk_total_bytes{path="C:\\\\hertzbeat\\\\repo\\\\testpath",} 4.29496725504E11
-                # HELP go_gc_cycles_automatic_gc_cycles_total Count of completed GC cycles generated by the Go runtime.
-                # TYPE go_gc_cycles_automatic_gc_cycles_total counter
-                go_gc_cycles_automatic_gc_cycles_total 0
-                # HELP go_gc_cycles_forced_gc_cycles_total Count of completed GC cycles forced by the application.
-                # TYPE go_gc_cycles_forced_gc_cycles_total counter
-                go_gc_cycles_forced_gc_cycles_total 0
-                # HELP go_gc_cycles_total_gc_cycles_total Count of all completed GC cycles.
-                # TYPE go_gc_cycles_total_gc_cycles_total counter
-                go_gc_cycles_total_gc_cycles_total 0
-                # HELP go_gc_duration_seconds A summary of the pause duration of garbage collection cycles.
-                # TYPE go_gc_duration_seconds summary
-                go_gc_duration_seconds{quantile="0"} 0
-                go_gc_duration_seconds{quantile="0.25"} 0
-                go_gc_duration_seconds{quantile="0.5"} 0
-                go_gc_duration_seconds{quantile="0.75"} 0
-                go_gc_duration_seconds{quantile="1"} 0
-                # TYPE jvm info
-                # HELP jvm VM version info
-                jvm_info{runtime="OpenJDK Runtime Environment",vendor="Azul Systems, Inc.",version="11.0.13+8-LTS"} 1.0
-                # TYPE jvm_gc_collection_seconds summary
-                # HELP jvm_gc_collection_seconds Time spent in a given JVM garbage collector in seconds.
-                jvm_gc_collection_seconds_count{gc="G1 Young Generation"} 10.0
-                jvm_gc_collection_seconds_sum{gc="G1 Young Generation"} 0.051
-                jvm_gc_collection_seconds_count{gc="G1 Old Generation"} 0.0
-                jvm_gc_collection_seconds_sum{gc="G1 Old Generation"} 0.0
-                # TYPE resource_group_aggregate_usage_secs summary
-                resource_group_aggregate_usage_secs{cluster="standalone",quantile="0.5"} 2.69245E-4
-                resource_group_aggregate_usage_secs{cluster="standalone",quantile="0.9"} 3.49601E-4
-                resource_group_aggregate_usage_secs_count{cluster="standalone"} 13.0
-                resource_group_aggregate_usage_secs_sum{cluster="standalone"} 0.004832498
-                resource_group_aggregate_usage_secs_created{cluster="standalone"} 1.715842140749E9
-                # TYPE metadata_store_ops_latency_ms histogram
-                metadata_store_ops_latency_ms_bucket{cluster="standalone",name="metadata-store",type="get",status="success",le="1.0"} 59.0
-                metadata_store_ops_latency_ms_bucket{cluster="standalone",name="metadata-store",type="get",status="success",le="3.0"} 61.0
-                metadata_store_ops_latency_ms_bucket{cluster="standalone",name="metadata-store",type="get",status="success",le="5.0"} 61.0
-                # EOF""";
-
-        ExporterParser parser = new ExporterParser();
-        Map<String, MetricFamily> metricFamilyMap = parser.textToMetric(resp);
-        assertEquals(5, metricFamilyMap.get("resource_group_aggregate_usage_secs").getMetricList().size());
-        assertEquals(3, metricFamilyMap.get("metadata_store_ops_latency_ms").getMetricList().size());
-        assertEquals(5, metricFamilyMap.get("go_gc_duration_seconds").getMetricList().size());
-        assertEquals(9, metricFamilyMap.size());
-    }
-}

hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java

@@ -17,8 +17,6 @@
 
 package org.apache.hertzbeat.collector.collect.prometheus.parser;
 
-import org.apache.hertzbeat.common.entity.dto.MetricFamily;
-import org.apache.hertzbeat.common.util.OnlineParser;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
@@ -27,8 +25,12 @@ import java.io.InputStream;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.fail;
 
 class OnlineParserTest {
 
@@ -46,13 +48,13 @@ class OnlineParserTest {
         String str = """
                 # HELP go_gc_duration_seconds A summary of the pause duration of garbage collection cycles.
                 # TYPE go_gc_duration_seconds summary
-                go_gc_duration_seconds {  quantile="0"} 2.0209e-05 321312
-                go_gc_duration_seconds{  quantile = "0.25"  }   6.6917e-05
+                go_gc_duration_seconds{quantile="0"} 2.0209e-05
+                go_gc_duration_seconds{quantile="0.25"} 6.6917e-05
                 go_gc_duration_seconds{quantile="0.5"} -Inf
-                go_gc_duration_seconds{ quantile = "0.75"} +Inf
+                go_gc_duration_seconds{quantile="0.75"} +Inf
                 go_gc_duration_seconds{quantile="1"} NaN
-                go_gc_duration_seconds_sum 0.001134793 321314
-                go_gc_duration_seconds_count 5 43
+                go_gc_duration_seconds_sum 0.001134793
+                go_gc_duration_seconds_count 5
                 # HELP go_goroutines Number of goroutines that currently exist.
                 # TYPE go_goroutines gauge
                 go_goroutines 32
@@ -75,7 +77,26 @@ class OnlineParserTest {
                 # TYPE go_memstats_gc_sys_bytes gauge
                 go_memstats_gc_sys_bytes 4.614808e+06""";
         InputStream inputStream = new ByteArrayInputStream(str.getBytes(StandardCharsets.UTF_8));
-        Map<String, MetricFamily> metricFamilyMap = OnlineParser.parseMetrics(inputStream);
-        assertNotNull(metricFamilyMap);
+        Map<String, MetricFamily> metricFamilyMap1 = OnlineParser.parseMetrics(inputStream);
+        Map<String, MetricFamily> metricFamilyMap2 = TextParser.textToMetricFamilies(str);
+        assertNotNull(metricFamilyMap1);
+        assertNotNull(metricFamilyMap2);
+        assertEquals(metricFamilyMap1.size(), metricFamilyMap2.size());
+        metricFamilyMap2.forEach((metricFamilyName, metricFamily2) -> {
+            if (!metricFamilyMap1.containsKey(metricFamilyName)) {
+                fail("parse failed, different result from two parser.");
+            }
+            MetricFamily metricFamily1 = metricFamilyMap1.get(metricFamilyName);
+            assertEquals(metricFamily1.getName(), metricFamily1.getName());
+            Set<Double> metricValueSet = metricFamily2.getMetricList().stream().map(MetricFamily.Metric::getValue).collect(Collectors.toSet());
+            metricFamily1.getMetricList().forEach(metric -> {
+                // this is for something different between two algorithms above, and both of them is current on this parsing behavior.
+                if (!(metric.getValue() == Double.POSITIVE_INFINITY || metric.getValue() == Double.NEGATIVE_INFINITY)) {
+                    if (!metricValueSet.contains(metric.getValue())) {
+                        fail();
+                    }
+                }
+            });
+        });
     }
 }

hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/query/DatasourceQuery.java

@@ -0,0 +1,59 @@
+/*
+ * 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.dto.query;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Metric History Range Query Data
+ */
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+@Schema(description = "Query Request Data")
+public class DatasourceQuery {
+    
+    @Schema(title = "Ref Id, unique id for the query")
+    private String refId;
+    
+    @Schema(title = "datasource name")
+    private String datasource;
+    
+    @Schema(title = "query expr, like prometheus query")
+    private String expr;
+    
+    @Schema(title = "query expr type, like promql or sql or influxql")
+    private String exprType;
+    
+    @Schema(title = "query range type, like range or instant")
+    private String timeType;
+
+    @Schema(title = "query range start time")
+    private Long start;
+
+    @Schema(title = "query range end time")
+    private Long end;
+    
+    @Schema(title = "query time step, like 5m or 1h")
+    private String step;
+}

hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/query/DatasourceQueryData.java

@@ -18,14 +18,13 @@
 package org.apache.hertzbeat.common.entity.dto.query;
 
 import io.swagger.v3.oas.annotations.media.Schema;
+import java.util.List;
+import java.util.Map;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
-import java.util.List;
-import java.util.Map;
-
 /**
  * Metric History Range Query Data
  */
@@ -34,14 +33,36 @@ import java.util.Map;
 @AllArgsConstructor
 @NoArgsConstructor
 @Schema(description = "Metric Query Data")
-public class MetricQueryData {
+public class DatasourceQueryData {
+
+    @Schema(title = "Ref Id, unique id for the query")
+    private String refId;
     
-    @Schema(title = "Metric Schema")
-    private MetricSchema schema;
+    @Schema(title = "query status code, 200 for success, other for error")
+    private Integer status;
+    
+    @Schema(title = "query error message")
+    private String msg;
+    
+    @Schema(title = "query result data frames")
+    private List<SchemaData> frames;
+    
+    /**
+     * Schema Data
+     */
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    @Builder
+    public static final class SchemaData {
 
-    @Schema(title = "metrics row values, first is the timestamp-ts", example = "[[29,32,44],[32,34,true]]")
-    private List<List<Object>> values;
+        @Schema(title = "Metric Schema")
+        private MetricSchema schema;
 
+        @Schema(title = "metrics row values, first is the timestamp-ts", example = "[[29,32,44],[32,34,true]]")
+        private List<Object[]> data;
+    }
+    
     /**
      * Metric Schema
      */
@@ -54,6 +75,9 @@ public class MetricQueryData {
         @Schema(title = "Metrics Field")
         private List<MetricField> fields;
         
+        @Schema(title = "This frame labels")
+        private Map<String, String> labels;
+        
         @Schema(title = "Meta Information")
         private Map<String, String> meta;
     }
@@ -75,8 +99,5 @@ public class MetricQueryData {
 
         @Schema(title = "Field Unit: %, Mb, Kbps etc.")
         private String unit;
-
-        @Schema(title = "Whether is a label")
-        private Boolean label;
     }
 }

hertzbeat-common/src/main/java/org/apache/hertzbeat/common/util/TimePeriodUtil.java

@@ -49,4 +49,46 @@ public final class TimePeriodUtil {
             return Duration.parse("PT" + tokenTime);
         }
     }
+
+    /**
+     * transform any timestamp to milliseconds
+     * @param timestamp timestamp
+     * @return milliseconds
+     */
+    public static long normalizeToMilliseconds(Object timestamp) {
+        if (timestamp instanceof String timestampStr) {
+            // string type, may be second, millisecond or decimal second
+            // eg: "1672531199000", "1672531199", "1672531199.123"
+            if (timestampStr.contains(".")) {
+                // contains decimal point, parse as second timestamp
+                double seconds = Double.parseDouble(timestampStr);
+                return (long) (seconds * 1000);
+            } else {
+                // integer form, determine second or millisecond
+                long numericTimestamp = Long.parseLong(timestampStr);
+                return convertNumericTimestamp(numericTimestamp);
+            }
+        } else if (timestamp instanceof Number) {
+            // number eg Integer、Long、Double
+            if (timestamp instanceof Double || timestamp instanceof Float) {
+                // float type, treat as second timestamp
+                double seconds = ((Number) timestamp).doubleValue();
+                return (long) (seconds * 1000);
+            } else {
+                // integer type, directly determine second or millisecond
+                long numericTimestamp = ((Number) timestamp).longValue();
+                return convertNumericTimestamp(numericTimestamp);
+            }
+        } else {
+            throw new IllegalArgumentException("Not support this timestamp type: " + timestamp.getClass().getName());
+        }
+    }
+
+    private static long convertNumericTimestamp(long numericTimestamp) {
+        if (String.valueOf(numericTimestamp).length() <= 10) {
+            return numericTimestamp * 1000;
+        } else {
+            return numericTimestamp;
+        }
+    }
 }

hertzbeat-push/pom.xml

@@ -29,10 +29,11 @@
     <name>${project.artifactId}</name>
 
     <dependencies>
-        <!-- common -->
+        <!-- collector basic -->
         <dependency>
             <groupId>org.apache.hertzbeat</groupId>
-            <artifactId>hertzbeat-common</artifactId>
+            <artifactId>hertzbeat-collector-basic</artifactId>
+            <version>2.0-SNAPSHOT</version>
         </dependency>
         <!-- spring -->
         <dependency>

hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java

@@ -28,12 +28,12 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.collector.collect.prometheus.parser.MetricFamily;
+import org.apache.hertzbeat.collector.collect.prometheus.parser.OnlineParser;
 import org.apache.hertzbeat.common.constants.CommonConstants;
-import org.apache.hertzbeat.common.entity.dto.MetricFamily;
 import org.apache.hertzbeat.common.entity.manager.Monitor;
 import org.apache.hertzbeat.common.entity.message.CollectRep;
 import org.apache.hertzbeat.common.queue.CommonDataQueue;
-import org.apache.hertzbeat.common.util.OnlineParser;
 import org.apache.hertzbeat.common.util.SnowFlakeIdGenerator;
 import org.apache.hertzbeat.push.dao.PushMonitorDao;
 import org.apache.hertzbeat.push.service.PushGatewayService;

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/constants/WarehouseConstants.java

@@ -60,4 +60,8 @@ public interface WarehouseConstants {
 
     String SQL = "sql";
 
+    String RANGE = "range";
+
+    String INSTANT = "instant";
+
 }

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/DataQueryController.java

@@ -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.warehouse.controller;
+
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import java.util.List;
+import org.apache.hertzbeat.common.entity.dto.Message;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQuery;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQueryData;
+import org.apache.hertzbeat.warehouse.service.DatasourceQueryService;
+import org.springframework.beans.factory.annotation.Autowired;
+import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Metrics Data Query API
+ */
+@RestController
+@RequestMapping(produces = {APPLICATION_JSON_VALUE})
+@Tag(name = "Metrics Data Query API")
+public class DataQueryController {
+    
+    @Autowired(required = false)
+    private DatasourceQueryService datasourceQueryService;
+
+    @PostMapping("/api/warehouse/query")
+    @Operation(summary = "Warehouse Query")
+    public ResponseEntity<Message<List<DatasourceQueryData>>> query(
+            @Parameter(description = "Query Expr") @RequestBody List<DatasourceQuery> queries) {
+        return ResponseEntity.ok(Message.success(datasourceQueryService.query(queries)));
+    }
+}

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataQueryController.java

@@ -1,78 +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.warehouse.controller;
-
-import io.swagger.v3.oas.annotations.Operation;
-import io.swagger.v3.oas.annotations.Parameter;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import org.apache.hertzbeat.common.entity.dto.Message;
-import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
-import org.apache.hertzbeat.warehouse.service.MetricsDataQueryService;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.http.ResponseEntity;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestParam;
-import org.springframework.web.bind.annotation.RestController;
-
-import java.util.List;
-
-import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
-
-/**
- * Metrics Data Query API
- */
-@RestController
-@RequestMapping(produces = {APPLICATION_JSON_VALUE})
-@Tag(name = "Metrics Data Query API")
-public class MetricsDataQueryController {
-    
-
-    @Autowired
-    private MetricsDataQueryService queryService;
-    
-
-    @GetMapping("/api/warehouse/query")
-    @Operation(summary = "Query Real Time Metrics Data")
-    public ResponseEntity<Message<List<MetricQueryData>>> queryMetricsData(
-            @Parameter(description = "Query PromQL expr list", example = "cpu")
-            @RequestParam List<String> queries,
-            @Parameter(description = "Query type", example = "promql")
-            @RequestParam String type,
-            @Parameter(description = "Query timestamp", example = "1725854804451")
-            @RequestParam long time) {
-        return ResponseEntity.ok(Message.success(queryService.query(queries, type, time)));
-    }
-
-    @GetMapping("/api/warehouse/query/range")
-    @Operation(summary = "Query Range Metrics Data")
-    public ResponseEntity<Message<List<MetricQueryData>>> queryMetricsDataRange(
-            @Parameter(description = "Query PromQL expr list", example = "cpu")
-            @RequestParam List<String> queries,
-            @Parameter(description = "Query type", example = "promql")
-            @RequestParam String type,
-            @Parameter(description = "Query start timestamp", example = "1725854804451")
-            @RequestParam long start,
-            @Parameter(description = "Query end timestamp", example = "1733630804452")
-            @RequestParam long end,
-            @Parameter(description = "Query step", example = "4m")
-            @RequestParam String step
-    ) {
-        return ResponseEntity.ok(Message.success(queryService.queryRange(queries, type, start, end, step)));
-    }
-}

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimePromqlQueryExecutor.java

@@ -33,7 +33,9 @@ import org.springframework.web.client.RestTemplate;
 @Slf4j
 public class GreptimePromqlQueryExecutor extends PromqlQueryExecutor {
 
-    private static final String QUERY_PATH = "/v1/prometheus/api/v1/query";
+    private static final String QUERY_PATH = "/v1/prometheus";
+
+    private static final String Datasource = "Greptime";
 
     private final GreptimeProperties greptimeProperties;
 
@@ -42,4 +44,10 @@ public class GreptimePromqlQueryExecutor extends PromqlQueryExecutor {
                 greptimeProperties.username(), greptimeProperties.password()));
         this.greptimeProperties = greptimeProperties;
     }
+
+    @Override
+    public String getDatasource() {
+        return Datasource;
+    }
+
 }

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/PromqlQueryExecutor.java

@@ -20,10 +20,16 @@
 package org.apache.hertzbeat.warehouse.db;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.codec.binary.Base64;
 import org.apache.hertzbeat.common.constants.NetworkConstants;
 import org.apache.hertzbeat.common.constants.SignConstants;
-import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQuery;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQueryData;
 import org.apache.hertzbeat.common.util.Base64Util;
+import org.apache.hertzbeat.common.util.TimePeriodUtil;
+import static org.apache.hertzbeat.warehouse.constants.WarehouseConstants.INSTANT;
+import static org.apache.hertzbeat.warehouse.constants.WarehouseConstants.PROMQL;
+import static org.apache.hertzbeat.warehouse.constants.WarehouseConstants.RANGE;
 import org.apache.hertzbeat.warehouse.store.history.vm.PromQlQueryContent;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpEntity;
@@ -35,15 +41,12 @@ import org.springframework.web.client.RestTemplate;
 import org.springframework.web.util.UriComponentsBuilder;
 
 import java.net.URI;
-import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.hertzbeat.warehouse.constants.WarehouseConstants.PROMQL;
-
 /**
  * abstract class for promql query executor
  */
@@ -51,11 +54,15 @@ import static org.apache.hertzbeat.warehouse.constants.WarehouseConstants.PROMQL
 public abstract class PromqlQueryExecutor implements QueryExecutor {
 
     private static final String supportQueryLanguage = PROMQL;
+    private static final String QUERY_RANGE_PATH = "/api/v1/query_range";
+    private static final String QUERY_PATH = "/api/v1/query";
     protected static final String HTTP_QUERY_PARAM = "query";
     protected static final String HTTP_TIME_PARAM = "time";
     protected static final String HTTP_START_PARAM = "start";
     protected static final String HTTP_END_PARAM = "end";
     protected static final String HTTP_STEP_PARAM = "step";
+    private static final String INNER_KEY_TIME = "__ts__";
+    private static final String INNER_KEY_VALUE = "__value__";
 
     private final RestTemplate restTemplate;
 
@@ -69,14 +76,15 @@ public abstract class PromqlQueryExecutor implements QueryExecutor {
     /**
      * record class for promql http connection
      */
-    protected record HttpPromqlProperties (
-        String url,
-        String username,
-        String password
-    ){}
+    protected record HttpPromqlProperties(
+            String url,
+            String username,
+            String password
+    ) {
+    }
 
-    protected List<Map<String, Object>> http_promql(Map<String, Object> params) {
-        // http run the promql query
+    @Override
+    public List<Map<String, Object>> execute(String queryString) {
         List<Map<String, Object>> results = new LinkedList<>();
         try {
             HttpHeaders headers = new HttpHeaders();
@@ -86,13 +94,12 @@ public abstract class PromqlQueryExecutor implements QueryExecutor {
                     && StringUtils.hasText(httpPromqlProperties.password())) {
                 String authStr = httpPromqlProperties.username() + ":" + httpPromqlProperties.password();
                 String encodedAuth = Base64Util.encode(authStr);
-                headers.add(HttpHeaders.AUTHORIZATION,  NetworkConstants.BASIC + SignConstants.BLANK + encodedAuth);
+                headers.add(HttpHeaders.AUTHORIZATION, NetworkConstants.BASIC + SignConstants.BLANK + encodedAuth);
             }
             HttpEntity<Void> httpEntity = new HttpEntity<>(headers);
+
             UriComponentsBuilder uriComponentsBuilder = UriComponentsBuilder.fromHttpUrl(httpPromqlProperties.url);
-            for (Map.Entry<String, Object> entry : params.entrySet()) {
-                uriComponentsBuilder.queryParam(entry.getKey(), entry.getValue());
-            }
+            uriComponentsBuilder.queryParam(HTTP_QUERY_PARAM, queryString);
             URI uri = uriComponentsBuilder.build(true).toUri();
             ResponseEntity<PromQlQueryContent> responseEntity = restTemplate.exchange(uri,
                     HttpMethod.GET, httpEntity, PromQlQueryContent.class);
@@ -126,41 +133,85 @@ public abstract class PromqlQueryExecutor implements QueryExecutor {
         return results;
     }
 
-    public MetricQueryData convertToMetricQueryData(Object object) {
-        MetricQueryData metricQueryData = new MetricQueryData();
+    @Override
+    public DatasourceQueryData query(DatasourceQuery datasourceQuery) {
+        DatasourceQueryData.DatasourceQueryDataBuilder queryDataBuilder = DatasourceQueryData.builder()
+                .refId(datasourceQuery.getRefId()).status(200);
         try {
-            List<Map<String, Object>> metrics = (List<Map<String, Object>>) object;
-            // todo
+            HttpHeaders headers = new HttpHeaders();
+            headers.setContentType(MediaType.APPLICATION_JSON);
+            headers.setAccept(List.of(MediaType.APPLICATION_JSON));
+            if (StringUtils.hasText(httpPromqlProperties.username())
+                    && StringUtils.hasText(httpPromqlProperties.password())) {
+                String authStr = httpPromqlProperties.username() + ":" + httpPromqlProperties.password();
+                String encodedAuth = new String(Base64.encodeBase64(authStr.getBytes(StandardCharsets.UTF_8)), StandardCharsets.UTF_8);
+                headers.add(HttpHeaders.AUTHORIZATION, NetworkConstants.BASIC + " " + encodedAuth);
+            }
+            HttpEntity<Void> httpEntity = new HttpEntity<>(headers);
+            URI uri;
+            if (datasourceQuery.getTimeType().equals(RANGE)) {
+                uri = UriComponentsBuilder.fromHttpUrl(httpPromqlProperties.url() + QUERY_RANGE_PATH)
+                        .queryParam(HTTP_QUERY_PARAM, datasourceQuery.getExpr())
+                        .queryParam(HTTP_START_PARAM, datasourceQuery.getStart())
+                        .queryParam(HTTP_END_PARAM, datasourceQuery.getEnd())
+                        .queryParam(HTTP_STEP_PARAM, datasourceQuery.getStep())
+                        .build().toUri();
+            } else if (datasourceQuery.getTimeType().equals(INSTANT)) {
+                uri = UriComponentsBuilder.fromHttpUrl(httpPromqlProperties.url() + QUERY_PATH)
+                        .queryParam(HTTP_QUERY_PARAM, datasourceQuery.getExpr())
+                        .build().toUri();
+            } else {
+                throw new IllegalArgumentException(String.format("no such time type for query id {}.", datasourceQuery.getRefId()));
+            }
+            ResponseEntity<PromQlQueryContent> responseEntity = restTemplate.exchange(uri, HttpMethod.GET, httpEntity,
+                    PromQlQueryContent.class);
+            if (responseEntity.getStatusCode().is2xxSuccessful()) {
+                log.debug("query metrics data from promql http api success. {}", uri);
+                if (responseEntity.getBody() != null && responseEntity.getBody().getData() != null
+                        && responseEntity.getBody().getData().getResult() != null) {
+                    List<PromQlQueryContent.ContentData.Content> contents = responseEntity.getBody().getData().getResult();
+                    List<DatasourceQueryData.SchemaData> schemaDataList = new LinkedList<>();
+                    for (PromQlQueryContent.ContentData.Content content : contents) {
+                        DatasourceQueryData.MetricSchema.MetricSchemaBuilder schemaBuilder = DatasourceQueryData.MetricSchema
+                                .builder().fields(List.of(
+                                        // todo: unit?
+                                        DatasourceQueryData.MetricField.builder().name(INNER_KEY_TIME)
+                                                .type("time").build(),
+                                        DatasourceQueryData.MetricField.builder().name(INNER_KEY_VALUE)
+                                                .type("number").build()
+                                )).labels(content.getMetric());
+                        List<Object[]> values;
+                        if (datasourceQuery.getTimeType().equals(RANGE)) {
+                            values = content.getValues();
+                        }
+                        else {
+                            values = List.<Object[]>of(content.getValue());
+                        }
+                        values.forEach(objects -> {
+                            objects[0] = TimePeriodUtil.normalizeToMilliseconds(objects[0]);
+                        });
+                        DatasourceQueryData.SchemaData.SchemaDataBuilder schemaData = DatasourceQueryData.SchemaData.builder()
+                                .schema(schemaBuilder.build()).data(values);
+                        schemaDataList.add(schemaData.build());
+                    }
+                    queryDataBuilder.frames(schemaDataList);
+                }
+            } else {
+                log.error("query metrics data from victoria-metrics failed. {}", responseEntity);
+                queryDataBuilder.msg("query metrics data from victoria-metrics failed. ");
+                queryDataBuilder.status(responseEntity.getStatusCode().value());
+            }
         } catch (Exception e) {
-            log.error("converting to metric query data failed.");
+            log.error("query metrics data from victoria-metrics error. {}.", e.getMessage(), e);
+            queryDataBuilder.msg("query metrics data from victoria-metrics error: " + e.getMessage());
+            queryDataBuilder.status(400);
         }
-        return metricQueryData;
+        return queryDataBuilder.build();
     }
-
-    public List<Map<String, Object>> execute(String queryString) {
-        Map<String, Object> params = new HashMap<>();
-        params.put(HTTP_QUERY_PARAM, URLEncoder.encode(queryString, StandardCharsets.UTF_8));
-        return http_promql(params);
-    }
-
-    public List<Map<String, Object>> query(String queryString, long time) {
-        Map<String, Object> params = new HashMap<>();
-        params.put(HTTP_QUERY_PARAM, URLEncoder.encode(queryString, StandardCharsets.UTF_8));
-        params.put(HTTP_TIME_PARAM, time);
-        return http_promql(params);
-    }
-
-    public List<Map<String, Object>> query_range(String queryString, long start, long end, String step) {
-        Map<String, Object> params = new HashMap<>();
-        params.put(HTTP_QUERY_PARAM, URLEncoder.encode(queryString, StandardCharsets.UTF_8));
-        params.put(HTTP_START_PARAM, start);
-        params.put(HTTP_END_PARAM, end);
-        params.put(HTTP_STEP_PARAM, step);
-        return http_promql(params);
-    }
-
-    public boolean support(String datasource) {
-        return supportQueryLanguage.equals(datasource);
+    
+    @Override
+    public boolean support(String queryLanguage) {
+        return StringUtils.hasText(queryLanguage) && queryLanguage.equalsIgnoreCase(supportQueryLanguage);
     }
 
 }

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/QueryExecutor.java

@@ -17,7 +17,8 @@
 
 package org.apache.hertzbeat.warehouse.db;
 
-import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQuery;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQueryData;
 
 import java.util.List;
 import java.util.Map;
@@ -27,13 +28,11 @@ import java.util.Map;
  */
 public interface QueryExecutor {
 
-    MetricQueryData convertToMetricQueryData(Object object);
-
     List<Map<String, Object>> execute(String query);
-
-    List<Map<String, Object>> query(String query, long time);
-
-    List<Map<String, Object>> query_range(String query, long start, long end, String step);
     
-    boolean support(String datasource);
+    DatasourceQueryData query(DatasourceQuery datasourceQuery);
+
+    String getDatasource();
+
+    boolean support(String queryLanguage);
 }

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/SqlQueryExecutor.java

@@ -20,13 +20,15 @@
 package org.apache.hertzbeat.warehouse.db;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQuery;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQueryData;
+
+import static org.apache.hertzbeat.warehouse.constants.WarehouseConstants.SQL;
+import org.springframework.util.StringUtils;
 
 import java.util.List;
 import java.util.Map;
 
-import static org.apache.hertzbeat.warehouse.constants.WarehouseConstants.SQL;
-
 /**
  * abstract class for sql query executor
  */
@@ -40,28 +42,19 @@ public abstract class SqlQueryExecutor implements QueryExecutor {
      */
     protected record ConnectorSqlProperties () {}
 
-    protected abstract List<Map<String, Object>> do_sql(Map<String, Object> params);
-
-    public MetricQueryData convertToMetricQueryData(Object object) {
-        MetricQueryData metricQueryData = new MetricQueryData();
-        try {
-            List<Map<String, Object>> metrics = (List<Map<String, Object>>) object;
-            // todo
-        } catch (Exception e) {
-            log.error("converting to metric query data failed.");
-        }
-        return metricQueryData;
+    @Override
+    public List<Map<String, Object>> execute(String query) {
+        return null;
     }
 
-    public abstract List<Map<String, Object>> execute(String query);
-
-    public abstract List<Map<String, Object>> query(String query, long time);
-
-    public abstract List<Map<String, Object>> query_range(String query, long start, long end, String step);
-
-
-    public boolean support(String datasource) {
-        return supportQueryLanguage.equals(datasource);
+    @Override
+    public DatasourceQueryData query(DatasourceQuery datasourceQuery) {
+        return null;
     }
 
-}
+    @Override
+    public boolean support(String queryLanguage) {
+        return StringUtils.hasText(queryLanguage) && queryLanguage.equalsIgnoreCase(supportQueryLanguage);
+    }
+
+}

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/VictoriaMetricsQueryExecutor.java

@@ -31,15 +31,19 @@ import org.springframework.web.client.RestTemplate;
 @ConditionalOnProperty(prefix = "warehouse.store.victoria-metrics", name = "enabled", havingValue = "true")
 @Slf4j
 public class VictoriaMetricsQueryExecutor extends PromqlQueryExecutor {
-
-    private static final String QUERY_PATH = "/api/v1/query";
+    private static final String Datasource = "VictoriaMetrics";
     
     private final VictoriaMetricsProperties victoriaMetricsProp;
 
     public VictoriaMetricsQueryExecutor(VictoriaMetricsProperties victoriaMetricsProp, RestTemplate restTemplate) {
-        super(restTemplate, new HttpPromqlProperties(victoriaMetricsProp.url() + QUERY_PATH,
+        super(restTemplate, new HttpPromqlProperties(victoriaMetricsProp.url(),
                 victoriaMetricsProp.username(), victoriaMetricsProp.password()));
         this.victoriaMetricsProp = victoriaMetricsProp;
     }
 
+    @Override
+    public String getDatasource() {
+        return Datasource;
+    }
+
 }

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/DatasourceQueryService.java

@@ -17,30 +17,19 @@
 
 package org.apache.hertzbeat.warehouse.service;
 
-import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
-
 import java.util.List;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQuery;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQueryData;
 
 /**
  * metrics data query service
  */
-public interface MetricsDataQueryService {
-    
+public interface DatasourceQueryService {
+
     /**
      * Query metrics data
      * @param queries query expr
-     * @param time time
      * @return data
      */
-    List<MetricQueryData> query(List<String> queries, String queryType, long time);
-
-    /**
-     * Query metrics data range
-     * @param queries query expr
-     * @param start start
-     * @param end end
-     * @param step step
-     * @return data
-     */
-    List<MetricQueryData> queryRange(List<String> queries, String queryType, long start, long end, String step);
+    List<DatasourceQueryData> query(List<DatasourceQuery> queries);
 }

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/DatasourceQueryServiceImpl.java

@@ -0,0 +1,60 @@
+/*
+ * 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.warehouse.service.impl;
+
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQuery;
+import org.apache.hertzbeat.common.entity.dto.query.DatasourceQueryData;
+import org.apache.hertzbeat.warehouse.db.QueryExecutor;
+import org.apache.hertzbeat.warehouse.service.DatasourceQueryService;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * datasource query service impl
+ */
+@Service
+public class DatasourceQueryServiceImpl implements DatasourceQueryService {
+    Map<String, QueryExecutor> executorMap;
+
+    DatasourceQueryServiceImpl(List<QueryExecutor> executors) {
+        executorMap = executors.stream().collect(Collectors.toMap(QueryExecutor::getDatasource, executor -> executor));
+    }
+
+    @Override
+    public List<DatasourceQueryData> query(List<DatasourceQuery> queries) {
+        if (queries == null) {
+            throw new IllegalArgumentException("No query found");
+        }
+        List<DatasourceQueryData> datasourceQueryDataList = new ArrayList<>();
+        for (DatasourceQuery datasourceQuery : queries) {
+            QueryExecutor executor = executorMap.get(datasourceQuery.getDatasource());
+            if (executor == null) {
+                throw new IllegalArgumentException("Unsupported datasource: " + datasourceQuery.getDatasource());
+            }
+            datasourceQueryDataList.add(executor.query(datasourceQuery));
+        }
+
+        return datasourceQueryDataList;
+    }
+}

hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataQueryServiceImpl.java

@@ -1,67 +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.warehouse.service.impl;
-
-import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
-import org.apache.hertzbeat.warehouse.db.QueryExecutor;
-import org.apache.hertzbeat.warehouse.service.MetricsDataQueryService;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import java.util.ArrayList;
-import java.util.List;
-
-@Service
-public class MetricsDataQueryServiceImpl implements MetricsDataQueryService {
-    @Autowired(required = false)
-    List<QueryExecutor> executors;
-
-    @Override
-    public List<MetricQueryData> query(List<String> queries, String queryType, long time) {
-        if (queries == null || executors.isEmpty()) {
-            throw new IllegalArgumentException("No query executor found");
-        }
-        QueryExecutor executor = executors.stream().filter(e -> e.support(queryType)).findFirst().orElse(null);
-        if (executor == null) {
-            throw new IllegalArgumentException("Unsupported datasource: " + queryType);
-        }
-        List<MetricQueryData> metricQueryDataList = new ArrayList<>();
-        for (String query : queries) {
-            metricQueryDataList.add(executor.convertToMetricQueryData(executor.query(query, time)));
-        }
-        return metricQueryDataList;
-    }
-
-    @Override
-    public List<MetricQueryData> queryRange(List<String> queries, String queryType, long start, long end, String step) {
-        if (queries == null || executors.isEmpty()) {
-            throw new IllegalArgumentException("No query executor found");
-        }
-        QueryExecutor executor = executors.stream().filter(e -> e.support(queryType)).findFirst().orElse(null);
-        if (executor == null) {
-            throw new IllegalArgumentException("Unsupported datasource: " + queryType);
-        }
-        List<MetricQueryData> metricQueryDataList = new ArrayList<>();
-        for (String query : queries) {
-            metricQueryDataList.add(executor.convertToMetricQueryData(executor.query_range(query, start, end, step)));
-        }
-        return metricQueryDataList;
-    }
-}