refactor: remove resource monitor initialization from ingest, query, and server handlers

Author: vkhinvasara

src/handlers/http/modal/ingest_server.rs

@@ -133,18 +133,12 @@ impl ParseableServer for IngestServer {
         let (cancel_tx, cancel_rx) = oneshot::channel();
         thread::spawn(|| sync::handler(cancel_rx));
 
-        // Start resource monitor
-        let (resource_shutdown_tx, resource_shutdown_rx) = oneshot::channel();
-        resource_check::spawn_resource_monitor(resource_shutdown_rx);
-
         tokio::spawn(airplane::server());
 
         // Ingestors shouldn't have to deal with OpenId auth flow
         let result = self.start(shutdown_rx, prometheus.clone(), None).await;
         // Cancel sync jobs
         cancel_tx.send(()).expect("Cancellation should not fail");
-        // Shutdown resource monitor
-        let _ = resource_shutdown_tx.send(());
         if let Err(join_err) = startup_sync_handle.await {
             tracing::warn!("startup sync task panicked: {join_err}");
         }

src/handlers/http/modal/mod.rs

@@ -45,7 +45,7 @@ use crate::{
     utils::get_node_id,
 };
 
-use super::{audit, cross_origin_config, health_check, API_BASE_PATH, API_VERSION};
+use super::{audit, cross_origin_config, health_check, resource_check, API_BASE_PATH, API_VERSION};
 
 pub mod ingest;
 pub mod ingest_server;
@@ -107,6 +107,10 @@ pub trait ParseableServer {
             &PARSEABLE.options.trusted_ca_certs_path,
         )?;
 
+        // Start resource monitor
+        let (resource_shutdown_tx, resource_shutdown_rx) = oneshot::channel();
+        resource_check::spawn_resource_monitor(resource_shutdown_rx);
+
         // fn that creates the app
         let create_app_fn = move || {
             App::new()
@@ -142,6 +146,9 @@ pub trait ParseableServer {
 
             health_check::shutdown().await;
 
+            // Shutdown resource monitor
+            let _ = resource_shutdown_tx.send(());
+
             // Initiate graceful shutdown
             info!("Graceful shutdown of HTTP server triggered");
             srv_handle.stop(true).await;

src/handlers/http/modal/query_server.rs

@@ -150,19 +150,13 @@ impl ParseableServer for QueryServer {
         let (cancel_tx, cancel_rx) = oneshot::channel();
         thread::spawn(|| sync::handler(cancel_rx));
 
-        // Start resource monitor
-        let (resource_shutdown_tx, resource_shutdown_rx) = oneshot::channel();
-        resource_check::spawn_resource_monitor(resource_shutdown_rx);
-
         tokio::spawn(airplane::server());
 
         let result = self
             .start(shutdown_rx, prometheus.clone(), PARSEABLE.options.openid())
             .await?;
         // Cancel sync jobs
         cancel_tx.send(()).expect("Cancellation should not fail");
-        // Shutdown resource monitor
-        let _ = resource_shutdown_tx.send(());
         if let Err(join_err) = startup_sync_handle.await {
             tracing::warn!("startup sync task panicked: {join_err}");
         }

src/handlers/http/modal/server.rs

@@ -152,10 +152,6 @@ impl ParseableServer for Server {
         let (cancel_tx, cancel_rx) = oneshot::channel();
         thread::spawn(|| sync::handler(cancel_rx));
 
-        // Start resource monitor
-        let (resource_shutdown_tx, resource_shutdown_rx) = oneshot::channel();
-        resource_check::spawn_resource_monitor(resource_shutdown_rx);
-
         if PARSEABLE.options.send_analytics {
             analytics::init_analytics_scheduler()?;
         }
@@ -168,8 +164,6 @@ impl ParseableServer for Server {
             .await;
         // Cancel sync jobs
         cancel_tx.send(()).expect("Cancellation should not fail");
-        // Shutdown resource monitor
-        let _ = resource_shutdown_tx.send(());
         if let Err(join_err) = startup_sync_handle.await {
             tracing::warn!("startup sync task panicked: {join_err}");
         }