diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b7db7d7..9810cc35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,8 @@ * issue#710: Fixing Typo in thold_daemons.service File * issue#714: Increase the Name column to 255 characters * issue#719: Plugin Disabled due to mix of string and int +* issue#784: Retry failed queued email notifications with bounded exponential backoff +* issue#812: Recover stale notification claims, scope worker drains, and deprecate the ignored notification --thread option * issue: All Columns checkd on Thresholds page * issue: Special character previous value handling broken on data query indexes with special characters diff --git a/INFO b/INFO index b6c2c483..7099aeb2 100644 --- a/INFO +++ b/INFO @@ -21,7 +21,7 @@ [info] name = thold -version = 1.8.2 +version = 1.8.3 longname = Thresholds author = The Cacti Group email = diff --git a/README.md b/README.md index 46310dce..4f355898 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,28 @@ and become familiar with its settings. From there, you can provide overall control of thold, and set defaults for things like Email bodies, weekend exemptions, alert log retention, logging, etc. +When the notification queue is enabled, transient email failures are retried +up to five times with exponential backoff from one to eight minutes. The +Notification Queue page shows the attempt count and next eligible retry time. +After the fifth failed attempt the row becomes a terminal error so a permanent +SMTP or address failure cannot retry forever. + +Notification workers claim queue rows with their process ID and drain only +that claim. Unfinished rows are released when a worker stops or notifications +are suspended, while claims left by a hard-killed worker are recovered after +its process registration disappears. A cross-platform database advisory lease +prevents a second worker from taking the same slot. Because a database +reconnect can drop that lease while PHP is still running, a stale process row +is reclaimed immediately when an operating-system probe confirms that the old +PID is gone. Unknown liveness waits for one expired worker timeout before +recovery. A live PID is compared with the process registration time before a +stale worker is trusted, preventing a recycled operating-system PID from +blocking notification processing indefinitely. + +The legacy `thold_notify.php --thread=N` option remains accepted for operator +compatibility but is deprecated and ignored; queue ownership and worker +serialization are automatic. + As with much of Cacti, settings should be documented in line with the actual setting. If you find that any of these settings are ambiguous, please create a pull request with your proposed changes. diff --git a/includes/database.php b/includes/database.php index 279aff65..6512e6f4 100644 --- a/includes/database.php +++ b/includes/database.php @@ -1755,6 +1755,27 @@ function thold_upgrade_database($force = false) { db_execute('UPDATE plugin_notification_lists SET enabled = "on"'); } + if (cacti_version_compare($oldv, '1.8.3', '<')) { + db_add_column('notification_queue', [ + 'name' => 'attempt_count', + 'type' => 'int', + 'unsigned' => true, + 'NULL' => false, + 'default' => '0', + 'after' => 'error_message'] + ); + + db_add_column('notification_queue', [ + 'name' => 'next_attempt', + 'type' => 'timestamp', + 'NULL' => true, + 'default' => null, + 'after' => 'attempt_count'] + ); + + db_add_index('notification_queue', 'INDEX', 'retry_ready', ['event_processed', 'process_id', 'next_attempt']); + } + db_add_column('thold_data', [ 'name' => 'external_id', 'type' => 'varchar(20)', @@ -2123,12 +2144,15 @@ function thold_setup_database() { $data['columns'][] = ['name' => 'event_data', 'type' => 'longblob', 'NULL' => false, 'default' => '']; $data['columns'][] = ['name' => 'error_code', 'type' => 'int', 'NULL' => false, 'default' => '0']; $data['columns'][] = ['name' => 'error_message', 'type' => 'varchar(128)', 'NULL' => false, 'default' => '']; + $data['columns'][] = ['name' => 'attempt_count', 'type' => 'int', 'unsigned' => true, 'NULL' => false, 'default' => '0']; + $data['columns'][] = ['name' => 'next_attempt', 'type' => 'timestamp', 'NULL' => true, 'default' => null]; $data['columns'][] = ['name' => 'process_id', 'type' => 'int', 'unsigned' => true, 'NULL' => false, 'default' => '0']; $data['columns'][] = ['name' => 'event_processed', 'type' => 'tinyint', 'unsigned' => true, 'NULL' => false, 'default' => '0']; $data['columns'][] = ['name' => 'event_processed_time', 'type' => 'timestamp', 'NULL' => false, 'default' => '0000-00-00']; $data['columns'][] = ['name' => 'event_processed_runtime', 'type' => 'double', 'unsigned' => true, 'NULL' => false, 'default' => '0']; $data['primary'] = 'id'; $data['keys'][] = ['name' => 'topic_processed', 'columns' => 'topic`, `event_processed']; + $data['keys'][] = ['name' => 'retry_ready', 'columns' => 'event_processed`, `process_id`, `next_attempt']; $data['keys'][] = ['name' => 'process_id', 'columns' => 'process_id']; $data['keys'][] = ['name' => 'object_id', 'columns' => 'object_id']; $data['keys'][] = ['name' => 'host_id', 'columns' => 'host_id']; diff --git a/notify_queue.php b/notify_queue.php index 12f24645..c288a213 100644 --- a/notify_queue.php +++ b/notify_queue.php @@ -463,6 +463,18 @@ function clearFilter() { 'sort' => 'DESC', 'tip' => __('Did this notification result in an error. Hover on the error column for details.', 'thold') ], + 'attempt_count' => [ + 'display' => __('Attempts', 'thold'), + 'align' => 'right', + 'sort' => 'DESC', + 'tip' => __('The number of delivery attempts made for this notification.', 'thold') + ], + 'next_attempt' => [ + 'display' => __('Next Attempt', 'thold'), + 'align' => 'right', + 'sort' => 'DESC', + 'tip' => __('When a failed notification is eligible for its next retry.', 'thold') + ], 'event_processed_runtime' => [ 'display' => __('Run Time', 'thold'), 'align' => 'right', @@ -491,14 +503,9 @@ function clearFilter() { form_selectable_cell($n['id'], $n['id'], '', 'right'); form_selectable_cell($n['event_time'], $n['id'], '', 'right'); - form_selectable_cell($n['event_processed'] == 0 ? __('Pending', 'thold') : __('Done', 'thold'), $n['id'], '', 'right'); - if ($n['event_processed'] > 0) { - form_selectable_cell($n['error_code'] > 0 ? __('Errored', 'thold') : __('Success', 'thold'), $n['id'], '', 'right'); - form_selectable_cell(number_format_i18n($n['event_processed_runtime'], 2), $n['id'], '', 'right'); - } else { - form_selectable_cell(__('N/A', 'thold'), $n['id'], '', 'right'); - form_selectable_cell(__('N/A', 'thold'), $n['id'], '', 'right'); + foreach (thold_notification_queue_status_cells($n) as $cell) { + form_selectable_cell($cell, $n['id'], '', 'right'); } form_checkbox_cell($n['object_name'], $n['id']); diff --git a/tests/Helpers/CactiStubs.php b/tests/Helpers/CactiStubs.php index b7515c9e..d35013be 100644 --- a/tests/Helpers/CactiStubs.php +++ b/tests/Helpers/CactiStubs.php @@ -105,8 +105,8 @@ public static function reset() { /** * Record one Cacti function call. * - * @param string $fn Cacti function name. - * @param string $sql SQL text, or '' for non-query calls. + * @param string $fn Cacti function name. + * @param string $sql SQL text, or '' for non-query calls. * @param array $params Bound parameters, if any. * * @return void diff --git a/tests/Helpers/ThresholdScenario.php b/tests/Helpers/ThresholdScenario.php index de3fc6d4..d2b709d4 100644 --- a/tests/Helpers/ThresholdScenario.php +++ b/tests/Helpers/ThresholdScenario.php @@ -260,19 +260,11 @@ public function inMaintenance() { CactiStubs::willAlwaysReturn('plugin_maint_check_cacti_host', true); /* - * thold include_once()s the maint plugin when it reports enabled. The - * fixture supplies an empty file so the include succeeds; the function - * it would define is already stubbed. + * thold include_once()s the maint plugin when it reports enabled. Point + * base_path at a tracked, repository-contained Cacti root fixture so a + * unit test never writes into the caller's Cacti checkout. */ - $maint = dirname(__DIR__, 3) . '/maint'; - - if (!is_dir($maint)) { - mkdir($maint, 0755, true); - } - - if (!file_exists($maint . '/functions.php')) { - file_put_contents($maint . '/functions.php', "queueQueries(); - - $this->assertNotEmpty($queries); - - foreach ($queries as $sql) { - $this->assertStringNotContainsString('process_id =', $sql); - } + $this->assertSame([], $this->queueQueries()); + $this->assertNotEmpty(CactiStubs::$log); } /** @@ -109,49 +113,848 @@ public function testTheDrainRespectsARecordLimit(): void { } /** - * The collector claims only rows nobody holds, so a second instance - * cannot take rows the first is already working on. - * * @return void */ - public function testTheClaimTakesOnlyUnheldRows(): void { - $src = file_get_contents(dirname(__DIR__, 2) . '/thold_notify.php'); + public function testAClaimRecoversOrphansThenTakesOnlyUnheldRows(): void { + $this->assertSame(0, thold_notification_claim(0)); + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + CactiStubs::willReturn('db_affected_rows', 3); + + $this->assertSame(3, thold_notification_claim(4242)); + + $calls = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared'; + })); + + $this->assertCount(2, $calls); + $this->assertStringContainsString('LEFT JOIN processes', $calls[0]['sql']); + $this->assertStringContainsString('p.pid IS NULL', $calls[0]['sql']); + $this->assertSame(['thold_notify', 'child'], $calls[0]['params']); + $this->assertStringContainsString('(next_attempt IS NULL OR next_attempt <= NOW())', $calls[1]['sql']); + $this->assertStringContainsString('AND process_id = 0', $calls[1]['sql']); + $this->assertSame([4242], $calls[1]['params']); + } + + /** + * @return void + */ + public function testAClaimSkipsTheRecoveryUpdateWhenNoOrphanExists(): void { + thold_notification_claim(4242); + $updates = CactiStubs::callsTo('db_execute_prepared'); + + $this->assertCount(1, $updates); + $this->assertStringContainsString('SET process_id = ?', $updates[0]['sql']); + } + + /** + * @return void + */ + public function testAReleaseReturnsOnlyTheWorkersUnfinishedRows(): void { + $this->assertTrue(thold_notification_release_claim(0)); + $this->assertTrue(thold_notification_release_claim(4242)); + + $calls = CactiStubs::$calls; + $call = end($calls); + + $this->assertSame('db_execute_prepared', $call['fn']); + $this->assertStringContainsString('SET process_id = 0', $call['sql']); + $this->assertStringContainsString('AND event_processed = 0', $call['sql']); + $this->assertSame([4242], $call['params']); + } + + /** + * @return void + */ + public function testDatabaseLeaseOperationsAreConnectionScoped(): void { + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + CactiStubs::willReturn('db_fetch_cell_prepared', 1); - $this->assertMatchesRegularExpression( - '/SET process_id = \?\s+WHERE event_processed = 0\s+AND process_id = 0/', - $src + $this->assertSame('thold_notify_child_2', thold_notification_lock_name(2)); + $this->assertTrue(thold_notification_acquire_lock(2)); + $this->assertTrue(thold_notification_owns_lock(2)); + $this->assertTrue(thold_notification_release_lock(2)); + + $this->assertSame( + [ + "SELECT GET_LOCK(CONCAT(DATABASE(), ':', ?), 0)", + "SELECT IS_USED_LOCK(CONCAT(DATABASE(), ':', ?)) = CONNECTION_ID()", + "SELECT RELEASE_LOCK(CONCAT(DATABASE(), ':', ?))", + ], + array_column(CactiStubs::$calls, 'sql') ); + + foreach (CactiStubs::$calls as $call) { + $this->assertSame(['thold_notify_child_2'], $call['params']); + } } /** - * The claim has to follow the registration, or a second instance stamps - * its identifier over the first instance's rows before discovering that - * it should exit. - * * @return void */ - public function testTheClaimFollowsTheProcessRegistration(): void { - $src = file_get_contents(dirname(__DIR__, 2) . '/thold_notify.php'); + public function testRegistrationRequiresTheLeaseAndHandlesQueryFailures(): void { + $this->assertFalse(thold_notification_register_process(2, 300, static function () { + return false; + })); + $this->assertSame([], CactiStubs::$calls); + $this->assertNotEmpty(CactiStubs::$log); - $registered = strpos($src, "register_process_start('thold_notify'"); - $claimed = strpos($src, 'SET process_id = ?'); + CactiStubs::reset(); + CactiStubs::willReturn('db_fetch_row_prepared', false); + + $this->assertFalse(thold_notification_register_process(2, 300, static function () { + return true; + })); + $this->assertSame( + ['db_fetch_row_prepared', 'db_fetch_cell_prepared'], + array_column(CactiStubs::$calls, 'fn') + ); + + CactiStubs::reset(); + CactiStubs::willReturn('db_fetch_row_prepared', []); + $this->assertTrue(thold_notification_register_process(2, 300, static function () { + return true; + })); + $this->assertSame( + ['db_fetch_row_prepared', 'register_process_start'], + array_column(CactiStubs::$calls, 'fn') + ); - $this->assertNotFalse($registered); - $this->assertNotFalse($claimed); - $this->assertLessThan($claimed, $registered); + CactiStubs::reset(); + CactiStubs::willReturn('db_fetch_row_prepared', []); + CactiStubs::willReturn('register_process_start', false); + + $this->assertFalse(thold_notification_register_process(2, 300, static function () { + return true; + })); + $last = end(CactiStubs::$calls); + $this->assertSame('db_fetch_cell_prepared', $last['fn']); + $this->assertStringContainsString('RELEASE_LOCK', $last['sql']); + } + + /** + * @return void + */ + public function testRegistrationUsesLivenessAndBoundedHeartbeatFallbacks(): void { + $process = [ + 'pid' => 42, + 'started_at' => 500, + 'heartbeat_at' => 600, + 'current_timestamp' => 1000, + ]; + $lock = static function () { + return true; + }; + $same_process = static function () { + return 600; + }; + + CactiStubs::willReturn('db_fetch_row_prepared', $process); + $this->assertFalse(thold_notification_register_process(2, 300, $lock, static function () { + return true; + }, $same_process)); + + CactiStubs::reset(); + $fresh = $process; + $fresh['heartbeat_at'] = 900; + CactiStubs::willReturn('db_fetch_row_prepared', $fresh); + $this->assertFalse(thold_notification_register_process(2, 300, $lock, static function () { + return null; + })); + + CactiStubs::reset(); + CactiStubs::willReturn('db_fetch_row_prepared', $fresh); + $this->assertTrue(thold_notification_register_process(2, 300, $lock, static function () { + return false; + })); + + CactiStubs::reset(); + CactiStubs::willReturn('db_fetch_row_prepared', $process); + $this->assertTrue(thold_notification_register_process(2, 300, $lock, static function () { + return null; + })); + + CactiStubs::reset(); + $expired = $process; + $expired['heartbeat_at'] = 100; + $expired['current_timestamp'] = 2000; + CactiStubs::willReturn('db_fetch_row_prepared', $expired); + $this->assertFalse(thold_notification_register_process(2, 300, $lock, static function () { + return true; + }, static function () { + return 1600; + })); + + CactiStubs::reset(); + CactiStubs::willReturn('db_fetch_row_prepared', $expired); + $this->assertTrue(thold_notification_register_process(2, 300, $lock, static function () { + return true; + }, static function () { + return 700; + })); + + CactiStubs::reset(); + CactiStubs::willReturn('db_fetch_row_prepared', $expired); + $this->assertFalse(thold_notification_register_process(2, 300, $lock, static function () { + return true; + }, static function () { + return false; + })); + + CactiStubs::reset(); + CactiStubs::willReturn('db_fetch_row_prepared', $process); + $this->assertTrue(thold_notification_register_process(2, 300, $lock, static function () { + return false; + })); + $this->assertSame( + ['db_fetch_row_prepared', 'db_execute_prepared', 'unregister_process', 'register_process_start'], + array_column(CactiStubs::$calls, 'fn') + ); + $this->assertSame(['thold_notify', 'child', 2, 42], CactiStubs::$calls[2]['params']); + } + + /** + * @return void + */ + public function testProcessProbeDistinguishesDeathFromPermissionErrors(): void { + $missing = static function () { + return false; + }; + + $this->assertTrue(thold_notification_probe_process(42, static function () { + return 42; + })); + $this->assertTrue(thold_notification_probe_process(42, $missing, static function () { + return true; + })); + $this->assertTrue(thold_notification_probe_process(42, $missing, $missing, static function () { + return 1; + })); + $this->assertFalse(thold_notification_probe_process(42, $missing, $missing, static function () { + return 3; + })); + $this->assertNull(thold_notification_probe_process(42, $missing, false, false)); + $this->assertNull(thold_notification_probe_process(42, $missing, $missing, false)); + $this->assertNull(thold_notification_probe_process(42, $missing, $missing, static function () { + return 99; + })); + $this->assertTrue(thold_notification_probe_process(getmypid())); + $this->assertTrue(thold_notification_probe_process(getmypid(), $missing)); + $this->assertContains(thold_notification_probe_process(2147483647, $missing, $missing), [false, null], true); + $this->assertFalse(thold_notification_probe_process(0)); + } + + /** + * @return void + */ + public function testProcessIdentityDetectsPidReuse(): void { + $this->assertNull(thold_notification_process_matches(0, 500, 1000)); + $this->assertTrue(thold_notification_process_matches(42, 500, 1000, static function () { + return 600; + })); + $this->assertTrue(thold_notification_process_matches(42, 500, 1000, static function () { + return 495; + })); + $this->assertFalse(thold_notification_process_matches(42, 500, 1000, static function () { + return 494; + })); + $this->assertNull(thold_notification_process_matches(42, 500, 1000, static function () { + return false; + })); + $this->assertNull(thold_notification_process_matches(42, 500, 1000, static function () { + throw new RuntimeException('process probe failed'); + })); + } + + /** + * @return void + */ + public function testElapsedProcessTimeAcceptsLinuxAndPortableFormats(): void { + $this->assertSame(83, thold_notification_elapsed_seconds('83')); + $this->assertSame(83, thold_notification_elapsed_seconds('01:23')); + $this->assertSame(3723, thold_notification_elapsed_seconds('01:02:03')); + $this->assertSame(90123, thold_notification_elapsed_seconds('1-01:02:03')); + $this->assertFalse(thold_notification_elapsed_seconds('unknown')); + } + + /** + * @return void + */ + public function testProcessAgeProbeFallsBackToPortablePsOutput(): void { + $fields = []; + $runner = static function ($field, $pid) use (&$fields) { + $fields[] = [$field, $pid]; + + return $field === 'etimes' ? [1, []] : [0, ['01:23']]; + }; + + $this->assertSame(83, thold_notification_probe_elapsed(42, $runner)); + $this->assertSame([['etimes', 42], ['etime', 42]], $fields); + $this->assertFalse(thold_notification_probe_elapsed(42, false)); + $this->assertFalse(thold_notification_probe_elapsed(42, static function () { + return [0, ['unknown']]; + })); + $this->assertFalse(thold_notification_probe_elapsed(42, static function () { + throw new RuntimeException('ps failed'); + })); + } + + /** + * @return void + */ + public function testDefaultProcessIdentityProbeIsTimezoneIndependent(): void { + $timezone = date_default_timezone_get(); + $now = time(); + + try { + foreach (['UTC', 'America/Los_Angeles'] as $candidate) { + date_default_timezone_set($candidate); + $this->assertTrue(thold_notification_process_matches(getmypid(), $now, $now)); + } + } finally { + date_default_timezone_set($timezone); + } + } + + /** + * @return void + */ + public function testDefaultUnixProbeKeepsALiveWorkerRegistered(): void { + $now = time(); + CactiStubs::willReturn('db_fetch_row_prepared', [ + 'pid' => getmypid(), + 'started_at' => $now, + 'heartbeat_at' => $now - 400, + 'current_timestamp' => $now, + ]); + + $this->assertFalse(thold_notification_register_process(2, 300, static function () { + return true; + })); + $this->assertSame( + ['db_fetch_row_prepared', 'db_fetch_cell_prepared'], + array_column(CactiStubs::$calls, 'fn') + ); + } + + /** + * @return void + */ + public function testRegistrationFailureAfterReclaimReleasesTheLease(): void { + CactiStubs::willReturn('db_fetch_row_prepared', [ + 'pid' => 42, + 'started_at' => 500, + 'heartbeat_at' => 600, + 'current_timestamp' => 1000, + ]); + CactiStubs::willReturn('register_process_start', false); + + $this->assertFalse(thold_notification_register_process(2, 300, static function () { + return true; + }, static function () { + return false; + })); + $last = end(CactiStubs::$calls); + $this->assertSame('db_fetch_cell_prepared', $last['fn']); + $this->assertStringContainsString('RELEASE_LOCK', $last['sql']); + } + + /** + * @return void + */ + public function testASuspendedRunReleasesItsClaimAndRemainsScoped(): void { + CactiStubs::$configOptions['thold_notification_suspended'] = '1'; + CactiStubs::willReturn('db_affected_rows', 2); + $heartbeats = 0; + + $this->assertSame(2, thold_notification_run(77, 'all', static function () use (&$heartbeats) { + $heartbeats++; + })); + $this->assertSame(5, $heartbeats); + + foreach ($this->queueQueries() as $sql) { + if (strpos($sql, 'SELECT *') !== false) { + $this->assertStringContainsString('process_id = 77', $sql); + } + } + + $calls = CactiStubs::$calls; + $release = end($calls); + $this->assertSame('db_execute_prepared', $release['fn']); + $this->assertStringContainsString('SET process_id = 0', $release['sql']); + $this->assertSame([77], $release['params']); + } + + /** + * @return void + */ + public function testAHeartbeatFailureCannotSkipClaimRelease(): void { + CactiStubs::$configOptions['thold_notification_suspended'] = '1'; + $heartbeats = 0; + + try { + thold_notification_run(77, 'all', static function () use (&$heartbeats) { + $heartbeats++; + + if ($heartbeats === 5) { + throw new RuntimeException('heartbeat failed'); + } + }); + + $this->fail('Expected the heartbeat failure to propagate.'); + } catch (RuntimeException $error) { + $this->assertSame('heartbeat failed', $error->getMessage()); + } + + $calls = CactiStubs::$calls; + $release = end($calls); + $this->assertSame('db_execute_prepared', $release['fn']); + $this->assertStringContainsString('SET process_id = 0', $release['sql']); + $this->assertSame([77], $release['params']); + } + + /** + * @return void + */ + public function testQueueLoopsHeartbeatForEveryRecord(): void { + $records = [ + ['id' => 91, 'topic' => 'unknown-device'], + ['id' => 92, 'topic' => 'unknown-device'], + ]; + CactiStubs::willReturn('db_fetch_assoc_prepared', $records); + CactiStubs::willReturn('db_fetch_assoc_prepared', $records); + $heartbeats = 0; + $heartbeat = static function () use (&$heartbeats) { + $heartbeats++; + }; + + process_device_notifications(77, 'all', 0, $heartbeat); + process_non_device_notifications(77, 'all', 0, $heartbeat); + + $this->assertSame(4, $heartbeats); + + $terminal = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared' + && strpos($call['sql'], 'error_code = 1, error_message') !== false; + })); + $this->assertCount(4, $terminal); + + foreach ($terminal as $call) { + $this->assertStringContainsString('event_processed = 1', $call['sql']); + $this->assertSame(77, $call['params'][2]); + } + } + + /** + * @return void + */ + public function testUnknownTopicMessageIsBoundAndTruncated(): void { + $this->assertTrue(thold_notification_reject_unknown_topic(91, 77, str_repeat('é', 200))); + $call = end(CactiStubs::$calls); + + $this->assertSame('db_execute_prepared', $call['fn']); + $this->assertSame(128, mb_strlen($call['params'][0], 'UTF-8')); + $this->assertSame([91, 77], array_slice($call['params'], 1)); + $this->assertStringContainsString('AND process_id = ?', $call['sql']); + } + + /** + * @return void + */ + public function testCompletionDatabaseFailureUsesASafeTerminalFallback(): void { + CactiStubs::willReturn('db_execute_prepared', false); + + $this->assertTrue(thold_notification_complete('UPDATE notification_queue SET event_processed = 1', [77], [91], 77)); + $calls = CactiStubs::callsTo('db_execute_prepared'); + $this->assertCount(2, $calls); + $this->assertStringContainsString('Completion update failed', $calls[1]['sql']); + $this->assertSame([91, 77], $calls[1]['params']); + } + + /** + * @return void + */ + public function testCompletionAbortsWhenBothDatabaseUpdatesFail(): void { + CactiStubs::willReturn('db_execute_prepared', false); + CactiStubs::willReturn('db_execute_prepared', false); + + $this->expectException(UnexpectedValueException::class); + $this->expectExceptionMessage('database update failed'); + thold_notification_complete('UPDATE notification_queue SET event_processed = 1', [77], [91], 77); + } + + /** + * @return void + */ + public function testNotificationErrorsNormalizeInvalidUtf8(): void { + $message = thold_notification_error_message("\xff\xfe" . str_repeat('a', 200)); + + $this->assertTrue(mb_check_encoding($message, 'UTF-8')); + $this->assertLessThanOrEqual(128, mb_strlen($message, 'UTF-8')); + } + + /** + * @return void + */ + public function testCompletionRejectsAnEmptyClaim(): void { + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('queue row(s)'); + thold_notification_complete('UPDATE notification_queue SET event_processed = 1', [77], [], 77); + } + + /** + * @return void + */ + public function testRevokedSingleRowClaimIsLoggedAndAbortsTheDrain(): void { + CactiStubs::willReturn('db_affected_rows', 0); + + try { + thold_notification_complete( + 'UPDATE notification_queue SET event_processed = 1 WHERE id = ? AND process_id = ?', + [91, 77], + [91], + 77 + ); + $this->fail('Expected revoked ownership to stop the drain.'); + } catch (RuntimeException $error) { + $this->assertStringContainsString('queue row(s) 91', $error->getMessage()); + } + + $this->assertStringContainsString('process 77', end(CactiStubs::$log)); } /** - * Without a way to ask whether the recorded process is still alive, the - * run must stand down rather than proceed beside it. It previously fell - * through and drained the queue a second time. - * * @return void */ - public function testAnInstanceThatCannotCheckForAPeerStandsDown(): void { - $src = file_get_contents(dirname(__DIR__, 2) . '/thold_notify.php'); + public function testPartiallyRevokedGroupedClaimIsLoggedAndAbortsTheDrain(): void { + CactiStubs::willReturn('db_affected_rows', 1); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('queue row(s) 91, 92'); + thold_notification_complete( + 'UPDATE notification_queue SET event_processed = 1 WHERE id IN (91, 92) AND process_id = ?', + [77], + [91, 92], + 77 + ); + } + + /** + * @return void + */ + public function testDeviceCommandAndGroupedMailComplete(): void { + CactiStubs::$configOptions['alert_deadnotify_one_mail'] = 'on'; + CactiStubs::$configOptions['alert_deadnotify_subject'] = 'Device alerts'; + CactiStubs::willReturn('mailer', str_repeat('é', 200)); + CactiStubs::willReturn('db_fetch_assoc_prepared', [ + [ + 'id' => 101, + 'topic' => 'thold_dhost_cmd', + 'event_data' => json_encode([ + 'environment' => ['THOLD_DEVICE_TEST=1'], + 'command' => "printf '%0130d' 0", + 'data' => ['id' => 7], + ]), + ], + [ + 'id' => 102, + 'topic' => 'thold_dhost_mail', + 'event_data' => json_encode([ + 'from' => ['sender@example.com'], + 'to' => 'recipient@example.com', + 'cc' => '', + 'bcc' => '', + 'replyto' => '', + 'subject' => 'Device down', + 'body' => 'Down', + 'body_text' => 'Down', + 'attachments' => [], + 'headers' => [], + 'html' => true, + ]), + ], + ]); + $heartbeats = 0; + + process_device_notifications(77, 'all', 0, static function () use (&$heartbeats) { + $heartbeats++; + }); + putenv('THOLD_DEVICE_TEST'); + + $updates = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared' + && (strpos($call['sql'], 'event_processed = 1') !== false + || strpos($call['sql'], 'attempt_count = CASE id') !== false); + })); + + $this->assertCount(2, $updates); + $this->assertSame(101, $updates[0]['params'][3]); + $this->assertSame(77, $updates[0]['params'][4]); + $this->assertSame(128, strlen($updates[0]['params'][1])); + $this->assertSame(1, $updates[1]['params'][0]); + $this->assertSame(128, mb_strlen($updates[1]['params'][1], 'UTF-8')); + $this->assertSame([102, 77], array_slice($updates[1]['params'], -2)); + $this->assertStringContainsString('AND process_id = ?', $updates[0]['sql']); + $this->assertStringContainsString('AND process_id = ?', $updates[1]['sql']); + $this->assertSame(3, $heartbeats); + } - $this->assertMatchesRegularExpression('/\$running = true;/', $src); - $this->assertMatchesRegularExpression('/if \(\$running\) \{\s+exit\(1\);/', $src); + /** + * @return void + */ + public function testIndividualDeviceMailCompletionRequiresItsOwner(): void { + CactiStubs::willReturn('mailer', str_repeat('é', 200)); + CactiStubs::willReturn('db_fetch_assoc_prepared', [[ + 'id' => 104, + 'topic' => 'thold_dhost_mail', + 'event_data' => json_encode([ + 'from' => ['sender@example.com'], + 'to' => 'recipient@example.com', + 'bcc' => '', + 'replyto' => '', + 'subject' => 'Device down', + 'body' => 'Down', + 'body_text' => 'Down', + 'attachments' => [['attachment' => base64_encode('attachment')]], + 'headers' => [], + 'html' => true, + ]), + ]]); + + process_device_notifications(77, 'all', 0); + $call = end(CactiStubs::$calls); + + $this->assertStringContainsString('AND process_id = ?', $call['sql']); + $this->assertSame(128, mb_strlen($call['params'][1], 'UTF-8')); + $this->assertSame([104, 77], array_slice($call['params'], -2)); + } + + /** + * @return void + */ + public function testNonDeviceCommandCompletesWithItsEnvironment(): void { + CactiStubs::willReturn('db_fetch_assoc_prepared', [[ + 'id' => 103, + 'topic' => 'thold_cmd', + 'event_data' => json_encode([ + 'environment' => ['THOLD_COMMAND_TEST=1'], + 'command' => "printf '%0130d' 0", + 'data' => ['id' => 8], + ]), + ]]); + + process_non_device_notifications(77, 'all', 0); + putenv('THOLD_COMMAND_TEST'); + + $call = end(CactiStubs::$calls); + $this->assertSame('db_execute_prepared', $call['fn']); + $this->assertStringContainsString('event_processed = 1', $call['sql']); + $this->assertSame(128, strlen($call['params'][1])); + $this->assertSame(103, $call['params'][3]); + $this->assertSame(77, $call['params'][4]); + $this->assertStringContainsString('AND process_id = ?', $call['sql']); + } + + /** + * @return void + */ + public function testNonDeviceMailCompletionRequiresItsOwner(): void { + CactiStubs::willReturn('mailer', str_repeat('é', 200)); + CactiStubs::willReturn('db_fetch_assoc_prepared', [[ + 'id' => 105, + 'topic' => 'thold_mail', + 'event_data' => json_encode([ + 'from' => ['sender@example.com'], + 'to' => 'recipient@example.com', + 'bcc' => '', + 'replyto' => '', + 'subject' => 'Threshold alert', + 'body' => 'Alert', + 'body_text' => 'Alert', + 'attachments' => [], + 'headers' => [], + 'html' => true, + ]), + ]]); + + process_non_device_notifications(77, 'all', 0); + $call = end(CactiStubs::$calls); + + $this->assertStringContainsString('AND process_id = ?', $call['sql']); + $this->assertSame(128, mb_strlen($call['params'][1], 'UTF-8')); + $this->assertSame([105, 77], array_slice($call['params'], -2)); + } + + /** + * @return void + */ + public function testCleanupReleasesAndUnregistersOnlyOnce(): void { + $registered = true; + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + + $this->assertTrue(thold_notification_cleanup(77, 2, $registered)); + $this->assertFalse($registered); + $this->assertSame( + ['db_execute_prepared', 'unregister_process', 'db_fetch_cell_prepared'], + array_column(CactiStubs::$calls, 'fn') + ); + + $this->assertTrue(thold_notification_cleanup(77, 2, $registered)); + $this->assertCount(3, CactiStubs::$calls); + } + + /** + * @return void + */ + public function testCleanupReportsAClaimReleaseFailure(): void { + $registered = true; + CactiStubs::willReturn('db_execute_prepared', false); + + $this->assertFalse(thold_notification_cleanup(77, 2, $registered)); + $this->assertFalse($registered); + $this->assertNotEmpty(CactiStubs::$log); + } + + /** + * @return void + */ + public function testNamedShutdownIsIdempotent(): void { + $GLOBALS['notification_registered'] = true; + $GLOBALS['pid'] = 77; + $GLOBALS['thread'] = 2; + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + + thold_notification_shutdown(); + thold_notification_shutdown(); + + $this->assertFalse($GLOBALS['notification_registered']); + $this->assertCount(3, CactiStubs::$calls); + } + + /** + * @return void + */ + public function testMainInstallsShutdownBeforeRegistrationAndFailsClosed(): void { + $events = []; + + $this->assertFalse(thold_notification_main( + 2, + 300, + 77, + static function ($callback) use (&$events) { + $events[] = 'shutdown:' . $callback; + }, + static function ($thread, $timeout) use (&$events) { + $events[] = "register:$thread:$timeout"; + + return false; + }, + static function () use (&$events) { + $events[] = 'run'; + + return 0; + } + )); + $this->assertSame(['shutdown:thold_notification_shutdown', 'register:2:300'], $events); + $this->assertFalse($GLOBALS['notification_registered']); + } + + /** + * @return void + */ + public function testMainLogsAndCleansUpARegistrationException(): void { + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + + $this->assertFalse(thold_notification_main( + 2, + 300, + 77, + static function () { + }, + static function () { + throw new RuntimeException('registration failed'); + }, + static function () { + throw new RuntimeException('drain must not run'); + } + )); + + $this->assertStringContainsString('registration failed', end(CactiStubs::$log)); + $this->assertFalse($GLOBALS['notification_registered']); + $this->assertSame( + ['db_execute_prepared', 'unregister_process', 'db_fetch_cell_prepared'], + array_column(CactiStubs::$calls, 'fn') + ); + } + + /** + * @return void + */ + public function testMainRunsTheLeaseHeartbeatAndAlwaysCleansUp(): void { + $events = []; + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + + $this->assertTrue(thold_notification_main( + 2, + 300, + 77, + static function ($callback) use (&$events) { + $events[] = 'shutdown:' . $callback; + }, + static function ($thread, $timeout) use (&$events) { + $events[] = "register:$thread:$timeout"; + + return true; + }, + static function ($pid, $limit, $heartbeat) use (&$events) { + $events[] = "run:$pid:$limit"; + $heartbeat(); + + return 4; + } + )); + $this->assertSame( + ['shutdown:thold_notification_shutdown', 'register:2:300', 'run:77:all'], + $events + ); + $this->assertFalse($GLOBALS['notification_registered']); + $this->assertSame(2, $GLOBALS['thread']); + $this->assertContains('heartbeat_process', array_column(CactiStubs::$calls, 'fn')); + $this->assertStringContainsString('Notifications:4', end(CactiStubs::$log)); + } + + /** + * @return void + */ + public function testHeartbeatThrowsAsSoonAsTheLeaseIsLost(): void { + CactiStubs::willReturn('db_fetch_cell_prepared', 0); + + $this->expectException(RuntimeException::class); + $this->expectExceptionMessage('Notification worker lease was lost.'); + thold_notification_heartbeat(2); + } + + /** + * @return void + */ + public function testMainLogsLeaseLossAndReturnsFailureAfterCleanup(): void { + CactiStubs::willReturn('db_fetch_cell_prepared', 1); + + $this->assertFalse(thold_notification_main( + 2, + 300, + 77, + static function () { + }, + static function () { + return true; + }, + static function () { + throw new RuntimeException('lease lost'); + } + )); + $this->assertFalse($GLOBALS['notification_registered']); + $this->assertStringContainsString('lease lost', end(CactiStubs::$log)); } } diff --git a/tests/Unit/NotificationQueueRetryTest.php b/tests/Unit/NotificationQueueRetryTest.php new file mode 100644 index 00000000..6d2ee421 --- /dev/null +++ b/tests/Unit/NotificationQueueRetryTest.php @@ -0,0 +1,304 @@ + + */ + private function lastPreparedCall() { + $calls = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared'; + })); + + $this->assertNotEmpty($calls); + + return end($calls); + } + + /** + * @return array + * @param mixed $id + * @param mixed $topic + * @param mixed $attempts + */ + private function mailRow($id, $topic = 'thold_mail', $attempts = 0) { + return [ + 'id' => $id, + 'topic' => $topic, + 'attempt_count' => $attempts, + 'event_data' => json_encode([ + 'from' => ['sender@example.com'], + 'to' => 'recipient@example.com', + 'cc' => '', + 'bcc' => '', + 'replyto' => '', + 'subject' => 'Threshold alert', + 'body' => 'Alert', + 'body_text' => 'Alert', + 'attachments' => [], + 'headers' => [], + 'html' => true, + ]), + ]; + } + + /** + * @return void + */ + public function testRetryDelayUsesBoundedExponentialBackoff(): void { + $this->assertSame(60, thold_notification_retry_delay(-1)); + $this->assertSame(60, thold_notification_retry_delay(0)); + $this->assertSame(60, thold_notification_retry_delay(1)); + $this->assertSame(120, thold_notification_retry_delay(2)); + $this->assertSame(480, thold_notification_retry_delay(4)); + $this->assertSame(1920, thold_notification_retry_delay(6)); + $this->assertSame(3600, thold_notification_retry_delay(7)); + } + + /** + * @return void + */ + public function testQueueStatusCellsFollowTheirHeaderOrder(): void { + $cells = thold_notification_queue_status_cells([ + 'event_processed' => 1, + 'error_code' => 1, + 'attempt_count' => 4, + 'next_attempt' => null, + 'event_processed_runtime' => 0.25, + ]); + + $this->assertSame( + ['event_processed', 'error_code', 'attempt_count', 'next_attempt', 'event_processed_runtime'], + array_keys($cells) + ); + $this->assertSame(['Done', 'Errored', 4, 'N/A', '0.25'], array_values($cells)); + + $pending = thold_notification_queue_status_cells(['event_processed' => 0]); + $this->assertSame(['Pending', 'N/A', 0, 'N/A', 'N/A'], array_values($pending)); + } + + /** + * @return void + */ + public function testSuccessfulDeliveryIsTerminal(): void { + thold_notification_record_delivery(42, 77, '', 0.25, 2); + + $call = $this->lastPreparedCall(); + $sql = preg_replace('/\s+/', ' ', $call['sql']); + + $this->assertSame(0, $call['params'][0]); + $this->assertSame('', $call['params'][1]); + $this->assertStringContainsString('next_attempt = CASE id', $sql); + $this->assertStringContainsString('THEN NULL', $sql); + $this->assertSame([42, 3], array_slice($call['params'], 2, 2)); + $this->assertSame(1, $call['params'][7]); + $this->assertSame([42, 0.25, 42, 77], array_slice($call['params'], -4)); + $this->assertStringContainsString('AND process_id = ?', $sql); + } + + /** + * @return void + */ + public function testTransientFailureReleasesTheClaimAndSchedulesRetry(): void { + thold_notification_record_delivery(42, 77, "smtp\ndown", 0.5); + + $call = $this->lastPreparedCall(); + $sql = preg_replace('/\s+/', ' ', $call['sql']); + + $this->assertStringContainsString('THEN FROM_UNIXTIME', $sql); + $this->assertStringContainsString('process_id = CASE id', $sql); + $this->assertStringContainsString('THEN 0', $sql); + $this->assertSame([1, 'smtp down', 42, 1, 42, 60], array_slice($call['params'], 0, 6)); + $this->assertSame(0, $call['params'][8]); + $this->assertSame([42, 0.5, 42, 77], array_slice($call['params'], -4)); + } + + /** + * @return void + */ + public function testFifthFailureIsTerminal(): void { + thold_notification_record_delivery(42, 77, 'permanent failure', 0.5, 4); + + $call = $this->lastPreparedCall(); + $sql = preg_replace('/\s+/', ' ', $call['sql']); + + $this->assertStringContainsString('THEN NULL', $sql); + $this->assertSame([1, 'permanent failure', 42, 5], array_slice($call['params'], 0, 4)); + $this->assertSame(1, $call['params'][7]); + $this->assertSame([42, 0.5, 42, 77], array_slice($call['params'], -4)); + } + + /** + * @return void + */ + public function testFailureMessageFitsTheQueueColumn(): void { + thold_notification_record_delivery(42, 77, str_repeat('x', 200), 0.5); + + $call = $this->lastPreparedCall(); + + $this->assertSame(128, strlen($call['params'][1])); + } + + /** + * @return void + */ + public function testClaimAndBothDrainsIgnoreRetriesThatAreNotReady(): void { + thold_notification_claim(77); + + $claims = array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared' && + strpos($call['sql'], 'SET process_id = ?') !== false; + }); + + $this->assertCount(1, $claims); + $claim = reset($claims); + $this->assertStringContainsString('(next_attempt IS NULL OR next_attempt <= NOW())', $claim['sql']); + + thold_notification_execute(77); + + $queries = array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_fetch_assoc_prepared' && + strpos($call['sql'], 'notification_queue') !== false; + }); + + $this->assertCount(2, $queries); + + foreach ($queries as $call) { + $this->assertStringContainsString('(next_attempt IS NULL OR next_attempt <= NOW())', $call['sql']); + } + } + + /** + * @return void + */ + public function testIndividualDeviceMailFailureUsesTheRetryRecorder(): void { + CactiStubs::$configOptions['alert_deadnotify_one_mail'] = ''; + CactiStubs::willReturnFor('db_fetch_assoc_prepared', "topic IN ('thold_dhost_mail'", [ + $this->mailRow(51, 'thold_dhost_mail', 2), + ]); + CactiStubs::willReturn('mailer', 'temporary SMTP failure'); + + process_device_notifications(77, 'all', 0); + + $call = $this->lastPreparedCall(); + + $this->assertSame([1, 'temporary SMTP failure', 51, 3, 51, 240], array_slice($call['params'], 0, 6)); + $this->assertSame([51, $call['params'][10], 51, 77], array_slice($call['params'], -4)); + $this->assertSame(0, $call['params'][8]); + $this->assertStringContainsString('process_id = CASE id', $call['sql']); + } + + /** + * @return void + */ + public function testGroupedDeviceMailRecordsEveryAttempt(): void { + CactiStubs::$configOptions['alert_deadnotify_one_mail'] = 'on'; + CactiStubs::$configOptions['alert_deadnotify_subject'] = 'Device alerts'; + CactiStubs::willReturnFor('db_fetch_assoc_prepared', "topic IN ('thold_dhost_mail'", [ + $this->mailRow(61, 'thold_dhost_mail', 0), + $this->mailRow(63, 'thold_dhost_mail', 0), + $this->mailRow(62, 'thold_uhost_mail', 3), + $this->mailRow(64, 'thold_uhost_mail', 4), + ]); + CactiStubs::willReturn('mailer', 'temporary SMTP failure'); + CactiStubs::willReturn('db_affected_rows', 4); + + process_device_notifications(77, 'all', 0); + + $calls = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared' && strpos($call['sql'], 'attempt_count') !== false; + })); + + $this->assertCount(1, $calls); + $this->assertStringContainsString('attempt_count = CASE id', $calls[0]['sql']); + $this->assertStringContainsString('event_processed = CASE id', $calls[0]['sql']); + $this->assertSame([61, 1, 63, 1, 62, 4, 64, 5], array_slice($calls[0]['params'], 2, 8)); + $this->assertSame([61, 60, 63, 60, 62, 480, 64], array_slice($calls[0]['params'], 10, 7)); + $this->assertSame([61, 63, 62, 64], array_slice($calls[0]['params'], 17, 4)); + $this->assertSame([61, 0, 63, 0, 62, 0, 64, 1], array_slice($calls[0]['params'], 21, 8)); + $this->assertSame([61, 63, 62, 64], array_slice($calls[0]['params'], 29, 4)); + $this->assertSame([61, 63, 62, 64, 77], array_slice($calls[0]['params'], -5)); + $this->assertSame('temporary SMTP failure', $calls[0]['params'][1]); + } + + /** + * @return void + */ + public function testGroupedDeliveryHandlesEmptyInvalidAndSuccessfulBatches(): void { + $this->assertTrue(thold_notification_record_deliveries([], 77, '', 0.25)); + $this->assertTrue(thold_notification_record_deliveries([-1 => 0], 77, '', 0.25)); + $this->assertSame([], CactiStubs::$calls); + + CactiStubs::willReturn('db_affected_rows', 2); + $this->assertTrue(thold_notification_record_deliveries([81 => 0, 82 => 4], 77, '', 0.25)); + + $calls = array_values(array_filter(CactiStubs::$calls, static function ($call) { + return $call['fn'] === 'db_execute_prepared' && strpos($call['sql'], 'attempt_count = CASE id') !== false; + })); + + $this->assertCount(1, $calls); + $this->assertSame(0, $calls[0]['params'][0]); + $this->assertSame('', $calls[0]['params'][1]); + $this->assertSame([81, 82, 77], array_slice($calls[0]['params'], -3)); + $this->assertStringNotContainsString('FROM_UNIXTIME', $calls[0]['sql']); + $this->assertStringContainsString('process_id = CASE id', $calls[0]['sql']); + $this->assertStringContainsString('THEN process_id', $calls[0]['sql']); + $this->assertStringContainsString('THEN NOW()', $calls[0]['sql']); + $this->assertSame([81, 1, 82, 1], array_slice($calls[0]['params'], 10, 4)); + } + + /** + * @return void + */ + public function testGroupedTerminalFailuresStayClaimedAndComplete(): void { + CactiStubs::willReturn('db_affected_rows', 2); + $this->assertTrue(thold_notification_record_deliveries([91 => 4, 92 => 5], 77, 'permanent failure', 0.5)); + + $call = $this->lastPreparedCall(); + + $this->assertSame([91, 5, 92, 6], array_slice($call['params'], 2, 4)); + $this->assertSame([91, 1, 92, 1], array_slice($call['params'], 10, 4)); + $this->assertStringNotContainsString('FROM_UNIXTIME', $call['sql']); + $this->assertStringContainsString('THEN process_id', $call['sql']); + $this->assertStringContainsString('THEN NOW()', $call['sql']); + } + + /** + * @return void + */ + public function testNonDeviceMailFailureUsesTheRetryRecorder(): void { + CactiStubs::willReturnFor('db_fetch_assoc_prepared', "topic NOT IN ('thold_dhost_mail'", [ + $this->mailRow(71, 'thold_mail', 1), + ]); + CactiStubs::willReturn('mailer', 'temporary SMTP failure'); + + process_non_device_notifications(77, 'all', 0); + + $call = $this->lastPreparedCall(); + + $this->assertSame([1, 'temporary SMTP failure', 71, 2, 71, 120], array_slice($call['params'], 0, 6)); + $this->assertSame([71, $call['params'][10], 71, 77], array_slice($call['params'], -4)); + $this->assertSame(0, $call['params'][8]); + $this->assertStringContainsString('process_id = CASE id', $call['sql']); + } +} diff --git a/tests/Unit/ThresholdTimeBasedCharacterizationTest.php b/tests/Unit/ThresholdTimeBasedCharacterizationTest.php index e5a87fd5..5bc0e5af 100644 --- a/tests/Unit/ThresholdTimeBasedCharacterizationTest.php +++ b/tests/Unit/ThresholdTimeBasedCharacterizationTest.php @@ -110,9 +110,16 @@ public function testRestoralResetsTheFailCounts(): void { * @return void */ public function testMaintenanceWindowSuppressesNotification(): void { - $outcome = $this->bounded(['lastread' => 95, 'time_fail_trigger' => 1]) - ->inMaintenance() - ->poll(); + $scenario = $this->bounded(['lastread' => 95, 'time_fail_trigger' => 1]) + ->inMaintenance(); + $fixture_root = realpath(dirname(__DIR__) . '/fixtures'); + $base_path = realpath($GLOBALS['config']['base_path']); + + $this->assertNotFalse($fixture_root); + $this->assertNotFalse($base_path); + $this->assertStringStartsWith($fixture_root . DIRECTORY_SEPARATOR, $base_path . DIRECTORY_SEPARATOR); + + $outcome = $scenario->poll(); $this->assertSame(0, $outcome->mailCount()); } diff --git a/tests/bin/patch-coverage.php b/tests/bin/patch-coverage.php index 43e0a936..07d45250 100644 --- a/tests/bin/patch-coverage.php +++ b/tests/bin/patch-coverage.php @@ -77,6 +77,13 @@ function changed_lines($base_ref) { foreach (explode("\n", $diff) as $line) { if (strncmp($line, '+++ b/', 6) === 0) { $file = substr($line, 6); + + if (strncmp($file, 'tests/', 6) === 0) { + $file = null; + + continue; + } + $changed[$file] = []; } elseif (strncmp($line, '@@', 2) === 0 && $file !== null) { if (preg_match('/\+(\d+)(?:,(\d+))?/', $line, $match)) { @@ -102,9 +109,10 @@ function changed_lines($base_ref) { exit(2); } -$covered = 0; -$total = 0; -$missing = []; +$covered = 0; +$total = 0; +$missing = []; +$measured = []; foreach ($clover->xpath('//file') as $file) { $path = (string) $file['name']; @@ -122,6 +130,8 @@ function changed_lines($base_ref) { continue; } + $measured[$relative] = true; + foreach ($file->line as $line) { $number = (int) $line['num']; @@ -140,6 +150,34 @@ function changed_lines($base_ref) { } } +/* + * These entry points require a complete running Cacti and cannot safely be + * loaded into the isolated unit process. Keep the exception explicit: any + * newly changed production PHP file must either appear in Clover or be added + * here with reviewable justification. + */ +$unmeasured_allowlist = [ + // Database migration/schema declarations require a live Cacti database. + 'includes/database.php', + // Authenticated web entry point; its status mapping lives in the covered + // thold_notification_queue_status_cells() helper. + 'notify_queue.php', + 'thold_notify.php', +]; +$unmeasured = array_values(array_diff(array_keys($changed), array_keys($measured))); +$unexpected_unmeasured = array_values(array_diff($unmeasured, $unmeasured_allowlist)); + +if ($unmeasured !== []) { + print "Changed production PHP files absent from Clover:\n " . implode("\n ", $unmeasured) . "\n"; +} + +if ($unexpected_unmeasured !== []) { + print "FAIL: changed production PHP files are not measured or allowlisted:\n " + . implode("\n ", $unexpected_unmeasured) . "\n"; + + exit(1); +} + if ($total === 0) { print "Patch coverage: no measured lines changed.\n"; diff --git a/tests/bootstrap-unit.php b/tests/bootstrap-unit.php index 6c4a8e4f..1a0214f2 100644 --- a/tests/bootstrap-unit.php +++ b/tests/bootstrap-unit.php @@ -44,7 +44,7 @@ throw new RuntimeException("Expected Cacti version file is not readable: $expected"); } -$cacti_version = trim((string) file_get_contents($version)); +$cacti_version = trim((string) file_get_contents($version)); $expected_version = trim((string) file_get_contents($expected)); if ($cacti_version === '') { @@ -75,6 +75,7 @@ 'cacti_version' => $cacti_version, 'cacti_server_os' => 'unix', ]; +$GLOBALS['__test_original_base_path'] = $cacti_root; // thold_expand_string() include_once()s library_path/variables.php at call time. $GLOBALS['config']['library_path'] = __DIR__ . '/fixtures/cacti-lib'; @@ -104,6 +105,30 @@ function db_execute_prepared($sql, $params = [], $log = true, $db_conn = false) } } +if (!function_exists('register_process_start')) { + function register_process_start($tasktype, $taskname, $taskid = 0, $timeout = 300) { + CactiStubs::record('register_process_start', '', [$tasktype, $taskname, $taskid, $timeout]); + + return CactiStubs::nextReturn('register_process_start', true); + } +} + +if (!function_exists('unregister_process')) { + function unregister_process($tasktype, $taskname, $taskid = 0, $pid = -1) { + CactiStubs::record('unregister_process', '', [$tasktype, $taskname, $taskid, $pid]); + + return CactiStubs::nextReturn('unregister_process', true); + } +} + +if (!function_exists('heartbeat_process')) { + function heartbeat_process($tasktype, $taskname, $taskid = 0) { + CactiStubs::record('heartbeat_process', '', [$tasktype, $taskname, $taskid]); + + return CactiStubs::nextReturn('heartbeat_process', true); + } +} + if (!function_exists('db_fetch_assoc')) { function db_fetch_assoc($sql, $log = true, $db_conn = false) { CactiStubs::record('db_fetch_assoc', $sql); diff --git a/tests/fixtures/cacti-root/plugins/maint/functions.php b/tests/fixtures/cacti-root/plugins/maint/functions.php new file mode 100644 index 00000000..417fe871 --- /dev/null +++ b/tests/fixtures/cacti-root/plugins/maint/functions.php @@ -0,0 +1,2 @@ + 0) { - $sql_where = ' AND process_id = ' . $pid; + if ($last_error === null && function_exists('posix_get_last_error')) { + $last_error = 'posix_get_last_error'; + } + + if (!is_callable($last_error)) { + return null; + } + + $error = $last_error(); + + if ($error === 1) { + return true; + } + + if ($error === 3) { + return false; + } + + return null; +} + +/** + * Verify that a live PID was already running when its process row was created. + * + * @param int $pid + * @param int $registered_at + * @param int $current_timestamp + * @param callable|null $elapsed Optional process-age probe used by tests. + * + * @return bool|null True for the registered process, false for a reused PID, + * or null when process identity cannot be verified. + */ +function thold_notification_process_matches($pid, $registered_at, $current_timestamp, $elapsed = null) { + $pid = (int) $pid; + $registered_at = (int) $registered_at; + $current_timestamp = (int) $current_timestamp; + + if ($pid <= 0 || $registered_at <= 0 || $current_timestamp < $registered_at) { + return null; + } + + $process_age = false; + + if (is_callable($elapsed)) { + try { + $process_age = $elapsed($pid); + } catch (Throwable $error) { + return null; + } } else { - $sql_where = ''; + $process_age = thold_notification_probe_elapsed($pid); + } + + if (!is_int($process_age) || $process_age < 0) { + return null; + } + + $row_age = $current_timestamp - $registered_at; + + // The worker necessarily predates its database row. A younger process proves + // that the operating system reused the recorded PID. + return $process_age + 5 >= $row_age; +} + +/** + * Read process age using procps seconds with a portable ps fallback. + * + * @param int $pid + * @param mixed $runner Optional command runner used by tests. + * + * @return int|false + */ +function thold_notification_probe_elapsed($pid, $runner = null) { + $pid = (int) $pid; + + if ($runner === null && function_exists('exec')) { + $runner = static function ($field, $process_id) { + $output = []; + $status = 1; + exec('LC_ALL=C ps -o ' . $field . '= -p ' . (int) $process_id, $output, $status); + + return [$status, $output]; + }; + } + + if (!is_callable($runner)) { + return false; + } + + foreach (['etimes', 'etime'] as $field) { + try { + [$status, $output] = $runner($field, $pid); + } catch (Throwable $error) { + return false; + } + + if ((int) $status === 0 && isset($output[0])) { + $seconds = thold_notification_elapsed_seconds($output[0]); + + if ($seconds !== false) { + return $seconds; + } + } + } + + return false; +} + +/** + * Parse procps seconds or the portable [[days-]hours:]minutes:seconds format. + * + * @param mixed $elapsed + * + * @return int|false + */ +function thold_notification_elapsed_seconds($elapsed) { + $elapsed = trim((string) $elapsed); + + if (ctype_digit($elapsed)) { + return (int) $elapsed; + } + + if (!preg_match('/^(?:(\d+)-)?(?:(\d+):)?(\d+):(\d+)$/D', $elapsed, $parts)) { + return false; + } + + $days = isset($parts[1]) && $parts[1] !== '' ? (int) $parts[1] : 0; + $hours = isset($parts[2]) && $parts[2] !== '' ? (int) $parts[2] : 0; + $minutes = (int) $parts[3]; + $seconds = (int) $parts[4]; + + return $days * 86400 + $hours * 3600 + $minutes * 60 + $seconds; +} + +/** + * A process row is stale only after its latest heartbeat exceeds the timeout. + * + * @param array $process + * @param int $timeout + * + * @return bool + */ +function thold_notification_process_is_stale(array $process, $timeout) { + $heartbeat = (int) ($process['heartbeat_at'] ?? ($process['started_at'] ?? 0)); + $now = (int) ($process['current_timestamp'] ?? time()); + $timeout = max(1, (int) $timeout); + + return $heartbeat > 0 && $now >= $heartbeat && ($now - $heartbeat) >= $timeout; +} + +/** + * Register a worker only after acquiring its database-backed lease. + * + * @param int $thread + * @param int $timeout + * @param callable|null $lock Optional lease acquisition used by tests. + * @param callable|null $probe Optional process liveness probe used by tests. + * @param callable|null $identity Optional process-start probe used by tests. + * + * @return bool + */ +function thold_notification_register_process($thread, $timeout = 3600, $lock = null, $probe = null, $identity = null) { + global $config; + + $acquired = is_callable($lock) ? (bool) $lock($thread) : thold_notification_acquire_lock($thread); + + if (!$acquired) { + cacti_log(sprintf('WARNING: Notification thread %s already has an active worker lease.', $thread), false, 'THOLD'); + + return false; + } + + $process = db_fetch_row_prepared('SELECT pid, + UNIX_TIMESTAMP(started) AS started_at, + GREATEST(UNIX_TIMESTAMP(started), UNIX_TIMESTAMP(last_update)) AS heartbeat_at, + UNIX_TIMESTAMP() AS current_timestamp + FROM processes + WHERE tasktype = ? + AND taskname = ? + AND taskid = ?', + ['thold_notify', 'child', $thread]); + + if ($process === false) { + thold_notification_release_lock($thread); + + return false; + } + + if (!cacti_sizeof($process)) { + $registered = register_process_start('thold_notify', 'child', $thread, $timeout); + + if (!$registered) { + thold_notification_release_lock($thread); + } + + return $registered; + } + + $running_pid = (int) ($process['pid'] ?? 0); + $running = null; + + if (is_callable($probe)) { + $running = $probe($running_pid); + } elseif (($config['cacti_server_os'] ?? '') === 'unix') { + $running = thold_notification_probe_process($running_pid); + } + + // The advisory lock prevents a new overlap, but it is not evidence that + // the recorded process died: a reconnect drops GET_LOCK while PHP lives. + // Confirmed death can recover immediately. Unknown liveness waits for an + // expired heartbeat. A live PID is trusted while fresh; after timeout its + // elapsed age must prove that it predates the recorded worker row. + $stale = thold_notification_process_is_stale($process, $timeout); + $match = $running === true && $stale + ? thold_notification_process_matches( + $running_pid, + $process['started_at'] ?? 0, + $process['current_timestamp'] ?? time(), + $identity + ) + : null; + + if (($running === true && (!$stale || $match !== false)) || ($running === null && !$stale)) { + thold_notification_release_lock($thread); + cacti_log(sprintf('WARNING: Notification thread %s has an existing worker that is live or cannot be verified dead.', $thread), false, 'THOLD'); + + return false; + } + + thold_notification_release_claim($running_pid); + unregister_process('thold_notify', 'child', $thread, $running_pid); + + $registered = register_process_start('thold_notify', 'child', $thread, $timeout); + + if (!$registered) { + thold_notification_release_lock($thread); + } + + return $registered; +} + +/** + * Claim all currently unowned queue rows for one registered worker. + * + * @param int $pid + * + * @return int + */ +function thold_notification_claim($pid) { + $pid = (int) $pid; + + if ($pid <= 0) { + return 0; + } + + // SIGKILL cannot run cleanup. Probe the small set of currently owned rows + // before running the recovery UPDATE; process_id > 0 can use its index and + // avoids scanning completed queue history on every poller cycle. + $orphaned = db_fetch_cell_prepared('SELECT 1 + FROM notification_queue AS nq + LEFT JOIN processes AS p + ON p.pid = nq.process_id + AND p.tasktype = ? + AND p.taskname = ? + WHERE nq.process_id > 0 + AND nq.event_processed = 0 + AND p.pid IS NULL + LIMIT 1', + ['thold_notify', 'child']); + + if ($orphaned) { + db_execute_prepared('UPDATE notification_queue AS nq + LEFT JOIN processes AS p + ON p.pid = nq.process_id + AND p.tasktype = ? + AND p.taskname = ? + SET nq.process_id = 0 + WHERE nq.process_id > 0 + AND nq.event_processed = 0 + AND p.pid IS NULL', + ['thold_notify', 'child']); + } + + db_execute_prepared('UPDATE notification_queue + SET process_id = ? + WHERE event_processed = 0 + AND (next_attempt IS NULL OR next_attempt <= NOW()) + AND process_id = 0', + [$pid]); + + return db_affected_rows(); +} + +/** + * Release unfinished rows when a worker stops or suspends. + * + * @param int $pid + * + * @return bool + */ +function thold_notification_release_claim($pid) { + $pid = (int) $pid; + + if ($pid <= 0) { + return true; + } + + return db_execute_prepared('UPDATE notification_queue + SET process_id = 0 + WHERE process_id = ? + AND event_processed = 0', + [$pid]); +} + +/** + * Release queue and process ownership exactly once during shutdown. + * + * @param int $pid + * @param int $thread + * @param bool $registered + * + * @return bool + */ +function thold_notification_cleanup($pid, $thread, &$registered) { + if (!$registered) { + return true; + } + + $released = thold_notification_release_claim($pid); + + if (!$released) { + cacti_log(sprintf('WARNING: Failed to release unfinished notification claims for process %s.', $pid), false, 'THOLD'); + } + + unregister_process('thold_notify', 'child', $thread, $pid); + thold_notification_release_lock($thread); + $registered = false; + + return $released; +} + +/** + * Shutdown callback available before the CLI installs signal handlers. + * + * @return void + */ +function thold_notification_shutdown() { + global $notification_registered, $pid, $thread; + + thold_notification_cleanup($pid, $thread, $notification_registered); +} + +/** + * Execute a terminal queue update and prove the worker still owns every row. + * + * @param string $sql + * @param array $params + * @param array $ids + * @param int $pid + * + * @return bool + * + * @throws RuntimeException When the claim was revoked before completion. + */ +function thold_notification_complete($sql, array $params, array $ids, $pid) { + $ids = array_values(array_filter(array_map('intval', $ids))); + $expected = cacti_sizeof($ids); + $updated = db_execute_prepared($sql, $params); + + if (!$updated) { + $message = sprintf( + 'ERROR: Notification process %s could not complete queue row(s) %s because the database update failed.', + (int) $pid, + implode(', ', $ids) + ); + cacti_log($message, false, 'THOLD'); + $placeholders = implode(', ', array_fill(0, $expected, '?')); + $fallback = $expected > 0 && db_execute_prepared("UPDATE notification_queue + SET error_code = 1, error_message = 'Completion update failed', + event_processed = 1, event_processed_time = NOW() + WHERE id IN ($placeholders) + AND process_id = ?", array_merge($ids, [(int) $pid])); + + if ($fallback && db_affected_rows() >= $expected) { + return true; + } + + throw new UnexpectedValueException($message); + } + + $affected = db_affected_rows(); + + if ($expected === 0 || $affected < $expected) { + $message = sprintf( + 'ERROR: Notification process %s lost ownership before completing queue row(s) %s.', + (int) $pid, + implode(', ', $ids) + ); + cacti_log($message, false, 'THOLD'); + + throw new RuntimeException($message); + } + + return true; +} + +/** + * Bound queue error text to the notification_queue.error_message column. + * + * @param mixed $message + * + * @return string + */ +function thold_notification_error_message($message) { + $message = str_replace(["\r", "\n"], ' ', (string) $message); + + if (function_exists('mb_check_encoding') && !mb_check_encoding($message, 'UTF-8')) { + $message = mb_convert_encoding($message, 'UTF-8', 'UTF-8'); + } + + return function_exists('mb_substr') + ? mb_substr($message, 0, 128, 'UTF-8') + : substr($message, 0, 128); +} + +/** + * Refresh a worker heartbeat only while this connection owns its lease. + * + * @param int $thread + * + * @return void + * + * @throws RuntimeException When the lease was lost. + */ +function thold_notification_heartbeat($thread) { + if (!thold_notification_owns_lock($thread)) { + throw new RuntimeException('Notification worker lease was lost.'); + } + + heartbeat_process('thold_notify', 'child', $thread); +} + +/** + * Own and run one notification worker from registration through cleanup. + * + * Optional callables keep the startup ordering executable in unit tests. The + * CLI passes none of them and therefore uses the production implementations. + * + * @param int $thread + * @param int $timeout + * @param int|null $worker_pid + * @param callable|null $shutdown_registrar + * @param callable|null $register_worker + * @param callable|null $run_worker + * + * @return bool False when registration fails; true after a completed drain. + */ +function thold_notification_main($thread, $timeout = 3600, $worker_pid = null, $shutdown_registrar = null, $register_worker = null, $run_worker = null) { + global $notification_registered, $pid; + + $pid = $worker_pid === null ? getmypid() : (int) $worker_pid; + $GLOBALS['thread'] = (int) $thread; + $notification_registered = true; + $shutdown_registrar = is_callable($shutdown_registrar) ? $shutdown_registrar : 'register_shutdown_function'; + $register_worker = is_callable($register_worker) ? $register_worker : 'thold_notification_register_process'; + $run_worker = is_callable($run_worker) ? $run_worker : 'thold_notification_run'; + + // Install cleanup before ownership acquisition. A signal can then release + // a lease acquired by register_worker even before that call returns. + $shutdown_registrar('thold_notification_shutdown'); + + $start = microtime(true); + + try { + if (!$register_worker($thread, $timeout)) { + $notification_registered = false; + + return false; + } + + $total_rows = $run_worker($pid, 'all', static function () use ($thread) { + thold_notification_heartbeat($thread); + }); + } catch (Throwable $error) { + cacti_log('ERROR: Notification worker stopped: ' . $error->getMessage(), false, 'THOLD'); + + return false; + } finally { + thold_notification_shutdown(); + } + + $end = microtime(true); + cacti_log(sprintf('THOLD NOTIFY STATS: Time:%0.2f Notifications:%s', $end - $start, $total_rows), false, 'SYSTEM'); + + return true; +} + +/** + * Mark an unsupported queue topic terminal so it cannot poison every run. + * + * @param int $id + * @param int $pid + * @param string $topic + * + * @return bool + */ +function thold_notification_reject_unknown_topic($id, $pid, $topic) { + $message = thold_notification_error_message('Unsupported notification topic: ' . (string) $topic); + + return thold_notification_complete('UPDATE notification_queue + SET error_code = 1, error_message = ?, event_processed = 1, + event_processed_time = NOW() + WHERE id = ? + AND process_id = ?', + [$message, (int) $id, (int) $pid], + [(int) $id], + $pid); +} + +/** + * Claim, drain, and always release one worker's queue slice. + * + * @param int $pid + * @param int|string $max_records + * @param callable|null $heartbeat + * + * @return int + */ +function thold_notification_run($pid, $max_records = 'all', $heartbeat = null) { + if (is_callable($heartbeat)) { + $heartbeat(); + } + + $total_rows = 0; + + try { + $total_rows = thold_notification_claim($pid); + thold_notification_execute($pid, $max_records, $heartbeat); + } finally { + try { + if (is_callable($heartbeat)) { + $heartbeat(); + } + } finally { + thold_notification_release_claim($pid); + } + } + + return $total_rows; +} + +function thold_notification_execute($pid = 0, $max_records = 'all', $heartbeat = null) { + $pid = (int) $pid; + + if ($pid <= 0) { + cacti_log('ERROR: Refusing to drain an unclaimed Thold notification queue.', false, 'THOLD'); + + return; } // See if and administrator has suspended notifications $prev_suspended = read_config_option('thold_notification_suspended', true); + if (is_callable($heartbeat)) { + $heartbeat(); + } + /** * See if notification delay is active and mark the events as such, * which will potentially leave less events to process. * This process will also enable notification once the delay is over * for the devices or allow them to be sent. */ - pre_process_device_notifications($pid, $max_records); + pre_process_device_notifications(); + + if (is_callable($heartbeat)) { + $heartbeat(); + } /** * Process any non-device up/down notifications first. These * notifications are not subject to notification delay */ - process_non_device_notifications($pid, $max_records, $prev_suspended); + process_non_device_notifications($pid, $max_records, $prev_suspended, $heartbeat); + + if (is_callable($heartbeat)) { + $heartbeat(); + } /** * Last process expired notification delays or device * notifications that are not subject to notification delay * not matching any rule type. */ - process_device_notifications($pid, $max_records, $prev_suspended); + process_device_notifications($pid, $max_records, $prev_suspended, $heartbeat); +} + +function thold_notification_retry_delay($attempt) { + $attempt = max(1, (int) $attempt); + + return min(3600, 60 * (2 ** ($attempt - 1))); +} + +/** + * Values for the delivery-status columns, in the same order as their headers. + * + * @param array $notification + * + * @return array + */ +function thold_notification_queue_status_cells(array $notification) { + $processed = (int) ($notification['event_processed'] ?? 0); + + return [ + 'event_processed' => $processed === 0 ? __('Pending', 'thold') : __('Done', 'thold'), + 'error_code' => $processed === 0 ? __('N/A', 'thold') : ((int) ($notification['error_code'] ?? 0) > 0 ? __('Errored', 'thold') : __('Success', 'thold')), + 'attempt_count' => (int) ($notification['attempt_count'] ?? 0), + 'next_attempt' => !empty($notification['next_attempt']) ? $notification['next_attempt'] : __('N/A', 'thold'), + 'event_processed_runtime' => $processed === 0 ? __('N/A', 'thold') : number_format_i18n($notification['event_processed_runtime'] ?? 0, 2), + ]; +} + +/** + * Record one queued email delivery without losing transient failures. + * + * The fifth failed attempt is terminal. Earlier failures release the claim and + * schedule a bounded exponential retry, so a permanent SMTP error cannot spin + * every poller cycle forever. + * + * @param mixed $id + * @param int $pid + * @param mixed $error + * @param mixed $runtime + * @param mixed $previous_attempts + */ +function thold_notification_record_delivery($id, $pid, $error, $runtime, $previous_attempts = 0) { + return thold_notification_record_deliveries([(int) $id => $previous_attempts], $pid, $error, $runtime); +} + +/** + * Record one grouped mail result with a single prepared update. + * + * @param array $records Record ID => previous attempt count. + * @param int $pid Owning notification worker. + * @param string $error + * @param float $runtime + * + * @return bool + */ +function thold_notification_record_deliveries(array $records, $pid, $error, $runtime) { + $pid = (int) $pid; + + if ($pid <= 0 || !cacti_sizeof($records)) { + return true; + } + + $error = thold_notification_error_message($error); + $attempt_cases = []; + $attempt_params = []; + $next_cases = []; + $next_params = []; + $process_cases = []; + $process_params = []; + $done_cases = []; + $done_params = []; + $time_cases = []; + $time_params = []; + $ids = []; + + foreach ($records as $id => $previous_attempts) { + $id = (int) $id; + + if ($id <= 0) { + continue; + } + + $attempt = max(0, (int) $previous_attempts) + 1; + $retryable = $error !== '' && $attempt < 5; + $done = $retryable ? 0 : 1; + + $attempt_cases[] = 'WHEN ? THEN ?'; + $attempt_params[] = $id; + $attempt_params[] = $attempt; + + if ($retryable) { + $next_cases[] = 'WHEN ? THEN FROM_UNIXTIME(UNIX_TIMESTAMP() + ?)'; + $next_params[] = $id; + $next_params[] = thold_notification_retry_delay($attempt); + $process_cases[] = 'WHEN ? THEN 0'; + } else { + $next_cases[] = 'WHEN ? THEN NULL'; + $next_params[] = $id; + $process_cases[] = 'WHEN ? THEN process_id'; + } + + $process_params[] = $id; + $done_cases[] = 'WHEN ? THEN ?'; + $done_params[] = $id; + $done_params[] = $done; + $time_cases[] = $done ? 'WHEN ? THEN NOW()' : 'WHEN ? THEN event_processed_time'; + $time_params[] = $id; + $ids[] = $id; + } + + if (!cacti_sizeof($ids)) { + return true; + } + + $placeholders = implode(',', array_fill(0, cacti_sizeof($ids), '?')); + $params = array_merge( + [$error === '' ? 0 : 1, $error], + $attempt_params, + $next_params, + $process_params, + $done_params, + $time_params, + [$runtime], + $ids, + [$pid] + ); + + return thold_notification_complete('UPDATE notification_queue + SET error_code = ?, error_message = ?, + attempt_count = CASE id ' . implode(' ', $attempt_cases) . ' ELSE attempt_count END, + next_attempt = CASE id ' . implode(' ', $next_cases) . ' ELSE next_attempt END, + process_id = CASE id ' . implode(' ', $process_cases) . ' ELSE process_id END, + event_processed = CASE id ' . implode(' ', $done_cases) . ' ELSE event_processed END, + event_processed_time = CASE id ' . implode(' ', $time_cases) . ' ELSE event_processed_time END, + event_processed_runtime = ? + WHERE id IN (' . $placeholders . ') + AND process_id = ?', + $params, + $ids, + $pid); } -function process_device_notifications($pid, $max_records, $prev_suspended) { +function process_device_notifications($pid, $max_records, $prev_suspended, $heartbeat = null) { $one_email = read_config_option('alert_deadnotify_one_mail') == 'on' ? true : false; $emails = []; @@ -7266,25 +8045,25 @@ function process_device_notifications($pid, $max_records, $prev_suspended) { if ($max_records == 'all') { $sql_limit = ''; } else { - $sql_limit = 'LIMIT ' . $max_records; + $sql_limit = 'LIMIT ' . (int) $max_records; } - if ($pid > 0) { - $sql_where = ' AND process_id = ' . $pid; - } else { - $sql_where = ''; - } - - $records = db_fetch_assoc("SELECT * + $records = db_fetch_assoc_prepared("SELECT * FROM notification_queue WHERE event_processed = 0 + AND (next_attempt IS NULL OR next_attempt <= NOW()) AND topic IN ('thold_dhost_mail', 'thold_uhost_mail', 'thold_dhost_cmd', 'thold_uhost_cmd') - $sql_where + AND process_id = ? ORDER BY event_time ASC - $sql_limit"); + $sql_limit", + [$pid]); if ($prev_suspended == 0) { foreach ($records as $index => $r) { + if (is_callable($heartbeat)) { + $heartbeat(); + } + $nstart = microtime(true); // if notification is suspended, break from this loop @@ -7338,19 +8117,11 @@ function process_device_notifications($pid, $max_records, $prev_suspended) { if ($error != '') { cacti_log("ERROR: Sending Email Failed To:$to Subject:$subject. Error was:'$error'", true, 'THOLD'); - - $any_error = $error; - $error_code = 1; - } else { - $error_code = 0; } $nend = microtime(true); - db_execute_prepared('UPDATE notification_queue - SET error_code = ?, error_message = ?, event_processed = 1, event_processed_time=NOW(), event_processed_runtime = ? - WHERE id = ?', - [$error_code, str_replace("\n", ' ', $error), $nend - $nstart, $r['id']]); + thold_notification_record_delivery($r['id'], $pid, $error, $nend - $nstart, $r['attempt_count'] ?? 0); } else { $id = md5(json_encode([$from, $to, $cc, $bcc, $replyto])); @@ -7384,7 +8155,7 @@ function process_device_notifications($pid, $max_records, $prev_suspended) { } } - $emails[$id]['ids'][] = $r['id']; + $emails[$id]['records'][$r['id']] = $r['attempt_count'] ?? 0; } break; @@ -7407,8 +8178,8 @@ function process_device_notifications($pid, $max_records, $prev_suspended) { $output = []; $return = 0; - if (cacti_sizeof($data['environment'])) { - foreach ($data['environment'] as $e) { + if (cacti_sizeof($environment)) { + foreach ($environment as $e) { putenv($e); } } @@ -7419,19 +8190,27 @@ function process_device_notifications($pid, $max_records, $prev_suspended) { $nend = microtime(true); - db_execute_prepared('UPDATE notification_queue + thold_notification_complete('UPDATE notification_queue SET error_code = ?, error_message = ?, event_processed = 1, event_processed_time=NOW(), event_processed_runtime = ? - WHERE id = ?', - [$return, implode("\n", $output), $nend - $nstart, $r['id']]); + WHERE id = ? + AND process_id = ?', + [$return, thold_notification_error_message(implode("\n", $output)), $nend - $nstart, $r['id'], $pid], + [$r['id']], + $pid); break; default: cacti_log(sprintf('ERROR: Unable to process Thold Notification of topic %s', $topic), false, 'THOLD'); + thold_notification_reject_unknown_topic($r['id'] ?? 0, $pid, $topic); } } if (cacti_sizeof($emails) && $one_email) { foreach ($emails as $email) { + if (is_callable($heartbeat)) { + $heartbeat(); + } + $attachments = []; $from = $email['from']; $to = $email['to']; @@ -7461,21 +8240,11 @@ function process_device_notifications($pid, $max_records, $prev_suspended) { if ($error != '') { cacti_log("ERROR: Sending Email Failed To:$to Subject:$subject. Error was:'$error'", true, 'THOLD'); - - $any_error = $error; - $error_code = 1; - } else { - $error_code = 0; } $nend = microtime(true); - $ids = implode(', ', $email['ids']); - - db_execute_prepared("UPDATE notification_queue - SET error_code = ?, error_message = ?, event_processed = 1, event_processed_time=NOW(), event_processed_runtime = ? - WHERE id IN ($ids)", - [$error_code, str_replace("\n", ' ', $error), $nend - $nstart]); + thold_notification_record_deliveries($email['records'], $pid, $error, $nend - $nstart); } } } else { @@ -7483,29 +8252,29 @@ function process_device_notifications($pid, $max_records, $prev_suspended) { } } -function process_non_device_notifications($pid, $max_records, $prev_suspended) { +function process_non_device_notifications($pid, $max_records, $prev_suspended, $heartbeat = null) { if ($max_records == 'all') { $sql_limit = ''; } else { - $sql_limit = 'LIMIT ' . $max_records; - } - - if ($pid > 0) { - $sql_where = ' AND process_id = ' . $pid; - } else { - $sql_where = ''; + $sql_limit = 'LIMIT ' . (int) $max_records; } - $records = db_fetch_assoc("SELECT * + $records = db_fetch_assoc_prepared("SELECT * FROM notification_queue WHERE event_processed = 0 + AND (next_attempt IS NULL OR next_attempt <= NOW()) AND topic NOT IN ('thold_dhost_mail', 'thold_uhost_mail', 'thold_dhost_cmd', 'thold_uhost_cmd') - $sql_where + AND process_id = ? ORDER BY event_time ASC - $sql_limit"); + $sql_limit", + [$pid]); if ($prev_suspended == 0) { foreach ($records as $r) { + if (is_callable($heartbeat)) { + $heartbeat(); + } + $nstart = microtime(true); // if notification is suspended, break from this loop @@ -7548,19 +8317,11 @@ function process_non_device_notifications($pid, $max_records, $prev_suspended) { if ($error != '') { cacti_log("ERROR: Sending Email Failed To:$to Subject:$subject. Error was:'$error'", true, 'THOLD'); - - $any_error = $error; - $error_code = 1; - } else { - $error_code = 0; } $nend = microtime(true); - db_execute_prepared('UPDATE notification_queue - SET error_code = ?, error_message = ?, event_processed = 1, event_processed_time=NOW(), event_processed_runtime = ? - WHERE id = ?', - [$error_code, str_replace("\n", ' ', $error), $nend - $nstart, $r['id']]); + thold_notification_record_delivery($r['id'], $pid, $error, $nend - $nstart, $r['attempt_count'] ?? 0); break; case 'thold_cmd': @@ -7581,8 +8342,8 @@ function process_non_device_notifications($pid, $max_records, $prev_suspended) { $output = []; $return = 0; - if (cacti_sizeof($data['environment'])) { - foreach ($data['environment'] as $e) { + if (cacti_sizeof($environment)) { + foreach ($environment as $e) { putenv($e); } } @@ -7593,14 +8354,18 @@ function process_non_device_notifications($pid, $max_records, $prev_suspended) { $nend = microtime(true); - db_execute_prepared('UPDATE notification_queue + thold_notification_complete('UPDATE notification_queue SET error_code = ?, error_message = ?, event_processed = 1, event_processed_time=NOW(), event_processed_runtime = ? - WHERE id = ?', - [$return, implode("\n", $output), $nend - $nstart, $r['id']]); + WHERE id = ? + AND process_id = ?', + [$return, thold_notification_error_message(implode("\n", $output)), $nend - $nstart, $r['id'], $pid], + [$r['id']], + $pid); break; default: cacti_log(sprintf('ERROR: Unable to process Thold Notification of topic %s', $topic), false, 'THOLD'); + thold_notification_reject_unknown_topic($r['id'] ?? 0, $pid, $topic); } } } else { diff --git a/thold_notify.php b/thold_notify.php index 90b43951..3834b234 100644 --- a/thold_notify.php +++ b/thold_notify.php @@ -22,6 +22,10 @@ +-------------------------------------------------------------------------+ */ +$notification_registered = false; +$pid = 0; +$thread = 1; + if (function_exists('pcntl_async_signals')) { pcntl_async_signals(true); } else { @@ -51,7 +55,6 @@ $parms = $_SERVER['argv']; array_shift($parms); -$thread = false; $debug = false; global $thread; @@ -72,14 +75,14 @@ break; case '--thread': - $thread = $value; - - if (!is_numeric($thread) || $thread <= 0) { + if (!ctype_digit((string) $value) || (int) $value <= 0) { print 'FATAL: The Thread ID must be numeric and greater than 0.' . PHP_EOL; display_help(); exit(1); } + cacti_log('WARNING: thold_notify.php --thread is deprecated and ignored; notification ownership is automatic.', false, 'THOLD'); + break; case '-v': case '--version': @@ -94,86 +97,19 @@ default: print 'ERROR: Invalid Parameter ' . $parameter . "\n\n"; display_help(); + exit(1); } } } -// Record start time for the pid's processing -$start = microtime(true); +thold_cli_debug('Thold Notification Main Collector Started'); -// This is where we can parallelize -$collector = ($thread === false); -$pid = 0; -$total_rows = 0; - -if ($collector) { - thold_cli_debug('Thold Notification Main Collector Started'); - - $thread = 1; -} else { - thold_cli_debug("Thold Notification Child Thread $thread Started"); -} - -$timeout = 9999999999; - -// kill any running services that have run outside of their timeout -if (!register_process_start('thold_notify', 'child', $thread, $timeout)) { - $running_pid = db_fetch_cell_prepared('SELECT pid - FROM processes - WHERE tasktype = "thold_notify" - AND taskname = "child" - AND taskid = ?', - [$thread]); - - if ($config['cacti_server_os'] == 'unix' && function_exists('posix_getpgid')) { - $running = posix_getpgid($running_pid); - } elseif ($config['cacti_server_os'] == 'unix' && function_exists('posix_kill')) { - $running = posix_kill($running_pid, 0); - } else { - /* - * Without a way to ask whether the recorded process is alive, assume - * it is. Carrying on regardless is what let a second instance run - * beside the first and mail the same queue twice. - */ - $running = true; - } - - if ($running) { - exit(1); - } +$timeout = 3600; - unregister_process('thold_notify', 'child', $thread); - register_process_start('thold_notify', 'child', $thread, $timeout); +if (!thold_notification_main($thread, $timeout)) { + exit(1); } -/* - * Claim the queue only once this instance is the registered one, and only the - * rows nobody else holds. Claiming before the registration above meant a - * second instance stamped its own identifier over the first instance's rows - * even in the case where it went on to exit. - */ -if ($collector) { - $pid = getmypid(); - - db_execute_prepared('UPDATE notification_queue - SET process_id = ? - WHERE event_processed = 0 - AND process_id = 0', - [$pid]); - - $total_rows = db_affected_rows(); -} - -// Drain only what was claimed. Passing nothing selected every unprocessed row, -// so two overlapping runs both mailed the same notifications. -thold_notification_execute($pid); - -$end = microtime(true); - -cacti_log(sprintf('THOLD NOTIFY STATS: Time:%0.2f Notifications:%s', $end - $start, $total_rows), false, 'SYSTEM'); - -unregister_process('thold_notify', 'child', $thread); - exit(0); /** @@ -190,7 +126,7 @@ function sig_handler($signo) { case SIGTERM: case SIGINT: thold_cacti_log('WARNING: Thold Daemon Notification Child Process with PID[' . getmypid() . '] terminated by user', $thread); - unregister_process('thold_notify', 'child', $thread); + thold_notification_shutdown(); exit;