RockteMQ-AI commented on code in PR #2533: URL: https://github.com/apache/rocketmq-dashboard/pull/2533#discussion_r3834758553
########## server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java: ########## @@ -0,0 +1,96 @@ +/* + * 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 lombok.RequiredArgsConstructor; +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.NativeAlertProcessor; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.util.List; +import java.time.Duration; +import java.time.Instant; + +/** Runs native collectors independently for each configured instance. */ +@Slf4j +@Component +@RequiredArgsConstructor +@EnableConfigurationProperties(AlertingProperties.class) +public class CollectorScheduler { + private final AlertingProperties properties; + private final InstanceRepository instanceRepository; + private final List<ClusterMetricsCollector> clusterCollectors; + private final List<BusinessMetricsCollector> businessCollectors; + private final MetricSnapshotRepository snapshotRepository; + private final NativeAlertProcessor alertProcessor; + private final AlertCollectionLease collectionLease; + + @Scheduled(fixedDelayString = "${studio.alerting.collection-interval:PT30S}") + public void collect() { + if (!properties.isCollectionEnabled()) { + return; + } + if (!collectionLease.tryAcquire()) { + log.debug("Skipping native alert collection because another Studio replica holds the lease"); + return; + } + for (InstanceVO instance : instanceRepository.findAll()) { + collectClusterMetrics(instance); Review Comment: **[Info]** The `collect()` method iterates all instances and all collectors synchronously within a single scheduled invocation. If a collector blocks (e.g., slow broker response), it delays subsequent collections. Consider wrapping each instance+collector pair in a CompletableFuture or at least adding a per-collector timeout to prevent one slow instance from starving others. ########## server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertStateMachine.java: ########## @@ -0,0 +1,65 @@ +/* + * 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.springframework.stereotype.Component; + +import java.time.Instant; + +@Component +public class AlertStateMachine { + + public AlertStateUpdate advance(AlertRuleState previous, AlertEvaluationResult evaluation, + int requiredConsecutiveSamples, Instant now) { + if (requiredConsecutiveSamples < 1) { + throw new IllegalArgumentException("requiredConsecutiveSamples must be positive"); + } + AlertRuleState state = previous == null ? AlertRuleState.initial() : previous; + if (!evaluation.matches() || evaluation.availability() != MetricAvailability.AVAILABLE) { + return new AlertStateUpdate(state, AlertStateTransition.NONE); + } + if (evaluation.conditionMet()) { + return advanceHit(state, evaluation.currentValue(), requiredConsecutiveSamples, now); + } + return advanceClear(state, evaluation.currentValue(), now); + } + + private AlertStateUpdate advanceHit(AlertRuleState state, Double value, int required, Instant now) { + if (state.status() == AlertStateStatus.FIRING || state.status() == AlertStateStatus.ACKED) { + return new AlertStateUpdate(new AlertRuleState(state.status(), state.consecutiveHits(), value, + state.firstPendingAt(), state.firedAt(), null), AlertStateTransition.NONE); + } + int hits = state.consecutiveHits() + 1; + Instant pendingAt = state.firstPendingAt() == null ? now : state.firstPendingAt(); + if (hits < required) { + return new AlertStateUpdate(new AlertRuleState(AlertStateStatus.PENDING, hits, value, pendingAt, Review Comment: **[Info]** When the state machine transitions from PENDING to FIRING, it creates a new AlertRuleState with `firstPendingAt` preserved from the pending phase. This is correct for tracking when the alert condition first started, but the `firedAt` timestamp is set to `now` rather than the time the threshold was actually crossed. For `requiredConsecutiveSamples > 1`, the actual threshold crossing happened `requiredConsecutiveSamples * interval` ago. This is a minor semantic point — the current behavior is reasonable. ########## server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java: ########## @@ -0,0 +1,92 @@ +/* + * 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.MetricSample; +import org.apache.rocketmq.studio.common.domain.enums.AlertLevel; +import org.springframework.stereotype.Component; + +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.List; +import org.springframework.util.StringUtils; + +/** Applies native samples to persisted rule state and emits only lifecycle transitions. */ +@Component +@RequiredArgsConstructor +public class NativeAlertProcessor { + private final AlertService alertService; + private final AlertRuleEvaluator evaluator; + private final AlertStateMachine stateMachine; + private final AlertStateRepository stateRepository; + private final AlertRepository alertRepository; + private final NotificationOutboxService notificationOutboxService; + + public void process(List<MetricSample> samples) { + for (MetricSample sample : samples) { + for (AlertRuleVO rule : alertService.listRules(sample.domain())) { + if (rule.getId() == null) { + continue; + } + if (!matchesNativeScope(rule, sample)) { + continue; + } + AlertEvaluationResult evaluation = evaluator.evaluate(rule, sample); + if (!evaluation.matches()) { + continue; + } Review Comment: **[Warning]** The `process()` method iterates `samples × rules` with an O(n*m) loop. For each sample, it calls `alertService.listRules(sample.domain())` which loads all rules and filters in-memory. If the rule count grows large, consider caching rules by domain or indexing them to avoid repeated full scans per sample. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
