[feature] Support Collector Alarm

alerter/src/main/java/org/apache/hertzbeat/alert/calculate/CollectorAlarmHandler.java

@@ -0,0 +1,199 @@
+/*
+ * 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.hertzbeat.alert.calculate;
+
+import jakarta.persistence.criteria.Predicate;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.alert.AlerterWorkerPool;
+import org.apache.hertzbeat.alert.dao.AlertCollectorDao;
+import org.apache.hertzbeat.alert.service.AlertService;
+import org.apache.hertzbeat.common.constants.CommonConstants;
+import org.apache.hertzbeat.common.entity.alerter.Alert;
+import org.apache.hertzbeat.common.entity.manager.Collector;
+import org.apache.hertzbeat.common.support.event.CollectorDeletedEvent;
+import org.apache.hertzbeat.common.support.event.SystemConfigChangeEvent;
+import org.apache.hertzbeat.common.util.ResourceBundleUtil;
+import org.springframework.context.event.EventListener;
+import org.springframework.data.jpa.domain.Specification;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.ResourceBundle;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.hertzbeat.common.constants.CommonConstants.ALERT_STATUS_CODE_PENDING;
+
+/**
+ * handle collector alarm
+ */
+@Component
+@Slf4j
+public class CollectorAlarmHandler {
+
+    private final Map<String, Alert> offlineAlertMap;
+
+    private final AlertService alertService;
+
+    private final AlertCollectorDao alertCollectorDao;
+
+    private final AlerterWorkerPool workerPool;
+
+    private ResourceBundle bundle;
+
+    public CollectorAlarmHandler(AlertService alertService, AlertCollectorDao alertCollectorDao,
+                                 AlerterWorkerPool workerPool) {
+        this.offlineAlertMap = new ConcurrentHashMap<>(16);
+        this.alertService = alertService;
+        this.alertCollectorDao = alertCollectorDao;
+        this.workerPool = workerPool;
+        this.bundle = ResourceBundleUtil.getBundle("alerter");
+        List<Collector> collectors = this.alertCollectorDao.findCollectorsByStatus(CommonConstants.COLLECTOR_STATUS_OFFLINE);
+        if (!CollectionUtils.isEmpty(collectors)) {
+            for (Collector collector : collectors) {
+                Map<String, String> tags = new HashMap<>(8);
+                tags.put(CommonConstants.TAG_COLLECTOR_ID, String.valueOf(collector.getId()));
+                tags.put(CommonConstants.TAG_COLLECTOR_NAME, collector.getName());
+                this.offlineAlertMap.put(collector.getName(),
+                        Alert.builder().tags(tags).target(CommonConstants.AVAILABILITY).status(ALERT_STATUS_CODE_PENDING).build());
+            }
+        }
+    }
+
+    /**
+     * handle collector online
+     *
+     * @param identity collector name
+     */
+    public void online(final String identity) {
+        Collector collector = alertCollectorDao.findCollectorByName(identity);
+        if (collector == null) {
+            return;
+        }
+        long currentTimeMill = System.currentTimeMillis();
+        Alert preAlert = offlineAlertMap.remove(identity);
+        if (preAlert != null) {
+            Map<String, String> tags = preAlert.getTags();
+            tags.put(CommonConstants.TAG_COLLECTOR_HOST, collector.getIp());
+            String content = this.bundle.getString("alerter.availability.collector.recover");
+            Alert resumeAlert = Alert.builder()
+                    .tags(tags)
+                    .target(CommonConstants.AVAILABILITY_COLLECTOR)
+                    .content(content)
+                    .priority(CommonConstants.ALERT_PRIORITY_CODE_WARNING)
+                    .status(CommonConstants.ALERT_STATUS_CODE_RESTORED)
+                    .firstAlarmTime(currentTimeMill)
+                    .lastAlarmTime(preAlert.getLastAlarmTime())
+                    .build();
+            workerPool.executeJob(() -> recoverAlarm(identity, resumeAlert));
+        }
+    }
+
+    private void recoverAlarm(String identity, Alert restoreAlert) {
+        List<Long> alertIds = queryAvailabilityAlerts(identity, restoreAlert)
+                .stream()
+                .filter(alert -> Objects.equals(alert.getTags().get(CommonConstants.TAG_COLLECTOR_NAME), identity))
+                .map(Alert::getId)
+                .toList();
+
+        if (!alertIds.isEmpty()) {
+            alertService.editAlertStatus(CommonConstants.ALERT_STATUS_CODE_SOLVED, alertIds);
+
+            // Recovery notifications are generated only after an alarm has occurred
+            alertService.addAlert(restoreAlert);
+        }
+    }
+
+    private List<Alert> queryAvailabilityAlerts(String identity, Alert restoreAlert) {
+        //create query condition
+        Specification<Alert> specification = (root, query, criteriaBuilder) -> {
+            List<Predicate> andList = new ArrayList<>();
+
+            Predicate predicateTags = criteriaBuilder.like(root.get("tags").as(String.class), "%" + identity + "%");
+            andList.add(predicateTags);
+
+            Predicate predicatePriority = criteriaBuilder.equal(root.get("priority"), CommonConstants.ALERT_PRIORITY_CODE_EMERGENCY);
+            andList.add(predicatePriority);
+
+            Predicate predicateStatus = criteriaBuilder.equal(root.get("status"), ALERT_STATUS_CODE_PENDING);
+            andList.add(predicateStatus);
+
+            Predicate predicateAlertTime = criteriaBuilder.lessThanOrEqualTo(root.get("lastAlarmTime"), restoreAlert.getLastAlarmTime());
+            andList.add(predicateAlertTime);
+
+            Predicate[] predicates = new Predicate[andList.size()];
+            return criteriaBuilder.and(andList.toArray(predicates));
+        };
+
+        //query results
+        return alertService.getAlerts(specification);
+    }
+
+    /**
+     * handle collector offline
+     *
+     * @param identity collector name
+     */
+    public void offline(final String identity) {
+        Collector collector = alertCollectorDao.findCollectorByName(identity);
+        if (collector == null) {
+            return;
+        }
+        long currentTimeMill = System.currentTimeMillis();
+
+        Alert preAlert = offlineAlertMap.get(identity);
+        if (preAlert == null) {
+            Map<String, String> tags = new HashMap<>();
+            tags.put(CommonConstants.TAG_COLLECTOR_ID, String.valueOf(collector.getId()));
+            tags.put(CommonConstants.TAG_COLLECTOR_NAME, collector.getName());
+            tags.put(CommonConstants.TAG_COLLECTOR_HOST, collector.getIp());
+            tags.put(CommonConstants.TAG_CODE, "OFFLINE");
+
+            String content =  this.bundle.getString("alerter.availability.collector.offline");
+            Alert alert = Alert.builder()
+                    .tags(tags)
+                    .priority(CommonConstants.ALERT_PRIORITY_CODE_EMERGENCY)
+                    .status(CommonConstants.ALERT_STATUS_CODE_PENDING)
+                    .target(CommonConstants.AVAILABILITY_COLLECTOR)
+                    .content(content)
+                    .firstAlarmTime(currentTimeMill)
+                    .lastAlarmTime(currentTimeMill)
+                    .times(1)
+                    .build();
+            this.offlineAlertMap.put(identity, alert);
+            this.alertService.addAlert(alert);
+        }
+    }
+
+
+    @EventListener(SystemConfigChangeEvent.class)
+    public void onSystemConfigChangeEvent(SystemConfigChangeEvent event) {
+        log.info("calculate alarm receive system config change event: {}.", event.getSource());
+        this.bundle = ResourceBundleUtil.getBundle("alerter");
+    }
+
+    @EventListener(CollectorDeletedEvent.class)
+    public void onCollectorDeletedEvent(CollectorDeletedEvent event) {
+        log.info("collector alarm handler receive collector {} has been deleted.", event.getIdentity());
+        offlineAlertMap.remove(event.getIdentity());
+    }
+}

