conmonrs/streaming_server/
exec.rs

1//! Exec sessions, which start a process in the container for their client.
2
3use super::{
4    SessionCommon, StreamingServer,
5    session::{ClientRx, InputEvent, ResizeRx, SessionInput},
6};
7use crate::conmon_capnp::conmon::CgroupManager;
8use crate::{
9    attach::{SharedContainerAttach, StopOutput},
10    channel::Receiver as IOReceiver,
11    child_reaper::{ChildReaper, ExitChannelData, WatchedExec},
12    config::Config,
13    container_io::{ContainerIO, Drain, EarlyStdio, Message as IOMessage, Pipe, SharedContainerIO},
14    exec_start::ExecStart,
15    process::Process,
16    streaming_server::remotecommand::{MESSAGE_TOO_LARGE, Operation, Output, Status},
17    terminal::{Terminal, TerminalResizer},
18};
19use anyhow::{Context, Result, bail};
20use bytes::Bytes;
21use nix::sys::signal::Signal;
22use std::{
23    collections::VecDeque,
24    future::{self, Future},
25    sync::{
26        Arc,
27        atomic::{AtomicBool, Ordering},
28    },
29};
30use tokio::sync::broadcast::{self, error::RecvError};
31use tokio_util::sync::CancellationToken;
32use tracing::debug;
33
34/// The largest frame of the output collected while the runtime of an exec
35/// ran, like the largest reads of the pipes.
36const EARLY_FRAME_SIZE: usize = 32 * 1024;
37
38/// Required exec session data. Not `Debug`, the command may carry secrets.
39pub(crate) struct ExecSession {
40    pub(crate) child_reaper: Arc<ChildReaper>,
41
42    /// Whether the process gets a terminal. Its console socket gets created
43    /// only once the client came.
44    pub(crate) tty: bool,
45    pub(crate) server_config: Arc<Config>,
46    pub(crate) cgroup_manager: CgroupManager,
47    pub(crate) command: Vec<String>,
48    pub(crate) common: SessionCommon,
49}
50
51impl StreamingServer {
52    /// Run the runtime of an exec session and register the process it
53    /// started. A terminal gets the sizes of the client from `resize`
54    /// meanwhile.
55    pub(super) async fn start_exec(session: ExecSession, resize: ResizeRx) -> Result<StartedExec> {
56        let SessionCommon {
57            container_id,
58            stdin,
59            stdout,
60            stderr,
61            ..
62        } = session.common;
63        let mut exec = ExecStart::new(
64            &session.server_config,
65            &container_id,
66            session.tty,
67            &session.command,
68            session.cgroup_manager,
69            "exec_streaming",
70        )?;
71
72        // Read while the runtime runs, see `ContainerIO::collect_stdio`.
73        exec.collect_stdio(stdout, stderr)
74            .context("collect exec output")?;
75        let stdio = exec
76            .stdio()
77            .context("retrieve stdout and stderr channels")?;
78
79        // Clients send their size right away. The runtime connects the
80        // terminal before it starts the command, so applying the size while
81        // it runs gives the command its size from the start, usually, like
82        // with CRI-O, not only once the loop of the session runs.
83        let resizer = exec.terminal().map(Terminal::resizer);
84        // The process has no timeout, a runtime which hangs still gets
85        // killed.
86        let deadline = ChildReaper::exec_runtime_deadline(None);
87        let create = exec.run(&session.child_reaper, stdin, deadline);
88        let res = Self::resizing(create, resize, resizer).await;
89        let started = match res {
90            Ok(started) => started,
91            Err(e) => {
92                let e = exec.runtime_error(e).await;
93                return Err(e.context("create new child process"));
94            }
95        };
96
97        let WatchedExec {
98            io,
99            token,
100            process,
101            exit_rx,
102            stop_output,
103        } = exec
104            .watch(&session.child_reaper, container_id, started, None)
105            .context("watch exec process")?;
106        let teardown = ExecTeardown {
107            stdio,
108            _stop_output: stop_output,
109            attach: io.attach().clone(),
110            token,
111            process,
112            kill: Arc::default(),
113        };
114        let early_stdio = io.take_early_stdio().await;
115        Ok(StartedExec {
116            io,
117            early_stdio,
118            exit_rx,
119            stdin_enabled: stdin,
120            stdout_enabled: stdout,
121            stderr_enabled: stderr,
122            teardown,
123        })
124    }
125
126    /// The exec specific read/write loop. `status` gets the exit status once
127    /// the process exited, which goes out after the loop.
128    pub(super) async fn exec_loop(
129        started: StartedExec,
130        sender: &mut Output,
131        client: ClientRx,
132        status: &mut Option<Status>,
133    ) -> Result<()> {
134        let StartedExec {
135            io,
136            early_stdio,
137            mut exit_rx,
138            stdin_enabled,
139            stdout_enabled,
140            stderr_enabled,
141            teardown,
142        } = started;
143        let (stdout_rx, stderr_rx) = teardown.stdio.clone();
144        let mut input = SessionInput::new(io, teardown.attach.clone(), client, stdin_enabled);
145
146        // Streams which are disabled or whose read loop signalled the end are
147        // not polled anymore. A disabled one is not read at all, so let its
148        // read loop discard the output instead of waiting.
149        let mut stdout_done = !stdout_enabled;
150        let mut stderr_done = !stderr_enabled;
151        if stdout_done {
152            stdout_rx.close();
153        }
154        if stderr_done {
155            stderr_rx.close();
156        }
157
158        // What the process wrote while the runtime ran comes first, in the
159        // loop, so that a slow client does not hold up the stdin, resizing
160        // and the exit meanwhile. Output which got dropped for its size would
161        // leave a gap, so the stream ends with it, and the session fails.
162        let truncated = early_stdio.stdout.truncated || early_stdio.stderr.truncated;
163        let (mut early_stdout, mut early_stderr) = (VecDeque::new(), VecDeque::new());
164        for (early, rx, frames, done) in [
165            (
166                early_stdio.stdout,
167                &stdout_rx,
168                &mut early_stdout,
169                &mut stdout_done,
170            ),
171            (
172                early_stdio.stderr,
173                &stderr_rx,
174                &mut early_stderr,
175                &mut stderr_done,
176            ),
177        ] {
178            if *done {
179                continue;
180            }
181            *frames = Self::early_frames(early.data);
182            if early.done || early.truncated {
183                *done = true;
184                rx.close();
185            }
186        }
187
188        loop {
189            tokio::select! {
190                () = future::ready(()), if !early_stdout.is_empty() || !early_stderr.is_empty() => {
191                    for (frames, pipe) in [(&mut early_stdout, Pipe::StdOut), (&mut early_stderr, Pipe::StdErr)] {
192                        if let Some(frame) = frames.pop_front() {
193                            sender.data(pipe, &frame).await?;
194                        }
195                    }
196                },
197
198                res = stdout_rx.recv(), if !stdout_done && early_stdout.is_empty() => match res {
199                    Ok(IOMessage::Data(data, _)) => {
200                        sender.data(Pipe::StdOut, &data).await?;
201                    }
202                    Ok(IOMessage::Done) | Err(_) => stdout_done = true,
203                },
204
205                res = stderr_rx.recv(), if !stderr_done && early_stderr.is_empty() => match res {
206                    Ok(IOMessage::Data(data, _)) => {
207                        sender.data(Pipe::StdErr, &data).await?;
208                    }
209                    Ok(IOMessage::Done) | Err(_) => stderr_done = true,
210                },
211
212                event = input.next() => match event {
213                    InputEvent::Resize(data) => input.resize(&data),
214                    InputEvent::Ping => sender.ping().await?,
215                    // Close the process stdin as well and keep forwarding the
216                    // output until it exits.
217                    InputEvent::StdinClosed => input.close_container_stdin(),
218                    // The client closed the connection, everything it sent
219                    // reached the process.
220                    InputEvent::Ended => break,
221                    InputEvent::TooLarge => bail!(MESSAGE_TOO_LARGE),
222                },
223
224                res = exit_rx.recv() => {
225                    // Nothing ends the process otherwise, kill it like after
226                    // the idle timeout.
227                    if res.is_err() {
228                        teardown.kill.store(true, Ordering::Release);
229                    }
230                    *status = Some(Self::exit_status(res, truncated));
231                    // Forward what the read loops still deliver before closing.
232                    for (frames, pipe) in [(&mut early_stdout, Pipe::StdOut), (&mut early_stderr, Pipe::StdErr)] {
233                        for frame in frames.drain(..) {
234                            sender.data(pipe, &frame).await?;
235                        }
236                    }
237                    if !stdout_done {
238                        Self::drain_stdio(&stdout_rx, Pipe::StdOut, sender).await?;
239                    }
240                    if !stderr_done {
241                        Self::drain_stdio(&stderr_rx, Pipe::StdErr, sender).await?;
242                    }
243                    break
244                },
245            }
246        }
247
248        Ok(())
249    }
250
251    /// The status of the exit the watcher of the process sent. It sends it
252    /// once, and only goes away without it if its task got aborted, the
253    /// session must not wait for the exit forever then. A panic aborts the
254    /// whole server in release builds. A session whose output got
255    /// `truncated` while the runtime ran fails, the client must not take
256    /// output with a gap for the complete one, see
257    /// [`ContainerIO::collect_stdio`].
258    fn exit_status(res: Result<ExitChannelData, RecvError>, truncated: bool) -> Status {
259        match res {
260            Ok(exit_data) if truncated => Status::failed(
261                Operation::Exec,
262                format_args!(
263                    "the command wrote more than {} MiB while the runtime started it, \
264                     the rest of it got dropped (exit code {})",
265                    ContainerIO::MAX_STDIO_STREAM_SIZE / (1024 * 1024),
266                    exit_data.exit_code,
267                ),
268            ),
269            Ok(exit_data) => Status::exited(exit_data.exit_code),
270            Err(e) => Status::failed(
271                Operation::Exec,
272                format_args!("unable to get the exit of the command: {e}"),
273            ),
274        }
275    }
276
277    /// Run the runtime and give the terminal the sizes of the client
278    /// meanwhile, which it keeps until the runtime connected it, see
279    /// [`TerminalResizer::resize`]. The loop of the session takes over the
280    /// resizing once the runtime ran.
281    async fn resizing<F: Future>(
282        run: F,
283        mut resize: ResizeRx,
284        resizer: Option<TerminalResizer>,
285    ) -> F::Output {
286        let Some(resizer) = resizer else {
287            return run.await;
288        };
289        tokio::pin!(run);
290        loop {
291            tokio::select! {
292                output = &mut run => return output,
293                // A client which went away sends nothing anymore.
294                Ok(()) = resize.changed() => {
295                    if let Some(data) = resize.borrow_and_update().clone() {
296                        SessionInput::resize_with(&data, |width, height| {
297                            resizer.resize(width, height).map(|()| true)
298                        });
299                    }
300                }
301            }
302        }
303    }
304
305    /// Split the output collected while the runtime ran into frames, without
306    /// copying it.
307    fn early_frames(data: Vec<u8>) -> VecDeque<Bytes> {
308        let mut data = Bytes::from(data);
309        let mut frames = VecDeque::with_capacity(data.len().div_ceil(EARLY_FRAME_SIZE));
310        while !data.is_empty() {
311            frames.push_back(data.split_to(data.len().min(EARLY_FRAME_SIZE)));
312        }
313        frames
314    }
315
316    /// Forward the remaining output of the `pipe` of an exec after the process
317    /// exited, until the read loop signals that it is done or the [`Drain`]
318    /// gives up.
319    async fn drain_stdio(
320        rx: &IOReceiver<IOMessage>,
321        pipe: Pipe,
322        sender: &mut Output,
323    ) -> Result<()> {
324        let mut drain = Drain::default();
325        while let Some(message) = drain.recv_stdio(rx).await {
326            if let IOMessage::Data(data, _) = message {
327                sender.data(pipe, &data).await?;
328            }
329        }
330        Ok(())
331    }
332}
333
334/// An exec process which got started and registered for a session.
335pub(super) struct StartedExec {
336    io: SharedContainerIO,
337
338    /// The output written while the runtime ran.
339    early_stdio: EarlyStdio,
340    exit_rx: broadcast::Receiver<ExitChannelData>,
341    stdin_enabled: bool,
342    stdout_enabled: bool,
343    stderr_enabled: bool,
344    pub(super) teardown: ExecTeardown,
345}
346
347/// Ends an exec session however it ends, also if the client went away and
348/// the loop got aborted, or before the process even got started. The loop
349/// ends before the exit of the process if the connection is gone, if the
350/// client sent a message which is too large, or if the session hit the idle
351/// timeout or lost the exit. Only the last two kill the process, see
352/// [`ExecTeardown::kill`], the others close its stdin or hang up its
353/// terminal. All but the hang-up stop the output, which nothing takes
354/// anymore.
355pub(super) struct ExecTeardown {
356    /// The stdout and stderr channels of the process.
357    stdio: (IOReceiver<IOMessage>, IOReceiver<IOMessage>),
358    attach: SharedContainerAttach,
359
360    /// Cancelled once the process exited.
361    token: CancellationToken,
362
363    /// The process, which its watcher may reap before the token got
364    /// cancelled, so it does not get signalled by its PID.
365    process: Arc<Process>,
366
367    /// Set once the session ended with the idle timeout or lost the exit,
368    /// which kills the process, like the reaper kills one after its timeout.
369    /// Without a session the process would run on with nothing waiting for
370    /// it. Otherwise it gets the end of its stdin and its output pipes
371    /// closed, or the hang-up of its terminal, like a process started by the
372    /// client itself once the client went away.
373    pub(super) kill: Arc<AtomicBool>,
374
375    /// Stops the read loops once the session took the output, after the
376    /// exit, dropped after the rest.
377    _stop_output: StopOutput,
378}
379
380impl Drop for ExecTeardown {
381    fn drop(&mut self) {
382        // Nothing forwards the output anymore, the read loops discard it
383        // instead of waiting for this session.
384        let (stdout_rx, stderr_rx) = &self.stdio;
385        stdout_rx.close();
386        stderr_rx.close();
387
388        if self.token.is_cancelled() {
389            return;
390        }
391        if self.kill.load(Ordering::Acquire) {
392            let pid = self.process.pid();
393            debug!("Killing exec process {pid}, its session ended");
394            if let Err(e) = self.process.signal_group(Signal::SIGKILL) {
395                debug!("Not killing exec process {pid}, it is gone: {e}");
396            }
397            // A lost exit never cancels the token, which `StopOutput` waits
398            // for.
399            self.attach.stop_output();
400        } else if self.attach.close_container_stdin() {
401            // Without a client the process would wait for stdin forever.
402            // Close it, like the end of the session would for a process
403            // started by the client itself. Nothing takes the output
404            // anymore either, the read loops stop and close the pipes, so
405            // that the writes of a process like `tail -f`, which never reads
406            // stdin, fail.
407            self.attach.stop_output();
408        } else {
409            // A terminal cannot be closed without ending its output, hang it
410            // up instead, like a closed connection does. The stdin loop
411            // keeps its master open, so its output keeps getting read until
412            // the exit: a process which ignores the hang-up would block on
413            // a full terminal otherwise.
414            self.hang_up();
415        }
416    }
417}
418
419impl ExecTeardown {
420    /// Hang up the terminal like a closed connection does: signal the process
421    /// group and the process. Only while the process was not reaped yet, its
422    /// group may be gone and taken by another one otherwise, see
423    /// [`Process::signal_group`].
424    fn hang_up(&self) {
425        let pid = self.process.pid();
426        debug!("Hanging up the terminal of exec process {pid}");
427        if let Err(e) = self.process.signal_group(Signal::SIGHUP) {
428            debug!("Not hanging up exec process {pid}, it is gone: {e}");
429        }
430    }
431}
432
433#[cfg(test)]
434mod tests;
conmonrs/streaming_server/exec.rs

conmonrs/streaming_server/
exec.rs
