conmonrs/oom_watcher/
inotify.rs

1//! A single inotify instance which watches the memory events files of all
2//! containers.
3
4use crate::{
5    async_fd::{self, Fd},
6    mutex::lock,
7};
8use anyhow::{Context, Result};
9use nix::{
10    errno::Errno,
11    sys::inotify::{AddWatchFlags, InitFlags, Inotify, InotifyEvent, WatchDescriptor},
12};
13use std::{
14    collections::HashMap,
15    io,
16    path::Path,
17    sync::{
18        Arc, Mutex, Weak,
19        atomic::{AtomicBool, AtomicU64, Ordering},
20    },
21};
22use tokio::{
23    io::{ErrorKind, Interest, unix::AsyncFd},
24    runtime::Handle,
25    sync::mpsc::Sender,
26    task::{self, JoinHandle},
27    time::{self, Duration},
28};
29use tracing::{error, trace};
30
31/// A watch event as delivered to subscribers.
32#[derive(Clone, Debug, PartialEq, Eq)]
33pub(super) struct WatchEvent {
34    /// The file which changed. None if the kernel dropped events, every
35    /// watched file may have changed then.
36    pub(super) path: Option<Arc<Path>>,
37}
38
39impl WatchEvent {
40    /// Events got lost, by the kernel or a full channel.
41    pub(super) const LOST: Self = Self { path: None };
42}
43
44/// What a watch reports. cgroupfs notifies a change of the memory events
45/// with `IN_MODIFY` and the removal of a cgroup with `IN_IGNORED` only,
46/// which every watch gets anyway.
47const WATCH_FLAGS: AddWatchFlags = AddWatchFlags::IN_MODIFY;
48
49/// How often the events get dispatched at most. The kernel notifies a change
50/// of the memory events of a cgroup up to every 10 ms, and one of the
51/// hierarchical ones of every ancestor, which alternate in the queue, so
52/// that inotify does not merge them. Under memory pressure, like at
53/// `memory.high` or `memory.max`, a single container would wake the server
54/// up to a hundred times a second otherwise, and every one of its watchers
55/// would read its counters as often. The queue of the kernel holds 16384
56/// events by default, which takes more than 1600 watched files changing in
57/// one interval, far more than the containers of a pod. An overflow gets
58/// dispatched to every watcher, see [`Watches::dispatch`].
59const MIN_DISPATCH_INTERVAL: Duration = Duration::from_millis(100);
60
61/// The watches by descriptor. A watch belongs to a file rather than to its
62/// path: a file which got recreated gets a watch of its own, and the events
63/// of the old one, like its removal, must not reach the subscribers of the
64/// new one.
65#[derive(Default)]
66struct Watches(HashMap<WatchDescriptor, Watch>);
67
68/// A watch of a file and its subscribers.
69struct Watch {
70    path: Arc<Path>,
71    subscribers: Vec<(u64, Sender<WatchEvent>)>,
72}
73
74impl Watches {
75    /// Hand an event of the kernel to the subscribers of its watch.
76    fn dispatch(&mut self, event: &InotifyEvent) {
77        if event.mask.contains(AddWatchFlags::IN_Q_OVERFLOW) {
78            self.lost();
79            return;
80        }
81        let Some(watch) = self.0.get(&event.wd) else {
82            trace!("Watch event with unknown descriptor: {event:?}");
83            return;
84        };
85        // A removal of a cgroup only removes the watch, which tells nothing
86        // about the file, see [`OOMWatcher::watch_cgroup_v2`].
87        if event.mask.intersects(WATCH_FLAGS) {
88            let watch_event = WatchEvent {
89                path: Some(watch.path.clone()),
90            };
91            for (_, tx) in &watch.subscribers {
92                Self::deliver(tx, &watch_event);
93            }
94        }
95        // The last event of a watch, the kernel may reuse the descriptor.
96        if event.mask.contains(AddWatchFlags::IN_IGNORED) {
97            self.0.remove(&event.wd);
98        }
99    }
100
101    /// Tell every subscriber that events got lost, which may concern every
102    /// file.
103    fn lost(&self) {
104        for (_, tx) in self.0.values().flat_map(|watch| &watch.subscribers) {
105            Self::deliver(tx, &WatchEvent::LOST);
106        }
107    }
108
109    /// Deliver an event without ever blocking the reader of the events. The
110    /// last free slot of a channel takes a lost event instead, so that a
111    /// full channel drops events without its receiver missing that it did:
112    /// a dropped event of the pod cgroup must still refresh its baseline.
113    fn deliver(tx: &Sender<WatchEvent>, event: &WatchEvent) {
114        let event = if tx.capacity() > 1 {
115            event.clone()
116        } else {
117            WatchEvent::LOST
118        };
119        if tx.try_send(event).is_err() {
120            trace!("Dropping watch event, subscriber not ready");
121        }
122    }
123}
124
125/// A single inotify instance shared by all OOM watchers.
126///
127/// Creating an inotify instance per container does not scale with the number
128/// of containers in a pod. All memory event files are watched through this
129/// instance instead, a task reads its events and dispatches them to the
130/// subscribers of their path. Adding and removing a watch does not block, the
131/// subscribers do that right away.
132pub(super) struct SharedWatcher {
133    inotify: Arc<Inotify>,
134
135    /// The subscribers and the watches. The reader of the events takes this
136    /// lock as well, it never waits while holding it.
137    watches: Arc<Mutex<Watches>>,
138
139    /// Set once the reader of the events stopped, the next subscription gets
140    /// a new watcher then.
141    failed: Arc<AtomicBool>,
142
143    /// Set once a subscriber gave up watching after the reader stopped, see
144    /// [`Subscription::give_up`].
145    given_up: AtomicBool,
146
147    /// The task which reads the events, it ends with the watcher.
148    reader: JoinHandle<()>,
149
150    /// Source of unique subscription identifiers.
151    next_id: AtomicU64,
152}
153
154/// Where the shared watcher is kept, see [`SharedWatcher::get`].
155type Slot = Mutex<Weak<SharedWatcher>>;
156
157/// The shared watcher is kept alive by its subscriptions. It gets created on
158/// the first subscription and dropped with the last one, which frees the
159/// inotify instance while no container is watched. A failed creation is
160/// retried on the next subscription instead of being cached, and so is a
161/// watcher whose reader stopped. The server has a single runtime, which reads
162/// the events.
163#[cfg(not(test))]
164static SHARED_WATCHER: Slot = Mutex::new(Weak::new());
165
166/// The unit tests run a runtime per test, each one gets a watcher of its own,
167/// which its runtime reads.
168#[cfg(test)]
169static SHARED_WATCHERS: Mutex<Vec<(tokio::runtime::Id, &'static Slot)>> = Mutex::new(Vec::new());
170
171impl SharedWatcher {
172    /// Retrieve the shared watcher, creating it if necessary.
173    pub(super) fn get() -> Result<Arc<Self>> {
174        // The reader of the events is a task of the runtime.
175        let runtime = Handle::try_current().context("get runtime")?;
176        Self::get_in(Self::slot(&runtime))
177    }
178
179    /// Retrieve the watcher kept in the slot, or create one there.
180    fn get_in(slot: &Slot) -> Result<Arc<Self>> {
181        let mut slot = lock(slot);
182        if let Some(watcher) = slot
183            .upgrade()
184            .filter(|watcher| !watcher.failed.load(Ordering::Acquire))
185        {
186            return Ok(watcher);
187        }
188        let watcher = Arc::new(Self::new().context("create shared file watcher")?);
189        *slot = Arc::downgrade(&watcher);
190        Ok(watcher)
191    }
192
193    #[cfg(not(test))]
194    fn slot(_: &Handle) -> &'static Slot {
195        &SHARED_WATCHER
196    }
197
198    #[cfg(test)]
199    fn slot(runtime: &Handle) -> &'static Slot {
200        let id = runtime.id();
201        let mut shared = lock(&SHARED_WATCHERS);
202        if let Some((_, slot)) = shared.iter().find(|(i, _)| *i == id) {
203            return slot;
204        }
205        // One per test, which lives as long as the static.
206        let slot = Box::leak(Box::default());
207        shared.push((id, slot));
208        slot
209    }
210
211    fn new() -> Result<Self> {
212        let inotify = Inotify::init(InitFlags::IN_NONBLOCK | InitFlags::IN_CLOEXEC)
213            .context("create inotify instance")?;
214        let inotify = Arc::new(inotify);
215        let registered = async_fd::register(inotify.clone(), Interest::READABLE)
216            .context("register inotify instance")?;
217        let watches = Arc::<Mutex<Watches>>::default();
218        let failed = Arc::<AtomicBool>::default();
219        let reader = task::spawn(Self::read_events(
220            registered,
221            watches.clone(),
222            failed.clone(),
223        ));
224        Ok(Self {
225            inotify,
226            watches,
227            failed,
228            given_up: AtomicBool::new(false),
229            reader,
230            next_id: AtomicU64::new(0),
231        })
232    }
233
234    /// Read the events of the kernel and dispatch them, until the watcher
235    /// got dropped or reading failed.
236    ///
237    /// After a dispatch, the events which come meanwhile get dispatched
238    /// together once [`MIN_DISPATCH_INTERVAL`] passed. The instance gets
239    /// deregistered until then: tokio registers it edge triggered, and
240    /// every new event would wake the runtime while it stays registered,
241    /// even if nothing reads it.
242    async fn read_events(
243        mut registered: AsyncFd<Fd<Arc<Inotify>>>,
244        watches: Arc<Mutex<Watches>>,
245        failed: Arc<AtomicBool>,
246    ) {
247        loop {
248            let events = match registered
249                .async_io(Interest::READABLE, |fd| Self::read_queued(fd))
250                .await
251            {
252                Ok(events) => events,
253                Err(e) if e.kind() == ErrorKind::Interrupted => continue,
254                Err(e) => {
255                    error!(
256                        "Unable to read watch events, watching again on a new instance a few times at most: {e}"
257                    );
258                    break;
259                }
260            };
261            {
262                let mut watches = lock(&watches);
263                for event in &events {
264                    watches.dispatch(event);
265                }
266            }
267
268            let inotify = Arc::clone(registered.get_ref());
269            drop(registered);
270            time::sleep(MIN_DISPATCH_INTERVAL).await;
271            // Events which are queued already get reported right away.
272            registered = match async_fd::register(inotify, Interest::READABLE) {
273                Ok(registered) => registered,
274                Err(e) => {
275                    error!(
276                        "Unable to register inotify instance, watching again on a new one a few times at most: {e}"
277                    );
278                    break;
279                }
280            };
281        }
282        Self::stopped(&watches, &failed);
283    }
284
285    /// Read all queued events, which a single read takes up to 4 KiB of.
286    /// Fails with `WouldBlock` only if there is none.
287    fn read_queued(inotify: &Inotify) -> io::Result<Vec<InotifyEvent>> {
288        let mut events = inotify.read_events()?;
289        loop {
290            match inotify.read_events() {
291                Ok(more) => events.extend(more),
292                Err(Errno::EAGAIN) => return Ok(events),
293                // Recurs with the next read, after the dispatch of these.
294                Err(e) => {
295                    trace!("Unable to read more watch events: {e}");
296                    return Ok(events);
297                }
298            }
299        }
300    }
301
302    /// The reader stopped. The next subscription gets a new watcher. The
303    /// watches get dropped with their senders, so that the channels of the
304    /// subscribers end once they took the events so far, they watch again
305    /// then.
306    fn stopped(watches: &Mutex<Watches>, failed: &AtomicBool) {
307        failed.store(true, Ordering::Release);
308        let watches = std::mem::take(&mut *lock(watches));
309        drop(watches);
310    }
311
312    /// Start watching `path` on behalf of `tx`. The watch gets removed when the
313    /// returned subscription is dropped and no other subscriber of the watch
314    /// is left.
315    pub(super) fn subscribe(
316        self: &Arc<Self>,
317        path: &Path,
318        tx: Sender<WatchEvent>,
319    ) -> Result<Subscription> {
320        let id = self.next_id.fetch_add(1, Ordering::Relaxed);
321        let mut watches = lock(&self.watches);
322
323        // inotify returns the existing descriptor for a file which is watched
324        // already, the subscriber joins its watch then.
325        let wd = self
326            .inotify
327            .add_watch(path, WATCH_FLAGS)
328            .with_context(|| format!("watch {}", path.display()))?;
329        watches
330            .0
331            .entry(wd)
332            .or_insert_with(|| Watch {
333                path: path.into(),
334                subscribers: vec![],
335            })
336            .subscribers
337            .push((id, tx));
338
339        Ok(Subscription {
340            watcher: self.clone(),
341            id,
342            wd,
343        })
344    }
345
346    fn unsubscribe(&self, id: u64, wd: WatchDescriptor) {
347        let mut watches = lock(&self.watches);
348        // The kernel removed the watch already.
349        let Some(watch) = watches.0.get_mut(&wd) else {
350            return;
351        };
352        watch.subscribers.retain(|(i, _)| *i != id);
353        if !watch.subscribers.is_empty() {
354            return;
355        }
356        let Some(watch) = watches.0.remove(&wd) else {
357            return;
358        };
359        if let Err(e) = self.inotify.rm_watch(wd) {
360            // The watch is gone already if the cgroup got removed.
361            trace!("Unable to unwatch {}: {e:#}", watch.path.display());
362        }
363    }
364}
365
366impl Drop for SharedWatcher {
367    fn drop(&mut self) {
368        self.reader.abort();
369    }
370}
371
372/// A subscription to the [`SharedWatcher`] which unsubscribes on drop.
373pub(super) struct Subscription {
374    watcher: Arc<SharedWatcher>,
375    id: u64,
376    wd: WatchDescriptor,
377}
378
379impl Subscription {
380    /// Whether the reader of the watcher stopped, which ended the watch.
381    pub(super) fn watcher_failed(&self) -> bool {
382        self.watcher.failed.load(Ordering::Acquire)
383    }
384
385    /// Give up watching after the reader of the watcher stopped. Only the
386    /// first subscriber of the watcher which does gets true, so that a
387    /// failure gets one warning instead of one per container.
388    pub(super) fn give_up(&self) -> bool {
389        !self.watcher.given_up.swap(true, Ordering::Relaxed)
390    }
391}
392
393/// Ways for the tests of the watches to act like the kernel.
394#[cfg(test)]
395impl SharedWatcher {
396    /// Stop the reader, like a failure to read the events does.
397    pub(super) fn stop(&self) {
398        Self::stopped(&self.watches, &self.failed);
399    }
400
401    /// Remove the watch of a file, which the kernel reports with IN_IGNORED
402    /// only, like for the files of a cgroup which got removed.
403    pub(super) fn remove_watch(&self, path: &Path) -> Result<()> {
404        let wd = lock(&self.watches)
405            .0
406            .iter()
407            .find(|(_, watch)| *watch.path == *path)
408            .map(|(wd, _)| *wd)
409            .context("not watched")?;
410        Ok(self.inotify.rm_watch(wd)?)
411    }
412
413    /// Whether the file is watched.
414    pub(super) fn watches(&self, path: &Path) -> bool {
415        lock(&self.watches)
416            .0
417            .values()
418            .any(|watch| *watch.path == *path)
419    }
420}
421
422impl Drop for Subscription {
423    fn drop(&mut self) {
424        self.watcher.unsubscribe(self.id, self.wd);
425    }
426}
427
428#[cfg(test)]
429mod tests;
conmonrs/oom_watcher/inotify.rs

conmonrs/oom_watcher/
inotify.rs
