19417 FIX Piggyback-hub: Fix crash when many piggyback hosts are updated at once

Handle event queue overflow occurred by rescanning for messages and delete watchees of deleted host folders to free up watch descriptors.

SUP-28282

Change-Id: I52b6844c317e791ed60f634f01343314f2d3e8db

Author: rseltmann

.werks/19417.md

@@ -0,0 +1,50 @@
+[//]: # (werk v2)
+# Piggyback Hub: Fix crash when many piggyback hosts are updated at once
+
+key        | value
+---------- | ---
+date       | 2026-04-07T16:47:36+00:00
+version    | 2.5.0b5
+class      | fix
+edition    | cre
+component  | checks
+level      | 1
+compatible | yes
+
+On sites with a large number of piggybacked hosts, the piggyback hub process
+could crash when many of them were updated at the same time.
+The hub now recovers gracefully: Any updates that were missed during the burst
+are detected and reprocessed automatically.
+
+If you see the following message in the piggyback hub logs under
+`var/log/piggyback-hub.log`, it means that the recovery path was triggered:
+```
+Too many messages for the piggyback-hub to progress at once, rescanning data. Consider raising /proc/sys/fs/inotify/max_queued_events.
+```
+In systems where a large number of piggybacked hosts are updated at the same
+time, the `fs.inotify.max_queued_events` limit should be raised to reduce how
+often the recovery path is triggered.
+The value should be at least the number of piggyback hosts on the site, ideally
+at least double that number to account for bursts.
+To do so, run:
+```
+sysctl -w fs.inotify.max_queued_events=65536
+```
+Additionally, the piggyback hub registers an inotify watch for each piggybacked
+host directory. If `fs.inotify.max_user_watches` is lower than the number of
+piggybacked hosts, the hub may fail to monitor all of them. Check the current
+value with:
+```
+sysctl fs.inotify.max_user_watches
+```
+and raise it if necessary:
+```
+sysctl -w fs.inotify.max_user_watches=65536
+```
+To make these changes persist across reboots, add the respective lines to
+`/etc/sysctl.conf` (or a file in `/etc/sysctl.d/`):
+```
+fs.inotify.max_queued_events = 65536
+fs.inotify.max_user_watches = 65536
+```
+

cmk/piggyback/backend/_inotify.py

@@ -16,6 +16,7 @@
 # mypy: disable-error-code="no-untyped-call"
 
 import enum
+import errno
 import os
 from collections.abc import Iterator, Sequence
 from ctypes import c_int, CDLL, get_errno
@@ -111,6 +112,10 @@ def iterate_parsed_events(self, data: bytes) -> Iterator[Event]:
             ].split(b"\x00", 1)[0]
             offset = offset + self._FIXED_EVENT_PART_LEN + bytes_remaining
 