alerter/src/main/java/org/apache/hertzbeat/alert/dao/AlertCollectorDao.java

@@ -0,0 +1,44 @@
+/*
+ * 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.hertzbeat.alert.dao;
+
+import org.apache.hertzbeat.common.entity.manager.Collector;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+
+import java.util.List;
+
+/**
+ * Alert Collector Dao
+ */
+public interface AlertCollectorDao extends JpaRepository<Collector, Long>, JpaSpecificationExecutor<Collector> {
+
+    /**
+     * Query the collector in the specified state
+     * @param status status value
+     * @return collector list
+     */
+    List<Collector> findCollectorsByStatus(Byte status);
+
+    /**
+     * Query collector by name
+     * @param name collector name
+     * @return collector
+     */
+    Collector findCollectorByName(String name);
+}

alerter/src/main/resources/alerter_en_US.properties

@@ -14,6 +14,8 @@
 # limitations under the License.
 
 alerter.availability.recover = Availability Alert Resolved, Monitor Status Normal Now
+alerter.availability.collector.recover = Collector Availability Alert Resolved, The collector is online
+alerter.availability.collector.offline = Collector Availability Alert Notify, The collector is offline
 alerter.alarm.recover = Alert Resolved Notice
 alerter.notify.title = HertzBeat Alert Notify
 alerter.notify.target = Monitor Target

