feat(search-coordinator): Add initial scaffolding.

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "components/clp-rust-utils",
     "components/clp-tdl-package",
     "components/compression-coordinator",
-    "components/log-ingestor"
+    "components/log-ingestor",
+    "components/search-coordinator"
 ]
 resolver = "3"

components/search-coordinator/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "search-coordinator"
+version = "0.13.1-dev"
+edition = "2024"
+
+[[bin]]
+name = "search-coordinator"
+path = "src/bin/search_coordinator.rs"
+
+[dependencies]
+anyhow = "1.0.100"
+clap = { version = "4.6.4", features = ["derive"] }
+clp-rust-utils = { path = "../clp-rust-utils" }
+tokio = { version = "1.52.3", features = ["macros", "rt-multi-thread", "signal", "time"] }
+tokio-util = "0.7.18"
+tracing = "0.1.44"

components/search-coordinator/src/bin/search_coordinator.rs

@@ -0,0 +1,86 @@
+use std::path::PathBuf;
+use std::time::Duration;
+
+use clap::Parser;
+use clp_rust_utils::clp_config::package;
+use clp_rust_utils::serde::yaml;
+use search_coordinator::coordination::SearchCoordinator;
+
+/// Command-line arguments for the search coordinator.
+#[derive(Debug, Parser)]
+#[command(about = "Run the search coordinator.")]
+struct Cli {
+    /// Path to the configuration file.
+    #[arg(short, long, value_name = "PATH")]
+    config: PathBuf,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let args = Cli::parse();
+
+    let _guard = clp_rust_utils::logging::set_up_logging("search_coordinator.log");
+
+    let _: package::config::Config = yaml::from_path(args.config).inspect_err(|e| {
+        tracing::error!(error = % e, "Failed to load the configuration file.");
+    })?;
+
+    let (coordinator, cancellation_token) = SearchCoordinator::new();
+    let mut coordinator_handle = tokio::spawn(coordinator.run());
+
+    let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+        .expect("failed to listen for SIGTERM");
+
+    // `None` if a shutdown signal arrived while the coordinator is still running; `Some` if the
+    // coordinator returned on its own (an early exit, possibly on error).
+    let early_exit_result = tokio::select! {
+        _ = sigterm.recv() => {
+            tracing::info!("Received SIGTERM.");
+            None
+        }
+        result = tokio::signal::ctrl_c() => {
+            if let Err(e) = result {
+                tracing::error!(error = % e, "Failed to listen to ctrl-c.");
+            }
+            tracing::info!("Forcefully shutting down.");
+            None
+        }
+        join_result = &mut coordinator_handle => Some(join_result),
+    };
+
+    // Request a graceful stop. A no-op if the coordinator has already returned.
+    cancellation_token.cancel();
+
+    let join_result = if let Some(join_result) = early_exit_result {
+        join_result
+    } else {
+        const TERMINATION_TIMEOUT: Duration = Duration::from_secs(30);
+        if let Ok(join_result) =
+            tokio::time::timeout(TERMINATION_TIMEOUT, &mut coordinator_handle).await
+        {
+            join_result
+        } else {
+            tracing::warn!(
+                "The search coordinator did not stop within {TERMINATION_TIMEOUT:?}. Aborting."
+            );
+            coordinator_handle.abort();
+            return Ok(());
+        }
+    };
+
+    match join_result {
+        Ok(Ok(())) => {
+            tracing::info!("Search coordinator stopped.");
+            Ok(())
+        }
+        Ok(Err(e)) => {
+            tracing::error!(error = % e, "Search coordinator returned on error.");
+            Err(anyhow::anyhow!("Search coordinator returned on error."))
+        }
+        Err(err) => {
+            const ERROR_MESSAGE: &str = "Failed to join the search coordinator.";
+            tracing::error!(error = % err, ERROR_MESSAGE);
+            Err(anyhow::anyhow!(ERROR_MESSAGE))
+        }
+    }
+}

components/search-coordinator/src/coordination.rs

@@ -0,0 +1,45 @@
+//! The search coordinator.
+
+use tokio_util::sync::CancellationToken;
+
+/// Coordinator for CLP search jobs.
+pub struct SearchCoordinator {
+    _cancellation_token: CancellationToken,
+}
+
+impl SearchCoordinator {
+    /// Creates a search coordinator and a token that can be used to request its shutdown.
+    #[must_use]
+    pub fn new() -> (Self, CancellationToken) {
+        let cancellation_token = CancellationToken::new();
+        (
+            Self {
+                _cancellation_token: cancellation_token.clone(),
+            },
+            cancellation_token,
+        )
+    }
+
+    /// Runs the search coordinator.
+    ///
+    /// This is currently a no-op.
+    ///
+    /// # Errors
+    ///
+    /// This implementation never returns an error.
+    pub async fn run(self) -> anyhow::Result<()> {
+        std::future::ready(()).await;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::SearchCoordinator;
+
+    #[tokio::test]
+    async fn run_succeeds() {
+        let (coordinator, _cancellation_token) = SearchCoordinator::new();
+        assert!(coordinator.run().await.is_ok());
+    }
+}

components/search-coordinator/src/lib.rs

@@ -0,0 +1,3 @@
+//! Coordination for CLP search jobs.
+
+pub mod coordination;

tools/docker-images/clp-package/Dockerfile

@@ -48,4 +48,5 @@ COPY --link --chown=${UID} ./build/python-libs/ lib/python3/site-packages/
 COPY --link --chown=${UID} ./build/rust-targets/release/api_server bin/
 COPY --link --chown=${UID} ./build/rust-targets/release/compression-coordinator bin/
 COPY --link --chown=${UID} ./build/rust-targets/release/log-ingestor bin/
+COPY --link --chown=${UID} ./build/rust-targets/release/search-coordinator bin/
 COPY --link --chown=${UID} ./build/webui/ var/www/webui/