From e41d15c5e020cc5c77611956c38f57175cc6c5ba Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 24 Jul 2026 01:07:04 -0700 Subject: [PATCH] fix: derive dashboard summary from clusters --- .../ops/dashboard/DashboardProviderStub.java | 99 ------------- .../RepositoryDashboardProvider.java | 96 +++++++++++++ .../RepositoryDashboardProviderTest.java | 132 ++++++++++++++++++ 3 files changed, 228 insertions(+), 99 deletions(-) delete mode 100644 server/src/main/java/com/rocketmq/studio/ops/dashboard/DashboardProviderStub.java create mode 100644 server/src/main/java/com/rocketmq/studio/ops/dashboard/RepositoryDashboardProvider.java create mode 100644 server/src/test/java/com/rocketmq/studio/ops/dashboard/RepositoryDashboardProviderTest.java diff --git a/server/src/main/java/com/rocketmq/studio/ops/dashboard/DashboardProviderStub.java b/server/src/main/java/com/rocketmq/studio/ops/dashboard/DashboardProviderStub.java deleted file mode 100644 index 3964ed94..00000000 --- a/server/src/main/java/com/rocketmq/studio/ops/dashboard/DashboardProviderStub.java +++ /dev/null @@ -1,99 +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 com.rocketmq.studio.ops.dashboard; - -import com.rocketmq.studio.common.domain.enums.ClusterStatus; -import com.rocketmq.studio.common.domain.enums.ClusterType; -import org.springframework.stereotype.Component; - -import java.util.Arrays; -import java.util.List; - -@Component -public class DashboardProviderStub implements DashboardProvider { - - @Override - public DashboardDataVO getDashboardData() { - DashboardStatsVO stats = DashboardStatsVO.builder() - .totalClusters(3) - .healthyClusters(2) - .totalBrokers(12) - .totalProxies(6) - .totalNameServers(6) - .totalTopics(156) - .totalConsumerGroups(89) - .totalMessagesToday(12_500_000L) - .messagesPerSecond(14_467L) - .tpsIn(7_200L) - .tpsOut(7_267L) - .build(); - - List clusters = List.of( - ClusterOverviewVO.builder() - .id("cluster-1") - .name("production-cluster") - .type(ClusterType.V5_PROXY_CLUSTER) - .status(ClusterStatus.healthy) - .brokers(6) - .proxies(3) - .topics(80) - .groups(45) - .tpsIn(4200) - .tpsOut(4300) - .version("5.1.4") - .throughput(Arrays.asList(3200, 3500, 4100, 4200, 3800, 4000, - 4500, 4200, 3900, 4100, 4300, 4200)) - .build(), - ClusterOverviewVO.builder() - .id("cluster-2") - .name("staging-cluster") - .type(ClusterType.V5_PROXY_LOCAL) - .status(ClusterStatus.warning) - .brokers(4) - .proxies(2) - .topics(50) - .groups(30) - .tpsIn(2000) - .tpsOut(1967) - .version("5.1.4") - .throughput(Arrays.asList(1800, 1900, 2100, 2000, 1700, 2200, - 2000, 1800, 1900, 2100, 2000, 1967)) - .build(), - ClusterOverviewVO.builder() - .id("cluster-3") - .name("dev-cluster") - .type(ClusterType.V4_DIRECT) - .status(ClusterStatus.healthy) - .brokers(2) - .proxies(1) - .topics(26) - .groups(14) - .tpsIn(1000) - .tpsOut(1000) - .version("4.9.8") - .throughput(Arrays.asList(800, 900, 1100, 1000, 950, 1050, - 1000, 900, 1100, 1000, 950, 1000)) - .build() - ); - - return DashboardDataVO.builder() - .stats(stats) - .clusters(clusters) - .build(); - } -} diff --git a/server/src/main/java/com/rocketmq/studio/ops/dashboard/RepositoryDashboardProvider.java b/server/src/main/java/com/rocketmq/studio/ops/dashboard/RepositoryDashboardProvider.java new file mode 100644 index 00000000..3b96ccb8 --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/dashboard/RepositoryDashboardProvider.java @@ -0,0 +1,96 @@ +/* + * 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.ops.dashboard; + +import com.rocketmq.studio.cluster.broker.BrokerVO; +import com.rocketmq.studio.cluster.broker.ClusterRepository; +import com.rocketmq.studio.cluster.broker.ClusterVO; +import com.rocketmq.studio.common.domain.enums.ClusterStatus; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Component; + +import java.util.Collection; +import java.util.List; +import java.util.Objects; + +@Component +@RequiredArgsConstructor +public class RepositoryDashboardProvider implements DashboardProvider { + + private final ClusterRepository clusterRepository; + + @Override + public DashboardDataVO getDashboardData() { + List sourceClusters = clusterRepository.findAll(); + List clusters = sourceClusters.stream() + .map(this::toClusterOverview) + .toList(); + + DashboardStatsVO stats = DashboardStatsVO.builder() + .totalClusters(clusters.size()) + .healthyClusters((int) clusters.stream() + .filter(cluster -> cluster.getStatus() == ClusterStatus.healthy) + .count()) + .totalBrokers(clusters.stream().mapToInt(ClusterOverviewVO::getBrokers).sum()) + .totalProxies(clusters.stream().mapToInt(ClusterOverviewVO::getProxies).sum()) + .totalNameServers(sourceClusters.stream() + .map(ClusterVO::getNameServers) + .filter(Objects::nonNull) + .mapToInt(Collection::size) + .sum()) + .totalTopics(clusters.stream().mapToInt(ClusterOverviewVO::getTopics).sum()) + .totalConsumerGroups(clusters.stream().mapToInt(ClusterOverviewVO::getGroups).sum()) + .totalMessagesToday(0L) + .messagesPerSecond(clusters.stream() + .mapToLong(cluster -> (long) cluster.getTpsIn() + cluster.getTpsOut()) + .sum()) + .tpsIn(clusters.stream().mapToLong(ClusterOverviewVO::getTpsIn).sum()) + .tpsOut(clusters.stream().mapToLong(ClusterOverviewVO::getTpsOut).sum()) + .build(); + + return DashboardDataVO.builder() + .stats(stats) + .clusters(clusters) + .build(); + } + + private ClusterOverviewVO toClusterOverview(ClusterVO cluster) { + List brokers = nullToEmpty(cluster.getBrokers()); + int tpsIn = brokers.stream().mapToInt(BrokerVO::getTpsIn).sum(); + int tpsOut = brokers.stream().mapToInt(BrokerVO::getTpsOut).sum(); + + return ClusterOverviewVO.builder() + .id(cluster.getId()) + .name(cluster.getName()) + .type(cluster.getType()) + .status(cluster.getStatus()) + .brokers(brokers.size()) + .proxies(nullToEmpty(cluster.getProxies()).size()) + .topics(cluster.getTopicCount()) + .groups(cluster.getGroupCount()) + .tpsIn(tpsIn) + .tpsOut(tpsOut) + .version(cluster.getVersion()) + .throughput(nullToEmpty(cluster.getTpsHistory())) + .build(); + } + + private List nullToEmpty(List values) { + return values == null ? List.of() : values; + } +} diff --git a/server/src/test/java/com/rocketmq/studio/ops/dashboard/RepositoryDashboardProviderTest.java b/server/src/test/java/com/rocketmq/studio/ops/dashboard/RepositoryDashboardProviderTest.java new file mode 100644 index 00000000..c10a27de --- /dev/null +++ b/server/src/test/java/com/rocketmq/studio/ops/dashboard/RepositoryDashboardProviderTest.java @@ -0,0 +1,132 @@ +/* + * 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.ops.dashboard; + +import com.rocketmq.studio.cluster.broker.BrokerVO; +import com.rocketmq.studio.cluster.broker.ClusterRepository; +import com.rocketmq.studio.cluster.broker.ClusterVO; +import com.rocketmq.studio.cluster.nameserver.NameServerVO; +import com.rocketmq.studio.cluster.proxy.ProxyVO; +import com.rocketmq.studio.common.domain.enums.BrokerStatus; +import com.rocketmq.studio.common.domain.enums.ClusterStatus; +import com.rocketmq.studio.common.domain.enums.ClusterType; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; + +class RepositoryDashboardProviderTest { + + @Test + void getDashboardDataShouldSummarizeRepositoryClusters() { + RepositoryDashboardProvider provider = new RepositoryDashboardProvider(new StaticClusterRepository(List.of( + cluster("cluster-001", ClusterStatus.healthy, 12, 4, List.of(10, 20), 120, 80), + cluster("cluster-002", ClusterStatus.warning, 8, 3, List.of(30), 50, 40) + ))); + + DashboardDataVO dashboard = provider.getDashboardData(); + + assertThat(dashboard.getStats().getTotalClusters()).isEqualTo(2); + assertThat(dashboard.getStats().getHealthyClusters()).isEqualTo(1); + assertThat(dashboard.getStats().getTotalBrokers()).isEqualTo(4); + assertThat(dashboard.getStats().getTotalProxies()).isEqualTo(2); + assertThat(dashboard.getStats().getTotalNameServers()).isEqualTo(4); + assertThat(dashboard.getStats().getTotalTopics()).isEqualTo(20); + assertThat(dashboard.getStats().getTotalConsumerGroups()).isEqualTo(7); + assertThat(dashboard.getStats().getTpsIn()).isEqualTo(170L); + assertThat(dashboard.getStats().getTpsOut()).isEqualTo(120L); + assertThat(dashboard.getStats().getMessagesPerSecond()).isEqualTo(290L); + assertThat(dashboard.getStats().getTotalMessagesToday()).isZero(); + assertThat(dashboard.getClusters()) + .extracting(ClusterOverviewVO::getId) + .containsExactly("cluster-001", "cluster-002"); + } + + @Test + void getDashboardDataShouldHandleMissingNodeLists() { + ClusterVO cluster = ClusterVO.builder() + .name("empty") + .type(ClusterType.V4_DIRECT) + .status(ClusterStatus.healthy) + .version("4.9.8") + .topicCount(2) + .groupCount(1) + .build(); + cluster.setId("cluster-empty"); + RepositoryDashboardProvider provider = new RepositoryDashboardProvider(new StaticClusterRepository(List.of(cluster))); + + DashboardDataVO dashboard = provider.getDashboardData(); + + assertThat(dashboard.getStats().getTotalBrokers()).isZero(); + assertThat(dashboard.getStats().getTotalProxies()).isZero(); + assertThat(dashboard.getStats().getTotalNameServers()).isZero(); + assertThat(dashboard.getClusters().getFirst().getThroughput()).isEmpty(); + assertThat(dashboard.getClusters().getFirst().getTpsIn()).isZero(); + assertThat(dashboard.getClusters().getFirst().getTpsOut()).isZero(); + } + + private ClusterVO cluster(String id, ClusterStatus status, int topics, int groups, + List tpsHistory, int firstBrokerIn, int firstBrokerOut) { + ClusterVO cluster = ClusterVO.builder() + .name("cluster-" + id) + .type(ClusterType.V5_PROXY_CLUSTER) + .status(status) + .version("5.2.0") + .brokers(List.of( + broker(firstBrokerIn, firstBrokerOut), + broker(0, 0) + )) + .proxies(List.of(ProxyVO.builder().addr(id + "-proxy").status(status).build())) + .nameServers(List.of( + NameServerVO.builder().addr(id + "-namesrv-a").status(ClusterStatus.healthy).build(), + NameServerVO.builder().addr(id + "-namesrv-b").status(ClusterStatus.healthy).build() + )) + .topicCount(topics) + .groupCount(groups) + .tpsHistory(tpsHistory) + .build(); + cluster.setId(id); + return cluster; + } + + private BrokerVO broker(int tpsIn, int tpsOut) { + return BrokerVO.builder() + .name("broker-a") + .status(BrokerStatus.running) + .tpsIn(tpsIn) + .tpsOut(tpsOut) + .build(); + } + + private record StaticClusterRepository(List clusters) implements ClusterRepository { + @Override + public List findAll() { + return clusters; + } + + @Override + public Optional findById(String id) { + return clusters.stream().filter(cluster -> cluster.getId().equals(id)).findFirst(); + } + + @Override + public void updateConfig(String clusterId, com.rocketmq.studio.cluster.config.ClusterConfigVO config) { + } + } +}