Retrying compartmentless containers (#1608)

* changes in internal event handling

* fixing namespace tests compilation

* rustfmt + PR suggestions

* suggestions from PR, removing unnused vector

* removing unnecessary create uuid

* reverting Cargo.lock

* suggestions from PR

Author: grnmeira

Cargo.lock

@@ -436,7 +436,7 @@ async fn handle_jemalloc_pprof_heapgen(
         .expect("builder with known status code should not fail"))
 }
 
-#[cfg(not(feature = "jemalloc"))]
+#[cfg(all(target_os = "linux", not(feature = "jemalloc")))]
 async fn handle_jemalloc_pprof_heapgen(
     _req: Request<Incoming>,
 ) -> anyhow::Result<Response<Full<Bytes>>> {

src/admin.rs

@@ -24,6 +24,8 @@ pub enum Error {
     ProtocolError(String),
     #[error("announce error: {0}")]
     AnnounceError(String),
+    #[error("namespace error: {0}")]
+    NamespaceError(String),
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize)]

src/inpod.rs

@@ -34,8 +34,7 @@ mod workloadmanager;
 #[cfg(any(test, feature = "testing"))]
 pub mod test_helpers;
 
-
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct WorkloadData {
     workload_uid: WorkloadUid,
     workload_info: Option<WorkloadInfo>,

src/inpod/windows.rs

@@ -24,6 +24,7 @@ use std::sync::RwLock;
 #[derive(Debug, Serialize, Deserialize, Clone, Copy)]
 pub enum State {
     Pending,
+    WaitingCompartment,
     Up,
 }
 
@@ -63,6 +64,7 @@ pub struct WorkloadManagerAdminHandler {
 impl WorkloadManagerAdminHandler {
     pub fn proxy_pending(
         &self,
+        workload_state: State,
         uid: &crate::inpod::WorkloadUid,
         workload_info: &WorkloadInfo,
     ) {
@@ -71,13 +73,13 @@ impl WorkloadManagerAdminHandler {
         // don't increment count here, as it is only for up and down. see comment in count.
         match state.get_mut(uid) {
             Some(key) => {
-                key.state = State::Pending;
+                key.state = workload_state;
             }
             None => {
                 state.insert(
                     uid.clone(),
                     ProxyState {
-                        state: State::Pending,
+                        state: workload_state,
                         connections: None,
                         count: 0,
                         info: workload_info.clone(),

src/inpod/windows/admin.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-
-
 use crate::inpod::windows::namespace::InpodNamespace;
 use crate::proxy::DefaultSocketFactory;
 use crate::{config, socket};
@@ -68,8 +66,8 @@ impl InPodSocketFactory {
     fn from_cfg(inner: DefaultSocketFactory, _: &InPodConfig, netns: InpodNamespace) -> Self {
         Self::new(inner, netns)
     }
-    fn new(inner: DefaultSocketFactory,netns: InpodNamespace) -> Self {
-        Self {inner, netns }
+    fn new(inner: DefaultSocketFactory, netns: InpodNamespace) -> Self {
+        Self { inner, netns }
     }
 
     fn run_in_ns<S, F: FnOnce() -> std::io::Result<S>>(&self, f: F) -> std::io::Result<S> {
@@ -96,7 +94,7 @@ impl crate::proxy::SocketFactory for InPodSocketFactory {
     }
 
     fn tcp_bind(&self, addr: std::net::SocketAddr) -> std::io::Result<socket::Listener> {
-        let std_sock = self.configure( || std::net::TcpListener::bind(addr))?;
+        let std_sock = self.configure(|| std::net::TcpListener::bind(addr))?;
         std_sock.set_nonblocking(true)?;
         tokio::net::TcpListener::from_std(std_sock).map(socket::Listener::new)
     }

src/inpod/windows/config.rs

@@ -1,24 +1,16 @@
-use std::sync::Arc;
 use tracing::warn;
 use windows::Win32::NetworkManagement::IpHelper::{
     GetCurrentThreadCompartmentId, SetCurrentThreadCompartmentId,
 };
 
 #[derive(Debug, Clone, Eq, Hash, PartialEq)]
-pub struct Namespace {
-    pub id: u32,
-    pub guid: String,
-}
-
-#[derive(Clone, Debug)]
 pub struct InpodNamespace {
-    inner: Arc<NetnsInner>,
-}
-
-#[derive(Debug, Eq, PartialEq)]
-struct NetnsInner {
-    current_namespace: u32,
-    workload_namespace: Namespace,
+    // On Windows every network namespace is based
+    // on a network compartment ID. This is the reference
+    // we need when we want to create sockets inside
+    // a network namespace or change IP stack configuration.
+    pub compartment_id: u32,
+    pub namespace_guid: String,
 }
 
 impl InpodNamespace {
@@ -32,56 +24,50 @@ impl InpodNamespace {
     }
 
     pub fn capable() -> std::io::Result<()> {
-        // set the netns to our current netns. This is intended to be a no-op,
-        // and meant to be used as a test, so we can fail early if we can't set the netns
-        let curr_namespace = Self::current()?;
-        setns(curr_namespace)
+        // Set the network compartment to the host compartment. This is intended to be a no-op,
+        // and meant to be used as a test, so we can fail early if we can't set the netns.
+        set_compartment(1)
     }
 
-    pub fn new(cur_namespace: u32, workload_namespace: String) -> std::io::Result<Self> {
+    pub fn new(workload_namespace: String) -> std::io::Result<Self> {
         let ns = hcn::get_namespace(&workload_namespace);
         match ns {
             Err(e) => {
                 warn!("Failed to get namespace: {}", e);
                 Err(std::io::Error::last_os_error())
             }
             Ok(ns) => Ok(InpodNamespace {
-                inner: Arc::new(NetnsInner {
-                    current_namespace: cur_namespace,
-                    workload_namespace: Namespace {
-                        id: ns
-                            .namespace_id
-                            .expect("There must always be a namespace id"),
-                        guid: ns.id,
-                    },
-                }),
+                compartment_id: ns
+                    .namespace_id
+                    // Compartment ID 0 means undefined compartment ID.
+                    // At the moment the JSON serialization ommits the field
+                    // if it is set to 0. This happens when the compartment
+                    // for the container is not yet available.
+                    .unwrap_or(0),
+                namespace_guid: ns.id,
             }),
         }
     }
 
-    pub fn workload_namespace(&self) -> u32 {
-        self.inner.workload_namespace.id
-    }
-
-    // Useful for logging / debugging
-    pub fn workload_namespace_guid(&self) -> String {
-        self.inner.workload_namespace.guid.clone()
-    }
-
     pub fn run<F, T>(&self, f: F) -> std::io::Result<T>
     where
         F: FnOnce() -> T,
     {
-        setns(self.inner.workload_namespace.id)?;
+        set_compartment(self.compartment_id)?;
         let ret = f();
-        setns(self.inner.current_namespace).expect("this must never fail");
+        // The Windows API defines the network compartment ID 1 as the
+        // comapartment backing up the host network namespace.
+        set_compartment(1).expect("failed to switch to host namespace");
         Ok(ret)
     }
 }
 
-// hop into a namespace
-fn setns(namespace: u32) -> std::io::Result<()> {
-    let error = unsafe { SetCurrentThreadCompartmentId(namespace) };
+// Hop into a network compartment
+fn set_compartment(compartment_id: u32) -> std::io::Result<()> {
+    if compartment_id == 0 {
+        return Err(std::io::Error::other("undefined compartment ID"));
+    }
+    let error = unsafe { SetCurrentThreadCompartmentId(compartment_id) };
     if error.0 != 0 {
         return Err(std::io::Error::from_raw_os_error(error.0 as i32));
     }
@@ -90,13 +76,13 @@ fn setns(namespace: u32) -> std::io::Result<()> {
 
 #[cfg(test)]
 mod tests {
-    use hcn::schema::HostComputeQuery;
     use hcn::api;
+    use hcn::schema::HostComputeQuery;
     use windows::core::GUID;
 
     use super::*;
 
-    fn new_namespace() -> Namespace {
+    fn new_namespace() -> InpodNamespace {
         let api_namespace = hcn::schema::HostComputeNamespace::default();
 
         let api_namespace = serde_json::to_string(&api_namespace).unwrap();
@@ -111,9 +97,9 @@ mod tests {
         let api_namespace: hcn::schema::HostComputeNamespace =
             serde_json::from_str(&api_namespace).unwrap();
 
-        Namespace {
-            id: api_namespace.namespace_id.unwrap(),
-            guid: api_namespace.id,
+        InpodNamespace {
+            compartment_id: api_namespace.namespace_id.unwrap(),
+            namespace_guid: api_namespace.id,
         }
     }
 

src/inpod/windows/namespace.rs

@@ -1,13 +1,12 @@
 use crate::drain::DrainWatcher;
-use crate::inpod::windows::{WorkloadData, WorkloadMessage, WorkloadUid};
 use crate::inpod::istio::zds::{
-    self, workload_request::Payload, Ack, Version, WorkloadRequest, WorkloadResponse, ZdsHello,
+    self, Ack, Version, WorkloadRequest, WorkloadResponse, ZdsHello, workload_request::Payload,
 };
+use crate::inpod::windows::{WorkloadData, WorkloadMessage, WorkloadUid};
 use prost::Message;
-use tracing::info;
 use std::io::{IoSlice, IoSliceMut};
 use tokio::net::windows::named_pipe::*;
-
+use tracing::info;
 
 pub struct WorkloadStreamProcessor {
     client: NamedPipeClient,