From 3398399b962a0858231881ec09510776671759a1 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Mon, 20 Jul 2026 22:41:26 -0400 Subject: [PATCH 1/9] Done. --- Cargo.lock | 2 + .../clp-rust-utils/src/task_io/compression.rs | 3 +- components/compression-coordinator/Cargo.toml | 2 + .../compression-coordinator/src/error.rs | 11 +- .../compression-coordinator/src/job_handle.rs | 315 ++++++++++++++++++ components/compression-coordinator/src/lib.rs | 1 + 6 files changed, 332 insertions(+), 2 deletions(-) create mode 100644 components/compression-coordinator/src/job_handle.rs diff --git a/Cargo.lock b/Cargo.lock index 1eba6ec59..2719ea65b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -996,7 +996,9 @@ dependencies = [ "clp-rust-utils", "serde", "spider-core", + "sqlx", "thiserror", + "tracing", ] [[package]] diff --git a/components/clp-rust-utils/src/task_io/compression.rs b/components/clp-rust-utils/src/task_io/compression.rs index db6c5ca62..16890ad1f 100644 --- a/components/clp-rust-utils/src/task_io/compression.rs +++ b/components/clp-rust-utils/src/task_io/compression.rs @@ -9,8 +9,9 @@ use crate::clp_config::AwsAuthentication; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ClpSCompressionOption { pub target_encoded_size: u64, - pub compression_level: i32, + pub compression_level: u8, pub timestamp_key: Option, + pub unstructured: bool, } /// Input source for a compression task. diff --git a/components/compression-coordinator/Cargo.toml b/components/compression-coordinator/Cargo.toml index 861b59641..717649cde 100644 --- a/components/compression-coordinator/Cargo.toml +++ b/components/compression-coordinator/Cargo.toml @@ -12,4 +12,6 @@ async-trait = "0.1.89" clp-rust-utils = { path = "../clp-rust-utils" } serde = { version = "1.0.228", features = ["derive"] } spider-core = { git = "https://github.com/y-scope/spider.git", branch = "main" } +sqlx = { version = "0.8.6", features = ["runtime-tokio", "mysql"] } thiserror = "2.0.18" +tracing = "0.1.44" diff --git a/components/compression-coordinator/src/error.rs b/components/compression-coordinator/src/error.rs index 027c335c9..c860a23f5 100644 --- a/components/compression-coordinator/src/error.rs +++ b/components/compression-coordinator/src/error.rs @@ -2,4 +2,13 @@ /// Errors returned by the compression coordinator. #[derive(Debug, thiserror::Error)] -pub enum Error {} +pub enum Error { + #[error("sqlx error: {0}")] + Sqlx(#[from] sqlx::Error), + + #[error("invalid dataset: {0}")] + InvalidDataset(String), + + #[error("unsupported input config")] + UnsupportedInputConfig, +} diff --git a/components/compression-coordinator/src/job_handle.rs b/components/compression-coordinator/src/job_handle.rs new file mode 100644 index 000000000..8d47da8a4 --- /dev/null +++ b/components/compression-coordinator/src/job_handle.rs @@ -0,0 +1,315 @@ +//! Handle for driving a single S3 compression job to completion. + +use std::{sync::Arc, time::Duration}; + +use clp_rust_utils::{ + clp_config::package::config::Database, + dataset::VALID_DATASET_NAME_REGEX, + job_config::{ClpIoConfig, CompressionJobId, CompressionJobStatus, InputConfig}, + task_io::compression::{ClpSCompressionOption, S3InputSource}, +}; +use spider_core::{ + task::ExecutionPolicy, + types::id::{JobId as SpiderJobId, ResourceGroupId}, +}; +use sqlx::MySqlPool; + +use crate::{Error, compression_job_submitter::S3CompressionJobSubmitter}; + +/// Options for a compression job running in Spider. +pub struct SpiderOption { + pub compression_task_max_retry: u32, + pub commit_task_execution_policy: ExecutionPolicy, + pub initial_poll_backoff: Duration, + pub max_poll_backoff: Duration, +} + +/// Handles the asynchronous submission of an S3 compression job and the retrieval of its result. +/// +/// # Type Parameters +/// +/// * `SubmitterType` - The type of the job submitter for Spider job submission. +pub struct S3CompressionJobHandle { + _db_pool: MySqlPool, + _db_config: Database, + compression_job_id: CompressionJobId, + job_submitter: SubmitterType, + resource_group_id: ResourceGroupId, + + _input_config: InputConfig, + clp_s_compression_option: ClpSCompressionOption, + dataset: Option, + _target_archive_size: u64, + + spider_option: Arc, +} + +impl S3CompressionJobHandle { + /// Factory function. + /// + /// # Returns + /// + /// A newly created [`S3CompressionJobHandle`] for the given compression job, with the `clp-s` + /// compression options derived from `clp_io_config`'s output config. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`Error::UnsupportedInputConfig`] if `clp_io_config`'s input config is not an + /// [`InputConfig::S3ObjectMetadataInputConfig`]. + /// * [`Error::InvalidDataset`] if the configured dataset name is not a valid dataset name. + pub fn new( + db_pool: MySqlPool, + db_config: Database, + compression_job_id: CompressionJobId, + job_submitter: SubmitterType, + resource_group_id: ResourceGroupId, + clp_io_config: ClpIoConfig, + spider_option: Arc, + ) -> Result { + let input_config = clp_io_config.input; + let InputConfig::S3ObjectMetadataInputConfig { + config: s3_object_metadata_config, + } = &input_config + else { + return Err(Error::UnsupportedInputConfig); + }; + let dataset: Option = s3_object_metadata_config.dataset.clone().map(String::from); + if let Some(dataset_name) = dataset.as_ref() + && !VALID_DATASET_NAME_REGEX.is_match(dataset_name) + { + return Err(Error::InvalidDataset(dataset_name.clone())); + } + + let output_config = clp_io_config.output; + let clp_s_compression_option = ClpSCompressionOption { + target_encoded_size: output_config.target_segment_size + + output_config.target_dictionaries_size, + compression_level: output_config.compression_level, + timestamp_key: s3_object_metadata_config + .timestamp_key + .clone() + .map(String::from), + unstructured: s3_object_metadata_config.unstructured, + }; + + Ok(Self { + _db_pool: db_pool, + _db_config: db_config, + compression_job_id, + job_submitter, + resource_group_id, + _input_config: input_config, + clp_s_compression_option, + dataset, + _target_archive_size: output_config.target_archive_size, + spider_option, + }) + } + + /// Submits the compression job to Spider and drives it to completion. + /// + /// This method prepares the compression tasks' inputs, ensures the dataset's metadata tables + /// exist, submits the job, persists the Spider job ID it was assigned, and then waits for the + /// job to reach a terminal state. On any failure, the compression job is marked as + /// [`CompressionJobStatus::Failed`] before the error is returned. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Self::submit_and_wait`]'s return values on failure. + pub async fn run(self) -> Result<(), Error> { + tracing::info!(compression_job_id = % self.compression_job_id, "Starting compression job."); + + let result = self.submit_and_wait().await; + if let Err(err) = &result { + self.report_failure(err).await; + } + + result + } + + /// Resumes a compression job that was already submitted to Spider. + /// + /// + /// This method skips submission and waits for the Spider job identified by `spider_job_id` to + /// reach a terminal state. On failure, the compression job is marked as + /// [`CompressionJobStatus::Failed`] before the error is returned. + /// + /// NOTE: It's the caller's responsibility to ensure that the given Spider job ID is associated + /// with the compression job. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Self::to_completion`]'s return values on failure. + pub async fn recover(self, spider_job_id: SpiderJobId) -> Result<(), Error> { + tracing::info!( + compression_job_id = % self.compression_job_id, + spider_job_id = % spider_job_id, + "Recovering compression job.", + ); + match self.to_completion(spider_job_id).await { + Ok(()) => Ok(()), + Err(err) => { + self.report_failure(&err).await; + Err(err) + } + } + } + + /// Submits the compression job to Spider and waits for it to reach a terminal state. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Self::prepare_task_inputs`]'s return values on failure. + /// * Forwards [`Self::upsert_metadata_tables`]'s return values on failure. + /// * Forwards [`S3CompressionJobSubmitter::submit_s3_compression_job`]'s return values on + /// failure. + /// * Forwards [`Self::persist_spider_job_id`]'s return values on failure. + /// * Forwards [`Self::to_completion`]'s return values on failure. + async fn submit_and_wait(&self) -> Result<(), Error> { + let input_sources = self.prepare_task_inputs().await?; + + self.upsert_metadata_tables().await?; + + let spider_job_id = self + .job_submitter + .submit_s3_compression_job( + self.compression_job_id, + self.resource_group_id, + self.clp_s_compression_option.clone(), + self.dataset.clone(), + input_sources, + self.spider_option.commit_task_execution_policy.clone(), + ) + .await?; + tracing::info!( + compression_job_id = % self.compression_job_id, + spider_job_id = % spider_job_id, + "Compression job submitted.", + ); + + self.persist_spider_job_id(spider_job_id).await?; + tracing::info!( + compression_job_id = % self.compression_job_id, + spider_job_id = % spider_job_id, + "Compression job submission persisted.", + ); + + self.to_completion(spider_job_id).await + } + + /// Reports a compression job failure. + /// + /// This method logs the original error and attempts to mark the compression job as + /// [`CompressionJobStatus::Failed`] in the CLP database. The stored status message includes the + /// original error message. + /// + /// If updating the job status fails, the status-update error is logged for observability and + /// otherwise ignored. + async fn report_failure(&self, err: &Error) { + tracing::error!( + compression_job_id = % self.compression_job_id, + error = % err, + "Compression job failed.", + ); + let status_message = format!("Compression job failed: {err}"); + if let Err(e) = self + .update_job_status(CompressionJobStatus::Failed, Some(status_message)) + .await + { + tracing::error!( + compression_job_id = % self.compression_job_id, + error = % e, + "Failed to update job status on a job failure.", + ); + } + } + + /// Prepares the task inputs for the compression job. + /// + /// This method retrieves object metadata from the S3 object metadata table in the CLP database, + /// partitions the objects into compression task inputs, and derives an execution policy for + /// each task based on the number of objects it contains. + /// + /// # Returns + /// + /// A vector of tuples on success, where each tuple contains: + /// + /// * An [`S3InputSource`] representing the input to a single compression task. + /// * The [`ExecutionPolicy`] for that task. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// TODO + async fn prepare_task_inputs(&self) -> Result, Error> { + todo!("implement me!") + } + + /// Ensures that the required metadata tables exist for the configured dataset. + /// + /// This method creates the following tables in the CLP database if they do not already exist: + /// + /// * The archive metadata table. + /// * The column metadata table. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// TODO + async fn upsert_metadata_tables(&self) -> Result<(), Error> { + todo!("implement me!") + } + + /// Persists the Spider job ID and marks the compression job as running. + /// + /// This method associates the given Spider job ID with the compression job in the CLP database + /// and updates the compression job status to [`CompressionJobStatus::Running`]. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// TODO + async fn persist_spider_job_id(&self, _spider_job_id: SpiderJobId) -> Result<(), Error> { + todo!("implement me!") + } + + /// Waits for the associated Spider job to complete and finalizes the compression job. + /// + /// This method monitors the specified Spider job until it reaches a terminal state, then + /// updates the compression job according to the Spider job's result. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// TODO + async fn to_completion(&self, _spider_job_id: SpiderJobId) -> Result<(), Error> { + todo!("implement me!") + } + + /// Updates the compression job status in the CLP database. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// TODO + async fn update_job_status( + &self, + _job_status: CompressionJobStatus, + _status_message: Option, + ) -> Result<(), Error> { + todo!("implement me!") + } +} diff --git a/components/compression-coordinator/src/lib.rs b/components/compression-coordinator/src/lib.rs index 47921c004..419601eec 100644 --- a/components/compression-coordinator/src/lib.rs +++ b/components/compression-coordinator/src/lib.rs @@ -3,5 +3,6 @@ pub mod compression_job_submitter; mod error; +pub mod job_handle; pub use error::Error; From dc8594730422c4e90856b3e3158697928bbc2eb0 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Wed, 22 Jul 2026 14:42:31 -0400 Subject: [PATCH 2/9] Done. --- Cargo.lock | 27 ++++++++++++++----- components/compression-coordinator/Cargo.toml | 8 +++++- .../src/bin/compression_coordinator.rs | 19 +++++++++++++ 3 files changed, 46 insertions(+), 8 deletions(-) create mode 100644 components/compression-coordinator/src/bin/compression_coordinator.rs diff --git a/Cargo.lock b/Cargo.lock index 76e837c11..a4bba8f27 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -900,9 +900,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.6.1" +version = "4.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51" +checksum = "d91e0c145792ef73a6ad36d27c75ac09f1832222a3c209689d90f534685ee5b7" dependencies = [ "clap_builder", "clap_derive", @@ -910,9 +910,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.6.0" +version = "4.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f" +checksum = "f09628afdcc538b57f3c6341e9c8e9970f18e4a481690a64974d7023bd33548b" dependencies = [ "anstream", "anstyle", @@ -922,14 +922,14 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.6.1" +version = "4.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +checksum = "d012d2b9d65aca7f18f4d9878a045bc17899bba951561ba5ec3c2ba1eed9a061" dependencies = [ "heck", "proc-macro2", "quote", - "syn 2.0.117", + "syn 3.0.3", ] [[package]] @@ -1005,7 +1005,9 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" name = "compression-coordinator" version = "0.13.1-dev" dependencies = [ + "anyhow", "async-trait", + "clap", "clp-rust-utils", "non-empty-string", "rmp-serde", @@ -4255,6 +4257,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" diff --git a/components/compression-coordinator/Cargo.toml b/components/compression-coordinator/Cargo.toml index 455995775..2eaff68c7 100644 --- a/components/compression-coordinator/Cargo.toml +++ b/components/compression-coordinator/Cargo.toml @@ -7,7 +7,12 @@ edition = "2024" name = "compression_coordinator" path = "src/lib.rs" +[[bin]] +name = "compression-coordinator" +path = "src/bin/compression_coordinator.rs" + [dependencies] +anyhow = "1.0.100" async-trait = "0.1.89" clp-rust-utils = { path = "../clp-rust-utils" } rmp-serde = "1.3.1" @@ -16,8 +21,9 @@ spider-client = { git = "https://github.com/y-scope/spider.git", branch = "main" spider-core = { git = "https://github.com/y-scope/spider.git", branch = "main" } strsim = "0.11.1" thiserror = "2.0.18" -tokio = { version = "1.52.3", features = ["time"] } +tokio = { version = "1.52.3", features = ["time", "rt-multi-thread"] } tracing = "0.1.44" +clap = { version = "4.6.4", features = ["derive"] } [dev-dependencies] non-empty-string = "0.2.6" diff --git a/components/compression-coordinator/src/bin/compression_coordinator.rs b/components/compression-coordinator/src/bin/compression_coordinator.rs new file mode 100644 index 000000000..8e18f0072 --- /dev/null +++ b/components/compression-coordinator/src/bin/compression_coordinator.rs @@ -0,0 +1,19 @@ +use std::path::PathBuf; + +use clap::Parser; + +/// Command-line arguments for the compression coordinator. +#[derive(Debug, Parser)] +#[command(about = "Run the compression coordinator.")] +struct Cli { + /// Path to the configuration file. + #[arg(short, long, value_name = "PATH")] + config: PathBuf, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let _cli = Cli::parse(); + + Ok(()) +} From 0758f407a749a343cdf4e4e8e8a6e76d1e76b57f Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Wed, 22 Jul 2026 16:09:56 -0400 Subject: [PATCH 3/9] Update docker file to include compression coordinator. --- tools/docker-images/clp-package/Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/tools/docker-images/clp-package/Dockerfile b/tools/docker-images/clp-package/Dockerfile index 26c02e702..9be5a45a6 100644 --- a/tools/docker-images/clp-package/Dockerfile +++ b/tools/docker-images/clp-package/Dockerfile @@ -46,5 +46,6 @@ COPY --link --chown=${UID} ./build/spider/spider-build/src/spider/spider_worker COPY --link --chown=${UID} ./build/nodejs-22/bin/node bin/node-22 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/webui/ var/www/webui/ From e513c393a84ed71dea63fa9225941a6c78cbed36 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Wed, 22 Jul 2026 23:36:43 -0400 Subject: [PATCH 4/9] Done. --- Cargo.lock | 4 + components/api-server/src/error.rs | 6 +- .../src/clp_config/package/config.rs | 95 ++++ components/clp-rust-utils/src/error.rs | 3 + .../src/job_config/clp_io_config.rs | 12 +- .../src/job_config/compression.rs | 1 + .../src/serde/brotli_msgpack.rs | 27 +- components/compression-coordinator/Cargo.toml | 6 +- .../src/bin/compression_coordinator.rs | 128 ++++- .../src/coordination.rs | 519 ++++++++++++++++++ .../compression-coordinator/src/error.rs | 3 + components/compression-coordinator/src/lib.rs | 1 + 12 files changed, 788 insertions(+), 17 deletions(-) create mode 100644 components/compression-coordinator/src/coordination.rs diff --git a/Cargo.lock b/Cargo.lock index 640fcda85..44cc667e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1009,8 +1009,10 @@ dependencies = [ "async-trait", "clap", "clp-rust-utils", + "const_format", "non-empty-string", "rmp-serde", + "secrecy", "serde", "spider-client", "spider-core", @@ -1018,6 +1020,8 @@ dependencies = [ "strsim", "thiserror", "tokio", + "tokio-util", + "tonic", "tracing", ] diff --git a/components/api-server/src/error.rs b/components/api-server/src/error.rs index 484b462b4..edce2fef9 100644 --- a/components/api-server/src/error.rs +++ b/components/api-server/src/error.rs @@ -80,9 +80,9 @@ impl From for ClientError { impl From for ClientError { fn from(value: clp_rust_utils::Error) -> Self { match value { - clp_rust_utils::Error::MsgpackEncode(_) | clp_rust_utils::Error::SerdeYaml(_) => { - Self::MalformedData - } + clp_rust_utils::Error::MsgpackEncode(_) + | clp_rust_utils::Error::MsgpackDecode(_) + | clp_rust_utils::Error::SerdeYaml(_) => Self::MalformedData, clp_rust_utils::Error::Io(error) => error.into(), clp_rust_utils::Error::Sqlx(error) => error.into(), clp_rust_utils::Error::TelemetryExporterBuildError(error) => Self::Telemetry(error), diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs index 29863686b..44cdd5b0d 100644 --- a/components/clp-rust-utils/src/clp_config/package/config.rs +++ b/components/clp-rust-utils/src/clp_config/package/config.rs @@ -1,3 +1,6 @@ +use std::num::{NonZeroU32, NonZeroU64}; + +use non_empty_string::NonEmptyString; use serde::Deserialize; use crate::clp_config::{AwsAuthentication, S3Config}; @@ -22,6 +25,8 @@ pub struct Config { pub logs_input: LogsInput, pub archive_output: ArchiveOutput, pub telemetry: Telemetry, + pub spider: Option, + pub compression_coordinator: Option, } impl Default for Config { @@ -39,6 +44,8 @@ impl Default for Config { }, archive_output: ArchiveOutput::default(), telemetry: Telemetry::default(), + spider: None, + compression_coordinator: None, } } } @@ -337,6 +344,94 @@ impl Default for Telemetry { } } +/// Compression coordinator configuration. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq)] +#[serde(default)] +pub struct CompressionCoordinator { + /// The Spider resource group the coordinator submits jobs to. + pub resource_group: SpiderResourceGroup, + + /// The interval, in milliseconds, between polls for new compression jobs. + pub job_polling_interval_millisecs: NonZeroU64, + + /// The backoff schedule for polling a submitted job's result. + pub result_polling: PollingBackoff, + + /// The maximum number of retries for a compression task. + pub compression_task_max_retry: u32, + + /// The maximum number of retries for a commit task. + pub commit_task_max_retry: u32, + + /// The size of the database connection pool. + pub database_connection_pool_size: NonZeroU32, + + /// The timeout, in seconds, for graceful shutdown. + pub termination_timeout_secs: NonZeroU64, + + /// The soft timeout, in seconds, for a commit task. + pub commit_task_soft_timeout_secs: NonZeroU64, + + /// The hard timeout, in seconds, for a commit task. + pub commit_task_hard_timeout_secs: NonZeroU64, +} + +impl Default for CompressionCoordinator { + fn default() -> Self { + Self { + resource_group: SpiderResourceGroup { + name: NonEmptyString::new("compression-coordinator".to_owned()) + .expect("default resource group name should not be empty"), + }, + job_polling_interval_millisecs: NonZeroU64::new(100) + .expect("default jobs poll delay should not be zero"), + result_polling: PollingBackoff { + init_backoff_millisecs: NonZeroU64::new(100) + .expect("default result polling init backoff should not be zero"), + max_backoff_millisecs: NonZeroU64::new(1000) + .expect("default result polling max backoff should not be zero"), + }, + compression_task_max_retry: 1, + commit_task_max_retry: 1, + database_connection_pool_size: NonZeroU32::new(10) + .expect("default database connection pool size should not be zero"), + termination_timeout_secs: NonZeroU64::new(30) + .expect("default termination timeout should not be zero"), + commit_task_soft_timeout_secs: NonZeroU64::new(45) + .expect("default commit task soft timeout should not be zero"), + commit_task_hard_timeout_secs: NonZeroU64::new(60) + .expect("default commit task hard timeout should not be zero"), + } + } +} + +/// Spider configuration. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq)] +pub struct Spider { + /// The Spider cluster's host. + pub host: NonEmptyString, + + /// The Spider cluster's port. + pub port: u16, +} + +/// Spider resource group configuration. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq)] +pub struct SpiderResourceGroup { + /// The name of the Spider resource group. + pub name: NonEmptyString, +} + +/// Polling backoff configuration. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq)] +pub struct PollingBackoff { + /// The initial backoff, in milliseconds, between polls. + pub init_backoff_millisecs: NonZeroU64, + + /// The maximum backoff, in milliseconds, between polls. + pub max_backoff_millisecs: NonZeroU64, +} + #[cfg(test)] mod tests { use super::LogsInput; diff --git a/components/clp-rust-utils/src/error.rs b/components/clp-rust-utils/src/error.rs index 61271f7ad..8bc43de12 100644 --- a/components/clp-rust-utils/src/error.rs +++ b/components/clp-rust-utils/src/error.rs @@ -5,6 +5,9 @@ pub enum Error { #[error("`rmp_serde::encode::Error`: {0}")] MsgpackEncode(#[from] rmp_serde::encode::Error), + #[error("`rmp_serde::decode::Error`: {0}")] + MsgpackDecode(#[from] rmp_serde::decode::Error), + #[error("`std::io::Error`: {0}")] Io(#[from] std::io::Error), diff --git a/components/clp-rust-utils/src/job_config/clp_io_config.rs b/components/clp-rust-utils/src/job_config/clp_io_config.rs index 7f4e19cb5..a9d54a70c 100644 --- a/components/clp-rust-utils/src/job_config/clp_io_config.rs +++ b/components/clp-rust-utils/src/job_config/clp_io_config.rs @@ -1,5 +1,5 @@ use non_empty_string::NonEmptyString; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use crate::{ clp_config::S3Config, @@ -8,14 +8,14 @@ use crate::{ }; /// Represents CLP IO config. -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ClpIoConfig { pub input: InputConfig, pub output: OutputConfig, } /// An enum representing CLP input config. -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "type")] pub enum InputConfig { #[serde(rename = "s3")] @@ -32,7 +32,7 @@ pub enum InputConfig { } /// Represents S3 input config. -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct S3InputConfig { #[serde(flatten)] pub s3_config: S3Config, @@ -43,7 +43,7 @@ pub struct S3InputConfig { pub unstructured: bool, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct S3ObjectMetadataInputConfig { #[serde(flatten)] pub s3_config: S3Config, @@ -56,7 +56,7 @@ pub struct S3ObjectMetadataInputConfig { } /// Represents CLP output config. -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct OutputConfig { pub target_archive_size: u64, pub target_dictionaries_size: u64, diff --git a/components/clp-rust-utils/src/job_config/compression.rs b/components/clp-rust-utils/src/job_config/compression.rs index ddebf2f88..99b73a12d 100644 --- a/components/clp-rust-utils/src/job_config/compression.rs +++ b/components/clp-rust-utils/src/job_config/compression.rs @@ -18,6 +18,7 @@ pub type CompressionJobId = i32; Serialize, ToSchema, TryFromPrimitive, + sqlx::Type, )] #[repr(i32)] #[strum(ascii_case_insensitive)] diff --git a/components/clp-rust-utils/src/serde/brotli_msgpack.rs b/components/clp-rust-utils/src/serde/brotli_msgpack.rs index a85afc15a..0ac1beae9 100644 --- a/components/clp-rust-utils/src/serde/brotli_msgpack.rs +++ b/components/clp-rust-utils/src/serde/brotli_msgpack.rs @@ -1,7 +1,7 @@ -use std::io::Write; +use std::io::{Read, Write}; -use brotli::CompressorWriter; -use serde::Serialize; +use brotli::{CompressorWriter, Decompressor}; +use serde::{Serialize, de::DeserializeOwned}; use crate::Error; @@ -11,7 +11,8 @@ pub struct BrotliMsgpack {} impl BrotliMsgpack { /// Serialize a value to a Brotli-compressed `MessagePack` byte sequence. /// - /// # Return + /// # Returns + /// /// A vector of bytes containing the serialized byte sequence. /// /// # Errors @@ -26,4 +27,22 @@ impl BrotliMsgpack { brotli_compressor.write_all(&msgpack_data)?; Ok(brotli_compressor.into_inner()) } + + /// Deserialize an owned value from a Brotli-compressed `MessagePack` byte sequence. + /// + /// # Returns + /// + /// The deserialized value. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`rmp_serde::from_slice`]'s errors on failure. + /// * Forwards [`std::io::Read::read_to_end`]'s errors on failure. + pub fn deserialize(data: &[u8]) -> Result { + let mut msgpack_data = Vec::new(); + Decompressor::new(data, 4096).read_to_end(&mut msgpack_data)?; + rmp_serde::from_slice(&msgpack_data).map_err(Into::into) + } } diff --git a/components/compression-coordinator/Cargo.toml b/components/compression-coordinator/Cargo.toml index 875f3e95b..636033298 100644 --- a/components/compression-coordinator/Cargo.toml +++ b/components/compression-coordinator/Cargo.toml @@ -15,14 +15,18 @@ path = "src/bin/compression_coordinator.rs" anyhow = "1.0.100" async-trait = "0.1.89" clp-rust-utils = { path = "../clp-rust-utils" } +const_format = "0.2.35" rmp-serde = "1.3.1" +secrecy = "0.10.3" serde = { version = "1.0.228", features = ["derive"] } spider-client = { git = "https://github.com/y-scope/spider.git", branch = "main" } spider-core = { git = "https://github.com/y-scope/spider.git", branch = "main" } sqlx = { version = "0.8.6", features = ["runtime-tokio", "mysql"] } strsim = "0.11.1" thiserror = "2.0.18" -tokio = { version = "1.52.3", features = ["time", "rt-multi-thread"] } +tokio-util = "0.7.18" +tokio = { version = "1.52.3", features = ["time", "rt-multi-thread", "signal"] } +tonic = "0.14.6" tracing = "0.1.44" clap = { version = "4.6.4", features = ["derive"] } diff --git a/components/compression-coordinator/src/bin/compression_coordinator.rs b/components/compression-coordinator/src/bin/compression_coordinator.rs index 8e18f0072..6da2bf02d 100644 --- a/components/compression-coordinator/src/bin/compression_coordinator.rs +++ b/components/compression-coordinator/src/bin/compression_coordinator.rs @@ -1,6 +1,11 @@ -use std::path::PathBuf; +use std::{path::PathBuf, time::Duration}; use clap::Parser; +use clp_rust_utils::{ + clp_config::package::{self}, + database::mysql::create_clp_db_mysql_pool, + serde::yaml, +}; /// Command-line arguments for the compression coordinator. #[derive(Debug, Parser)] @@ -11,9 +16,126 @@ struct Cli { config: PathBuf, } +#[allow(clippy::too_many_lines)] #[tokio::main] async fn main() -> anyhow::Result<()> { - let _cli = Cli::parse(); + let args = Cli::parse(); - Ok(()) + let _guard = clp_rust_utils::logging::set_up_logging("compression_coordinator.log"); + + let config: package::config::Config = yaml::from_path(args.config).inspect_err(|e| { + tracing::error!(error = % e, "Failed to load the configuration file."); + })?; + + let credentials = package::credentials::Credentials { + database: package::credentials::Database { + password: secrecy::SecretString::new( + std::env::var("CLP_DB_PASS") + .inspect_err(|e| { + tracing::error!( + error = % e, + "Failed to read the database password from `CLP_DB_PASS`." + ); + })? + .into_boxed_str(), + ), + user: std::env::var("CLP_DB_USER").inspect_err(|e| { + tracing::error!( + error = % e, + "Failed to read the database user from `CLP_DB_USER`." + ); + })?, + }, + }; + + let coordinator_config = config.compression_coordinator.ok_or_else(|| { + const ERROR_MESSAGE: &str = "Compression coordinator configuration is missing."; + tracing::error!(ERROR_MESSAGE); + anyhow::anyhow!(ERROR_MESSAGE) + })?; + + let spider_config = config.spider.ok_or_else(|| { + const ERROR_MESSAGE: &str = "Spider configuration is missing."; + tracing::error!(ERROR_MESSAGE); + anyhow::anyhow!(ERROR_MESSAGE) + })?; + + let db_pool = create_clp_db_mysql_pool( + &config.database, + &credentials.database, + coordinator_config.database_connection_pool_size.get(), + ) + .await + .inspect_err(|e| tracing::error!(error = % e, "Failed to create the database pool."))?; + + let (coordinator, cancellation_token) = + compression_coordinator::coordination::Coordinator::new( + &coordinator_config, + &spider_config, + db_pool, + config.database, + ) + .await?; + + 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 = 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 { + join_result + } else { + let termination_timeout = + Duration::from_secs(coordinator_config.termination_timeout_secs.get()); + if let Ok(join_result) = + tokio::time::timeout(termination_timeout, &mut coordinator_handle).await + { + join_result + } else { + tracing::warn!( + "The compression coordinator did not stop within {termination_timeout:?}. \ + Aborting." + ); + coordinator_handle.abort(); + return Ok(()); + } + }; + + match join_result { + Ok(Ok(())) => { + tracing::info!("Compression coordinator stopped."); + Ok(()) + } + Ok(Err(e)) => { + tracing::error!(error = % e, "Compression coordinator returned on error."); + Err(anyhow::anyhow!( + "Compression coordinator returned on error." + )) + } + Err(err) => { + const ERROR_MESSAGE: &str = "Failed to join the compression coordinator."; + tracing::error!(error = % err, ERROR_MESSAGE); + Err(anyhow::anyhow!(ERROR_MESSAGE)) + } + } } diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs new file mode 100644 index 000000000..9da7bb1b4 --- /dev/null +++ b/components/compression-coordinator/src/coordination.rs @@ -0,0 +1,519 @@ +//! The coordinator poll loop that discovers pending CLP compression jobs and dispatches them to +//! Spider. + +use std::{sync::Arc, time::Duration}; + +use clp_rust_utils::{ + clp_config::package::config::{ + CompressionCoordinator as CoordinatorConfig, + Database as DatabaseConfig, + Spider as SpiderConfig, + SpiderResourceGroup, + }, + job_config::{ClpIoConfig, CompressionJobId, CompressionJobStatus}, + serde::BrotliMsgpack, +}; +use const_format::formatcp; +use spider_client::SpiderClient; +use spider_core::{ + task::{ExecutionPolicy, TimeoutPolicy}, + types::id::{JobId as SpiderJobId, ResourceGroupId}, +}; +use tokio::{select, time::Instant}; +use tokio_util::sync::CancellationToken; +use tonic::transport::Endpoint; + +use crate::{ + Error, + job_handle::{S3CompressionJobHandle, SpiderOption}, +}; + +/// Coordinator for fetching new compression jobs and submitting them to Spider. +pub struct Coordinator { + resource_group_id: ResourceGroupId, + spider_client: SpiderClient, + db_pool: sqlx::MySqlPool, + db_config: DatabaseConfig, + spider_option: Arc, + last_polled_job_id: Option, + job_polling_interval: Duration, + cancellation_token: CancellationToken, +} + +impl Coordinator { + /// Factory function. + /// + /// On construction, this recovers compression jobs that a previous coordinator instance had + /// already submitted to Spider (those still `RUNNING` with a Spider job ID) by spawning a + /// detached handle to drive each one to completion. + /// + /// # Returns + /// + /// A tuple on success, containing: + /// + /// * The constructed [`Coordinator`]. + /// * The [`CancellationToken`] the caller uses to request shutdown. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`Error::InvalidEndpoint`] if the Spider host and port do not form a valid endpoint. + /// * Forwards [`SpiderClient::builder`]'s connection return values on failure. + /// * Forwards [`get_or_create_resource_group_id`]'s return values on failure. + /// * Forwards [`Self::fetch_submitted_running_jobs`]'s return values on failure. + pub async fn new( + coordinator_config: &CoordinatorConfig, + spider_config: &SpiderConfig, + db_pool: sqlx::MySqlPool, + db_config: DatabaseConfig, + ) -> Result<(Self, CancellationToken), Error> { + let spider_host = spider_config.host.as_str(); + let spider_port = spider_config.port; + let endpoint_str = format!("http://{spider_host}:{spider_port}"); + let endpoint = Endpoint::from_shared(endpoint_str) + .inspect_err(|e| { + tracing::error!(error = % e, "Failed to create Spider endpoint."); + }) + .map_err(|e| Error::InvalidEndpoint(e.to_string()))?; + let spider_client = SpiderClient::builder(endpoint) + .connect() + .await + .inspect_err(|e| { + tracing::error!(error = % e, "Failed to connect to Spider."); + })?; + let resource_group_id = get_or_create_resource_group_id( + &coordinator_config.resource_group, + &spider_client, + &db_pool, + ) + .await + .inspect_err(|e| { + tracing::error!(error = % e, "Failed to get or create resource group."); + })?; + + let spider_option = Arc::new(SpiderOption { + compression_task_max_retry: coordinator_config.compression_task_max_retry, + commit_task_execution_policy: ExecutionPolicy { + max_num_instances: 1, + max_num_retry: coordinator_config.commit_task_max_retry, + timeout_policy: TimeoutPolicy { + soft_timeout_ms: coordinator_config.commit_task_soft_timeout_secs.get() * 1000, + hard_timeout_ms: coordinator_config.commit_task_hard_timeout_secs.get() * 1000, + }, + }, + initial_poll_backoff: Duration::from_millis( + coordinator_config + .result_polling + .init_backoff_millisecs + .get(), + ), + max_poll_backoff: Duration::from_millis( + coordinator_config + .result_polling + .max_backoff_millisecs + .get(), + ), + }); + + let cancellation_token = CancellationToken::new(); + + let coordinator = Self { + resource_group_id, + spider_client, + db_pool, + db_config, + spider_option, + last_polled_job_id: None, + job_polling_interval: Duration::from_millis( + coordinator_config.job_polling_interval_millisecs.get(), + ), + cancellation_token: cancellation_token.clone(), + }; + + for (job_id, spider_job_id, clp_io_config) in + coordinator.fetch_submitted_running_jobs().await? + { + tracing::info!( + job_id = % job_id, + spider_job_id = % spider_job_id, + "Recovering a previously submitted job." + ); + let Ok(job_handle) = coordinator.create_job_handle(job_id, clp_io_config).await else { + continue; + }; + tokio::spawn(async move { + let _ = job_handle.recover(spider_job_id).await.inspect_err(|e| { + tracing::error!( + error = % e, + job_id = % job_id, + spider_job_id = % spider_job_id, + "The recovered compression job failed." + ); + }); + }); + } + + Ok((coordinator, cancellation_token)) + } + + /// Runs the coordinator's poll loop until cancelled. + /// + /// On each iteration, this method fetches the pending compression jobs, spawns a detached + /// handle to drive each one, and then sleeps until the next poll or until the cancellation + /// token is triggered. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Self::schedule_new_jobs`]'s return values on failure. + pub async fn run(mut self) -> Result<(), Error> { + let cancellation_token = self.cancellation_token.clone(); + loop { + let now = Instant::now(); + + select! { + () = cancellation_token.cancelled() => { + break; + } + result = self.schedule_new_jobs() => { + result.inspect_err(|e| { + tracing::error!(error = % e, "Failed to schedule new jobs."); + })?; + } + } + + let elapsed = now.elapsed(); + let sleep_duration = self.job_polling_interval.saturating_sub(elapsed); + if sleep_duration.is_zero() { + tokio::task::yield_now().await; + } else if tokio::time::timeout(sleep_duration, cancellation_token.cancelled()) + .await + .is_ok() + { + break; + } + } + + tracing::info!("Coordinator shutting down."); + Ok(()) + } + + /// Marks the compression job identified by `job_id` as [`CompressionJobStatus::Failed`]. + /// + /// This is a best-effort update; if it fails, the error is logged and otherwise ignored. + async fn mark_job_failed(&self, job_id: CompressionJobId, status_msg: &str) { + const QUERY: &str = formatcp!( + "UPDATE `{table}` SET `status` = ?, `status_msg` = ?, `update_time` = \ + CURRENT_TIMESTAMP() WHERE `id` = ?;", + table = COMPRESSION_JOB_TABLE_NAME, + ); + tracing::info!(job_id = % job_id, "Failing the compression job."); + if let Err(e) = sqlx::query(QUERY) + .bind(CompressionJobStatus::Failed) + .bind(status_msg) + .bind(job_id) + .execute(&self.db_pool) + .await + { + tracing::error!( + error = % e, + job_id = % job_id, + "Failed to mark the compression job as failed." + ); + } + } + + /// Fetches the pending compression jobs and spawns a detached handle to drive each one. + /// + /// A job whose config cannot be deserialized is marked `FAILED` and skipped; a job whose handle + /// cannot be constructed is skipped as well (and marked `FAILED` unless its input config is + /// unsupported, in which case it is left for another handler). + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`Self::fetch_new_job_rows`]'s return values on failure. + async fn schedule_new_jobs(&mut self) -> Result<(), Error> { + let new_job_rows = self.fetch_new_job_rows().await.inspect_err(|e| { + tracing::error!(error = % e, "Failed to fetch new jobs from database."); + })?; + for job_row in new_job_rows { + let job_id = job_row.id; + let clp_io_config: ClpIoConfig = + match BrotliMsgpack::deserialize(&job_row.serialized_clp_io_config) { + Ok(clp_io_config) => clp_io_config, + Err(e) => { + tracing::error!( + error = % e, + job_id = % job_id, + "Failed to deserialize CLP I/O config. Skipping." + ); + self.mark_job_failed( + job_id, + &format!("Failed to deserialize CLP I/O config: {e}"), + ) + .await; + continue; + } + }; + tracing::info!(job_id = % job_id, "Scheduling new job."); + let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { + continue; + }; + tokio::spawn(async move { + let _ = job_handle.run().await.inspect_err(|e| { + tracing::error!( + error = % e, + job_id = % job_id, + "Failed to schedule S3 compression job." + ); + }); + }); + } + Ok(()) + } + + /// Constructs an [`S3CompressionJobHandle`] for the given job. + /// + /// A construction failure is logged, and the job is marked `FAILED` for any failure other than + /// an unsupported input config, which is only warned and left for another handler. + /// + /// # Returns + /// + /// The constructed [`S3CompressionJobHandle`] on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`S3CompressionJobHandle::new`]'s return values on failure. + async fn create_job_handle( + &self, + job_id: CompressionJobId, + clp_io_config: ClpIoConfig, + ) -> Result, Error> { + let result = S3CompressionJobHandle::new( + self.db_pool.clone(), + self.db_config.clone(), + job_id, + self.spider_client.clone(), + self.resource_group_id, + clp_io_config, + self.spider_option.clone(), + ); + + if let Err(e) = &result { + if matches!(e, Error::UnsupportedInputConfig) { + tracing::warn!( + error = % e, + job_id = % job_id, + "Unsupported input config. Skipping." + ); + } else { + tracing::error!( + error = % e, + job_id = % job_id, + "Failed to create S3 job handle. Skipping." + ); + self.mark_job_failed( + job_id, + &format!("Failed to create the compression job handle: {e}"), + ) + .await; + } + } + + result + } + + /// Fetches the pending compression jobs newer than the last polled job and advances the poll + /// cursor. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`sqlx::query::QueryAs::fetch_all`]'s return values on failure. + async fn fetch_new_job_rows(&mut self) -> Result, Error> { + let query = self.last_polled_job_id.map_or_else( + || { + sqlx::query_as::<_, PendingJobRowProjection>(formatcp!( + "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? ORDER BY `id` \ + ASC;", + table = COMPRESSION_JOB_TABLE_NAME, + )) + .bind(CompressionJobStatus::Pending) + }, + |last_polled_job_id| { + sqlx::query_as::<_, PendingJobRowProjection>(formatcp!( + "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? AND `id` > ? \ + ORDER BY `id` ASC;", + table = COMPRESSION_JOB_TABLE_NAME, + )) + .bind(CompressionJobStatus::Pending) + .bind(last_polled_job_id) + }, + ); + + let rows = query.fetch_all(&self.db_pool).await?; + if let Some(last_row) = rows.last() { + self.last_polled_job_id = Some(last_row.id); + } + + Ok(rows) + } + + /// Fetches jobs that are still in [`CompressionJobStatus::Running`] and were previously + /// submitted by the compression coordinator. + /// + /// A running job whose config cannot be deserialized is marked `FAILED` and skipped. + /// + /// # Returns + /// + /// A vector of tuples on success, each tuple containing: + /// + /// * The compression job ID. + /// * The Spider job ID. + /// * The IO config of the compression job. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`sqlx::query::QueryAs::fetch_all`]'s return values on failure. + async fn fetch_submitted_running_jobs( + &self, + ) -> Result, Error> { + const QUERY: &str = formatcp!( + "SELECT `id`, `spider_id`, `clp_config` FROM `{table}` WHERE `status` = ? AND \ + `spider_id` IS NOT NULL;", + table = COMPRESSION_JOB_TABLE_NAME, + ); + + let mut recovery_context = Vec::new(); + for row in sqlx::query_as::<_, RunningJobRowProjection>(QUERY) + .bind(CompressionJobStatus::Running) + .fetch_all(&self.db_pool) + .await? + { + let clp_io_config: ClpIoConfig = + match BrotliMsgpack::deserialize(&row.serialized_clp_io_config) { + Ok(clp_io_config) => clp_io_config, + Err(e) => { + tracing::error!( + error = % e, + job_id = % row.id, + "Failed to deserialize CLP I/O config of a running job. The database \ + might be corrupted. Skipping." + ); + self.mark_job_failed( + row.id, + &format!("Failed to deserialize CLP I/O config: {e}"), + ) + .await; + continue; + } + }; + recovery_context.push((row.id, row.spider_job_id, clp_io_config)); + } + + Ok(recovery_context) + } +} + +const COMPRESSION_JOB_TABLE_NAME: &str = "compression_jobs"; + +/// A projection of the columns read from a [`CompressionJobStatus::Pending`] compression job row. +#[derive(Debug, sqlx::FromRow)] +struct PendingJobRowProjection { + id: CompressionJobId, + #[sqlx(rename = "clp_config")] + serialized_clp_io_config: Vec, +} + +/// A projection of the columns read from a [`CompressionJobStatus::Running`] compression job row. +#[derive(Debug, sqlx::FromRow)] +struct RunningJobRowProjection { + id: CompressionJobId, + #[sqlx(rename = "spider_id")] + spider_job_id: SpiderJobId, + #[sqlx(rename = "clp_config")] + serialized_clp_io_config: Vec, +} + +/// Retrieves the Spider resource group ID for the configured resource group, registering it if it +/// does not yet exist. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. +/// * Forwards [`SpiderClient::add_resource_group`]'s return values on failure. +async fn get_or_create_resource_group_id( + resource_group_config: &SpiderResourceGroup, + spider_client: &SpiderClient, + db_pool: &sqlx::MySqlPool, +) -> Result { + const SPIDER_RESOURCE_GROUP_TABLE_NAME: &str = "spider_resource_groups"; + + const CREATE_TABLE_QUERY: &str = formatcp!( + "CREATE TABLE IF NOT EXISTS `{table}` ( + `rg_name` VARCHAR(255) NOT NULL, + `rg_id` BIGINT UNSIGNED NOT NULL, + PRIMARY KEY (`rg_name`) USING BTREE + ) ROW_FORMAT=DYNAMIC", + table = SPIDER_RESOURCE_GROUP_TABLE_NAME, + ); + const SELECT_QUERY: &str = formatcp!( + "SELECT `rg_id` FROM `{table}` WHERE `rg_name` = ?;", + table = SPIDER_RESOURCE_GROUP_TABLE_NAME, + ); + const INSERT_QUERY: &str = formatcp!( + "INSERT INTO `{table}` (`rg_name`, `rg_id`) VALUES (?, ?);", + table = SPIDER_RESOURCE_GROUP_TABLE_NAME, + ); + + sqlx::query(CREATE_TABLE_QUERY).execute(db_pool).await?; + + let resource_group = resource_group_config.name.as_str(); + let existing_rg_id: Option = sqlx::query_scalar(SELECT_QUERY) + .bind(resource_group) + .fetch_optional(db_pool) + .await?; + if let Some(spider_rg_id) = existing_rg_id { + tracing::info!( + resource_group = % resource_group, + spider_rg_id = % spider_rg_id, + "Resource group already registered. Returning Spider resource group ID." + ); + return Ok(ResourceGroupId::from(spider_rg_id)); + } + + // NOTE: For now, Spider does not enforce resource group credential validation. The password is + // hardcoded to be the same as the username. + let resource_group_id = spider_client + .add_resource_group( + resource_group.to_owned(), + resource_group.as_bytes().to_vec(), + ) + .await?; + + sqlx::query(INSERT_QUERY) + .bind(resource_group) + .bind(resource_group_id.get()) + .execute(db_pool) + .await + .inspect_err(|e| { + tracing::error!( + error = % e, + "Failed to insert resource group into database. This might be a race condition. \ + Restart the service to retry." + ); + })?; + + Ok(resource_group_id) +} diff --git a/components/compression-coordinator/src/error.rs b/components/compression-coordinator/src/error.rs index 129503bbf..76f31899f 100644 --- a/components/compression-coordinator/src/error.rs +++ b/components/compression-coordinator/src/error.rs @@ -6,6 +6,9 @@ pub enum Error { #[error("invalid dataset: {0}")] InvalidDataset(String), + #[error("invalid endpoint: {0}")] + InvalidEndpoint(String), + #[error("S3 bucket mismatch: expected `{0}`, but got `{1}`")] S3BucketMismatch(String, String), diff --git a/components/compression-coordinator/src/lib.rs b/components/compression-coordinator/src/lib.rs index bf3c59997..868332378 100644 --- a/components/compression-coordinator/src/lib.rs +++ b/components/compression-coordinator/src/lib.rs @@ -1,6 +1,7 @@ //! Compression-job submission and S3 input partitioning for a Spider cluster. pub mod compression_job_submitter; +pub mod coordination; mod error; pub mod job_handle; pub mod partition; From a7ea194c57094b1561ff151d63cb4182f74ad928 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Wed, 22 Jul 2026 23:55:33 -0400 Subject: [PATCH 5/9] Polish docstring. --- .../src/coordination.rs | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index 9da7bb1b4..f289d2a6f 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -44,8 +44,8 @@ impl Coordinator { /// Factory function. /// /// On construction, this recovers compression jobs that a previous coordinator instance had - /// already submitted to Spider (those still `RUNNING` with a Spider job ID) by spawning a - /// detached handle to drive each one to completion. + /// already submitted to Spider (those still [`CompressionJobStatus::Running`] with a Spider job + /// ID) by spawning a detached handle to drive each one to completion. /// /// # Returns /// @@ -227,9 +227,11 @@ impl Coordinator { /// Fetches the pending compression jobs and spawns a detached handle to drive each one. /// - /// A job whose config cannot be deserialized is marked `FAILED` and skipped; a job whose handle - /// cannot be constructed is skipped as well (and marked `FAILED` unless its input config is - /// unsupported, in which case it is left for another handler). + /// + /// A job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] and + /// skipped; a job whose handle cannot be constructed is skipped as well (and marked + /// [`CompressionJobStatus::Failed`] unless its input config is unsupported, in which case it is + /// left for the legacy Celery-based compression scheduler). /// /// # Errors /// @@ -278,8 +280,9 @@ impl Coordinator { /// Constructs an [`S3CompressionJobHandle`] for the given job. /// - /// A construction failure is logged, and the job is marked `FAILED` for any failure other than - /// an unsupported input config, which is only warned and left for another handler. + /// A construction failure is logged, and the job is marked [`CompressionJobStatus::Failed`] for + /// any failure other than an unsupported input config, which is only warned and left for + /// another handler. /// /// # Returns /// @@ -369,7 +372,8 @@ impl Coordinator { /// Fetches jobs that are still in [`CompressionJobStatus::Running`] and were previously /// submitted by the compression coordinator. /// - /// A running job whose config cannot be deserialized is marked `FAILED` and skipped. + /// A running job whose config cannot be deserialized is marked [`CompressionJobStatus::Failed`] + /// and skipped. /// /// # Returns /// From 0ddd1e5621aafaa085f3a911e5b3e33859f8e6fb Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Thu, 23 Jul 2026 00:01:22 -0400 Subject: [PATCH 6/9] Update config docstrings. --- .../src/clp_config/package/config.rs | 24 ------------------- 1 file changed, 24 deletions(-) diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs index 44cdd5b0d..a47a75e91 100644 --- a/components/clp-rust-utils/src/clp_config/package/config.rs +++ b/components/clp-rust-utils/src/clp_config/package/config.rs @@ -348,31 +348,14 @@ impl Default for Telemetry { #[derive(Clone, Debug, Deserialize, Eq, PartialEq)] #[serde(default)] pub struct CompressionCoordinator { - /// The Spider resource group the coordinator submits jobs to. pub resource_group: SpiderResourceGroup, - - /// The interval, in milliseconds, between polls for new compression jobs. pub job_polling_interval_millisecs: NonZeroU64, - - /// The backoff schedule for polling a submitted job's result. pub result_polling: PollingBackoff, - - /// The maximum number of retries for a compression task. pub compression_task_max_retry: u32, - - /// The maximum number of retries for a commit task. pub commit_task_max_retry: u32, - - /// The size of the database connection pool. pub database_connection_pool_size: NonZeroU32, - - /// The timeout, in seconds, for graceful shutdown. pub termination_timeout_secs: NonZeroU64, - - /// The soft timeout, in seconds, for a commit task. pub commit_task_soft_timeout_secs: NonZeroU64, - - /// The hard timeout, in seconds, for a commit task. pub commit_task_hard_timeout_secs: NonZeroU64, } @@ -408,27 +391,20 @@ impl Default for CompressionCoordinator { /// Spider configuration. #[derive(Clone, Debug, Deserialize, Eq, PartialEq)] pub struct Spider { - /// The Spider cluster's host. pub host: NonEmptyString, - - /// The Spider cluster's port. pub port: u16, } /// Spider resource group configuration. #[derive(Clone, Debug, Deserialize, Eq, PartialEq)] pub struct SpiderResourceGroup { - /// The name of the Spider resource group. pub name: NonEmptyString, } /// Polling backoff configuration. #[derive(Clone, Debug, Deserialize, Eq, PartialEq)] pub struct PollingBackoff { - /// The initial backoff, in milliseconds, between polls. pub init_backoff_millisecs: NonZeroU64, - - /// The maximum backoff, in milliseconds, between polls. pub max_backoff_millisecs: NonZeroU64, } From f1623b0bbebfdd770b417161813d7c480a81be5c Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 26 Jul 2026 21:53:31 -0400 Subject: [PATCH 7/9] Update dispatch handling. --- .../src/coordination.rs | 106 +++++++++++++----- 1 file changed, 77 insertions(+), 29 deletions(-) diff --git a/components/compression-coordinator/src/coordination.rs b/components/compression-coordinator/src/coordination.rs index f289d2a6f..0873e9d42 100644 --- a/components/compression-coordinator/src/coordination.rs +++ b/components/compression-coordinator/src/coordination.rs @@ -35,7 +35,7 @@ pub struct Coordinator { db_pool: sqlx::MySqlPool, db_config: DatabaseConfig, spider_option: Arc, - last_polled_job_id: Option, + is_first_fetch: bool, job_polling_interval: Duration, cancellation_token: CancellationToken, } @@ -124,7 +124,7 @@ impl Coordinator { db_pool, db_config, spider_option, - last_polled_job_id: None, + is_first_fetch: true, job_polling_interval: Duration::from_millis( coordinator_config.job_polling_interval_millisecs.get(), ), @@ -161,24 +161,27 @@ impl Coordinator { /// /// On each iteration, this method fetches the pending compression jobs, spawns a detached /// handle to drive each one, and then sleeps until the next poll or until the cancellation - /// token is triggered. + /// token is triggered. The jobs dispatched in the iteration are marked once the sleep elapses, + /// so their update does not contend with concurrent job submissions during the poll interval. /// /// # Errors /// /// Returns an error if: /// /// * Forwards [`Self::schedule_new_jobs`]'s return values on failure. + /// * Forwards [`Self::mark_jobs_dispatched`]'s return values on failure. pub async fn run(mut self) -> Result<(), Error> { let cancellation_token = self.cancellation_token.clone(); loop { let now = Instant::now(); + let dispatched_job_ids; select! { () = cancellation_token.cancelled() => { break; } result = self.schedule_new_jobs() => { - result.inspect_err(|e| { + dispatched_job_ids = result.inspect_err(|e| { tracing::error!(error = % e, "Failed to schedule new jobs."); })?; } @@ -194,6 +197,8 @@ impl Coordinator { { break; } + + self.mark_jobs_dispatched(&dispatched_job_ids).await?; } tracing::info!("Coordinator shutting down."); @@ -233,15 +238,21 @@ impl Coordinator { /// [`CompressionJobStatus::Failed`] unless its input config is unsupported, in which case it is /// left for the legacy Celery-based compression scheduler). /// + /// # Returns + /// + /// The IDs of the fetched jobs that were dispatched in this poll. + /// /// # Errors /// /// Returns an error if: /// /// * Forwards [`Self::fetch_new_job_rows`]'s return values on failure. - async fn schedule_new_jobs(&mut self) -> Result<(), Error> { + async fn schedule_new_jobs(&mut self) -> Result, Error> { let new_job_rows = self.fetch_new_job_rows().await.inspect_err(|e| { tracing::error!(error = % e, "Failed to fetch new jobs from database."); })?; + let dispatched_job_ids: Vec = + new_job_rows.iter().map(|row| row.id).collect(); for job_row in new_job_rows { let job_id = job_row.id; let clp_io_config: ClpIoConfig = @@ -275,6 +286,38 @@ impl Coordinator { }); }); } + Ok(dispatched_job_ids) + } + + /// Marks the compression jobs identified by `job_ids` with the current dispatch time. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`sqlx::Pool::begin`]'s return values on failure. + /// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. + /// * Forwards [`sqlx::Transaction::commit`]'s return values on failure. + async fn mark_jobs_dispatched(&self, job_ids: &[CompressionJobId]) -> Result<(), Error> { + if job_ids.is_empty() { + return Ok(()); + } + + let mut tx = self.db_pool.begin().await?; + for chunk in job_ids.chunks(1000) { + let mut query_builder = sqlx::QueryBuilder::::new(formatcp!( + "UPDATE `{table}` SET `dispatch_time` = CURRENT_TIMESTAMP() WHERE `id` IN (", + table = COMPRESSION_JOB_TABLE_NAME, + )); + let mut separated_ids = query_builder.separated(", "); + for job_id in chunk { + separated_ids.push_bind(job_id); + } + query_builder.push(");"); + query_builder.build().execute(&mut *tx).await?; + } + tx.commit().await?; + Ok(()) } @@ -332,8 +375,17 @@ impl Coordinator { result } - /// Fetches the pending compression jobs newer than the last polled job and advances the poll - /// cursor. + /// Fetches the pending compression jobs to dispatch. + /// + /// The first fetch after startup returns every [`CompressionJobStatus::Pending`] job so that + /// jobs a previous coordinator instance had already dispatched but not started are + /// re-dispatched. Every subsequent fetch returns only [`CompressionJobStatus::Pending`] jobs + /// whose dispatch time is still not set. + /// + /// # Returns + /// + /// A vector of rows projected from the compression job table on success, each row represents a + /// pending compression job. /// /// # Errors /// @@ -341,30 +393,26 @@ impl Coordinator { /// /// * Forwards [`sqlx::query::QueryAs::fetch_all`]'s return values on failure. async fn fetch_new_job_rows(&mut self) -> Result, Error> { - let query = self.last_polled_job_id.map_or_else( - || { - sqlx::query_as::<_, PendingJobRowProjection>(formatcp!( - "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? ORDER BY `id` \ - ASC;", - table = COMPRESSION_JOB_TABLE_NAME, - )) - .bind(CompressionJobStatus::Pending) - }, - |last_polled_job_id| { - sqlx::query_as::<_, PendingJobRowProjection>(formatcp!( - "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? AND `id` > ? \ - ORDER BY `id` ASC;", - table = COMPRESSION_JOB_TABLE_NAME, - )) - .bind(CompressionJobStatus::Pending) - .bind(last_polled_job_id) - }, + const FIRST_FETCH_QUERY: &str = formatcp!( + "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? ORDER BY `id` ASC;", + table = COMPRESSION_JOB_TABLE_NAME, + ); + const SUBSEQUENT_FETCH_QUERY: &str = formatcp!( + "SELECT `id`, `clp_config` FROM `{table}` WHERE `status` = ? AND `dispatch_time` IS \ + NULL ORDER BY `id` ASC;", + table = COMPRESSION_JOB_TABLE_NAME, ); - let rows = query.fetch_all(&self.db_pool).await?; - if let Some(last_row) = rows.last() { - self.last_polled_job_id = Some(last_row.id); - } + let query = if self.is_first_fetch { + self.is_first_fetch = false; + FIRST_FETCH_QUERY + } else { + SUBSEQUENT_FETCH_QUERY + }; + let rows = sqlx::query_as::<_, PendingJobRowProjection>(query) + .bind(CompressionJobStatus::Pending) + .fetch_all(&self.db_pool) + .await?; Ok(rows) } From 34e88e0ecafea457625b713c511e855fb0372885 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 26 Jul 2026 22:04:05 -0400 Subject: [PATCH 8/9] Apply code review comments. --- .../src/bin/compression_coordinator.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/components/compression-coordinator/src/bin/compression_coordinator.rs b/components/compression-coordinator/src/bin/compression_coordinator.rs index 6da2bf02d..359c33570 100644 --- a/components/compression-coordinator/src/bin/compression_coordinator.rs +++ b/components/compression-coordinator/src/bin/compression_coordinator.rs @@ -84,7 +84,7 @@ async fn main() -> anyhow::Result<()> { // `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 = tokio::select! { + let early_exit_result = tokio::select! { _ = sigterm.recv() => { tracing::info!("Received SIGTERM."); None @@ -102,7 +102,7 @@ async fn main() -> anyhow::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 { + let join_result = if let Some(join_result) = early_exit_result { join_result } else { let termination_timeout = From 3530a638a6761b3b5d3d697d666d689c4c013e60 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Tue, 28 Jul 2026 20:20:26 -0400 Subject: [PATCH 9/9] Add config check at the beginning of the main. --- .../src/bin/compression_coordinator.rs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/components/compression-coordinator/src/bin/compression_coordinator.rs b/components/compression-coordinator/src/bin/compression_coordinator.rs index 359c33570..cf069567d 100644 --- a/components/compression-coordinator/src/bin/compression_coordinator.rs +++ b/components/compression-coordinator/src/bin/compression_coordinator.rs @@ -27,6 +27,21 @@ async fn main() -> anyhow::Result<()> { tracing::error!(error = % e, "Failed to load the configuration file."); })?; + if !matches!(config.logs_input, package::config::LogsInput::S3 { .. }) { + const ERROR_MESSAGE: &str = "The compression coordinator only supports S3 logs inputs."; + tracing::error!(ERROR_MESSAGE); + anyhow::bail!(ERROR_MESSAGE); + } + + if !matches!( + config.archive_output.storage, + package::config::ArchiveOutputStorage::S3 { .. } + ) { + const ERROR_MESSAGE: &str = "The compression coordinator only supports S3 archive outputs."; + tracing::error!(ERROR_MESSAGE); + anyhow::bail!(ERROR_MESSAGE); + } + let credentials = package::credentials::Credentials { database: package::credentials::Database { password: secrecy::SecretString::new(