feat(query-coordinator): Implement Spider query graph submission.

Cargo.toml

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

components/clp-rust-utils/src/job_config/search.rs

@@ -1,10 +1,15 @@
+use non_empty_string::NonEmptyString;
 use num_enum::IntoPrimitive;
 use num_enum::TryFromPrimitive;
 use serde::Deserialize;
 use serde::Serialize;
 
 pub const QUERY_JOBS_TABLE_NAME: &str = "query_jobs";
 
+pub type ArchiveId = NonEmptyString;
+
+pub type QueryJobId = i32;
+
 /// Mirror of `job_orchestration.scheduler.job_config.AggregationConfig`. Must be kept in sync.
 #[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
 #[serde(default)]

components/clp-rust-utils/src/task_io.rs

@@ -1 +1,2 @@
 pub mod compression;
+pub mod query;

components/clp-rust-utils/src/task_io/query.rs

@@ -0,0 +1,31 @@
+//! Protocol types exchanged with the Spider (Huntsman) tasks that run CLP query jobs.
+
+use std::num::NonZeroU32;
+
+use non_empty_string::NonEmptyString;
+use serde::Deserialize;
+use serde::Serialize;
+
+/// `clp-s` options for a query job.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct ClpSQueryOption {
+    /// The query string passed positionally to `clp-s`.
+    pub query_string: NonEmptyString,
+
+    /// The per-archive result limit. When absent, the task omits `--max-num-results` and uses the
+    /// `clp-s` default.
+    pub max_num_results: Option<NonZeroU32>,
+
+    /// Inclusive `--tge` bound in Unix epoch milliseconds.
+    pub begin_timestamp_millisecs: Option<i64>,
+
+    /// Inclusive `--tle` bound in Unix epoch milliseconds.
+    pub end_timestamp_millisecs: Option<i64>,
+
+    /// Whether `clp-s` performs a case-insensitive search.
+    pub ignore_case: bool,
+}
+
+/// The output handler that `clp-s` writes a query task's results to.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub enum OutputHandle {}

components/clp-tdl-package/src/lib.rs

@@ -1,4 +1,4 @@
-//! Spider TDL task package `clp`: the CLP compression tasks the Spider task executor loads.
+//! Spider TDL package `clp`, providing CLP compression and query tasks for Spider task executors.
 
 pub mod common;
 mod task;
