conmonrs/streaming_server/port_forward/
namespace.rs

1//! The thread which creates the sockets of a session in the network
2//! namespace of its pod.
3
4use super::{CONNECT_BUDGET, SOCKET_THREAD_LINGER, SOCKET_THREAD_STACK_SIZE};
5use crate::{mutex::lock, streaming_server::budget::Budget};
6use anyhow::{Context, Result, bail};
7use nix::{
8    sched::{CloneFlags, setns},
9    sys::socket::{AddressFamily, SockFlag, SockType, socket},
10};
11#[cfg(test)]
12use std::sync::atomic::{AtomicUsize, Ordering};
13use std::{
14    fs::{self, File},
15    io,
16    net::{IpAddr, Ipv4Addr, Ipv6Addr, TcpStream as StdTcpStream},
17    os::unix::fs::MetadataExt,
18    path::{Path, PathBuf},
19    sync::{Arc, Mutex, MutexGuard, mpsc as std_mpsc},
20    thread,
21};
22use tokio::{sync::oneshot, time::Duration};
23
24/// The sockets for a connection into the pod, one per address family.
25type Sockets = Vec<(IpAddr, io::Result<StdTcpStream>)>;
26
27/// The requests for sockets to the thread in the network namespace.
28type SocketRequests = std_mpsc::Sender<oneshot::Sender<Result<Sockets>>>;
29
30/// Creates the sockets of a session in its network namespace, which stay in
31/// the namespace they got created in. The runtime connects them. A thread
32/// enters the namespace and serves the connections which follow within the
33/// linger time, then it ends and leaves the namespace, at once when the
34/// session ends. Failures are not kept, every connection which fails gets its
35/// own error.
36pub(super) struct Namespace {
37    pub(super) path: PathBuf,
38    linger: Duration,
39
40    /// The requests of the thread which got started last.
41    requests: Mutex<Option<SocketRequests>>,
42
43    /// What the connections which get established hold, see
44    /// [`CONNECT_BUDGET`].
45    pub(super) connect_budget: Budget,
46
47    /// How many threads got started and ended.
48    #[cfg(test)]
49    pub(super) threads: AtomicUsize,
50    #[cfg(test)]
51    pub(super) ended: Arc<AtomicUsize>,
52}
53
54impl Namespace {
55    /// The network namespace at `path`, whose thread lingers for
56    /// [`SOCKET_THREAD_LINGER`].
57    pub(super) fn new(path: PathBuf) -> Arc<Self> {
58        Self::with_linger(path, SOCKET_THREAD_LINGER)
59    }
60
61    /// The network namespace at `path`, whose thread lingers for `linger`.
62    pub(super) fn with_linger(path: PathBuf, linger: Duration) -> Arc<Self> {
63        Arc::new(Self {
64            path,
65            linger,
66            requests: Mutex::default(),
67            connect_budget: Budget::new(CONNECT_BUDGET),
68            #[cfg(test)]
69            threads: AtomicUsize::new(0),
70            #[cfg(test)]
71            ended: Arc::default(),
72        })
73    }
74
75    /// New sockets in the namespace from its thread, which gets started if it
76    /// ended.
77    pub(super) async fn sockets(&self) -> Result<Sockets> {
78        // A thread which ended before it took the request gets replaced
79        // once.
80        for _ in 0..2 {
81            let (tx, rx) = oneshot::channel();
82            self.send(tx)?;
83            if let Ok(sockets) = rx.await {
84                return sockets;
85            }
86        }
87        bail!("socket thread ended")
88    }
89
90    /// Send a request to the thread, which gets started if it ended.
91    fn send(&self, reply: oneshot::Sender<Result<Sockets>>) -> Result<()> {
92        let running = self.lock().clone();
93        let reply = match running {
94            Some(requests) => match requests.send(reply) {
95                Ok(()) => return Ok(()),
96                Err(std_mpsc::SendError(reply)) => reply,
97            },
98            None => reply,
99        };
100        let requests = self.start()?;
101        requests
102            .send(reply)
103            .map_err(|_| anyhow::anyhow!("socket thread ended"))?;
104        *self.lock() = Some(requests);
105        Ok(())
106    }
107
108    fn lock(&self) -> MutexGuard<'_, Option<SocketRequests>> {
109        lock(&self.requests)
110    }
111
112    fn start(&self) -> Result<SocketRequests> {
113        let (tx, rx) = std_mpsc::channel();
114        let (path, linger) = (self.path.clone(), self.linger);
115        #[cfg(test)]
116        let ended = self.ended.clone();
117        thread::Builder::new()
118            .name("portforward".into())
119            .stack_size(SOCKET_THREAD_STACK_SIZE)
120            .spawn(move || {
121                Self::serve(&path, linger, &rx);
122                #[cfg(test)]
123                ended.fetch_add(1, Ordering::Relaxed);
124            })
125            .context("spawn socket thread")?;
126        #[cfg(test)]
127        self.threads.fetch_add(1, Ordering::Relaxed);
128        Ok(tx)
129    }
130
131    /// Create sockets until no request arrived for the linger time or the
132    /// session is gone. The thread enters the namespace for its first
133    /// request, a request which finds it outside of the namespace tries again
134    /// and gets its own error.
135    fn serve(
136        path: &Path,
137        linger: Duration,
138        requests: &std_mpsc::Receiver<oneshot::Sender<Result<Sockets>>>,
139    ) {
140        let mut entered = false;
141        while let Ok(reply) = requests.recv_timeout(linger) {
142            if !entered {
143                if let Err(e) = Self::enter(path) {
144                    let _ = reply.send(Err(e));
145                    continue;
146                }
147                entered = true;
148            }
149            let _ = reply.send(Ok(Self::create_sockets()));
150        }
151    }
152
153    /// Enter the network namespace with the current thread.
154    fn enter(path: &Path) -> Result<()> {
155        let netns = File::open(path)
156            .with_context(|| format!("open network namespace {}", path.display()))?;
157        // Entering the own namespace needs privileges as well, which a server
158        // for a host network pod or in tests may not have.
159        let current = fs::metadata("/proc/thread-self/ns/net").context("stat network namespace")?;
160        let target = netns.metadata().context("stat target network namespace")?;
161        if (current.dev(), current.ino()) != (target.dev(), target.ino()) {
162            setns(&netns, CloneFlags::CLONE_NEWNET)
163                .with_context(|| format!("enter network namespace {}", path.display()))?;
164        }
165        Ok(())
166    }
167
168    /// A socket of each family, for one connection. The IPv6 one is only
169    /// needed if IPv4 fails, creating it right away costs less than another
170    /// round trip into the namespace thread.
171    fn create_sockets() -> Sockets {
172        [
173            (AddressFamily::Inet, IpAddr::V4(Ipv4Addr::LOCALHOST)),
174            (AddressFamily::Inet6, IpAddr::V6(Ipv6Addr::LOCALHOST)),
175        ]
176        .into_iter()
177        .map(|(family, ip)| {
178            let socket = socket(
179                family,
180                SockType::Stream,
181                SockFlag::SOCK_NONBLOCK | SockFlag::SOCK_CLOEXEC,
182                None,
183            )
184            .map(StdTcpStream::from)
185            .map_err(io::Error::from);
186            (ip, socket)
187        })
188        .collect()
189    }
190}
conmonrs/streaming_server/port_forward/namespace.rs

conmonrs/streaming_server/port_forward/
namespace.rs