alerter/src/main/resources/alerter_zh_CN.properties

@@ -14,6 +14,8 @@
 # limitations under the License.
 
 alerter.availability.recover = 可用性告警恢复通知, 任务状态已恢复正常
+alerter.availability.collector.recover = 采集器可用性恢复通知，采集器已上线
+alerter.availability.collector.offline = 采集器可用性告警通知，采集器已下线
 alerter.alarm.recover = 告警恢复通知
 alerter.notify.title = HertzBeat告警通知
 alerter.notify.target = 告警目标对象

common/src/main/java/org/apache/hertzbeat/common/constants/CommonConstants.java

@@ -147,6 +147,11 @@ public interface CommonConstants {
      */
     String AVAILABILITY = "availability";
 
+    /**
+     * Collector availability
+     */
+    String AVAILABILITY_COLLECTOR = "collectorAvailability";
+
     /**
      * Parameter Type Number
      */
@@ -192,6 +197,21 @@ public interface CommonConstants {
      */
     byte AUTH_TYPE_GITEE = 5;
 
+    /**
+     * Inside the tag: monitorId
+     */
+    String TAG_COLLECTOR_ID = "collectorId";
+
+    /**
+     * Inside the tag: collectorName
+     */
+    String TAG_COLLECTOR_NAME = "collectorName";
+
+    /**
+     * Inside the tag: collectorHost
+     */
+    String TAG_COLLECTOR_HOST = "collectorHost";
+
     /**
      * Inside the tag: monitorId Monitor task ID
      */

common/src/main/java/org/apache/hertzbeat/common/support/event/CollectorDeletedEvent.java

@@ -0,0 +1,37 @@
+/*
+ * 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.hertzbeat.common.support.event;
+
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * the event for collector delete
+ */
+public class CollectorDeletedEvent extends ApplicationEvent {
+
+    private final String identity;
+
+    public CollectorDeletedEvent(Object source, String identity) {
+        super(source);
+        this.identity = identity;
+    }
+
+    public String getIdentity() {
+        return identity;
+    }
+}

common/src/main/java/org/apache/hertzbeat/common/support/event/MonitorDeletedEvent.java

@@ -20,7 +20,7 @@
 import org.springframework.context.ApplicationEvent;
 
 /**
- * the event for system config change
+ * the event for monitor delete
  */
 public class MonitorDeletedEvent extends ApplicationEvent {
 

manager/src/main/java/org/apache/hertzbeat/manager/scheduler/netty/ManageServer.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.hertzbeat.alert.calculate.CollectorAlarmHandler;
 import org.apache.hertzbeat.common.entity.message.ClusterMsg;
 import org.apache.hertzbeat.common.support.CommonThreadPool;
 import org.apache.hertzbeat.manager.scheduler.CollectorJobScheduler;
@@ -55,6 +56,8 @@ public class ManageServer implements CommandLineRunner {
 
     private final CollectorJobScheduler collectorJobScheduler;
 
+    private final CollectorAlarmHandler collectorAlarmHandler;
+
     private ScheduledExecutorService channelSchedule;
 
     private RemotingServer remotingServer;
@@ -63,9 +66,11 @@ public class ManageServer implements CommandLineRunner {
 
     public ManageServer(final SchedulerProperties schedulerProperties,
                         final CollectorJobScheduler collectorJobScheduler,
-                        final CommonThreadPool threadPool) {
+                        final CommonThreadPool threadPool,
+                        final CollectorAlarmHandler collectorAlarmHandler) {
         this.collectorJobScheduler = collectorJobScheduler;
         this.collectorJobScheduler.setManageServer(this);
+        this.collectorAlarmHandler = collectorAlarmHandler;
         this.init(schedulerProperties, threadPool);
     }
 
@@ -96,6 +101,7 @@ public void start() {
                         channel.closeFuture();
                         this.clientChannelTable.remove(collector);
                         this.collectorJobScheduler.collectorGoOffline(collector);
+                        this.collectorAlarmHandler.offline(collector);
                     }
                 });   
             } catch (Exception e) {
@@ -129,6 +135,7 @@ public void addChannel(final String identity, Channel channel) {
             preChannel.close();
         }
         this.clientChannelTable.put(identity, channel);
+        this.collectorAlarmHandler.online(identity);
     }
 
     public void closeChannel(final String identity) {