This is an automated email from the ASF dual-hosted git repository. lizhimins pushed a commit to branch rocketmq-studio in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
commit 194cf63e0107208334fa8b44609868d7fd848e68 Author: wizcraft_kris <[email protected]> AuthorDate: Wed Jul 22 17:30:51 2026 +0800 feat: add Prometheus range query adapter (#432) Add a real Prometheus /api/v1/query_range adapter as the foundation for observability (#431): configurable base URL, timeouts, Basic/Bearer auth, error mapping, and tests. --- docs/api-spec.md | 76 ++++- .../studio/cluster/metrics/MetricDataVO.java | 35 ++- .../studio/cluster/metrics/MetricQueryDTO.java | 23 ++ .../studio/cluster/metrics/MetricsController.java | 17 +- .../studio/cluster/metrics/MetricsService.java | 4 +- ...{MetricDataVO.java => PrometheusException.java} | 25 +- .../cluster/metrics/PrometheusMetricsSource.java | 260 +++++++++++++++-- ...tricsService.java => PrometheusProperties.java} | 30 +- .../common/exception/GlobalExceptionHandler.java | 28 ++ server/src/main/resources/application.yml | 10 + .../cluster/metrics/MetricsControllerTest.java | 55 ++++ .../studio/cluster/metrics/MetricsServiceTest.java | 80 +++-- .../metrics/PrometheusMetricsSourceTest.java | 321 +++++++++++++++++++++ web/src/api/metrics.ts | 44 ++- 14 files changed, 912 insertions(+), 96 deletions(-) diff --git a/docs/api-spec.md b/docs/api-spec.md index 581e8d0a..2b0bd143 100644 --- a/docs/api-spec.md +++ b/docs/api-spec.md @@ -1652,17 +1652,83 @@ POST /api/metrics/query | 字段 | 类型 | 必填 | 说明 | |------|------|------|------| -| `metric` | `string` | 是 | 指标名称（如 `tps_in`、`tps_out`、`message_count`、`disk_usage`） | +| `metric` | `string` | 是 | PromQL 表达式，最大 4096 个字符 | | `start` | `number` | 是 | 起始时间（Unix 时间戳，秒） | | `end` | `number` | 是 | 结束时间（Unix 时间戳，秒） | -| `step` | `string` | 否 | 采样步长（如 `"60s"`、`"5m"`、`"1h"`） | +| `step` | `string` | 是 | 查询分辨率，可以是持续时间或秒数（如 `"30s"`、`"5m"`、`"1h"`） | -**Response `data`:** `MetricsResult` +**Request 示例：** + +```json +{ + "metric": "sum(rate(rocketmq_messages_in_total[1m])) by (node_id)", + "start": 1784112606, + "end": 1784114406, + "step": "30s" +} +``` + +**Response `data`:** `MetricData` + +| 字段 | 类型 | 说明 | +|------|------|------| +| `resultType` | `string` | Prometheus 结果类型；范围查询通常为 `matrix` | +| `series` | `MetricSeries[]` | 查询返回的时间序列 | +| `warnings` | `string[]` | Prometheus 返回的非致命警告，没有警告时为空数组 | + +**MetricSeries：** + +| 字段 | 类型 | 说明 | +|------|------|------| +| `labels` | `object` | 序列的完整标签集合，包括可能存在的 `__name__` | +| `values` | `MetricSample[]` | 浮点样本；没有浮点样本时为空数组 | +| `histograms` | `MetricHistogramSample[]` | Native Histogram 样本；没有 Histogram 样本时为空数组 | + +同一序列可能只有 `values`、只有 `histograms`，或同时包含两者。 + +**MetricSample：** + +| 字段 | 类型 | 说明 | +|------|------|------| +| `timestamp` | `number` | Unix 时间戳，可能包含小数秒 | +| `value` | `string` | Prometheus 样本原始字符串，保留小数精度以及 `NaN`、`+Inf`、`-Inf` | + +**MetricHistogramSample：** | 字段 | 类型 | 说明 | |------|------|------| -| `metric` | `string` | 指标名称 | -| `values` | `[number, number][]` | 数据点数组，每项为 `[timestamp, value]` | +| `timestamp` | `number` | Unix 时间戳，可能包含小数秒 | +| `histogram` | `object` | Prometheus Native Histogram 原始对象，包含 `count`、`sum` 和 `buckets` | + +**Prometheus 配置：** + +```yaml +studio: + metrics: + prometheus: + base-url: ${STUDIO_METRICS_PROMETHEUS_BASE_URL:} + connect-timeout: ${STUDIO_METRICS_PROMETHEUS_CONNECT_TIMEOUT:3s} + read-timeout: ${STUDIO_METRICS_PROMETHEUS_READ_TIMEOUT:10s} + username: ${STUDIO_METRICS_PROMETHEUS_USERNAME:} + password: ${STUDIO_METRICS_PROMETHEUS_PASSWORD:} + bearer-token: ${STUDIO_METRICS_PROMETHEUS_BEARER_TOKEN:} +``` + +- `base-url` 是 Prometheus 或 Prometheus-compatible 服务的 URL 前缀；服务会在其后追加 `/api/v1/query_range`。 +- 未配置 `base-url` 时，查询接口返回 HTTP 503。 +- `connect-timeout` 默认 3 秒，`read-timeout` 默认 10 秒。 +- Bearer Token 的优先级高于 Basic Auth；Basic Auth 必须同时配置 `username` 和 `password`。 +- 密码和 Token 等敏感配置应通过环境变量或其他外部化配置传入，不应提交到代码仓库。 + +**错误响应：** + +| HTTP 状态 | 场景 | +|-----------|------| +| `400` | JSON 无法解析、字段类型错误、请求校验失败或 PromQL 参数错误 | +| `422` | Prometheus 无法执行 PromQL 表达式 | +| `502` | 无法连接 Prometheus，或 Prometheus 返回非法响应 | +| `503` | Prometheus 未配置或暂时不可用 | +| `504` | Prometheus 查询超时 | --- diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricDataVO.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricDataVO.java index 06fbf70c..29846dd1 100644 --- a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricDataVO.java +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricDataVO.java @@ -16,11 +16,13 @@ */ package com.rocketmq.studio.cluster.metrics; +import com.fasterxml.jackson.databind.JsonNode; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; import lombok.NoArgsConstructor; +import java.util.Map; import java.util.List; @Data @@ -28,6 +30,35 @@ import java.util.List; @NoArgsConstructor @AllArgsConstructor public class MetricDataVO { - private String metric; - private List<long[]> values; + private String resultType; + private List<MetricSeriesVO> series; + private List<String> warnings; + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class MetricSeriesVO { + private Map<String, String> labels; + private List<MetricSampleVO> values; + private List<MetricHistogramSampleVO> histograms; + } + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class MetricSampleVO { + private double timestamp; + private String value; + } + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class MetricHistogramSampleVO { + private double timestamp; + private JsonNode histogram; + } } diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricQueryDTO.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricQueryDTO.java index 0884ea48..de032a07 100644 --- a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricQueryDTO.java +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricQueryDTO.java @@ -16,6 +16,10 @@ */ package com.rocketmq.studio.cluster.metrics; +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.Positive; +import jakarta.validation.constraints.Size; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; @@ -25,9 +29,28 @@ import lombok.NoArgsConstructor; @Builder @NoArgsConstructor @AllArgsConstructor +@Schema(description = "Prometheus range query") public class MetricQueryDTO { + @Schema(description = "PromQL expression evaluated by Prometheus", + example = "sum(rate(rocketmq_messages_in_total[1m])) by (node_id)", minLength = 1, + requiredMode = Schema.RequiredMode.REQUIRED) + @NotBlank(message = "Metric query is required") + @Size(max = 4096, message = "Metric query must not exceed 4096 characters") private String metric; + + @Schema(description = "Range start as a Unix timestamp in seconds", example = "1784112606", + requiredMode = Schema.RequiredMode.REQUIRED) + @Positive(message = "Metric query start must be positive") private long start; + + @Schema(description = "Range end as a Unix timestamp in seconds", example = "1784114406", + requiredMode = Schema.RequiredMode.REQUIRED) + @Positive(message = "Metric query end must be positive") private long end; + + @Schema(description = "Prometheus query resolution step as a duration or number of seconds", example = "30s", + minLength = 1, requiredMode = Schema.RequiredMode.REQUIRED) + @NotBlank(message = "Metric query step is required") + @Size(max = 32, message = "Metric query step must not exceed 32 characters") private String step; } diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsController.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsController.java index 9015c010..6ee8334f 100644 --- a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsController.java +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsController.java @@ -17,6 +17,10 @@ package com.rocketmq.studio.cluster.metrics; import com.rocketmq.studio.common.domain.Result; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.responses.ApiResponse; +import io.swagger.v3.oas.annotations.responses.ApiResponses; +import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; @@ -30,8 +34,19 @@ public class MetricsController { private final MetricsService metricsService; + @Operation(summary = "Query Prometheus range metrics", + description = "Executes a PromQL range query against the configured Prometheus server") + @ApiResponses({ + @ApiResponse(responseCode = "200", description = "Range query completed successfully", + useReturnTypeSchema = true), + @ApiResponse(responseCode = "400", description = "Invalid request or PromQL expression"), + @ApiResponse(responseCode = "422", description = "Prometheus could not execute the expression"), + @ApiResponse(responseCode = "502", description = "Prometheus connection or response failure"), + @ApiResponse(responseCode = "503", description = "Prometheus is unavailable or not configured"), + @ApiResponse(responseCode = "504", description = "Prometheus query timed out") + }) @PostMapping("/query") - public Result<MetricDataVO> query(@RequestBody MetricQueryDTO query) { + public Result<MetricDataVO> query(@Valid @RequestBody MetricQueryDTO query) { return Result.ok(metricsService.query(query)); } } diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsService.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsService.java index 0a381275..65bbe77f 100644 --- a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsService.java +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsService.java @@ -28,8 +28,8 @@ public class MetricsService { private final MetricsSource metricsSource; public MetricDataVO query(MetricQueryDTO query) { - log.info("Querying metrics: metric={}, start={}, end={}, step={}", - query.getMetric(), query.getStart(), query.getEnd(), query.getStep()); + log.debug("Querying metrics: start={}, end={}, step={}", + query.getStart(), query.getEnd(), query.getStep()); return metricsSource.query(query); } } diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricDataVO.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusException.java similarity index 67% copy from server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricDataVO.java copy to server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusException.java index 06fbf70c..e286c66e 100644 --- a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricDataVO.java +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusException.java @@ -16,18 +16,19 @@ */ package com.rocketmq.studio.cluster.metrics; -import lombok.AllArgsConstructor; -import lombok.Builder; -import lombok.Data; -import lombok.NoArgsConstructor; +import lombok.Getter; -import java.util.List; +@Getter +public class PrometheusException extends RuntimeException { + private final int statusCode; -@Data -@Builder -@NoArgsConstructor -@AllArgsConstructor -public class MetricDataVO { - private String metric; - private List<long[]> values; + public PrometheusException(int statusCode, String message) { + super(message); + this.statusCode = statusCode; + } + + public PrometheusException(int statusCode, String message, Throwable cause) { + super(message, cause); + this.statusCode = statusCode; + } } diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusMetricsSource.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusMetricsSource.java index cc48fe5a..701bab1a 100644 --- a/server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusMetricsSource.java +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusMetricsSource.java @@ -16,53 +16,263 @@ */ package com.rocketmq.studio.cluster.metrics; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.client.SimpleClientHttpRequestFactory; import org.springframework.stereotype.Component; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; +import org.springframework.util.StringUtils; +import org.springframework.web.client.ResourceAccessException; +import org.springframework.web.client.RestClient; +import org.springframework.web.client.RestClientException; +import org.springframework.web.client.RestClientResponseException; -import java.util.ArrayList; +import java.io.IOException; +import java.net.SocketTimeoutException; +import java.net.URI; +import java.util.Iterator; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; +import java.util.stream.StreamSupport; @Slf4j @Component public class PrometheusMetricsSource implements MetricsSource { + private static final String QUERY_RANGE_PATH = "/api/v1/query_range"; + + private final RestClient restClient; + private final ObjectMapper objectMapper; + private final PrometheusProperties properties; + + public PrometheusMetricsSource(RestClient.Builder restClientBuilder, ObjectMapper objectMapper, + PrometheusProperties properties) { + SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory(); + requestFactory.setConnectTimeout(properties.getConnectTimeout()); + requestFactory.setReadTimeout(properties.getReadTimeout()); + this.restClient = restClientBuilder.requestFactory(requestFactory).build(); + this.objectMapper = objectMapper; + this.properties = properties; + } + @Override public MetricDataVO query(MetricQueryDTO query) { - log.info("Querying Prometheus: metric={}, start={}, end={}, step={}", - query.getMetric(), query.getStart(), query.getEnd(), query.getStep()); + validateQuery(query); + URI queryRangeUri = queryRangeUri(); + MultiValueMap<String, String> form = new LinkedMultiValueMap<>(); + form.add("query", query.getMetric()); + form.add("start", Long.toString(query.getStart())); + form.add("end", Long.toString(query.getEnd())); + form.add("step", query.getStep()); + + log.debug("Querying Prometheus range: start={}, end={}, step={}", + query.getStart(), query.getEnd(), query.getStep()); + + try { + JsonNode response = restClient.post() + .uri(queryRangeUri) + .contentType(MediaType.APPLICATION_FORM_URLENCODED) + .headers(this::applyAuthentication) + .body(form) + .retrieve() + .body(JsonNode.class); + return parseResponse(response); + } catch (PrometheusException exception) { + throw exception; + } catch (RestClientResponseException exception) { + throw responseException(exception); + } catch (ResourceAccessException exception) { + if (hasCause(exception, SocketTimeoutException.class)) { + throw new PrometheusException(HttpStatus.GATEWAY_TIMEOUT.value(), + "Prometheus query timed out", exception); + } + throw new PrometheusException(HttpStatus.BAD_GATEWAY.value(), + "Failed to connect to Prometheus", exception); + } catch (RestClientException exception) { + if (hasCause(exception, SocketTimeoutException.class)) { + throw new PrometheusException(HttpStatus.GATEWAY_TIMEOUT.value(), + "Prometheus query timed out", exception); + } + throw new PrometheusException(HttpStatus.BAD_GATEWAY.value(), + "Prometheus query failed", exception); + } + } + + private void validateQuery(MetricQueryDTO query) { + if (query == null) { + throw new PrometheusException(HttpStatus.BAD_REQUEST.value(), "Metric query is required"); + } + if (query.getEnd() < query.getStart()) { + throw new PrometheusException(HttpStatus.BAD_REQUEST.value(), + "Metric query end must not be earlier than start"); + } + } - // Stub: generate sample data points - List<long[]> values = new ArrayList<>(); - long stepSeconds = parseStep(query.getStep()); - long start = query.getStart(); - long end = query.getEnd(); + private URI queryRangeUri() { + if (!StringUtils.hasText(properties.getBaseUrl())) { + throw new PrometheusException(HttpStatus.SERVICE_UNAVAILABLE.value(), + "Prometheus base URL is not configured"); + } + try { + String baseUrl = properties.getBaseUrl().strip(); + while (baseUrl.endsWith("/")) { + baseUrl = baseUrl.substring(0, baseUrl.length() - 1); + } + URI uri = URI.create(baseUrl + QUERY_RANGE_PATH); + if (!"http".equalsIgnoreCase(uri.getScheme()) && !"https".equalsIgnoreCase(uri.getScheme())) { + throw new IllegalArgumentException("Unsupported Prometheus URL scheme"); + } + return uri; + } catch (IllegalArgumentException exception) { + throw new PrometheusException(HttpStatus.SERVICE_UNAVAILABLE.value(), + "Prometheus base URL is invalid", exception); + } + } + + private void applyAuthentication(HttpHeaders headers) { + if (StringUtils.hasText(properties.getBearerToken())) { + headers.setBearerAuth(properties.getBearerToken()); + return; + } + boolean hasUsername = StringUtils.hasText(properties.getUsername()); + boolean hasPassword = StringUtils.hasText(properties.getPassword()); + if (hasUsername != hasPassword) { + throw new PrometheusException(HttpStatus.SERVICE_UNAVAILABLE.value(), + "Prometheus basic authentication is incomplete"); + } + if (hasUsername) { + headers.setBasicAuth(properties.getUsername(), properties.getPassword()); + } + } - for (long ts = start; ts <= end; ts += stepSeconds) { - values.add(new long[]{ts, (long) (Math.random() * 100)}); + private MetricDataVO parseResponse(JsonNode response) { + if (response == null || !"success".equals(response.path("status").asText())) { + throw responseBodyException(response, HttpStatus.BAD_GATEWAY.value()); } + JsonNode data = response.path("data"); + JsonNode result = data.path("result"); + if (!data.isObject() || !result.isArray() || !StringUtils.hasText(data.path("resultType").asText())) { + throw new PrometheusException(HttpStatus.BAD_GATEWAY.value(), + "Prometheus returned a malformed response"); + } + + List<MetricDataVO.MetricSeriesVO> series = StreamSupport.stream(result.spliterator(), false) + .map(this::parseSeries) + .toList(); + List<String> warnings = parseWarnings(response.path("warnings")); + return MetricDataVO.builder() - .metric(query.getMetric()) - .values(values) + .resultType(data.path("resultType").asText()) + .series(series) + .warnings(warnings) + .build(); + } + + private MetricDataVO.MetricSeriesVO parseSeries(JsonNode seriesNode) { + JsonNode metric = seriesNode.path("metric"); + JsonNode values = seriesNode.path("values"); + JsonNode histograms = seriesNode.path("histograms"); + boolean hasValues = values.isArray(); + boolean hasHistograms = histograms.isArray(); + boolean hasSamples = hasValues || hasHistograms; + boolean invalidValues = !values.isMissingNode() && !hasValues; + boolean invalidHistograms = !histograms.isMissingNode() && !hasHistograms; + if (!metric.isObject() || invalidValues || invalidHistograms || !hasSamples) { + throw new PrometheusException(HttpStatus.BAD_GATEWAY.value(), + "Prometheus returned a malformed time series"); + } + + Map<String, String> labels = new LinkedHashMap<>(); + Iterator<Map.Entry<String, JsonNode>> fields = metric.fields(); + fields.forEachRemaining(entry -> labels.put(entry.getKey(), entry.getValue().asText())); + + List<MetricDataVO.MetricSampleVO> samples = hasValues + ? StreamSupport.stream(values.spliterator(), false).map(this::parseSample).toList() + : List.of(); + List<MetricDataVO.MetricHistogramSampleVO> histogramSamples = hasHistograms + ? StreamSupport.stream(histograms.spliterator(), false).map(this::parseHistogramSample).toList() + : List.of(); + return MetricDataVO.MetricSeriesVO.builder() + .labels(labels) + .values(samples) + .histograms(histogramSamples) + .build(); + } + + private MetricDataVO.MetricSampleVO parseSample(JsonNode sampleNode) { + if (!sampleNode.isArray() || sampleNode.size() != 2 || !sampleNode.get(0).isNumber()) { + throw new PrometheusException(HttpStatus.BAD_GATEWAY.value(), + "Prometheus returned a malformed sample"); + } + return MetricDataVO.MetricSampleVO.builder() + .timestamp(sampleNode.get(0).asDouble()) + .value(sampleNode.get(1).asText()) .build(); } - private long parseStep(String step) { - if (step == null || step.isEmpty()) { - return 60; + private MetricDataVO.MetricHistogramSampleVO parseHistogramSample(JsonNode sampleNode) { + if (!sampleNode.isArray() || sampleNode.size() != 2 + || !sampleNode.get(0).isNumber() || !sampleNode.get(1).isObject()) { + throw new PrometheusException(HttpStatus.BAD_GATEWAY.value(), + "Prometheus returned a malformed histogram sample"); } + return MetricDataVO.MetricHistogramSampleVO.builder() + .timestamp(sampleNode.get(0).asDouble()) + .histogram(sampleNode.get(1)) + .build(); + } + + private List<String> parseWarnings(JsonNode warningsNode) { + if (!warningsNode.isArray()) { + return List.of(); + } + return StreamSupport.stream(warningsNode.spliterator(), false) + .map(JsonNode::asText) + .toList(); + } + + private PrometheusException responseException(RestClientResponseException exception) { + JsonNode response = null; try { - if (step.endsWith("s")) { - return Long.parseLong(step.substring(0, step.length() - 1)); - } else if (step.endsWith("m")) { - return Long.parseLong(step.substring(0, step.length() - 1)) * 60; - } else if (step.endsWith("h")) { - return Long.parseLong(step.substring(0, step.length() - 1)) * 3600; + response = objectMapper.readTree(exception.getResponseBodyAsString()); + } catch (IOException ignored) { + log.debug("Failed to parse Prometheus error response"); + } + int upstreamStatus = exception.getStatusCode().value(); + int statusCode = switch (upstreamStatus) { + case 400, 422, 503 -> upstreamStatus; + default -> HttpStatus.BAD_GATEWAY.value(); + }; + return responseBodyException(response, statusCode); + } + + private PrometheusException responseBodyException(JsonNode response, int statusCode) { + String errorType = response == null ? "" : response.path("errorType").asText(); + String error = response == null ? "" : response.path("error").asText(); + if (StringUtils.hasText(error)) { + String message = StringUtils.hasText(errorType) + ? "Prometheus query failed (" + errorType + "): " + error + : "Prometheus query failed: " + error; + return new PrometheusException(statusCode, message); + } + return new PrometheusException(statusCode, "Prometheus query failed"); + } + + private boolean hasCause(Throwable throwable, Class<? extends Throwable> causeType) { + Throwable current = throwable; + while (current != null) { + if (causeType.isInstance(current)) { + return true; } - return Long.parseLong(step); - } catch (NumberFormatException e) { - log.warn("Failed to parse step '{}', defaulting to 60s", step); - return 60; + current = current.getCause(); } + return false; } } diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsService.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusProperties.java similarity index 60% copy from server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsService.java copy to server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusProperties.java index 0a381275..e37e5826 100644 --- a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricsService.java +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/PrometheusProperties.java @@ -16,20 +16,22 @@ */ package com.rocketmq.studio.cluster.metrics; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Service; +import lombok.Getter; +import lombok.Setter; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; -@Slf4j -@Service -@RequiredArgsConstructor -public class MetricsService { +import java.time.Duration; - private final MetricsSource metricsSource; - - public MetricDataVO query(MetricQueryDTO query) { - log.info("Querying metrics: metric={}, start={}, end={}, step={}", - query.getMetric(), query.getStart(), query.getEnd(), query.getStep()); - return metricsSource.query(query); - } +@Getter +@Setter +@Component +@ConfigurationProperties(prefix = "studio.metrics.prometheus") +public class PrometheusProperties { + private String baseUrl; + private Duration connectTimeout = Duration.ofSeconds(3); + private Duration readTimeout = Duration.ofSeconds(10); + private String username; + private String password; + private String bearerToken; } diff --git a/server/src/main/java/com/rocketmq/studio/common/exception/GlobalExceptionHandler.java b/server/src/main/java/com/rocketmq/studio/common/exception/GlobalExceptionHandler.java index 3bc1c19e..26cffb94 100644 --- a/server/src/main/java/com/rocketmq/studio/common/exception/GlobalExceptionHandler.java +++ b/server/src/main/java/com/rocketmq/studio/common/exception/GlobalExceptionHandler.java @@ -16,10 +16,14 @@ */ package com.rocketmq.studio.common.exception; +import com.rocketmq.studio.cluster.metrics.PrometheusException; import com.rocketmq.studio.common.domain.Result; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.http.converter.HttpMessageNotReadableException; +import org.springframework.web.bind.MethodArgumentNotValidException; import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestControllerAdvice; @@ -36,6 +40,30 @@ public class GlobalExceptionHandler { return Result.error(ex.getCode(), ex.getMessage()); } + @ExceptionHandler(PrometheusException.class) + public ResponseEntity<Result<?>> handlePrometheusException(PrometheusException ex) { + log.warn("Prometheus exception: status={}, message={}", ex.getStatusCode(), ex.getMessage()); + return ResponseEntity.status(ex.getStatusCode()) + .body(Result.error(ex.getStatusCode(), ex.getMessage())); + } + + @ExceptionHandler(MethodArgumentNotValidException.class) + @ResponseStatus(HttpStatus.BAD_REQUEST) + public Result<?> handleValidationException(MethodArgumentNotValidException ex) { + String message = ex.getBindingResult().getFieldErrors().stream() + .findFirst() + .map(error -> error.getDefaultMessage() == null ? "Invalid request" : error.getDefaultMessage()) + .orElse("Invalid request"); + return Result.error(HttpStatus.BAD_REQUEST.value(), message); + } + + @ExceptionHandler(HttpMessageNotReadableException.class) + @ResponseStatus(HttpStatus.BAD_REQUEST) + public Result<?> handleHttpMessageNotReadableException(HttpMessageNotReadableException ex) { + log.warn("Invalid request body"); + return Result.error(HttpStatus.BAD_REQUEST.value(), "Invalid request body"); + } + @ExceptionHandler(Exception.class) @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR) public Result<?> handleException(Exception ex) { diff --git a/server/src/main/resources/application.yml b/server/src/main/resources/application.yml index 3c98b4c6..a213118d 100644 --- a/server/src/main/resources/application.yml +++ b/server/src/main/resources/application.yml @@ -12,3 +12,13 @@ springdoc: path: /api-docs swagger-ui: path: /swagger-ui.html + +studio: + metrics: + prometheus: + base-url: ${STUDIO_METRICS_PROMETHEUS_BASE_URL:} + connect-timeout: ${STUDIO_METRICS_PROMETHEUS_CONNECT_TIMEOUT:3s} + read-timeout: ${STUDIO_METRICS_PROMETHEUS_READ_TIMEOUT:10s} + username: ${STUDIO_METRICS_PROMETHEUS_USERNAME:} + password: ${STUDIO_METRICS_PROMETHEUS_PASSWORD:} + bearer-token: ${STUDIO_METRICS_PROMETHEUS_BEARER_TOKEN:} diff --git a/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsControllerTest.java b/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsControllerTest.java new file mode 100644 index 00000000..70b64bcf --- /dev/null +++ b/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsControllerTest.java @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.rocketmq.studio.cluster.metrics; + +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc; +import org.springframework.boot.test.autoconfigure.web.servlet.WebMvcTest; +import org.springframework.boot.test.mock.mockito.MockBean; +import org.springframework.http.MediaType; +import org.springframework.test.web.servlet.MockMvc; + +import static org.mockito.Mockito.verifyNoInteractions; +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; + +@WebMvcTest(MetricsController.class) +@AutoConfigureMockMvc(addFilters = false) +class MetricsControllerTest { + + @Autowired + private MockMvc mockMvc; + + @MockBean + private MetricsService metricsService; + + @Test + void queryShouldReturnBadRequestWhenFieldTypeIsInvalid() throws Exception { + mockMvc.perform(post("/api/metrics/query") + .contentType(MediaType.APPLICATION_JSON) + .content(""" + {"metric":"up","start":"abc","end":123,"step":"30s"} + """)) + .andExpect(status().isBadRequest()) + .andExpect(jsonPath("$.code").value(400)) + .andExpect(jsonPath("$.message").value("Invalid request body")); + + verifyNoInteractions(metricsService); + } +} diff --git a/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsServiceTest.java b/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsServiceTest.java index b6897758..a8f7118a 100644 --- a/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsServiceTest.java +++ b/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsServiceTest.java @@ -22,9 +22,9 @@ import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; @@ -48,22 +48,21 @@ class MetricsServiceTest { .end(1700003600L) .step("1m") .build(); - List<long[]> values = Arrays.asList( - new long[]{1700000000L, 45}, - new long[]{1700000060L, 52}, - new long[]{1700000120L, 48} + List<MetricDataVO.MetricSampleVO> values = List.of( + sample(1700000000L, "45.5"), + sample(1700000060L, "52"), + sample(1700000120L, "48") ); - MetricDataVO data = MetricDataVO.builder().metric("cpu_usage").values(values).build(); + MetricDataVO data = metricData("cpu_usage", values); when(metricsSource.query(query)).thenReturn(data); MetricDataVO result = metricsService.query(query); - assertThat(result.getMetric()).isEqualTo("cpu_usage"); - assertThat(result.getValues()).hasSize(3); - assertThat(result.getValues().get(0)[0]).isEqualTo(1700000000L); - assertThat(result.getValues().get(0)[1]).isEqualTo(45); - assertThat(result.getValues().get(1)[1]).isEqualTo(52); - assertThat(result.getValues().get(2)[1]).isEqualTo(48); + assertThat(result.getSeries()).hasSize(1); + assertThat(result.getSeries().get(0).getLabels()).containsEntry("__name__", "cpu_usage"); + assertThat(result.getSeries().get(0).getValues()).hasSize(3); + assertThat(result.getSeries().get(0).getValues().get(0).getTimestamp()).isEqualTo(1700000000D); + assertThat(result.getSeries().get(0).getValues().get(0).getValue()).isEqualTo("45.5"); verify(metricsSource).query(query); } @@ -75,13 +74,12 @@ class MetricsServiceTest { .end(1700003600L) .step("5m") .build(); - MetricDataVO data = MetricDataVO.builder().metric("disk_io").values(Collections.emptyList()).build(); + MetricDataVO data = emptyMetricData(); when(metricsSource.query(query)).thenReturn(data); MetricDataVO result = metricsService.query(query); - assertThat(result.getMetric()).isEqualTo("disk_io"); - assertThat(result.getValues()).isEmpty(); + assertThat(result.getSeries()).isEmpty(); } @Test @@ -92,7 +90,7 @@ class MetricsServiceTest { .end(1700086400L) .step("1h") .build(); - MetricDataVO data = MetricDataVO.builder().metric("tps").values(Collections.emptyList()).build(); + MetricDataVO data = emptyMetricData(); when(metricsSource.query(any(MetricQueryDTO.class))).thenReturn(data); metricsService.query(query); @@ -104,7 +102,7 @@ class MetricsServiceTest { void queryShouldHandleVariousStepSizes() { MetricQueryDTO query15s = MetricQueryDTO.builder().metric("cpu").start(1L).end(2L).step("15s").build(); MetricQueryDTO query1h = MetricQueryDTO.builder().metric("cpu").start(1L).end(2L).step("1h").build(); - MetricDataVO data = MetricDataVO.builder().metric("cpu").values(Collections.emptyList()).build(); + MetricDataVO data = emptyMetricData(); when(metricsSource.query(any(MetricQueryDTO.class))).thenReturn(data); MetricDataVO result15s = metricsService.query(query15s); @@ -124,20 +122,19 @@ class MetricsServiceTest { .end(1700003600L) .step("1m") .build(); - List<long[]> values = Arrays.asList( - new long[]{1700000000L, 72}, - new long[]{1700000060L, 73}, - new long[]{1700000120L, 71}, - new long[]{1700000180L, 74}, - new long[]{1700000240L, 75} + List<MetricDataVO.MetricSampleVO> values = List.of( + sample(1700000000L, "72"), + sample(1700000060L, "73"), + sample(1700000120L, "71"), + sample(1700000180L, "74"), + sample(1700000240L, "75") ); - MetricDataVO data = MetricDataVO.builder().metric("memory_usage").values(values).build(); + MetricDataVO data = metricData("memory_usage", values); when(metricsSource.query(query)).thenReturn(data); MetricDataVO result = metricsService.query(query); - assertThat(result.getMetric()).isEqualTo("memory_usage"); - assertThat(result.getValues()).hasSize(5); + assertThat(result.getSeries().get(0).getValues()).hasSize(5); } @Test @@ -145,12 +142,39 @@ class MetricsServiceTest { String[] metrics = {"rocketmq_tps", "rocketmq_latency_p99", "broker_disk_usage", "consumer_lag"}; for (String metricName : metrics) { MetricQueryDTO query = MetricQueryDTO.builder().metric(metricName).start(1L).end(2L).step("1m").build(); - MetricDataVO data = MetricDataVO.builder().metric(metricName).values(Collections.emptyList()).build(); + MetricDataVO data = metricData(metricName, Collections.emptyList()); when(metricsSource.query(query)).thenReturn(data); MetricDataVO result = metricsService.query(query); - assertThat(result.getMetric()).isEqualTo(metricName); + assertThat(result.getSeries().get(0).getLabels()).containsEntry("__name__", metricName); } } + + private MetricDataVO emptyMetricData() { + return MetricDataVO.builder() + .resultType("matrix") + .series(Collections.emptyList()) + .warnings(Collections.emptyList()) + .build(); + } + + private MetricDataVO metricData(String metricName, List<MetricDataVO.MetricSampleVO> values) { + MetricDataVO.MetricSeriesVO series = MetricDataVO.MetricSeriesVO.builder() + .labels(Map.of("__name__", metricName)) + .values(values) + .build(); + return MetricDataVO.builder() + .resultType("matrix") + .series(List.of(series)) + .warnings(Collections.emptyList()) + .build(); + } + + private MetricDataVO.MetricSampleVO sample(double timestamp, String value) { + return MetricDataVO.MetricSampleVO.builder() + .timestamp(timestamp) + .value(value) + .build(); + } } diff --git a/server/src/test/java/com/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java b/server/src/test/java/com/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java new file mode 100644 index 00000000..3d2fbb26 --- /dev/null +++ b/server/src/test/java/com/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java @@ -0,0 +1,321 @@ +/* + * 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 com.rocketmq.studio.cluster.metrics; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpStatus; +import org.springframework.web.client.RestClient; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Base64; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class PrometheusMetricsSourceTest { + + private HttpServer server; + private String baseUrl; + + @BeforeEach + void setUp() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + baseUrl = "http://127.0.0.1:"; + server.getAddress().getPort(); + server.start(); + } + + @AfterEach + void tearDown() { + server.stop(0); + } + + @Test + void queryShouldPreserveSeriesLabelsDecimalValuesAndWarnings() { + AtomicReference<String> requestBody = new AtomicReference<>(); + AtomicReference<String> requestMethod = new AtomicReference<>(); + AtomicReference<String> contentType = new AtomicReference<>(); + server.createContext("/api/v1/query_range", exchange -> { + requestMethod.set(exchange.getRequestMethod()); + contentType.set(exchange.getRequestHeaders().getFirst("Content-Type")); + requestBody.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + respond(exchange, 200, """ + { + "status": "success", + "data": { + "resultType": "matrix", + "result": [ + { + "metric": {"node_id": "broker-a", "cluster": "cluster-a"}, + "values": [[1784107658, "0.30000000000000004"], [1784107688, "NaN"]] + }, + { + "metric": {"node_id": "broker-b", "cluster": "cluster-a"}, + "values": [[1784107658.5, "1.25"]] + } + ] + }, + "warnings": ["partial response"] + } + """); + }); + + PrometheusMetricsSource source = source(Duration.ofSeconds(2)); + MetricDataVO result = source.query(query()); + + assertThat(result.getResultType()).isEqualTo("matrix"); + assertThat(result.getSeries()).hasSize(2); + assertThat(result.getSeries().get(0).getLabels()) + .containsEntry("node_id", "broker-a") + .containsEntry("cluster", "cluster-a"); + assertThat(result.getSeries().get(0).getValues().get(0).getValue()) + .isEqualTo("0.30000000000000004"); + assertThat(result.getSeries().get(0).getValues().get(1).getValue()).isEqualTo("NaN"); + assertThat(result.getSeries().get(1).getValues().get(0).getTimestamp()).isEqualTo(1784107658.5D); + assertThat(result.getWarnings()).containsExactly("partial response"); + + assertThat(requestMethod.get()).isEqualTo("POST"); + assertThat(contentType.get()).startsWith("application/x-www-form-urlencoded"); + String decodedBody = URLDecoder.decode(requestBody.get(), StandardCharsets.UTF_8); + assertThat(decodedBody).contains("query=sum(rate(rocketmq_messages_in_total[1m]))"); + assertThat(decodedBody).contains("start=1784107658"); + assertThat(decodedBody).contains("end=1784108558"); + assertThat(decodedBody).contains("step=30s"); + } + + @Test + void queryShouldPreserveHistogramOnlySeries() { + server.createContext("/api/v1/query_range", exchange -> respond(exchange, 200, """ + { + "status": "success", + "data": { + "resultType": "matrix", + "result": [{ + "metric": {"__name__": "rocketmq_rpc_latency"}, + "histograms": [[1784107658, { + "count": "12", + "sum": "3.5", + "buckets": [[3, "-0.5", "0.5", "4"], [0, "0.5", "+Inf", "8"]] + }]] + }] + } + } + """)); + + MetricDataVO.MetricSeriesVO series = source(Duration.ofSeconds(2)).query(query()).getSeries().get(0); + + assertThat(series.getValues()).isEmpty(); + assertThat(series.getHistograms()).hasSize(1); + assertThat(series.getHistograms().get(0).getTimestamp()).isEqualTo(1784107658D); + assertThat(series.getHistograms().get(0).getHistogram().path("count").asText()).isEqualTo("12"); + assertThat(series.getHistograms().get(0).getHistogram().path("sum").asText()).isEqualTo("3.5"); + assertThat(series.getHistograms().get(0).getHistogram().path("buckets")).hasSize(2); + } + + @Test + void queryShouldPreserveFloatAndHistogramSamplesInSameSeries() { + server.createContext("/api/v1/query_range", exchange -> respond(exchange, 200, """ + { + "status": "success", + "data": { + "resultType": "matrix", + "result": [{ + "metric": {"__name__": "request_duration_seconds"}, + "values": [[1784107658, "1.25"]], + "histograms": [[1784107688, { + "count": "2", + "sum": "1.5", + "buckets": [[3, "-0.5", "0.5", "2"]] + }]] + }] + } + } + """)); + + MetricDataVO.MetricSeriesVO series = source(Duration.ofSeconds(2)).query(query()).getSeries().get(0); + + assertThat(series.getValues()).hasSize(1); + assertThat(series.getValues().get(0).getValue()).isEqualTo("1.25"); + assertThat(series.getHistograms()).hasSize(1); + assertThat(series.getHistograms().get(0).getHistogram().path("count").asText()).isEqualTo("2"); + } + + @Test + void queryShouldApplyBasicAuthentication() { + AtomicReference<String> authorization = new AtomicReference<>(); + server.createContext("/api/v1/query_range", exchange -> { + authorization.set(exchange.getRequestHeaders().getFirst("Authorization")); + respond(exchange, 200, successResponse()); + }); + PrometheusProperties properties = properties(Duration.ofSeconds(2)); + properties.setUsername("studio"); + properties.setPassword("secret"); + + source(properties).query(query()); + + String credentials = Base64.getEncoder().encodeToString("studio:secret".getBytes(StandardCharsets.UTF_8)); + assertThat(authorization.get()).isEqualTo("Basic " + credentials); + } + + @Test + void queryShouldPreferBearerAuthentication() { + AtomicReference<String> authorization = new AtomicReference<>(); + server.createContext("/api/v1/query_range", exchange -> { + authorization.set(exchange.getRequestHeaders().getFirst("Authorization")); + respond(exchange, 200, successResponse()); + }); + PrometheusProperties properties = properties(Duration.ofSeconds(2)); + properties.setBearerToken("test-token"); + properties.setUsername("ignored-user"); + properties.setPassword("ignored-password"); + + source(properties).query(query()); + + assertThat(authorization.get()).isEqualTo("Bearer test-token"); + } + + @Test + void queryShouldExposePrometheusErrorDetails() { + server.createContext("/api/v1/query_range", exchange -> respond(exchange, 422, """ + {"status":"error","errorType":"execution","error":"invalid expression"} + """)); + + PrometheusMetricsSource source = source(Duration.ofSeconds(2)); + + assertThatThrownBy(() -> source.query(query())) + .isInstanceOf(PrometheusException.class) + .satisfies(exception -> { + PrometheusException prometheusException = (PrometheusException) exception; + assertThat(prometheusException.getStatusCode()).isEqualTo(422); + assertThat(prometheusException.getMessage()) + .isEqualTo("Prometheus query failed (execution): invalid expression"); + }); + } + + @Test + void queryShouldRejectMalformedPrometheusResponse() { + server.createContext("/api/v1/query_range", exchange -> respond(exchange, 200, """ + {"status":"success","data":{"resultType":"matrix","result":{}}} + """)); + + assertThatThrownBy(() -> source(Duration.ofSeconds(2)).query(query())) + .isInstanceOf(PrometheusException.class) + .satisfies(exception -> assertThat(((PrometheusException) exception).getStatusCode()) + .isEqualTo(HttpStatus.BAD_GATEWAY.value())) + .hasMessage("Prometheus returned a malformed response"); + } + + @Test + void queryShouldRejectEndEarlierThanStart() { + MetricQueryDTO invalidQuery = MetricQueryDTO.builder() + .metric("up") + .start(2L) + .end(1L) + .step("30s") + .build(); + + assertThatThrownBy(() -> source(Duration.ofSeconds(2)).query(invalidQuery)) + .isInstanceOf(PrometheusException.class) + .satisfies(exception -> assertThat(((PrometheusException) exception).getStatusCode()) + .isEqualTo(HttpStatus.BAD_REQUEST.value())) + .hasMessage("Metric query end must not be earlier than start"); + } + + @Test + void queryShouldFailLoudWhenPrometheusIsNotConfigured() { + PrometheusProperties properties = new PrometheusProperties(); + PrometheusMetricsSource source = new PrometheusMetricsSource( + RestClient.builder(), new ObjectMapper(), properties); + + assertThatThrownBy(() -> source.query(query())) + .isInstanceOf(PrometheusException.class) + .satisfies(exception -> assertThat(((PrometheusException) exception).getStatusCode()) + .isEqualTo(HttpStatus.SERVICE_UNAVAILABLE.value())) + .hasMessage("Prometheus base URL is not configured"); + } + + @Test + void queryShouldReportReadTimeout() { + server.createContext("/api/v1/query_range", exchange -> { + try { + Thread.sleep(300); + respond(exchange, 200, "{\"status\":\"success\",\"data\":{\"resultType\":\"matrix\",\"result\":[]}}"); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + } catch (IOException ignored) { + // The client closes the exchange after the expected timeout. + } + }); + + PrometheusMetricsSource source = source(Duration.ofMillis(50)); + + assertThatThrownBy(() -> source.query(query())) + .isInstanceOf(PrometheusException.class) + .satisfies(exception -> { + PrometheusException prometheusException = (PrometheusException) exception; + assertThat(prometheusException.getStatusCode()) + .isEqualTo(HttpStatus.GATEWAY_TIMEOUT.value()); + }) + .hasMessage("Prometheus query timed out"); + } + + private PrometheusMetricsSource source(Duration readTimeout) { + return source(properties(readTimeout)); + } + + private PrometheusMetricsSource source(PrometheusProperties properties) { + return new PrometheusMetricsSource(RestClient.builder(), new ObjectMapper(), properties); + } + + private PrometheusProperties properties(Duration readTimeout) { + PrometheusProperties properties = new PrometheusProperties(); + properties.setBaseUrl(baseUrl); + properties.setConnectTimeout(Duration.ofSeconds(1)); + properties.setReadTimeout(readTimeout); + return properties; + } + + private MetricQueryDTO query() { + return MetricQueryDTO.builder() + .metric("sum(rate(rocketmq_messages_in_total[1m]))") + .start(1784107658L) + .end(1784108558L) + .step("30s") + .build(); + } + + private void respond(HttpExchange exchange, int statusCode, String body) throws IOException { + byte[] response = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(statusCode, response.length); + exchange.getResponseBody().write(response); + exchange.close(); + } + + private String successResponse() { + return "{\"status\":\"success\",\"data\":{\"resultType\":\"matrix\",\"result\":[]}}"; + } +} diff --git a/web/src/api/metrics.ts b/web/src/api/metrics.ts index ddc0cc18..d0b71a92 100644 --- a/web/src/api/metrics.ts +++ b/web/src/api/metrics.ts @@ -40,6 +40,41 @@ export interface DashboardData { clusters: ClusterOverview[]; } +export interface MetricSample { + timestamp: number; + value: string; +} + +export interface MetricHistogram { + count: string; + sum: string; + buckets: [number, string, string, string][]; +} + +export interface MetricHistogramSample { + timestamp: number; + histogram: MetricHistogram; +} + +export interface MetricSeries { + labels: Record<string, string>; + values: MetricSample[]; + histograms: MetricHistogramSample[]; +} + +export interface MetricData { + resultType: string; + series: MetricSeries[]; + warnings: string[]; +} + +export interface MetricQuery { + metric: string; + start: number; + end: number; + step: string; +} + // ─── Dashboard ────────────────────────────────────────────────── export async function getDashboard() { const res = await client.get<{ data: DashboardData }>('/dashboard'); @@ -47,12 +82,7 @@ export async function getDashboard() { } // ─── Metrics ──────────────────────────────────────────────────── -export async function queryMetrics(query: { - metric: string; - start: number; - end: number; - step?: string; -}) { - const res = await client.post<{ data: unknown }>('/metrics/query', query); +export async function queryMetrics(query: MetricQuery) { + const res = await client.post<{ data: MetricData }>('/metrics/query', query); return res.data.data; }
