Fix notification consistency and Syslog alert priorities for v1.6

.github/workflows/ci.yml

@@ -4,6 +4,7 @@ on:
   pull_request:
     branches:
       - main
+      - "release/**"
 
 jobs:
   validate:

README.md

@@ -250,6 +250,21 @@ syslog:
   facility: local0
 ```
 
+Syslog is an **alert and error stream**, not a per-poll health feed. It sends
+one structured, single-line event="watchdog_alert" record per emitted incident
+(including INFO recoveries and reboot summaries), plus watchdog ERROR/CRITICAL
+diagnostics. Routine INFO/WARNING polling messages stay local; per-container
+poll output is available at log_level: DEBUG. The alert records include the
+severity, UTC time, host, container, failure type, detection evidence and action,
+with special previous/new active-node fields for HA. The syslog priority is
+CRITICAL=2, HIGH=3 (error), WARNING=4 and INFO=6; the facility remains
+configurable. Values are quoted, escaped and length-limited for ingestion.
+Syslog is configured independently from alert_channels.
+Local log_level controls only console/file verbosity; INFO recoveries and
+WARNING HA alerts still reach Syslog when local log_level is WARNING or ERROR.
+HA and reboot-summary records omit pseudo-container fields.
+
+
 ### SMTP (Email)
 
 Add `smtp` to `alert_channels` to enable.
@@ -289,6 +304,10 @@ snmp_trap:
   # v3_local_engine_id: ""                 # hex engine ID pinned from first-run log; see DEPLOYMENT.md
 ```
 
+SNMP retains its five existing var-binds and OIDs for NMS compatibility.
+The human-readable cwSummary contains severity; cwProbeDetail contains
+detection evidence. No additional SNMP objects are introduced in v1.6.0.
+
 Each trap carries five var-binds, all under the Radware enterprise OID arc (`1.3.6.1.4.1.89.110`):
 
 | OID | Object | Value |

tests/test_ha_monitor.py

@@ -183,6 +183,19 @@ def test_real_watchdog_event_and_persistence(self):
                 restarted._poll_ha(restarted._session)
             restarted._dispatch.assert_not_called()
 
+    def test_ha_exporter_failure_is_error_but_throttled(self):
+        with tempfile.TemporaryDirectory() as td:
+            w = self.create_shell(f"{td}/state.json")
+            with patch("watchdog.read_ha_observation", side_effect=TimeoutError("offline")), \
+                 patch("watchdog.time.monotonic", side_effect=[1000, 1000, 1100, 1301, 1301]), \
+                 patch("watchdog.log.error") as err:
+                w._poll_ha(w._session)
+                w._poll_ha(w._session)
+                w._poll_ha(w._session)
+            self.assertEqual(err.call_count, 2)
+            self.assertIn("HA check unavailable", err.call_args.args[0])
+            w._dispatch.assert_not_called()
+
     def test_failed_exporter_does_not_overwrite_state_or_alert(self):
         with tempfile.TemporaryDirectory() as td:
             w = self.create_shell(f"{td}/state.json")
@@ -351,9 +364,12 @@ def test_slack_ha_fields_and_no_false_container(self):
 
     def test_syslog_logging_includes_transition_and_dispatches_snmp(self):
         alert = self.make_alert()
-        with patch("watchdog.log.warning") as warn, patch("watchdog.send_snmp_trap") as trap:
+        with patch("watchdog.log.log") as logged, patch("watchdog.send_snmp_trap") as trap:
             watchdog.dispatch_alert(alert, {"alert_channels": ["snmp_trap", "syslog"]})
-        self.assertIn(f"{A} -> {B}", warn.call_args.args[1])
+        self.assertEqual(logged.call_args.args[0], watchdog.logging.WARNING)
+        self.assertIn(f'previous_active="{A}"', logged.call_args.args[1])
+        self.assertIn(f'new_active="{B}"', logged.call_args.args[1])
+        self.assertTrue(logged.call_args.kwargs["extra"]["watchdog_alert"])
         trap.assert_called_once()
 
     def test_smtp_ha_notification_identifies_role_change(self):

tests/test_notifications.py

