From 267b74a59b1bd7df7c86b6f8029b21e8c7b91630 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Sat, 1 Aug 2026 03:51:47 -0400 Subject: [PATCH] Add scaffold --- Cargo.lock | 12 +++ Cargo.toml | 3 +- components/search-coordinator/Cargo.toml | 16 ++++ .../src/bin/search_coordinator.rs | 86 +++++++++++++++++++ .../search-coordinator/src/coordination.rs | 45 ++++++++++ components/search-coordinator/src/lib.rs | 3 + tools/docker-images/clp-package/Dockerfile | 1 + 7 files changed, 165 insertions(+), 1 deletion(-) create mode 100644 components/search-coordinator/Cargo.toml create mode 100644 components/search-coordinator/src/bin/search_coordinator.rs create mode 100644 components/search-coordinator/src/coordination.rs create mode 100644 components/search-coordinator/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 03b01ad06..0d4af4db2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3566,6 +3566,18 @@ version = "3.0.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "490dcfcbfef26be6800d11870ff2df8774fa6e86d047e3e8c8a76b25655e41ca" +[[package]] +name = "search-coordinator" +version = "0.13.1-dev" +dependencies = [ + "anyhow", + "clap", + "clp-rust-utils", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "sec1" version = "0.3.0" diff --git a/Cargo.toml b/Cargo.toml index 2d1b6414f..874776e50 100644 --- a/Cargo.toml +++ b/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" diff --git a/components/search-coordinator/Cargo.toml b/components/search-coordinator/Cargo.toml new file mode 100644 index 000000000..bd525fdc7 --- /dev/null +++ b/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" diff --git a/components/search-coordinator/src/bin/search_coordinator.rs b/components/search-coordinator/src/bin/search_coordinator.rs new file mode 100644 index 000000000..5ed01b42d --- /dev/null +++ b/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)) + } + } +} diff --git a/components/search-coordinator/src/coordination.rs b/components/search-coordinator/src/coordination.rs new file mode 100644 index 000000000..505335adc --- /dev/null +++ b/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()); + } +} diff --git a/components/search-coordinator/src/lib.rs b/components/search-coordinator/src/lib.rs new file mode 100644 index 000000000..174279a77 --- /dev/null +++ b/components/search-coordinator/src/lib.rs @@ -0,0 +1,3 @@ +//! Coordination for CLP search jobs. + +pub mod coordination; diff --git a/tools/docker-images/clp-package/Dockerfile b/tools/docker-images/clp-package/Dockerfile index 0516b95bb..a2143e0f9 100644 --- a/tools/docker-images/clp-package/Dockerfile +++ b/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/