From ad77856dd0858ff2a9218ab4091e9fc1aee94654 Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 24 Jul 2026 00:59:05 -0700 Subject: [PATCH] feat: add RocketMQ metric profiles --- .../studio/cluster/metrics/MetricProfile.java | 50 ++++++++ .../cluster/metrics/MetricProfileService.java | 106 +++++++++++++++++ .../cluster/metrics/MetricProfileVO.java | 48 ++++++++ .../cluster/metrics/MetricsController.java | 13 +++ .../cluster/metrics/SemanticMetric.java | 49 ++++++++ .../metrics/MetricProfileServiceTest.java | 110 ++++++++++++++++++ .../metrics/MetricsControllerTest.java | 30 +++++ 7 files changed, 406 insertions(+) create mode 100644 server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfile.java create mode 100644 server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfileService.java create mode 100644 server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfileVO.java create mode 100644 server/src/main/java/com/rocketmq/studio/cluster/metrics/SemanticMetric.java create mode 100644 server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricProfileServiceTest.java diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfile.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfile.java new file mode 100644 index 00000000..1d0d6e3b --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfile.java @@ -0,0 +1,50 @@ +/* + * 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; + +public enum MetricProfile { + ROCKETMQ_4_EXPORTER( + "rocketmq4-exporter", + "RocketMQ 4.x Exporter", + "RocketMQ 4.x clusters scraped through the standalone rocketmq-exporter"), + ROCKETMQ_5_NATIVE( + "rocketmq5-native", + "RocketMQ 5.x Native", + "RocketMQ 5.1+ native OpenTelemetry and Prometheus metrics"); + + private final String id; + private final String displayName; + private final String description; + + MetricProfile(String id, String displayName, String description) { + this.id = id; + this.displayName = displayName; + this.description = description; + } + + public String getId() { + return id; + } + + public String getDisplayName() { + return displayName; + } + + public String getDescription() { + return description; + } +} diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfileService.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfileService.java new file mode 100644 index 00000000..d01af4da --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfileService.java @@ -0,0 +1,106 @@ +/* + * 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.springframework.stereotype.Service; + +import java.util.List; + +@Service +public class MetricProfileService { + + public List listProfiles() { + return List.of( + profile(MetricProfile.ROCKETMQ_4_EXPORTER, rocketmq4ExporterMetrics()), + profile(MetricProfile.ROCKETMQ_5_NATIVE, rocketmq5NativeMetrics()) + ); + } + + private MetricProfileVO profile(MetricProfile profile, + List metrics) { + return MetricProfileVO.builder() + .id(profile.getId()) + .name(profile.getDisplayName()) + .description(profile.getDescription()) + .metrics(metrics) + .build(); + } + + private List rocketmq4ExporterMetrics() { + return List.of( + mapping(SemanticMetric.MESSAGE_IN_TPS, "rocketmq_broker_tps", + "sum(rocketmq_broker_tps) by (cluster, broker)", + "cluster", "broker"), + mapping(SemanticMetric.MESSAGE_OUT_TPS, "rocketmq_consumer_tps", + "sum(rocketmq_consumer_tps) by (cluster, group, topic)", + "cluster", "group", "topic"), + mapping(SemanticMetric.THROUGHPUT_IN, "rocketmq_producer_message_size", + "sum(rocketmq_producer_message_size) by (cluster, topic)", + "cluster", "topic"), + mapping(SemanticMetric.THROUGHPUT_OUT, "rocketmq_consumer_message_size", + "sum(rocketmq_consumer_message_size) by (cluster, group, topic)", + "cluster", "group", "topic"), + mapping(SemanticMetric.CONSUMER_LAG_MESSAGES, "rocketmq_message_accumulation", + "sum(rocketmq_message_accumulation) by (cluster, group, topic)", + "cluster", "group", "topic"), + mapping(SemanticMetric.CONSUMER_LAG_LATENCY, "rocketmq_group_get_latency_by_storetime", + "max(rocketmq_group_get_latency_by_storetime) by (cluster, group, topic)", + "cluster", "group", "topic"), + mapping(SemanticMetric.BROKER_HEALTH, "up", + "min(up{job=~\".*rocketmq.*\"}) by (job, instance)", + "job", "instance") + ); + } + + private List rocketmq5NativeMetrics() { + return List.of( + mapping(SemanticMetric.MESSAGE_IN_TPS, "rocketmq_messages_in_total", + "sum(rate(rocketmq_messages_in_total[1m])) by (cluster, node_id)", + "cluster", "node_id", "topic", "message_type"), + mapping(SemanticMetric.MESSAGE_OUT_TPS, "rocketmq_messages_out_total", + "sum(rate(rocketmq_messages_out_total[1m])) by (cluster, node_id, consumer_group)", + "cluster", "node_id", "topic", "consumer_group"), + mapping(SemanticMetric.THROUGHPUT_IN, "rocketmq_throughput_in_total", + "sum(rate(rocketmq_throughput_in_total[1m])) by (cluster, node_id)", + "cluster", "node_id", "topic", "message_type"), + mapping(SemanticMetric.THROUGHPUT_OUT, "rocketmq_throughput_out_total", + "sum(rate(rocketmq_throughput_out_total[1m])) by (cluster, node_id, consumer_group)", + "cluster", "node_id", "topic", "consumer_group"), + mapping(SemanticMetric.CONSUMER_LAG_MESSAGES, "rocketmq_consumer_lag_messages", + "sum(rocketmq_consumer_lag_messages) by (cluster, topic, consumer_group)", + "cluster", "topic", "consumer_group"), + mapping(SemanticMetric.CONSUMER_LAG_LATENCY, "rocketmq_consumer_lag_latency", + "max(rocketmq_consumer_lag_latency) by (cluster, topic, consumer_group)", + "cluster", "topic", "consumer_group"), + mapping(SemanticMetric.BROKER_HEALTH, "rocketmq_processor_watermark", + "max(rocketmq_processor_watermark) by (cluster, node_id, processor)", + "cluster", "node_id", "processor") + ); + } + + private MetricProfileVO.MetricMappingVO mapping(SemanticMetric semanticMetric, String prometheusMetric, + String promql, String... labels) { + return MetricProfileVO.MetricMappingVO.builder() + .semanticMetric(semanticMetric.getKey()) + .name(semanticMetric.getDisplayName()) + .unit(semanticMetric.getUnit()) + .prometheusMetric(prometheusMetric) + .promql(promql) + .labels(List.of(labels)) + .build(); + } +} diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfileVO.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfileVO.java new file mode 100644 index 00000000..1f3fe0fe --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/MetricProfileVO.java @@ -0,0 +1,48 @@ +/* + * 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 lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.List; + +@Data +@Builder +@NoArgsConstructor +@AllArgsConstructor +public class MetricProfileVO { + private String id; + private String name; + private String description; + private List metrics; + + @Data + @Builder + @NoArgsConstructor + @AllArgsConstructor + public static class MetricMappingVO { + private String semanticMetric; + private String name; + private String unit; + private String prometheusMetric; + private String promql; + private List labels; + } +} 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 6ee8334f..1b3d3cea 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 @@ -22,17 +22,30 @@ import io.swagger.v3.oas.annotations.responses.ApiResponses; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; +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.RestController; +import java.util.List; + @RestController @RequestMapping("/api/metrics") @RequiredArgsConstructor public class MetricsController { private final MetricsService metricsService; + private final MetricProfileService metricProfileService; + + @Operation(summary = "List RocketMQ metric profiles", + description = "Returns version-aware semantic PromQL mappings for RocketMQ metrics") + @ApiResponse(responseCode = "200", description = "Metric profiles listed successfully", + useReturnTypeSchema = true) + @GetMapping("/profiles") + public Result> listProfiles() { + return Result.ok(metricProfileService.listProfiles()); + } @Operation(summary = "Query Prometheus range metrics", description = "Executes a PromQL range query against the configured Prometheus server") diff --git a/server/src/main/java/com/rocketmq/studio/cluster/metrics/SemanticMetric.java b/server/src/main/java/com/rocketmq/studio/cluster/metrics/SemanticMetric.java new file mode 100644 index 00000000..2130e8a2 --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/cluster/metrics/SemanticMetric.java @@ -0,0 +1,49 @@ +/* + * 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; + +public enum SemanticMetric { + MESSAGE_IN_TPS("message_in_tps", "Message In TPS", "messages/s"), + MESSAGE_OUT_TPS("message_out_tps", "Message Out TPS", "messages/s"), + THROUGHPUT_IN("throughput_in", "Throughput In", "bytes/s"), + THROUGHPUT_OUT("throughput_out", "Throughput Out", "bytes/s"), + CONSUMER_LAG_MESSAGES("consumer_lag_messages", "Consumer Lag", "messages"), + CONSUMER_LAG_LATENCY("consumer_lag_latency", "Consumer Lag Latency", "ms"), + BROKER_HEALTH("broker_health", "Broker Health", "up"); + + private final String key; + private final String displayName; + private final String unit; + + SemanticMetric(String key, String displayName, String unit) { + this.key = key; + this.displayName = displayName; + this.unit = unit; + } + + public String getKey() { + return key; + } + + public String getDisplayName() { + return displayName; + } + + public String getUnit() { + return unit; + } +} diff --git a/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricProfileServiceTest.java b/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricProfileServiceTest.java new file mode 100644 index 00000000..fc0ec8eb --- /dev/null +++ b/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricProfileServiceTest.java @@ -0,0 +1,110 @@ +/* + * 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 java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +class MetricProfileServiceTest { + + private final MetricProfileService service = new MetricProfileService(); + + @Test + void listProfilesShouldExposeRocketmq4And5Mappings() { + Map profiles = service.listProfiles().stream() + .collect(Collectors.toMap(MetricProfileVO::getId, Function.identity())); + + assertThat(profiles.keySet()).containsExactlyInAnyOrder("rocketmq4-exporter", "rocketmq5-native"); + assertThat(semanticMetrics(profiles.get("rocketmq4-exporter"))) + .containsExactlyInAnyOrderElementsOf(allSemanticMetricKeys()); + assertThat(semanticMetrics(profiles.get("rocketmq5-native"))) + .containsExactlyInAnyOrderElementsOf(allSemanticMetricKeys()); + } + + @Test + void rocketmq5ProfileShouldUseNativeMetricNames() { + MetricProfileVO profile = findProfile("rocketmq5-native"); + + assertThat(mapping(profile, SemanticMetric.MESSAGE_IN_TPS)) + .extracting(MetricProfileVO.MetricMappingVO::getPrometheusMetric, + MetricProfileVO.MetricMappingVO::getPromql) + .containsExactly("rocketmq_messages_in_total", + "sum(rate(rocketmq_messages_in_total[1m])) by (cluster, node_id)"); + assertThat(mapping(profile, SemanticMetric.CONSUMER_LAG_MESSAGES).getPrometheusMetric()) + .isEqualTo("rocketmq_consumer_lag_messages"); + assertThat(mapping(profile, SemanticMetric.BROKER_HEALTH).getPrometheusMetric()) + .isEqualTo("rocketmq_processor_watermark"); + } + + @Test + void rocketmq4ProfileShouldUseExporterMetricNames() { + MetricProfileVO profile = findProfile("rocketmq4-exporter"); + + assertThat(mapping(profile, SemanticMetric.MESSAGE_IN_TPS).getPrometheusMetric()) + .isEqualTo("rocketmq_broker_tps"); + assertThat(mapping(profile, SemanticMetric.THROUGHPUT_IN).getPrometheusMetric()) + .isEqualTo("rocketmq_producer_message_size"); + assertThat(mapping(profile, SemanticMetric.THROUGHPUT_OUT).getPrometheusMetric()) + .isEqualTo("rocketmq_consumer_message_size"); + assertThat(mapping(profile, SemanticMetric.CONSUMER_LAG_MESSAGES).getPrometheusMetric()) + .isEqualTo("rocketmq_message_accumulation"); + assertThat(mapping(profile, SemanticMetric.CONSUMER_LAG_LATENCY).getPrometheusMetric()) + .isEqualTo("rocketmq_group_get_latency_by_storetime"); + } + + @Test + void mappingsShouldExposeLabelsAndUnitsForDashboardRendering() { + MetricProfileVO profile = findProfile("rocketmq5-native"); + MetricProfileVO.MetricMappingVO messageOut = mapping(profile, SemanticMetric.MESSAGE_OUT_TPS); + + assertThat(messageOut.getLabels()).contains("cluster", "node_id", "topic", "consumer_group"); + assertThat(messageOut.getUnit()).isEqualTo("messages/s"); + assertThat(messageOut.getName()).isEqualTo("Message Out TPS"); + } + + private MetricProfileVO findProfile(String id) { + return service.listProfiles().stream() + .filter(profile -> profile.getId().equals(id)) + .findFirst() + .orElseThrow(); + } + + private MetricProfileVO.MetricMappingVO mapping(MetricProfileVO profile, SemanticMetric semanticMetric) { + return profile.getMetrics().stream() + .filter(metric -> metric.getSemanticMetric().equals(semanticMetric.getKey())) + .findFirst() + .orElseThrow(); + } + + private List semanticMetrics(MetricProfileVO profile) { + return profile.getMetrics().stream() + .map(MetricProfileVO.MetricMappingVO::getSemanticMetric) + .toList(); + } + + private List allSemanticMetricKeys() { + return List.of(SemanticMetric.values()).stream() + .map(SemanticMetric::getKey) + .toList(); + } +} 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 index 70b64bcf..43f987b9 100644 --- a/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsControllerTest.java +++ b/server/src/test/java/com/rocketmq/studio/cluster/metrics/MetricsControllerTest.java @@ -24,7 +24,11 @@ import org.springframework.http.MediaType; import org.springframework.test.web.servlet.MockMvc; +import java.util.List; + +import static org.mockito.Mockito.when; import static org.mockito.Mockito.verifyNoInteractions; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get; 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; @@ -39,6 +43,32 @@ class MetricsControllerTest { @MockBean private MetricsService metricsService; + @MockBean + private MetricProfileService metricProfileService; + + @Test + void listProfilesShouldReturnVersionAwareMappings() throws Exception { + MetricProfileVO profile = MetricProfileVO.builder() + .id("rocketmq5-native") + .name("RocketMQ 5.x Native") + .metrics(List.of(MetricProfileVO.MetricMappingVO.builder() + .semanticMetric("message_in_tps") + .prometheusMetric("rocketmq_messages_in_total") + .promql("sum(rate(rocketmq_messages_in_total[1m])) by (cluster, node_id)") + .labels(List.of("cluster", "node_id")) + .build())) + .build(); + when(metricProfileService.listProfiles()).thenReturn(List.of(profile)); + + mockMvc.perform(get("/api/metrics/profiles")) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.code").value(200)) + .andExpect(jsonPath("$.data[0].id").value("rocketmq5-native")) + .andExpect(jsonPath("$.data[0].metrics[0].semanticMetric").value("message_in_tps")) + .andExpect(jsonPath("$.data[0].metrics[0].prometheusMetric") + .value("rocketmq_messages_in_total")); + } + @Test void queryShouldReturnBadRequestWhenFieldTypeIsInvalid() throws Exception { mockMvc.perform(post("/api/metrics/query")