Add optional per-rule Alarm Manager forwarding through native hooks

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 --- develop ---
 
+* feature: Add optional Alarm Manager forwarding per alert rule, retain Syslog Alarming as the default and fallback, and preserve threshold episode recovery behavior.
+
 * issue#317: Expire partitions by their actual UTC upper boundaries instead of partition count, preserving cutoff-spanning partitions and supporting legacy TO_DAYS layouts.
 
 * issue#317: Defer automatic partition reorganization when overflow estimates exceed 100,000 rows or 128 MiB, or cannot be read; preserve the safety partition and report the reason through maintenance health.

README.md

@@ -551,3 +551,5 @@ nesting levels. Invalid searches return no results and CSV export rejects them.
 
 -----------------------------------------------
 Copyright (c) 2004-2026 - The Cacti Group, Inc.
+
+Optional [Alarm Manager forwarding](docs/ALARM_MANAGER.md) uses Cacti native in-process hooks with existing Syslog email as fallback.

docs/ALARM_MANAGER.md

@@ -0,0 +1,66 @@
+# Alarm Manager forwarding
+
+In **Console → Configuration → Settings → Syslog**, enable **Forward alarms to
+Alarm Manager**. It is off by default. Install and enable `plugin_alarmmanager`
+on each processing collector, with matching enabled routing rules and delivery
+destinations. Cacti's normal settings replication carries this setting; Syslog's
+existing rule replication continues unchanged. No HTTP endpoint or credentials
+are needed by Syslog.
+
+In each Alert Rule, select **Alarm Destination → Syslog Alarming** (the default)
+or **Alarm Manager**. The selector appears only when Alarm Manager is installed.
+The global forwarding setting must also be enabled for Alarm Manager rules.
+Their Email Options and Actions sections are hidden, and ticket, alert, and recovery commands never
+run, including when forwarding is unavailable. Configured email remains the
+fallback. Removing or disabling Alarm Manager retains that email fallback.
+
+Syslog checks Cacti's `api_plugin_installed()` and `api_plugin_is_enabled()` and
+the enabled `alarmmanager_submit` hook registration, then calls
+`api_plugin_hook_function()` with contract version 1. Only an unchanged event
+returned with `accepted` or `already_accepted` and a positive integer
+`event_row_id` suppresses the corresponding email/SMS email. `unhandled`,
+`failed` (including invalid-event rejection), exceptions, and malformed results
+retain normal Syslog email. Acceptance means committed ingestion and processing
+job storage, not successful routing or external delivery. Syslog Alarming rules retain their existing commands and email. Alarm history,
+maintenance and suppression checks remain active for both destinations.
+
+Events include the rule, source message IDs, message text (bounded to 16 KiB),
+hosts, severity, timestamp, count, and processing collector. Syslog host IDs are
+not Cacti device IDs; they are sent as external identities with a null device
+ID. Event identity uses the rule, collector and sorted source sequence/time
+pairs. Retrying the same batch sends the same event; separate occurrences have
+separate stream keys. Keep the Cacti base URL, Syslog database name, collector ID
+and collector timezone stable across retries. Failures are logged by rule and
+safe reason code without message or provider exception contents.
+
+Windowed threshold events use their existing breach key to correlate raises
+and recovery. Only accepted raises enable forwarding of their quiet-period
+clear. The existing collector-owned recovery queue retains clear timestamps and
+retries failed submissions. Alarm Manager recovery does not run commands.
+Occurrence alarms and thresholds without breach state send occurrences
+and never invent clears. Replicated history and foreign-owned breaches cannot
+submit events. Alarm Manager owns its own native queue/history replication.
+
+Raise forwarding is synchronous at the existing notification boundary; this
+change does not add a crash-durable raise outbox. The local recovery-ownership
+write follows Alarm Manager's ingestion commit; a crash or failed local write
+between them can leave an accepted raise without a forwarded clear (logged when
+the write fails). Existing Syslog notification retry semantics otherwise apply.
+Disabling forwarding, deleting/disabling/editing a rule, or changing its recovery
+command can cancel queued recovery under the existing lifecycle rules.
+
+Local checks:
+
+```sh
+node tests/regression/alarm_destination_ui_test.js
+php tests/regression/alarmmanager_forward_test.php
+SYSLOG_TEST_MYSQL_SOCKET=/path/to/disposable/mysql.sock php tests/regression/threshold_recovery_database_test.php
+php tests/regression/threshold_alert_host_test.php
+```
+
+The forwarding tests use Alarm Manager's real event validator from the sibling
+checkout (an alternate validator path can be passed as the first argument to
+`alarmmanager_forward_test.php`). Ingestion and mail delivery are stubbed. The
+MariaDB test creates and drops randomly named test databases and exercises real
+Syslog recovery SQL and local fixture commands. Neither test verifies live
+remote collectors, Alarm Manager ingestion commits, routing, or provider delivery.

