Feat/pipeline health alerts

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 --- develop ---
 
+
+* feature: Add poller-driven pipeline health alerts for stalled processing, incoming backlogs, remote synchronization failures or delays, and blocked partition maintenance; expand the existing Status health card with check results and email delivery status, with optional problem, reminder, and recovery notifications
+* bug: Treat an empty incoming queue as drained rather than missing data, and evaluate stale queued messages even when new messages continue arriving
+
 * feature: Add an auto-refresh interval dropdown and a manual refresh glyph to the Syslog Status toolbar (mirroring the intropage refresh concept); the interval persists through the page's auto-reload and the refresh glyphs spin while their cards reload
 * bug: Show the Syslog Status card refresh glyph again by using Font Awesome's fa-sync-alt (the bundled Font Awesome 6 no longer renders the Font Awesome 4 fa-refresh alias)
 * change: Repaint the Syslog Status card headers with the active theme's native html_start_box title bar (sampled from the live cactiTableTitle at runtime) instead of a flat surface tint so they match Cacti on every theme, widen the card grid from 6 to 8 columns, drop the card corner radius, and give each header one of four translucent severity overlays - a neutral grey "no severity" default plus intropage's good / warning / error palette (#94b36b / #ff8f1e / #cc0000)

README.md

@@ -53,6 +53,38 @@ be quieted using syslog's 'Re-Alert' setting.
 
 * Ability to run alert specific commands at Alert and Re-alert times
 
+## Pipeline Health Alerts
+
+The Status tab includes a **Pipeline health** card in the existing Collector
+health card position. It reports stalled processing, excessive or old incoming
+backlogs, failed or delayed remote synchronization, and blocked partition
+maintenance. An empty incoming queue is healthy when processing is completing;
+quiet devices do not trigger source-silence alerts. Unavailable queue metrics
+are reported as problems rather than healthy values.
+
+Checks run from Cacti's poller before launching the Syslog processor, so a
+stalled processor can still be detected. The delay is the greater of the
+Collector Staleness Threshold, three poller intervals, and 300 seconds. New or
+re-enabled monitoring gets a startup grace period. Incoming age uses the
+configured incoming timestamp field; delayed historical events or incorrect
+source clocks can therefore trigger an old-backlog warning.
+
+In **Settings > Syslog**, select a **Pipeline Health Notification List** from
+Thold and/or set **Pipeline Health Email Recipients** to enable problem,
+changed-problem, reminder, and recovery emails using Cacti's mail settings.
+Enabled lists add their current recipients on each check, preserving BCC
+recipients; disabled or missing lists contribute no addresses. Direct email
+recipients work without Thold. Email is disabled by default. Persistent problems repeat at the
+configured reminder interval, and failed delivery attempts retry no more than
+once every five minutes. The card shows the last health check, email attempt
+status, and last successful send. Health state changes also appear in Cacti's
+log. Intentionally disabled processing does not generate health notifications.
+
+Monitoring requires a running Cacti poller and an accessible Syslog database.
+Use external monitoring for a complete Cacti or database outage. Remote health
+checks run on each enabled remote collector; this does not infer a collector
+outage merely from a quiet stream of device messages.
+
 ## Table Architecture and Storage Engine Support
 
 New installs always create partitioned `syslog` tables.  The traditional

includes/functions.php

@@ -1577,6 +1577,7 @@ function syslog_status_get(): array {
 	syslog_status_ensure_table();
 
 	$status = [
+		'pipeline_health_monitor' => '',
 		'last_polling_time' => '',
 		'last_start_time'   => '',
 		'last_end_time'     => '',
@@ -1596,6 +1597,8 @@ function syslog_status_get(): array {
 		'partition_maintenance_last_attempt' => '',
 		'partition_maintenance_last_success' => '',
 		'partition_maintenance_outcome'      => '',
+		'partition_maintenance_blocked'      => '',
+		'partition_maintenance_reason'       => '',
 		'partition_recovery_progress'        => '',
 		'partition_maintenance_history'      => '',
 		'replication_state' => '',
@@ -1607,6 +1610,7 @@ function syslog_status_get(): array {
 	$rows = syslog_db_fetch_assoc("SELECT `name`, `value`, `updated`
 		FROM `$syslogdb_default`.`syslog_status`
 		WHERE `name` IN (
+			'pipeline_health_monitor',
 			'last_polling_time',
 			'last_start_time',
 			'last_end_time',
@@ -1626,6 +1630,8 @@ function syslog_status_get(): array {
 			'partition_maintenance_last_attempt',
 			'partition_maintenance_last_success',
 			'partition_maintenance_outcome',
+			'partition_maintenance_blocked',
+			'partition_maintenance_reason',
 			'partition_recovery_progress',
 			'partition_maintenance_history',
 			'replication_state',
@@ -1641,6 +1647,163 @@ function syslog_status_get(): array {
 	return $status;
 }
 
+/** Read pipeline health without changing state or sending notifications. */
+function syslog_pipeline_health(): array {
+	global $config, $syslogdb_default, $syslog_incoming_config;
+
+	$status = syslog_status_get();
+	$monitor = json_decode($status['pipeline_health_monitor'], true);
+	$monitor = is_array($monitor) ? $monitor : [];
+	$health = ['state' => 'up', 'issues' => [], 'checks' => [], 'monitor' => $monitor];
+
+	if (read_config_option('syslog_enabled') !== 'on'
+		|| ((int) ($config['poller_id'] ?? 1) > 1 && read_config_option('syslog_remote_enabled') !== 'on')) {
+		$health['state'] = 'disabled';
+		return $health;
+	}
+
+	$now = time();
+	$interval = max(60, (int) read_config_option('poller_interval'));
+	$delay = max($interval * 3, (int) read_config_option('syslog_stale_threshold'), 300);
+	$limit = (int) read_config_option('syslog_backlog_threshold');
+	$limit = $limit > 0 ? $limit : 10000;
+	$completed = (int) $status['last_end_time'];
+	$baseline = max($completed, (int) ($monitor['first_checked'] ?? 0));
+	if (($monitor['signature'] ?? '') === 'disabled:') {
+		$baseline = 0;
+	}
+	$health['checks']['processing'] = $completed === 0 || $baseline === 0
+		? __('Waiting for a completed processing run.', 'syslog')
+		: __('Processing is completing within the allowed delay.', 'syslog');
+	if ($baseline > 0 && $now - $baseline > $delay) {
+		$health['issues']['processing'] = __('No processing run has completed for more than %d seconds.', $delay, 'syslog');
+	}
+
+	$column = $syslog_incoming_config['timeField'] ?? 'logtime';
+	$incoming = [];
+	if (is_string($column) && preg_match('/^[a-zA-Z_][a-zA-Z0-9_]*$/', $column)) {
+		// Above the threshold, the exact count and oldest timestamp cannot change the verdict.
+		$sample_limit = $limit + 1;
+		$incoming = syslog_db_fetch_row("SELECT COUNT(*) AS pending, UNIX_TIMESTAMP(MIN(`$column`)) AS oldest
+			FROM (SELECT `$column` FROM `$syslogdb_default`.`syslog_incoming` LIMIT $sample_limit) AS incoming_sample", false);
+	}
+	$health['checks']['backlog'] = __('Incoming messages are within the configured limits.', 'syslog');
+	if (!isset($incoming['pending']) || !is_numeric($incoming['pending'])) {
+		$health['issues']['backlog'] = __('Incoming queue metrics are unavailable.', 'syslog');
+	} elseif ((int) $incoming['pending'] > $limit) {
+		$health['issues']['backlog'] = __('Incoming backlog of at least %s messages exceeds the limit of %s.', number_format((int) $incoming['pending']), number_format($limit), 'syslog');
+	} elseif ((int) $incoming['pending'] > 0) {
+		if (!isset($incoming['oldest']) || !is_numeric($incoming['oldest'])) {
+			$health['issues']['backlog'] = __('The oldest incoming message timestamp is unavailable.', 'syslog');
+		} elseif ($now - (int) $incoming['oldest'] > $delay) {
+			$health['issues']['backlog'] = __('Unprocessed messages are older than %d seconds.', $delay, 'syslog');
+		}
+	}
+
+	$replication = syslog_replication_operational_status(true);
+	$health['checks']['replication'] = __('Synchronization is not enabled on this collector.', 'syslog');
+	if (!empty($replication['enabled'])) {
+		$health['checks']['replication'] = __('Synchronization is within the allowed delay.', 'syslog');
+		$oldest = $replication['oldest_pending_epoch'];
+		if ($replication['pending'] === null) {
+			$health['issues']['replication'] = __('Synchronization queue metrics are unavailable.', 'syslog');
+		} elseif ($replication['pending'] > 0 && $oldest === null) {
+			$health['issues']['replication'] = __('The oldest synchronization record timestamp is unavailable.', 'syslog');
+		} elseif ($replication['pending'] > 0 && ($replication['last_error'] !== ''
+			|| ($oldest !== null && $now - $oldest > $delay))) {
+			$health['issues']['replication'] = __('Synchronization has failed or queued records exceed the allowed delay.', 'syslog');
+		}
+	}
+
+	$health['checks']['partition'] = __('Partition maintenance is not blocked.', 'syslog');
+	if ((int) ($status['partition_maintenance_blocked'] ?? 0) === 1) {
+		$health['issues']['partition'] = __('Partition maintenance is blocked: %s', $status['partition_maintenance_reason'] ?? '', 'syslog');
+	}
+
+	if ($health['issues']) {
+		$health['state'] = 'down';
+	} elseif ($completed === 0 || $baseline === 0) {
+		$health['state'] = 'recovering';
+	}
+	return $health;
+}
+
+/** Check independently of the Syslog processor; serialize notifications per collector. */
+function syslog_pipeline_health_check(): void {
+	global $config, $syslogdb_default;
+
+	$lock = 'syslog_health_' . sha1($syslogdb_default . ':' . (int) ($config['poller_id'] ?? 1));
+	if ((int) syslog_db_fetch_cell_prepared('SELECT GET_LOCK(?, 0)', [$lock], false) !== 1) {
+		return;
+	}
+
+	try {
+		$health = syslog_pipeline_health();
+		$monitor = $health['monitor'];
+		$now = time();
+		$keys = array_keys($health['issues']);
+		sort($keys);
+		$signature = $health['state'] . ':' . implode(',', $keys);
+		$prior = $monitor['signature'] ?? '';
+		$monitor['first_checked'] = $prior === 'disabled:' ? $now : ($monitor['first_checked'] ?? $now);
+		$monitor['checked'] = $now;
+		$monitor['signature'] = $signature;
+
+		if ($signature !== $prior && $health['state'] !== 'disabled' && ($health['issues'] || str_starts_with($prior, 'down:'))) {
+			cacti_log('SYSLOG PIPELINE: ' . ($health['issues'] ? implode(' ', $health['issues']) : 'Pipeline recovered'), false, 'SYSLOG');
+		}
+
+		$recipients = trim((string) read_config_option('syslog_pipeline_health_email'), ' ,');
+		$bcc = '';
+		$list_id = (int) read_config_option('syslog_pipeline_health_notify');
+		if ($list_id > 0 && db_table_exists('plugin_notification_lists')) {
+			$bcc_column = db_column_exists('plugin_notification_lists', 'bcc_emails') ? 'bcc_emails' : "'' AS bcc_emails";
+			$list = db_fetch_row_prepared("SELECT emails, $bcc_column FROM plugin_notification_lists WHERE id = ? AND enabled = ?", [$list_id, 'on']);
+			$recipients = implode(', ', array_filter([$recipients, trim((string) ($list['emails'] ?? ''), ' ,')]));
+			$bcc = trim((string) ($list['bcc_emails'] ?? ''), ' ,');
+		}
+		if (($recipients === '' && $bcc === '') || $health['state'] === 'disabled') {
+			$monitor['notification'] = 'disabled';
+			unset($monitor['notified_signature']);
+		} else {
+			$notified = $monitor['notified_signature'] ?? '';
+			$repeat = max(300, (int) (read_config_option('syslog_pipeline_health_repeat') ?: 3600));
+			$due = ($health['issues'] && ($signature !== $notified || $now - (int) ($monitor['last_notified'] ?? 0) >= $repeat))
+				|| (!$health['issues'] && str_starts_with($notified, 'down:'));
+			// Bound failed mail retries even if poller hooks overlap or run more often.
+			if ($due && $now - (int) ($monitor['last_attempt'] ?? 0) >= 300) {
+				$monitor['last_attempt'] = $now;
+				$monitor['notification'] = 'pending';
+				if (!syslog_status_set('pipeline_health_monitor', (string) json_encode($monitor))) {
+					return;
+				}
+				$subject = __('Syslog pipeline health - collector %d', (int) ($config['poller_id'] ?? 1), 'syslog');
+				$message = $health['issues'] ? implode("\n", $health['issues']) : __('The Syslog pipeline has recovered.', 'syslog');
+				$from = [(string) (read_config_option('settings_from_email') ?: 'Cacti@cacti.net'), (string) (read_config_option('settings_from_name') ?: 'Cacti')];
+				try {
+					$error = mailer($from, $recipients, '', $bcc, '', $subject, '', $message, '', '', false);
+					$monitor['notification'] = $error === '' ? 'sent' : 'failed';
+				} catch (Throwable $error) {
+					$monitor['notification'] = 'failed';
+				}
+				if ($monitor['notification'] === 'sent') {
+					$monitor['notified_signature'] = $signature;
+					$monitor['last_notified'] = $now;
+				} else {
+					cacti_log('SYSLOG PIPELINE: Health notification failed; retrying on a later poller run', false, 'SYSLOG');
+				}
+			} elseif ($due && ($monitor['notification'] ?? '') !== 'failed') {
+				$monitor['notification'] = 'pending';
+			} elseif (!isset($monitor['notification']) || $monitor['notification'] === 'disabled') {
+				$monitor['notification'] = 'ready';
+			}
+		}
+		syslog_status_set('pipeline_health_monitor', (string) json_encode($monitor));
+	} finally {
+		syslog_db_fetch_cell_prepared('SELECT RELEASE_LOCK(?)', [$lock], false);
+	}
+}
+
 /**
  * syslog_worker_stats_get - Collect the parallel worker statistics shown
  * on the Syslog Status tab.
@@ -5624,14 +5787,19 @@ function syslog_replication_record_error(string $message): void {
 }
 
 /** Read on-demand, indexed operational status for this local collector. */
-function syslog_replication_operational_status(): array {
+function syslog_replication_operational_status(bool $bounded = false): array {
 	global $syslogdb_default;
 	if (!syslog_replication_is_enabled()) {
 		return ['enabled' => false];
 	}
 
-	$row = syslog_db_fetch_row("SELECT COUNT(*) AS pending, MIN(created_at) AS oldest_pending
-		FROM `$syslogdb_default`.`syslog_replication_output`", false);
+	// The backlog index starts with acknowledged_at, created_at; health only needs the oldest pending row.
+	$queue = $bounded
+		? "(SELECT created_at FROM `$syslogdb_default`.`syslog_replication_output`
+			WHERE acknowledged_at IS NULL ORDER BY created_at LIMIT 1) AS pending_sample"
+		: "`$syslogdb_default`.`syslog_replication_output`";
+	$row = syslog_db_fetch_row("SELECT COUNT(*) AS pending, MIN(created_at) AS oldest_pending,
+		UNIX_TIMESTAMP(MIN(created_at)) AS oldest_pending_epoch FROM $queue", false);
 	$lease = syslog_db_fetch_row("SELECT acquired_at, heartbeat_at FROM `$syslogdb_default`.`syslog_replication_recovery`
 		WHERE name = 'recovery'", false);
 	$status = syslog_status_get();
@@ -5640,6 +5808,7 @@ function syslog_replication_operational_status(): array {
 		'state' => syslog_replication_get_state(),
 		'pending' => isset($row['pending']) ? (int) $row['pending'] : null,
 		'oldest_pending' => (string) ($row['oldest_pending'] ?? ''),
+		'oldest_pending_epoch' => isset($row['oldest_pending_epoch']) ? (int) $row['oldest_pending_epoch'] : null,
 		'recovery_active' => !empty($lease) && isset($lease['heartbeat_at']) && (int) $lease['heartbeat_at'] >= time() - SYSLOG_REPLICATION_RECOVERY_LEASE_SECONDS,
 		'recovery_started' => (string) ($lease['acquired_at'] ?? ''),
 		'last_success' => $status['replication_last_success'] ?? '',

includes/processing.php

@@ -78,6 +78,7 @@ function syslog_poller_bottom(): void {
 
 	if (syslog_config_safe()) {
 		syslog_connect();
+		syslog_pipeline_health_check();
 		syslog_status_set('last_polling_time', time());
 
 		$command_string = (string) read_config_option('path_php_binary');

includes/settings.php

@@ -52,6 +52,13 @@ function syslog_config_settings(): void {
 
 	$tabs['syslog'] = __('Syslog', 'syslog');
 
+	$notification_lists = [0 => __('None', 'syslog')];
+	if (db_table_exists('plugin_notification_lists')) {
+		foreach (db_fetch_assoc('SELECT id, name FROM plugin_notification_lists ORDER BY name') as $list) {
+			$notification_lists[(int) $list['id']] = $list['name'];
+		}
+	}
+
 	$temp = [
 		'syslog_header' => [
 			'friendly_name' => __('General Settings', 'syslog'),
@@ -191,7 +198,7 @@ function syslog_config_settings(): void {
 		],
 		'syslog_stale_threshold' => [
 			'friendly_name' => __('Collector Staleness Threshold', 'syslog'),
-			'description'   => __('The Syslog Status page shows a warning when no new message has arrived within this many seconds.  Set to 0 to use the default of 300 seconds.', 'syslog'),
+			'description'   => __('Pipeline health alerts warn when unprocessed messages, processing completion, or queued synchronization exceed this delay. The effective delay is at least three poller intervals and 300 seconds. Set to 0 to use 300 seconds.', 'syslog'),
 			'method'        => 'drop_array',
 			'default'       => '300',
 			'array'         => [
@@ -216,6 +223,37 @@ function syslog_config_settings(): void {
 				'50000' => __('%d Messages', 50000, 'syslog')
 			]
 		],
+		'syslog_pipeline_health_header' => [
+			'friendly_name' => __('Pipeline Health Alerts', 'syslog'),
+			'method'        => 'spacer',
+		],
+		'syslog_pipeline_health_notify' => [
+			'friendly_name' => __('Pipeline Health Notification List', 'syslog'),
+			'description'   => __('Select an existing Thold notification list for pipeline health alerts. Enabled list recipients are combined with the email addresses below, preserving BCC recipients. Lists require the Thold notification-list table.', 'syslog'),
+			'method'        => 'drop_array',
+			'array'         => $notification_lists,
+			'default'       => '0'
+		],
+		'syslog_pipeline_health_email' => [
+			'friendly_name' => __('Pipeline Health Email Recipients', 'syslog'),
+			'description'   => __('Comma-separated email recipients for pipeline failures and recovery. Leave blank and select no notification list to disable email; health remains visible on the Status tab and state changes are logged. Checks require the Cacti poller to run.', 'syslog'),
+			'method'        => 'textbox',
+			'size'          => 80,
+			'max_length'    => 255,
+			'default'       => ''
+		],
+		'syslog_pipeline_health_repeat' => [
+			'friendly_name' => __('Pipeline Health Reminder Interval', 'syslog'),
+			'description'   => __('Repeat notifications for ongoing pipeline problems after this interval. Failed email attempts are retried no more than once every five minutes.', 'syslog'),
+			'method'        => 'drop_array',
+			'default'       => '3600',
+			'array'         => [
+				'900' => __('%d minutes', 15, 'syslog'),
+				'3600' => __('%d hours', 1, 'syslog'),
+				'21600' => __('%d hours', 6, 'syslog'),
+				'86400' => __('%d days', 1, 'syslog')
+			]
+		],
 		'syslog_alert_retention' => [
 			'friendly_name' => __('Syslog Alert Retention', 'syslog'),
 			'description'   => __('This is the number of days to keep alert logs.', 'syslog'),