WIP: Add resource-based alert rules with evaluation engine, event bus integration, and UI

api/src/main/java/com/cloud/event/EventTypes.java

@@ -822,6 +822,11 @@ public class EventTypes {
     public static final String EVENT_QUOTA_TARIFF_DELETE = "QUOTA.TARIFF.DELETE";
     public static final String EVENT_QUOTA_TARIFF_UPDATE = "QUOTA.TARIFF.UPDATE";
 
+    // Resource alert rules
+    public static final String EVENT_RESOURCE_ALERT_RULE_CREATE = "RESOURCE.ALERT.RULE.CREATE";
+    public static final String EVENT_RESOURCE_ALERT_RULE_UPDATE = "RESOURCE.ALERT.RULE.UPDATE";
+    public static final String EVENT_RESOURCE_ALERT_RULE_DELETE = "RESOURCE.ALERT.RULE.DELETE";
+
     // Routing
     public static final String EVENT_ZONE_IP4_SUBNET_CREATE = "ZONE.IP4.SUBNET.CREATE";
     public static final String EVENT_ZONE_IP4_SUBNET_UPDATE = "ZONE.IP4.SUBNET.UPDATE";

api/src/main/java/org/apache/cloudstack/api/ApiConstants.java

@@ -1089,6 +1089,13 @@ public class ApiConstants {
     public static final String AGGR_FUNCTION = "aggrfunction";
     public static final String AGGR_VALUE = "aggrvalue";
     public static final String THRESHOLD = "threshold";
+    public static final String METRIC = "metric";
+    public static final String CONDITION = "condition";
+    public static final String SEVERITY = "severity";
+    public static final String RESET_INTERVAL = "resetinterval";
+    public static final String WEBHOOK_IDS = "webhookids";
+    public static final String CLEANUP_WEBHOOKS = "cleanupwebhooks";
+    public static final String ALERT_RULE_ID = "alertruleid";
     public static final String RELATIONAL_OPERATOR = "relationaloperator";
     public static final String OTHER_DEPLOY_PARAMS = "otherdeployparams";
     public static final String MIN_MEMBERS = "minmembers";

client/pom.xml

@@ -432,6 +432,11 @@
             <artifactId>cloud-mom-webhook</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.cloudstack</groupId>
+            <artifactId>cloud-plugin-resource-alerts</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.cloudstack</groupId>
             <artifactId>cloud-framework-agent-lb</artifactId>

engine/schema/src/main/java/com/cloud/storage/dao/VolumeDao.java

@@ -34,6 +34,8 @@ public interface VolumeDao extends GenericDao<VolumeVO, Long>, StateDao<Volume.S
 
     List<VolumeVO> findByAccount(long accountId);
 
+    List<Long> listIdsByAccountOrDomainsAndState(Long accountId, List<Long> domainIds, Volume.State state);
+
     List<VolumeVO> findIncludingRemovedByAccount(long accountId);
 
     Pair<Long, Long> getCountAndTotalByPool(long poolId);

engine/schema/src/main/java/com/cloud/storage/dao/VolumeDaoImpl.java

@@ -75,6 +75,7 @@ public class VolumeDaoImpl extends GenericDaoBase<VolumeVO, Long> implements Vol
     private final SearchBuilder<VolumeVO> storeAndInstallPathSearch;
     private final SearchBuilder<VolumeVO> volumeIdSearch;
     protected GenericSearchBuilder<VolumeVO, Long> CountByAccount;
+    protected final GenericSearchBuilder<VolumeVO, Long> IdsByAccountOrDomainsAndStateSearch;
     protected final SearchBuilder<VolumeVO> ExternalUuidSearch;
     protected GenericSearchBuilder<VolumeVO, SumCount> primaryStorageSearch;
     protected GenericSearchBuilder<VolumeVO, SumCount> primaryStorageSearch2;
@@ -118,6 +119,21 @@ public List<VolumeVO> findByAccount(long accountId) {
         return listBy(sc);
     }
 
+    @Override
+    public List<Long> listIdsByAccountOrDomainsAndState(Long accountId, List<Long> domainIds, Volume.State state) {
+        SearchCriteria<Long> sc = IdsByAccountOrDomainsAndStateSearch.create();
+        if (accountId != null) {
+            sc.setParameters("accountId", accountId);
+        }
+        if (domainIds != null) {
+            sc.setParameters("domainIds", domainIds.toArray());
+        }
+        if (state != null) {
+            sc.setParameters("state", state);
+        }
+        return customSearch(sc, null);
+    }
+
     @Override
     public List<VolumeVO> findIncludingRemovedByAccount(long accountId) {
         SearchCriteria<VolumeVO> sc = AllFieldsSearch.create();
@@ -420,6 +436,13 @@ public VolumeDaoImpl() {
         AllFieldsSearch.and("kmsWrappedKeyId", AllFieldsSearch.entity().getKmsWrappedKeyId(), Op.EQ);
         AllFieldsSearch.done();
 
+        IdsByAccountOrDomainsAndStateSearch = createSearchBuilder(Long.class);
+        IdsByAccountOrDomainsAndStateSearch.selectFields(IdsByAccountOrDomainsAndStateSearch.entity().getId());
+        IdsByAccountOrDomainsAndStateSearch.and("accountId", IdsByAccountOrDomainsAndStateSearch.entity().getAccountId(), Op.EQ);
+        IdsByAccountOrDomainsAndStateSearch.and("domainIds", IdsByAccountOrDomainsAndStateSearch.entity().getDomainId(), Op.IN);
+        IdsByAccountOrDomainsAndStateSearch.and("state", IdsByAccountOrDomainsAndStateSearch.entity().getState(), Op.EQ);
+        IdsByAccountOrDomainsAndStateSearch.done();
+
         RootDiskStateSearch = createSearchBuilder();
         RootDiskStateSearch.and("state", RootDiskStateSearch.entity().getState(), Op.IN);
         RootDiskStateSearch.and("vType", RootDiskStateSearch.entity().getVolumeType(), Op.EQ);

engine/schema/src/main/java/com/cloud/vm/dao/UserVmDao.java

@@ -31,6 +31,8 @@
 public interface UserVmDao extends GenericDao<UserVmVO, Long> {
     List<UserVmVO> listByAccountId(long id);
 
+    List<Long> listIdsByAccountOrDomainsAndState(Long accountId, List<Long> domainIds, State state);
+
     List<UserVmVO> listByAccountAndPod(long accountId, long podId);
 
     List<UserVmVO> listByAccountAndDataCenter(long accountId, long dcId);

engine/schema/src/main/java/com/cloud/vm/dao/UserVmDaoImpl.java

@@ -82,6 +82,7 @@ public class UserVmDaoImpl extends GenericDaoBase<UserVmVO, Long> implements Use
     protected SearchBuilder<UserVmVO> AccountDataCenterVirtualSearch;
     protected GenericSearchBuilder<UserVmVO, Long> CountByAccountPod;
     protected GenericSearchBuilder<UserVmVO, Long> CountByAccount;
+    protected GenericSearchBuilder<UserVmVO, Long> IdsByAccountOrDomainsAndStateSearch;
     protected GenericSearchBuilder<UserVmVO, Long> CountActiveAccount;
     protected GenericSearchBuilder<UserVmVO, Long> PodsHavingVmsForAccount;
 
@@ -143,6 +144,13 @@ void init() {
         AccountSearch.and("account", AccountSearch.entity().getAccountId(), SearchCriteria.Op.EQ);
         AccountSearch.done();
 
+        IdsByAccountOrDomainsAndStateSearch = createSearchBuilder(Long.class);
+        IdsByAccountOrDomainsAndStateSearch.selectFields(IdsByAccountOrDomainsAndStateSearch.entity().getId());
+        IdsByAccountOrDomainsAndStateSearch.and("accountId", IdsByAccountOrDomainsAndStateSearch.entity().getAccountId(), SearchCriteria.Op.EQ);
+        IdsByAccountOrDomainsAndStateSearch.and("domainIds", IdsByAccountOrDomainsAndStateSearch.entity().getDomainId(), SearchCriteria.Op.IN);
+        IdsByAccountOrDomainsAndStateSearch.and("state", IdsByAccountOrDomainsAndStateSearch.entity().getState(), SearchCriteria.Op.EQ);
+        IdsByAccountOrDomainsAndStateSearch.done();
+
         IdsSearch = createSearchBuilder();
         IdsSearch.and("ids", IdsSearch.entity().getId(), SearchCriteria.Op.IN);
         IdsSearch.done();
@@ -318,6 +326,21 @@ public List<UserVmVO> listByAccountId(long id) {
         return listBy(sc);
     }
 
+    @Override
+    public List<Long> listIdsByAccountOrDomainsAndState(Long accountId, List<Long> domainIds, State state) {
+        SearchCriteria<Long> sc = IdsByAccountOrDomainsAndStateSearch.create();
+        if (accountId != null) {
+            sc.setParameters("accountId", accountId);
+        }
+        if (domainIds != null) {
+            sc.setParameters("domainIds", domainIds.toArray());
+        }
+        if (state != null) {
+            sc.setParameters("state", state);
+        }
+        return customSearch(sc, null);
+    }
+
     @Override
     public List<UserVmVO> listByHostId(Long id) {
         SearchCriteria<UserVmVO> sc = HostSearch.create();

engine/schema/src/main/resources/META-INF/db/schema-42300to2400.sql

@@ -18,3 +18,57 @@
 --;
 -- Schema upgrade from 4.23.0.0 to 24.0.0
 --;
+
+-- resource_alert_rules: stores per-resource or generic metric threshold rules
+CREATE TABLE IF NOT EXISTS `cloud`.`resource_alert_rules` (
+    `id` bigint unsigned NOT NULL AUTO_INCREMENT,
+    `uuid` varchar(255) NOT NULL UNIQUE,
+    `name` varchar(255) NOT NULL,
+    `resource_type` varchar(64) NOT NULL COMMENT 'VirtualMachine, Volume, Host, StoragePool',
+    `resource_id` bigint unsigned DEFAULT NULL COMMENT 'null = applies to all resources of the type in scope',
+    `account_id` bigint unsigned NOT NULL,
+    `domain_id` bigint unsigned NOT NULL,
+    `metric` varchar(64) NOT NULL,
+    `condition_operator` varchar(8) NOT NULL COMMENT 'GT, GTE, LT, LTE, EQ',
+    `threshold` double NOT NULL,
+    `severity` varchar(32) NOT NULL COMMENT 'CRITICAL, HIGH, MEDIUM, LOW',
+    `message` varchar(4096) DEFAULT NULL,
+    `email` tinyint(1) NOT NULL DEFAULT 0,
+    `reset_interval` int unsigned NOT NULL DEFAULT 600 COMMENT 'minimum seconds between repeat firings of this rule',
+    `created` datetime DEFAULT NULL,
+    `updated` datetime DEFAULT NULL,
+    `removed` datetime DEFAULT NULL,
+    PRIMARY KEY (`id`),
+    INDEX `i_resource_alert_rules__account_id`(`account_id`),
+    INDEX `i_resource_alert_rules__domain_id`(`domain_id`),
+    CONSTRAINT `fk_resource_alert_rules__account_id` FOREIGN KEY (`account_id`) REFERENCES `account`(`id`) ON DELETE CASCADE,
+    CONSTRAINT `fk_resource_alert_rules__domain_id` FOREIGN KEY (`domain_id`) REFERENCES `domain`(`id`) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- resource_alerts: immutable log of fired alerts
+CREATE TABLE IF NOT EXISTS `cloud`.`resource_alerts` (
+    `id` bigint unsigned NOT NULL AUTO_INCREMENT,
+    `uuid` varchar(255) NOT NULL UNIQUE,
+    `alert_rule_id` bigint unsigned NOT NULL,
+    `resource_id` bigint unsigned DEFAULT NULL COMMENT 'the specific resource that triggered the alert',
+    `metric_type` varchar(64) NOT NULL,
+    `metric_value` double NOT NULL,
+    `severity` varchar(32) NOT NULL,
+    `message` varchar(4096) DEFAULT NULL,
+    `alert_timestamp` datetime NOT NULL,
+    PRIMARY KEY (`id`),
+    INDEX `i_resource_alerts__alert_rule_id`(`alert_rule_id`),
+    INDEX `i_resource_alerts__alert_timestamp`(`alert_timestamp`),
+    CONSTRAINT `fk_resource_alerts__alert_rule_id` FOREIGN KEY (`alert_rule_id`) REFERENCES `resource_alert_rules`(`id`) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+-- resource_alert_rules_webhook: webhooks a rule delivers its alerts to
+CREATE TABLE IF NOT EXISTS `cloud`.`resource_alert_rules_webhook` (
+    `id` bigint unsigned NOT NULL AUTO_INCREMENT,
+    `resource_alert_rule_id` bigint unsigned NOT NULL,
+    `webhook_id` bigint unsigned NOT NULL,
+    PRIMARY KEY (`id`),
+    UNIQUE KEY `uc_resource_alert_rules_webhook__rule_webhook`(`resource_alert_rule_id`, `webhook_id`),
+    CONSTRAINT `fk_resource_alert_rules_webhook__rule_id` FOREIGN KEY (`resource_alert_rule_id`) REFERENCES `resource_alert_rules`(`id`) ON DELETE CASCADE,
+    CONSTRAINT `fk_resource_alert_rules_webhook__webhook_id` FOREIGN KEY (`webhook_id`) REFERENCES `webhook`(`id`) ON DELETE CASCADE
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

engine/schema/src/main/resources/META-INF/db/views/cloud.resource_alert_rule_view.sql

@@ -0,0 +1,48 @@
+-- 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.
+
+-- VIEW `cloud`.`resource_alert_rule_view`;
+
+DROP VIEW IF EXISTS `cloud`.`resource_alert_rule_view`;
+CREATE VIEW `cloud`.`resource_alert_rule_view` AS
+    SELECT
+        r.id,
+        r.uuid,
+        r.name,
+        r.resource_type,
+        r.resource_id,
+        r.metric,
+        r.condition_operator,
+        r.threshold,
+        r.severity,
+        r.message,
+        r.email,
+        r.reset_interval,
+        r.created,
+        r.updated,
+        r.removed,
+        a.id account_id,
+        a.uuid account_uuid,
+        a.account_name,
+        a.type account_type,
+        d.id domain_id,
+        d.uuid domain_uuid,
+        d.name domain_name,
+        d.path domain_path
+    FROM `cloud`.`resource_alert_rules` r
+    INNER JOIN `cloud`.`account` a ON r.account_id = a.id
+    INNER JOIN `cloud`.`domain` d ON r.domain_id = d.id;

plugins/event-bus/webhook/src/main/java/org/apache/cloudstack/mom/webhook/WebhookServiceImpl.java

@@ -121,13 +121,27 @@ protected WebhookDeliveryThread getDeliveryJob(Event event, Webhook webhook, Del
                 .setContext(context);
         WebhookDeliveryThread job = new WebhookDeliveryThread(webhook, event, caller);
         job = ComponentContext.inject(job);
+        applyDeliveryConfig(job, config);
+        return job;
+    }
+
+    protected DeliveryConfig getDeliveryConfig(long domainId) {
+        return new DeliveryConfig(
+                WebhookDeliveryTries.valueIn(domainId),
+                WebhookDeliveryTimeout.valueIn(domainId),
+                WebhookDeliveryBlocklist.valueIn(domainId),
+                WebhookDeliveryBlockLocalAddresses.value(),
+                WebhookDeliveryAllowRedirects.valueIn(domainId),
+                WebhookDeliveryAllowHttp.valueIn(domainId));
+    }
+
+    protected void applyDeliveryConfig(WebhookDeliveryThread job, DeliveryConfig config) {
         job.setDeliveryTries(config.tries);
         job.setDeliveryTimeout(config.timeout);
         job.setDestinationBlocklist(config.blocklist);
         job.setBlockLocalAddresses(config.blockLocalAddresses);
         job.setAllowRedirects(config.allowRedirects);
         job.setAllowHttp(config.allowHttp);
-        return job;
     }
 
     protected String getEventValueByFilterType(Event event, WebhookFilter.Type filterType) {
@@ -214,17 +228,7 @@ protected List<Runnable> getDeliveryJobs(Event event) throws EventBusException {
                 logger.debug("Skipping delivering {} to {} as it doesn't match filters", event, webhook);
                 continue;
             }
-            if (!domainConfigs.containsKey(webhook.getDomainId())) {
-                domainConfigs.put(webhook.getDomainId(),
-                        new DeliveryConfig(
-                                WebhookDeliveryTries.valueIn(webhook.getDomainId()),
-                                WebhookDeliveryTimeout.valueIn(webhook.getDomainId()),
-                                WebhookDeliveryBlocklist.valueIn(webhook.getDomainId()),
-                                WebhookDeliveryBlockLocalAddresses.value(),
-                                WebhookDeliveryAllowRedirects.valueIn(webhook.getDomainId()),
-                                WebhookDeliveryAllowHttp.valueIn(webhook.getDomainId())));
-            }
-            DeliveryConfig config = domainConfigs.get(webhook.getDomainId());
+            DeliveryConfig config = domainConfigs.computeIfAbsent(webhook.getDomainId(), this::getDeliveryConfig);
             WebhookDeliveryThread job = getDeliveryJob(event, webhook, config);
             jobs.add(job);
         }
@@ -274,6 +278,54 @@ protected Runnable getManualDeliveryJob(WebhookDelivery existingDelivery, Webhoo
         return job;
     }
 
+    protected List<Runnable> getDirectDeliveryJobs(List<Long> webhookIds, long accountId, String eventType,
+            String payload) {
+        List<Runnable> jobs = new ArrayList<>();
+        if (CollectionUtils.isEmpty(webhookIds)) {
+            return jobs;
+        }
+        Account account = accountManager.getAccount(accountId);
+        Event event = new Event(ManagementService.Name, EventCategory.ALERT_EVENT.getName(), eventType, null, null);
+        event.setEventUuid(UUID.randomUUID().toString());
+        event.setDescription(payload);
+        event.setResourceAccountUuid(account != null ? account.getUuid() : null);
+        for (Long webhookId : webhookIds) {
+            WebhookVO webhook = webhookDao.findById(webhookId);
+            if (webhook == null || !Webhook.State.Enabled.equals(webhook.getState())) {
+                logger.debug("Skipping delivering {} to webhook ID: {} as it is missing or disabled", event, webhookId);
+                continue;
+            }
+            if (!isEventMatchingFilters(event, webhookFiltersCache.get(webhook.getId()))) {
+                logger.debug("Skipping delivering {} to {} as it doesn't match filters", event, webhook);
+                continue;
+            }
+            WebhookDeliveryThread.WebhookDeliveryContext<WebhookDeliveryThread.WebhookDeliveryResult> context =
+                    new WebhookDeliveryThread.WebhookDeliveryContext<>(null, null, webhook.getId());
+            AsyncCallbackDispatcher<WebhookServiceImpl, WebhookDeliveryThread.WebhookDeliveryResult> caller =
+                    AsyncCallbackDispatcher.create(this);
+            caller.setCallback(caller.getTarget().directDeliveryCompleteCallback(null, null))
+                    .setContext(context);
+            WebhookDeliveryThread job = new WebhookDeliveryThread(webhook, event, caller);
+            job = ComponentContext.inject(job);
+            applyDeliveryConfig(job, getDeliveryConfig(webhook.getDomainId()));
+            jobs.add(job);
+        }
+        return jobs;
+    }
+
+    // Not persisted: webhook_delivery.event_id must reference a row in the event table.
+    protected Void directDeliveryCompleteCallback(
+            AsyncCallbackDispatcher<WebhookServiceImpl, WebhookDeliveryThread.WebhookDeliveryResult> callback,
+            WebhookDeliveryThread.WebhookDeliveryContext<Webhook> context) {
+        WebhookDeliveryThread.WebhookDeliveryResult result = callback.getResult();
+        if (result.isSuccess()) {
+            logger.debug("Delivered alert to webhook ID: {}", context.getRuleId());
+        } else {
+            logger.warn("Failed to deliver alert to webhook ID: {} due to: {}", context.getRuleId(), result.getResult());
+        }
+        return null;
+    }
+
     protected Void deliveryCompleteCallback(
             AsyncCallbackDispatcher<WebhookServiceImpl, WebhookDeliveryThread.WebhookDeliveryResult> callback,
             WebhookDeliveryThread.WebhookDeliveryContext<Webhook> context) {
@@ -386,6 +438,35 @@ public List<? extends ControlledEntity> listWebhooksByAccount(long accountId) {
         return webhookDao.listByAccount(accountId);
     }
 
+    @Override
+    public ControlledEntity findWebhookByUuid(String uuid) {
+        return webhookDao.findByUuid(uuid);
+    }
+
+    @Override
+    public String getWebhookUuid(long webhookId) {
+        WebhookVO webhook = webhookDao.findByIdIncludingRemoved(webhookId);
+        return webhook != null ? webhook.getUuid() : null;
+    }
+
+    @Override
+    public void deliverToWebhooks(List<Long> webhookIds, long accountId, String eventType, String payload) {
+        for (Runnable job : getDirectDeliveryJobs(webhookIds, accountId, eventType, payload)) {
+            webhookJobExecutor.submit(loggingFailures(job, eventType));
+        }
+    }
+
+    // The executor swallows exceptions, e.g. a payload URL rejected by the blocklist at delivery time.
+    protected Runnable loggingFailures(Runnable job, String eventType) {
+        return () -> {
+            try {
+                job.run();
+            } catch (Exception e) {
+                logger.warn("Failed to deliver {} to webhook: {}", eventType, e.getMessage());
+            }
+        };
+    }
+
     @Override
     public void handleEvent(Event event) throws EventBusException {
         List<Runnable> jobs = getDeliveryJobs(event);