From 9728fad5b70322fa9bfb6a6100d21b2127fd93cd Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 24 Jul 2026 01:43:56 -0700 Subject: [PATCH] feat: add metadata list AI tools --- .../ai/tool/ConsumerGroupListToolHandler.java | 88 ++++++++++ .../ops/ai/tool/TopicListToolHandler.java | 82 ++++++++++ .../resources/tool-catalog/rmq-tools.yaml | 122 ++++++++++++++ .../studio/ops/ai/tool/ToolCatalogTest.java | 6 +- .../ops/ai/tool/ToolGatewayServiceTest.java | 154 +++++++++++++++++- 5 files changed, 444 insertions(+), 8 deletions(-) create mode 100644 server/src/main/java/com/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandler.java create mode 100644 server/src/main/java/com/rocketmq/studio/ops/ai/tool/TopicListToolHandler.java diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandler.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandler.java new file mode 100644 index 00000000..00e75e6e --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/ConsumerGroupListToolHandler.java @@ -0,0 +1,88 @@ +/* + * 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.ai.tool; + +import com.rocketmq.studio.instance.group.ConsumerGroupVO; +import com.rocketmq.studio.instance.topic.MetadataService; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Component; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +@Component +@RequiredArgsConstructor +public class ConsumerGroupListToolHandler implements ToolHandler { + + private static final String NAME = "rmq.group.list"; + + private final MetadataService metadataService; + + @Override + public String name() { + return NAME; + } + + @Override + public Object execute(Map input) { + String clusterId = (String) input.get("cluster"); + String search = (String) input.get("search"); + return metadataService.listConsumerGroups(clusterId, search).stream() + .map(ConsumerGroupListToolHandler::safeProjection) + .toList(); + } + + private static Map safeProjection(ConsumerGroupVO group) { + Map result = new LinkedHashMap<>(); + result.put("name", require(group.getName(), "name")); + result.put("namespace", blankIfNull(group.getNamespace())); + result.put("clusterId", blankIfNull(group.getClusterId())); + result.put("subscriptionMode", requiredEnumName( + group.getSubscriptionMode(), "subscriptionMode", group.getName())); + result.put("consumeType", requiredEnumName( + group.getConsumeType(), "consumeType", group.getName())); + result.put("onlineInstances", group.getOnlineInstances()); + result.put("totalLag", group.getTotalLag()); + result.put("subscribedTopics", copyList(group.getSubscribedTopics())); + result.put("retryMaxTimes", group.getRetryMaxTimes()); + return result; + } + + private static String requiredEnumName(Enum value, String field, String groupName) { + if (value == null) { + throw new IllegalStateException("Consumer group " + field + + " is unavailable: " + groupName); + } + return value.name(); + } + + private static String require(String value, String field) { + if (value == null || value.isBlank()) { + throw new IllegalStateException("Consumer group " + field + " is unavailable"); + } + return value; + } + + private static List copyList(List value) { + return value == null ? List.of() : List.copyOf(value); + } + + private static String blankIfNull(String value) { + return value == null ? "" : value; + } +} diff --git a/server/src/main/java/com/rocketmq/studio/ops/ai/tool/TopicListToolHandler.java b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/TopicListToolHandler.java new file mode 100644 index 00000000..57aac568 --- /dev/null +++ b/server/src/main/java/com/rocketmq/studio/ops/ai/tool/TopicListToolHandler.java @@ -0,0 +1,82 @@ +/* + * 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.ai.tool; + +import com.rocketmq.studio.instance.topic.MetadataService; +import com.rocketmq.studio.instance.topic.TopicVO; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Component; + +import java.util.LinkedHashMap; +import java.util.Map; + +@Component +@RequiredArgsConstructor +public class TopicListToolHandler implements ToolHandler { + + private static final String NAME = "rmq.topic.list"; + + private final MetadataService metadataService; + + @Override + public String name() { + return NAME; + } + + @Override + public Object execute(Map input) { + String clusterId = (String) input.get("cluster"); + String type = (String) input.get("type"); + String search = (String) input.get("search"); + return metadataService.listTopics(clusterId, type, search).stream() + .map(TopicListToolHandler::safeProjection) + .toList(); + } + + private static Map safeProjection(TopicVO topic) { + Map result = new LinkedHashMap<>(); + result.put("name", require(topic.getName(), "name")); + result.put("namespace", blankIfNull(topic.getNamespace())); + result.put("clusterId", blankIfNull(topic.getClusterId())); + result.put("type", requiredEnumName(topic.getType(), "type", topic.getName())); + result.put("writeQueues", topic.getWriteQueues()); + result.put("readQueues", topic.getReadQueues()); + result.put("perm", requiredEnumName(topic.getPerm(), "perm", topic.getName())); + result.put("messageCount", topic.getMessageCount()); + result.put("tps", topic.getTps()); + result.put("consumerGroupCount", topic.getConsumerGroupCount()); + return result; + } + + private static String requiredEnumName(Enum value, String field, String topicName) { + if (value == null) { + throw new IllegalStateException("Topic " + field + " is unavailable: " + topicName); + } + return value.name(); + } + + private static String require(String value, String field) { + if (value == null || value.isBlank()) { + throw new IllegalStateException("Topic " + field + " is unavailable"); + } + return value; + } + + private static String blankIfNull(String value) { + return value == null ? "" : value; + } +} diff --git a/server/src/main/resources/tool-catalog/rmq-tools.yaml b/server/src/main/resources/tool-catalog/rmq-tools.yaml index 4327667c..ea089e54 100644 --- a/server/src/main/resources/tool-catalog/rmq-tools.yaml +++ b/server/src/main/resources/tool-catalog/rmq-tools.yaml @@ -74,3 +74,125 @@ tools: type: string viewHint: object deprecated: false + - name: rmq.topic.list + cli: + resource: topic + verb: list + description: List RocketMQ topics in one Studio cluster. + riskLevel: L1 + permission: topic:read + requiredCapabilities: + - REMOTING + inputSchema: + type: object + required: + - cluster + additionalProperties: false + properties: + cluster: + type: string + minLength: 1 + type: + type: string + minLength: 1 + search: + type: string + minLength: 1 + outputSchema: + type: array + items: + type: object + required: + - name + - namespace + - clusterId + - type + - writeQueues + - readQueues + - perm + - messageCount + - tps + - consumerGroupCount + additionalProperties: false + properties: + name: + type: string + namespace: + type: string + clusterId: + type: string + type: + type: string + writeQueues: + type: integer + readQueues: + type: integer + perm: + type: string + messageCount: + type: integer + tps: + type: number + consumerGroupCount: + type: integer + viewHint: table + deprecated: false + - name: rmq.group.list + cli: + resource: group + verb: list + description: List RocketMQ consumer groups in one Studio cluster. + riskLevel: L1 + permission: group:read + requiredCapabilities: + - REMOTING + inputSchema: + type: object + required: + - cluster + additionalProperties: false + properties: + cluster: + type: string + minLength: 1 + search: + type: string + minLength: 1 + outputSchema: + type: array + items: + type: object + required: + - name + - namespace + - clusterId + - subscriptionMode + - consumeType + - onlineInstances + - totalLag + - subscribedTopics + - retryMaxTimes + additionalProperties: false + properties: + name: + type: string + namespace: + type: string + clusterId: + type: string + subscriptionMode: + type: string + consumeType: + type: string + onlineInstances: + type: integer + totalLag: + type: integer + subscribedTopics: + type: array + items: + type: string + retryMaxTimes: + type: integer + viewHint: table + deprecated: false diff --git a/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolCatalogTest.java b/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolCatalogTest.java index 285965b8..43bd71cd 100644 --- a/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolCatalogTest.java +++ b/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolCatalogTest.java @@ -38,7 +38,11 @@ void loadsAndIndexesTheCanonicalCatalog() { assertThat(catalog.getMinimumClientVersion()).isEqualTo("1.0.0"); assertThat(catalog.getDigest()).matches("[0-9a-f]{64}"); assertThat(catalog.list()).extracting(ToolDefinition::getName) - .containsExactly("rmq.cluster.list", "rmq.capabilities"); + .containsExactly( + "rmq.cluster.list", + "rmq.capabilities", + "rmq.topic.list", + "rmq.group.list"); assertThat(catalog.find("rmq.cluster.list")).isPresent(); assertThat(catalog.find("rmq.unknown")).isEmpty(); } diff --git a/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java b/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java index 7d7cb37c..66c0c38c 100644 --- a/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java +++ b/server/src/test/java/com/rocketmq/studio/ops/ai/tool/ToolGatewayServiceTest.java @@ -21,7 +21,14 @@ import com.rocketmq.studio.cluster.broker.ClusterVO; import com.rocketmq.studio.common.domain.enums.ClusterStatus; import com.rocketmq.studio.common.domain.enums.ClusterType; +import com.rocketmq.studio.common.domain.enums.ConsumeType; +import com.rocketmq.studio.common.domain.enums.SubscriptionMode; +import com.rocketmq.studio.common.domain.enums.TopicPerm; +import com.rocketmq.studio.common.domain.enums.TopicType; import com.rocketmq.studio.common.exception.BusinessException; +import com.rocketmq.studio.instance.group.ConsumerGroupVO; +import com.rocketmq.studio.instance.topic.MetadataService; +import com.rocketmq.studio.instance.topic.TopicVO; import com.rocketmq.studio.ops.ai.AiToolVO; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -43,19 +50,30 @@ class ToolGatewayServiceTest { private ToolCatalog catalog; private ClusterService clusterService; + private MetadataService metadataService; private CapabilityResolver capabilityResolver; private ClusterListToolHandler clusterListHandler; private CapabilitiesToolHandler capabilitiesHandler; + private TopicListToolHandler topicListHandler; + private ConsumerGroupListToolHandler consumerGroupListHandler; private ToolGatewayService gateway; @BeforeEach void setUp() { catalog = canonicalCatalog(); clusterService = mock(ClusterService.class); + metadataService = mock(MetadataService.class); capabilityResolver = new CapabilityResolver(clusterService); clusterListHandler = new ClusterListToolHandler(clusterService); capabilitiesHandler = new CapabilitiesToolHandler(clusterService, capabilityResolver); - gateway = gateway(catalog, clusterListHandler, capabilitiesHandler); + topicListHandler = new TopicListToolHandler(metadataService); + consumerGroupListHandler = new ConsumerGroupListToolHandler(metadataService); + gateway = gateway( + catalog, + clusterListHandler, + capabilitiesHandler, + topicListHandler, + consumerGroupListHandler); } @Test @@ -72,7 +90,11 @@ void discoveryWithClusterExposesRegisteredSupportedTools() { assertThat(gateway.discover("cluster-v5")) .extracting(AiToolVO::getName) - .containsExactly("rmq.cluster.list", "rmq.capabilities"); + .containsExactly( + "rmq.cluster.list", + "rmq.capabilities", + "rmq.topic.list", + "rmq.group.list"); } @Test @@ -152,6 +174,73 @@ void rejectsClusterWithMissingType() { .hasMessageContaining("Cluster type is unavailable"); } + @Test + void executesTopicListThroughADataMinimizingProjection() { + when(clusterService.getCluster("cluster-v5")).thenReturn(cluster(ClusterType.V5_PROXY_CLUSTER)); + TopicVO topic = topic(); + topic.setRemark("do-not-expose"); + when(metadataService.listTopics("cluster-v5", "NORMAL", "order")) + .thenReturn(List.of(topic)); + + Object output = gateway.execute("rmq.topic.list", Map.of( + "cluster", "cluster-v5", + "type", "NORMAL", + "search", "order")); + + assertThat(output).isEqualTo(List.of(Map.of( + "name", "order-topic", + "namespace", "default", + "clusterId", "cluster-v5", + "type", "NORMAL", + "writeQueues", 8, + "readQueues", 8, + "perm", "RW", + "messageCount", 1200L, + "tps", 23.5D, + "consumerGroupCount", 3))); + assertThat(output.toString()).doesNotContain("do-not-expose"); + } + + @Test + void rejectsTopicListWithoutAClusterBeforeHandlerRuns() { + assertThatThrownBy(() -> gateway.execute("rmq.topic.list", Map.of("type", "NORMAL"))) + .isInstanceOf(BusinessException.class) + .hasMessageContaining("input validation failed"); + verifyNoInteractions(metadataService); + } + + @Test + void executesConsumerGroupListThroughADataMinimizingProjection() { + when(clusterService.getCluster("cluster-v5")).thenReturn(cluster(ClusterType.V5_PROXY_CLUSTER)); + ConsumerGroupVO group = consumerGroup(); + group.setDelaySeconds(30); + when(metadataService.listConsumerGroups("cluster-v5", "order")).thenReturn(List.of(group)); + + Object output = gateway.execute("rmq.group.list", Map.of( + "cluster", "cluster-v5", + "search", "order")); + + assertThat(output).isEqualTo(List.of(Map.of( + "name", "cg-order", + "namespace", "default", + "clusterId", "cluster-v5", + "subscriptionMode", "Push", + "consumeType", "CLUSTERING", + "onlineInstances", 2, + "totalLag", 42L, + "subscribedTopics", List.of("order-topic"), + "retryMaxTimes", 16))); + assertThat(output.toString()).doesNotContain("delaySeconds"); + } + + @Test + void rejectsConsumerGroupListWithoutAClusterBeforeHandlerRuns() { + assertThatThrownBy(() -> gateway.execute("rmq.group.list", Map.of())) + .isInstanceOf(BusinessException.class) + .hasMessageContaining("input validation failed"); + verifyNoInteractions(metadataService); + } + @Test void rejectsCapabilitiesExecutionWhenClusterTypeIsMissing() { when(clusterService.getCluster("unknown")).thenReturn(cluster("unknown", null)); @@ -201,7 +290,12 @@ void refusesNonL1CatalogEntriesEvenWhenAHandlerIsRegistered() throws IOException ToolCatalog l2Catalog = ToolCatalog.load( new ByteArrayResource(yaml.getBytes(StandardCharsets.UTF_8)), new ClassPathResource("tool-catalog/rmq-tools.schema.json")); - ToolGatewayService l2Gateway = gateway(l2Catalog, clusterListHandler, capabilitiesHandler); + ToolGatewayService l2Gateway = gateway( + l2Catalog, + clusterListHandler, + capabilitiesHandler, + topicListHandler, + consumerGroupListHandler); assertThatThrownBy(() -> l2Gateway.execute("rmq.cluster.list", Map.of())) .isInstanceOf(BusinessException.class) @@ -212,7 +306,12 @@ void refusesNonL1CatalogEntriesEvenWhenAHandlerIsRegistered() throws IOException @Test void failsStartupForDuplicateHandlerNames() { assertThatThrownBy(() -> gateway( - catalog, clusterListHandler, clusterListHandler, capabilitiesHandler)) + catalog, + clusterListHandler, + clusterListHandler, + capabilitiesHandler, + topicListHandler, + consumerGroupListHandler)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("duplicate handler"); } @@ -241,7 +340,11 @@ void failsStartupWhenToolSchemaContainsAnUnresolvedReference() throws IOExceptio new ClassPathResource("tool-catalog/rmq-tools.schema.json")); assertThatThrownBy(() -> gateway( - invalidCatalog, clusterListHandler, capabilitiesHandler)) + invalidCatalog, + clusterListHandler, + capabilitiesHandler, + topicListHandler, + consumerGroupListHandler)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("input schema") .hasMessageContaining("rmq.cluster.list"); @@ -263,7 +366,11 @@ void failsStartupWhenToolSchemaViolatesTheJsonSchemaMetaSchema() throws IOExcept new ClassPathResource("tool-catalog/rmq-tools.schema.json")); assertThatThrownBy(() -> gateway( - invalidCatalog, clusterListHandler, capabilitiesHandler)) + invalidCatalog, + clusterListHandler, + capabilitiesHandler, + topicListHandler, + consumerGroupListHandler)) .isInstanceOf(IllegalStateException.class) .hasMessageContaining("input schema") .hasMessageContaining("rmq.cluster.list"); @@ -283,7 +390,11 @@ public Object execute(Map input) { } }; ToolGatewayService invalidGateway = gateway( - catalog, invalidClusterListHandler, capabilitiesHandler); + catalog, + invalidClusterListHandler, + capabilitiesHandler, + topicListHandler, + consumerGroupListHandler); assertThatThrownBy(() -> invalidGateway.execute("rmq.cluster.list", Map.of())) .isInstanceOf(IllegalStateException.class) @@ -323,4 +434,33 @@ private static ClusterVO cluster(String id, ClusterType type) { cluster.setId(id); return cluster; } + + private static TopicVO topic() { + TopicVO topic = new TopicVO(); + topic.setName("order-topic"); + topic.setNamespace("default"); + topic.setClusterId("cluster-v5"); + topic.setType(TopicType.NORMAL); + topic.setWriteQueues(8); + topic.setReadQueues(8); + topic.setPerm(TopicPerm.RW); + topic.setMessageCount(1200L); + topic.setTps(23.5D); + topic.setConsumerGroupCount(3); + return topic; + } + + private static ConsumerGroupVO consumerGroup() { + ConsumerGroupVO group = new ConsumerGroupVO(); + group.setName("cg-order"); + group.setNamespace("default"); + group.setClusterId("cluster-v5"); + group.setSubscriptionMode(SubscriptionMode.Push); + group.setConsumeType(ConsumeType.CLUSTERING); + group.setOnlineInstances(2); + group.setTotalLag(42L); + group.setSubscribedTopics(List.of("order-topic")); + group.setRetryMaxTimes(16); + return group; + } }