From 03858926742f28a486d26448760752cb3cc6d3d0 Mon Sep 17 00:00:00 2001 From: leo <1552443053@qq.com> Date: Sat, 5 Apr 2025 17:28:11 +0800 Subject: [PATCH] [feature] add pushgateway to push module (#3204) Signed-off-by: tomsun28 Signed-off-by: leo <1552443053@qq.com> Co-authored-by: tomsun28 Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> --- .../alert/service/DataSourceServiceTest.java | 2 +- .../prometheus/PrometheusAutoCollectImpl.java | 2 +- .../collect/prometheus/parser/TextParser.java | 1 + .../prometheus/parser/OnlineParserTest.java | 2 + .../common/constants/CommonConstants.java | 15 ++ .../common/entity/dto}/MetricFamily.java | 2 +- .../entity/dto/query/MetricQueryData.java | 82 +++++++++ .../common/entity/manager/Monitor.java | 3 + .../hertzbeat/common/util}/OnlineParser.java | 101 +++++++---- .../org/apache/hertzbeat/manager/Manager.java | 2 + .../service/impl/MonitorServiceImpl.java | 2 +- .../push/config/PushErrorRequestWrapper.java | 42 +++++ .../push/config/PushFilterConfig.java | 47 +++++ .../PushPrometheusStreamReadingFilter.java | 82 +++++++++ .../config/PushSuccessRequestWrapper.java | 41 +++++ .../push/controller/PushController.java | 62 ------- ...ler.java => PushPrometheusController.java} | 31 ++-- .../hertzbeat/push/dao/PushMonitorDao.java | 8 + .../push/service/PushGatewayService.java | 13 +- .../hertzbeat/push/service/PushService.java | 31 ---- .../service/impl/PushGatewayServiceImpl.java | 99 ++++++++++- .../push/service/impl/PushServiceImpl.java | 143 --------------- .../push/controller/PushControllerTest.java | 93 ---------- ...java => PushPrometheusControllerTest.java} | 21 +-- .../push/dao/PushMetricsDaoTest.java | 77 -------- .../push/service/PushServiceTest.java | 132 -------------- .../constants/WarehouseConstants.java | 4 + .../MetricsDataQueryController.java | 78 ++++++++ .../db/GreptimePromqlQueryExecutor.java | 81 +-------- .../warehouse/db/PromqlQueryExecutor.java | 166 ++++++++++++++++++ .../hertzbeat/warehouse/db/QueryExecutor.java | 10 +- .../warehouse/db/SqlQueryExecutor.java | 67 +++++++ .../db/VictoriaMetricsQueryExecutor.java | 79 +-------- .../service/MetricsDataQueryService.java | 31 +++- .../impl/MetricsDataQueryServiceImpl.java | 67 +++++++ web-app/src/app/pojo/Monitor.ts | 2 + 36 files changed, 946 insertions(+), 775 deletions(-) rename {hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser => hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto}/MetricFamily.java (95%) create mode 100644 hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/query/MetricQueryData.java rename {hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser => hertzbeat-common/src/main/java/org/apache/hertzbeat/common/util}/OnlineParser.java (82%) create mode 100644 hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushErrorRequestWrapper.java create mode 100644 hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushFilterConfig.java create mode 100644 hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushPrometheusStreamReadingFilter.java create mode 100644 hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushSuccessRequestWrapper.java delete mode 100644 hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushController.java rename hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/{PushGatewayController.java => PushPrometheusController.java} (59%) delete mode 100644 hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/PushService.java delete mode 100644 hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushServiceImpl.java delete mode 100644 hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushControllerTest.java rename hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/{PushGatewayControllerTest.java => PushPrometheusControllerTest.java} (75%) delete mode 100644 hertzbeat-push/src/test/java/org/apache/hertzbeat/push/dao/PushMetricsDaoTest.java delete mode 100644 hertzbeat-push/src/test/java/org/apache/hertzbeat/push/service/PushServiceTest.java create mode 100644 hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataQueryController.java create mode 100644 hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/PromqlQueryExecutor.java create mode 100644 hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/SqlQueryExecutor.java rename hertzbeat-push/src/main/java/org/apache/hertzbeat/push/dao/PushMetricsDao.java => hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/MetricsDataQueryService.java (53%) create mode 100644 hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataQueryServiceImpl.java diff --git a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/DataSourceServiceTest.java b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/DataSourceServiceTest.java index f37faa1563..add057139b 100644 --- a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/DataSourceServiceTest.java +++ b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/service/DataSourceServiceTest.java @@ -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} diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/PrometheusAutoCollectImpl.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/PrometheusAutoCollectImpl.java index f2ea5f4e29..f92c3d6455 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/PrometheusAutoCollectImpl.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/PrometheusAutoCollectImpl.java @@ -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; diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/TextParser.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/TextParser.java index 67e977de07..9f622a856f 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/TextParser.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/TextParser.java @@ -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; /** diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java index ab3bd0eb69..f01da4a5ac 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java @@ -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; diff --git a/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/constants/CommonConstants.java b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/constants/CommonConstants.java index 4fae43283d..24e1d15433 100644 --- a/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/constants/CommonConstants.java +++ b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/constants/CommonConstants.java @@ -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; } diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/MetricFamily.java b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/MetricFamily.java similarity index 95% rename from hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/MetricFamily.java rename to hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/MetricFamily.java index 1d9e5a3e93..b135b0fae4 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/MetricFamily.java +++ b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/MetricFamily.java @@ -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; diff --git a/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/query/MetricQueryData.java b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/query/MetricQueryData.java new file mode 100644 index 0000000000..f6ba8562c2 --- /dev/null +++ b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/dto/query/MetricQueryData.java @@ -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> values; + + /** + * Metric Schema + */ + @Data + @AllArgsConstructor + @NoArgsConstructor + @Builder + public static final class MetricSchema { + + @Schema(title = "Metrics Field") + private List fields; + + @Schema(title = "Meta Information") + private Map 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; + } +} diff --git a/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/manager/Monitor.java b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/manager/Monitor.java index 634f168e1a..7133f87d85 100644 --- a/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/manager/Monitor.java +++ b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/entity/manager/Monitor.java @@ -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) diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/util/OnlineParser.java similarity index 82% rename from hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java rename to hertzbeat-common/src/main/java/org/apache/hertzbeat/common/util/OnlineParser.java index 5a754e3e5c..da1cb36a4b 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java +++ b/hertzbeat-common/src/main/java/org/apache/hertzbeat/common/util/OnlineParser.java @@ -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 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 labelList = new LinkedList<>(); + metric.setLabels(labelList); if (i == '{') { - List labelList = new LinkedList<>(); parseLabels(inputStream, stringBuilder, labelList); - metric.setLabels(labelList); i = skipSpaces(inputStream).getInt(); } @@ -307,8 +336,8 @@ public class OnlineParser { public static Map parseMetrics(InputStream inputStream) throws IOException { Map 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()); diff --git a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/Manager.java b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/Manager.java index ce10a4e51b..94584238e5 100644 --- a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/Manager.java +++ b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/Manager.java @@ -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); diff --git a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/service/impl/MonitorServiceImpl.java b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/service/impl/MonitorServiceImpl.java index 0e45558407..6644ee4a9a 100644 --- a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/service/impl/MonitorServiceImpl.java +++ b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/service/impl/MonitorServiceImpl.java @@ -531,7 +531,7 @@ public class MonitorServiceImpl implements MonitorService { MonitorDto monitorDto = new MonitorDto(); List 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 metricsDataList = warehouseService.queryMonitorMetricsData(id); List metrics = metricsDataList.stream().map(CollectRep.MetricsData::getMetrics).collect(Collectors.toList()); monitorDto.setMetrics(metrics); diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushErrorRequestWrapper.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushErrorRequestWrapper.java new file mode 100644 index 0000000000..96ec20288f --- /dev/null +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushErrorRequestWrapper.java @@ -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; + } +} diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushFilterConfig.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushFilterConfig.java new file mode 100644 index 0000000000..9c339d6bae --- /dev/null +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushFilterConfig.java @@ -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 contentTypeFilter() { + FilterRegistrationBean registrationBean = new FilterRegistrationBean<>(); + registrationBean.setFilter(new PushPrometheusStreamReadingFilter(pushGatewayService)); + registrationBean.addUrlPatterns(URI_PREFIX); + registrationBean.setOrder(Integer.MIN_VALUE); + return registrationBean; + } +} diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushPrometheusStreamReadingFilter.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushPrometheusStreamReadingFilter.java new file mode 100644 index 0000000000..47b2abd558 --- /dev/null +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushPrometheusStreamReadingFilter.java @@ -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() {} +} diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushSuccessRequestWrapper.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushSuccessRequestWrapper.java new file mode 100644 index 0000000000..817f3cc110 --- /dev/null +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/config/PushSuccessRequestWrapper.java @@ -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; + } +} diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushController.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushController.java deleted file mode 100644 index e0598b470c..0000000000 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushController.java +++ /dev/null @@ -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> 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> 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)); - } -} diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushGatewayController.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushPrometheusController.java similarity index 59% rename from hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushGatewayController.java rename to hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushPrometheusController.java index f355a4a739..02e4d4bb28 100644 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushGatewayController.java +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/controller/PushPrometheusController.java @@ -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> 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> 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()))); } } diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/dao/PushMonitorDao.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/dao/PushMonitorDao.java index b0f2870e98..57463c539e 100644 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/dao/PushMonitorDao.java +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/dao/PushMonitorDao.java @@ -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 { + + /** + * Find all monitoring entities by type + * @param type Monitoring type + * @return Monitoring entity list + */ + List findMonitorsByType(byte type); } diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/PushGatewayService.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/PushGatewayService.java index daa0cf1a8f..b3aa4581e1 100644 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/PushGatewayService.java +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/PushGatewayService.java @@ -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); } diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/PushService.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/PushService.java deleted file mode 100644 index 41143e3c36..0000000000 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/PushService.java +++ /dev/null @@ -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); -} diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java index ffbb4b2127..b85943a1af 100644 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java @@ -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 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 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 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 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 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; + } } } diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushServiceImpl.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushServiceImpl.java deleted file mode 100644 index d3780a3995..0000000000 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushServiceImpl.java +++ /dev/null @@ -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 monitorIdCache; // key: monitorId, value: time stamp of last query - - private static final long cacheTimeout = 5000L; // ms - - private final Map 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 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 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> 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; - } - -} diff --git a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushControllerTest.java b/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushControllerTest.java deleted file mode 100644 index 03d9f9a61a..0000000000 --- a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushControllerTest.java +++ /dev/null @@ -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(); - } - -} diff --git a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushGatewayControllerTest.java b/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushPrometheusControllerTest.java similarity index 75% rename from hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushGatewayControllerTest.java rename to hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushPrometheusControllerTest.java index c71377e56d..4bd3834dac 100644 --- a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushGatewayControllerTest.java +++ b/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/controller/PushPrometheusControllerTest.java @@ -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()); } } diff --git a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/dao/PushMetricsDaoTest.java b/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/dao/PushMetricsDaoTest.java deleted file mode 100644 index c90a9c7e4d..0000000000 --- a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/dao/PushMetricsDaoTest.java +++ /dev/null @@ -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); - } -} diff --git a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/service/PushServiceTest.java b/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/service/PushServiceTest.java deleted file mode 100644 index d5ce5e0ebe..0000000000 --- a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/service/PushServiceTest.java +++ /dev/null @@ -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 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()); - } - -} diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/constants/WarehouseConstants.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/constants/WarehouseConstants.java index aee6e4ff95..822bcd0aa8 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/constants/WarehouseConstants.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/constants/WarehouseConstants.java @@ -56,4 +56,8 @@ public interface WarehouseConstants { String MEMORY = "memory"; } + String PROMQL = "promql"; + + String SQL = "sql"; + } diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataQueryController.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataQueryController.java new file mode 100644 index 0000000000..fef41cae7f --- /dev/null +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/controller/MetricsDataQueryController.java @@ -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>> queryMetricsData( + @Parameter(description = "Query PromQL expr list", example = "cpu") + @RequestParam List 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>> queryMetricsDataRange( + @Parameter(description = "Query PromQL expr list", example = "cpu") + @RequestParam List 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))); + } +} diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimePromqlQueryExecutor.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimePromqlQueryExecutor.java index aba37c5d7e..ac48e885fa 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimePromqlQueryExecutor.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/GreptimePromqlQueryExecutor.java @@ -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> execute(String query) { - // http run the promql query - List> 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 httpEntity = new HttpEntity<>(headers); - URI uri = UriComponentsBuilder.fromHttpUrl(greptimeProperties.httpEndpoint() + QUERY_PATH) - .queryParam("query", URLEncoder.encode(query, StandardCharsets.UTF_8)) - .build(true).toUri(); - ResponseEntity 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 contents = responseEntity.getBody().getData().getResult(); - for (PromQlQueryContent.ContentData.Content content : contents) { - Map labels = content.getMetric(); - Map 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 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); } } diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/PromqlQueryExecutor.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/PromqlQueryExecutor.java new file mode 100644 index 0000000000..4e4c0c03fa --- /dev/null +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/PromqlQueryExecutor.java @@ -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> http_promql(Map params) { + // http run the promql query + List> 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 httpEntity = new HttpEntity<>(headers); + UriComponentsBuilder uriComponentsBuilder = UriComponentsBuilder.fromHttpUrl(httpPromqlProperties.url); + for (Map.Entry entry : params.entrySet()) { + uriComponentsBuilder.queryParam(entry.getKey(), entry.getValue()); + } + URI uri = uriComponentsBuilder.build(true).toUri(); + ResponseEntity 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 contents = responseEntity.getBody().getData().getResult(); + for (PromQlQueryContent.ContentData.Content content : contents) { + Map labels = content.getMetric(); + Map 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 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> metrics = (List>) object; + // todo + } catch (Exception e) { + log.error("converting to metric query data failed."); + } + return metricQueryData; + } + + public List> execute(String queryString) { + Map params = new HashMap<>(); + params.put(HTTP_QUERY_PARAM, URLEncoder.encode(queryString, StandardCharsets.UTF_8)); + return http_promql(params); + } + + public List> query(String queryString, long time) { + Map 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> query_range(String queryString, long start, long end, String step) { + Map 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); + } + +} diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/QueryExecutor.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/QueryExecutor.java index 97b7ad4264..dc22a6ec92 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/QueryExecutor.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/QueryExecutor.java @@ -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> execute(String query); + + List> query(String query, long time); + + List> query_range(String query, long start, long end, String step); boolean support(String datasource); } diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/SqlQueryExecutor.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/SqlQueryExecutor.java new file mode 100644 index 0000000000..230b8005f9 --- /dev/null +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/SqlQueryExecutor.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.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> do_sql(Map params); + + public MetricQueryData convertToMetricQueryData(Object object) { + MetricQueryData metricQueryData = new MetricQueryData(); + try { + List> metrics = (List>) object; + // todo + } catch (Exception e) { + log.error("converting to metric query data failed."); + } + return metricQueryData; + } + + public abstract List> execute(String query); + + public abstract List> query(String query, long time); + + public abstract List> query_range(String query, long start, long end, String step); + + + public boolean support(String datasource) { + return supportQueryLanguage.equals(datasource); + } + +} \ No newline at end of file diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/VictoriaMetricsQueryExecutor.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/VictoriaMetricsQueryExecutor.java index 90311d83de..7d0ac85774 100644 --- a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/VictoriaMetricsQueryExecutor.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/db/VictoriaMetricsQueryExecutor.java @@ -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> execute(String query) { - // http run the promql query - List> 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 httpEntity = new HttpEntity<>(headers); - URI uri = UriComponentsBuilder.fromHttpUrl(victoriaMetricsProp.url() + QUERY_PATH) - .queryParam("query", URLEncoder.encode(query, StandardCharsets.UTF_8)) - .build(true).toUri(); - ResponseEntity 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 contents = responseEntity.getBody().getData().getResult(); - for (PromQlQueryContent.ContentData.Content content : contents) { - Map labels = content.getMetric(); - Map 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 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); - } } diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/dao/PushMetricsDao.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/MetricsDataQueryService.java similarity index 53% rename from hertzbeat-push/src/main/java/org/apache/hertzbeat/push/dao/PushMetricsDao.java rename to hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/MetricsDataQueryService.java index b575ce66d2..73ac03f575 100644 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/dao/PushMetricsDao.java +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/MetricsDataQueryService.java @@ -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 { +public interface MetricsDataQueryService { - PushMetrics findFirstByMonitorIdOrderByTimeDesc(Long monitorId); + /** + * Query metrics data + * @param queries query expr + * @param time time + * @return data + */ + List query(List 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 queryRange(List queries, String queryType, long start, long end, String step); } diff --git a/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataQueryServiceImpl.java b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataQueryServiceImpl.java new file mode 100644 index 0000000000..333148949a --- /dev/null +++ b/hertzbeat-warehouse/src/main/java/org/apache/hertzbeat/warehouse/service/impl/MetricsDataQueryServiceImpl.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.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 executors; + + @Override + public List query(List 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 metricQueryDataList = new ArrayList<>(); + for (String query : queries) { + metricQueryDataList.add(executor.convertToMetricQueryData(executor.query(query, time))); + } + return metricQueryDataList; + } + + @Override + public List queryRange(List 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 metricQueryDataList = new ArrayList<>(); + for (String query : queries) { + metricQueryDataList.add(executor.convertToMetricQueryData(executor.query_range(query, start, end, step))); + } + return metricQueryDataList; + } +} diff --git a/web-app/src/app/pojo/Monitor.ts b/web-app/src/app/pojo/Monitor.ts index 2f42c70532..e84c67831c 100644 --- a/web-app/src/app/pojo/Monitor.ts +++ b/web-app/src/app/pojo/Monitor.ts @@ -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; annotations!: Record;