@@ -0,0 +1,192 @@
+"""Notification consistency and syslog regression tests (no network)."""
+import logging
+import socket
+import tempfile
+import unittest
+from unittest.mock import patch
+from email.parser import BytesParser
+from email.policy import default
+from pathlib import Path
+
+import watchdog
+
+
+def alert(severity="CRITICAL", failure="crashed"):
+    return watchdog.AlertPayload(
+        severity, "demo-service", "abcdef123456789", "vision", failure, 137,
+        "", "Inspect Docker events and logs.", "https://example.invalid/runbook",
+        'Crash probe failed - SIGKILL; OOM not confirmed',
+    )
+
+
+class NotificationTests(unittest.TestCase):
+    def test_oom_evidence_only_when_confirmed(self):
+        p = alert("CRITICAL", "oom")
+        p.probe_detail = ("OOM probe failed - container was OOM-killed by the kernel; "
+                          "post-event memory snapshot: usage=63.0 MB, limit=64.0 MB")
+        data = watchdog.format_syslog_alert(p)
+        self.assertIn('failure_type="oom"', data)
+        self.assertIn('post-event memory snapshot: usage=63.0 MB', data)
+        self.assertNotIn('failure_type="crashed"', data)
+
+    def test_real_syslog_keeps_recovery_and_ha_at_local_warning_or_error(self):
+        for level in ("WARNING", "ERROR"):
+            with self.subTest(level=level), tempfile.TemporaryDirectory() as td:
+                with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as receiver:
+                    receiver.bind(("127.0.0.1", 0))
+                    receiver.settimeout(0.6)
+                    watchdog.setup_logging(
+                        level, f"{td}/watchdog.log",
+                        {"enabled": True, "host": "127.0.0.1",
+                         "port": receiver.getsockname()[1], "protocol": "udp"}
+                    )
+                    try:
+                        watchdog.log.info("routine polling must stay local")
+                        recovery = alert("INFO", "recovered")
+                        recovery.exit_code = None
+                        watchdog.dispatch_alert(recovery, {"alert_channels": []})
+                        ha = watchdog.AlertPayload(
+                            "WARNING", "(HA cluster)", "", "vision", "ha-failover",
+                            None, "", "Check cluster health", "",
+                            "HA Operator promoted to new node",
+                            ha_previous_node="node-a", ha_new_node="node-b"
+                        )
+                        watchdog.dispatch_alert(ha, {"alert_channels": []})
+                        watchdog.log.error("HA check unavailable (TimeoutError): offline")
+                        received = [receiver.recvfrom(8192)[0].decode()
+                                    for _ in range(3)]
+                        with self.assertRaises(socket.timeout):
+                            receiver.recvfrom(8192)
+                        self.assertEqual(
+                            [int(msg.split(">", 1)[0][1:]) % 8 for msg in received],
+                            [6, 4, 3])
+                        self.assertIn('failure_type="recovered"', received[0])
+                        self.assertIn('failure_type="ha-failover"', received[1])
+                        self.assertIn("HA check unavailable", received[2])
+                        self.assertFalse(any("routine polling" in msg for msg in received))
+                        text = Path(f"{td}/watchdog.log").read_text()
+                        self.assertNotIn("routine polling", text)
+                        self.assertNotIn('failure_type="recovered"', text)
+                        if level == "ERROR":
+                            self.assertNotIn('failure_type="ha-failover"', text)
+                        else:
+                            self.assertIn('failure_type="ha-failover"', text)
+                    finally:
+                        logging.basicConfig(
+                            handlers=[logging.NullHandler()],
+                            level=logging.WARNING, force=True)
+
+    def test_pseudo_container_fields_absent_in_ha_and_reboot_events(self):
+        for failure, container_name in (
+            ("ha-failover", "(HA cluster)"),
+            ("reboot-summary", "(post-reboot summary)")
+        ):
+            with self.subTest(failure=failure):
+                p = watchdog.AlertPayload(
+                    "INFO", container_name, "", "vision", failure, None, "",
+                    "Check cluster", "", "Event details",
+                    ha_previous_node="node-a", ha_new_node="node-b")
+                record = watchdog.format_syslog_alert(p)
+                self.assertIn('event="watchdog_alert"', record)
+                self.assertIn(f'failure_type="{failure}"', record)
+                self.assertNotIn('container=', record)
+                self.assertNotIn('container_id=', record)
+                self.assertIn('host="vision"', record)
+                if failure == "ha-failover":
+                    self.assertIn('previous_active="node-a"', record)
+                    self.assertIn('new_active="node-b"', record)
+
+    def test_memory_snapshot_omits_missing_zero_and_unavailable_stats(self):
+        from unittest.mock import Mock
+        container = Mock()
+        container.stats.return_value = {
+            "memory_stats": {"usage": 0, "limit": 67108864, "max_usage": 12582912}
+        }
+        sample = watchdog.get_memory_stats(container)
+        self.assertIn("post-event memory snapshot:", sample)
+        self.assertIn("limit=64.0 MB", sample)
+        self.assertIn("peak=12.0 MB", sample)
+        self.assertNotIn("usage=", sample)
+        container.stats.return_value = {"memory_stats": {}}
+        self.assertEqual(watchdog.get_memory_stats(container), "")
+        container.stats.side_effect = RuntimeError("unavailable after OOM")
+        self.assertEqual(watchdog.get_memory_stats(container), "")
+
+    def test_native_levels_match_incident_severity(self):
+        for severity, expected in (("CRITICAL", logging.CRITICAL),
+                                   ("HIGH", logging.ERROR),
+                                   ("WARNING", logging.WARNING),
+                                   ("INFO", logging.INFO)):
+            with self.subTest(severity=severity):
+                p = alert(severity)
+                with patch("watchdog.log.log") as log:
+                    watchdog.dispatch_alert(p, {"alert_channels": []})
+                self.assertEqual(log.call_args.args[0], expected)
+                self.assertIn(f'severity="{severity}"', log.call_args.args[1])
+                self.assertTrue(log.call_args.kwargs["extra"]["watchdog_alert"])
+
+    def test_syslog_filter_rejects_poll_noise_and_keeps_recovery_and_errors(self):
+        f = watchdog.AlertSyslogFilter()
+        def record(level, is_alert=False):
+            r = logging.LogRecord("watchdog", level, __file__, 1, "message", (), None)
+            if is_alert:
+                r.watchdog_alert = True
+            return r
+        self.assertFalse(f.filter(record(logging.INFO)))
+        self.assertFalse(f.filter(record(logging.WARNING)))
+        self.assertTrue(f.filter(record(logging.INFO, True)))
+        self.assertTrue(f.filter(record(logging.WARNING, True)))
+        self.assertTrue(f.filter(record(logging.ERROR)))
+
+    def test_syslog_has_evidence_without_false_oom_and_escapes_content(self):
+        p = alert()
+        p.probe_detail = 'line one\nline "two"'
+        data = watchdog.format_syslog_alert(p)
+        for fragment in ('event="watchdog_alert"', 'severity="CRITICAL"',
+                         'failure_type="crashed"', 'exit_code="137"',
+                         'host="vision"', 'container="demo-service"',
+                         'container_id="abcdef123456"',
+                         'probe_type="Docker event (crash)"',
+                         'action="Inspect Docker events and logs."'):
+            self.assertIn(fragment, data)
+        self.assertIn(r'line one\nline \"two\"', data)
+        self.assertNotIn('failure_type="oom"', data)
+        self.assertNotIn("\n", data)
+
+    def test_long_untrusted_detection_is_bounded(self):
+        p = alert()
+        p.probe_detail = "x" * 20000
+        data = watchdog.format_syslog_alert(p)
+        self.assertLess(len(data), 2000)
+        self.assertIn('...', data)
+
+    def test_email_includes_detection_without_assuming_oom(self):
+        p = alert()
+        cfg = {"smtp": {"enabled": True, "auth": False, "tls": False,
+                "host": "mail.invalid", "port": 25,
+                "sender": "alerts@example.invalid",
+                "recipients": ["ops@example.invalid"]}}
+        with patch("smtplib.SMTP") as smtp:
+            watchdog.send_smtp(p, cfg)
+        sent = smtp.return_value.__enter__.return_value.sendmail.call_args.args[2]
+        body = BytesParser(policy=default).parsebytes(sent).get_body(
+            preferencelist=("plain",)).get_content()
+        for item in ("Container ID: abcdef123456", "Severity: CRITICAL",
+                     "Exit Code: 137", "Probe Type: Docker event (crash)",
+                     "OOM not confirmed", "Runbook: https://example.invalid/runbook"):
+            self.assertIn(item, body)
+
+    def test_slack_fields_match_alert_payload(self):
+        p = alert()
+        with patch.dict("os.environ", {"SLACK_WEBHOOK_URL": "https://example.invalid/slack"}):
+            with patch("watchdog.requests.post") as post:
+                watchdog.send_slack(p, {"slack": {"enabled": True}})
+        fields = post.call_args.kwargs["json"]["attachments"][0]["fields"]
+        values = {field["title"]: field["value"] for field in fields}
+        self.assertEqual(values["Severity"], "CRITICAL")
+        self.assertEqual(values["Exit Code"], "137")
+        self.assertEqual(values["Detection"], p.probe_detail)
+
+
+if __name__ == "__main__":
+    unittest.main()