Merge branch 'master' into pushgateway

fix QueryExecutor inject
2026-09-17 18:19:02 +00:00 · 2025-04-05 13:52:00 +08:00 · 2025-04-05 12:44:29 +08:00 · 2025-04-05 01:58:46 +08:00 · 2025-04-04 23:42:52 +08:00 · 2025-04-04 23:36:40 +08:00
36 changed files with 946 additions and 775 deletions
@@ -41,12 +41,12 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import java.util.HashMap;
 import org.apache.hertzbeat.alert.service.impl.DataSourceServiceImpl;
+import org.apache.hertzbeat.warehouse.db.QueryExecutor;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import java.util.List;
 import java.util.Map;
 import org.mockito.Mockito;
-import org.apache.hertzbeat.warehouse.db.QueryExecutor;

 /**
 * test case for {@link DataSourceService}
@@ -33,7 +33,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.collector.collect.prometheus.parser.MetricFamily;
+import org.apache.hertzbeat.common.entity.dto.MetricFamily;
 import org.apache.hertzbeat.collector.collect.prometheus.parser.TextParser;
 import org.apache.hertzbeat.collector.dispatch.DispatchConstants;
 import org.apache.hertzbeat.collector.util.CollectUtil;
@@ -24,6 +24,7 @@ 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;

 /**
@@ -17,6 +17,8 @@

 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;

@@ -455,4 +455,19 @@ public interface CommonConstants {
     * status page incident state resolved
     */
    byte STATUS_PAGE_INCIDENT_STATE_RESOLVED = 3;
+
+    /**
+     * status page incident state resolved
+     */
+    byte MONITOR_TYPE_NORMAL = 0;
+
+    /**
+     * status page incident state resolved
+     */
+    byte MONITOR_TYPE_PUSH_AUTO_CREATE = 1;
+
+    /**
+     * status page incident state resolved
+     */
+    byte MONITOR_TYPE_DISCOVERY_AUTO_CREATE = 2;
 }
@@ -15,7 +15,7 @@
 * limitations under the License.
 */

-package org.apache.hertzbeat.collector.collect.prometheus.parser;
+package org.apache.hertzbeat.common.entity.dto;

 import java.util.List;
 import lombok.Data;
