feat(query-coordinator): Add query job lifecycle handling

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-py-utils/clp_py_utils/initialize-orchestration-db.py

@@ -134,15 +134,18 @@ def main(argv):
                     `id` INT NOT NULL AUTO_INCREMENT,
                     `type` INT NOT NULL,
                     `status` INT NOT NULL DEFAULT '{QueryJobStatus.PENDING}',
+                    `status_msg` VARCHAR(512) NOT NULL DEFAULT '',
                     `creation_time` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
                     `num_tasks` INT NOT NULL DEFAULT '0',
                     `num_tasks_completed` INT NOT NULL DEFAULT '0',
                     `start_time` DATETIME(3) NULL DEFAULT NULL,
                     `duration` FLOAT NULL DEFAULT NULL,
                     `job_config` MEDIUMBLOB NOT NULL,
+                    `spider_id` BIGINT UNSIGNED NULL DEFAULT NULL,
                     PRIMARY KEY (`id`) USING BTREE,
                     INDEX `CREATION_TIME` (`creation_time`) USING BTREE,
-                    INDEX `JOB_STATUS` (`status`) USING BTREE
+                    INDEX `JOB_STATUS` (`status`) USING BTREE,
+                    INDEX `JOB_SPIDER_ID` (`spider_id`) USING BTREE
                 ) ROW_FORMAT=DYNAMIC
                 """
             )

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/compression/compress.rs

@@ -10,10 +10,7 @@ use std::process::Command;
 use std::process::Stdio;
 
 use anyhow::Context;
-use aws_config::BehaviorVersion;
-use aws_sdk_s3::config::ProvideCredentials;
 use clp_rust_utils::aws::AWS_DEFAULT_REGION;
-use clp_rust_utils::clp_config::AwsAuthentication;
 use clp_rust_utils::clp_config::S3Config;
 use clp_rust_utils::clp_config::package::config::ArchiveOutput;
 use clp_rust_utils::clp_config::package::config::ArchiveOutputStorage;
@@ -30,6 +27,8 @@ use non_empty_string::NonEmptyString;
 
 use crate::common::clp_home;
 use crate::common::runtime;
+use crate::task::utils::clp_binary_path;
+use crate::task::utils::s3_credential_env;
 
 /// Compresses the given S3 objects into archives, uploads them to S3, and returns their metadata
 /// for the commit task.
@@ -345,74 +344,6 @@ fn build_s3_logs_list(input_source: &S3InputSource) -> anyhow::Result<String> {
     Ok(list)
 }
 
-/// Resolves the AWS credential env vars clp-s needs to access the S3 objects.
-///
-/// # Returns
-///
-/// The env-var name-value pairs with the following environment variables set:
-///
-/// * `AWS_ACCESS_KEY_ID`
-/// * `AWS_SECRET_ACCESS_KEY`
-/// * `AWS_SESSION_TOKEN` (if any)
-///
-/// # Errors
-///
-/// Returns an error if:
-///
-/// * The default AWS SDK credential provider chain has no provider.
-/// * Forwards [`ProvideCredentials::provide_credentials`]'s return values on failure.
-fn s3_credential_env(
-    runtime: &tokio::runtime::Handle,
-    region: &str,
-    auth: &AwsAuthentication,
-) -> anyhow::Result<Vec<(&'static str, String)>> {
-    /// The env var holding the AWS access key ID.
-    const AWS_ACCESS_KEY_ID_ENV_VAR: &str = "AWS_ACCESS_KEY_ID";
-
-    /// The env var holding the AWS secret access key.
-    const AWS_SECRET_ACCESS_KEY_ENV_VAR: &str = "AWS_SECRET_ACCESS_KEY";
-
-    /// The env var holding the AWS session token.
-    const AWS_SESSION_TOKEN_ENV_VAR: &str = "AWS_SESSION_TOKEN";
-
-    let (access_key_id, secret_access_key, session_token) = match auth {
-        AwsAuthentication::Credentials { credentials } => (
-            credentials.access_key_id.clone(),
-            credentials.secret_access_key.clone(),
-            credentials.session_token.clone(),
-        ),
-        AwsAuthentication::Default => {
-            let sdk_config = runtime.block_on(
-                aws_config::defaults(BehaviorVersion::latest())
-                    .region(aws_sdk_s3::config::Region::new(region.to_string()))
-                    .load(),
-            );
-            let provider = sdk_config
-                .credentials_provider()
-                .context("default AWS SDK credential provider is unavailable")?;
-            let credentials = runtime
-                .block_on(provider.provide_credentials())
-                .context("failed to resolve credentials from the default AWS SDK provider chain")?;
-            (
-                credentials.access_key_id().to_string(),
-                credentials.secret_access_key().to_string(),
-                credentials
-                    .session_token()
-                    .map(std::string::ToString::to_string),
-            )
-        }
-    };
-
-    let mut env = vec![
-        (AWS_ACCESS_KEY_ID_ENV_VAR, access_key_id),
-        (AWS_SECRET_ACCESS_KEY_ENV_VAR, secret_access_key),
-    ];
-    if let Some(session_token) = session_token {
-        env.push((AWS_SESSION_TOKEN_ENV_VAR, session_token));
-    }
-    Ok(env)
-}
-
 /// Parses a single clp-s `--print-archive-stats` stdout line into an [`ArchiveMetadata`].
 ///
 /// NOTE: clp-s emits a superset of [`ArchiveMetadata`]'s fields per line; unknown fields are
@@ -585,15 +516,6 @@ fn build_log_converter_args(output_dir: &Path, inputs_from_path: &Path) -> Vec<O
     ]
 }
 
-/// Resolves the path of a CLP binary under `clp_home`, joining `bin/{binary}`.
-///
-/// # Returns
-///
-/// The path to the named binary under the CLP installation.
-fn clp_binary_path(clp_home: &Path, binary: &str) -> PathBuf {
-    clp_home.join("bin").join(binary)
-}
-
 /// Resolves the S3 config the archives are uploaded to from `config`.
 ///
 /// # Returns
@@ -905,7 +827,6 @@ mod tests {
     use std::path::PathBuf;
 
     use clp_rust_utils::clp_config::AwsAuthentication;
-    use clp_rust_utils::clp_config::AwsCredentials;
     use clp_rust_utils::clp_config::S3Config;
     use clp_rust_utils::clp_config::package::config::ArchiveOutput;
     use clp_rust_utils::clp_config::package::config::ArchiveOutputStorage;
@@ -923,7 +844,6 @@ mod tests {
     use super::build_s3_logs_list;
     use super::create_archive_s3_key;
     use super::parse_archive_stats;
-    use super::s3_credential_env;
 
     #[test]
     fn build_s3_logs_list_default_endpoint() -> anyhow::Result<()> {
@@ -947,28 +867,6 @@ mod tests {
         Ok(())
     }
 
-    #[test]
-    fn s3_credential_env_credentials() {
-        let runtime = tokio::runtime::Runtime::new().expect("failed to create Tokio runtime");
-        let auth = AwsAuthentication::Credentials {
-            credentials: AwsCredentials {
-                access_key_id: "the-access-key".to_string(),
-                secret_access_key: "the-secret-key".to_string(),
-                session_token: Some("the-session-token".to_string()),
-            },
-        };
-
-        assert_eq!(
-            s3_credential_env(runtime.handle(), "us-east-1", &auth)
-                .expect("failed to resolve credentials"),
-            vec![
-                ("AWS_ACCESS_KEY_ID", "the-access-key".to_string()),
-                ("AWS_SECRET_ACCESS_KEY", "the-secret-key".to_string()),
-                ("AWS_SESSION_TOKEN", "the-session-token".to_string()),
-            ]
-        );
-    }
-
     #[test]
     fn parse_archive_stats_ignores_extra_keys() {
         let line = concat!(

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

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

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")
+}