includes/alarmmanager.php

@@ -0,0 +1,86 @@
+<?php
+/* Optional in-process Alarm Manager submission. No dependency on its PHP files. */
+
+function syslog_alarmmanager_selected(array $alert): bool {
+	return ($alert['alarm_destination'] ?? 'syslog') === 'alarmmanager';
+}
+
+/** True only when ingestion has committed this exact event (including retries). */
+function syslog_alarmmanager_forward(array $alert, array $matches, int $count, bool $recovered = false, ?int $cleared_at = null): bool {
+	global $config, $syslogdb_default;
+
+	if (!syslog_alarmmanager_selected($alert) || read_config_option('syslog_alarmmanager_enabled') !== 'on') {
+		return false;
+	}
+
+	try {
+		if (!api_plugin_installed('alarmmanager') || !api_plugin_is_enabled('alarmmanager')) {
+			throw new RuntimeException('plugin_unavailable');
+		}
+		// Cacti exposes no hook-presence helper; inspect its enabled registry.
+		if (!db_fetch_cell_prepared('SELECT COUNT(*) FROM plugin_hooks WHERE name = ? AND hook = ? AND status = 1', ['alarmmanager', 'alarmmanager_submit'])) {
+			throw new RuntimeException('hook_unavailable');
+		}
+		$collector = (int) ($config['poller_id'] ?? 1);
+		$state = $alert['_threshold_state'] ?? [];
+		if ($collector < 1 || (isset($state['source_poller_id']) && (int) $state['source_poller_id'] !== $collector) || !$matches || ($recovered && (!$state || $cleared_at === null))) {
+			throw new RuntimeException('invalid_owner_or_event');
+		}
+		foreach ($matches as $match) {
+			if (isset($match['replication_source_poller_id']) && (int) $match['replication_source_poller_id'] > 0 && (int) $match['replication_source_poller_id'] !== $collector) {
+				throw new RuntimeException('foreign_collector');
+			}
+			if (empty($match['seq']) || empty($match['logtime'])) {
+				throw new RuntimeException('source_identity_missing');
+			}
+		}
+		usort($matches, static fn($a, $b) => (int) $a['seq'] <=> (int) $b['seq']);
+		$first = $matches[0];
+		$identity = hash('sha256', json_encode([$alert['id'], $collector, array_map(static fn($m) => [$m['seq'], $m['logtime']], $matches)], JSON_THROW_ON_ERROR));
+		$episode = $state['breach_key'] ?? $identity;
+		$transition = $recovered ? 'recovered' : ($state ? 'raised' : 'occurrence');
+		$time = $recovered ? (new DateTimeImmutable('@' . $cleared_at)) : new DateTimeImmutable($first['logtime']);
+		$event = [
+			'event_id' => $recovered ? hash('sha256', $episode . ':clear') : $identity,
+			'source' => 'syslog',
+			'source_instance' => hash('sha256', read_config_option('base_url') . ':' . $syslogdb_default . ':' . $collector),
+			'resource_id' => 'rule:' . $alert['id'],
+			// Independent episodes must not share a fixed-sequence stream.
+			'alarm_key' => 'rule:' . $alert['id'] . ':' . $episode,
+			'episode_id' => $episode,
+			'sequence' => $recovered ? PHP_INT_MAX : (int) $first['seq'],
+			'transition' => $transition,
+			'severity' => $recovered ? 'clear' : ([0 => 'info', 1 => 'warning', 2 => 'critical'][(int) $alert['severity']] ?? 'warning'),
+			'occurred_at' => $time->format('Y-m-d\TH:i:sP'),
+			'collector_id' => $collector,
+			// Syslog host IDs are not Cacti device IDs.
+			'device' => ['id' => null, 'hostname' => substr((string) $first['host'], 0, 255), 'external_id' => 'syslog:' . ($first['host_id'] ?? $first['host'])],
+			'summary' => substr((string) $alert['name'], 0, 512),
+			'details' => substr(implode("\n", array_map(static fn($m) => $m['host'] . ' ' . $m['logtime'] . ' #' . $m['seq'] . ' [priority ' . ($m['priority_id'] ?? '') . ', facility ' . ($m['facility_id'] ?? '') . ']: ' . $m['message'], $matches)), 0, 16384),
+			'context' => ['rule_name' => substr((string) $alert['name'], 0, 512), 'rule_type' => (string) $alert['method'], 'metric_value' => (string) $count, 'threshold' => (string) ($alert['num'] ?? 1), 'source_notification_channels' => ['email']]
+		];
+		$envelope = ['contract_version' => 1, 'event' => $event];
+		$result = api_plugin_hook_function('alarmmanager_submit', $envelope);
+		if (is_array($result) && in_array($result['status'] ?? null, ['accepted', 'already_accepted'], true)
+			&& is_int($result['event_row_id'] ?? null) && $result['event_row_id'] > 0
+			&& ($result['event'] ?? null) === $event) {
+			if (!$recovered && $state) {
+				try {
+					if (!syslog_db_execute_prepared("UPDATE syslog_threshold_state SET recovery_context = JSON_SET(recovery_context, '$.alarmmanager_accepted', true) WHERE source_poller_id = ? AND breach_key = ?", [$collector, $state['breach_key']])) {
+						throw new RuntimeException('recovery_context_write_failed');
+					}
+				} catch (Throwable $error) {
+					// Acceptance is already committed; a local failure must not resend email.
+					cacti_log('SYSLOG: Unable to retain Alarm Manager recovery ownership for rule ' . (int) $alert['id'], false, 'SYSLOG');
+				}
+			}
+			return true;
+		}
+		throw new RuntimeException(is_array($result) && in_array($result['status'] ?? null, ['failed', 'rejected', 'unhandled'], true) ? 'submission_' . $result['status'] : 'submission_not_accepted');
+	} catch (Throwable $error) {
+		// Never log provider exceptions or message contents.
+		$reason = in_array($error->getMessage(), ['plugin_unavailable', 'hook_unavailable', 'invalid_owner_or_event', 'foreign_collector', 'source_identity_missing', 'submission_not_accepted', 'submission_failed', 'submission_rejected', 'submission_unhandled'], true) ? $error->getMessage() : 'submission_exception';
+		cacti_log('SYSLOG: Alarm Manager ' . $reason . ' for rule ' . (int) $alert['id'] . '; retaining Syslog notification', false, 'SYSLOG');
+		return false;
+	}
+}

