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 6604841c5 fix(cluster): harden proxy discovery, nameserver reload and
metrics collection (#2807)
6604841c5 is described below
commit 6604841c5f00eb7e1d089883dc8239bf3346d206
Author: yyqdbngt <[email protected]>
AuthorDate: Mon Aug 31 19:21:24 2026 +0800
fix(cluster): harden proxy discovery, nameserver reload and metrics
collection (#2807)
* fix(proxy): retry discovery after transient failures
* fix(proxy): reject malformed bracketed IPv6 addresses
* fix(nameserver): handle create reload races
* fix(metrics): bound the complete collection pass
* fix(metrics): isolate unscoped snapshot history
---------
Co-authored-by: Yue Wang <[email protected]>
---
.../studio/cluster/metrics/CollectorScheduler.java | 11 +++-
.../MybatisPlusMetricSnapshotRepository.java | 1 +
.../nameserver/NameserverRegistryService.java | 6 ++-
.../studio/cluster/proxy/ProxyAddressService.java | 7 +++
.../provider/apache/ProxyConsumerResolver.java | 2 +
.../cluster/metrics/CollectorSchedulerTest.java | 34 ++++++++++++
.../MybatisPlusMetricSnapshotRepositoryTest.java | 61 ++++++++++++++++++++++
.../nameserver/NameserverRegistryServiceTest.java | 18 +++++++
.../cluster/proxy/ProxyAddressServiceTest.java | 5 +-
.../provider/apache/ProxyConsumerResolverTest.java | 18 +++++++
10 files changed, 159 insertions(+), 4 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
index e20c717f6..4cdfe880a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
@@ -84,12 +84,19 @@ public class CollectorScheduler {
.filter(java.util.Objects::nonNull)
.toList();
Duration timeout =
parsePositiveDuration(properties.getCollectionTimeout(),
Duration.ofSeconds(15));
+ long passStartedAt = System.nanoTime();
+ long timeoutNanos = timeout.toNanos();
for (Future<?> job : jobs) {
try {
- job.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ long remainingNanos = timeoutNanos - (System.nanoTime() -
passStartedAt);
+ if (remainingNanos <= 0) {
+ job.cancel(true);
+ continue;
+ }
+ job.get(remainingNanos, TimeUnit.NANOSECONDS);
} catch (java.util.concurrent.TimeoutException error) {
job.cancel(true);
- log.warn("Native metric collection exceeded {} for one
instance and was cancelled", timeout);
+ log.warn("Native metric collection pass exceeded {} and
unfinished work was cancelled", timeout);
} catch (InterruptedException error) {
Thread.currentThread().interrupt();
return;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepository.java
index 18baf8105..8cdd35480 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepository.java
@@ -60,6 +60,7 @@ public class MybatisPlusMetricSnapshotRepository implements
MetricSnapshotReposi
return mapper.selectList(new QueryWrapper<RmqMetricSnapshot>()
.eq("instance_id",
scope.instanceId()).eq("metric_key", scope.metricKey())
.eq("domain", scope.domain().name()).eq("labels_hash",
sha256(labelsJson))
+ .isNull(scope.clusterId() == null, "cluster_id")
.eq(scope.clusterId() != null, "cluster_id",
scope.clusterId())
.eq("availability",
MetricAvailability.AVAILABLE.name())
.ge("collected_at", LocalDateTime.ofInstant(since,
ZoneOffset.UTC))
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryService.java
index 2d2c6e750..615499813 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryService.java
@@ -57,7 +57,11 @@ public class NameserverRegistryService {
// The unique index is the final guard against concurrent
duplicate creates.
throw duplicateName(name);
}
- return toVO(nameserverMapper.selectById(entity.getId()));
+ RmqNameserver stored = nameserverMapper.selectById(entity.getId());
+ if (stored == null) {
+ throw concurrentlyDeleted(entity.getId());
+ }
+ return toVO(stored);
}
public NameserverRegistryVO update(UpdateNameserverRegistryDTO command) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
index 14a27f9cd..145f67355 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressService.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.cluster.proxy;
+import org.apache.commons.validator.routines.InetAddressValidator;
import org.apache.rocketmq.studio.cluster.broker.ClusterService;
import org.apache.rocketmq.studio.common.exception.BusinessException;
import
org.apache.rocketmq.studio.common.util.NoRedirectClientHttpRequestFactory;
@@ -53,6 +54,7 @@ public class ProxyAddressService {
private static final Pattern PROXY_ADDR_PATTERN =
Pattern.compile("^(\\[[0-9a-fA-F:.]+]|[A-Za-z0-9._-]+):(\\d{1,5})$");
+ private static final InetAddressValidator INET_ADDRESS_VALIDATOR =
InetAddressValidator.getInstance();
private static final int MIN_PORT = 1;
private static final int MAX_PORT = 65535;
@@ -311,6 +313,11 @@ public class ProxyAddressService {
if (!matcher.matches()) {
throw new BusinessException(400, fieldName + " must be in
host:port or [ipv6]:port format");
}
+ String host = matcher.group(1);
+ if (host.startsWith("[")
+ &&
!INET_ADDRESS_VALIDATOR.isValidInet6Address(host.substring(1, host.length() -
1))) {
+ throw new BusinessException(400, fieldName + " contains a
malformed IPv6 address");
+ }
int port = Integer.parseInt(matcher.group(2));
if (port < MIN_PORT || port > MAX_PORT) {
throw new BusinessException(400, fieldName + " port must be
between 1 and 65535");
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
index 192c55c75..15d5e2ac0 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolver.java
@@ -173,6 +173,8 @@ public class ProxyConsumerResolver {
} catch (Exception e) {
log.debug("Proxy discovery via heartbeat syncer failed for
instance {}: {}",
instanceId, e.getMessage());
+ // A failed lookup is transient and must not suppress discovery
for the full cache TTL.
+ return List.of();
}
List<String> addresses = ips.stream()
.map(ip -> ip + ":" + PROXY_REMOTING_PORT)
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
index be7283447..ea4c8fdad 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
@@ -83,6 +83,40 @@ class CollectorSchedulerTest {
scheduler.stopCollectionExecutor();
}
+ @Test
+ void appliesCollectionTimeoutToTheWholePassInsteadOfEachInstanceTest() {
+ AlertingProperties properties = new AlertingProperties();
+ properties.setCollectionTimeout("PT0.25S");
+ InstanceRepository instances = mock(InstanceRepository.class);
+ when(instances.findAll()).thenReturn(List.of(
+ InstanceVO.builder().name("slow-a").build(),
+ InstanceVO.builder().name("slow-b").build(),
+ InstanceVO.builder().name("slow-c").build()));
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ when(collector.supports(any(InstanceVO.class))).thenReturn(true);
+ when(collector.collect(any(InstanceVO.class))).thenAnswer(invocation
-> {
+ try {
+ Thread.sleep(TimeUnit.SECONDS.toMillis(5));
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ }
+ return List.of();
+ });
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ when(lease.tryAcquire()).thenReturn(true);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CollectorScheduler scheduler = new CollectorScheduler(properties,
instances, List.of(collector), List.of(),
+ mock(MetricSnapshotRepository.class),
mock(NativeAlertProcessor.class), lease, executor);
+
+ long startedAt = System.nanoTime();
+ scheduler.collect();
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startedAt);
+ scheduler.stopCollectionExecutor();
+
+ assertTrue(elapsedMillis < 600,
+ "the configured timeout should cap the complete pass,
elapsed=" + elapsedMillis + "ms");
+ }
+
@Test
void collectsAndPersistsSupportedSamplesTest() {
AlertingProperties properties = new AlertingProperties();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepositoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepositoryTest.java
new file mode 100644
index 000000000..7d89fa996
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepositoryTest.java
@@ -0,0 +1,61 @@
+/*
+ * 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 org.apache.rocketmq.studio.cluster.metrics;
+
+import com.baomidou.mybatisplus.core.conditions.Wrapper;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.studio.ops.alert.AlertDomain;
+import org.apache.rocketmq.studio.persistence.entity.RmqMetricSnapshot;
+import org.apache.rocketmq.studio.persistence.mapper.RmqMetricSnapshotMapper;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class MybatisPlusMetricSnapshotRepositoryTest {
+
+ @Mock
+ private RmqMetricSnapshotMapper mapper;
+
+ @Test
+ void nullClusterScopeShouldOnlyReadUnscopedSnapshotsTest() {
+ when(mapper.selectList(any(Wrapper.class))).thenReturn(List.of());
+ MybatisPlusMetricSnapshotRepository repository =
+ new MybatisPlusMetricSnapshotRepository(mapper, new
ObjectMapper());
+ MetricSample scope = new MetricSample("broker.availability",
AlertDomain.CLUSTER,
+ "local", null, Map.of(), 1D, MetricAvailability.AVAILABLE,
Instant.now());
+
+ repository.findRecent(scope, Instant.EPOCH);
+
+ ArgumentCaptor<Wrapper<RmqMetricSnapshot>> queryCaptor =
ArgumentCaptor.forClass(Wrapper.class);
+ verify(mapper).selectList(queryCaptor.capture());
+ QueryWrapper<RmqMetricSnapshot> query =
(QueryWrapper<RmqMetricSnapshot>) queryCaptor.getValue();
+ assertThat(query.getSqlSegment()).contains("cluster_id IS NULL");
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryServiceTest.java
index d1bc1e83e..e253677b1 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/nameserver/NameserverRegistryServiceTest.java
@@ -200,6 +200,24 @@ class NameserverRegistryServiceTest {
.hasMessageContaining("already exists");
}
+ @Test
+ void createShouldThrowWhenEntryVanishesBeforeReloadTest() {
+ when(nameserverMapper.selectCount(any())).thenReturn(0L);
+
when(nameserverMapper.insert(any(RmqNameserver.class))).thenAnswer(invocation
-> {
+ RmqNameserver entity = invocation.getArgument(0);
+ entity.setId(12L);
+ return 1;
+ });
+ when(nameserverMapper.selectById(12L)).thenReturn(null);
+
+ assertThatThrownBy(() ->
service.create(CreateNameserverRegistryDTO.builder()
+ .name("prod")
+ .namesrvAddr("10.0.0.1:9876")
+ .build()))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("deleted concurrently");
+ }
+
@Test
void updateShouldPersistAndReturnStoredEntryTest() {
RmqNameserver existing = new RmqNameserver();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
index 0b727eba7..cada952fd 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/proxy/ProxyAddressServiceTest.java
@@ -104,7 +104,10 @@ class ProxyAddressServiceTest {
"10.0.0.1:0",
"10.0.0.1:65536",
"http://10.0.0.1:8081";,
- "10.0.0.1:8081/path"
+ "10.0.0.1:8081/path",
+ "[:::]:8081",
+ "[2001:db8::1::2]:8081",
+ "[127.0.0.1]:8081"
);
for (String invalidProxyAddr : invalidProxyAddrs) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
index 6957a1512..137269cf9 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/ProxyConsumerResolverTest.java
@@ -94,6 +94,24 @@ class ProxyConsumerResolverTest {
.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic");
}
+ @Test
+ void discoverProxyAddressesShouldRetryAfterATransientFailureTest() throws
Exception {
+ ConsumerConnection syncer = new ConsumerConnection();
+ Connection proxy = new Connection();
+ proxy.setClientId("proxy-a");
+ proxy.setClientAddr("10.0.4.66:10911");
+ syncer.setConnectionSet(new HashSet<>(List.of(proxy)));
+
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic"))
+ .thenThrow(new IllegalStateException("nameserver unavailable"))
+ .thenReturn(syncer);
+
+ assertThat(resolver.discoverProxyAddresses("instance-a")).isEmpty();
+
assertThat(resolver.discoverProxyAddresses("instance-a")).containsExactly("10.0.4.66:8080");
+
+ org.mockito.Mockito.verify(adminExt, org.mockito.Mockito.times(2))
+
.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic");
+ }
+
@Test
void resolveConsumerConnectionShouldReturnNullWhenNoProxyDiscoveredTest()
throws Exception {
when(adminExt.examineConsumerConnectionInfo("CID_DefaultHeartBeatSyncerTopic"))