+            if raw_event_type & Masks.Q_OVERFLOW:
+                yield Event(Watchee(-1, Path()), Masks(raw_event_type), Cookie(0), "")
+                continue
+
             yield Event(
                 Watchee(int(raw_watch_descriptor), self._wd_map[raw_watch_descriptor]),
                 Masks(raw_event_type),
@@ -183,7 +188,13 @@ def add_watch(self, path: Path, mask: Masks) -> Watchee:
         return Watchee(watch_descriptor, path)
 
     def rm_watch(self, watchee: Watchee) -> None:
-        self._libc.rm_watch(self._fileio.fileno(), watchee.wd)
+        try:
+            self._libc.rm_watch(self._fileio.fileno(), watchee.wd)
+        except OSError as e:
+            # EINVAL means the watch was already removed (e.g. the watched file/directory
+            # was deleted), so we can safely ignore it. Re-raise any other OS error.
+            if e.errno != errno.EINVAL:
+                raise
         self._parser.drop(watchee.wd)
 
     def read_forever(self) -> Iterator[Event]:

cmk/piggyback/backend/_storage.py

@@ -90,21 +90,42 @@ class PiggybackMessage:
 def watch_new_messages(omd_root: Path) -> Iterator[PiggybackMessage]:
     """Yields piggyback messages as they come in."""
 
+    host_folder_mask = Masks.MOVED_TO | Masks.DELETE_SELF
+
     with INotify() as inotify:
         watch_for_new_piggybacked_hosts = inotify.add_watch(payload_dir(omd_root), Masks.CREATE)
         watch_for_deleted_status_files = inotify.add_watch(
             source_status_dir(omd_root), Masks.DELETE
         )
         for folder in _get_piggybacked_host_folders(omd_root):
-            inotify.add_watch(folder, Masks.MOVED_TO)
+            inotify.add_watch(folder, host_folder_mask)
+
+        _last_processed_time: int = int(time.time())
 
         for event in inotify.read_forever():
+            if event.type & Masks.Q_OVERFLOW:
+                logger.warning(
+                    "Too many messages for the piggyback-hub to progress at once, rescanning data. "
+                    "Consider raising /proc/sys/fs/inotify/max_queued_events."
+                )
+                # check if any data was missed when the event queue overflowed
+                for source_file in _get_source_state_files(omd_root):
+                    if (
+                        mtime := _get_mtime(source_file)
+                    ) is not None and mtime >= _last_processed_time:
+                        source = HostName(source_file.name)
+                        for piggybacked_host in _get_piggybacked_hosts_for_source(omd_root, source):
+                            yield from get_messages_for(HostAddress(piggybacked_host), omd_root)
+                _last_processed_time = int(time.time())
+                continue
+
             # check if a new piggybacked host folder was created
             if event.watchee == watch_for_new_piggybacked_hosts:
                 if event.type & Masks.CREATE:
-                    inotify.add_watch(event.watchee.path / event.name, Masks.MOVED_TO)
+                    inotify.add_watch(event.watchee.path / event.name, host_folder_mask)
                     # Handle all files already in the folder (we rather have duplicates than missing files)
                     yield from get_messages_for(HostAddress(event.name), omd_root)
+                _last_processed_time = int(time.time())
                 continue
             if event.watchee == watch_for_deleted_status_files:
                 if event.type & Masks.DELETE:
@@ -119,9 +140,15 @@ def watch_new_messages(omd_root: Path) -> Iterator[PiggybackMessage]:
                             ),
                             b"",
                         )
+                _last_processed_time = int(time.time())
+                continue
+
+            if event.type & Masks.DELETE_SELF:
+                inotify.rm_watch(event.watchee)
                 continue
 
             if message := _make_message_from_event(event, omd_root):
+                _last_processed_time = int(time.time())
                 yield message
 
 

cmk/piggyback/hub/_main.py

@@ -101,7 +101,7 @@ def _parse_arguments(argv: list[str]) -> Arguments:
 
 
 def _setup_logging(args: Arguments) -> logging.Logger:
-    logger = getLogger(__name__)
+    logger = getLogger("cmk.piggyback")
     handler: logging.StreamHandler | WatchedFileHandler = (
         logging.StreamHandler(stream=sys.stderr)
         if args.foreground

tests/unit/cmk/piggyback/test__inotify.py

@@ -3,14 +3,13 @@
 # This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
 # conditions defined in the file COPYING, which is part of this source code package.
 
-# mypy: disable-error-code="no-untyped-call"
-
 from pathlib import Path
+from struct import pack
 from unittest.mock import ANY
 
 import pytest
 
-from cmk.piggyback.backend._inotify import Cookie, Event, INotify, Masks, Watchee
+from cmk.piggyback.backend._inotify import _EventParser, Cookie, Event, INotify, Masks, Watchee
 
 
 @pytest.mark.skip(reason="CMK-26458")
@@ -100,3 +99,14 @@ def test_basic_event_observing(tmp_path: Path) -> None:
         ]
         assert isinstance(actual[4].cookie, Cookie)
         assert actual[3].cookie == actual[4].cookie
+
+
+def test_overflow_event_does_not_crash() -> None:
+    """IN_Q_OVERFLOW must not cause a KeyError regardless of the watch descriptor value."""
+    parser = _EventParser()
+    # Raw inotify event: wd=-1, mask=IN_Q_OVERFLOW (0x4000), cookie=0, name_len=0
+    data = pack("iIII", -1, 0x4000, 0, 0)
+    (event,) = parser.iterate_parsed_events(data)
+
+    assert event.type & Masks.Q_OVERFLOW
+    assert event.watchee == Watchee(-1, Path())