add mechanism for passing metrics TypedSender to BackendManager

sender is acquired at the start of the drone loop, so should always be valid
while the drone<->controller connection is active.

plane/src/drone/backend_manager.rs

@@ -4,6 +4,8 @@ use super::{
 };
 use crate::{
     names::BackendName,
+    protocol::BackendMetricsMessage,
+    typed_socket::TypedSocketSender,
     types::{BackendState, BackendStatus, ExecutorConfig, PullPolicy, TerminationKind},
     util::{ExponentialBackoff, GuardHandle},
 };
@@ -14,7 +16,7 @@ use std::{error::Error, fmt::Debug};
 use std::{future::pending, pin::Pin};
 use std::{
     net::IpAddr,
-    sync::{Arc, Mutex},
+    sync::{Arc, Mutex, RwLock},
 };
 
 /// The backend manager uses a state machine internally to manage the state of the backend.
@@ -84,6 +86,7 @@ impl BackendManager {
         state: BackendState,
         docker: PlaneDocker,
         state_callback: impl Fn(&BackendState) -> Result<(), Box<dyn Error>> + Send + Sync + 'static,
+        _metrics_sender: Arc<RwLock<TypedSocketSender<BackendMetricsMessage>>>,
         ip: IpAddr,
     ) -> Arc<Self> {
         let container_id = ContainerId::from(format!("plane-{}", backend_id));

plane/src/drone/executor.rs

@@ -1,7 +1,8 @@
 use super::{backend_manager::BackendManager, docker::PlaneDocker, state_store::StateStore};
 use crate::{
     names::BackendName,
-    protocol::{BackendAction, BackendEventId, BackendStateMessage},
+    protocol::{BackendAction, BackendEventId, BackendMetricsMessage, BackendStateMessage},
+    typed_socket::TypedSocketSender,
     types::BackendState,
     util::GuardHandle,
 };
@@ -68,6 +69,13 @@ impl Executor {
             .register_listener(listener)
     }
 
+    pub fn register_metrics_sender(&self, sender: TypedSocketSender<BackendMetricsMessage>) {
+        self.state_store
+            .lock()
+            .expect("State store lock poisoned")
+            .register_metrics_sender(sender);
+    }
+
     pub fn ack_event(&self, event_id: BackendEventId) -> Result<()> {
         self.state_store
             .lock()
@@ -96,12 +104,19 @@ impl Executor {
                     }
                 };
 
+                let metrics_sender = self
+                    .state_store
+                    .lock()
+                    .expect("State store lock poisoned")
+                    .get_metrics_sender()?;
+
                 let manager = BackendManager::new(
                     backend_id.clone(),
                     executable.as_ref().clone(),
                     BackendState::default(),
                     self.docker.clone(),
                     callback,
+                    metrics_sender,
                     self.ip,
                 );
                 tracing::info!("Inserting backend {}.", backend_id);

plane/src/drone/mod.rs

@@ -44,6 +44,9 @@ pub async fn drone_loop(
         let mut socket = connection.connect_with_retry(&name).await;
         let _heartbeat_guard = HeartbeatLoop::start(socket.sender(MessageFromDrone::Heartbeat));
 
+        let metrics_sender = socket.sender(MessageFromDrone::BackendMetrics);
+        executor.register_metrics_sender(metrics_sender);
+
         key_manager
             .lock()
             .expect("Key manager lock poisoned")