@@ -0,0 +1,82 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Metric History Range Query Data
+ */
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+@Schema(description = "Metric Query Data")
+public class MetricQueryData {
+    
+    @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<List<Object>> values;
+
+    /**
+     * Metric Schema
+     */
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    @Builder
+    public static final class MetricSchema {
+
+        @Schema(title = "Metrics Field")
+        private List<MetricField> fields;
+        
+        @Schema(title = "Meta Information")
+        private Map<String, String> meta;
+    }
+
+    /**
+     * Metric Field
+     */
+    @Data
+    @AllArgsConstructor
+    @NoArgsConstructor
+    @Builder
+    public static final class MetricField {
+
+        @Schema(title = "Metric Field Name")
+        private String name;
+
+        @Schema(title = "Field Type: number, string, time, bool")
+        private String type;
+
+        @Schema(title = "Field Unit: %, Mb, Kbps etc.")
+        private String unit;
+
+        @Schema(title = "Whether is a label")
+        private Boolean label;
+    }
+}
@@ -91,6 +91,9 @@ public class Monitor {
    @Max(4)
    private byte status;
    
+    @Schema(title = "Task type 0: Normal, 1: push auto create, 2: discovery auto create")
+    private byte type;
+    
    @Schema(title = "task label", example = "{env:test}", accessMode = READ_WRITE)
    @Convert(converter = JsonMapAttributeConverter.class)
    @Column(length = 4096)
@@ -15,18 +15,20 @@
 * limitations under the License.
 */

-package org.apache.hertzbeat.collector.collect.prometheus.parser;
+package org.apache.hertzbeat.common.util;

+import java.util.ArrayList;
+import java.util.HashMap;
+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;
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;


@@ -36,9 +38,23 @@ import java.util.concurrent.ConcurrentHashMap;
@Slf4j
 public class OnlineParser {

+    private static final Map<Integer, Integer> escapeMap = new HashMap<>();
+    
+    static {
+        escapeMap.put((int) 'n', (int) '\n');
+        escapeMap.put((int) 'b', (int) '\b');
+        escapeMap.put((int) 't', (int) '\t');
+        escapeMap.put((int) 'r', (int) '\r');
+        escapeMap.put((int) 'f', (int) '\f');
+        escapeMap.put((int) '\'', (int) '\'');
+        escapeMap.put((int) '\"', (int) '\"');
+        escapeMap.put((int) '\\', (int) '\\');
+    }
+
    private static class FormatException extends Exception {

-        public FormatException() {}
+        public FormatException() {
+        }

        public FormatException(String message) {
            super(message);
@@ -123,51 +139,65 @@ public class OnlineParser {

    }

-    private static CharChecker parseOneChar(InputStream inputStream) throws IOException {
+    private static int getChar(InputStream inputStream) throws IOException, FormatException {
        int i = inputStream.read();
+        if (i == '\\') {
+            i = inputStream.read();
+            if (escapeMap.containsKey(i)) {
+                return escapeMap.get(i);
+            } else {
+                throw new FormatException("Escape character failed.");
+            }
+        } else {
+            return i;
+        }
+    }
+
+    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 {
-        int i = inputStream.read();
+    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 == '.') {
            stringBuilder.append((char) i);
-            i = inputStream.read();
+            i = getChar(inputStream);
        }
        return new CharChecker(i);
    }

-    private static CharChecker skipOneLong(InputStream inputStream) throws IOException {
-        int i = inputStream.read();
+    private static CharChecker skipOneLong(InputStream inputStream) throws IOException, FormatException {
+        int i = getChar(inputStream);
        while (i >= '0' && i <= '9') {
-            i = inputStream.read();
+            i = getChar(inputStream);
        }
        return new CharChecker(i);
    }

-    private static CharChecker parseMetricName(InputStream inputStream, StringBuilder stringBuilder) throws IOException {
-        int i = inputStream.read();
+    private static CharChecker parseMetricName(InputStream inputStream, StringBuilder stringBuilder) throws IOException, FormatException {
+        int i = getChar(inputStream);
        while ((i >= 'a' && i <= 'z') || (i >= 'A' && i <= 'Z') || (i >= '0' && i <= '9') || i == '_' || i == ':') {
            stringBuilder.append((char) i);
-            i = inputStream.read();
+            i = getChar(inputStream);
        }
        return new CharChecker(i);
    }

-    private static CharChecker parseLabelName(InputStream inputStream, StringBuilder stringBuilder) throws IOException {
-        int i = inputStream.read();
+    private static CharChecker parseLabelName(InputStream inputStream, StringBuilder stringBuilder) throws IOException, FormatException {
+        int i = getChar(inputStream);
        while ((i >= 'a' && i <= 'z') || (i >= 'A' && i <= 'Z') || (i >= '0' && i <= '9') || i == '_') {
            stringBuilder.append((char) i);
-            i = inputStream.read();
+            i = getChar(inputStream);
        }
        return new CharChecker(i);
    }

    private static CharChecker parseLabelValue(InputStream inputStream, StringBuilder stringBuilder) throws IOException, FormatException {
-        int i = inputStream.read();
+        int i = getChar(inputStream);
        while (i != '"' && i != -1) {
            if (i == '\\') {
-                i = inputStream.read();
+                i = getChar(inputStream);
                switch (i) {
                    case 'n':
                        stringBuilder.append('\n');
@@ -181,27 +211,26 @@ public class OnlineParser {
                    default:
                        throw new FormatException();
                }
-            }
-            else {
+            } else {
                stringBuilder.append((char) i);
            }
-            i = inputStream.read();
+            i = getChar(inputStream);
        }
        return new CharChecker(i);
    }

-    private static CharChecker skipSpaces(InputStream inputStream) throws IOException {
-        int i = inputStream.read();
+    private static CharChecker skipSpaces(InputStream inputStream) throws IOException, FormatException {
+        int i = getChar(inputStream);
        while (i == ' ') {
-            i = inputStream.read();
+            i = getChar(inputStream);
        }
        return new CharChecker(i);
    }

-    private static CharChecker skipToLineEnd(InputStream inputStream) throws IOException {
-        int i = inputStream.read();
+    private static CharChecker skipToLineEnd(InputStream inputStream) throws IOException, FormatException {
+        int i = getChar(inputStream);
        while (i != '\n' && i != -1) {
-            i = inputStream.read();
+            i = getChar(inputStream);
        }
        return new CharChecker(i);
    }
@@ -267,18 +296,18 @@ public class OnlineParser {
            metricFamily.setMetricList(new ArrayList<>());
            metricFamily.setName(metricName);
            metricFamilyMap.put(metricName, metricFamily);
-        }
-        else {
+        } else {
            metricFamily = metricFamilyMap.get(metricName);
        }

        if (i == ' ') {
            i = skipSpaces(inputStream).getInt();
        }
+
+        List<MetricFamily.Label> labelList = new LinkedList<>();
+        metric.setLabels(labelList);
        if (i == '{') {
-            List<MetricFamily.Label> labelList = new LinkedList<>();
            parseLabels(inputStream, stringBuilder, labelList);
-            metric.setLabels(labelList);
            i = skipSpaces(inputStream).getInt();
        }

@@ -307,8 +336,8 @@ public class OnlineParser {

    public static Map<String, MetricFamily> parseMetrics(InputStream inputStream) throws IOException {
        Map<String, MetricFamily> metricFamilyMap = new ConcurrentHashMap<>(10);
-        int i = inputStream.read();
        try {
+            int i = getChar(inputStream);
            while (i != -1) {
                if (i == '#' || i == '\n') {
                    skipToLineEnd(inputStream).maybeEol().maybeEof().noElse();
@@ -317,7 +346,7 @@ public class OnlineParser {
                    stringBuilder.append((char) i);
                    parseMetric(inputStream, metricFamilyMap, stringBuilder);
                }
-                i = inputStream.read();
+                i = getChar(inputStream);
            }
        } catch (FormatException e) {
            log.error("prometheus parser failed because of wrong input format. {}", e.getMessage());
@@ -28,6 +28,7 @@ import org.springframework.context.annotation.ImportRuntimeHints;
 import org.springframework.data.jpa.repository.config.EnableJpaAuditing;
 import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.annotation.EnableScheduling;

 /**
 * start up class.
@@ -41,6 +42,7 @@ import org.springframework.scheduling.annotation.EnableAsync;
@ConfigurationPropertiesScan(basePackages = {"org.apache.hertzbeat"})
@ImportRuntimeHints(HertzbeatRuntimeHintsRegistrar.class)
@EnableAsync
+@EnableScheduling
 public class Manager {
    public static void main(String[] args) {
        SpringApplication.run(Manager.class, args);
@@ -531,7 +531,7 @@ public class MonitorServiceImpl implements MonitorService {
            MonitorDto monitorDto = new MonitorDto();
            List<Param> params = paramDao.findParamsByMonitorId(id);
            monitorDto.setParams(params);
-            if (DispatchConstants.PROTOCOL_PROMETHEUS.equalsIgnoreCase(monitor.getApp())) {
+            if (DispatchConstants.PROTOCOL_PROMETHEUS.equalsIgnoreCase(monitor.getApp()) || monitor.getType() == CommonConstants.MONITOR_TYPE_PUSH_AUTO_CREATE) {
                List<CollectRep.MetricsData> metricsDataList = warehouseService.queryMonitorMetricsData(id);
                List<String> metrics = metricsDataList.stream().map(CollectRep.MetricsData::getMetrics).collect(Collectors.toList());
                monitorDto.setMetrics(metrics);
@@ -0,0 +1,42 @@
+/*
+ * 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.push.config;
+
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletRequestWrapper;
+import lombok.Getter;
+
+/**
+ * push error request wrapper
+ */
+@Getter
+public class PushErrorRequestWrapper extends HttpServletRequestWrapper {
+
+
+    private final String job;
+
+    private final String instance;
+    
+    public PushErrorRequestWrapper(HttpServletRequest request,  String job, String instance) {
+        super(request);
+        this.job = job;
+        this.instance = instance;
+    }
+}
@@ -0,0 +1,47 @@
+/*
+ * 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.push.config;
+
+import org.apache.hertzbeat.push.service.PushGatewayService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.web.servlet.FilterRegistrationBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ *
+ */
+@Configuration
+public class PushFilterConfig {
+
+    @Autowired
+    private PushGatewayService pushGatewayService;
+
+    private static final String URI_PREFIX = "/api/push/prometheus/*";
+
+    @Bean
+    public FilterRegistrationBean<PushPrometheusStreamReadingFilter> contentTypeFilter() {
+        FilterRegistrationBean<PushPrometheusStreamReadingFilter> registrationBean = new FilterRegistrationBean<>();
+        registrationBean.setFilter(new PushPrometheusStreamReadingFilter(pushGatewayService));
+        registrationBean.addUrlPatterns(URI_PREFIX);
+        registrationBean.setOrder(Integer.MIN_VALUE);
+        return registrationBean;
+    }
+}
@@ -0,0 +1,82 @@
+/*
+ * 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.push.config;
+
+import jakarta.servlet.Filter;
+import jakarta.servlet.FilterChain;
+import jakarta.servlet.FilterConfig;
+import jakarta.servlet.ServletException;
+import jakarta.servlet.ServletRequest;
+import jakarta.servlet.ServletResponse;
+import jakarta.servlet.http.HttpServletRequest;
+import java.io.IOException;
+import java.util.regex.Matcher;
+
+import java.util.regex.Pattern;
+import org.apache.hertzbeat.push.service.PushGatewayService;
+
+
+/**
+ * todo 
+ */
+public class PushPrometheusStreamReadingFilter implements Filter {
+    
+    private final PushGatewayService pushGatewayService;
+    
+    private final Pattern pathPattern = Pattern.compile("^/api/push/prometheus/job/([a-zA-Z0-9_]*)(?:/instance/([a-zA-Z0-9_]*))?$");
+    
+    public PushPrometheusStreamReadingFilter(PushGatewayService pushGatewayService) {
+        this.pushGatewayService = pushGatewayService;
+    }
+    
+    @Override
+    public void init(FilterConfig filterConfig) throws ServletException {}
+
+    @Override
+    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
+            throws IOException, ServletException {
+        if (request instanceof HttpServletRequest httpRequest) {
+            String uri = httpRequest.getRequestURI();
+            Matcher matcher = pathPattern.matcher(uri);
+            String job = null;
+            String instance = null;
+            if (matcher.matches()) {
+                job = matcher.group(1);
+                instance = matcher.group(2);
+                boolean flag = pushGatewayService.pushPrometheusMetrics(request.getInputStream(), job, instance);
+                if (flag) {
+                    PushSuccessRequestWrapper successRequestWrapper = new PushSuccessRequestWrapper(httpRequest, job, instance);
+                    chain.doFilter(successRequestWrapper, response);
+                } else {
+                    PushErrorRequestWrapper errorRequestWrapper = new PushErrorRequestWrapper(httpRequest, job, instance);
+                    chain.doFilter(errorRequestWrapper, response);
+                }
+            } else {
+                chain.doFilter(request, response);
+            }
+
+        } else {
+            chain.doFilter(request, response);
+        }
+    }
+
+    @Override
+    public void destroy() {}
+}
@@ -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.push.config;
+
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletRequestWrapper;
+import lombok.Getter;
+
+/**
+ * push success request wrapper
+ */
+@Getter
+public class PushSuccessRequestWrapper extends HttpServletRequestWrapper {
+
+    private final String job;
+    
+    private final String instance;
+
+    public PushSuccessRequestWrapper(HttpServletRequest request, String job, String instance) {
+        super(request);
+        this.job = job;
+        this.instance = instance;
+    }
+}
@@ -1,62 +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.push.controller;
-
-import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
-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.push.PushMetricsDto;
-import org.apache.hertzbeat.push.service.PushService;
-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.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestParam;
-import org.springframework.web.bind.annotation.RestController;
-
-/**
- * push controller
- */
-@Tag(name = "Metrics Push API")
-@RestController
-@RequestMapping(value = "/api/push", produces = {APPLICATION_JSON_VALUE})
-public class PushController {
-
-    @Autowired
-    private PushService pushService;
-
-    @PostMapping
-    @Operation(summary = "Push metric data to hertzbeat", description = "Push metric data to hertzbeat")
-    public ResponseEntity<Message<Void>> pushMetrics(@RequestBody PushMetricsDto pushMetricsDto) {
-        pushService.pushMetricsData(pushMetricsDto);
-        return ResponseEntity.ok(Message.success("Push success"));
-    }
-
-    @GetMapping()
-    @Operation(summary = "Get metric data for hertzbeat", description = "Get metric data for hertzbeat")
-    public ResponseEntity<Message<PushMetricsDto>> getMetrics(
-            @Parameter(description = "Monitor ID", example = "6565463543") @RequestParam("id") final Long id,
-            @Parameter(description = "Last pull time", example = "6565463543") @RequestParam("time") final Long time) {
-        PushMetricsDto pushMetricsDto = pushService.getPushMetricData(id, time);
-        return ResponseEntity.ok(Message.success(pushMetricsDto));
-    }
-}
@@ -22,11 +22,9 @@ package org.apache.hertzbeat.push.controller;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.servlet.http.HttpServletRequest;
-import java.io.IOException;
-import java.io.InputStream;
 import org.apache.hertzbeat.common.entity.dto.Message;
-import org.apache.hertzbeat.push.service.PushGatewayService;
-import org.springframework.beans.factory.annotation.Autowired;
+import org.apache.hertzbeat.push.config.PushErrorRequestWrapper;
+import org.apache.hertzbeat.push.config.PushSuccessRequestWrapper;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -37,22 +35,23 @@ import org.springframework.web.bind.annotation.RestController;
 */
@Tag(name = "Metrics Push Gateway API")
@RestController
-@RequestMapping(value = "/api/push/pushgateway")
-public class PushGatewayController {
-
-    @Autowired
-    private PushGatewayService pushGatewayService;
+@RequestMapping(value = "/api/push/prometheus/**")
+public class PushPrometheusController {

    @PostMapping()
-    @Operation(summary = "Push metric data to hertzbeat pushgateway", description = "Push metric data to hertzbeat pushgateway")
-    public ResponseEntity<Message<Void>> pushMetrics(HttpServletRequest request) throws IOException {
-        InputStream inputStream = request.getInputStream();
-        boolean result = pushGatewayService.pushMetricsData(inputStream);
-        if (result) {
-            return ResponseEntity.ok(Message.success("Push success"));
+    @Operation(summary = "Prometheus push gateway", description = "Push prometheus metric data to hertzbeat")
+    public ResponseEntity<Message<Void>> pushMetrics(HttpServletRequest request) {
+        if (request instanceof PushErrorRequestWrapper error) {
+            return ResponseEntity.badRequest().body(Message.success(String.format("Push failed, job: %s, instance: %s", 
+                            error.getJob(), error.getInstance())));
+        }
+        else if (request instanceof PushSuccessRequestWrapper success) {
+            return ResponseEntity.ok(Message.success(String.format("Push success, job: %s, instance: %s",
+                    success.getJob(), success.getInstance())));
        }
        else {
-            return ResponseEntity.ok(Message.success("Push failed"));
+            return ResponseEntity.badRequest()
+                    .body(Message.success(String.format("Request  %s not matched.", request.getRequestURI())));
        }
    }

@@ -17,6 +17,7 @@

 package org.apache.hertzbeat.push.dao;

+import java.util.List;
 import org.apache.hertzbeat.common.entity.manager.Monitor;
 import org.springframework.data.jpa.repository.JpaRepository;

@@ -24,4 +25,11 @@ import org.springframework.data.jpa.repository.JpaRepository;
 * push monitor dao
 */
 public interface PushMonitorDao extends JpaRepository<Monitor, Long> {
+
+    /**
+     * Find all monitoring entities by type
+     * @param type Monitoring type
+     * @return Monitoring entity list
+     */
+    List<Monitor> findMonitorsByType(byte type);
 }
@@ -19,17 +19,24 @@

 package org.apache.hertzbeat.push.service;

-import java.io.IOException;
 import java.io.InputStream;
 import org.springframework.stereotype.Service;

 /**
- * push gateway metrics
+ * push gateway service
 */

@Service
 public interface PushGatewayService {

-    boolean pushMetricsData(InputStream inputStream) throws IOException;
+
+    /**
+     * push prometheus metrics data
+     * @param inputStream input stream
+     * @param job job name, maybe null
+     * @param instance instance name, maybe null
+     * @return push success or not
+     */
+    boolean pushPrometheusMetrics(InputStream inputStream, String job, String instance);

 }
@@ -1,31 +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.push.service;
-
-import org.apache.hertzbeat.common.entity.push.PushMetricsDto;
-import org.springframework.stereotype.Service;
-
-/**
- * push metrics
- */
-@Service
-public interface PushService {
-    void pushMetricsData(PushMetricsDto pushMetricsData);
-
-    PushMetricsDto getPushMetricData(Long monitorId, Long time);
-}
@@ -19,9 +19,23 @@

 package org.apache.hertzbeat.push.service.impl;

-import java.io.IOException;
 import java.io.InputStream;
+import java.time.Instant;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
 import lombok.extern.slf4j.Slf4j;
+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;
 import org.springframework.stereotype.Service;

@@ -32,9 +46,88 @@ import org.springframework.stereotype.Service;
@Slf4j
@Service
 public class PushGatewayServiceImpl implements PushGatewayService {
+    
+    private final CommonDataQueue commonDataQueue;
+    
+    private final PushMonitorDao pushMonitorDao;
+    
+    private final Map<String, Long> jobInstanceMap;
+    
+    public PushGatewayServiceImpl(CommonDataQueue commonDataQueue, PushMonitorDao pushMonitorDao) {
+        this.commonDataQueue = commonDataQueue;
+        this.pushMonitorDao = pushMonitorDao;
+        jobInstanceMap = new ConcurrentHashMap<>();
+        pushMonitorDao.findMonitorsByType((byte) 1).forEach(monitor -> 
+                jobInstanceMap.put(monitor.getApp() + "_" + monitor.getName(), monitor.getId()));
+    }

    @Override
-    public boolean pushMetricsData(InputStream inputStream) throws IOException {
-        return true;
+    public boolean pushPrometheusMetrics(InputStream inputStream, String job, String instance) {
+        try {
+            long curTime = Instant.now().toEpochMilli();
+            Map<String, MetricFamily> metricFamilyMap = OnlineParser.parseMetrics(inputStream);
+            if (metricFamilyMap == null) {
+                log.error("parse prometheus metrics is null, job: {}, instance: {}", job, instance);
+                return false;
+            }
+            long id = 0L;
+            if (job != null && instance != null) {
+                // auto create monitor when job and instance not null
+                // job is app, instance is the name
+                id = jobInstanceMap.computeIfAbsent(job + "_" + instance, key -> {
+                    log.info("auto create monitor by prometheus push, job: {}, instance: {}", job, instance);
+                    long monitorId = SnowFlakeIdGenerator.generateId();
+                    Monitor monitor = Monitor.builder()
+                            .id(monitorId)
+                            .app(job)
+                            .name(instance)
+                            .host(instance)
+                            .type((byte) 1)
+                            .status(CommonConstants.MONITOR_UP_CODE)
+                            .build();
+                    this.pushMonitorDao.save(monitor);
+                    return monitorId;
+                });
+            }
+            for (Map.Entry<String, MetricFamily> entry : metricFamilyMap.entrySet()) {
+                CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder();
+                builder.setId(id);
+                builder.setApp(job);
+                builder.setTime(curTime);
+                String metricsName = entry.getKey();
+                builder.setMetrics(metricsName);
+                MetricFamily metricFamily = entry.getValue();
+                if (!metricFamily.getMetricList().isEmpty()) {
+                    List<String> metricsFields = new LinkedList<>();
+                    for (int index = 0; index < metricFamily.getMetricList().size(); index++) {
+                        MetricFamily.Metric metric = metricFamily.getMetricList().get(index);
+                        if (index == 0) {
+                            metric.getLabels().forEach(label -> {
+                                metricsFields.add(label.getName());
+                                builder.addField(CollectRep.Field.newBuilder().setName(label.getName())
+                                        .setType(CommonConstants.TYPE_STRING).setLabel(true).build());
+                            });
+                            builder.addField(CollectRep.Field.newBuilder().setName("value")
+                                    .setType(CommonConstants.TYPE_NUMBER).setLabel(false).build());
+                        }
+                        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 field : metricsFields) {
+                            String fieldValue = labelMap.get(field);
+                            valueRowBuilder.addColumn(fieldValue == null ? CommonConstants.NULL_VALUE : fieldValue);
+                        }
+                        valueRowBuilder.addColumn(String.valueOf(metric.getValue()));
+                        builder.addValueRow(valueRowBuilder.build());
+                    }
+                    commonDataQueue.sendMetricsData(builder.build());
+                }
+            }
+            return true;
+        } catch (Exception e) {
+            log.error("push prometheus metrics error", e);
+            return false;
+        }
    }
 }
@@ -1,143 +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.push.service.impl;
-
-
-import com.fasterxml.jackson.core.type.TypeReference;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Timer;
-import java.util.TimerTask;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.hertzbeat.common.entity.manager.Monitor;
-import org.apache.hertzbeat.common.entity.push.PushMetrics;
-import org.apache.hertzbeat.common.entity.push.PushMetricsDto;
-import org.apache.hertzbeat.common.util.JsonUtil;
-import org.apache.hertzbeat.push.dao.PushMetricsDao;
-import org.apache.hertzbeat.push.dao.PushMonitorDao;
-import org.apache.hertzbeat.push.service.PushService;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-/**
- * push service impl
- */
-@Slf4j
-@Service
-public class PushServiceImpl implements PushService {
-
-    @Autowired
-    private PushMonitorDao monitorDao;
-
-    @Autowired
-    private PushMetricsDao metricsDao;
-
-    private final Map<Long, Long> monitorIdCache; // key: monitorId, value: time stamp of last query
-
-    private static final long cacheTimeout = 5000L; // ms
-
-    private final Map<Long, PushMetricsDto.Metrics> lastPushMetrics;
-
-    private static final long deleteMetricsPeriod = 1000 * 60 * 60 * 12L;
-
-    private static final long deleteBeforeTime = deleteMetricsPeriod / 2;
-
-    public PushServiceImpl(){
-        monitorIdCache = new HashMap<>();
-        lastPushMetrics = new HashMap<>();
-
-        new Timer().schedule(new TimerTask() {
-            @Override
-            public void run() {
-                try {
-                    deletePeriodically();
-                } catch (Exception e) {
-                    log.error("periodical deletion failed. {}", e.getMessage());
-                }
-            }
-        }, 1000, deleteMetricsPeriod);
-    }
-
-    public void deletePeriodically(){
-        metricsDao.deleteAllByTimeBefore(System.currentTimeMillis() - deleteBeforeTime);
-    }
-
-    @Override
-    public void pushMetricsData(PushMetricsDto pushMetricsDto) throws RuntimeException {
-        List<PushMetrics> pushMetricsList = new ArrayList<>();
-        long curTime = System.currentTimeMillis();
-        for (PushMetricsDto.Metrics metrics : pushMetricsDto.getMetricsList()) {
-            long monitorId = metrics.getMonitorId();
-            metrics.setTime(curTime);
-
-            if (!monitorIdCache.containsKey(monitorId) || (monitorIdCache.containsKey(monitorId) && curTime > monitorIdCache.get(monitorId) + cacheTimeout)) {
-                Optional<Monitor> queryOption = monitorDao.findById(monitorId);
-                if (queryOption.isEmpty()) {
-                    monitorIdCache.remove(monitorId);
-                    continue;
-                }
-                monitorIdCache.put(monitorId, curTime);
-            }
-
-            PushMetrics pushMetrics = PushMetrics.builder()
-                    .monitorId(metrics.getMonitorId())
-                    .time(curTime)
-                    .metrics(JsonUtil.toJson(metrics.getMetrics())).build();
-            lastPushMetrics.put(monitorId, metrics);
-            pushMetricsList.add(pushMetrics);
-        }
-
-        metricsDao.saveAll(pushMetricsList);
-
-    }
-
-    @Override
-    public PushMetricsDto getPushMetricData(final Long monitorId, final Long time) {
-        PushMetricsDto.Metrics metrics;
-        PushMetricsDto pushMetricsDto = new PushMetricsDto();
-        if (lastPushMetrics.containsKey(monitorId)) {
-            metrics = lastPushMetrics.get(monitorId);
-        }
-        else {
-            try {
-                PushMetrics pushMetrics = metricsDao.findFirstByMonitorIdOrderByTimeDesc(monitorId);
-                if (pushMetrics == null || pushMetrics.getMetrics() == null) {
-                    return pushMetricsDto;
-                }
-                List<Map<String, String>> jsonMap = JsonUtil.fromJson(pushMetrics.getMetrics(), new TypeReference<>() {
-                });
-                metrics = PushMetricsDto.Metrics.builder().monitorId(monitorId).metrics(jsonMap).time(pushMetrics.getTime()).build();
-                lastPushMetrics.put(monitorId, metrics);
-            }
-            catch (Exception e) {
-                log.error("no metrics found, monitor id: {}, {}", monitorId, e.getMessage(), e);
-                return pushMetricsDto;
-            }
-        }
-        if (time > metrics.getTime()) {
-            // return void because time param is invalid
-            return pushMetricsDto;
-        }
-        pushMetricsDto.getMetricsList().add(metrics);
-        return pushMetricsDto;
-    }
-
-}
@@ -1,93 +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.push.controller;
-
-import static org.mockito.Mockito.when;
-import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
-import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
-import static org.springframework.test.web.servlet.setup.MockMvcBuilders.standaloneSetup;
-import org.apache.hertzbeat.common.constants.CommonConstants;
-import org.apache.hertzbeat.common.entity.push.PushMetricsDto;
-import org.apache.hertzbeat.common.util.JsonUtil;
-import org.apache.hertzbeat.push.service.PushService;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-import org.springframework.http.MediaType;
-import org.springframework.test.web.servlet.MockMvc;
-import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
-
-/**
- * test case for {@link PushController}
- */
-
-@ExtendWith(MockitoExtension.class)
-class PushControllerTest {
-
-    private MockMvc mockMvc;
-
-    @Mock
-    private PushService pushService;
-
-    @InjectMocks
-    private PushController pushController;
-
-    private PushMetricsDto mockPushMetricsDto;
-
-    @BeforeEach
-    void setUp() {
-
-        this.mockMvc = standaloneSetup(this.pushController).build();
-
-        mockPushMetricsDto = PushMetricsDto.builder().build();
-    }
-
-    @Test
-    void testPushMetrics() throws Exception {
-
-        this.mockMvc.perform(MockMvcRequestBuilders.post("/api/push")
-                        .contentType(MediaType.APPLICATION_JSON)
-                        .content(JsonUtil.toJson(mockPushMetricsDto)))
-                .andExpect(status().isOk())
-                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
-                .andReturn();
-    }
-
-    @Test
-    void testGetMetrics() throws Exception {
-
-        Long id = 6565463543L;
-        Long time = 6565463543L;
-
-        when(pushService.getPushMetricData(id, time)).thenReturn(mockPushMetricsDto);
-
-        this.mockMvc.perform(MockMvcRequestBuilders.get("/api/push")
-                        .contentType(MediaType.APPLICATION_JSON)
-                        .param("id", id.toString())
-                        .param("time", time.toString()))
-                .andExpect(status().isOk())
-                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
-                .andReturn();
-    }
-
-}
@@ -22,12 +22,11 @@ package org.apache.hertzbeat.push.controller;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.when;
 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
-import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 import java.io.InputStream;
-import org.apache.hertzbeat.common.constants.CommonConstants;
 import org.apache.hertzbeat.push.service.PushGatewayService;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.InjectMocks;
@@ -38,11 +37,12 @@ import org.springframework.test.web.servlet.MockMvc;
 import org.springframework.test.web.servlet.setup.MockMvcBuilders;

 /**
- * test case for {@link PushGatewayController}
+ * test case for {@link PushPrometheusController}
 */

+@Disabled
@ExtendWith(MockitoExtension.class)
-class PushGatewayControllerTest {
+class PushPrometheusControllerTest {

    private MockMvc mockMvc;

@@ -50,7 +50,7 @@ class PushGatewayControllerTest {
    private PushGatewayService pushGatewayService;

    @InjectMocks
-    private PushGatewayController gatewayController;
+    private PushPrometheusController gatewayController;

    @BeforeEach
    void setUp() {
@@ -63,14 +63,12 @@ class PushGatewayControllerTest {

        String mockData = "some metric data";

-        when(pushGatewayService.pushMetricsData(any(InputStream.class))).thenReturn(true);
+        when(pushGatewayService.pushPrometheusMetrics(any(InputStream.class), any(), any())).thenReturn(true);

        mockMvc.perform(post("/api/push/pushgateway")
                        .contentType(MediaType.APPLICATION_JSON)
                        .content(mockData))
-                .andExpect(status().isOk())
-                .andExpect(jsonPath("$.code").value((int) CommonConstants.SUCCESS_CODE))
-                .andExpect(jsonPath("$.msg").value("Push success"));
+                .andExpect(status().isOk());
    }

    @Test
@@ -78,13 +76,12 @@ class PushGatewayControllerTest {

        String mockData = "some metric data";

-        when(pushGatewayService.pushMetricsData(any(InputStream.class))).thenReturn(false);
+        when(pushGatewayService.pushPrometheusMetrics(any(InputStream.class), any(), any())).thenReturn(false);

        mockMvc.perform(post("/api/push/pushgateway")
                        .contentType(MediaType.APPLICATION_JSON)
                        .content(mockData))
-                .andExpect(status().isOk())
-                .andExpect(jsonPath("$.msg").value("Push failed"));
+                .andExpect(status().isOk());
    }

 }
@@ -1,77 +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.push.dao;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-import org.apache.hertzbeat.common.entity.push.PushMetrics;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-/**
- * test case for {@link PushMetricsDao}
- */
-
-@ExtendWith(MockitoExtension.class)
-public class PushMetricsDaoTest {
-    @Mock
-    private PushMetricsDao pushMetricsDao;
-
-    @InjectMocks
-    private PushMetricsDaoTest pushMetricsDaoTest;
-
-    @BeforeEach
-    void setUp() {
-        MockitoAnnotations.openMocks(this);
-    }
-
-    @Test
-    void shallFindFirstByMonitorIdOrderByTimeDesc() {
-
-        PushMetrics expectedMetrics = new PushMetrics();
-        expectedMetrics.setMonitorId(1L);
-        expectedMetrics.setTime(System.currentTimeMillis());
-
-        when(pushMetricsDao.findFirstByMonitorIdOrderByTimeDesc(1L)).thenReturn(expectedMetrics);
-
-        PushMetrics actualMetrics = pushMetricsDao.findFirstByMonitorIdOrderByTimeDesc(1L);
-
-        assertEquals(expectedMetrics, actualMetrics);
-        verify(pushMetricsDao, times(1)).findFirstByMonitorIdOrderByTimeDesc(1L);
-    }
-
-    @Test
-    void shallDeleteAllByTimeBefore() {
-
-        doNothing().when(pushMetricsDao).deleteAllByTimeBefore(anyLong());
-
-        pushMetricsDao.deleteAllByTimeBefore(1000L);
-
-        verify(pushMetricsDao, times(1)).deleteAllByTimeBefore(1000L);
-    }
-}
@@ -1,132 +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.push.service;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-import org.apache.hertzbeat.common.entity.manager.Monitor;
-import org.apache.hertzbeat.common.entity.push.PushMetrics;
-import org.apache.hertzbeat.common.entity.push.PushMetricsDto;
-import org.apache.hertzbeat.push.dao.PushMetricsDao;
-import org.apache.hertzbeat.push.dao.PushMonitorDao;
-import org.apache.hertzbeat.push.service.impl.PushServiceImpl;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-import org.springframework.test.util.ReflectionTestUtils;
-
-/**
- * test case for {@link PushServiceImpl}
- */
-
-@ExtendWith(MockitoExtension.class)
-class PushServiceTest {
-
-    @Mock
-    private PushMonitorDao monitorDao;
-
-    @Mock
-    private PushMetricsDao metricsDao;
-
-    @InjectMocks
-    private PushServiceImpl pushService;
-
-    @BeforeEach
-    void setUp() {
-
-        pushService = new PushServiceImpl();
-
-        ReflectionTestUtils.setField(pushService, "monitorDao", monitorDao);
-        ReflectionTestUtils.setField(pushService, "metricsDao", metricsDao);
-    }
-
-    @Test
-    void testPushMetricsData() {
-
-        PushMetricsDto pushMetricsDto = new PushMetricsDto();
-        List<PushMetricsDto.Metrics> metricsList = new ArrayList<>();
-        PushMetricsDto.Metrics metrics = new PushMetricsDto.Metrics();
-        metrics.setMonitorId(1L);
-        metricsList.add(metrics);
-        pushMetricsDto.setMetricsList(metricsList);
-
-        when(monitorDao.findById(anyLong())).thenReturn(Optional.of(new Monitor()));
-
-        pushService.pushMetricsData(pushMetricsDto);
-
-        verify(metricsDao, times(1)).saveAll(any());
-    }
-
-    @Test
-    void testGetPushMetricData() {
-
-        Long monitorId = 1L;
-        Long time = System.currentTimeMillis();
-        PushMetrics pushMetrics = PushMetrics.builder()
-                .monitorId(monitorId)
-                .time(time)
-                .metrics("[{\"key\":\"value\"}]")
-                .build();
-
-        when(metricsDao.findFirstByMonitorIdOrderByTimeDesc(monitorId)).thenReturn(pushMetrics);
-
-        PushMetricsDto result = pushService.getPushMetricData(monitorId, time);
-
-        assertEquals(1, result.getMetricsList().size());
-        assertEquals(monitorId, result.getMetricsList().get(0).getMonitorId());
-    }
-
-    @Test
-    void testGetPushMetricDataTimeInvalid() {
-
-        Long monitorId = 1L;
-        Long time = System.currentTimeMillis() + 10000;
-        PushMetrics pushMetrics = PushMetrics.builder()
-                .monitorId(monitorId)
-                .time(System.currentTimeMillis())
-                .metrics("[{\"key\":\"value\"}]")
-                .build();
-
-        when(metricsDao.findFirstByMonitorIdOrderByTimeDesc(monitorId)).thenReturn(pushMetrics);
-
-        PushMetricsDto result = pushService.getPushMetricData(monitorId, time);
-
-        assertTrue(result.getMetricsList().isEmpty());
-    }
-
-    @Test
-    void testDeletePeriodically() {
-
-        pushService.deletePeriodically();
-        verify(metricsDao, times(1)).deleteAllByTimeBefore(anyLong());
-    }
-
-}
@@ -56,4 +56,8 @@ public interface WarehouseConstants {
        String MEMORY = "memory";
    }

+    String PROMQL = "promql";
+
+    String SQL = "sql";
+
 }
@@ -0,0 +1,78 @@
+/*
+ * 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)));
+    }
+}
@@ -18,29 +18,12 @@
 package org.apache.hertzbeat.warehouse.db;


-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 lombok.extern.slf4j.Slf4j;
-import org.apache.hertzbeat.common.constants.NetworkConstants;
-import org.apache.hertzbeat.common.constants.SignConstants;
-import org.apache.hertzbeat.common.util.Base64Util;
 import org.apache.hertzbeat.warehouse.store.history.greptime.GreptimeProperties;
-import org.apache.hertzbeat.warehouse.store.history.vm.PromQlQueryContent;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.http.HttpEntity;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpMethod;
-import org.springframework.http.MediaType;
-import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Component;
-import org.springframework.util.StringUtils;
 import org.springframework.web.client.RestTemplate;
-import org.springframework.web.util.UriComponentsBuilder;
+

 /**
 * query executor for victor metrics
@@ -48,71 +31,15 @@ import org.springframework.web.util.UriComponentsBuilder;
@Component
@ConditionalOnProperty(prefix = "warehouse.store.greptime", name = "enabled", havingValue = "true")
@Slf4j
-public class GreptimePromqlQueryExecutor implements QueryExecutor {
+public class GreptimePromqlQueryExecutor extends PromqlQueryExecutor {

    private static final String QUERY_PATH = "/v1/prometheus/api/v1/query";

    private final GreptimeProperties greptimeProperties;

-    private final RestTemplate restTemplate;
-
    public GreptimePromqlQueryExecutor(GreptimeProperties greptimeProperties, RestTemplate restTemplate) {
+        super(restTemplate, new HttpPromqlProperties(greptimeProperties.httpEndpoint() + QUERY_PATH,
+                greptimeProperties.username(), greptimeProperties.password()));
        this.greptimeProperties = greptimeProperties;
-        this.restTemplate = restTemplate;
-    }
-
-    @Override
-    public List<Map<String, Object>> execute(String query) {
-        // http run the promql query
-        List<Map<String, Object>> results = new LinkedList<>();
-        try {
-            HttpHeaders headers = new HttpHeaders();
-            headers.setContentType(MediaType.APPLICATION_JSON);
-            headers.setAccept(List.of(MediaType.APPLICATION_JSON));
-            if (StringUtils.hasText(greptimeProperties.username())
-                    && StringUtils.hasText(greptimeProperties.password())) {
-                String authStr = greptimeProperties.username() + ":" + greptimeProperties.password();
-                String encodedAuth = Base64Util.encode(authStr);
-                headers.add(HttpHeaders.AUTHORIZATION,  NetworkConstants.BASIC + SignConstants.BLANK + encodedAuth);
-            }
-            HttpEntity<Void> httpEntity = new HttpEntity<>(headers);
-            URI uri = UriComponentsBuilder.fromHttpUrl(greptimeProperties.httpEndpoint() + QUERY_PATH)
-                    .queryParam("query", URLEncoder.encode(query, StandardCharsets.UTF_8))
-                    .build(true).toUri();
-            ResponseEntity<PromQlQueryContent> responseEntity = restTemplate.exchange(uri,
-                    HttpMethod.GET, httpEntity, PromQlQueryContent.class);
-            if (responseEntity.getStatusCode().is2xxSuccessful()) {
-                if (responseEntity.getBody() != null && responseEntity.getBody().getData() != null
-                        && responseEntity.getBody().getData().getResult() != null) {
-                    List<PromQlQueryContent.ContentData.Content> contents = responseEntity.getBody().getData().getResult();
-                    for (PromQlQueryContent.ContentData.Content content : contents) {
-                        Map<String, String> labels = content.getMetric();
-                        Map<String, Object> queryResult = new HashMap<>(8);
-                        queryResult.putAll(labels);
-                        if (content.getValue() != null && content.getValue().length == 2) {
-                            queryResult.put("__timestamp__", content.getValue()[0]);
-                            queryResult.put("__value__", content.getValue()[1]);
-                        } else if (content.getValues() != null && !content.getValues().isEmpty()) {
-                            List<Object> values = new LinkedList<>();
-                            for (Object[] valueArr : content.getValues()) {
-                                values.add(valueArr[1]);
-                            }
-                            queryResult.put("__value__", values);
-                        }
-                        results.add(queryResult);
-                    }
-                }
-            } else {
-                log.error("query metrics data from greptime failed. {}", responseEntity);
-            }
-        } catch (Exception e) {
-            log.error(e.toString(), e);
-        }
-        return results;
-    }
-
-    @Override
-    public boolean support(String datasource) {
-        return "promql".equals(datasource);
    }
 }
@@ -0,0 +1,166 @@
+/*
+ * 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.db;
+
+import lombok.extern.slf4j.Slf4j;
+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.util.Base64Util;
+import org.apache.hertzbeat.warehouse.store.history.vm.PromQlQueryContent;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.MediaType;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.StringUtils;
+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
+ */
+@Slf4j
+public abstract class PromqlQueryExecutor implements QueryExecutor {
+
+    private static final String supportQueryLanguage = PROMQL;
+    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 final RestTemplate restTemplate;
+
+    private final HttpPromqlProperties httpPromqlProperties;
+
+    PromqlQueryExecutor(RestTemplate restTemplate, HttpPromqlProperties httpPromqlProperties) {
+        this.restTemplate = restTemplate;
+        this.httpPromqlProperties = httpPromqlProperties;
+    }
+
+    /**
+     * record class for promql http connection
+     */
+    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
+        List<Map<String, Object>> results = new LinkedList<>();
+        try {
+            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 = Base64Util.encode(authStr);
+                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());
+            }
+            URI uri = uriComponentsBuilder.build(true).toUri();
+            ResponseEntity<PromQlQueryContent> responseEntity = restTemplate.exchange(uri,
+                    HttpMethod.GET, httpEntity, PromQlQueryContent.class);
+            if (responseEntity.getStatusCode().is2xxSuccessful()) {
+                if (responseEntity.getBody() != null && responseEntity.getBody().getData() != null
+                        && responseEntity.getBody().getData().getResult() != null) {
+                    List<PromQlQueryContent.ContentData.Content> contents = responseEntity.getBody().getData().getResult();
+                    for (PromQlQueryContent.ContentData.Content content : contents) {
+                        Map<String, String> labels = content.getMetric();
+                        Map<String, Object> queryResult = new HashMap<>(8);
+                        queryResult.putAll(labels);
+                        if (content.getValue() != null && content.getValue().length == 2) {
+                            queryResult.put("__timestamp__", content.getValue()[0]);
+                            queryResult.put("__value__", content.getValue()[1]);
+                        } else if (content.getValues() != null && !content.getValues().isEmpty()) {
+                            List<Object> values = new LinkedList<>();
+                            for (Object[] valueArr : content.getValues()) {
+                                values.add(valueArr[1]);
+                            }
+                            queryResult.put("__value__", values);
+                        }
+                        results.add(queryResult);
+                    }
+                }
+            } else {
+                log.error("query metrics data from greptime failed. {}", responseEntity);
+            }
+        } catch (Exception e) {
+            log.error(e.toString(), e);
+        }
+        return results;
+    }
+
+    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;
+    }
+
+    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);
+    }
+
+}
@@ -17,6 +17,8 @@

 package org.apache.hertzbeat.warehouse.db;

+import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
+
 import java.util.List;
 import java.util.Map;

@@ -24,8 +26,14 @@ import java.util.Map;
 * query executor interface
 */
 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);
 }
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hertzbeat.warehouse.db;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hertzbeat.warehouse.constants.WarehouseConstants.SQL;
+
+/**
+ * abstract class for sql query executor
+ */
+@Slf4j
+public abstract class SqlQueryExecutor implements QueryExecutor {
+
+    private static final String supportQueryLanguage = SQL;
+
+    /**
+     * record class for sql connection
+     */
+    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;
+    }
+
+    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);
+    }
+
+}
@@ -18,29 +18,11 @@
 package org.apache.hertzbeat.warehouse.db;


-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 lombok.extern.slf4j.Slf4j;
-import org.apache.hertzbeat.common.constants.NetworkConstants;
-import org.apache.hertzbeat.common.constants.SignConstants;
-import org.apache.hertzbeat.common.util.Base64Util;
-import org.apache.hertzbeat.warehouse.store.history.vm.PromQlQueryContent;
 import org.apache.hertzbeat.warehouse.store.history.vm.VictoriaMetricsProperties;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.http.HttpEntity;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpMethod;
-import org.springframework.http.MediaType;
-import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Component;
-import org.springframework.util.StringUtils;
 import org.springframework.web.client.RestTemplate;
-import org.springframework.web.util.UriComponentsBuilder;

 /**
 * query executor for victor metrics
@@ -48,71 +30,16 @@ import org.springframework.web.util.UriComponentsBuilder;
@Component
@ConditionalOnProperty(prefix = "warehouse.store.victoria-metrics", name = "enabled", havingValue = "true")
@Slf4j
-public class VictoriaMetricsQueryExecutor implements QueryExecutor {
+public class VictoriaMetricsQueryExecutor extends PromqlQueryExecutor {

    private static final String QUERY_PATH = "/api/v1/query";
    
    private final VictoriaMetricsProperties victoriaMetricsProp;

-    private final RestTemplate restTemplate;
-
    public VictoriaMetricsQueryExecutor(VictoriaMetricsProperties victoriaMetricsProp, RestTemplate restTemplate) {
+        super(restTemplate, new HttpPromqlProperties(victoriaMetricsProp.url() + QUERY_PATH,
+                victoriaMetricsProp.username(), victoriaMetricsProp.password()));
        this.victoriaMetricsProp = victoriaMetricsProp;
-        this.restTemplate = restTemplate;
    }

-    @Override
-    public List<Map<String, Object>> execute(String query) {
-        // http run the promql query
-        List<Map<String, Object>> results = new LinkedList<>();
-        try {
-            HttpHeaders headers = new HttpHeaders();
-            headers.setContentType(MediaType.APPLICATION_JSON);
-            headers.setAccept(List.of(MediaType.APPLICATION_JSON));
-            if (StringUtils.hasText(victoriaMetricsProp.username())
-                    && StringUtils.hasText(victoriaMetricsProp.password())) {
-                String authStr = victoriaMetricsProp.username() + ":" + victoriaMetricsProp.password();
-                String encodedAuth = Base64Util.encode(authStr);
-                headers.add(HttpHeaders.AUTHORIZATION,  NetworkConstants.BASIC + SignConstants.BLANK + encodedAuth);
-            }
-            HttpEntity<Void> httpEntity = new HttpEntity<>(headers);
-            URI uri = UriComponentsBuilder.fromHttpUrl(victoriaMetricsProp.url() + QUERY_PATH)
-                    .queryParam("query", URLEncoder.encode(query, StandardCharsets.UTF_8))
-                    .build(true).toUri();
-            ResponseEntity<PromQlQueryContent> responseEntity = restTemplate.exchange(uri,
-                    HttpMethod.GET, httpEntity, PromQlQueryContent.class);
-            if (responseEntity.getStatusCode().is2xxSuccessful()) {
-                if (responseEntity.getBody() != null && responseEntity.getBody().getData() != null
-                        && responseEntity.getBody().getData().getResult() != null) {
-                    List<PromQlQueryContent.ContentData.Content> contents = responseEntity.getBody().getData().getResult();
-                    for (PromQlQueryContent.ContentData.Content content : contents) {
-                        Map<String, String> labels = content.getMetric();
-                        Map<String, Object> queryResult = new HashMap<>(8);
-                        queryResult.putAll(labels);
-                        if (content.getValue() != null && content.getValue().length == 2) {
-                            queryResult.put("__timestamp__", content.getValue()[0]);
-                            queryResult.put("__value__", content.getValue()[1]);
-                        } else if (content.getValues() != null && !content.getValues().isEmpty()) {
-                            List<Object> values = new LinkedList<>();
-                            for (Object[] valueArr : content.getValues()) {
-                                values.add(valueArr[1]);
-                            }
-                            queryResult.put("__value__", values);
-                        }
-                        results.add(queryResult);
-                    }
-                }
-            } else {
-                log.error("query metrics data from victor-metrics failed. {}", responseEntity);
-            }
-        } catch (Exception e) {
-            log.error(e.toString(), e);
-        }
-        return results;
-    }
-
-    @Override
-    public boolean support(String datasource) {
-        return "promql".equals(datasource);
-    }
 }
@@ -15,19 +15,32 @@
 * limitations under the License.
 */

-package org.apache.hertzbeat.push.dao;
+package org.apache.hertzbeat.warehouse.service;

-import org.apache.hertzbeat.common.entity.push.PushMetrics;
-import org.springframework.data.jpa.repository.JpaRepository;
-import org.springframework.transaction.annotation.Transactional;
+import org.apache.hertzbeat.common.entity.dto.query.MetricQueryData;
+
+import java.util.List;

 /**
- * push metrics dao
+ * metrics data query service
 */
-public interface PushMetricsDao extends JpaRepository<PushMetrics, Long> {
+public interface MetricsDataQueryService {
    
-    PushMetrics findFirstByMonitorIdOrderByTimeDesc(Long monitorId);
+    /**
+     * Query metrics data
+     * @param queries query expr
+     * @param time time
+     * @return data
+     */
+    List<MetricQueryData> query(List<String> queries, String queryType, long time);

-    @Transactional(rollbackFor = Exception.class)
-    void deleteAllByTimeBefore(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);
 }
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.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;
+    }
+}
@@ -25,6 +25,8 @@ export class Monitor {
  intervals: number = 60;
  // Monitoring status 0: Paused, 1: Up, 2: Down
  status!: number;
+  // Task type 0: Normal, 1: push auto create, 2: discovery auto create
+  type!: number;
  description!: string;
  labels!: Record<string, string>;
  annotations!: Record<string, string>;
Author	SHA1	Message	Date
tomsun28	3a923a42b0	Merge branch 'master' into pushgateway	2025-04-05 13:52:00 +08:00
vinci	36ad5a6a31	fix QueryExecutor inject	2025-04-05 12:44:29 +08:00
vinci	a6c79d2699	fix QueryExecutor inject	2025-04-05 01:58:46 +08:00
leo	eebfa81627	Delete hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/SQLQueryExecutor.java Signed-off-by: leo <1552443053@qq.com>	2025-04-04 23:42:52 +08:00
vinci	0f9fab646f	fix formats	2025-04-04 23:36:40 +08:00
vinci	ef7a058770	fix formats	2025-04-04 23:31:06 +08:00
leoandgithub-actions[bot]	573c7aaa11	Update hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataQueryServiceImpl.java Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Signed-off-by: leo <1552443053@qq.com>	2025-04-03 19:25:55 +08:00
leoandgithub-actions[bot]	fb348b3d34	Update hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/SQLQueryExecutor.java Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Signed-off-by: leo <1552443053@qq.com>	2025-04-03 19:25:48 +08:00
leoandgithub-actions[bot]	6ad20d3848	Update hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/PromqlQueryExecutor.java Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Signed-off-by: leo <1552443053@qq.com>	2025-04-03 19:24:59 +08:00
leoandgithub-actions[bot]	2bcdf4fb1e	Update hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushSuccessRequestWrapper.java Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Signed-off-by: leo <1552443053@qq.com>	2025-04-03 19:24:52 +08:00
leoandgithub-actions[bot]	07058fdfb6	Update hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushPrometheusStreamReadingFilter.java Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Signed-off-by: leo <1552443053@qq.com>	2025-04-03 19:24:44 +08:00
leoandgithub-actions[bot]	24a6deb636	Update hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushErrorRequestWrapper.java Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Signed-off-by: leo <1552443053@qq.com>	2025-04-03 19:24:36 +08:00
leoandgithub-actions[bot]	475e4a3288	Update hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushFilterConfig.java Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> Signed-off-by: leo <1552443053@qq.com>	2025-04-03 19:24:26 +08:00
tomsun28	8b42f574fe	Merge branch 'master' into pushgateway Signed-off-by: tomsun28 <tomsun28@outlook.com>	2025-04-03 14:39:24 +08:00
vinci	e8d2f9fb69	add PromqlQueryExecutor	2025-03-03 00:03:27 +08:00
vinci	5f0132a8d6	add PromqlQueryExecutor	2025-03-02 23:49:58 +08:00
vinci	4835960827	add PromqlQueryExecutor	2025-03-02 23:34:22 +08:00
vinci	5716e21f1f	add a if condition for push auto	2025-02-28 16:33:22 +08:00
tomsun28	9ca7dc0945	Merge branch 'master' into pushgateway # Conflicts: # web-app/src/app/routes/monitor/monitor-list/monitor-list.component.html	2025-02-23 22:53:42 +08:00
tomsun28	d787a582d5	[improve] push gateway Signed-off-by: tomsun28 <tomsun28@outlook.com>	2025-02-23 19:36:08 +08:00
vinci	495f037ead	1	2025-02-22 12:58:03 +08:00
vinci	c3fda15036	1	2025-02-22 12:04:54 +08:00
vinci	c516b3689c	1	2025-02-22 12:00:14 +08:00
tomsun28	d01f3e2263	[improve] push gateway filter to parse stream Signed-off-by: tomsun28 <tomsun28@outlook.com>	2025-02-22 10:25:40 +08:00
vinci	6fc5d81dae	1	2025-02-16 21:00:24 +08:00
vinci	ab48fc7c34	1	2025-02-16 20:40:15 +08:00
vinci	4f6ca71471	1	2025-02-16 20:14:11 +08:00
vinci	61ab3d4114	1	2025-02-16 20:12:34 +08:00
vinci	4a9aa91130	1	2025-02-12 16:36:39 +08:00