Stop existing modules when iotedged starts (#3299)

Normally iotedged will stop all modules when it shuts down. But if it crashed, modules will continue to run. On Linux systems where iotedged is responsible for creating/binding the socket (e.g., CentOS 7.5, which uses systemd but does not support systemd socket activation), modules will be left holding stale file descriptors for the workload and management APIs and calls on these APIs will begin to fail. It is expected that modules will be resilient to these sorts of failures, but in reality sometimes they aren't. This change updates iotedged to stop any existing modules when it is starting. The stopped modules will be started again naturally once iotedged (and Edge Agent) are running again.
Azure · Jul 29, 2020 · 2d1a609 · 2d1a609
1 parent e9dc6c1
commit 2d1a609
Show file tree

Hide file tree

Showing 7 changed files with 71 additions and 0 deletions.
diff --git a/edgelet/edgelet-core/src/module.rs b/edgelet/edgelet-core/src/module.rs
@@ -501,6 +501,7 @@ pub trait ModuleRuntime: Sized {
     type SystemInfoFuture: Future<Item = SystemInfo, Error = Self::Error> + Send;
     type SystemResourcesFuture: Future<Item = SystemResources, Error = Self::Error> + Send;
     type RemoveAllFuture: Future<Item = (), Error = Self::Error> + Send;
+    type StopAllFuture: Future<Item = (), Error = Self::Error> + Send;
 
     fn create(&self, module: ModuleSpec<Self::Config>) -> Self::CreateFuture;
     fn get(&self, id: &str) -> Self::GetFuture;
@@ -515,6 +516,7 @@ pub trait ModuleRuntime: Sized {
     fn logs(&self, id: &str, options: &LogOptions) -> Self::LogsFuture;
     fn registry(&self) -> &Self::ModuleRegistry;
     fn remove_all(&self) -> Self::RemoveAllFuture;
+    fn stop_all(&self, wait_before_kill: Option<Duration>) -> Self::StopAllFuture;
 }
 
 #[derive(Clone, Copy, Debug)]

diff --git a/edgelet/edgelet-docker/src/runtime.rs b/edgelet/edgelet-docker/src/runtime.rs
@@ -315,6 +315,7 @@ impl ModuleRuntime for DockerModuleRuntime {
     type SystemResourcesFuture =
         Box<dyn Future<Item = SystemResources, Error = Self::Error> + Send>;
     type RemoveAllFuture = Box<dyn Future<Item = (), Error = Self::Error> + Send>;
+    type StopAllFuture = Box<dyn Future<Item = (), Error = Self::Error> + Send>;
 
     fn create(&self, module: ModuleSpec<Self::Config>) -> Self::CreateFuture {
         info!("Creating module {}...", module.name());
@@ -823,6 +824,26 @@ impl ModuleRuntime for DockerModuleRuntime {
             future::join_all(n).map(|_| ())
         }))
     }
+
+    fn stop_all(&self, wait_before_kill: Option<Duration>) -> Self::StopAllFuture {
+        let self_for_stop = self.clone();
+        Box::new(self.list().and_then(move |list| {
+            let n = list.into_iter().map(move |c| {
+                <DockerModuleRuntime as ModuleRuntime>::stop(
+                    &self_for_stop,
+                    c.name(),
+                    wait_before_kill,
+                )
+                .or_else(|err| {
+                    match Fail::find_root_cause(&err).downcast_ref::<ErrorKind>() {
+                        Some(ErrorKind::NotFound(_)) | Some(ErrorKind::NotModified) => Ok(()),
+                        _ => Err(err),
+                    }
+                })
+            });
+            future::join_all(n).map(|_| ())
+        }))
+    }
 }
 
 impl Authenticator for DockerModuleRuntime {
@@ -1479,6 +1500,7 @@ mod tests {
         type SystemResourcesFuture =
             Box<dyn Future<Item = SystemResources, Error = Self::Error> + Send>;
         type RemoveAllFuture = FutureResult<(), Self::Error>;
+        type StopAllFuture = FutureResult<(), Self::Error>;
 
         fn create(&self, _module: ModuleSpec<Self::Config>) -> Self::CreateFuture {
             unimplemented!()
@@ -1531,6 +1553,10 @@ mod tests {
         fn remove_all(&self) -> Self::RemoveAllFuture {
             unimplemented!()
         }
+
+        fn stop_all(&self, _wait_before_kill: Option<Duration>) -> Self::StopAllFuture {
+            unimplemented!()
+        }
     }
 
     impl Authenticator for TestModuleList {

diff --git a/edgelet/edgelet-http-mgmt/src/client/module.rs b/edgelet/edgelet-http-mgmt/src/client/module.rs
@@ -168,6 +168,7 @@ impl ModuleRuntime for ModuleClient {
     type SystemResourcesFuture =
         Box<dyn Future<Item = SystemResources, Error = Self::Error> + Send>;
     type RemoveAllFuture = Box<dyn Future<Item = (), Error = Self::Error> + Send>;
+    type StopAllFuture = Box<dyn Future<Item = (), Error = Self::Error> + Send>;
 
     fn create(&self, _module: ModuleSpec<Self::Config>) -> Self::CreateFuture {
         unimplemented!()
@@ -346,6 +347,16 @@ impl ModuleRuntime for ModuleClient {
             future::join_all(n).map(|_| ())
         }))
     }
+
+    fn stop_all(&self, _wait_before_kill: Option<Duration>) -> Self::StopAllFuture {
+        let self_for_stop = self.clone();
+        Box::new(self.list().and_then(move |list| {
+            let n = list
+                .into_iter()
+                .map(move |c| <Self as ModuleRuntime>::stop(&self_for_stop, c.name(), None));
+            future::join_all(n).map(|_| ())
+        }))
+    }
 }
 
 pub struct Logs(String, Body);

diff --git a/edgelet/edgelet-kube/src/runtime.rs b/edgelet/edgelet-kube/src/runtime.rs
@@ -163,6 +163,7 @@ where
     type SystemResourcesFuture =
         Box<dyn Future<Item = SystemResources, Error = Self::Error> + Send>;
     type RemoveAllFuture = Box<dyn Future<Item = (), Error = Self::Error> + Send>;
+    type StopAllFuture = Box<dyn Future<Item = (), Error = Self::Error> + Send>;
 
     fn create(&self, module: ModuleSpec<Self::Config>) -> Self::CreateFuture {
         Box::new(create_module(self, module))
@@ -303,6 +304,10 @@ where
     fn remove_all(&self) -> Self::RemoveAllFuture {
         Box::new(future::ok(()))
     }
+
+    fn stop_all(&self, _wait_before_kill: Option<Duration>) -> Self::StopAllFuture {
+        Box::new(future::ok(()))
+    }
 }
 
 impl<T, S> Authenticator for KubeModuleRuntime<T, S>

diff --git a/edgelet/edgelet-test-utils/src/module.rs b/edgelet/edgelet-test-utils/src/module.rs
@@ -344,6 +344,7 @@ where
     type SystemInfoFuture = FutureResult<SystemInfo, Self::Error>;
     type SystemResourcesFuture = FutureResult<SystemResources, Self::Error>;
     type RemoveAllFuture = FutureResult<(), Self::Error>;
+    type StopAllFuture = FutureResult<(), Self::Error>;
 
     fn create(&self, _module: ModuleSpec<Self::Config>) -> Self::CreateFuture {
         match self.module.as_ref().unwrap() {
@@ -449,4 +450,8 @@ where
     fn remove_all(&self) -> Self::RemoveAllFuture {
         future::ok(())
     }
+
+    fn stop_all(&self, _wait_before_kill: Option<Duration>) -> Self::StopAllFuture {
+        future::ok(())
+    }
 }
diff --git a/edgelet/iotedged/src/error.rs b/edgelet/iotedged/src/error.rs
@@ -193,6 +193,7 @@ pub enum InitializeErrorReason {
     #[cfg(windows)]
     RegisterWindowsService,
     RemoveExistingModules,
+    StopExistingModules,
     SaveSettings,
     #[cfg(windows)]
     StartWindowsService,
@@ -341,6 +342,10 @@ impl fmt::Display for InitializeErrorReason {
                 write!(f, "Could not remove existing modules")
             }
 
+            InitializeErrorReason::StopExistingModules => {
+                write!(f, "Could not stop existing modules")
+            }
+
             InitializeErrorReason::SaveSettings => write!(f, "Could not save settings file"),
 
             #[cfg(windows)]

diff --git a/edgelet/iotedged/src/lib.rs b/edgelet/iotedged/src/lib.rs
@@ -33,6 +33,7 @@ use std::fs::{DirBuilder, File, OpenOptions};
 use std::io::{Read, Write};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::time::Duration;
 
 use failure::{Context, Fail, ResultExt};
 use futures::future::{Either, IntoFuture};
@@ -328,6 +329,22 @@ where
                     crypto.clone(),
                 )?;
 
+                // Normally iotedged will stop all modules when it shuts down. But if it crashed,
+                // modules will continue to run. On Linux systems where iotedged is responsible for
+                // creating/binding the socket (e.g., CentOS 7.5, which uses systemd but does not
+                // support systemd socket activation), modules will be left holding stale file
+                // descriptors for the workload and management APIs and calls on these APIs will
+                // begin to fail. Resilient modules should be able to deal with this, but we'll
+                // restart all modules to ensure a clean start.
+                const STOP_TIME: Duration = Duration::from_secs(30);
+                info!("Stopping all modules...");
+                tokio_runtime
+                    .block_on(runtime.stop_all(Some(STOP_TIME)))
+                    .context(ErrorKind::Initialize(
+                        InitializeErrorReason::StopExistingModules
+                    ))?;
+                info!("Finished stopping modules.");
+
                 if $force_reprovision ||
                     ($provisioning_result.reconfigure() != ReprovisioningStatus::DeviceDataNotUpdated) {
                     // If this device was re-provisioned and the device key was updated it causes