@@ -28,5 +28,9 @@ fn package_init() -> Result<(), TdlError> {
 spider_tdl::register_tdl_package! {
     package_name: "clp",
     init: package_init,
-    tasks: [task::compression::s3_compress_task, task::compression::commit_task],
+    tasks: [
+        task::compression::s3_compress_task,
+        task::compression::commit_task,
+        task::query::clp_s_search_task,
+    ],
 }

components/clp-tdl-package/src/task/mod.rs

@@ -1,3 +1,4 @@
 //! The task implementations this package registers with Spider.
 
 pub mod compression;
+pub mod query;

components/clp-tdl-package/src/task/query/mod.rs

@@ -0,0 +1,21 @@
+//! The query-task signatures registered with Spider.
+
+use clp_rust_utils::job_config::ArchiveId;
+use clp_rust_utils::job_config::QueryJobId;
+use clp_rust_utils::task_io::query::ClpSQueryOption;
+use clp_rust_utils::task_io::query::OutputHandle;
+use non_empty_string::NonEmptyString;
+use spider_tdl::TaskContext;
+use spider_tdl::task;
+
+#[task(name = "query::clp_s_search")]
+pub(crate) fn clp_s_search_task(
+    _ctx: TaskContext,
+    _query_job_id: QueryJobId,
+    _clp_s_query_option: ClpSQueryOption,
+    _dataset: Option<NonEmptyString>,
+    _archive_id: ArchiveId,
+    _output_handle: OutputHandle,
+) -> Result<(), spider_tdl::TdlError> {
+    todo!("clp-s search task is not implemented")
+}

components/query-coordinator/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "query-coordinator"
+version = { workspace = true }
+edition = { workspace = true }
+
+[dependencies]
+async-trait = { workspace = true }
+clp-rust-utils = { workspace = true }
+non-empty-string = { workspace = true }
+rmp-serde = { workspace = true }
+spider-client = { workspace = true }
+spider-core = { workspace = true }
+thiserror = { workspace = true }
+tracing = { workspace = true }

components/query-coordinator/src/error.rs

@@ -0,0 +1,14 @@
+//! The crate-level error type for the query coordinator.
+
+/// Errors returned by the query coordinator.
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error("spider request failure: {0}")]
+    SpiderClient(#[from] spider_client::error::ClientError),
+
+    #[error("failed to build the query task graph: {0}")]
+    TaskGraph(#[from] spider_core::task::Error),
+
+    #[error("failed to serialize a task input: {0}")]
+    TaskInputSerialization(#[from] rmp_serde::encode::Error),
+}

components/query-coordinator/src/lib.rs

@@ -0,0 +1,6 @@
+//! Coordination for CLP query jobs.
+
+mod error;
+pub mod query_job_submitter;
+
+pub use error::Error;

components/query-coordinator/src/query_job_submitter/mod.rs

@@ -0,0 +1,60 @@
+//! The query-job submission interface.
+
+mod spider;
+
+use async_trait::async_trait;
+use clp_rust_utils::job_config::ArchiveId;
+use clp_rust_utils::job_config::QueryJobId;
+use clp_rust_utils::task_io::query::ClpSQueryOption;
+use clp_rust_utils::task_io::query::OutputHandle;
+use non_empty_string::NonEmptyString;
+use spider_core::task::ExecutionPolicy;
+use spider_core::types::id::JobId;
+use spider_core::types::id::ResourceGroupId;
+
+use crate::Error;
+
+/// Identifies an archive handled by query tasks.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ArchiveMetadata {
+    /// The archive's ID.
+    pub id: ArchiveId,
+
+    /// The archive's dataset, or `None` for the default dataset.
+    pub dataset: Option<NonEmptyString>,
+
+    /// The archive's compressed size in bytes.
+    pub size: u64,
+}
+
+/// Drives CLP query jobs on a Spider (Huntsman) cluster.
+#[async_trait]
+pub trait QueryJobSubmitter: Clone + Send + Sync {
+    /// Builds the query task graph for the given archives and registers it with Spider, without
+    /// starting it.
+    ///
+    /// # Parameters
+    ///
+    /// * `query_job_id` - The unique ID of the CLP query job.
+    /// * `resource_group_id` - The Spider resource group to register the job under.
+    /// * `clp_s_query_option` - `clp-s` query options shared by every task in the job.
+    /// * `output_handle` - The output handle selecting how the query outputs are returned.
+    /// * `archives_to_search` - The archives to search, each represents a query task paired with
+    ///   the task execution policy.
+    ///
+    /// # Returns
+    ///
+    /// The job ID issued by Spider on success.
+    ///
+    /// # Errors
+    ///
+    /// Implementations must document their error conditions.
+    async fn submit_query_job(
+        &self,
+        query_job_id: QueryJobId,
+        resource_group_id: ResourceGroupId,
+        clp_s_query_option: ClpSQueryOption,
+        output_handle: OutputHandle,
+        archives_to_search: Vec<(ArchiveMetadata, ExecutionPolicy)>,
+    ) -> Result<JobId, Error>;
+}

components/query-coordinator/src/query_job_submitter/spider.rs

@@ -0,0 +1,120 @@
+//! [`QueryJobSubmitter`] implementation for [`spider_client::SpiderClient`].
+
+use async_trait::async_trait;
+use clp_rust_utils::job_config::QueryJobId;
+use clp_rust_utils::task_io::query::ClpSQueryOption;
+use clp_rust_utils::task_io::query::OutputHandle;
+use spider_client::SpiderClient;
+use spider_core::task::DataTypeDescriptor;
+use spider_core::task::ExecutionPolicy;
+use spider_core::task::TaskDescriptor;
+use spider_core::task::TaskGraph;
+use spider_core::task::TdlContext;
+use spider_core::task::ValueTypeDescriptor;
+use spider_core::types::id::JobId;
+use spider_core::types::id::ResourceGroupId;
+use spider_core::types::io::TaskInput;
+
+use crate::Error;
+use crate::query_job_submitter::ArchiveMetadata;
+use crate::query_job_submitter::QueryJobSubmitter;
+
+#[async_trait]
+impl QueryJobSubmitter for SpiderClient {
+    /// # Errors
+    ///
+    /// Returns an error if:
+    ///
+    /// * Forwards [`build_query_task_graph`]'s return values on failure.
+    /// * Forwards [`SpiderClient::submit_job`]'s return values on failure.
+    async fn submit_query_job(
+        &self,
+        query_job_id: QueryJobId,
+        resource_group_id: ResourceGroupId,
+        clp_s_query_option: ClpSQueryOption,
+        output_handle: OutputHandle,
+        archives_to_search: Vec<(ArchiveMetadata, ExecutionPolicy)>,
+    ) -> Result<JobId, Error> {
+        let (graph, inputs) = build_query_task_graph(
+            query_job_id,
+            &clp_s_query_option,
+            &output_handle,
+            archives_to_search,
+        )?;
+        let spider_job_id = self.submit_job(resource_group_id, &graph, inputs).await?;
+
+        tracing::info!(
+            query_job_id = % query_job_id,
+            spider_job_id = % spider_job_id,
+            num_tasks = graph.get_num_tasks(),
+            "Submitted query job to Spider.",
+        );
+
+        Ok(spider_job_id)
+    }
+}
+
+/// Builds independent archive-search tasks and their positionally ordered external inputs.
+///
+/// # Returns
+///
+/// A tuple on success, containing:
+///
+/// * The constructed task graph.
+/// * The positionally ordered external inputs.
+///
+/// # Errors
+///
+/// Returns an error if:
+///
+/// * Forwards [`TaskGraph::new`]'s return values on failure.
+/// * Forwards [`ValueTypeDescriptor::struct_from_name`]'s return values on failure.
+/// * Forwards [`TaskGraph::insert_task`]'s return values on failure.
+/// * Forwards [`rmp_serde::to_vec`]'s return values on failure.
+fn build_query_task_graph(
+    query_job_id: QueryJobId,
+    clp_s_query_option: &ClpSQueryOption,
+    output_handle: &OutputHandle,
+    archives_to_search: Vec<(ArchiveMetadata, ExecutionPolicy)>,
+) -> Result<(TaskGraph, Vec<TaskInput>), Error> {
+    // NOTE: Keep these names and the input order in sync with the TDL package definitions.
+    const CLP_TDL_PACKAGE_NAME: &str = "clp";
+    const QUERY_TASK_FUNC: &str = "query::clp_s_search";
+
+    let mut graph = TaskGraph::new(None, None)?;
+
+    let mut inputs = Vec::new();
+    for (archive, execution_policy) in archives_to_search {
+        graph.insert_task(TaskDescriptor {
+            tdl_context: TdlContext {
+                package: CLP_TDL_PACKAGE_NAME.to_owned(),
+                task_func: QUERY_TASK_FUNC.to_owned(),
+            },
+            execution_policy: Some(execution_policy),
+            inputs: vec![
+                DataTypeDescriptor::Value(ValueTypeDescriptor::int32()),
+                DataTypeDescriptor::Value(ValueTypeDescriptor::struct_from_name(
+                    "ClpSQueryOption",
+                )?),
+                DataTypeDescriptor::Value(ValueTypeDescriptor::struct_from_name(
+                    "Option<NonEmptyString>",
+                )?),
+                DataTypeDescriptor::Value(ValueTypeDescriptor::struct_from_name("NonEmptyString")?),
+                DataTypeDescriptor::Value(ValueTypeDescriptor::struct_from_name("OutputHandle")?),
+            ],
+            outputs: vec![],
+            input_sources: None,
+        })?;
+        inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(&query_job_id)?));
+        inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(
+            clp_s_query_option,
+        )?));
+        inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(
+            &archive.dataset,
+        )?));
+        inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(&archive.id)?));
+        inputs.push(TaskInput::ValuePayload(rmp_serde::to_vec(output_handle)?));
+    }
+
+    Ok((graph, inputs))
+}