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

css/dashboard.css

@@ -86,55 +86,14 @@
 	align-items: start;
 }
 
+/* Card surfaces, headers, titles and tools reuse the Status tab classes. */
 .syslogDashboardCard {
-	box-sizing: border-box;
 	position: relative;
-	border: 1px solid var(--search-border, #cbd8d1);
-	border-radius: 8px;
-	background: var(--search-card, #fff);
-	color: var(--search-text, #23362f);
-	box-shadow: 0 1px 2px rgba(35, 54, 47, 0.08);
-	overflow: hidden;
-}
-
-.syslogDashboardCardHeader {
-	display: flex;
-	align-items: center;
-	justify-content: space-between;
-	gap: 8px;
-	padding: 8px 12px;
-	border-bottom: 1px solid var(--search-tint, #e2ede6);
-	background: var(--search-tint, #e2ede6);
-}
-
-.syslogDashboardCardTitle {
-	margin: 0;
-	font-size: 13px;
-	font-weight: 600;
-	line-height: 1.4;
-	overflow-wrap: anywhere;
-}
-
-.syslogDashboardCardHeader .syslogDashboardCardButton {
-	border: 0;
-	background: transparent;
-	color: var(--search-muted, #596d64);
-	cursor: pointer;
-	padding: 3px 4px;
-	border-radius: 4px;
-	font-size: 12px;
-}
-
-.syslogDashboardCardHeader .syslogDashboardCardButton:hover,
-.syslogDashboardCardHeader .syslogDashboardCardButton:focus {
-	background: var(--search-accent, #276448);
-	color: #fff;
-	outline: none;
 }
 
 .syslogChart {
 	min-height: 260px;
-	padding: 6px 6px 0 0;
+	padding: 6px 0 0;
 }
 
 /* Bottom-right drag handle; expands the pointer target to the corner. */
@@ -182,7 +141,7 @@
 }
 
 .syslogDashboardCardStatus {
-	margin: 40px 12px;
+	margin: 8px 0;
 	font-size: 12px;
 	color: var(--search-muted, #596d64);
 }
@@ -204,12 +163,8 @@
 }
 
 /* Drag-and-drop reorder affordances. */
-.syslogDashboardCardHeader {
-	cursor: grab;
-}
-
 .syslogDashboardCardDragging {
-	opacity: 0.45;
+	opacity: 0.5;
 }
 
 .syslogDashboardCardDropBefore {
@@ -266,7 +221,12 @@
 		grid-template-columns: 1fr;
 	}
 
+	.syslogDashboardCard {
+		/* Fit saved multi-column panels into the mobile grid. */
+		grid-column: 1 / -1 !important;
+	}
+
 	.syslogDashboardPanelActions {
 		margin-left: 0;
 	}
-}
+}

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,161 @@ 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)) {
+		$incoming = syslog_db_fetch_row("SELECT COUNT(*) AS pending, UNIX_TIMESTAMP(MIN(`$column`)) AS oldest
+			FROM `$syslogdb_default`.`syslog_incoming`", 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 %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();
+	$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 = strtotime($replication['oldest_pending']);
+		if ($replication['pending'] === null) {
+			$health['issues']['replication'] = __('Synchronization queue metrics are unavailable.', 'syslog');
+		} elseif ($replication['pending'] > 0 && $oldest === false) {
+			$health['issues']['replication'] = __('The oldest synchronization record timestamp is unavailable.', 'syslog');
+		} elseif ($replication['pending'] > 0 && ($replication['last_error'] !== ''
+			|| ($oldest !== false && $now - $oldest > $delay))) {
+			$health['issues']['replication'] = __('Synchronization has failed or queued records exceed the allowed delay (%s pending).', number_format($replication['pending']), '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.

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');