RockteMQ-AI commented on code in PR #2533: URL: https://github.com/apache/rocketmq-dashboard/pull/2533#discussion_r3834866838
########## 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); + collectBusinessMetrics(instance); + } + } + Review Comment: **[Info]** Sequential instance iteration: `collect()` iterates all instances sequentially. If one instance has high latency (slow broker response), it delays collection for all subsequent instances. Consider parallelizing with `CompletableFuture.supplyAsync()` or a bounded thread pool for large multi-instance deployments. ########## server/src/main/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxService.java: ########## @@ -0,0 +1,192 @@ +/* + * 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. + */ +package org.apache.rocketmq.studio.ops.alert; + +import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.studio.audit.OperationAuditService; +import org.apache.rocketmq.studio.common.util.NoRedirectClientHttpRequestFactory; +import org.apache.rocketmq.studio.common.util.UrlHostGuard; +import org.apache.rocketmq.studio.persistence.entity.RmqAlertNotificationOutbox; +import org.apache.rocketmq.studio.persistence.mapper.RmqAlertNotificationOutboxMapper; +import org.apache.rocketmq.studio.settings.GeneralSettingsVO; +import org.apache.rocketmq.studio.settings.SettingsRepository; +import org.springframework.http.ResponseEntity; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; +import org.springframework.web.client.RestTemplate; + +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Persists notification work so collection never blocks on remote webhook availability. */ +@Slf4j +@Service +public class NotificationOutboxService { + private static final int MAX_ATTEMPTS = 5; + private static final int BATCH_SIZE = 20; + + private final RmqAlertNotificationOutboxMapper mapper; + private final SettingsRepository settingsRepository; + private final AlertSilenceService silenceService; + private final AlertRepository alertRepository; + private final OperationAuditService operationAuditService; + private final RestTemplate restTemplate; + + public NotificationOutboxService(RmqAlertNotificationOutboxMapper mapper, SettingsRepository settingsRepository, + AlertSilenceService silenceService, AlertRepository alertRepository, + OperationAuditService operationAuditService) { + this(mapper, settingsRepository, silenceService, alertRepository, operationAuditService, newClient()); + } + + NotificationOutboxService(RmqAlertNotificationOutboxMapper mapper, SettingsRepository settingsRepository, + AlertSilenceService silenceService, AlertRepository alertRepository, + OperationAuditService operationAuditService, RestTemplate restTemplate) { + this.mapper = mapper; + this.settingsRepository = settingsRepository; + this.silenceService = silenceService; + this.alertRepository = alertRepository; + this.operationAuditService = operationAuditService; + this.restTemplate = restTemplate; + } + + public void enqueue(SystemAlertVO alert, AlertRuleVO rule) { + enqueue(alert, rule, Map.of()); + } + + public void enqueue(SystemAlertVO alert, AlertRuleVO rule, Map<String, String> labels) { + boolean silenced = labels == null || labels.isEmpty() + ? silenceService.isActive(rule, alert.getInstanceId(), alert.getTime()) + : silenceService.isActive(rule, alert.getInstanceId(), labels, alert.getTime()); + if (alert.getId() == null || silenced) { + return; + } + Set<String> channels = new LinkedHashSet<>(); + if (rule.getChannels() != null) { + rule.getChannels().stream().filter(StringUtils::hasText) + .map(value -> value.trim().toLowerCase()).forEach(channels::add); + } + for (String channel : channels) { + if (!"dingtalk".equals(channel) && !"sms".equals(channel)) { + continue; + } + RmqAlertNotificationOutbox row = new RmqAlertNotificationOutbox(); + row.setAlertId(alert.getId()); + row.setChannel(channel); Review Comment: **[Info]** Timezone consistency: `dispatch()` uses `LocalDateTime.now()` for `dispatchedAt`, while `NativeAlertProcessor` uses `Instant.now()` and `ZoneOffset.UTC` for alert timestamps. Consider standardizing on `Instant`/UTC across the alert pipeline to avoid timezone-related ordering issues in multi-region deployments. ########## server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertSilenceService.java: ########## @@ -0,0 +1,111 @@ +/* + * 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.audit.OperationAuditService; +import org.apache.rocketmq.studio.auth.AuthenticatedUserContext; +import org.apache.rocketmq.studio.common.exception.BusinessException; +import org.springframework.stereotype.Service; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.LinkedHashMap; +import java.util.Map; + +@Service +@RequiredArgsConstructor +public class AlertSilenceService { + private final AlertSilenceRepository repository; + private final OperationAuditService operationAuditService; + + public List<AlertSilenceVO> list() { + return repository.findAll(); + } + + public AlertSilenceVO create(CreateAlertSilenceDTO request) { + if (request == null || request.getStartsAt() == null || request.getEndsAt() == null) { + throw new BusinessException(400, "Silence start and end times are required"); + } + if (!request.getEndsAt().isAfter(request.getStartsAt())) { + throw new BusinessException(400, "Silence end time must be after start time"); + } + if (request.getReason() != null && request.getReason().length() > 512) { + throw new BusinessException(400, "Silence reason must not exceed 512 characters"); + } + AlertSilenceVO silence = AlertSilenceVO.builder().domain(request.getDomain()) + .ruleId(request.getRuleId()).instanceId(trimToNull(request.getInstanceId())) + .labels(normalizeLabels(request.getLabels())) + .startsAt(request.getStartsAt()).endsAt(request.getEndsAt()) + .reason(trimToNull(request.getReason())) + .createdBy(AuthenticatedUserContext.currentUsernameOrSystem()).build(); + AlertSilenceVO saved = repository.save(silence); + operationAuditService.record("CREATE_ALERT_SILENCE", "ALERT_SILENCE", String.valueOf(saved.getId()), + saved.getInstanceId(), "ruleId=" + saved.getRuleId(), "SUCCESS", null); + return saved; + } + + public void delete(Long id) { + if (id == null) { + throw new BusinessException(400, "Silence ID is required"); + } + if (!repository.deleteById(id)) { + throw new BusinessException(404, "Alert silence not found: " + id); + } + operationAuditService.record("DELETE_ALERT_SILENCE", "ALERT_SILENCE", String.valueOf(id), null, null, + "SUCCESS", null); + } + + public boolean isActive(AlertRuleVO rule, String instanceId, LocalDateTime now) { + return isActive(rule, instanceId, Map.of(), now); + } + + public boolean isActive(AlertRuleVO rule, String instanceId, Map<String, String> labels, LocalDateTime now) { + AlertDomain domain = rule.getDomain() == null ? AlertDomain.BUSINESS : rule.getDomain(); + return repository.findAll().stream().anyMatch(silence -> matches(silence, rule.getId(), domain, instanceId, + labels == null ? Map.of() : labels, now)); + } + + private static boolean matches(AlertSilenceVO silence, Long ruleId, AlertDomain domain, String instanceId, + Map<String, String> labels, LocalDateTime now) { + return !now.isBefore(silence.getStartsAt()) && now.isBefore(silence.getEndsAt()) + && (silence.getDomain() == null || silence.getDomain() == domain) + && (silence.getRuleId() == null || silence.getRuleId().equals(ruleId)) + && (silence.getInstanceId() == null || silence.getInstanceId().equals(instanceId)) + && (silence.getLabels() == null || labels.entrySet().containsAll(silence.getLabels().entrySet())); + } + + private static String trimToNull(String value) { + return value == null || value.isBlank() ? null : value.trim(); + } + Review Comment: **[Info]** `isActive()` calls `repository.findAll()` on every evaluation. If there are many silences and many rules being evaluated per collection cycle, this could become a hot path. Consider caching active silences with a short TTL (e.g., 10s) or loading them once per collection cycle and passing as a parameter. -- 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]
