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 6772afcce fix: isolate native alert rule evaluations (#2697)
6772afcce is described below
commit 6772afccebccbdf1965fdf47d0b82dd80697c034
Author: xdz997 <[email protected]>
AuthorDate: Wed Sep 2 20:10:05 2026 +0800
fix: isolate native alert rule evaluations (#2697)
* fix: isolate native alert rule evaluations
* test: verify native alert transaction rollback
---
.../ops/alert/NativeAlertEvaluationService.java | 156 +++++++++++++++++++++
.../studio/ops/alert/NativeAlertProcessor.java | 75 +++-------
.../NativeAlertEvaluationTransactionTest.java | 101 +++++++++++++
.../studio/ops/alert/NativeAlertProcessorTest.java | 132 ++++++++++++++---
4 files changed, 388 insertions(+), 76 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
new file mode 100644
index 000000000..0de1fac79
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
@@ -0,0 +1,156 @@
+/*
+ * 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.ops.alert;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.rocketmq.studio.cluster.metrics.MetricAvailability;
+import org.apache.rocketmq.studio.cluster.metrics.MetricSample;
+import org.apache.rocketmq.studio.cluster.metrics.MetricSnapshotRepository;
+import org.apache.rocketmq.studio.common.domain.enums.AlertLevel;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeMap;
+
+/**
+ * Evaluates one rule against one native sample in an independent transaction.
A failure rolls
+ * back this evaluation's state, event, and outbox changes without
invalidating other evaluations
+ * from the same collection batch.
+ */
+@Service
+@RequiredArgsConstructor
+public class NativeAlertEvaluationService {
+ private final AlertRuleEvaluator evaluator;
+ private final AlertStateMachine stateMachine;
+ private final AlertStateRepository stateRepository;
+ private final MetricSnapshotRepository snapshotRepository;
+ private final AlertRepository alertRepository;
+ private final NotificationOutboxService notificationOutboxService;
+ private final AlertNotificationSuppressionService
notificationSuppressionService;
+
+ @Transactional(propagation = Propagation.REQUIRES_NEW)
+ public void evaluate(AlertRuleVO rule, MetricSample sample) {
+ MetricSample evaluatedSample = aggregate(rule, sample);
+ AlertEvaluationResult evaluation = evaluator.evaluate(rule,
evaluatedSample);
+ if (!evaluation.matches()) {
+ return;
+ }
+
+ AlertStateKey key = new AlertStateKey(rule.getId(),
+ AlertFingerprint.of(rule.getId(), sample.instanceId(),
sample.labels()));
+ AlertStateUpdate update =
stateMachine.advance(stateRepository.find(key).orElse(null), evaluation,
+ Math.max(1, rule.getConsecutiveSamples()),
AlertRuleDuration.parse(rule.getDuration()),
+ AlertRuleDuration.parse(rule.getReminderInterval()),
sample.collectedAt());
+ if (!stateRepository.save(key, update.state()) ||
!emitsLifecycleEvent(update.transition())) {
+ return;
+ }
+
+ 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();
+ applyNotificationSuppression(event, sample.domain(),
update.transition());
+
+ 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());
+ }
+ }
+
+ /** Reads the optional snapshot window inside this evaluation's
independent transaction. */
+ private MetricSample aggregate(AlertRuleVO rule, MetricSample sample) {
+ if (sample.availability() != MetricAvailability.AVAILABLE ||
rule.getWindowSeconds() <= 0) {
+ return sample;
+ }
+ List<MetricSample> window = snapshotRepository.findRecent(sample,
+
sample.collectedAt().minus(Duration.ofSeconds(rule.getWindowSeconds())));
+ if (window.isEmpty()) {
+ return sample;
+ }
+ double value = switch (rule.getAggregation() == null ? "LAST" :
rule.getAggregation().toUpperCase(Locale.ROOT)) {
+ case "MAX" -> window.stream().mapToDouble(item ->
item.value()).max().orElse(sample.value());
+ case "MIN" -> window.stream().mapToDouble(item ->
item.value()).min().orElse(sample.value());
+ case "AVG" -> window.stream().mapToDouble(item ->
item.value()).average().orElse(sample.value());
+ case "SUM" -> window.stream().mapToDouble(item ->
item.value()).sum();
+ default -> window.get(window.size() - 1).value();
+ };
+ return new MetricSample(sample.metricKey(), sample.domain(),
sample.instanceId(), sample.clusterId(),
+ sample.labels(), value, MetricAvailability.AVAILABLE,
sample.collectedAt());
+ }
+
+ private void applyNotificationSuppression(SystemAlertVO event, AlertDomain
domain,
+ AlertStateTransition transition) {
+ if (!shouldSuppress(domain, transition)) {
+ return;
+ }
+ Optional<SystemAlertVO> cause =
notificationSuppressionService.findSuppressingClusterAlert(event);
+ if (cause.isEmpty()) {
+ return;
+ }
+ SystemAlertVO suppressingAlert = cause.get();
+ event.setNotificationSuppressed(true);
+ event.setSuppressionCauseAlertId(suppressingAlert.getId());
+ event.setSuppressionReason("Suppressed by active cluster incident #" +
suppressingAlert.getId()
+ + ": " + suppressingAlert.getTitle());
+ }
+
+ private static boolean emitsLifecycleEvent(AlertStateTransition
transition) {
+ return transition == AlertStateTransition.FIRING
+ || transition == AlertStateTransition.REMINDER
+ || transition == AlertStateTransition.RESOLVED;
+ }
+
+ private static boolean shouldSuppress(AlertDomain domain,
AlertStateTransition transition) {
+ return domain == AlertDomain.BUSINESS
+ && (transition == AlertStateTransition.FIRING || transition ==
AlertStateTransition.REMINDER);
+ }
+
+ private static AlertLevel level(String severity) {
+ if ("critical".equalsIgnoreCase(severity)) {
+ return AlertLevel.error;
+ }
+ if ("warning".equalsIgnoreCase(severity)) {
+ return AlertLevel.warning;
+ }
+ return AlertLevel.info;
+ }
+}
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 f75560654..fad40db82 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
@@ -17,42 +17,39 @@
package org.apache.rocketmq.studio.ops.alert;
import lombok.RequiredArgsConstructor;
-import org.apache.rocketmq.studio.cluster.metrics.MetricSample;
-import org.apache.rocketmq.studio.cluster.metrics.MetricSnapshotRepository;
+import lombok.extern.slf4j.Slf4j;
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.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.LocalDateTime;
import java.time.ZoneOffset;
+import java.util.EnumMap;
import java.util.List;
-import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
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. */
+@Slf4j
@Component
@RequiredArgsConstructor
public class NativeAlertProcessor {
private final AlertService alertService;
- private final AlertRuleEvaluator evaluator;
+ private final NativeAlertEvaluationService evaluationService;
private final AlertStateMachine stateMachine;
private final AlertStateRepository stateRepository;
- private final MetricSnapshotRepository snapshotRepository;
private final AlertRepository alertRepository;
private final NotificationOutboxService notificationOutboxService;
private final AlertNotificationSuppressionService
notificationSuppressionService;
- @Transactional
public void process(List<MetricSample> samples) {
processSamples(samples);
}
@@ -76,37 +73,24 @@ public class NativeAlertProcessor {
private void processSamples(List<MetricSample> samples) {
Map<AlertDomain, List<AlertRuleVO>> rulesByDomain = new
EnumMap<>(AlertDomain.class);
+ int failedEvaluations = 0;
for (MetricSample sample : samples) {
for (AlertRuleVO rule :
rulesByDomain.computeIfAbsent(sample.domain(), alertService::listRules)) {
- if (rule.getId() == null) {
+ if (!eligible(rule, sample)) {
continue;
}
- if (!rule.isEnabled()) {
- continue;
- }
- if (!NativeAlertRuleScopeMatcher.matches(rule, sample)) {
- continue;
- }
- MetricSample evaluatedSample = aggregate(rule, sample);
- AlertEvaluationResult evaluation = evaluator.evaluate(rule,
evaluatedSample);
- if (!evaluation.matches()) {
- continue;
- }
- AlertStateKey key = new AlertStateKey(rule.getId(),
- AlertFingerprint.of(rule.getId(), sample.instanceId(),
sample.labels()));
- AlertStateUpdate update =
stateMachine.advance(stateRepository.find(key).orElse(null), evaluation,
- Math.max(1, rule.getConsecutiveSamples()),
AlertRuleDuration.parse(rule.getDuration()),
- AlertRuleDuration.parse(rule.getReminderInterval()),
sample.collectedAt());
- if (!stateRepository.save(key, update.state())) {
- continue;
- }
- if (update.transition() == AlertStateTransition.FIRING ||
update.transition() == AlertStateTransition.REMINDER
- || update.transition() ==
AlertStateTransition.RESOLVED) {
- emitLifecycleEvent(rule, key, update, sample.domain(),
sample.instanceId(), sample.metricKey(),
- sample.labels(), sample.collectedAt());
+ try {
+ evaluationService.evaluate(rule, sample);
+ } catch (RuntimeException error) {
+ failedEvaluations++;
+ log.warn("Native alert evaluation failed: ruleId={},
instanceId={}, metric={}, cause={}",
+ rule.getId(), sample.instanceId(),
sample.metricKey(), error.getClass().getSimpleName());
}
}
}
+ if (failedEvaluations > 0) {
+ log.warn("Native alert batch completed with {} failed rule
evaluation(s)", failedEvaluations);
+ }
}
private void reconcileMissingActiveStates(MetricCollectionScope scope,
List<MetricSample> samples) {
@@ -188,26 +172,6 @@ public class NativeAlertProcessor {
&& (transition == AlertStateTransition.FIRING || transition ==
AlertStateTransition.REMINDER);
}
- private MetricSample aggregate(AlertRuleVO rule, MetricSample sample) {
- if (sample.availability() != MetricAvailability.AVAILABLE ||
rule.getWindowSeconds() <= 0) {
- return sample;
- }
- List<MetricSample> window = snapshotRepository.findRecent(sample,
-
sample.collectedAt().minus(Duration.ofSeconds(rule.getWindowSeconds())));
- if (window.isEmpty()) {
- return sample;
- }
- double value = switch (rule.getAggregation() == null ? "LAST" :
rule.getAggregation().toUpperCase(Locale.ROOT)) {
- case "MAX" -> window.stream().mapToDouble(item ->
item.value()).max().orElse(sample.value());
- case "MIN" -> window.stream().mapToDouble(item ->
item.value()).min().orElse(sample.value());
- case "AVG" -> window.stream().mapToDouble(item ->
item.value()).average().orElse(sample.value());
- case "SUM" -> window.stream().mapToDouble(item ->
item.value()).sum();
- default -> window.get(window.size() - 1).value();
- };
- return new MetricSample(sample.metricKey(), sample.domain(),
sample.instanceId(), sample.clusterId(), sample.labels(),
- value, MetricAvailability.AVAILABLE, sample.collectedAt());
- }
-
private static AlertLevel level(String severity) {
if ("critical".equalsIgnoreCase(severity)) {
return AlertLevel.error;
@@ -218,4 +182,7 @@ public class NativeAlertProcessor {
return AlertLevel.info;
}
+ private static boolean eligible(AlertRuleVO rule, MetricSample sample) {
+ return rule.getId() != null && rule.isEnabled() &&
NativeAlertRuleScopeMatcher.matches(rule, sample);
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationTransactionTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationTransactionTest.java
new file mode 100644
index 000000000..5e582563f
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationTransactionTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.ops.alert;
+
+import org.apache.rocketmq.studio.cluster.metrics.MetricAvailability;
+import org.apache.rocketmq.studio.cluster.metrics.MetricSample;
+import org.apache.rocketmq.studio.cluster.metrics.MetricSnapshotRepository;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.time.Instant;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+@SpringBootTest
+class NativeAlertEvaluationTransactionTest {
+ @Autowired
+ private NativeAlertEvaluationService evaluationService;
+ @Autowired
+ private JdbcTemplate jdbcTemplate;
+ @MockBean
+ private AlertRuleEvaluator evaluator;
+ @MockBean
+ private AlertStateMachine stateMachine;
+ @MockBean
+ private AlertStateRepository stateRepository;
+ @MockBean
+ private MetricSnapshotRepository snapshotRepository;
+ @MockBean
+ private AlertRepository alertRepository;
+ @MockBean
+ private NotificationOutboxService outbox;
+ @MockBean
+ private AlertNotificationSuppressionService suppression;
+
+ @BeforeEach
+ void setUp() {
+ jdbcTemplate.execute("CREATE TABLE IF NOT EXISTS native_alert_tx_probe
(id BIGINT PRIMARY KEY)");
+ jdbcTemplate.update("DELETE FROM native_alert_tx_probe");
+ when(evaluator.evaluate(any(), any())).thenReturn(
+ new AlertEvaluationResult(true, true, 20D,
MetricAvailability.AVAILABLE));
+ when(stateRepository.find(any())).thenReturn(Optional.empty());
+ when(stateMachine.advance(any(), any(), any(Integer.class), any(),
any(), any()))
+ .thenReturn(new AlertStateUpdate(new
AlertRuleState(AlertStateStatus.FIRING, 1, 20D, null,
+ Instant.now(), Instant.now(), null),
AlertStateTransition.FIRING));
+ when(stateRepository.save(any(), any())).thenAnswer(invocation -> {
+ jdbcTemplate.update("INSERT INTO native_alert_tx_probe (id) VALUES
(1)");
+ return true;
+ });
+
when(suppression.findSuppressingClusterAlert(any())).thenReturn(Optional.empty());
+ }
+
+ @Test
+ void failedEvaluationRollsBackItsWritesButNextEvaluationCommits() {
+ AtomicBoolean fail = new AtomicBoolean(true);
+ when(alertRepository.saveAlert(any())).thenAnswer(invocation -> {
+ if (fail.get()) {
+ throw new IllegalStateException("event insert failed");
+ }
+ return invocation.getArgument(0);
+ });
+ AlertRuleVO rule =
AlertRuleVO.builder().id(1L).domain(AlertDomain.BUSINESS).name("Orders")
+
.metric("consumer.lag.total").operator(">").threshold(10).enabled(true).instanceId("local")
+ .consumerGroup("orders").build();
+ MetricSample sample = new MetricSample("consumer.lag.total",
AlertDomain.BUSINESS, "local", null,
+ Map.of("consumerGroup", "orders"), 20D,
MetricAvailability.AVAILABLE, Instant.now());
+
+ assertThatThrownBy(() -> evaluationService.evaluate(rule,
sample)).isInstanceOf(IllegalStateException.class);
+ assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM
native_alert_tx_probe", Integer.class))
+ .isZero();
+
+ fail.set(false);
+ evaluationService.evaluate(rule, sample);
+ assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM
native_alert_tx_probe", Integer.class))
+ .isEqualTo(1);
+ }
+}
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 1f94d3410..e7f530711 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
@@ -30,6 +30,8 @@ import java.util.Map;
import java.util.Optional;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
@@ -41,6 +43,73 @@ import static org.mockito.Mockito.when;
class NativeAlertProcessorTest {
+ @Test
+ void continuesWithLaterRulesWhenOneEvaluationFailsTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO failing = rule(1L, "local", "orders", 1);
+ failing.setWindowSeconds(300);
+ AlertRuleVO healthy = rule(2L, "local", "orders", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(failing,
healthy));
+ MetricSnapshotRepository snapshots =
mock(MetricSnapshotRepository.class);
+ when(snapshots.findRecent(any(MetricSample.class), any(Instant.class)))
+ .thenThrow(new IllegalStateException("snapshot read failed"));
+ AlertStateRepository states = mock(AlertStateRepository.class);
+
when(states.find(any(AlertStateKey.class))).thenReturn(Optional.empty());
+ when(states.save(any(AlertStateKey.class),
any(AlertRuleState.class))).thenReturn(true);
+
+ NativeAlertProcessor processor = processor(service, states, snapshots,
mock(AlertRepository.class),
+ mock(NotificationOutboxService.class), suppression());
+
+ assertThatCode(() ->
processor.process(List.of(sample("orders")))).doesNotThrowAnyException();
+
+ verify(states).save(org.mockito.ArgumentMatchers.argThat(key ->
key.ruleId().equals(2L)),
+ any(AlertRuleState.class));
+ }
+
+ @Test
+ void continuesWithLaterRulesWhenAlertPersistenceFailsTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO failing = rule(1L, "local", "orders", 1);
+ AlertRuleVO healthy = rule(2L, "local", "orders", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(failing,
healthy));
+ AlertStateRepository states = mock(AlertStateRepository.class);
+
when(states.find(any(AlertStateKey.class))).thenReturn(Optional.empty());
+ when(states.save(any(AlertStateKey.class),
any(AlertRuleState.class))).thenReturn(true);
+ AlertRepository alerts = mock(AlertRepository.class);
+ when(alerts.saveAlert(any(SystemAlertVO.class)))
+ .thenThrow(new IllegalStateException("event insert failed"))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+ NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
+
+ NativeAlertProcessor processor = processor(service, states,
mock(MetricSnapshotRepository.class), alerts,
+ outbox, suppression());
+
+ assertThatCode(() ->
processor.process(List.of(sample("orders")))).doesNotThrowAnyException();
+
+ verify(states).save(org.mockito.ArgumentMatchers.argThat(key ->
key.ruleId().equals(2L)),
+ any(AlertRuleState.class));
+ verify(outbox).enqueue(any(SystemAlertVO.class),
org.mockito.ArgumentMatchers.same(healthy),
+ org.mockito.ArgumentMatchers.anyMap());
+ }
+
+ @Test
+ void doesNotSwallowErrorsTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO rule = rule(1L, "local", "orders", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(rule));
+ NativeAlertEvaluationService evaluationService =
mock(NativeAlertEvaluationService.class);
+ org.mockito.Mockito.doThrow(new AssertionError("fatal evaluation
failure"))
+ .when(evaluationService).evaluate(any(AlertRuleVO.class),
any(MetricSample.class));
+
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
evaluationService,
+ new AlertStateMachine(), mock(AlertStateRepository.class),
mock(AlertRepository.class),
+ mock(NotificationOutboxService.class), suppression());
+
+ assertThatThrownBy(() -> processor.process(List.of(sample("orders"))))
+ .isInstanceOf(AssertionError.class)
+ .hasMessage("fatal evaluation failure");
+ }
+
@Test
void requiresInstanceScopeBeforeProcessingNativeSamplesTest() {
AlertService service = mock(AlertService.class);
@@ -155,8 +224,8 @@ class NativeAlertProcessorTest {
}
};
- new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states, snapshots,
- mock(AlertRepository.class),
mock(NotificationOutboxService.class), suppression()).process(List.of(current));
+ processor(service, states, snapshots, mock(AlertRepository.class),
mock(NotificationOutboxService.class),
+ suppression()).process(List.of(current));
assertThat(saved.values()).singleElement().satisfies(state -> {
assertThat(state.status()).isEqualTo(AlertStateStatus.FIRING);
@@ -198,8 +267,8 @@ class NativeAlertProcessorTest {
}
};
- new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states, snapshots,
- mock(AlertRepository.class),
mock(NotificationOutboxService.class), suppression()).process(List.of(current));
+ processor(service, states, snapshots, mock(AlertRepository.class),
mock(NotificationOutboxService.class),
+ suppression()).process(List.of(current));
assertThat(saved.values()).singleElement().satisfies(state -> {
assertThat(state.status()).isEqualTo(AlertStateStatus.FIRING);
@@ -225,9 +294,8 @@ class NativeAlertProcessorTest {
when(states.find(any(AlertStateKey.class))).thenReturn(Optional.empty());
when(states.save(any(AlertStateKey.class),
any(AlertRuleState.class))).thenReturn(true);
- new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states, snapshots,
- mock(AlertRepository.class),
mock(NotificationOutboxService.class), suppression())
- .process(List.of(current));
+ processor(service, states, snapshots, mock(AlertRepository.class),
mock(NotificationOutboxService.class),
+ suppression()).process(List.of(current));
org.mockito.ArgumentCaptor<AlertRuleState> saved =
org.mockito.ArgumentCaptor.forClass(AlertRuleState.class);
verify(states).save(any(AlertStateKey.class), saved.capture());
@@ -273,17 +341,26 @@ class NativeAlertProcessorTest {
});
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
- new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states,
- mock(MetricSnapshotRepository.class), alerts, outbox,
suppression()).process(List.of(sample("orders")));
+ processor(service, states, mock(MetricSnapshotRepository.class),
alerts, outbox, suppression())
+ .process(List.of(sample("orders")));
verify(outbox).enqueue(any(SystemAlertVO.class),
org.mockito.ArgumentMatchers.same(rule),
org.mockito.ArgumentMatchers.anyMap());
}
+ private static NativeAlertProcessor processor(AlertService service,
AlertStateRepository states,
+ MetricSnapshotRepository snapshots, AlertRepository alerts,
NotificationOutboxService outbox,
+ AlertNotificationSuppressionService suppression) {
+ NativeAlertEvaluationService evaluationService = new
NativeAlertEvaluationService(new AlertRuleEvaluator(),
+ new AlertStateMachine(), states, snapshots, alerts, outbox,
suppression);
+ return new NativeAlertProcessor(service, evaluationService, new
AlertStateMachine(), states, alerts, outbox,
+ suppression);
+ }
+
private static NativeAlertProcessor processor(AlertService service,
AlertStateRepository states,
AlertRepository alerts) {
- return new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states,
- mock(MetricSnapshotRepository.class), alerts,
mock(NotificationOutboxService.class), suppression());
+ return processor(service, states,
mock(MetricSnapshotRepository.class), alerts,
+ mock(NotificationOutboxService.class), suppression());
}
@Test
@@ -302,8 +379,8 @@ class NativeAlertProcessorTest {
SystemAlertVO clusterCause =
SystemAlertVO.builder().id(11L).title("Broker unavailable").build();
when(suppression.findSuppressingClusterAlert(any(SystemAlertVO.class))).thenReturn(Optional.of(clusterCause));
- new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states,
- mock(MetricSnapshotRepository.class), alerts, outbox,
suppression).process(List.of(sample("orders")));
+ processor(service, states, mock(MetricSnapshotRepository.class),
alerts, outbox, suppression)
+ .process(List.of(sample("orders")));
org.mockito.ArgumentCaptor<SystemAlertVO> event =
org.mockito.ArgumentCaptor.forClass(SystemAlertVO.class);
verify(alerts).saveAlert(event.capture());
@@ -328,8 +405,7 @@ class NativeAlertProcessorTest {
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
AlertNotificationSuppressionService suppression =
mock(AlertNotificationSuppressionService.class);
- new NativeAlertProcessor(service, new AlertRuleEvaluator(), new
AlertStateMachine(), states,
- mock(MetricSnapshotRepository.class), alerts, outbox,
suppression)
+ processor(service, states, mock(MetricSnapshotRepository.class),
alerts, outbox, suppression)
.process(List.of(sample("orders", 0D)));
verify(suppression, never()).findSuppressingClusterAlert(any());
@@ -355,8 +431,10 @@ class NativeAlertProcessorTest {
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())
+ new NativeAlertProcessor(service,
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts, outbox,
suppression()),
+ new AlertStateMachine(), states, alerts, outbox, suppression())
.processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
java.util.Set.of("consumer.lag.total")), List.of());
@@ -388,8 +466,11 @@ class NativeAlertProcessorTest {
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())
+ new NativeAlertProcessor(service,
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts,
mock(NotificationOutboxService.class),
+ suppression()),
+ new AlertStateMachine(), states, alerts,
mock(NotificationOutboxService.class), suppression())
.processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
java.util.Set.of("consumer.lag.total")),
List.of(current));
@@ -412,8 +493,11 @@ class NativeAlertProcessorTest {
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())
+ new NativeAlertProcessor(service,
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts,
mock(NotificationOutboxService.class),
+ suppression()),
+ new AlertStateMachine(), states, 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,
@@ -430,7 +514,11 @@ class NativeAlertProcessorTest {
}
private static AlertRuleVO rule(String instanceId, String group, int
consecutiveSamples) {
- return
AlertRuleVO.builder().id(1L).domain(AlertDomain.BUSINESS).name("Orders lag")
+ return rule(1L, instanceId, group, consecutiveSamples);
+ }
+
+ private static AlertRuleVO rule(Long id, String instanceId, String group,
int consecutiveSamples) {
+ return
AlertRuleVO.builder().id(id).domain(AlertDomain.BUSINESS).name("Orders lag")
.metric("consumer.lag.total").operator(">").threshold(10).enabled(true)
.instanceId(instanceId).consumerGroup(group).consecutiveSamples(consecutiveSamples).build();
}
