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 73cf84aa feat: route and audit cluster runtime operations by selected
instance (#1244)
73cf84aa is described below
commit 73cf84aa3f93424f21881c9eae565667b010ce82
Author: aias00 <[email protected]>
AuthorDate: Mon Aug 10 19:05:18 2026 +0800
feat: route and audit cluster runtime operations by selected instance
(#1244)
* feat: route cluster runtime operations by selected instance
* test: preserve legacy cluster runtime paths
* test: cover instance-scoped cluster operations
* chore: include instance in cluster list logs
* [ISSUE #1262] Populate runtime audit cluster IDs
* test: bind cluster controller mocks to instance context
---
.../studio/cluster/broker/ClusterController.java | 10 +++--
.../studio/cluster/broker/ClusterProvider.java | 8 ++++
.../studio/cluster/broker/ClusterService.java | 50 ++++++++++++++++++++--
.../studio/cluster/config/UpdateConfigDTO.java | 2 +
.../rocketmq/studio/ops/audit/AuditService.java | 5 +++
.../apache/RocketMQBrokerConfigService.java | 23 ++++++++--
.../provider/apache/RocketMQClusterProvider.java | 30 ++++++++++---
.../cluster/broker/ClusterControllerTest.java | 7 +--
.../studio/cluster/broker/ClusterServiceTest.java | 42 +++++++++++++++++-
.../studio/ops/audit/AuditServiceTest.java | 10 +++++
.../apache/RocketMQBrokerConfigServiceTest.java | 23 ++++++++--
.../apache/RocketMQClusterProviderTest.java | 3 +-
web/src/api/cluster.ts | 16 ++++---
web/src/pages/cluster/index.tsx | 30 ++++++++++++-
web/src/services/clusterService.ts | 12 +++---
15 files changed, 235 insertions(+), 36 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
index 7a72a7bc..1525c12a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
@@ -28,6 +28,7 @@ import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
@@ -42,8 +43,8 @@ public class ClusterController {
private final ClusterConnectionService clusterConnectionService;
@GetMapping
- public Result<List<ClusterVO>> listClusters() {
- return Result.ok(clusterService.listClusters());
+ public Result<List<ClusterVO>> listClusters(@RequestParam(required =
false) String instanceId) {
+ return Result.ok(clusterService.listClusters(instanceId));
}
@PostMapping("/test-connection")
@@ -52,8 +53,9 @@ public class ClusterController {
}
@GetMapping("/{id}")
- public Result<ClusterVO> getCluster(@PathVariable String id) {
- return Result.ok(clusterService.getCluster(id));
+ public Result<ClusterVO> getCluster(@PathVariable String id,
+ @RequestParam(required = false) String
instanceId) {
+ return Result.ok(clusterService.getCluster(id, instanceId));
}
@PostMapping("/config/update")
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProvider.java
index c7c583f0..4ce5228c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterProvider.java
@@ -23,5 +23,13 @@ public interface ClusterProvider {
List<ClusterVO> discoverClusters();
+ default List<ClusterVO> discoverClusters(String instanceId) {
+ return discoverClusters();
+ }
+
ClusterVO refreshClusterDetail(String clusterId);
+
+ default ClusterVO refreshClusterDetail(String clusterId, String
instanceId) {
+ return refreshClusterDetail(clusterId);
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
index 8d85d49e..16570fe7 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
@@ -60,6 +60,16 @@ public class ClusterService {
return List.of();
}
+ public List<ClusterVO> listClusters(String instanceId) {
+ log.info("Listing clusters for instance: {}", instanceId);
+ List<ClusterVO> discovered =
clusterProvider.discoverClusters(instanceId);
+ if (discovered != null && !discovered.isEmpty()) {
+ discovered.forEach(cluster -> enrichWithLiveConfig(cluster,
instanceId));
+ return discovered;
+ }
+ return List.of();
+ }
+
public ClusterVO getCluster(String id) {
log.info("Getting cluster detail: {}", id);
ClusterVO live = clusterProvider.refreshClusterDetail(id);
@@ -70,17 +80,31 @@ public class ClusterService {
throw new BusinessException(503, "Cluster details are unavailable: " +
id);
}
+ public ClusterVO getCluster(String id, String instanceId) {
+ log.info("Getting cluster detail: {}", id);
+ ClusterVO live = clusterProvider.refreshClusterDetail(id, instanceId);
+ if (live != null) {
+ enrichWithLiveConfig(live, instanceId);
+ return live;
+ }
+ throw new BusinessException(503, "Cluster details are unavailable: " +
id);
+ }
+
/**
* Attach live broker configuration (read from the first reachable master
broker via the
* admin API) to a discovered cluster. Falls back to the persisted config,
if any, when the
* live read is unavailable.
*/
private void enrichWithLiveConfig(ClusterVO cluster) {
+ enrichWithLiveConfig(cluster, null);
+ }
+
+ private void enrichWithLiveConfig(ClusterVO cluster, String instanceId) {
if (cluster.getBrokers() != null) {
for (BrokerVO broker : cluster.getBrokers()) {
if (broker.getAddr() != null && !broker.getAddr().isEmpty()) {
try {
-
cluster.setConfig(brokerConfigService.getBrokerConfig(broker.getAddr()));
+
cluster.setConfig(brokerConfigService.getBrokerConfig(broker.getAddr(),
instanceId));
return;
} catch (Exception e) {
log.warn("Failed to read live config from broker {}:
{}",
@@ -97,7 +121,7 @@ public class ClusterService {
public ClusterConfigUpdateResultVO updateClusterConfig(UpdateConfigDTO
command) {
log.info("Updating cluster config for: {}", command.getId());
requireMatchingDefaultQueueNums(command);
- ClusterVO cluster = resolveCluster(command.getId());
+ ClusterVO cluster = resolveCluster(command.getId(),
command.getInstanceId());
ClusterConfigVO config = copyConfig(cluster.getConfig());
applyConfig(command, config);
@@ -112,7 +136,12 @@ public class ClusterService {
continue;
}
try {
- brokerConfigService.updateBrokerConfig(address,
command.getId(), brokerProps);
+ if (command.getInstanceId() == null ||
command.getInstanceId().isBlank()) {
+ brokerConfigService.updateBrokerConfig(address,
command.getId(), brokerProps);
+ } else {
+ brokerConfigService.updateBrokerConfig(
+ address, command.getId(),
command.getInstanceId(), brokerProps);
+ }
successfulBrokers.add(address);
} catch (Exception e) {
failedBrokers.add(BrokerConfigUpdateFailureVO.builder()
@@ -190,7 +219,7 @@ public class ClusterService {
.map(failure -> failure.getAddress() + ": " +
failure.getMessage())
.toList();
try {
- auditService.record("UPDATE_CLUSTER_CONFIG", "CLUSTER:" +
clusterId, detail, status.name());
+ auditService.record("UPDATE_CLUSTER_CONFIG", "CLUSTER:" +
clusterId, clusterId, detail, status.name());
} catch (Exception e) {
log.warn("Failed to record cluster config update audit for {}:
{}", clusterId, e.getMessage());
}
@@ -206,6 +235,19 @@ public class ClusterService {
.orElseThrow(() -> new BusinessException(404, "Cluster not
found: " + clusterId));
}
+ private ClusterVO resolveCluster(String clusterId, String instanceId) {
+ if (instanceId == null || instanceId.isBlank()) {
+ return resolveCluster(clusterId);
+ }
+ ClusterVO live = clusterProvider.refreshClusterDetail(clusterId,
instanceId);
+ if (live != null) {
+ enrichWithLiveConfig(live, instanceId);
+ return live;
+ }
+ return clusterRepository.findById(clusterId)
+ .orElseThrow(() -> new BusinessException(404, "Cluster not
found: " + clusterId));
+ }
+
private ClusterConfigVO copyConfig(ClusterConfigVO config) {
if (config == null) {
return new ClusterConfigVO();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/config/UpdateConfigDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/config/UpdateConfigDTO.java
index a23d0c67..b9e9e1c5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/config/UpdateConfigDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/config/UpdateConfigDTO.java
@@ -32,6 +32,8 @@ public class UpdateConfigDTO {
@NotBlank(message = "id is required")
private String id;
+ private String instanceId;
+
private String flushDiskType;
private Boolean autoCreateTopicEnable;
private Boolean autoCreateSubscriptionGroup;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditService.java
index f9a733b3..4179ffeb 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/audit/AuditService.java
@@ -85,11 +85,16 @@ public class AuditService {
public void record(String operationType, String target, String detail,
String result) {
+ record(operationType, target, null, detail, result);
+ }
+
+ public void record(String operationType, String target, String clusterId,
String detail, String result) {
AuditRecordVO record = AuditRecordVO.builder()
.timestamp(LocalDateTime.now())
.operator(AuthenticatedUserContext.currentUsernameOrSystem())
.operationType(operationType)
.target(target)
+ .clusterId(clusterId)
.detail(detail)
.result(result)
.build();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigService.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigService.java
index 44010a09..2452120d 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigService.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.provider.apache;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
import org.apache.rocketmq.studio.common.domain.enums.FlushDiskType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
@@ -35,13 +36,18 @@ public class RocketMQBrokerConfigService {
private final MqAdminExtFactory adminFactory;
private final RocketMQProperties properties;
+ private final RuntimeAdminClientResolver runtimeAdminClientResolver;
private final AuditService auditService;
/**
* Read broker config from the live broker via admin API.
*/
public ClusterConfigVO getBrokerConfig(String brokerAddr) {
- return adminFactory.execute(namesrvAddr(), null, admin -> {
+ return getBrokerConfig(brokerAddr, null);
+ }
+
+ public ClusterConfigVO getBrokerConfig(String brokerAddr, String
instanceId) {
+ return execute(instanceId, admin -> {
try {
Properties props = admin.getBrokerConfig(brokerAddr);
return mapToClusterConfigVO(props);
@@ -52,11 +58,22 @@ public class RocketMQBrokerConfigService {
});
}
+ private <T> T execute(String instanceId, MqAdminExtFactory.AdminAction<T>
action) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.execute(instanceId, action);
+ }
+ return adminFactory.execute(namesrvAddr(), null, action);
+ }
+
/**
* Update broker config on the live broker via admin API, then record
audit.
*/
public void updateBrokerConfig(String brokerAddr, String clusterId,
Properties newConfig) {
- adminFactory.execute(namesrvAddr(), null, admin -> {
+ updateBrokerConfig(brokerAddr, clusterId, null, newConfig);
+ }
+
+ public void updateBrokerConfig(String brokerAddr, String clusterId, String
instanceId, Properties newConfig) {
+ execute(instanceId, admin -> {
try {
admin.updateBrokerConfig(brokerAddr, newConfig);
String detail = "brokerAddr=" + brokerAddr + ", config=" +
newConfig;
@@ -74,7 +91,7 @@ public class RocketMQBrokerConfigService {
private void recordAudit(String clusterId, String detail, String result) {
try {
- auditService.record("UPDATE_BROKER_CONFIG", "CLUSTER:" +
clusterId, detail, result);
+ auditService.record("UPDATE_BROKER_CONFIG", "CLUSTER:" +
clusterId, clusterId, detail, result);
} catch (Exception auditFailure) {
log.warn("Failed to record broker config audit for cluster {}:
{}", clusterId,
auditFailure.getMessage());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
index a5615753..51d333d5 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
@@ -23,6 +23,7 @@ import org.apache.rocketmq.studio.cluster.broker.BrokerVO;
import org.apache.rocketmq.studio.cluster.broker.ClusterProvider;
import org.apache.rocketmq.studio.cluster.broker.ClusterVO;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.cluster.nameserver.NameServerVO;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.common.domain.enums.BrokerStatus;
@@ -52,10 +53,16 @@ public class RocketMQClusterProvider implements
ClusterProvider {
private final MqAdminExtFactory adminFactory;
private final RocketMQProperties properties;
+ private final RuntimeAdminClientResolver runtimeAdminClientResolver;
@Override
public List<ClusterVO> discoverClusters() {
- String namesrvAddr = properties.getNamesrvAddr();
+ return discoverClusters(null);
+ }
+
+ @Override
+ public List<ClusterVO> discoverClusters(String instanceId) {
+ String namesrvAddr = resolveNamesrvAddr(instanceId);
if (!StringUtils.hasText(namesrvAddr)) {
log.debug("NameServer address not configured, returning empty
cluster list");
return Collections.emptyList();
@@ -77,7 +84,7 @@ public class RocketMQClusterProvider implements
ClusterProvider {
Set<String> brokerNames = entry.getValue();
List<BrokerVO> brokers = buildBrokerList(admin,
brokerNames, brokerAddrTable);
- List<NameServerVO> nameServers = buildNameServerList();
+ List<NameServerVO> nameServers =
buildNameServerList(namesrvAddr);
ClusterVO cluster = buildClusterVO(clusterName, brokers,
nameServers);
clusters.add(cluster);
@@ -93,7 +100,12 @@ public class RocketMQClusterProvider implements
ClusterProvider {
@Override
public ClusterVO refreshClusterDetail(String clusterId) {
- String namesrvAddr = properties.getNamesrvAddr();
+ return refreshClusterDetail(clusterId, null);
+ }
+
+ @Override
+ public ClusterVO refreshClusterDetail(String clusterId, String instanceId)
{
+ String namesrvAddr = resolveNamesrvAddr(instanceId);
if (!StringUtils.hasText(namesrvAddr)) {
log.debug("NameServer address not configured, cannot refresh
cluster detail");
return null;
@@ -115,7 +127,7 @@ public class RocketMQClusterProvider implements
ClusterProvider {
}
List<BrokerVO> brokers = buildBrokerList(admin, brokerNames,
brokerAddrTable);
- List<NameServerVO> nameServers = buildNameServerList();
+ List<NameServerVO> nameServers =
buildNameServerList(namesrvAddr);
return buildClusterVO(clusterId, brokers, nameServers);
});
@@ -253,9 +265,15 @@ public class RocketMQClusterProvider implements
ClusterProvider {
return 0;
}
- private List<NameServerVO> buildNameServerList() {
+ private String resolveNamesrvAddr(String instanceId) {
+ if (StringUtils.hasText(instanceId)) {
+ return runtimeAdminClientResolver.resolveEndpoint(instanceId);
+ }
+ return properties.getNamesrvAddr();
+ }
+
+ private List<NameServerVO> buildNameServerList(String namesrvAddr) {
List<NameServerVO> nameServers = new ArrayList<>();
- String namesrvAddr = properties.getNamesrvAddr();
if (namesrvAddr == null || namesrvAddr.isEmpty()) {
return nameServers;
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
index 2bd3b758..c5d0d6cd 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
@@ -41,6 +41,7 @@ import java.util.Collections;
import java.util.stream.Stream;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -111,7 +112,7 @@ class ClusterControllerTest {
void listClustersShouldReturnAllClusters() throws Exception {
ClusterVO cluster1 = buildCluster("cluster-1", "production-cluster",
ClusterStatus.healthy);
ClusterVO cluster2 = buildCluster("cluster-2", "staging-cluster",
ClusterStatus.warning);
- when(clusterService.listClusters()).thenReturn(Arrays.asList(cluster1,
cluster2));
+
when(clusterService.listClusters(isNull())).thenReturn(Arrays.asList(cluster1,
cluster2));
mockMvc.perform(get("/api/clusters"))
.andExpect(status().isOk())
@@ -128,7 +129,7 @@ class ClusterControllerTest {
@Test
void listClustersShouldReturnEmptyArrayWhenNoClusters() throws Exception {
-
when(clusterService.listClusters()).thenReturn(Collections.emptyList());
+
when(clusterService.listClusters(isNull())).thenReturn(Collections.emptyList());
mockMvc.perform(get("/api/clusters"))
.andExpect(status().isOk())
@@ -147,7 +148,7 @@ class ClusterControllerTest {
.maxMessageSize(4194304)
.autoCreateTopicEnable(true)
.build());
- when(clusterService.getCluster("cluster-1")).thenReturn(cluster);
+ when(clusterService.getCluster("cluster-1", null)).thenReturn(cluster);
mockMvc.perform(get("/api/clusters/cluster-1"))
.andExpect(status().isOk())
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
index ca9f193f..f5fba221 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
@@ -132,6 +132,17 @@ class ClusterServiceTest {
verify(clusterRepository, never()).findAll();
}
+ @Test
+ void listClustersShouldUseSelectedInstance() {
+
when(clusterProvider.discoverClusters("instance-1")).thenReturn(List.of(sampleCluster));
+
+ List<ClusterVO> result = clusterService.listClusters("instance-1");
+
+ assertThat(result).containsExactly(sampleCluster);
+ verify(clusterProvider).discoverClusters("instance-1");
+ verify(clusterProvider, never()).discoverClusters();
+ }
+
@Test
void
updateClusterConfigShouldRejectDifferentDefaultReadAndWriteQueueNums() {
UpdateConfigDTO command = UpdateConfigDTO.builder()
@@ -169,6 +180,17 @@ class ClusterServiceTest {
assertThat(result.getType()).isEqualTo(ClusterType.V5_PROXY_CLUSTER);
}
+ @Test
+ void getClusterShouldUseSelectedInstance() {
+ when(clusterProvider.refreshClusterDetail("cluster-1",
"instance-1")).thenReturn(sampleCluster);
+
+ ClusterVO result = clusterService.getCluster("cluster-1",
"instance-1");
+
+ assertThat(result).isSameAs(sampleCluster);
+ verify(clusterProvider).refreshClusterDetail("cluster-1",
"instance-1");
+ verify(clusterProvider, never()).refreshClusterDetail("cluster-1");
+ }
+
@Test
void getClusterShouldThrowWhenNotFound() {
when(clusterProvider.refreshClusterDetail("nonexistent")).thenReturn(null);
@@ -198,7 +220,7 @@ class ClusterServiceTest {
void updateConfigShouldSucceedWhenAuditRecordingFails() {
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
doThrow(new IllegalStateException("audit storage
unavailable")).when(auditService)
- .record(any(), any(), any(), any());
+ .record(any(), any(), any(), any(), any());
ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(UpdateConfigDTO.builder()
.id("cluster-1")
@@ -294,10 +316,28 @@ class ClusterServiceTest {
verify(auditService).record(
eq("UPDATE_CLUSTER_CONFIG"),
eq("CLUSTER:cluster-1"),
+ eq("cluster-1"),
org.mockito.ArgumentMatchers.contains("10.0.0.2:10911"),
eq("PARTIAL"));
}
+ @Test
+ void updateConfigShouldUseSelectedInstanceForBrokerUpdates() {
+
sampleCluster.setBrokers(List.of(BrokerVO.builder().name("broker-0").addr("10.0.0.1:10911").build()));
+ when(clusterProvider.refreshClusterDetail("cluster-1",
"instance-1")).thenReturn(sampleCluster);
+
+ ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(UpdateConfigDTO.builder()
+ .id("cluster-1")
+ .instanceId("instance-1")
+ .writeQueueNums(16)
+ .build());
+
+
assertThat(result.getStatus()).isEqualTo(ClusterConfigUpdateResultVO.Status.SUCCESS);
+ verify(clusterProvider).refreshClusterDetail("cluster-1",
"instance-1");
+ verify(brokerConfigService).updateBrokerConfig(
+ eq("10.0.0.1:10911"), eq("cluster-1"), eq("instance-1"),
any());
+ }
+
@Test
void updateConfigShouldThrowWhenClusterNotFound() {
when(clusterRepository.findById("missing")).thenReturn(Optional.empty());
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/audit/AuditServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/audit/AuditServiceTest.java
index 6ac8ace3..43593b19 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/audit/AuditServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/audit/AuditServiceTest.java
@@ -63,6 +63,16 @@ class AuditServiceTest {
assertThat(captor.getValue().getOperator()).isEqualTo("operator-user");
}
+ @Test
+ void recordShouldPreserveClusterIdWhenProvided() {
+ auditService.record("UPDATE_CLUSTER_CONFIG", "CLUSTER:prod-cn",
"prod-cn",
+ "updated broker config", "SUCCESS");
+
+ ArgumentCaptor<AuditRecordVO> captor =
ArgumentCaptor.forClass(AuditRecordVO.class);
+ verify(auditRepository).save(captor.capture());
+ assertThat(captor.getValue().getClusterId()).isEqualTo("prod-cn");
+ }
+
@Test
void queryLogsDelegatesPaginationAndFiltersToRepository() {
AuditRecordVO record =
AuditRecordVO.builder().operationType("CREATE").build();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigServiceTest.java
index 83b8fe4d..0c85b1ba 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigServiceTest.java
@@ -6,6 +6,7 @@
*/
package org.apache.rocketmq.studio.provider.apache;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.studio.ops.audit.AuditService;
@@ -24,6 +25,7 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.verify;
@ExtendWith(MockitoExtension.class)
class RocketMQBrokerConfigServiceTest {
@@ -36,6 +38,8 @@ class RocketMQBrokerConfigServiceTest {
private DefaultMQAdminExt adminExt;
@Mock
private AuditService auditService;
+ @Mock
+ private RuntimeAdminClientResolver runtimeAdminClientResolver;
private RocketMQBrokerConfigService brokerConfigService;
@@ -44,7 +48,8 @@ class RocketMQBrokerConfigServiceTest {
lenient().when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
lenient().when(adminFactory.execute(anyString(), any(),
any())).thenAnswer(invocation ->
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(2).apply(adminExt));
- brokerConfigService = new RocketMQBrokerConfigService(adminFactory,
properties, auditService);
+ brokerConfigService = new RocketMQBrokerConfigService(
+ adminFactory, properties, runtimeAdminClientResolver,
auditService);
}
@Test
@@ -53,7 +58,7 @@ class RocketMQBrokerConfigServiceTest {
config.setProperty("flushDiskType", "ASYNC_FLUSH");
doNothing().when(adminExt).updateBrokerConfig("broker-a:10911",
config);
doThrow(new IllegalStateException("audit db down")).when(auditService)
- .record(anyString(), anyString(), anyString(), anyString());
+ .record(anyString(), anyString(), anyString(), anyString(),
anyString());
brokerConfigService.updateBrokerConfig("broker-a:10911", "cluster-a",
config);
}
@@ -64,10 +69,22 @@ class RocketMQBrokerConfigServiceTest {
doThrow(new IllegalStateException("broker unavailable")).when(adminExt)
.updateBrokerConfig("broker-a:10911", config);
doThrow(new IllegalStateException("audit db down")).when(auditService)
- .record(anyString(), anyString(), anyString(), anyString());
+ .record(anyString(), anyString(), anyString(), anyString(),
anyString());
assertThatThrownBy(() ->
brokerConfigService.updateBrokerConfig("broker-a:10911", "cluster-a", config))
.isInstanceOf(BusinessException.class)
.hasMessage("Failed to update broker config: broker
unavailable");
}
+
+ @Test
+ void updateRecordsStructuredClusterId() throws Exception {
+ Properties config = new Properties();
+ doNothing().when(adminExt).updateBrokerConfig("broker-a:10911",
config);
+
+ brokerConfigService.updateBrokerConfig("broker-a:10911", "cluster-a",
config);
+
+ verify(auditService).record(
+ "UPDATE_BROKER_CONFIG", "CLUSTER:cluster-a", "cluster-a",
+ "brokerAddr=broker-a:10911, config={}", "SUCCESS");
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
index bf9d20ec..7cbf83b2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
@@ -21,6 +21,7 @@ import org.apache.rocketmq.remoting.protocol.body.KVTable;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.studio.cluster.broker.ClusterVO;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.cluster.broker.RuntimeAdminClientResolver;
import org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
@@ -113,7 +114,7 @@ class RocketMQClusterProviderTest {
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(2).apply(adminExt));
RocketMQProperties properties = new RocketMQProperties();
properties.setNamesrvAddr("10.0.0.1:9876");
- return new RocketMQClusterProvider(adminFactory, properties);
+ return new RocketMQClusterProvider(adminFactory, properties,
mock(RuntimeAdminClientResolver.class));
}
@Test
diff --git a/web/src/api/cluster.ts b/web/src/api/cluster.ts
index 0de78ce7..b7851da9 100644
--- a/web/src/api/cluster.ts
+++ b/web/src/api/cluster.ts
@@ -135,8 +135,10 @@ export interface NameServerConfigDiffResult {
}
// ─── Cluster ────────────────────────────────────────────────────
-export async function listClusters() {
- const res = await client.get<{ data: ClusterInfo[] }>('/clusters');
+export async function listClusters(instanceId?: string) {
+ const res = await client.get<{ data: ClusterInfo[] }>('/clusters', {
+ params: instanceId ? { instanceId } : undefined,
+ });
return res.data.data;
}
@@ -147,12 +149,16 @@ export async function testClusterConnection(namesrvAddr:
string) {
return res.data.data;
}
-export async function getCluster(id: string) {
- const res = await client.get<{ data: ClusterInfo
}>(`/clusters/${pathSegment(id)}`);
+export async function getCluster(id: string, instanceId?: string) {
+ const res = await client.get<{ data: ClusterInfo
}>(`/clusters/${pathSegment(id)}`, {
+ params: instanceId ? { instanceId } : undefined,
+ });
return res.data.data;
}
-export async function updateClusterConfig(data: { id: string } &
Partial<ClusterConfig>) {
+export async function updateClusterConfig(
+ data: { id: string; instanceId?: string } & Partial<ClusterConfig>,
+) {
const res = await client.post<{ data: ClusterConfigUpdateResult }>(
'/clusters/config/update',
data,
diff --git a/web/src/pages/cluster/index.tsx b/web/src/pages/cluster/index.tsx
index 0db8c857..c4088809 100644
--- a/web/src/pages/cluster/index.tsx
+++ b/web/src/pages/cluster/index.tsx
@@ -64,6 +64,8 @@ import {
updateClusterConfig,
updateNameServer,
} from '../../services/clusterService';
+import { listInstances } from '../../services/instanceService';
+import type { Instance } from '../../api/instance';
const { Text } = Typography;
@@ -83,6 +85,8 @@ const compareText = (left: string | null | undefined, right:
string | null | und
const ClusterPage = () => {
const { t } = useLang();
const [clusters, setClusters] = useState<ClusterInfo[]>([]);
+ const [instances, setInstances] = useState<Instance[]>([]);
+ const [selectedInstanceId, setSelectedInstanceId] = useState('');
const [loading, setLoading] = useState(true);
const [nsSearch, setNsSearch] = useState('');
const [brokerSearch, setBrokerSearch] = useState('');
@@ -148,6 +152,18 @@ const ClusterPage = () => {
Promise.resolve(),
);
const tRef = useRef(t);
+ const selectedInstanceIdRef = useRef('');
+
+ useEffect(() => {
+ void listInstances().then((nextInstances) => {
+ const apacheInstances = nextInstances.filter((instance) =>
instance.vendor === 'APACHE');
+ setInstances(apacheInstances);
+ const initialInstanceId = apacheInstances[0]?.id ?? '';
+ selectedInstanceIdRef.current = initialInstanceId;
+ setSelectedInstanceId(initialInstanceId);
+ void requestRefreshRef.current('manual');
+ });
+ }, []);
const clearRefreshTimer = useCallback(() => {
if (refreshTimerRef.current !== null) {
@@ -177,7 +193,7 @@ const ClusterPage = () => {
while (currentSource && mountedRef.current) {
try {
- const nextClusters = await listClusters();
+ const nextClusters = await
listClusters(selectedInstanceIdRef.current || undefined);
if (!mountedRef.current) return;
setClusters(nextClusters);
setSelectedProxy((current) => {
@@ -530,6 +546,7 @@ const ClusterPage = () => {
};
const result = await updateClusterConfig({
id: selectedCluster.id,
+ instanceId: selectedInstanceIdRef.current || undefined,
...nextConfig,
});
if (result.status === 'SUCCESS') {
@@ -730,6 +747,17 @@ const ClusterPage = () => {
<div>
<Flex justify="space-between" style={{ marginBottom: 16 }}>
<Space>
+ <Select
+ value={selectedInstanceId || undefined}
+ onChange={(instanceId) => {
+ selectedInstanceIdRef.current = instanceId;
+ setSelectedInstanceId(instanceId);
+ void requestRefresh('manual');
+ }}
+ placeholder="Select instance"
+ style={{ width: 180 }}
+ options={instances.map((instance) => ({ value: instance.id,
label: instance.name }))}
+ />
<Input.Search
placeholder={t('cluster.searchNs')}
allowClear
diff --git a/web/src/services/clusterService.ts
b/web/src/services/clusterService.ts
index b15d9760..65bef100 100644
--- a/web/src/services/clusterService.ts
+++ b/web/src/services/clusterService.ts
@@ -34,11 +34,11 @@ function copyCluster(cluster: ClusterInfo): ClusterInfo {
};
}
-export async function listClusters(): Promise<ClusterInfo[]> {
+export async function listClusters(instanceId?: string):
Promise<ClusterInfo[]> {
if (isMockMode()) {
return clusters.map(copyCluster);
}
- return clusterApi.listClusters();
+ return clusterApi.listClusters(instanceId);
}
export async function testClusterConnection(namesrvAddr: string):
Promise<ClusterProbeResult> {
@@ -59,13 +59,13 @@ export async function testClusterConnection(namesrvAddr:
string): Promise<Cluste
return clusterApi.testClusterConnection(namesrvAddr);
}
-export async function getCluster(id: string): Promise<ClusterInfo> {
+export async function getCluster(id: string, instanceId?: string):
Promise<ClusterInfo> {
if (isMockMode()) {
const cluster = clusters.find((item) => item.id === id);
if (!cluster) throw new Error('Cluster not found');
return copyCluster(cluster);
}
- return clusterApi.getCluster(id);
+ return clusterApi.getCluster(id, instanceId);
}
export async function getNameServerConfigDiff(
@@ -171,7 +171,9 @@ export async function deleteK8sCert(id: string):
Promise<void> {
return clusterApi.deleteK8sCert(id);
}
-export async function updateClusterConfig(data: { id: string } &
Partial<ClusterConfig>) {
+export async function updateClusterConfig(
+ data: { id: string; instanceId?: string } & Partial<ClusterConfig>,
+) {
if (isMockMode()) {
const { id, ...config } = data;
const cluster = getMockCluster(id);
