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 30e697660 fix(alert): consolidate native collection reliability (#2957)
30e697660 is described below
commit 30e697660be86af2acaa98910e1fc6e7f3e4560f
Author: aias00 <[email protected]>
AuthorDate: Wed Sep 2 15:05:50 2026 +0800
fix(alert): consolidate native collection reliability (#2957)
* Prevent native alert collection from dropping saturated instances
Use a sliding collection window so one pass keeps at most
collectionParallelism jobs in flight, starts the next instance only after a job
completes or times out, and keeps the executor queue bounded.
Constraint: preserve lease refresh before persistence, per-instance timeout
cancellation, and interrupt cancellation
Rejected: unbounded executor queue | it can hide scheduler saturation by
queueing the full instance inventory
Confidence: high
Scope-risk: narrow
Tested: JAVA_HOME=$(/usr/libexec/java_home -v 21) mvn
-Dtest=CollectorSchedulerTest test
Signed-off-by: liuhy <[email protected]>
* Resolve stale native alerts after complete collections
Native alert states can remain FIRING or ACKED when the monitored resource
disappears and no longer emits a sample. Reconcile active fingerprints only
after a collector has completed a scoped collection, and use the latest alert
event metadata to emit one resolved lifecycle event with the original labels.
Also batch-load latest alert event metadata for active states to avoid
per-state lookups when reconciling multiple active fingerprints.
Constraint: failed, partial, or lease-lost collection must not clear active
state
Rejected: per-state latest alert selectOne lookup | causes N+1 queries
during reconciliation
Confidence: high
Scope-risk: moderate
Tested: JAVA_HOME=$(/usr/libexec/java_home -v 21) mvn
-Dtest=MybatisPlusAlertStateRepositoryTest test
Tested: JAVA_HOME=$(/usr/libexec/java_home -v 21) mvn
-Dtest=NativeAlertProcessorTest,CollectorSchedulerTest,MybatisPlusAlertStateRepositoryTest
test
Tested: JAVA_HOME=$(/usr/libexec/java_home -v 21) mvn
-Dtest='org.apache.rocketmq.studio.ops.alert.*Test,org.apache.rocketmq.studio.cluster.metrics.*Test'
test
Tested: JAVA_HOME=$(/usr/libexec/java_home -v 21) mvn test
Tested: JAVA_HOME=$(/usr/libexec/java_home -v 21) mvn -DskipTests package
Signed-off-by: liuhy <[email protected]>
---------
Signed-off-by: liuhy <[email protected]>
---
.../cluster/metrics/BusinessMetricsCollector.java | 5 +
.../cluster/metrics/ClusterMetricsCollector.java | 5 +
.../studio/cluster/metrics/CollectorScheduler.java | 206 +++++++++++++++++----
.../cluster/metrics/MetricCollectionScope.java | 41 ++++
.../ApacheRocketMqBusinessMetricsCollector.java | 6 +
.../ApacheRocketMqClusterMetricsCollector.java | 7 +
.../ApacheRocketMqDlqMetricsCollector.java | 6 +
.../ApacheRocketMqProxyMetricsCollector.java | 6 +
.../CloudRocketMqBusinessMetricsCollector.java | 6 +
.../CloudRocketMqClusterMetricsCollector.java | 6 +
.../alert/ActiveAlertState.java} | 22 ++-
.../studio/ops/alert/AlertStateRepository.java | 6 +
.../ops/alert/MybatisPlusAlertStateRepository.java | 85 +++++++++
.../studio/ops/alert/NativeAlertProcessor.java | 130 ++++++++++---
.../cluster/metrics/CollectorSchedulerTest.java | 158 +++++++++++++++-
.../alert/MybatisPlusAlertStateRepositoryTest.java | 95 +++++++++-
.../studio/ops/alert/NativeAlertProcessorTest.java | 89 +++++++++
17 files changed, 800 insertions(+), 79 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/BusinessMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/BusinessMetricsCollector.java
index 327b2dc63..008f36801 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/BusinessMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/BusinessMetricsCollector.java
@@ -19,10 +19,15 @@ package org.apache.rocketmq.studio.cluster.metrics;
import org.apache.rocketmq.studio.instance.InstanceVO;
import java.util.List;
+import java.util.Set;
/** Collects business-flow metrics for one Studio-managed instance. */
public interface BusinessMetricsCollector {
boolean supports(InstanceVO instance);
List<MetricSample> collect(InstanceVO instance);
+
+ default Set<String> metricKeys() {
+ return Set.of();
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/ClusterMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/ClusterMetricsCollector.java
index 406b1207d..c07f0c5d9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/ClusterMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/ClusterMetricsCollector.java
@@ -19,10 +19,15 @@ package org.apache.rocketmq.studio.cluster.metrics;
import org.apache.rocketmq.studio.instance.InstanceVO;
import java.util.List;
+import java.util.Set;
/** Collects operational metrics for one Studio-managed instance. */
public interface ClusterMetricsCollector {
boolean supports(InstanceVO instance);
List<MetricSample> collect(InstanceVO instance);
+
+ default Set<String> metricKeys() {
+ return Set.of();
+ }
}
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 4cdfe880a..22fc26270 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
@@ -20,21 +20,31 @@ import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.studio.instance.InstanceRepository;
import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.ops.alert.AlertDomain;
import org.apache.rocketmq.studio.ops.alert.NativeAlertProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import
org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.time.Duration;
import java.time.Instant;
+import java.util.Set;
/** Runs independent, bounded collection jobs for each configured instance. */
@Slf4j
@@ -79,43 +89,138 @@ public class CollectorScheduler {
log.debug("Skipping native alert collection because another Studio
replica holds the lease");
return;
}
- List<Future<?>> jobs = instanceRepository.findAll().stream()
- .map(this::submitCollection)
- .filter(java.util.Objects::nonNull)
- .toList();
+ List<InstanceVO> instances = instanceRepository.findAll();
+ if (instances.isEmpty()) {
+ return;
+ }
Duration timeout =
parsePositiveDuration(properties.getCollectionTimeout(),
Duration.ofSeconds(15));
- long passStartedAt = System.nanoTime();
- long timeoutNanos = timeout.toNanos();
- for (Future<?> job : jobs) {
- try {
- long remainingNanos = timeoutNanos - (System.nanoTime() -
passStartedAt);
- if (remainingNanos <= 0) {
- job.cancel(true);
- continue;
+ Duration passTimeout = timeout.multipliedBy(instances.size());
+ Instant passDeadline = Instant.now().plus(passTimeout);
+ int parallelism = Math.max(1, properties.getCollectionParallelism());
+ CompletionService<InstanceVO> completionService = new
ExecutorCompletionService<>(collectionExecutor);
+ ArrayDeque<InstanceVO> pending = new ArrayDeque<>(instances);
+ List<CollectionJob> running = new ArrayList<>();
+ try {
+ submitAvailable(completionService, pending, running, parallelism,
timeout, passDeadline);
+ while (!running.isEmpty()) {
+ completeFinishedJobs(running);
+ cancelExpiredJobs(running, timeout, passDeadline, passTimeout);
+ submitAvailable(completionService, pending, running,
parallelism, timeout, passDeadline);
+ if (running.isEmpty()) {
+ break;
+ }
+ Future<InstanceVO> completed =
completionService.poll(nextWaitMillis(running, passDeadline),
+ TimeUnit.MILLISECONDS);
+ if (completed != null) {
+ completeJob(running, completed);
}
- job.get(remainingNanos, TimeUnit.NANOSECONDS);
- } catch (java.util.concurrent.TimeoutException error) {
- job.cancel(true);
- log.warn("Native metric collection pass exceeded {} and
unfinished work was cancelled", timeout);
- } catch (InterruptedException error) {
- Thread.currentThread().interrupt();
+ }
+ } catch (InterruptedException error) {
+ Thread.currentThread().interrupt();
+ cancelRunningJobs(running);
+ return;
+ } catch (RejectedExecutionException error) {
+ cancelRunningJobs(running);
+ log.warn("Native metric collection executor rejected new work;
aborting pass with {} unstarted instance(s)",
+ pending.size());
+ }
+ if (!pending.isEmpty()) {
+ log.warn("Native metric collection pass exceeded {}; {}
instance(s) were not started", passTimeout,
+ pending.size());
+ }
+ }
+
+ private void submitAvailable(CompletionService<InstanceVO>
completionService, ArrayDeque<InstanceVO> pending,
+ List<CollectionJob> running, int parallelism, Duration timeout,
Instant passDeadline) {
+ while (!pending.isEmpty() && running.size() < parallelism) {
+ Instant now = Instant.now();
+ if (!passDeadline.isAfter(now)) {
return;
- } catch (java.util.concurrent.ExecutionException error) {
- log.warn("Native metric collection job failed: {}",
error.getCause().getMessage());
+ }
+ InstanceVO instance = pending.removeFirst();
+ try {
+ Future<InstanceVO> future = completionService.submit(() -> {
+ collectClusterMetrics(instance);
+ collectBusinessMetrics(instance);
+ return instance;
+ });
+ running.add(new CollectionJob(instance, future,
now.plus(timeout)));
+ } catch (RejectedExecutionException error) {
+ pending.addFirst(instance);
+ throw error;
+ }
+ }
+ }
+
+ private void completeFinishedJobs(List<CollectionJob> running) throws
InterruptedException {
+ Iterator<CollectionJob> iterator = running.iterator();
+ while (iterator.hasNext()) {
+ CollectionJob job = iterator.next();
+ if (job.future().isDone()) {
+ iterator.remove();
+ awaitJob(job);
}
}
}
- private Future<?> submitCollection(InstanceVO instance) {
+ private void completeJob(List<CollectionJob> running, Future<InstanceVO>
completed) throws InterruptedException {
+ Iterator<CollectionJob> iterator = running.iterator();
+ while (iterator.hasNext()) {
+ CollectionJob job = iterator.next();
+ if (job.future() == completed) {
+ iterator.remove();
+ awaitJob(job);
+ return;
+ }
+ }
+ }
+
+ private void awaitJob(CollectionJob job) throws InterruptedException {
try {
- return collectionExecutor.submit(() -> {
- collectClusterMetrics(instance);
- collectBusinessMetrics(instance);
- });
- } catch (java.util.concurrent.RejectedExecutionException error) {
- log.warn("Skipping native metric collection for instance {}
because the collector is saturated", instance.getName());
- return null;
+ job.future().get();
+ } catch (CancellationException ignored) {
+ // Already logged when the job was cancelled.
+ } catch (ExecutionException error) {
+ log.warn("Native metric collection job failed for instance {}:
{}", job.instance().getName(),
+ error.getCause().getMessage());
+ }
+ }
+
+ private void cancelExpiredJobs(List<CollectionJob> running, Duration
timeout, Instant passDeadline,
+ Duration passTimeout) {
+ Instant now = Instant.now();
+ Iterator<CollectionJob> iterator = running.iterator();
+ while (iterator.hasNext()) {
+ CollectionJob job = iterator.next();
+ if (!passDeadline.isAfter(now)) {
+ job.future().cancel(true);
+ iterator.remove();
+ log.warn("Native metric collection pass exceeded {};
cancelling instance {}", passTimeout,
+ job.instance().getName());
+ } else if (!job.deadline().isAfter(now)) {
+ job.future().cancel(true);
+ iterator.remove();
+ log.warn("Native metric collection exceeded {} for instance {}
and was cancelled", timeout,
+ job.instance().getName());
+ }
+ }
+ }
+
+ private void cancelRunningJobs(List<CollectionJob> running) {
+ for (CollectionJob job : running) {
+ job.future().cancel(true);
}
+ running.clear();
+ }
+
+ private long nextWaitMillis(List<CollectionJob> running, Instant
passDeadline) {
+ Instant nextDeadline = passDeadline;
+ for (CollectionJob job : running) {
+ if (job.deadline().isBefore(nextDeadline)) {
+ nextDeadline = job.deadline();
+ }
+ }
+ return Math.max(1, Duration.between(Instant.now(),
nextDeadline).toMillis());
}
@Scheduled(fixedDelayString =
"${studio.alerting.snapshot-cleanup-interval:PT1H}")
@@ -130,7 +235,10 @@ public class CollectorScheduler {
private void collectClusterMetrics(InstanceVO instance) {
for (ClusterMetricsCollector collector : clusterCollectors) {
try {
- persist(collector.supports(instance) ?
collector.collect(instance) : List.of());
+ if (collector.supports(instance)) {
+ List<MetricSample> samples = collector.collect(instance);
+ persist(AlertDomain.CLUSTER, instance,
collector.metricKeys(), samples);
+ }
} catch (RuntimeException error) {
log.warn("Native metric collector failed for instance {}: {}",
instance.getName(), error.getMessage());
}
@@ -140,23 +248,40 @@ public class CollectorScheduler {
private void collectBusinessMetrics(InstanceVO instance) {
for (BusinessMetricsCollector collector : businessCollectors) {
try {
- persist(collector.supports(instance) ?
collector.collect(instance) : List.of());
+ if (collector.supports(instance)) {
+ List<MetricSample> samples = collector.collect(instance);
+ persist(AlertDomain.BUSINESS, instance,
collector.metricKeys(), samples);
+ }
} catch (RuntimeException error) {
log.warn("Native metric collector failed for instance {}: {}",
instance.getName(), error.getMessage());
}
}
}
- private void persist(List<MetricSample> samples) {
- if (!samples.isEmpty()) {
- // Refresh immediately before persisting so a slow remote
collection cannot write after lease loss.
- if (!collectionLease.tryAcquire()) {
- log.debug("Discarding native metric samples because the
collection lease was lost");
- return;
- }
- snapshotRepository.saveAll(samples);
- alertProcessor.process(samples);
+ private void persist(AlertDomain domain, InstanceVO instance, Set<String>
declaredMetricKeys,
+ List<MetricSample> samples) {
+ List<MetricSample> collected = samples == null ? List.of() : samples;
+ Set<String> scopeMetricKeys = metricKeys(declaredMetricKeys,
collected);
+ if (scopeMetricKeys.isEmpty()) {
+ return;
+ }
+ MetricCollectionScope actualScope = new MetricCollectionScope(domain,
instance.getName(), scopeMetricKeys);
+ // Refresh immediately before persisting so a slow remote collection
cannot write after lease loss.
+ if (!collectionLease.tryAcquire()) {
+ log.debug("Discarding native metric samples because the collection
lease was lost");
+ return;
}
+ if (!collected.isEmpty()) {
+ snapshotRepository.saveAll(collected);
+ }
+ alertProcessor.processSuccessfulCollection(actualScope, collected);
+ }
+
+ private static Set<String> metricKeys(Set<String> declared,
List<MetricSample> samples) {
+ if (declared != null && !declared.isEmpty()) {
+ return declared;
+ }
+ return
samples.stream().map(MetricSample::metricKey).collect(java.util.stream.Collectors.toSet());
}
@PreDestroy
@@ -175,6 +300,9 @@ public class CollectorScheduler {
new ArrayBlockingQueue<>(parallelism), threadFactory, new
ThreadPoolExecutor.AbortPolicy());
}
+ private record CollectionJob(InstanceVO instance, Future<InstanceVO>
future, Instant deadline) {
+ }
+
private static Duration parsePositiveDuration(String value, Duration
fallback) {
try {
Duration duration = Duration.parse(value);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MetricCollectionScope.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MetricCollectionScope.java
new file mode 100644
index 000000000..a3d49b165
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MetricCollectionScope.java
@@ -0,0 +1,41 @@
+/*
+ * 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 org.apache.rocketmq.studio.ops.alert.AlertDomain;
+
+import java.util.Objects;
+import java.util.Set;
+
+/** Identifies one successful native metric collection scope. */
+public record MetricCollectionScope(AlertDomain domain, String instanceId,
Set<String> metricKeys) {
+ public MetricCollectionScope {
+ Objects.requireNonNull(domain, "domain is required");
+ if (instanceId == null || instanceId.isBlank()) {
+ throw new IllegalArgumentException("instanceId is required");
+ }
+ metricKeys = Set.copyOf(metricKeys == null ? Set.of() : metricKeys);
+ if (metricKeys.isEmpty() || metricKeys.stream().anyMatch(key -> key ==
null || key.isBlank())) {
+ throw new IllegalArgumentException("metricKeys must not be empty");
+ }
+ }
+
+ public boolean contains(MetricSample sample) {
+ return sample != null && domain == sample.domain() &&
instanceId.equals(sample.instanceId())
+ && metricKeys.contains(sample.metricKey());
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollector.java
index dfeb918a5..e4cfe9af9 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollector.java
@@ -34,6 +34,7 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/** Collects consumer-group lag directly from Studio's Apache instance
provider. */
@Slf4j
@@ -53,6 +54,11 @@ public class ApacheRocketMqBusinessMetricsCollector
implements BusinessMetricsCo
&& instance.getName() != null;
}
+ @Override
+ public Set<String> metricKeys() {
+ return Set.of(CONSUMER_LAG_TOTAL, CONSUMER_LAG_MAX_QUEUE,
CONSUMER_DELAY_SECONDS, TOPIC_BACKLOG_TOTAL);
+ }
+
@Override
public List<MetricSample> collect(InstanceVO instance) {
if (!supports(instance)) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
index 27580f6ab..df1435be8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqClusterMetricsCollector.java
@@ -36,6 +36,7 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/** Native Apache RocketMQ health collection through Studio's managed admin
client. */
@Slf4j
@@ -57,6 +58,12 @@ public class ApacheRocketMqClusterMetricsCollector
implements ClusterMetricsColl
&& StringUtils.hasText(instance.getName()) &&
StringUtils.hasText(instance.getEndpoint());
}
+ @Override
+ public Set<String> metricKeys() {
+ return Set.of(NAMESERVER_AVAILABILITY, BROKER_AVAILABILITY,
BROKER_DISK_USAGE_RATIO,
+ BROKER_JVM_HEAP_USAGE_RATIO, BROKER_SEND_QUEUE_USAGE_RATIO);
+ }
+
@Override
public List<MetricSample> collect(InstanceVO instance) {
if (!supports(instance)) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqDlqMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqDlqMetricsCollector.java
index 6ef6fdfc8..d8002771b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqDlqMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqDlqMetricsCollector.java
@@ -31,6 +31,7 @@ import org.springframework.stereotype.Component;
import java.time.Instant;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/** Collects dead-letter message counts directly from the Apache DLQ provider.
*/
@Slf4j
@@ -47,6 +48,11 @@ public class ApacheRocketMqDlqMetricsCollector implements
BusinessMetricsCollect
&& instance.getName() != null;
}
+ @Override
+ public Set<String> metricKeys() {
+ return Set.of(DLQ_MESSAGE_COUNT);
+ }
+
@Override
public List<MetricSample> collect(InstanceVO instance) {
if (!supports(instance)) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqProxyMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqProxyMetricsCollector.java
index 15293254a..2a384ab74 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqProxyMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqProxyMetricsCollector.java
@@ -35,6 +35,7 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/** Collects TCP reachability for Proxies discovered from the selected Studio
instance. */
@Slf4j
@@ -54,6 +55,11 @@ public class ApacheRocketMqProxyMetricsCollector implements
ClusterMetricsCollec
&& StringUtils.hasText(instance.getName()) &&
StringUtils.hasText(instance.getEndpoint());
}
+ @Override
+ public Set<String> metricKeys() {
+ return Set.of(PROXY_AVAILABILITY);
+ }
+
@Override
public List<MetricSample> collect(InstanceVO instance) {
if (!supports(instance)) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollector.java
index 875208dc5..4cc1d6161 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqBusinessMetricsCollector.java
@@ -34,6 +34,7 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/** Collects cloud consumer lag using the provider's existing consumer
progress API. */
@Slf4j
@@ -52,6 +53,11 @@ public class CloudRocketMqBusinessMetricsCollector
implements BusinessMetricsCol
|| instance.getVendor() == InstanceVendor.TENCENT) &&
instance.getName() != null;
}
+ @Override
+ public Set<String> metricKeys() {
+ return Set.of(CONSUMER_LAG_TOTAL, CONSUMER_LAG_MAX_QUEUE,
TOPIC_BACKLOG_TOTAL);
+ }
+
@Override
public List<MetricSample> collect(InstanceVO instance) {
if (!supports(instance)) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqClusterMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqClusterMetricsCollector.java
index ddbec51d7..92b91b1b0 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqClusterMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/CloudRocketMqClusterMetricsCollector.java
@@ -31,6 +31,7 @@ import org.springframework.util.StringUtils;
import java.time.Instant;
import java.util.Map;
+import java.util.Set;
/** Collects the cloud provider's managed-instance lifecycle status for Aliyun
and Tencent. */
@Slf4j
@@ -49,6 +50,11 @@ public class CloudRocketMqClusterMetricsCollector implements
ClusterMetricsColle
&& StringUtils.hasText(instance.getCloudInstanceId());
}
+ @Override
+ public Set<String> metricKeys() {
+ return Set.of(CLOUD_INSTANCE_AVAILABILITY);
+ }
+
@Override
public java.util.List<MetricSample> collect(InstanceVO instance) {
if (!supports(instance)) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/BusinessMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/ActiveAlertState.java
similarity index 55%
copy from
server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/BusinessMetricsCollector.java
copy to
server/src/main/java/org/apache/rocketmq/studio/ops/alert/ActiveAlertState.java
index 327b2dc63..acafc73aa 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/BusinessMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/ActiveAlertState.java
@@ -14,15 +14,19 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.rocketmq.studio.cluster.metrics;
+package org.apache.rocketmq.studio.ops.alert;
-import org.apache.rocketmq.studio.instance.InstanceVO;
+import java.util.Map;
+import java.util.Objects;
-import java.util.List;
-
-/** Collects business-flow metrics for one Studio-managed instance. */
-public interface BusinessMetricsCollector {
- boolean supports(InstanceVO instance);
-
- List<MetricSample> collect(InstanceVO instance);
+/** Persisted active state plus labels needed to emit a lifecycle recovery
event. */
+public record ActiveAlertState(AlertStateKey key, AlertRuleState state, String
instanceId, Map<String, String> labels) {
+ public ActiveAlertState {
+ Objects.requireNonNull(key, "key is required");
+ Objects.requireNonNull(state, "state is required");
+ if (instanceId == null || instanceId.isBlank()) {
+ throw new IllegalArgumentException("instanceId is required");
+ }
+ labels = Map.copyOf(labels == null ? Map.of() : labels);
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertStateRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertStateRepository.java
index f33674d11..8f5607859 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertStateRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertStateRepository.java
@@ -16,6 +16,8 @@
*/
package org.apache.rocketmq.studio.ops.alert;
+import org.apache.rocketmq.studio.cluster.metrics.MetricCollectionScope;
+
import java.time.Instant;
import java.util.Optional;
import java.util.List;
@@ -37,6 +39,10 @@ public interface AlertStateRepository {
void deleteByRuleId(Long ruleId);
+ default List<ActiveAlertState> findActive(MetricCollectionScope scope,
List<AlertRuleVO> rules) {
+ return List.of();
+ }
+
default List<AlertRuleRuntimeVO> findRuntimeByRuleIds(List<AlertRuleVO>
rules) {
return List.of();
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertStateRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertStateRepository.java
index aa0d85262..b71d44628 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertStateRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertStateRepository.java
@@ -17,24 +17,35 @@
package org.apache.rocketmq.studio.ops.alert;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
+import org.apache.rocketmq.studio.cluster.metrics.MetricCollectionScope;
import org.apache.rocketmq.studio.persistence.entity.RmqAlertState;
+import org.apache.rocketmq.studio.persistence.entity.RmqSystemAlert;
import org.apache.rocketmq.studio.persistence.mapper.RmqAlertStateMapper;
+import org.apache.rocketmq.studio.persistence.mapper.RmqSystemAlertMapper;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Repository;
+import org.springframework.util.StringUtils;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Optional;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
@Repository
@RequiredArgsConstructor
public class MybatisPlusAlertStateRepository implements AlertStateRepository {
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
private final RmqAlertStateMapper mapper;
+ private final RmqSystemAlertMapper alertMapper;
@Override
public Optional<AlertRuleState> find(AlertStateKey key) {
@@ -82,6 +93,35 @@ public class MybatisPlusAlertStateRepository implements
AlertStateRepository {
}
}
+ @Override
+ public List<ActiveAlertState> findActive(MetricCollectionScope scope,
List<AlertRuleVO> rules) {
+ if (scope == null || rules == null || rules.isEmpty()) {
+ return List.of();
+ }
+ Set<String> metricKeys = scope.metricKeys();
+ Map<Long, AlertRuleVO> scopedRules = rules.stream()
+ .filter(rule -> rule.getId() != null)
+ .filter(AlertRuleVO::isEnabled)
+ .filter(rule -> ruleDomain(rule) == scope.domain())
+ .filter(rule -> metricKeys.contains(rule.getMetric()))
+ .filter(rule -> !StringUtils.hasText(rule.getInstanceId())
+ || scope.instanceId().equals(rule.getInstanceId()))
+ .collect(Collectors.toMap(AlertRuleVO::getId, rule -> rule,
(left, right) -> left));
+ if (scopedRules.isEmpty()) {
+ return List.of();
+ }
+ List<RmqAlertState> activeStates = mapper.selectList(new
QueryWrapper<RmqAlertState>()
+ .in("rule_id", scopedRules.keySet())
+ .in("status", List.of(AlertStateStatus.FIRING.name(),
AlertStateStatus.ACKED.name())));
+ Map<AlertStateKey, RmqSystemAlert> latestAlerts =
findLatestAlerts(scope, activeStates);
+ return activeStates
+ .stream()
+ .map(state -> toActiveState(state, latestAlerts.get(new
AlertStateKey(state.getRuleId(),
+ state.getFingerprint()))))
+ .flatMap(Optional::stream)
+ .toList();
+ }
+
@Override
public List<AlertRuleRuntimeVO> findRuntimeByRuleIds(List<AlertRuleVO>
rules) {
Map<Long, AlertRuleVO> byId = rules.stream().filter(rule ->
rule.getId() != null)
@@ -99,6 +139,40 @@ public class MybatisPlusAlertStateRepository implements
AlertStateRepository {
.currentValue(entity.getCurrentValue()).lastNotifiedAt(entity.getLastNotifiedAt()).nextReminderAt(next).build();
}
+ private Map<AlertStateKey, RmqSystemAlert>
findLatestAlerts(MetricCollectionScope scope,
+ List<RmqAlertState> activeStates) {
+ if (activeStates.isEmpty()) {
+ return Map.of();
+ }
+ Set<Long> ruleIds =
activeStates.stream().map(RmqAlertState::getRuleId).collect(Collectors.toSet());
+ Set<String> fingerprints =
activeStates.stream().map(RmqAlertState::getFingerprint).collect(Collectors.toSet());
+ Set<AlertStateKey> activeKeys = activeStates.stream()
+ .map(state -> new AlertStateKey(state.getRuleId(),
state.getFingerprint()))
+ .collect(Collectors.toCollection(HashSet::new));
+ return alertMapper.selectList(new QueryWrapper<RmqSystemAlert>()
+ .in("rule_id", ruleIds)
+ .in("fingerprint", fingerprints)
+ .eq("domain", scope.domain().name())
+ .eq("instance_id", scope.instanceId())
+ .orderByDesc("time", "id"))
+ .stream()
+ .filter(alert -> activeKeys.contains(new
AlertStateKey(alert.getRuleId(), alert.getFingerprint())))
+ .collect(Collectors.toMap(alert -> new
AlertStateKey(alert.getRuleId(), alert.getFingerprint()),
+ alert -> alert, (latest, ignored) -> latest));
+ }
+
+ private Optional<ActiveAlertState> toActiveState(RmqAlertState state,
RmqSystemAlert alert) {
+ if (alert == null) {
+ return Optional.empty();
+ }
+ return Optional.of(new ActiveAlertState(new
AlertStateKey(state.getRuleId(), state.getFingerprint()),
+ toState(state), alert.getInstanceId(),
readLabels(alert.getLabelsJson())));
+ }
+
+ private static AlertDomain ruleDomain(AlertRuleVO rule) {
+ return rule.getDomain() == null ? AlertDomain.BUSINESS :
rule.getDomain();
+ }
+
private static void apply(RmqAlertState entity, AlertRuleState state) {
entity.setStatus(state.status().name());
entity.setConsecutiveHits(state.consecutiveHits());
@@ -124,4 +198,15 @@ public class MybatisPlusAlertStateRepository implements
AlertStateRepository {
private static Instant toInstant(LocalDateTime value) {
return value == null ? null : value.toInstant(ZoneOffset.UTC);
}
+
+ private static Map<String, String> readLabels(String labelsJson) {
+ if (!StringUtils.hasText(labelsJson)) {
+ return Map.of();
+ }
+ try {
+ return OBJECT_MAPPER.readValue(labelsJson, new TypeReference<>() {
});
+ } catch (Exception error) {
+ throw new IllegalStateException("Unable to read alert labels",
error);
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java
index b4559a533..f75560654 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java
@@ -20,19 +20,24 @@ import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.studio.cluster.metrics.MetricSample;
import org.apache.rocketmq.studio.cluster.metrics.MetricSnapshotRepository;
import org.apache.rocketmq.studio.cluster.metrics.MetricAvailability;
+import org.apache.rocketmq.studio.cluster.metrics.MetricCollectionScope;
import org.apache.rocketmq.studio.common.domain.enums.AlertLevel;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.time.Duration;
+import java.time.Instant;
import java.time.ZoneOffset;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.TreeMap;
import java.util.EnumMap;
+import java.util.Objects;
+import java.util.stream.Collectors;
/** Applies native samples to persisted rule state and emits only lifecycle
transitions. */
@Component
@@ -49,6 +54,27 @@ public class NativeAlertProcessor {
@Transactional
public void process(List<MetricSample> samples) {
+ processSamples(samples);
+ }
+
+ @Transactional
+ public void processSuccessfulCollection(MetricCollectionScope scope,
List<MetricSample> samples) {
+ Objects.requireNonNull(scope, "scope is required");
+ List<MetricSample> collected = samples == null ? List.of() : samples;
+ processSamples(collected);
+ if (containsWholeScopeFailure(scope, collected)) {
+ return;
+ }
+ reconcileMissingActiveStates(scope, collected);
+ }
+
+ private static boolean containsWholeScopeFailure(MetricCollectionScope
scope, List<MetricSample> samples) {
+ return samples.stream().filter(scope::contains)
+ .anyMatch(sample -> sample.availability() !=
MetricAvailability.AVAILABLE
+ && sample.labels().isEmpty());
+ }
+
+ private void processSamples(List<MetricSample> samples) {
Map<AlertDomain, List<AlertRuleVO>> rulesByDomain = new
EnumMap<>(AlertDomain.class);
for (MetricSample sample : samples) {
for (AlertRuleVO rule :
rulesByDomain.computeIfAbsent(sample.domain(), alertService::listRules)) {
@@ -76,39 +102,87 @@ public class NativeAlertProcessor {
}
if (update.transition() == AlertStateTransition.FIRING ||
update.transition() == AlertStateTransition.REMINDER
|| update.transition() ==
AlertStateTransition.RESOLVED) {
- LocalDateTime eventTime =
LocalDateTime.ofInstant(sample.collectedAt(), ZoneOffset.UTC);
- SystemAlertVO event =
SystemAlertVO.builder().level(level(rule.getSeverity()))
-
.title(rule.getName()).description(update.transition() + " " +
sample.metricKey()
- + " on " +
sample.instanceId()).time(eventTime)
-
.acknowledged(false).domain(sample.domain()).ruleId(rule.getId())
-
.fingerprint(key.fingerprint()).transition(update.transition().name())
-
.instanceId(sample.instanceId()).currentValue(update.state().currentValue())
- .labels(Map.copyOf(new
TreeMap<>(sample.labels()))).build();
- boolean suppressNotification =
shouldSuppress(sample.domain(), update.transition());
- if (suppressNotification) {
- Optional<SystemAlertVO> cause =
notificationSuppressionService.findSuppressingClusterAlert(event);
- if (cause.isPresent()) {
- event.setNotificationSuppressed(true);
-
event.setSuppressionCauseAlertId(cause.get().getId());
- event.setSuppressionReason("Suppressed by active
cluster incident #" + cause.get().getId()
- + ": " + cause.get().getTitle());
- }
- }
- SystemAlertVO savedEvent =
alertRepository.saveAlert(event);
- if (savedEvent != null) {
- event = savedEvent;
- }
- if (update.transition() == AlertStateTransition.FIRING) {
- alertRepository.markRuleTriggered(rule.getId(),
eventTime.toString());
- }
- if (!event.isNotificationSuppressed()) {
- notificationOutboxService.enqueue(event, rule,
sample.labels());
- }
+ emitLifecycleEvent(rule, key, update, sample.domain(),
sample.instanceId(), sample.metricKey(),
+ sample.labels(), sample.collectedAt());
}
}
}
}
+ private void reconcileMissingActiveStates(MetricCollectionScope scope,
List<MetricSample> samples) {
+ List<AlertRuleVO> rules =
alertService.listRules(scope.domain()).stream()
+ .filter(rule -> rule.getId() != null)
+ .filter(AlertRuleVO::isEnabled)
+ .filter(rule -> scope.metricKeys().contains(rule.getMetric()))
+ .filter(rule -> rule.getInstanceId() == null ||
scope.instanceId().equals(rule.getInstanceId()))
+ .toList();
+ if (rules.isEmpty()) {
+ return;
+ }
+ Set<AlertStateKey> presentKeys = samples.stream()
+ .filter(scope::contains)
+ .flatMap(sample -> rules.stream()
+ .filter(rule ->
NativeAlertRuleScopeMatcher.matches(rule, sample))
+ .map(rule -> new AlertStateKey(rule.getId(),
+ AlertFingerprint.of(rule.getId(),
sample.instanceId(), sample.labels()))))
+ .collect(Collectors.toSet());
+ Map<Long, AlertRuleVO> byId =
rules.stream().collect(Collectors.toMap(AlertRuleVO::getId, rule -> rule));
+ Instant resolvedAt =
samples.stream().filter(scope::contains).map(MetricSample::collectedAt).max(Instant::compareTo)
+ .orElseGet(Instant::now);
+ AlertEvaluationResult clear = new AlertEvaluationResult(true, false,
null, MetricAvailability.AVAILABLE);
+ for (ActiveAlertState active : stateRepository.findActive(scope,
rules)) {
+ if (presentKeys.contains(active.key())) {
+ continue;
+ }
+ AlertRuleVO rule = byId.get(active.key().ruleId());
+ if (rule == null) {
+ continue;
+ }
+ AlertStateUpdate update = stateMachine.advance(active.state(),
clear,
+ Math.max(1, rule.getConsecutiveSamples()),
AlertRuleDuration.parse(rule.getDuration()),
+ AlertRuleDuration.parse(rule.getReminderInterval()),
resolvedAt);
+ if (update.transition() != AlertStateTransition.RESOLVED) {
+ continue;
+ }
+ if (!stateRepository.save(active.key(), update.state())) {
+ continue;
+ }
+ emitLifecycleEvent(rule, active.key(), update, scope.domain(),
active.instanceId(), rule.getMetric(),
+ active.labels(), resolvedAt);
+ }
+ }
+
+ private void emitLifecycleEvent(AlertRuleVO rule, AlertStateKey key,
AlertStateUpdate update, AlertDomain domain,
+ String instanceId, String metricKey, Map<String, String> labels,
Instant collectedAt) {
+ LocalDateTime eventTime = LocalDateTime.ofInstant(collectedAt,
ZoneOffset.UTC);
+ Map<String, String> eventLabels = Map.copyOf(new TreeMap<>(labels ==
null ? Map.of() : labels));
+ SystemAlertVO event =
SystemAlertVO.builder().level(level(rule.getSeverity()))
+ .title(rule.getName()).description(update.transition() + " " +
metricKey + " on " + instanceId)
+
.time(eventTime).acknowledged(false).domain(domain).ruleId(rule.getId())
+
.fingerprint(key.fingerprint()).transition(update.transition().name()).instanceId(instanceId)
+
.currentValue(update.state().currentValue()).labels(eventLabels).build();
+ boolean suppressNotification = shouldSuppress(domain,
update.transition());
+ if (suppressNotification) {
+ Optional<SystemAlertVO> cause =
notificationSuppressionService.findSuppressingClusterAlert(event);
+ if (cause.isPresent()) {
+ event.setNotificationSuppressed(true);
+ event.setSuppressionCauseAlertId(cause.get().getId());
+ event.setSuppressionReason("Suppressed by active cluster
incident #" + cause.get().getId()
+ + ": " + cause.get().getTitle());
+ }
+ }
+ SystemAlertVO savedEvent = alertRepository.saveAlert(event);
+ if (savedEvent != null) {
+ event = savedEvent;
+ }
+ if (update.transition() == AlertStateTransition.FIRING) {
+ alertRepository.markRuleTriggered(rule.getId(),
eventTime.toString());
+ }
+ if (!event.isNotificationSuppressed()) {
+ notificationOutboxService.enqueue(event, rule, eventLabels);
+ }
+ }
+
private static boolean shouldSuppress(AlertDomain domain,
AlertStateTransition transition) {
return domain == AlertDomain.BUSINESS
&& (transition == AlertStateTransition.FIRING || transition ==
AlertStateTransition.REMINDER);
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 ea4c8fdad..ecf956412 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
@@ -23,12 +23,18 @@ import
org.apache.rocketmq.studio.ops.alert.NativeAlertProcessor;
import org.junit.jupiter.api.Test;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -38,6 +44,103 @@ import static org.mockito.Mockito.when;
class CollectorSchedulerTest {
+ @Test
+ void collectsAllInstancesWhenInventoryExceedsExecutorCapacityTest() throws
Exception {
+ AlertingProperties properties = new AlertingProperties();
+ properties.setCollectionParallelism(2);
+ properties.setCollectionTimeout("PT2S");
+ InstanceRepository instances = mock(InstanceRepository.class);
+ List<InstanceVO> inventory = new ArrayList<>();
+ for (int index = 0; index < 5; index++) {
+ inventory.add(InstanceVO.builder().name("instance-" +
index).endpoint("instance-" + index + ":9876").build());
+ }
+ when(instances.findAll()).thenReturn(inventory);
+
+ CountDownLatch firstWaveStarted = new CountDownLatch(2);
+ CountDownLatch releaseFirstWave = new CountDownLatch(1);
+ AtomicInteger started = new AtomicInteger();
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ for (InstanceVO instance : inventory) {
+ when(collector.supports(instance)).thenReturn(true);
+ when(collector.collect(instance)).thenAnswer(invocation -> {
+ if (started.incrementAndGet() <= 2) {
+ firstWaveStarted.countDown();
+ assertTrue(releaseFirstWave.await(1, TimeUnit.SECONDS),
"test did not release the first wave");
+ }
+ return List.of(sampleFor(instance));
+ });
+ }
+
+ List<MetricSample> persisted = new CopyOnWriteArrayList<>();
+ MetricSnapshotRepository snapshots =
mock(MetricSnapshotRepository.class);
+ org.mockito.Mockito.doAnswer(invocation -> {
+ persisted.addAll(invocation.getArgument(0));
+ return null;
+ }).when(snapshots).saveAll(any());
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ when(lease.tryAcquire()).thenReturn(true);
+ CollectorScheduler scheduler = new CollectorScheduler(properties,
instances, List.of(collector), List.of(), snapshots,
+ mock(NativeAlertProcessor.class), lease);
+
+ Thread pass = new Thread(scheduler::collect);
+ pass.start();
+ assertTrue(firstWaveStarted.await(1, TimeUnit.SECONDS), "first wave
should occupy the collector executor");
+ TimeUnit.MILLISECONDS.sleep(100);
+ releaseFirstWave.countDown();
+ pass.join(TimeUnit.SECONDS.toMillis(3));
+ scheduler.stopCollectionExecutor();
+
+ assertTrue(!pass.isAlive(), "collection pass should finish");
+ assertEquals(inventory.size(), persisted.size(), "collector saturation
must not drop later instances");
+
assertEquals(inventory.stream().map(InstanceVO::getName).sorted().toList(),
+
persisted.stream().map(MetricSample::instanceId).sorted().toList());
+ }
+
+ @Test
+ void doesNotQueueMoreThanParallelismBeforeAnyCollectionCompletesTest()
throws Exception {
+ AlertingProperties properties = new AlertingProperties();
+ properties.setCollectionParallelism(2);
+ properties.setCollectionTimeout("PT2S");
+ InstanceRepository instances = mock(InstanceRepository.class);
+ List<InstanceVO> inventory = new ArrayList<>();
+ for (int index = 0; index < 5; index++) {
+ inventory.add(InstanceVO.builder().name("window-" +
index).endpoint("window-" + index + ":9876").build());
+ }
+ when(instances.findAll()).thenReturn(inventory);
+
+ CountDownLatch firstWaveStarted = new CountDownLatch(2);
+ CountDownLatch releaseFirstWave = new CountDownLatch(1);
+ AtomicInteger started = new AtomicInteger();
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ for (InstanceVO instance : inventory) {
+ when(collector.supports(instance)).thenReturn(true);
+ when(collector.collect(instance)).thenAnswer(invocation -> {
+ if (started.incrementAndGet() <= 2) {
+ firstWaveStarted.countDown();
+ assertTrue(releaseFirstWave.await(1, TimeUnit.SECONDS),
"test did not release the first wave");
+ }
+ return List.of(sampleFor(instance));
+ });
+ }
+
+ LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<>();
+ ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 2, 0,
TimeUnit.MILLISECONDS, queue);
+ CollectorScheduler scheduler = new CollectorScheduler(properties,
instances, List.of(collector), List.of(),
+ mock(MetricSnapshotRepository.class),
mock(NativeAlertProcessor.class), acquiredLease(), executor);
+
+ Thread pass = new Thread(scheduler::collect);
+ pass.start();
+ assertTrue(firstWaveStarted.await(1, TimeUnit.SECONDS), "first wave
should occupy the collector executor");
+ TimeUnit.MILLISECONDS.sleep(100);
+ int queuedBeforeAnyCompletion = queue.size();
+ releaseFirstWave.countDown();
+ pass.join(TimeUnit.SECONDS.toMillis(3));
+ scheduler.stopCollectionExecutor();
+
+ assertTrue(!pass.isAlive(), "collection pass should finish");
+ assertEquals(0, queuedBeforeAnyCompletion, "collector should submit
the next instance only after one finishes");
+ }
+
@Test
void persistsFastInstanceWhileAnotherInstanceTimesOutTest() throws
Exception {
AlertingProperties properties = new AlertingProperties();
@@ -136,7 +239,7 @@ class CollectorSchedulerTest {
new CollectorScheduler(properties, instances, List.of(collector),
List.of(), snapshots, processor, lease).collect();
verify(snapshots).saveAll(List.of(sample));
- verify(processor).process(List.of(sample));
+
verify(processor).processSuccessfulCollection(any(MetricCollectionScope.class),
org.mockito.ArgumentMatchers.eq(List.of(sample)));
}
@Test
@@ -184,5 +287,58 @@ class CollectorSchedulerTest {
verify(snapshots, never()).saveAll(any());
verify(processor, never()).process(any());
+ verify(processor,
never()).processSuccessfulCollection(any(MetricCollectionScope.class), any());
+ }
+
+ @Test
+ void doesNotReconcileCollectionScopeWhenCollectorFailsTest() {
+ AlertingProperties properties = new AlertingProperties();
+ InstanceRepository instances = mock(InstanceRepository.class);
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ MetricSnapshotRepository snapshots =
mock(MetricSnapshotRepository.class);
+ NativeAlertProcessor processor = mock(NativeAlertProcessor.class);
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ InstanceVO instance =
InstanceVO.builder().name("local").endpoint("localhost:9876").build();
+ when(instances.findAll()).thenReturn(List.of(instance));
+ when(collector.supports(instance)).thenReturn(true);
+ when(collector.collect(instance)).thenThrow(new
IllegalStateException("collector failed"));
+ when(lease.tryAcquire()).thenReturn(true);
+
+ new CollectorScheduler(properties, instances, List.of(collector),
List.of(), snapshots, processor, lease).collect();
+
+ verify(snapshots, never()).saveAll(any());
+ verify(processor,
never()).processSuccessfulCollection(any(MetricCollectionScope.class), any());
+ }
+
+ @Test
+ void
reconcilesSuccessfulEmptyCollectionWhenCollectorDeclaresMetricKeysTest() {
+ AlertingProperties properties = new AlertingProperties();
+ InstanceRepository instances = mock(InstanceRepository.class);
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ MetricSnapshotRepository snapshots =
mock(MetricSnapshotRepository.class);
+ NativeAlertProcessor processor = mock(NativeAlertProcessor.class);
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ InstanceVO instance =
InstanceVO.builder().name("local").endpoint("localhost:9876").build();
+ when(instances.findAll()).thenReturn(List.of(instance));
+ when(collector.supports(instance)).thenReturn(true);
+
when(collector.metricKeys()).thenReturn(java.util.Set.of("broker.availability"));
+ when(collector.collect(instance)).thenReturn(List.of());
+ when(lease.tryAcquire()).thenReturn(true);
+
+ new CollectorScheduler(properties, instances, List.of(collector),
List.of(), snapshots, processor, lease).collect();
+
+ verify(snapshots, never()).saveAll(any());
+
verify(processor).processSuccessfulCollection(any(MetricCollectionScope.class),
org.mockito.ArgumentMatchers.eq(List.of()));
+ }
+
+ private static MetricSample sampleFor(InstanceVO instance) {
+ return new MetricSample("nameserver.availability",
AlertDomain.CLUSTER, instance.getName(), null,
+ null, 1D, MetricAvailability.AVAILABLE, Instant.now());
+ }
+
+ private static AlertCollectionLease acquiredLease() {
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ when(lease.tryAcquire()).thenReturn(true);
+ return lease;
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertStateRepositoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertStateRepositoryTest.java
index e7fd02f3d..4598d69b8 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertStateRepositoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertStateRepositoryTest.java
@@ -6,17 +6,26 @@
*/
package org.apache.rocketmq.studio.ops.alert;
+import org.apache.rocketmq.studio.cluster.metrics.MetricCollectionScope;
import org.apache.rocketmq.studio.persistence.entity.RmqAlertState;
+import org.apache.rocketmq.studio.persistence.entity.RmqSystemAlert;
import org.apache.rocketmq.studio.persistence.mapper.RmqAlertStateMapper;
+import org.apache.rocketmq.studio.persistence.mapper.RmqSystemAlertMapper;
import org.junit.jupiter.api.Test;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -29,7 +38,8 @@ class MybatisPlusAlertStateRepositoryTest {
existing.setVersion(3);
existing.setStatus(AlertStateStatus.FIRING.name());
when(mapper.selectOne(any())).thenReturn(existing);
- MybatisPlusAlertStateRepository repository = new
MybatisPlusAlertStateRepository(mapper);
+ MybatisPlusAlertStateRepository repository = new
MybatisPlusAlertStateRepository(mapper,
+ mock(RmqSystemAlertMapper.class));
repository.save(new AlertStateKey(4L, "fingerprint"), new
AlertRuleState(AlertStateStatus.FIRING, 2,
10D, Instant.now(), Instant.now(), Instant.now(), null));
@@ -43,11 +53,92 @@ class MybatisPlusAlertStateRepositoryTest {
Instant firedAt = Instant.parse("2026-08-22T12:00:00Z");
when(mapper.acknowledgeFiring(eq(4L), eq("fingerprint"),
eq(LocalDateTime.ofInstant(firedAt, ZoneOffset.UTC)),
any())).thenReturn(1);
- MybatisPlusAlertStateRepository repository = new
MybatisPlusAlertStateRepository(mapper);
+ MybatisPlusAlertStateRepository repository = new
MybatisPlusAlertStateRepository(mapper,
+ mock(RmqSystemAlertMapper.class));
repository.acknowledge(new AlertStateKey(4L, "fingerprint"), firedAt);
verify(mapper).acknowledgeFiring(eq(4L), eq("fingerprint"),
eq(LocalDateTime.ofInstant(firedAt, ZoneOffset.UTC)), any());
}
+
+ @Test
+ void findsActiveStatesWithLabelsFromTheLatestAlertEventTest() {
+ RmqAlertStateMapper mapper = mock(RmqAlertStateMapper.class);
+ RmqAlertState state = new RmqAlertState();
+ state.setRuleId(4L);
+ state.setFingerprint("fingerprint");
+ state.setStatus(AlertStateStatus.ACKED.name());
+ state.setConsecutiveHits(1);
+ state.setCurrentValue(30D);
+ when(mapper.selectList(any())).thenReturn(List.of(state));
+ RmqSystemAlert alert = new RmqSystemAlert();
+ alert.setRuleId(4L);
+ alert.setFingerprint("fingerprint");
+ alert.setInstanceId("local");
+ alert.setLabelsJson("{\"consumerGroup\":\"orders\"}");
+ RmqSystemAlertMapper alertMapper = mock(RmqSystemAlertMapper.class);
+ when(alertMapper.selectList(any())).thenReturn(List.of(alert));
+ MybatisPlusAlertStateRepository repository = new
MybatisPlusAlertStateRepository(mapper, alertMapper);
+ AlertRuleVO rule =
AlertRuleVO.builder().id(4L).domain(AlertDomain.BUSINESS).enabled(true)
+ .instanceId("local").metric("consumer.lag.total").build();
+
+ List<ActiveAlertState> active = repository.findActive(new
MetricCollectionScope(AlertDomain.BUSINESS,
+ "local", Set.of("consumer.lag.total")), List.of(rule));
+
+ assertThat(active).singleElement().satisfies(item -> {
+ assertThat(item.key()).isEqualTo(new AlertStateKey(4L,
"fingerprint"));
+
assertThat(item.state().status()).isEqualTo(AlertStateStatus.ACKED);
+ assertThat(item.instanceId()).isEqualTo("local");
+ assertThat(item.labels()).isEqualTo(Map.of("consumerGroup",
"orders"));
+ });
+ verify(alertMapper).selectList(any());
+ verify(alertMapper, never()).selectOne(any());
+ }
+
+ @Test
+ void findsActiveStatesWithOneLatestAlertMetadataQueryTest() {
+ RmqAlertStateMapper mapper = mock(RmqAlertStateMapper.class);
+ RmqAlertState first = activeState(4L, "fingerprint-a",
AlertStateStatus.FIRING);
+ RmqAlertState second = activeState(5L, "fingerprint-b",
AlertStateStatus.ACKED);
+ when(mapper.selectList(any())).thenReturn(List.of(first, second));
+ RmqSystemAlertMapper alertMapper = mock(RmqSystemAlertMapper.class);
+ RmqSystemAlert firstAlert = alert(4L, "fingerprint-a", "local",
"{\"consumerGroup\":\"orders\"}");
+ RmqSystemAlert secondAlert = alert(5L, "fingerprint-b", "local",
"{\"consumerGroup\":\"payments\"}");
+ RmqSystemAlert staleFirstAlert = alert(4L, "fingerprint-a", "local",
"{\"consumerGroup\":\"old\"}");
+ when(alertMapper.selectList(any())).thenReturn(List.of(firstAlert,
secondAlert, staleFirstAlert));
+ MybatisPlusAlertStateRepository repository = new
MybatisPlusAlertStateRepository(mapper, alertMapper);
+ AlertRuleVO firstRule =
AlertRuleVO.builder().id(4L).domain(AlertDomain.BUSINESS).enabled(true)
+ .instanceId("local").metric("consumer.lag.total").build();
+ AlertRuleVO secondRule =
AlertRuleVO.builder().id(5L).domain(AlertDomain.BUSINESS).enabled(true)
+ .instanceId("local").metric("consumer.lag.total").build();
+
+ List<ActiveAlertState> active = repository.findActive(new
MetricCollectionScope(AlertDomain.BUSINESS,
+ "local", Set.of("consumer.lag.total")), List.of(firstRule,
secondRule));
+
+ assertThat(active).hasSize(2);
+ assertThat(active).extracting(item ->
item.labels().get("consumerGroup"))
+ .containsExactly("orders", "payments");
+ verify(alertMapper, times(1)).selectList(any());
+ verify(alertMapper, never()).selectOne(any());
+ }
+
+ private static RmqAlertState activeState(Long ruleId, String fingerprint,
AlertStateStatus status) {
+ RmqAlertState state = new RmqAlertState();
+ state.setRuleId(ruleId);
+ state.setFingerprint(fingerprint);
+ state.setStatus(status.name());
+ state.setConsecutiveHits(1);
+ state.setCurrentValue(30D);
+ return state;
+ }
+
+ private static RmqSystemAlert alert(Long ruleId, String fingerprint,
String instanceId, String labelsJson) {
+ RmqSystemAlert alert = new RmqSystemAlert();
+ alert.setRuleId(ruleId);
+ alert.setFingerprint(fingerprint);
+ alert.setInstanceId(instanceId);
+ alert.setLabelsJson(labelsJson);
+ return alert;
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
index f93c82ac9..1f94d3410 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.ops.alert;
import org.apache.rocketmq.studio.cluster.metrics.MetricAvailability;
+import org.apache.rocketmq.studio.cluster.metrics.MetricCollectionScope;
import org.apache.rocketmq.studio.cluster.metrics.MetricSample;
import org.apache.rocketmq.studio.cluster.metrics.MetricSnapshotRepository;
import org.junit.jupiter.api.Test;
@@ -30,6 +31,7 @@ import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -334,6 +336,93 @@ class NativeAlertProcessorTest {
verify(outbox).enqueue(any(), org.mockito.ArgumentMatchers.same(rule),
any());
}
+ @Test
+ void resolvesActiveFingerprintMissingFromSuccessfulCollectionScopeTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO rule = rule("local", "orders", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(rule));
+ MetricSample oldSample = sample("orders");
+ AlertStateKey oldKey = new AlertStateKey(rule.getId(),
+ AlertFingerprint.of(rule.getId(), oldSample.instanceId(),
oldSample.labels()));
+ AlertRuleState firing = new AlertRuleState(AlertStateStatus.FIRING, 1,
20D,
+ oldSample.collectedAt().minusSeconds(60),
oldSample.collectedAt().minusSeconds(60),
+ oldSample.collectedAt().minusSeconds(60), null);
+ ActiveAlertState active = new ActiveAlertState(oldKey, firing,
oldSample.instanceId(), oldSample.labels());
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
+ when(states.save(eq(oldKey),
any(AlertRuleState.class))).thenReturn(true);
+ AlertRepository alerts = mock(AlertRepository.class);
+ when(alerts.saveAlert(any(SystemAlertVO.class))).thenAnswer(invocation
-> invocation.getArgument(0));
+ NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
+
+ new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts, outbox,
suppression())
+ .processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ java.util.Set.of("consumer.lag.total")), List.of());
+
+ org.mockito.ArgumentCaptor<AlertRuleState> state =
org.mockito.ArgumentCaptor.forClass(AlertRuleState.class);
+ org.mockito.ArgumentCaptor<SystemAlertVO> event =
org.mockito.ArgumentCaptor.forClass(SystemAlertVO.class);
+ verify(states).save(eq(oldKey), state.capture());
+ verify(alerts).saveAlert(event.capture());
+
assertThat(state.getValue().status()).isEqualTo(AlertStateStatus.RESOLVED);
+
assertThat(event.getValue().getTransition()).isEqualTo(AlertStateTransition.RESOLVED.name());
+ assertThat(event.getValue().getLabels()).isEqualTo(oldSample.labels());
+ verify(outbox).enqueue(any(SystemAlertVO.class), eq(rule),
eq(oldSample.labels()));
+ }
+
+ @Test
+ void keepsActiveFingerprintWhenItAppearsInSuccessfulCollectionScopeTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO rule = rule("local", "orders", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(rule));
+ MetricSample current = sample("orders");
+ AlertStateKey key = new AlertStateKey(rule.getId(),
+ AlertFingerprint.of(rule.getId(), current.instanceId(),
current.labels()));
+ AlertRuleState firing = new AlertRuleState(AlertStateStatus.FIRING, 1,
20D,
+ current.collectedAt().minusSeconds(60),
current.collectedAt().minusSeconds(60),
+ current.collectedAt().minusSeconds(60), null);
+ ActiveAlertState active = new ActiveAlertState(key, firing,
current.instanceId(), current.labels());
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.find(key)).thenReturn(Optional.of(firing));
+ when(states.save(eq(key), any(AlertRuleState.class))).thenReturn(true);
+ when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
+ AlertRepository alerts = mock(AlertRepository.class);
+
+ new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts,
mock(NotificationOutboxService.class), suppression())
+ .processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ java.util.Set.of("consumer.lag.total")),
List.of(current));
+
+ verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
+ }
+
+ @Test
+ void
doesNotResolveMissingActiveStateWhenCollectionReportsWholeScopeUnavailableTest()
{
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO rule = rule("local", "orders", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(rule));
+ MetricSample oldSample = sample("orders");
+ AlertStateKey oldKey = new AlertStateKey(rule.getId(),
+ AlertFingerprint.of(rule.getId(), oldSample.instanceId(),
oldSample.labels()));
+ ActiveAlertState active = new ActiveAlertState(oldKey,
+ new AlertRuleState(AlertStateStatus.FIRING, 1, 20D,
oldSample.collectedAt().minusSeconds(60),
+ oldSample.collectedAt().minusSeconds(60),
oldSample.collectedAt().minusSeconds(60), null),
+ oldSample.instanceId(), oldSample.labels());
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
+ AlertRepository alerts = mock(AlertRepository.class);
+
+ new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts,
mock(NotificationOutboxService.class), suppression())
+ .processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ java.util.Set.of("consumer.lag.total")), List.of(new
MetricSample("consumer.lag.total",
+ AlertDomain.BUSINESS, "local", null, Map.of(), null,
MetricAvailability.UNAVAILABLE,
+ Instant.now(), "BUSINESS_METRICS_COLLECTION_FAILED")));
+
+ verify(states, never()).save(eq(oldKey), any(AlertRuleState.class));
+ verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
+ }
+
private static AlertNotificationSuppressionService suppression() {
AlertNotificationSuppressionService service =
mock(AlertNotificationSuppressionService.class);
when(service.findSuppressingClusterAlert(any(SystemAlertVO.class))).thenReturn(Optional.empty());
