feat(wip): rework impementation

Author: DavSanchez

agent-control/src/sub_agent/k8s/new_supervisor.rs

@@ -7,7 +7,7 @@ use crate::{
     sub_agent::{
         effective_agents_assembler::EffectiveAgent,
         k8s::supervisor::{NotStartedSupervisorK8s, StartedSupervisorK8s},
-        supervisor::{Supervisor, SupervisorStarter, starter::SupervisorStarterError},
+        supervisor::{ApplyError, Supervisor, SupervisorStarter, starter::SupervisorStarterError},
     },
     utils::thread_context::{ThreadCollectionStopperExt, ThreadContextStopperError},
 };
@@ -53,52 +53,64 @@ impl SupervisorStarter for NotStartedSupervisorK8s {
 }
 
 impl Supervisor for StartedSupervisorK8s {
-    type ApplyError = SupervisorStarterError;
     type StopError = ThreadContextStopperError;
 
-    fn apply(self, effective_agent: EffectiveAgent) -> Result<Self, Self::ApplyError> {
+    fn apply(self, effective_agent: EffectiveAgent) -> Result<Self, ApplyError<Self>> {
         let new_k8s_config = effective_agent
             .get_k8s_config()
-            .map_err(|e| SupervisorStarterError::ConfigError(e.to_string()))?;
+            .map_err(|e| SupervisorStarterError::ConfigError(e.to_string()));
 
-        if &self.k8s_config == new_k8s_config {
+        if let Ok(cfg) = new_k8s_config
+            && cfg == &self.k8s_config
+        {
             // No changes, return same supervisor
             return Ok(self);
         }
 
-        // Reuse started supervisor's contents
-        let Self {
-            thread_contexts,
-            agent_identity,
-            k8s_client,
-            sub_agent_internal_publisher,
-            ..
-        } = self;
-
         debug!(
-            agent_id = %agent_identity.id,
+            agent_id = %self.agent_identity.id,
             "Applying new configuration to K8s supervisor"
         );
 
-        // Attach the agent id info to the logs emitted by `thread_contexts.stop()`.
-        let span = info_span!("stopping_supervisor", agent_id = %agent_identity.id);
-        let stop_threads_result = span.in_scope(|| thread_contexts.stop());
-        if let Err(e) = stop_threads_result {
-            warn!(agent_id = %agent_identity.id, "Errors stopping supervisor threads: {e}");
-        }
-
         // A new non-started supervisor to build dynamic objects from the new config
-        let temp_starter = NotStartedSupervisorK8s::new(
-            agent_identity,
-            k8s_client.clone(),
-            new_k8s_config.clone(),
-        );
-        let resources = temp_starter.build_dynamic_objects()?;
-
-        // Apply resources directly
-        Self::apply_resources(resources.iter(), &k8s_client)?;
-
-        SupervisorStarter::start(temp_starter, sub_agent_internal_publisher)
+        let temp_starter = new_k8s_config.and_then(|new_k8s_config| {
+            Ok(NotStartedSupervisorK8s::new(
+                self.agent_identity.clone(),
+                self.k8s_client.clone(),
+                new_k8s_config.clone(),
+            ))
+        });
+
+        let resources = temp_starter
+            .clone() // try to remove, need a reference to `temp_starter`
+            .and_then(|temp_starter| temp_starter.build_dynamic_objects())
+            .and_then(|res| Self::apply_resources(res.iter(), &self.k8s_client.clone()));
+
+        let new_supervisor = temp_starter.and_then(|temp_starter| {
+            SupervisorStarter::start(temp_starter, self.sub_agent_internal_publisher.clone())
+        });
+
+        match new_supervisor {
+            Ok(supervisor) => {
+                // Stop old supervisor
+                let Self {
+                    thread_contexts,
+                    agent_identity,
+                    ..
+                } = self;
+                // Attach the agent id info to the logs emitted by `thread_contexts.stop()`.
+                let span = info_span!("stopping_supervisor", agent_id = %agent_identity.id);
+                let stop_threads_result = span.in_scope(|| thread_contexts.stop());
+                if let Err(e) = stop_threads_result {
+                    warn!(agent_id = %agent_identity.id, "Errors stopping supervisor threads: {e}");
+                }
+                Ok(supervisor)
+            }
+            Err(e) => Err(ApplyError {
+                reason: e.to_string(),
+                supervisor: self,
+            }),
+        }
     }
 
     fn stop(self) -> Result<(), Self::StopError> {

agent-control/src/sub_agent/k8s/supervisor.rs

@@ -32,6 +32,7 @@ use tracing::{debug, info, info_span, trace, warn};
 const OBJECTS_SUPERVISOR_INTERVAL_SECONDS: u64 = 30;
 const SUPERVISOR_THREAD_NAME: &str = "supervisor";
 
+#[derive(Debug, Clone)]
 pub struct NotStartedSupervisorK8s {
     pub(super) agent_identity: AgentIdentity,
     pub(super) k8s_client: Arc<SyncK8sClient>,

agent-control/src/sub_agent/supervisor.rs

@@ -191,10 +191,9 @@ pub mod tests {
     mock! {
         pub Supervisor {}
         impl Supervisor for Supervisor {
-            type ApplyError = MockError;
             type StopError = MockError;
 
-            fn apply(self, effective_agent: EffectiveAgent) -> Result<Self, <Self as Supervisor>::ApplyError>;
+            fn apply(self, effective_agent: EffectiveAgent) -> Result<Self, ApplyError<Self>>;
             fn stop(self) -> Result<(), <Self as Supervisor>::StopError>;
         }
     }

agent-control/src/sub_agent/supervisor/starter.rs

@@ -6,7 +6,7 @@ use crate::sub_agent::error::SubAgentBuilderError;
 use crate::sub_agent::supervisor::stopper::SupervisorStopper;
 use thiserror::Error;
 
-#[derive(Debug, Error)]
+#[derive(Debug, Error, Clone)]
 pub enum SupervisorStarterError {
     #[error("the kube client returned an error: {0}")]
     Generic(#[from] crate::k8s::error::K8sError),