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
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 652fcd3d fix(metrics): bound Prometheus query responses (#1006)
652fcd3d is described below
commit 652fcd3df316e48624bf447840fdf131c89159bd
Author: aias00 <[email protected]>
AuthorDate: Wed Aug 5 02:43:52 2026 -0700
fix(metrics): bound Prometheus query responses (#1006)
---
.../studio/cluster/metrics/MetricsController.java | 1 +
.../cluster/metrics/PrometheusMetricsSource.java | 62 +++++++++++++++++-----
.../metrics/PrometheusMetricsSourceTest.java | 53 ++++++++++++++++++
3 files changed, 102 insertions(+), 14 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MetricsController.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MetricsController.java
index 02b6d855..2c79b35b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MetricsController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MetricsController.java
@@ -54,6 +54,7 @@ public class MetricsController {
useReturnTypeSchema = true),
@ApiResponse(responseCode = "400", description = "Invalid request or
PromQL expression"),
@ApiResponse(responseCode = "422", description = "Prometheus could not
execute the expression"),
+ @ApiResponse(responseCode = "413", description = "Prometheus response
exceeds Studio query limits"),
@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")
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSource.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSource.java
index 913cd56c..921b7402 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSource.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSource.java
@@ -21,6 +21,7 @@ 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.HttpStatusCode;
import org.springframework.http.MediaType;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.stereotype.Component;
@@ -30,9 +31,10 @@ 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.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.InputStream;
import java.net.SocketTimeoutException;
import java.net.URI;
import java.util.Iterator;
@@ -46,6 +48,9 @@ import java.util.stream.StreamSupport;
public class PrometheusMetricsSource implements MetricsSource {
private static final String QUERY_RANGE_PATH = "/api/v1/query_range";
+ private static final int MAX_RESPONSE_BYTES = 5 * 1024 * 1024;
+ private static final int MAX_SERIES = 1_000;
+ private static final int MAX_TOTAL_SAMPLES = 100_000;
private final RestClient restClient;
private final ObjectMapper objectMapper;
@@ -80,13 +85,16 @@ public class PrometheusMetricsSource implements
MetricsSource {
.contentType(MediaType.APPLICATION_FORM_URLENCODED)
.headers(this::applyAuthentication)
.body(form)
- .retrieve()
- .body(JsonNode.class);
+ .exchange((request, clientResponse) -> {
+ JsonNode body =
objectMapper.readTree(readResponseBody(clientResponse.getBody()));
+ if (clientResponse.getStatusCode().isError()) {
+ throw responseBodyException(body,
responseStatus(clientResponse.getStatusCode()));
+ }
+ return body;
+ });
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(),
@@ -174,6 +182,7 @@ public class PrometheusMetricsSource implements
MetricsSource {
throw new PrometheusException(HttpStatus.BAD_GATEWAY.value(),
"Prometheus returned a malformed response");
}
+ validateResponseLimits(result);
List<MetricDataVO.MetricSeriesVO> series =
StreamSupport.stream(result.spliterator(), false)
.map(this::parseSeries)
@@ -250,19 +259,44 @@ public class PrometheusMetricsSource implements
MetricsSource {
.toList();
}
- private PrometheusException responseException(RestClientResponseException
exception) {
- JsonNode response = null;
- try {
- response =
objectMapper.readTree(exception.getResponseBodyAsString());
- } catch (IOException ignored) {
- log.debug("Failed to parse Prometheus error response");
+ private byte[] readResponseBody(InputStream input) throws IOException {
+ try (InputStream response = input; ByteArrayOutputStream output = new
ByteArrayOutputStream()) {
+ byte[] buffer = new byte[8 * 1024];
+ int read;
+ while ((read = response.read(buffer)) != -1) {
+ if (output.size() > MAX_RESPONSE_BYTES - read) {
+ throw new
PrometheusException(HttpStatus.PAYLOAD_TOO_LARGE.value(),
+ "Prometheus response exceeds 5 MiB; narrow the
query");
+ }
+ output.write(buffer, 0, read);
+ }
+ return output.toByteArray();
+ }
+ }
+
+ private void validateResponseLimits(JsonNode result) {
+ if (result.size() > MAX_SERIES) {
+ throw new PrometheusException(HttpStatus.PAYLOAD_TOO_LARGE.value(),
+ "Prometheus query returned too many series; narrow the
query");
}
- int upstreamStatus = exception.getStatusCode().value();
- int statusCode = switch (upstreamStatus) {
+ long totalSamples = 0;
+ for (JsonNode series : result) {
+ totalSamples += series.path("values").size();
+ totalSamples += series.path("histograms").size();
+ if (totalSamples > MAX_TOTAL_SAMPLES) {
+ throw new
PrometheusException(HttpStatus.PAYLOAD_TOO_LARGE.value(),
+ "Prometheus query returned too many samples; increase
step or narrow the query");
+ }
+ }
+ }
+
+ private int responseStatus(HttpStatusCode statusCode) {
+ int upstreamStatus = statusCode.value();
+ int mappedStatus = switch (upstreamStatus) {
case 400, 422, 503 -> upstreamStatus;
default -> HttpStatus.BAD_GATEWAY.value();
};
- return responseBodyException(response, statusCode);
+ return mappedStatus;
}
private PrometheusException responseBodyException(JsonNode response, int
statusCode) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java
index 32d0e502..ff0c0873 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/PrometheusMetricsSourceTest.java
@@ -31,6 +31,7 @@ import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Base64;
+import java.util.Collections;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
@@ -228,6 +229,40 @@ class PrometheusMetricsSourceTest {
.hasMessage("Prometheus returned a malformed response");
}
+ @Test
+ void queryShouldRejectResponseWithTooManySeries() {
+ server.createContext("/api/v1/query_range", exchange ->
respond(exchange, 200, responseWithSeries(1_001)));
+
+ assertThatThrownBy(() -> source(Duration.ofSeconds(2)).query(query()))
+ .isInstanceOf(PrometheusException.class)
+ .satisfies(exception -> assertThat(((PrometheusException)
exception).getStatusCode())
+ .isEqualTo(HttpStatus.PAYLOAD_TOO_LARGE.value()))
+ .hasMessage("Prometheus query returned too many series; narrow
the query");
+ }
+
+ @Test
+ void queryShouldRejectResponseWithTooManySamples() {
+ server.createContext("/api/v1/query_range", exchange ->
respond(exchange, 200, responseWithSamples(100_001)));
+
+ assertThatThrownBy(() -> source(Duration.ofSeconds(2)).query(query()))
+ .isInstanceOf(PrometheusException.class)
+ .satisfies(exception -> assertThat(((PrometheusException)
exception).getStatusCode())
+ .isEqualTo(HttpStatus.PAYLOAD_TOO_LARGE.value()))
+ .hasMessage("Prometheus query returned too many samples;
increase step or narrow the query");
+ }
+
+ @Test
+ void queryShouldRejectResponseLargerThanFiveMebibytes() {
+ server.createContext("/api/v1/query_range", exchange ->
respond(exchange, 200,
+ "{\"status\":\"success\",\"padding\":\"" + "x".repeat(5 * 1024
* 1024) + "\"}"));
+
+ assertThatThrownBy(() -> source(Duration.ofSeconds(2)).query(query()))
+ .isInstanceOf(PrometheusException.class)
+ .satisfies(exception -> assertThat(((PrometheusException)
exception).getStatusCode())
+ .isEqualTo(HttpStatus.PAYLOAD_TOO_LARGE.value()))
+ .hasMessage("Prometheus response exceeds 5 MiB; narrow the
query");
+ }
+
@Test
void queryShouldRejectEndEarlierThanStart() {
MetricQueryDTO invalidQuery = MetricQueryDTO.builder()
@@ -382,4 +417,22 @@ class PrometheusMetricsSourceTest {
private String successResponse() {
return
"{\"status\":\"success\",\"data\":{\"resultType\":\"matrix\",\"result\":[]}}";
}
+
+ private String responseWithSeries(int count) {
+ return
"{\"status\":\"success\",\"data\":{\"resultType\":\"matrix\",\"result\":["
+ + String.join(",", Collections.nCopies(count,
"{\"metric\":{},\"values\":[]}"))
+ + "]}}";
+ }
+
+ private String responseWithSamples(int count) {
+ StringBuilder values = new StringBuilder();
+ for (int index = 0; index < count; index++) {
+ if (index > 0) {
+ values.append(',');
+ }
+ values.append("[1784107658,\"1\"]");
+ }
+ return
"{\"status\":\"success\",\"data\":{\"resultType\":\"matrix\",\"result\":[{\"metric\":{},\"values\":["
+ + values + "]}]}}";
+ }
}