includes/functions.php

@@ -22,6 +22,8 @@
  +-------------------------------------------------------------------------+
 */
 
+require_once(__DIR__ . '/alarmmanager.php');
+
 $syslog_query_builder = dirname(__DIR__) . '/lib/QueryBuilder.php';
 
 if (file_exists($syslog_query_builder)) {
@@ -4686,6 +4688,9 @@ function syslog_array2xml(array $array, string $tag = 'template'): string {
  * @return void
  */
 function syslog_execute_ticket_command($alert, $hostlist, $error_message) {
+	if (syslog_alarmmanager_selected($alert)) {
+		return;
+	}
 	$command = read_config_option('syslog_ticket_command');
 
 	if ($command != '') {
@@ -4731,6 +4736,9 @@ function syslog_execute_ticket_command($alert, $hostlist, $error_message) {
  * @return bool True when the command exits successfully.
  */
 function syslog_execute_alert_command($alert, $results, $hostname, string $event = 'breach'): bool {
+	if (syslog_alarmmanager_selected($alert)) {
+		return true;
+	}
 	/* alert_replace_variables() escapes each substituted token (<ALERTID>,
 	 * <HOSTNAME>, <PRIORITY>, <FACILITY>, <MESSAGE>, <SEVERITY>) with
 	 * cacti_escapeshellarg(). The command template itself comes from admin
@@ -4998,7 +5006,9 @@ function syslog_alert_window_seconds(array $alert): int {
 
 /** Hash the rule settings that define a threshold episode. */
 function syslog_threshold_rule_hash(array $alert): string {
-	return sha1(json_encode(array_intersect_key($alert, array_flip(['type', 'message', 'level', 'num', 'threshold_window', 'threshold_unit', 'threshold_recovery_minutes', 'threshold_recovery_unit', 'recovery_command', 'date']))));
+	$settings = array_intersect_key($alert, array_flip(['type', 'message', 'level', 'num', 'threshold_window', 'threshold_unit', 'threshold_recovery_minutes', 'threshold_recovery_unit', 'recovery_command', 'alarm_destination', 'date']));
+
+	return substr(hash('sha256', json_encode($settings)), 0, 40);
 }
 
 /** Recover quiet breaches without scanning or rewriting message history. */
@@ -5013,10 +5023,10 @@ function syslog_threshold_recover(?int $now = null): void {
 	}
 	$queries = [
 		["INSERT INTO syslog_threshold_recovery_queue (source_poller_id, breach_key, alert_id, rule_date, context, available_at)
-			SELECT s.source_poller_id, s.breach_key, s.alert_id, s.rule_date, s.recovery_context, ?
+			SELECT s.source_poller_id, s.breach_key, s.alert_id, s.rule_date, JSON_SET(s.recovery_context, '$.cleared_at', s.expires_at), ?
 			FROM syslog_threshold_state AS s JOIN syslog_alert AS a ON a.id = s.alert_id
 			WHERE s.source_poller_id = ? AND s.expires_at <= ? AND a.enabled = 'on' AND a.method = 1
-			AND s.rule_date = a.date AND s.recovery_context IS NOT NULL AND COALESCE(a.recovery_command, '') <> ''
+			AND s.rule_date = a.date AND s.recovery_context IS NOT NULL AND ((a.alarm_destination = 'syslog' AND COALESCE(a.recovery_command, '') <> '') OR (a.alarm_destination = 'alarmmanager' AND '" . (read_config_option('syslog_alarmmanager_enabled') === 'on' ? 'on' : '') . "' = 'on'))
 			ON DUPLICATE KEY UPDATE breach_key = VALUES(breach_key)", [$now, $source, $now]],
 		["DELETE c FROM syslog_alert_counts AS c JOIN syslog_threshold_state AS s ON s.alert_id = c.alert_id AND s.scope_key = c.scope_key LEFT JOIN syslog_alert AS a ON a.id = s.alert_id WHERE s.source_poller_id = ? AND $ended", [$source, $now]],
 		["DELETE m FROM syslog_threshold_state AS s JOIN syslog_threshold_messages AS m ON m.source_poller_id = s.source_poller_id AND m.breach_key = s.breach_key LEFT JOIN syslog_alert AS a ON a.id = s.alert_id WHERE (s.source_poller_id = ? AND $ended) OR s.expires_at <= ?", [$source, $now, $now]],
@@ -5049,7 +5059,7 @@ function syslog_threshold_deliver_recoveries(?int $now = null): void {
 
 	try {
 		if (!syslog_db_execute_prepared("DELETE q FROM syslog_threshold_recovery_queue AS q LEFT JOIN syslog_alert AS a ON a.id = q.alert_id
-			WHERE q.source_poller_id = ? AND (a.id IS NULL OR a.enabled <> 'on' OR a.method <> 1 OR a.date <> q.rule_date OR COALESCE(a.recovery_command, '') = '')", [$source])) {
+			WHERE q.source_poller_id = ? AND (a.id IS NULL OR a.enabled <> 'on' OR a.method <> 1 OR a.date <> q.rule_date OR ((a.alarm_destination = 'alarmmanager' OR COALESCE(a.recovery_command, '') = '') AND (a.alarm_destination <> 'alarmmanager' OR '" . (read_config_option('syslog_alarmmanager_enabled') === 'on' ? 'on' : '') . "' <> 'on')))", [$source])) {
 			return;
 		}
 		$pending = syslog_db_fetch_assoc_prepared('SELECT breach_key, context, attempts FROM syslog_threshold_recovery_queue WHERE source_poller_id = ? AND available_at <= ? ORDER BY available_at, breach_key LIMIT 25', [$source, $now]);
@@ -5062,9 +5072,13 @@ function syslog_threshold_deliver_recoveries(?int $now = null): void {
 				&& isset($context['alert']['id'], $context['alert']['name'], $context['alert']['message'], $context['alert']['severity'], $context['alert']['command'],
 					$context['alert']['_threshold_state']['breach_key'], $context['alert']['_threshold_state']['source_poller_id'],
 					$context['results']['message'], $context['results']['priority_id'], $context['results']['facility_id'])) {
+				$forwarded = true;
+				if (syslog_alarmmanager_selected($context['alert']) && !empty($context['alarmmanager_accepted']) && isset($context['alarmmanager_matches'], $context['cleared_at']) && read_config_option('syslog_alarmmanager_enabled') === 'on') {
+					$forwarded = syslog_alarmmanager_forward($context['alert'], $context['alarmmanager_matches'], 0, true, (int) $context['cleared_at']);
+				}
 				alert_setup_environment($context['alert'], $context['results'], $context['hostlist'], $context['hostname']);
-
-				if (syslog_execute_alert_command($context['alert'], $context['results'], $context['hostname'], 'clear')) {
+				$command_done = trim($context['alert']['command']) === '' || syslog_execute_alert_command($context['alert'], $context['results'], $context['hostname'], 'clear');
+				if ($command_done && $forwarded) {
 					syslog_db_execute_prepared('DELETE FROM syslog_threshold_recovery_queue WHERE source_poller_id = ? AND breach_key = ?', [$source, $row['breach_key']]);
 
 					continue;
@@ -5416,11 +5430,12 @@ function syslog_mark_threshold_messages(array $alert, int $count, array $matches
 		return true;
 	}
 
-	if ($matches && !empty($alert['_threshold_state']) && trim($alert['recovery_command'] ?? '') !== '') {
-		$state   = $alert['_threshold_state'];
+	if ($matches && !empty($alert['_threshold_state']) && (syslog_alarmmanager_selected($alert) ? read_config_option('syslog_alarmmanager_enabled') === 'on' : trim($alert['recovery_command'] ?? '') !== '')) {
+		$state = $alert['_threshold_state'];
 		$context = json_encode([
-			'alert'    => ['id' => $alert['id'], 'name' => $alert['name'], 'severity' => $alert['severity'], 'message' => $alert['message'], 'command' => $alert['recovery_command'], '_threshold_state' => $state],
-			'results'  => ['priority_id' => $matches[0]['priority_id'], 'facility_id' => $matches[0]['facility_id'], 'message' => $matches[0]['message']],
+			'alert' => ['id' => $alert['id'], 'name' => $alert['name'], 'severity' => $alert['severity'], 'message' => $alert['message'], 'alarm_destination' => $alert['alarm_destination'] ?? 'syslog', 'method' => $alert['method'], 'num' => $alert['num'], 'command' => $alert['recovery_command'] ?? '', '_threshold_state' => $state],
+			'results' => ['priority_id' => $matches[0]['priority_id'], 'facility_id' => $matches[0]['facility_id'], 'message' => $matches[0]['message']],
+			'alarmmanager_matches' => [$matches[0]],
 			'hostname' => (int) $alert['level'] === 1 ? $matches[0]['host'] : '',
 			'hostlist' => array_values(array_unique(array_column($matches, 'host')))
 		], JSON_INVALID_UTF8_SUBSTITUTE | JSON_THROW_ON_ERROR);
@@ -5783,7 +5798,7 @@ function syslog_process_alert($alert, $sql, $params, $count, $hostname = '', $at
 					/**
 					 * Send the Email notification
 					 */
-					if ($alert['email'] != '' || $smsalert != '') {
+					if (!syslog_alarmmanager_forward($alert, $at, (int) $count) && ($alert['email'] != '' || $smsalert != '')) {
 						syslog_sendemail(trim($alert['email']), $from, __esc('Event Alert - %s', $alert['name'], 'syslog'), $message, $smsalert);
 					}
 
@@ -5808,7 +5823,7 @@ function syslog_process_alert($alert, $sql, $params, $count, $hostname = '', $at
 					 * so a threshold alert emails out only when a recipient
 					 * is configured.
 					 */
-					if ($alert['email'] != '') {
+					if (!syslog_alarmmanager_forward($alert, $at, (int) $count) && $alert['email'] != '') {
 						syslog_sendemail(trim($alert['email']), $from, __esc('Event Alert - %s', $alert['name'], 'syslog'), $message, $smsalert);
 					}