diff --git a/Cargo.lock b/Cargo.lock index abe78a8ca7..1d90ffb7b1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -910,6 +910,7 @@ checksum = "c3e64b0cc0439b12df2fa678eae89a1c56a529fd067a9115f7827f1fffd22b32" name = "clp-rust-utils" version = "0.12.1-dev" dependencies = [ + "anyhow", "aws-config", "aws-sdk-s3", "aws-sdk-sqs", diff --git a/components/clp-rust-utils/Cargo.toml b/components/clp-rust-utils/Cargo.toml index a4a9a26336..72a9e63eb1 100644 --- a/components/clp-rust-utils/Cargo.toml +++ b/components/clp-rust-utils/Cargo.toml @@ -4,6 +4,7 @@ version = "0.12.1-dev" edition = "2024" [dependencies] +anyhow = "1.0.100" aws-config = "1.8.12" aws-sdk-s3 = "1.121.0" aws-sdk-sqs = "1.92.0" 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 7f4e19cb51..212e020a5f 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 @@ -2,8 +2,12 @@ use non_empty_string::NonEmptyString; use serde::Serialize; use crate::{ - clp_config::S3Config, - job_config::ingestion::JobId as IngestionJobId, + clp_config::{ + AwsAuthentication, + S3Config, + package::{DEFAULT_DATASET_NAME, config::ArchiveOutput}, + }, + job_config::ingestion::{JobId as IngestionJobId, s3::BaseConfig}, s3::S3ObjectMetadataId, }; @@ -64,3 +68,55 @@ pub struct OutputConfig { pub target_segment_size: u64, pub compression_level: u8, } + +impl ClpIoConfig { + /// Creates a CLP IO config for compressing previously ingested S3 object metadata. + /// + /// # Returns + /// + /// A [`ClpIoConfig`] configured to read from the given S3 object metadata IDs and write using + /// the given archive output configuration. + #[must_use] + pub fn from_ingested_s3_object_metadata( + aws_authentication: AwsAuthentication, + archive_output_config: &ArchiveOutput, + ingestion_job_config: &BaseConfig, + ingestion_job_id: IngestionJobId, + s3_object_metadata_ids: Vec, + ) -> Self { + let s3_object_metadata_input_config = S3ObjectMetadataInputConfig { + s3_config: S3Config { + bucket: ingestion_job_config.bucket_name.clone(), + region_code: ingestion_job_config.region.clone(), + key_prefix: ingestion_job_config.key_prefix.clone(), + endpoint_url: ingestion_job_config.endpoint_url.clone(), + aws_authentication, + }, + ingestion_job_id, + s3_object_metadata_ids, + // NOTE: Workaround for #1735 + dataset: Some( + ingestion_job_config + .dataset + .clone() + .unwrap_or_else(|| DEFAULT_DATASET_NAME.clone()), + ), + timestamp_key: ingestion_job_config.timestamp_key.clone(), + unstructured: ingestion_job_config.unstructured, + }; + let output_config = OutputConfig { + target_archive_size: archive_output_config.target_archive_size, + target_dictionaries_size: archive_output_config.target_dictionaries_size, + target_encoded_file_size: archive_output_config.target_encoded_file_size, + target_segment_size: archive_output_config.target_segment_size, + compression_level: archive_output_config.compression_level, + }; + + Self { + input: InputConfig::S3ObjectMetadataInputConfig { + config: s3_object_metadata_input_config, + }, + output: output_config, + } + } +} diff --git a/components/clp-rust-utils/src/job_config/ingestion.rs b/components/clp-rust-utils/src/job_config/ingestion.rs index bc3f726ac2..6c53b9df5d 100644 --- a/components/clp-rust-utils/src/job_config/ingestion.rs +++ b/components/clp-rust-utils/src/job_config/ingestion.rs @@ -176,6 +176,24 @@ pub mod s3 { pub start_after: Option, } + /// Configuration for a S3 prefix ingestion job. + #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] + pub struct S3PrefixConfig { + #[serde(flatten)] + pub base: BaseConfig, + } + + /// Configuration for a one-time S3 keys ingestion job. + #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] + pub struct S3KeysConfig { + #[serde(flatten)] + pub base: BaseConfig, + + /// The explicit object keys to ingest. + #[schema(value_type = Vec)] + pub object_keys: Vec, + } + /// Configuration for buffer behavior. #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] #[serde(default)] diff --git a/components/clp-rust-utils/src/s3.rs b/components/clp-rust-utils/src/s3.rs index 6e9be865ec..69a6bad305 100644 --- a/components/clp-rust-utils/src/s3.rs +++ b/components/clp-rust-utils/src/s3.rs @@ -1,7 +1,9 @@ mod client; +mod url; pub use client::create_new_client; use non_empty_string::NonEmptyString; +pub use url::{NormalizedS3ObjectUrls, ParsedS3Url, normalize_s3_object_urls, parse_s3_url}; /// Represents the unique identifier for an S3 object metadata entry in CLP DB. pub type S3ObjectMetadataId = u64; diff --git a/components/clp-rust-utils/src/s3/url.rs b/components/clp-rust-utils/src/s3/url.rs new file mode 100644 index 0000000000..4d88aaec8c --- /dev/null +++ b/components/clp-rust-utils/src/s3/url.rs @@ -0,0 +1,233 @@ +use std::{collections::BTreeSet, sync::LazyLock}; + +use anyhow::Result; +use non_empty_string::NonEmptyString; +use regex::Regex; + +use crate::types::non_empty_string::ExpectedNonEmpty; + +const AWS_ENDPOINT: &str = "amazonaws.com"; +static HOST_STYLE_URL_REGEX: LazyLock = LazyLock::new(|| { + Regex::new(concat!( + r"^(?Phttps?)://", + r"(?P[a-z0-9.-]+)\.", + r"(?Ps3)\.", + r"(?:(?P[a-z0-9\-]+)\.)?", + r"(?P[A-Za-z0-9.-]+(?::[0-9]+)?)", + r"/(?P[^?]+)(?:\?.*)?$" + )) + .expect("host-style S3 URL regex should be valid") +}); +static PATH_STYLE_URL_REGEX: LazyLock = LazyLock::new(|| { + Regex::new(concat!( + r"^(?Phttps?)://", + r"(?:(?Ps3)\.(?:(?P[a-z0-9\-]+)\.)?)?", + r"(?P[A-Za-z0-9.-]+(?::[0-9]+)?)", + r"/(?P[a-z0-9.-]+)", + r"/(?P[^?]+)(?:\?.*)?$" + )) + .expect("path-style S3 URL regex should be valid") +}); + +/// Parsed components of an S3 object URL. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParsedS3Url { + pub endpoint_url: Option, + pub region: Option, + pub bucket_name: NonEmptyString, + pub key: NonEmptyString, +} + +/// Normalized representation of a set of S3 object URLs. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NormalizedS3ObjectUrls { + pub endpoint_url: Option, + pub region: Option, + pub bucket_name: NonEmptyString, + pub common_key_prefix: NonEmptyString, + pub object_keys: Vec, +} + +/// Parses an S3 object URL into endpoint, region, bucket, and key components. +/// +/// This function supports both host-style and path-style S3 URLs: +/// * For AWS endpoints, `endpoint_url` is normalized to `None`; +/// * For custom S3-compatible endpoints, the returned `endpoint_url` preserves the submitted scheme +/// and host; +/// +/// # Returns +/// +/// The parsed S3 URL on success. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * [`anyhow::Error`] if the URL does not match a supported S3 host-style or path-style format. +/// * [`anyhow::Error`] if the parsed bucket or key is empty. +/// +/// # Panics +/// +/// Panics if the internal host-style or path-style S3 URL regexes do not contain the expected +/// capture groups. +pub fn parse_s3_url(url: &str) -> Result { + let captures = HOST_STYLE_URL_REGEX + .captures(url) + .or_else(|| PATH_STYLE_URL_REGEX.captures(url)) + .ok_or_else(|| anyhow::anyhow!("unsupported URL format: {url}"))?; + + let scheme = captures + .name("scheme") + .expect("S3 URL regex should capture scheme") + .as_str(); + let endpoint = captures + .name("endpoint") + .expect("S3 URL regex should capture endpoint") + .as_str() + .to_string(); + let bucket_name = captures + .name("bucket_name") + .expect("S3 URL regex should capture bucket name") + .as_str(); + let key = NonEmptyString::new( + captures + .name("key_prefix") + .expect("S3 URL regex should capture key") + .as_str() + .to_string(), + ) + .map_err(|_| anyhow::anyhow!("parsed empty object key from URL: {url}"))?; + let region = captures + .name("region_code") + .map(|region| region.as_str().to_string()); + + let s3_prefix = if captures.name("s3").is_some() { + "s3." + } else { + "" + }; + let endpoint_url = if endpoint == AWS_ENDPOINT { + None + } else { + Some(NonEmptyString::from_string(format!( + "{scheme}://{s3_prefix}{endpoint}" + ))) + }; + let region = region.map(NonEmptyString::from_string); + + Ok(ParsedS3Url { + endpoint_url, + region, + bucket_name: NonEmptyString::from_string(bucket_name.to_string()), + key, + }) +} + +/// Parses and validates a set of S3 object URLs for one-time explicit-keys ingestion. +/// +/// * `urls` must be non-empty. +/// * All URLs must share the same endpoint, region, and bucket. +/// * Duplicate object keys are rejected. +/// * The object keys must share a non-empty common prefix. +/// +/// # Returns +/// +/// The normalized URL set on success. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * [`anyhow::Error`] if `urls` is empty. +/// * [`anyhow::Error`] if the URLs do not share the same endpoint. +/// * [`anyhow::Error`] if the URLs do not share the same region. +/// * [`anyhow::Error`] if the URLs do not share the same bucket. +/// * [`anyhow::Error`] if duplicate object keys are found. +/// * [`anyhow::Error`] if the object keys have no non-empty common prefix. +/// * Forwards [`parse_s3_url`]'s return values on failure. +pub fn normalize_s3_object_urls(urls: &[NonEmptyString]) -> Result { + let Some(first_url) = urls.first() else { + return Err(anyhow::anyhow!("no S3 URLs provided")); + }; + + let ParsedS3Url { + endpoint_url, + region, + bucket_name, + key: first_key, + } = parse_s3_url(first_url.as_str())?; + + let mut key_set = BTreeSet::from([first_key]); + for url in &urls[1..] { + let parsed = parse_s3_url(url.as_str())?; + if endpoint_url != parsed.endpoint_url { + return Err(anyhow::anyhow!( + "all S3 URLs must have the same endpoint: found {} and {}", + format_optional_non_empty_string(endpoint_url.as_ref()), + format_optional_non_empty_string(parsed.endpoint_url.as_ref()) + )); + } + if region != parsed.region { + return Err(anyhow::anyhow!( + "all S3 URLs must be in the same region: found {} and {}", + format_optional_non_empty_string(region.as_ref()), + format_optional_non_empty_string(parsed.region.as_ref()) + )); + } + if bucket_name != parsed.bucket_name { + return Err(anyhow::anyhow!( + "all S3 URLs must be in the same bucket: found {} and {}", + bucket_name, + parsed.bucket_name + )); + } + if !key_set.insert(parsed.key.clone()) { + return Err(anyhow::anyhow!("duplicate S3 key found: {}", parsed.key)); + } + } + + let object_keys: Vec = key_set.into_iter().collect(); + let common_key_prefix = compute_common_prefix(&object_keys) + .ok_or_else(|| anyhow::anyhow!("the given S3 URLs have no common prefix"))?; + + Ok(NormalizedS3ObjectUrls { + endpoint_url, + region, + bucket_name, + common_key_prefix, + object_keys, + }) +} + +/// Computes the non-empty common prefix shared by all keys. +/// +/// # Returns +/// +/// The common prefix if it is non-empty. +fn compute_common_prefix(keys: &[NonEmptyString]) -> Option { + let mut common_prefix = keys.first()?.as_str().to_string(); + for key in &keys[1..] { + let mut prefix_len = 0; + for (left_char, right_char) in common_prefix.chars().zip(key.as_str().chars()) { + if left_char != right_char { + break; + } + prefix_len += left_char.len_utf8(); + } + common_prefix.truncate(prefix_len); + if common_prefix.is_empty() { + return None; + } + } + + Some(NonEmptyString::new(common_prefix).expect("common prefix should not be empty")) +} + +/// Formats an optional non-empty string for human-readable validation errors. +/// +/// # Returns +/// +/// `"None"` if the value is absent, or the inner string otherwise. +fn format_optional_non_empty_string(value: Option<&NonEmptyString>) -> String { + value.map_or_else(|| "None".to_string(), std::string::ToString::to_string) +} diff --git a/components/clp-rust-utils/tests/s3_url_test.rs b/components/clp-rust-utils/tests/s3_url_test.rs new file mode 100644 index 0000000000..3b58cfafa6 --- /dev/null +++ b/components/clp-rust-utils/tests/s3_url_test.rs @@ -0,0 +1,122 @@ +use clp_rust_utils::{ + s3::{NormalizedS3ObjectUrls, ParsedS3Url, normalize_s3_object_urls, parse_s3_url}, + types::non_empty_string::ExpectedNonEmpty, +}; +use non_empty_string::NonEmptyString; + +const AWS_HOST_STYLE_URL: &str = "https://my-bucket.s3.us-east-1.amazonaws.com/logs/app.log"; +const AWS_PATH_STYLE_URL: &str = "https://s3.us-east-1.amazonaws.com/my-bucket/logs/app.log"; +const CUSTOM_HOST_STYLE_URL: &str = "http://my-bucket.s3.localhost.localstack.cloud/logs/app.log"; +const NORMALIZED_URL_1: &str = "https://my-bucket.s3.us-east-1.amazonaws.com/logs/app/0001.log"; +const NORMALIZED_URL_2: &str = "https://my-bucket.s3.us-east-1.amazonaws.com/logs/app/0002.log"; + +fn into_non_empty_string_vec(urls: &[&str]) -> Vec { + urls.iter() + .map(|url| NonEmptyString::from_string((*url).to_string())) + .collect() +} + +#[test] +fn test_parse_s3_url_supports_aws_host_and_path_style_urls() { + let expected = ParsedS3Url { + endpoint_url: None, + region: Some(NonEmptyString::from_static_str("us-east-1")), + bucket_name: NonEmptyString::from_static_str("my-bucket"), + key: NonEmptyString::from_static_str("logs/app.log"), + }; + + assert_eq!( + parse_s3_url(AWS_HOST_STYLE_URL).expect("AWS host-style URL should parse"), + expected + ); + assert_eq!( + parse_s3_url(AWS_PATH_STYLE_URL).expect("AWS path-style URL should parse"), + expected + ); +} + +#[test] +fn test_parse_s3_url_preserves_custom_region_and_strips_query_string() { + let parsed = parse_s3_url("https://my-bucket.s3.nyc3.example.com/logs/app.log?versionId=123") + .expect("custom-region URL with query string should parse"); + assert_eq!( + parsed, + ParsedS3Url { + endpoint_url: Some(NonEmptyString::from_static_str("https://s3.example.com")), + region: Some(NonEmptyString::from_static_str("nyc3")), + bucket_name: NonEmptyString::from_static_str("my-bucket"), + key: NonEmptyString::from_static_str("logs/app.log"), + } + ); +} + +#[test] +fn test_parse_s3_url_preserves_custom_host_style_endpoint_shape() { + let parsed = parse_s3_url(CUSTOM_HOST_STYLE_URL).expect("custom host-style URL should parse"); + assert_eq!( + parsed, + ParsedS3Url { + endpoint_url: Some(NonEmptyString::from_static_str( + "http://s3.localstack.cloud" + )), + region: Some(NonEmptyString::from_static_str("localhost")), + bucket_name: NonEmptyString::from_static_str("my-bucket"), + key: NonEmptyString::from_static_str("logs/app.log"), + } + ); +} + +#[test] +fn test_normalize_s3_object_urls_returns_sorted_keys_and_common_prefix() { + let normalized = normalize_s3_object_urls(&into_non_empty_string_vec(&[ + NORMALIZED_URL_2, + NORMALIZED_URL_1, + ])) + .expect("URL normalization should succeed"); + assert_eq!( + normalized, + NormalizedS3ObjectUrls { + endpoint_url: None, + region: Some(NonEmptyString::from_static_str("us-east-1")), + bucket_name: NonEmptyString::from_static_str("my-bucket"), + common_key_prefix: NonEmptyString::from_static_str("logs/app/000"), + object_keys: vec![ + NonEmptyString::from_static_str("logs/app/0001.log"), + NonEmptyString::from_static_str("logs/app/0002.log"), + ], + } + ); +} + +#[test] +fn test_normalize_s3_object_urls_rejects_duplicate_keys() { + let err = normalize_s3_object_urls(&into_non_empty_string_vec(&[ + AWS_HOST_STYLE_URL, + AWS_HOST_STYLE_URL, + ])) + .expect_err("duplicate keys should fail"); + assert_eq!(err.to_string(), "duplicate S3 key found: logs/app.log"); +} + +#[test] +fn test_normalize_s3_object_urls_rejects_incompatible_url_sets() { + let mixed_region = normalize_s3_object_urls(&into_non_empty_string_vec(&[ + "https://my-bucket.s3.us-east-1.amazonaws.com/logs/app/1.log", + "https://my-bucket.s3.us-west-2.amazonaws.com/logs/app/2.log", + ])) + .expect_err("mixed region should fail"); + assert_eq!( + mixed_region.to_string(), + "all S3 URLs must be in the same region: found us-east-1 and us-west-2" + ); + + let empty_prefix = normalize_s3_object_urls(&into_non_empty_string_vec(&[ + "https://my-bucket.s3.amazonaws.com/a.log", + "https://my-bucket.s3.amazonaws.com/b.log", + ])) + .expect_err("empty common prefix should fail"); + assert_eq!( + empty_prefix.to_string(), + "the given S3 URLs have no common prefix" + ); +} diff --git a/components/log-ingestor/src/compression/compression_job_submitter.rs b/components/log-ingestor/src/compression/compression_job_submitter.rs index b4f8f6287c..f04f8c32eb 100644 --- a/components/log-ingestor/src/compression/compression_job_submitter.rs +++ b/components/log-ingestor/src/compression/compression_job_submitter.rs @@ -1,18 +1,12 @@ use anyhow::Result; use async_trait::async_trait; use clp_rust_utils::{ - clp_config::{ - AwsAuthentication, - S3Config, - package::{DEFAULT_DATASET_NAME, config::ArchiveOutput}, - }, + clp_config::{AwsAuthentication, package::config::ArchiveOutput}, job_config::{ ClpIoConfig, CompressionJobId, CompressionJobStatus, InputConfig, - OutputConfig, - S3ObjectMetadataInputConfig, ingestion::s3::BaseConfig, }, s3::S3ObjectMetadataId, @@ -63,40 +57,13 @@ impl CompressionJobSubmitter { archive_output_config: &ArchiveOutput, ingestion_job_config: &BaseConfig, ) -> Self { - let ingestion_job_id = clp_compression_state.get_ingestion_job_id(); - let s3_object_metadata_input_config = S3ObjectMetadataInputConfig { - s3_config: S3Config { - bucket: ingestion_job_config.bucket_name.clone(), - region_code: ingestion_job_config.region.clone(), - key_prefix: ingestion_job_config.key_prefix.clone(), - endpoint_url: ingestion_job_config.endpoint_url.clone(), - aws_authentication, - }, - ingestion_job_id, - s3_object_metadata_ids: vec![], - // NOTE: Workaround for #1735 - dataset: Some( - ingestion_job_config - .dataset - .clone() - .unwrap_or_else(|| DEFAULT_DATASET_NAME.clone()), - ), - timestamp_key: ingestion_job_config.timestamp_key.clone(), - unstructured: ingestion_job_config.unstructured, - }; - let output_config = OutputConfig { - target_archive_size: archive_output_config.target_archive_size, - target_dictionaries_size: archive_output_config.target_dictionaries_size, - target_encoded_file_size: archive_output_config.target_encoded_file_size, - target_segment_size: archive_output_config.target_segment_size, - compression_level: archive_output_config.compression_level, - }; - let io_config_template = ClpIoConfig { - input: InputConfig::S3ObjectMetadataInputConfig { - config: s3_object_metadata_input_config, - }, - output: output_config, - }; + let io_config_template = ClpIoConfig::from_ingested_s3_object_metadata( + aws_authentication, + archive_output_config, + ingestion_job_config, + clp_compression_state.get_ingestion_job_id(), + Vec::new(), + ); Self { state: clp_compression_state, io_config_template, diff --git a/components/log-ingestor/src/ingestion_job.rs b/components/log-ingestor/src/ingestion_job.rs index e31f4c533c..e7cd400f68 100644 --- a/components/log-ingestor/src/ingestion_job.rs +++ b/components/log-ingestor/src/ingestion_job.rs @@ -1,13 +1,24 @@ +mod s3_keys; +mod s3_prefix; mod s3_scanner; mod scan; mod sqs_listener; mod state; pub use clp_rust_utils::job_config::ingestion::JobId as IngestionJobId; -pub use s3_scanner::*; +pub use s3_keys::S3KeysIngestion; +pub use s3_prefix::S3PrefixIngestion; +pub use s3_scanner::S3Scanner; pub use scan::scan_prefix; -pub use sqs_listener::*; -pub use state::*; +pub use sqs_listener::SqsListener; +pub use state::{ + FinalizeOutcome, + IngestionJobState, + OneTimeIngestionState, + S3ScannerState, + SqsListenerState, + ZeroFaultToleranceIngestionJobState, +}; /// Enum for different types of ingestion jobs. /// diff --git a/components/log-ingestor/src/ingestion_job/s3_keys.rs b/components/log-ingestor/src/ingestion_job/s3_keys.rs new file mode 100644 index 0000000000..b0da675e7d --- /dev/null +++ b/components/log-ingestor/src/ingestion_job/s3_keys.rs @@ -0,0 +1,186 @@ +use std::{ + collections::HashSet, + sync::{Arc, Mutex}, +}; + +use anyhow::Result; +use aws_sdk_s3::Client; +use clp_rust_utils::{job_config::ingestion::s3::S3KeysConfig, s3::ObjectMetadata}; + +use crate::{ + aws_client_manager::AwsClientManagerType, + ingestion_job::{FinalizeOutcome, IngestionJobId, OneTimeIngestionState, scan_prefix}, +}; + +/// Represents a one-time S3 explicit-keys ingestion task that scans a given prefix under the +/// bucket once and finalizes the metadata of the requested objects. +/// +/// # Type Parameters +/// +/// * [`S3ClientManager`]: The type of the AWS S3 client manager. +/// * [`State`]: The type that implements [`OneTimeIngestionState`] for managing one-time ingestion +/// states. +struct Task, State: OneTimeIngestionState> { + s3_client_manager: S3ClientManager, + config: S3KeysConfig, + state: State, +} + +impl, State: OneTimeIngestionState> + Task +{ + /// Runs the one-time explicit-keys ingestion task to scan the given bucket and finalize the + /// metadata of the requested objects. + /// + /// # Returns + /// + /// `Ok(())` on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`OneTimeIngestionState::start`]'s return values on failure. + /// * Forwards [`AwsClientManagerType::get`]'s return values on failure. + /// * Forwards [`scan_prefix`]'s return values on failure. + /// * [`anyhow::Error`] if one or more requested keys are missing after scanning the configured + /// prefix to exhaustion. + /// * Forwards [`OneTimeIngestionState::finalize`]'s return values on failure. + async fn run(self) -> Result<()> { + self.state.start().await?; + + let client = self.s3_client_manager.get().await?; + let found_objects = Arc::new(Mutex::new(Vec::new())); + let collected_objects = Arc::clone(&found_objects); + let remaining_keys = Arc::new(Mutex::new( + self.config + .object_keys + .iter() + .cloned() + .collect::>(), + )); + let pending_keys = Arc::clone(&remaining_keys); + scan_prefix( + &client, + &self.config.base.bucket_name, + &self.config.base.key_prefix, + None, + async move |page: Vec| -> Result<_> { + let last_key = page + .last() + .expect("scanned page should not be empty") + .key + .clone(); + let mut matched_objects = Vec::new(); + let mut remaining_keys = pending_keys + .lock() + .expect("remaining keys mutex should not be poisoned"); + for object in page { + if remaining_keys.remove(&object.key) { + matched_objects.push(object); + } + } + let should_continue_scanning = !remaining_keys.is_empty(); + drop(remaining_keys); + if !matched_objects.is_empty() { + collected_objects + .lock() + .expect("collected objects mutex should not be poisoned") + .extend(matched_objects); + } + Ok((should_continue_scanning, last_key)) + }, + ) + .await?; + + let remaining_keys = Arc::into_inner(remaining_keys) + .expect("remaining keys should have no remaining references") + .into_inner() + .expect("remaining keys mutex should not be poisoned"); + if !remaining_keys.is_empty() { + let mut remaining_keys: Vec = remaining_keys + .into_iter() + .map(|key| key.to_string()) + .collect(); + remaining_keys.sort(); + return Err(anyhow::anyhow!( + "failed to find requested S3 object keys: {}", + remaining_keys.join(", ") + )); + } + + let all_objects = Arc::into_inner(found_objects) + .expect("collected objects should have no remaining references") + .into_inner() + .expect("collected objects mutex should not be poisoned"); + let outcome = self.state.finalize(all_objects).await?; + match outcome { + FinalizeOutcome::Finished => { + tracing::info!("S3 keys ingestion finalized."); + } + FinalizeOutcome::Cancelled => { + tracing::info!("S3 keys ingestion finalization cancelled."); + } + } + + Ok(()) + } +} + +/// Represents a one-time S3 explicit-keys ingestion job that manages the lifecycle of a one-time +/// S3 explicit-keys ingestion task. +pub struct S3KeysIngestion { + id: IngestionJobId, +} + +impl S3KeysIngestion { + /// Creates and spawns a new [`S3KeysIngestion`] backed by a [`Task`]. + /// + /// This function spawns a [`Task`]. The spawned task will scan the configured S3 bucket and + /// prefix once, collect the metadata for the requested object keys, and finalize them using + /// the provided state. + /// + /// # Type Parameters + /// + /// * [`S3ClientManager`]: The type of the AWS S3 client manager. + /// * [`State`]: The type that implements [`OneTimeIngestionState`] for managing one-time + /// ingestion states. + /// + /// # Returns + /// + /// A newly created instance of [`S3KeysIngestion`]. + #[must_use] + pub fn spawn, State: OneTimeIngestionState>( + id: IngestionJobId, + s3_client_manager: S3ClientManager, + config: S3KeysConfig, + state: State, + ) -> Self { + let task = Task { + s3_client_manager, + config, + state: state.clone(), + }; + tokio::spawn(async move { + if let Err(err) = task.run().await { + tracing::error!( + error = ? err, + job_id = ? id, + "S3 keys ingestion task execution failed." + ); + state + .fail(format!("S3 keys ingestion task execution failed: {err}")) + .await; + } + }); + Self { id } + } + + /// # Returns + /// + /// The ID of this S3 keys ingestion job. + #[must_use] + pub const fn get_id(&self) -> IngestionJobId { + self.id + } +} diff --git a/components/log-ingestor/src/ingestion_job/s3_prefix.rs b/components/log-ingestor/src/ingestion_job/s3_prefix.rs new file mode 100644 index 0000000000..1cb261afe1 --- /dev/null +++ b/components/log-ingestor/src/ingestion_job/s3_prefix.rs @@ -0,0 +1,140 @@ +use std::sync::{Arc, Mutex}; + +use anyhow::Result; +use aws_sdk_s3::Client; +use clp_rust_utils::{job_config::ingestion::s3::S3PrefixConfig, s3::ObjectMetadata}; + +use crate::{ + aws_client_manager::AwsClientManagerType, + ingestion_job::{FinalizeOutcome, IngestionJobId, OneTimeIngestionState, scan_prefix}, +}; + +/// Represents a one-time S3 prefix scan task that collects all object metadata under the configured +/// prefix and finalizes them in a single pass. +/// +/// # Type Parameters +/// +/// * `S3ClientManager`: The type of the AWS S3 client manager. +/// * `State`: The type that implements [`OneTimeIngestionState`] for managing one-time ingestion +/// state. +struct Task, State: OneTimeIngestionState> { + s3_client_manager: S3ClientManager, + config: S3PrefixConfig, + state: State, +} + +impl, State: OneTimeIngestionState> + Task +{ + /// Runs the one-time prefix scan task by transitioning the job to the running status and + /// scanning the configured S3 prefix to exhaustion. + /// + /// # NOTE + /// + /// Persistence is delegated to the provided [`OneTimeIngestionState`] implementation. + /// + /// # Returns + /// + /// `Ok(())` on success, including the case where the finalize step was cancelled. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`OneTimeIngestionState::start`]'s return values on failure. + /// * Forwards [`AwsClientManagerType::get`]'s return values on failure. + /// * Forwards [`scan_prefix`]'s return values on failure. + /// * Forwards [`OneTimeIngestionState::finalize`]'s return values on failure. + async fn run(self) -> Result<()> { + self.state.start().await?; + + let client = self.s3_client_manager.get().await?; + let all_objects = Arc::new(Mutex::new(Vec::new())); + let collected_objects = Arc::clone(&all_objects); + scan_prefix( + &client, + &self.config.base.bucket_name, + &self.config.base.key_prefix, + None, + async move |page: Vec| -> Result<_> { + let last_key = page.last().expect("`page` should not be empty").key.clone(); + collected_objects + .lock() + .expect("collected objects mutex should not be poisoned") + .extend(page); + Ok((true, last_key)) + }, + ) + .await?; + + let all_objects = Arc::into_inner(all_objects) + .expect("collected objects should have no remaining references") + .into_inner() + .expect("collected objects mutex should not be poisoned"); + let outcome = self.state.finalize(all_objects).await?; + match outcome { + FinalizeOutcome::Finished => { + tracing::info!("S3 prefix ingestion finalized."); + } + FinalizeOutcome::Cancelled => { + tracing::info!("S3 prefix ingestion finalization cancelled."); + } + } + + Ok(()) + } +} + +/// Handle for a one-time S3 prefix ingestion job that scans the configured prefix once, collects +/// all object metadata, and submits a single compression job. +pub struct S3PrefixIngestion { + id: IngestionJobId, +} + +impl S3PrefixIngestion { + /// Creates and spawns a detached one-time S3 prefix ingestion job backed by a [`Task`]. + /// + /// # Type Parameters + /// + /// * `S3ClientManager`: The type of the AWS S3 client manager. + /// * `State`: The type that implements [`OneTimeIngestionState`] for managing one-time + /// ingestion state. + /// + /// # Returns + /// + /// A newly created instance of [`S3PrefixIngestion`]. + #[must_use] + pub fn spawn, State: OneTimeIngestionState>( + id: IngestionJobId, + s3_client_manager: S3ClientManager, + config: S3PrefixConfig, + state: State, + ) -> Self { + let task = Task { + s3_client_manager, + config, + state: state.clone(), + }; + tokio::spawn(async move { + if let Err(err) = task.run().await { + tracing::error!( + error = ? err, + job_id = ? id, + "S3 prefix ingestion task execution failed." + ); + state + .fail(format!("S3 prefix ingestion task execution failed: {err}")) + .await; + } + }); + Self { id } + } + + /// # Returns + /// + /// The ID of this S3 prefix ingestion job. + #[must_use] + pub const fn get_id(&self) -> IngestionJobId { + self.id + } +} diff --git a/components/log-ingestor/src/ingestion_job/state.rs b/components/log-ingestor/src/ingestion_job/state.rs index 43aaf172fb..4dae5aef99 100644 --- a/components/log-ingestor/src/ingestion_job/state.rs +++ b/components/log-ingestor/src/ingestion_job/state.rs @@ -88,6 +88,37 @@ pub trait S3ScannerState: Send + Sync + Clone + 'static { ) -> anyhow::Result<()>; } +/// Outcome of a one-time ingestion job's `finalize` step. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FinalizeOutcome { + /// The `finalize` step completed successfully. + Finished, + /// The `finalize` step was cancelled. + Cancelled, +} + +/// An abstract layer for managing one-time ingestion job states. +#[async_trait] +pub trait OneTimeIngestionState: IngestionJobState { + /// Persists the collected object metadata, submits a single compression job, and + /// transitions the ingestion job to a terminal state. + /// + /// # Parameters + /// + /// * `objects`: The collected object metadata to persist. + /// + /// # Returns + /// + /// [`FinalizeOutcome::Finished`] if all steps have been completed successfully, or + /// [`FinalizeOutcome::Cancelled`] if the job was cancelled before all steps were completed. + /// + /// # Errors + /// + /// Implementations **must document** the specific error variants they may return and the + /// conditions under which those errors occur. + async fn finalize(&self, objects: Vec) -> anyhow::Result; +} + /// An ingestion job state implementation that has no fault-tolerance. #[derive(Clone)] pub struct ZeroFaultToleranceIngestionJobState { @@ -143,3 +174,16 @@ impl S3ScannerState for ZeroFaultToleranceIngestionJobState { Ok(()) } } + +#[async_trait] +impl OneTimeIngestionState for ZeroFaultToleranceIngestionJobState { + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`mpsc::Sender::send`]'s return values on failure. + async fn finalize(&self, objects: Vec) -> anyhow::Result { + self.sender.send(objects).await?; + Ok(FinalizeOutcome::Finished) + } +} diff --git a/components/log-ingestor/src/ingestion_job_manager.rs b/components/log-ingestor/src/ingestion_job_manager.rs index 00639daf2c..b1b680cc6c 100644 --- a/components/log-ingestor/src/ingestion_job_manager.rs +++ b/components/log-ingestor/src/ingestion_job_manager.rs @@ -11,7 +11,14 @@ use clp_rust_utils::{ credentials::Credentials as ClpCredentials, }, }, - job_config::ingestion::s3::{ConfigError, S3IngestionJobConfig, ValidatedSqsListenerConfig}, + job_config::ingestion::s3::{ + BaseConfig, + ConfigError, + S3IngestionJobConfig, + S3KeysConfig, + S3PrefixConfig, + ValidatedSqsListenerConfig, + }, }; use serde::Serialize; use tokio::sync::Mutex; @@ -19,7 +26,14 @@ use utoipa::ToSchema; use crate::{ aws_client_manager::{S3ClientWrapper, SqsClientWrapper}, - ingestion_job::{IngestionJob, IngestionJobId, IngestionJobState, S3Scanner}, + ingestion_job::{ + IngestionJob, + IngestionJobId, + IngestionJobState, + S3KeysIngestion, + S3PrefixIngestion, + S3Scanner, + }, }; /// Errors for ingestion job manager operations. @@ -52,6 +66,70 @@ pub enum TerminalStatus { Failed, } +/// Configuration for one-time ingestion jobs. +#[derive(Debug, Clone)] +pub enum OneTimeIngestionJobConfig { + /// Configuration for a S3 prefix ingestion job. + S3Prefix(S3PrefixConfig), + + /// Configuration for a S3 explicit-keys ingestion job. + S3Keys(S3KeysConfig), +} + +impl OneTimeIngestionJobConfig { + #[must_use] + pub const fn job_type(&self) -> ClpIngestionJobType { + match self { + Self::S3Prefix(_) => ClpIngestionJobType::S3Prefix, + Self::S3Keys(_) => ClpIngestionJobType::S3Keys, + } + } + + #[must_use] + pub const fn base_config(&self) -> &BaseConfig { + match self { + Self::S3Prefix(config) => &config.base, + Self::S3Keys(config) => &config.base, + } + } + + /// Spawns the detached worker for this one-time ingestion job. + fn spawn(self, s3_client_manager: S3ClientWrapper, state: ClpOneTimeIngestionState) { + let job_id = state.get_job_id(); + match self { + Self::S3Prefix(config) => { + let _detached = S3PrefixIngestion::spawn(job_id, s3_client_manager, config, state); + } + Self::S3Keys(config) => { + let _detached = S3KeysIngestion::spawn(job_id, s3_client_manager, config, state); + } + } + } +} + +/// The context for recovering and spawning a one-time ingestion job. +pub struct OneTimeIngestionJobContext { + config: OneTimeIngestionJobConfig, + state: ClpOneTimeIngestionState, +} + +impl OneTimeIngestionJobContext { + #[must_use] + pub const fn new(config: OneTimeIngestionJobConfig, state: ClpOneTimeIngestionState) -> Self { + Self { config, state } + } + + #[must_use] + pub const fn get_job_id(&self) -> IngestionJobId { + self.state.get_job_id() + } + + #[must_use] + pub fn into_parts(self) -> (OneTimeIngestionJobConfig, ClpOneTimeIngestionState) { + (self.config, self.state) + } +} + /// An async-safe state for creating and managing ingestion jobs. #[derive(Clone)] pub struct IngestionJobManagerState { @@ -101,10 +179,16 @@ impl IngestionJobManagerState { let ingestion_job_manager = Self { inner }; // Recover log-ingestor - let (unfinished_compression_jobs, recoverable_ingestion_jobs, inactive_ingestion_jobs) = ( + let ( + unfinished_compression_jobs, + recoverable_ingestion_jobs, + inactive_ingestion_jobs, + recoverable_one_time_jobs, + ) = ( recovery_context.unfinished_compression_jobs, recovery_context.recoverable_ingestion_jobs, recovery_context.inactive_ingestion_jobs, + recovery_context.recoverable_one_time_jobs, ); try_recover_waiting_coroutines_for_unfinished_compression_jobs(unfinished_compression_jobs); try_recover_ingestion_job_instances( @@ -113,6 +197,7 @@ impl IngestionJobManagerState { ) .await; try_recover_inactive_ingestion_jobs(inactive_ingestion_jobs).await; + try_recover_one_time_jobs(ingestion_job_manager.clone(), recoverable_one_time_jobs).await; Ok(ingestion_job_manager) } @@ -128,12 +213,14 @@ impl IngestionJobManagerState { /// /// Returns an error if: /// + /// * [`Error::MissingRegionCode`] if no region is provided while using the default S3 endpoint. /// * Forwards [`Self::create_s3_ingestion_job_instance`]'s return values on failure. /// * Forwards [`ClpDbIngestionConnector::create_ingestion_job`]'s return values on failure. pub async fn register_and_create_s3_ingestion_job( &self, config: S3IngestionJobConfig, ) -> Result { + Self::validate_region_for_base_config(config.as_base_config())?; let ingestion_job_context = self .inner .clp_db_ingestion_connector @@ -154,6 +241,31 @@ impl IngestionJobManagerState { } } + /// Registers a one-time ingestion job in CLP DB. + /// + /// # Returns + /// + /// The newly created one-time ingestion job context on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`Error::MissingRegionCode`] if no region is provided while using the default S3 endpoint. + /// * Forwards [`ClpDbIngestionConnector::create_one_time_ingestion_job_context`]'s return + /// values on failure. + pub async fn register_one_time_ingestion_job( + &self, + config: OneTimeIngestionJobConfig, + ) -> Result { + Self::validate_region_for_base_config(config.base_config())?; + self.inner + .clp_db_ingestion_connector + .create_one_time_ingestion_job_context(config) + .await + .map_err(Into::into) + } + /// Shuts down and removes an ingestion job instance by its ID. /// /// # NOTE @@ -161,6 +273,9 @@ impl IngestionJobManagerState { /// This method only removes the running job instance. It does not remove the job from CLP DB. /// The job status and stats will still be accessible after calling this method. /// + /// For one-time jobs, this method sets the job status to [`ClpIngestionJobStatus::Paused`] + /// instead of [`ClpIngestionJobStatus::Failed`]. + /// /// # Returns /// /// The terminal status of the job indicating whether it has stopped with an error. @@ -171,7 +286,9 @@ impl IngestionJobManagerState { /// /// * [`Error::JobNotFound`] if the given job ID does not exist. /// * Forwards [`ClpIngestionJobContext::shutdown`]'s return values on failure. - /// * Forwards [`ClpDbIngestionConnector::get_job_status`]' return values on failure. + /// * Forwards [`ClpDbIngestionConnector::get_job_status`]'s return values on failure. + /// * Forwards [`ClpDbIngestionConnector::get_job_type`]'s return values on failure. + /// * Forwards [`ClpDbIngestionConnector::try_pause`]'s return values on failure. /// * Forwards [`ClpDbIngestionConnector::try_fail`]'s return values on failure. pub async fn shutdown_and_remove_job_instance( &self, @@ -199,21 +316,75 @@ impl IngestionJobManagerState { else { return Err(Error::JobNotFound(job_id)); }; - + // The job has already finished. Keep the operation idempotent by not overwriting the + // terminal status. if ClpIngestionJobStatus::Finished == status { - // The job has already finished. Keep the operation idempotent by not overwriting the - // status. - Ok(TerminalStatus::Finished) - } else { - self.inner - .clp_db_ingestion_connector - .try_fail( - job_id, - "Ingestion job instance not found on shutdown.".to_string(), - ) - .await?; - Ok(TerminalStatus::Failed) + return Ok(TerminalStatus::Finished); + } + if ClpIngestionJobStatus::Paused == status { + return Ok(TerminalStatus::Failed); + } + + let job_type = self + .inner + .clp_db_ingestion_connector + .get_job_type(job_id) + .await?; + match job_type { + ClpIngestionJobType::S3Prefix | ClpIngestionJobType::S3Keys => { + self.inner + .clp_db_ingestion_connector + .try_pause(job_id, None) + .await?; + } + ClpIngestionJobType::S3Scanner | ClpIngestionJobType::SqsListener => { + self.inner + .clp_db_ingestion_connector + .try_fail( + job_id, + "ingestion job instance not found on shutdown".to_string(), + ) + .await?; + } } + Ok(TerminalStatus::Failed) + } + + /// Spawns a detached one-time ingestion task for the given one-time job context. + /// + /// # NOTE + /// + /// Prefix-conflict detection against running long-running jobs is not performed. + pub async fn spawn_one_time_ingestion( + &self, + one_time_ingestion_job_context: OneTimeIngestionJobContext, + ) { + let (config, state) = one_time_ingestion_job_context.into_parts(); + let base_config = config.base_config(); + + let s3_client_manager = S3ClientWrapper::create( + base_config.region.as_ref(), + base_config.endpoint_url.as_ref(), + &self.inner.aws_authentication, + ) + .await; + + config.spawn(s3_client_manager, state); + } + + /// Validates region requirements for S3-backed ingestion jobs. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`Error::MissingRegionCode`] if `base_config.region` is `None` and + /// `base_config.endpoint_url` is `None`. + const fn validate_region_for_base_config(base_config: &BaseConfig) -> Result<(), Error> { + if base_config.region.is_none() && base_config.endpoint_url.is_none() { + return Err(Error::MissingRegionCode); + } + Ok(()) } /// Creates a new S3 ingestion job instance and adds it to the job table with key prefix @@ -476,3 +647,30 @@ async fn try_refill_ingestion_buffer(ingestion_job_context: &ClpIngestionJobCont ); } } + +/// Recovers one-time ingestion jobs by re-spawning them from scratch. +/// +/// # NOTE +/// +/// This function is best-effort. Failures spawning individual jobs are logged and skipped. +async fn try_recover_one_time_jobs( + ingestion_job_manager: IngestionJobManagerState, + recoverable_one_time_jobs: Vec, +) { + let num_jobs = recoverable_one_time_jobs.len(); + if num_jobs == 0 { + return; + } + + tracing::info!("Recovering {num_jobs} one-time ingestion job(s)."); + + for one_time_ingestion_job_context in recoverable_one_time_jobs { + let job_id = one_time_ingestion_job_context.get_job_id(); + tracing::info!(job_id = ? job_id, "Recovering one-time ingestion job."); + ingestion_job_manager + .spawn_one_time_ingestion(one_time_ingestion_job_context) + .await; + } + + tracing::info!("One-time ingestion job recovery completed. Total recovered jobs: {num_jobs}."); +} diff --git a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs index 2153d3c35f..2e6af74579 100644 --- a/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs +++ b/components/log-ingestor/src/ingestion_job_manager/clp_ingestion.rs @@ -16,7 +16,13 @@ use clp_rust_utils::{ CompressionJobId, CompressionJobStatus, InputConfig, - ingestion::s3::{S3IngestionJobConfig, S3ScannerConfig}, + ingestion::s3::{ + BaseConfig, + S3IngestionJobConfig, + S3KeysConfig, + S3PrefixConfig, + S3ScannerConfig, + }, }, s3::{ObjectMetadata, S3ObjectMetadataId}, }; @@ -35,8 +41,19 @@ use crate::{ Listener, wait_for_compression_job_completion_and_update_metadata, }, - ingestion_job::{IngestionJobState, S3ScannerState, SqsListenerState}, - ingestion_job_manager::{IngestionJobId, TerminalStatus}, + ingestion_job::{ + FinalizeOutcome, + IngestionJobState, + OneTimeIngestionState, + S3ScannerState, + SqsListenerState, + }, + ingestion_job_manager::{ + IngestionJobId, + OneTimeIngestionJobConfig, + OneTimeIngestionJobContext, + TerminalStatus, + }, }; /// A bundle of objects for log-ingestor to recovery from a restart. @@ -53,6 +70,14 @@ pub struct LogIngestorRecoveryContext { /// have active ingestion job instances, but for any already-ingested but still buffered object /// metadata, there should be a compression listener to re-submit them for compression. pub inactive_ingestion_jobs: Vec, + + /// A vector of recoverable one-time ingestion jobs that are re-spawned from scratch. + pub recoverable_one_time_jobs: Vec, +} + +enum RecoverableJob { + LongRunning(S3IngestionJobConfig), + OneTime(OneTimeIngestionJobConfig), } /// A bundle of objects to manage an already-submitted CLP compression job. @@ -68,8 +93,8 @@ impl ClpCompressionJobContext { self.compression_job_id } - /// Creates a detached coroutine to wait for the compression job to complete, and updates the - /// status of all ingested + /// Creates a detached coroutine to wait for the compression job to complete and updates the + /// status of all ingested object metadata rows linked to that job. pub fn detach_and_wait_for_completion_and_update_metadata(self) { tokio::spawn(async move { wait_for_compression_job_completion_and_update_metadata( @@ -215,12 +240,13 @@ impl ClpDbIngestionConnector { }; let unfinished_compression_jobs = connector.get_unfinished_compression_jobs().await?; - let (recoverable_ingestion_jobs, inactive_ingestion_jobs) = + let (recoverable_ingestion_jobs, inactive_ingestion_jobs, recoverable_one_time_jobs) = connector.load_ingestion_jobs().await?; let recovery_context = LogIngestorRecoveryContext { unfinished_compression_jobs, recoverable_ingestion_jobs, inactive_ingestion_jobs, + recoverable_one_time_jobs, }; Ok((connector, recovery_context)) } @@ -324,6 +350,36 @@ impl ClpDbIngestionConnector { .await } + /// Attempts to pause an ingestion job with the given ID. + /// + /// # NOTE + /// + /// This method doesn't perform any validation on the job ID before attempting to pause it, + /// which means the job ID may not exist in the CLP DB. + /// + /// # Returns + /// + /// `Ok(())` on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`update_job_status`]'s return values on failure. + pub async fn try_pause( + &self, + job_id: IngestionJobId, + error_msg: Option, + ) -> anyhow::Result<()> { + update_job_status( + self.db_pool.clone(), + job_id, + ClpIngestionJobStatus::Paused, + error_msg.as_deref(), + ) + .await + } + /// Retrieves all unfinished compression jobs that are previously submitted. /// /// # Returns @@ -450,6 +506,22 @@ impl ClpDbIngestionConnector { }) } + /// Creates a one-time ingestion job context for the given job ID and config. + fn create_one_time_ingestion_job_context_from_parts( + &self, + job_id: IngestionJobId, + config: OneTimeIngestionJobConfig, + ) -> OneTimeIngestionJobContext { + let state = ClpOneTimeIngestionState::new( + job_id, + self.db_pool.clone(), + self.aws_authentication.clone(), + self.archive_output_config.clone(), + config.base_config().clone(), + ); + OneTimeIngestionJobContext::new(config, state) + } + /// Loads ingestion contexts for all ingestion jobs from CLP DB. /// /// # NOTE @@ -461,9 +533,10 @@ impl ClpDbIngestionConnector { /// /// A tuple on success, containing: /// - /// * A vector of [`ClpIngestionJobContext`] instances representing all recoverable ingestion - /// jobs + /// * A vector of [`ClpIngestionJobContext`] instances representing all recoverable long-running + /// ingestion jobs. /// * A vector of [`ClpIngestionJobContext`] instances representing all inactive ingestion jobs. + /// * A vector of [`OneTimeIngestionJobContext`] values for recoverable one-time jobs. /// /// # Errors /// @@ -473,94 +546,76 @@ impl ClpDbIngestionConnector { /// metadata. async fn load_ingestion_jobs( &self, - ) -> anyhow::Result<(Vec, Vec)> { - // Load recoverable ingestion jobs. + ) -> anyhow::Result<( + Vec, + Vec, + Vec, + )> { let mut recoverable_ingestion_jobs: Vec = Vec::new(); - let requested_jod_id_and_config: Vec<(IngestionJobId, String)> = sqlx::query_as(formatcp!( - "SELECT `id`, `config` FROM `{table}` WHERE `status` = ?;", - table = INGESTION_JOB_TABLE_NAME, - )) - .bind(ClpIngestionJobStatus::Requested) - .fetch_all(&self.db_pool) - .await - .inspect_err(|e| { - tracing::error!( - error = ? e, - "Failed to fetch requested ingestion jobs from CLP DB during service start." - ); - })?; - for (job_id, config_str) in requested_jod_id_and_config { - let Some(config) = self.load_config(job_id, config_str).await else { - tracing::error!( - job_id = ? job_id, - "Failed to load config for a requested ingestion job at service start. The job \ - will be skipped." - ); + let mut recoverable_one_time_jobs: Vec = Vec::new(); + + let requested_jobs = self + .fetch_jobs_with_status(ClpIngestionJobStatus::Requested) + .await?; + for (job_id, config_str, job_type) in requested_jobs { + let Some(job) = self + .parse_recoverable_job(job_id, config_str, job_type) + .await + else { continue; }; - self.try_add_ingestion_job_context(job_id, config, &mut recoverable_ingestion_jobs); + self.recover_recoverable_job( + job_id, + job, + false, + &mut recoverable_ingestion_jobs, + &mut recoverable_one_time_jobs, + ) + .await; } - let running_jod_id_and_config: Vec<(IngestionJobId, String)> = sqlx::query_as(formatcp!( - "SELECT `id`, `config` FROM `{table}` WHERE `status` = ?;", - table = INGESTION_JOB_TABLE_NAME, - )) - .bind(ClpIngestionJobStatus::Running) - .fetch_all(&self.db_pool) - .await - .inspect_err(|e| { - tracing::error!( - error = ? e, - "Failed to fetch running ingestion jobs from CLP DB during service start." - ); - })?; - - for (job_id, config_str) in running_jod_id_and_config { - let Some(config) = self.load_config(job_id, config_str).await else { + let running_jobs = self + .fetch_jobs_with_status(ClpIngestionJobStatus::Running) + .await?; + for (job_id, config_str, job_type) in running_jobs { + let Some(job) = self + .parse_recoverable_job(job_id, config_str, job_type) + .await + else { continue; }; - let config = match config { - S3IngestionJobConfig::S3Scanner(config) => { - let updated_config = match self.update_start_after(job_id, config).await { - Ok(updated_config) => updated_config, - Err(e) => { - if let Err(e) = self.try_fail(job_id, e.to_string()).await { - tracing::error!( - error = ? e, - job_id = ? job_id, - "Failed to update ingestion job status to `failed` for a S3 \ - scanner ingestion job with invalid last ingested key \ - during service start." - ); - } - continue; - } - }; - S3IngestionJobConfig::S3Scanner(updated_config) - } - other @ S3IngestionJobConfig::SqsListener(_) => other, - }; - self.try_add_ingestion_job_context(job_id, config, &mut recoverable_ingestion_jobs); + self.recover_recoverable_job( + job_id, + job, + true, + &mut recoverable_ingestion_jobs, + &mut recoverable_one_time_jobs, + ) + .await; } - // Load inactive ingestion jobs. - let inactive_jod_id_and_config: Vec<(IngestionJobId, String)> = sqlx::query_as(formatcp!( - "SELECT `id`, `config` FROM `{table}` WHERE `status` NOT in (?, ?);", - table = INGESTION_JOB_TABLE_NAME, - )) - .bind(ClpIngestionJobStatus::Requested) - .bind(ClpIngestionJobStatus::Running) - .fetch_all(&self.db_pool) - .await - .inspect_err(|e| { - tracing::error!( - error = ? e, - "Failed to fetch inactive ingestion jobs from CLP DB during service start." - ); - })?; + let inactive_jobs: Vec<(IngestionJobId, String, ClpIngestionJobType)> = + sqlx::query_as(formatcp!( + "SELECT `id`, `config`, `job_type` FROM `{table}` WHERE `status` NOT in (?, ?);", + table = INGESTION_JOB_TABLE_NAME, + )) + .bind(ClpIngestionJobStatus::Requested) + .bind(ClpIngestionJobStatus::Running) + .fetch_all(&self.db_pool) + .await + .inspect_err(|e| { + tracing::error!( + error = ? e, + "Failed to fetch inactive ingestion jobs from CLP DB during service start." + ); + })?; let mut inactive_ingestion_jobs: Vec = - Vec::with_capacity(inactive_jod_id_and_config.len()); - for (job_id, config_str) in inactive_jod_id_and_config { + Vec::with_capacity(inactive_jobs.len()); + for (job_id, config_str, job_type) in inactive_jobs { + // One-time jobs don't have buffered objects that need resubmission. + if job_type.is_one_time() { + continue; + } let Some(config) = self.load_config(job_id, config_str).await else { tracing::error!( job_id = ? job_id, @@ -572,7 +627,165 @@ impl ClpDbIngestionConnector { self.try_add_ingestion_job_context(job_id, config, &mut inactive_ingestion_jobs); } - Ok((recoverable_ingestion_jobs, inactive_ingestion_jobs)) + Ok(( + recoverable_ingestion_jobs, + inactive_ingestion_jobs, + recoverable_one_time_jobs, + )) + } + + /// Fetches the IDs, configs, and types of all ingestion jobs in the given status. + /// + /// # Returns + /// + /// The matching `(job_id, config_str, job_type)` triples on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`sqlx::query::Query::fetch_all`]'s return values on failure. + async fn fetch_jobs_with_status( + &self, + status: ClpIngestionJobStatus, + ) -> anyhow::Result> { + const QUERY: &str = formatcp!( + "SELECT `id`, `config`, `job_type` FROM `{table}` WHERE `status` = ?;", + table = INGESTION_JOB_TABLE_NAME, + ); + sqlx::query_as(QUERY) + .bind(status) + .fetch_all(&self.db_pool) + .await + .map_err(|e| { + tracing::error!( + error = ? e, + status = ? status, + "Failed to fetch ingestion jobs from CLP DB during service start." + ); + e.into() + }) + } + + /// Parses a recoverable ingestion job from its persisted job type and config string. + /// + /// # Returns + /// + /// The parsed recoverable job on success, or `None` if the config cannot be deserialized. On + /// failure, the errors will be logged and the ingestion job will be marked as failed in CLP + /// DB. + async fn parse_recoverable_job( + &self, + job_id: IngestionJobId, + config_str: String, + job_type: ClpIngestionJobType, + ) -> Option { + match job_type { + ClpIngestionJobType::S3Prefix | ClpIngestionJobType::S3Keys => self + .load_one_time_config(job_id, config_str, job_type) + .await + .map(RecoverableJob::OneTime), + ClpIngestionJobType::S3Scanner | ClpIngestionJobType::SqsListener => self + .load_config(job_id, config_str) + .await + .map(RecoverableJob::LongRunning), + } + } + + /// Adds a parsed recoverable job to the appropriate recovery list. + /// + /// When `update_start_after` is true and the job is an S3 scanner, refreshes `start_after` + /// from the last persisted ingested key before recovery. + async fn recover_recoverable_job( + &self, + job_id: IngestionJobId, + job: RecoverableJob, + update_start_after: bool, + recoverable_ingestion_jobs: &mut Vec, + recoverable_one_time_jobs: &mut Vec, + ) { + match job { + RecoverableJob::OneTime(config) => { + recoverable_one_time_jobs + .push(self.create_one_time_ingestion_job_context_from_parts(job_id, config)); + } + RecoverableJob::LongRunning(config) => { + let config = if update_start_after { + match config { + S3IngestionJobConfig::S3Scanner(scanner_config) => { + match self.update_start_after(job_id, scanner_config).await { + Ok(updated_config) => { + S3IngestionJobConfig::S3Scanner(updated_config) + } + Err(e) => { + let status_msg = e.to_string(); + if let Err(e) = self.try_fail(job_id, status_msg).await { + tracing::error!( + error = ? e, + job_id = ? job_id, + "Failed to update ingestion job status to `failed` \ + for a S3 scanner ingestion job with invalid last \ + ingested key during service start." + ); + } + return; + } + } + } + other @ S3IngestionJobConfig::SqsListener(_) => other, + } + } else { + config + }; + self.try_add_ingestion_job_context(job_id, config, recoverable_ingestion_jobs); + } + } + } + + /// Loads the configuration of a one-time ingestion job with the given ID from CLP DB. + /// + /// # Returns + /// + /// The one-time ingestion job configuration on success, or `None` if the config cannot be + /// deserialized. On failure, the errors will be logged and the ingestion job will be marked as + /// failed in CLP DB. + async fn load_one_time_config( + &self, + job_id: IngestionJobId, + config_str: String, + job_type: ClpIngestionJobType, + ) -> Option { + let loaded_config = match job_type { + ClpIngestionJobType::S3Prefix => serde_json::from_str::(&config_str) + .map(OneTimeIngestionJobConfig::S3Prefix), + ClpIngestionJobType::S3Keys => serde_json::from_str::(&config_str) + .map(OneTimeIngestionJobConfig::S3Keys), + ClpIngestionJobType::S3Scanner | ClpIngestionJobType::SqsListener => { + panic!("non-one-time job type passed to load_one_time_config: {job_type}") + } + }; + match loaded_config { + Ok(config) => Some(config), + Err(e) => { + tracing::error!( + error = ? e, + job_id = ? job_id, + job_type = ? job_type, + "Failed to parse one-time ingestion job config from CLP DB during service \ + start. The job will be skipped." + ); + let status_msg = format!("failed to parse one-time ingestion job config: {e}"); + if let Err(e) = self.try_fail(job_id, status_msg).await { + tracing::error!( + error = ? e, + job_id = ? job_id, + "Failed to update job status to `failed` for a one-time job with invalid \ + config during service start." + ); + } + None + } + } } /// Loads the configuration of an ingestion job with the given ID from CLP DB. @@ -596,10 +809,8 @@ impl ClpDbIngestionConnector { "Failed to parse ingestion job config from CLP DB during service start. \ The job will be skipped." ); - if let Err(e) = self - .try_fail(job_id, format!("Failed to parse ingestion job config: {e}")) - .await - { + let status_msg = format!("failed to parse ingestion job config: {e}"); + if let Err(e) = self.try_fail(job_id, status_msg).await { tracing::error!( error = ? e, job_id = ? job_id, @@ -680,6 +891,77 @@ impl ClpDbIngestionConnector { ), } } + + /// Creates and registers a one-time ingestion job context in the database. + /// + /// # Returns + /// + /// The newly created one-time ingestion job context on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`serde_json::to_string`]'s return values on failure. + /// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. + pub async fn create_one_time_ingestion_job_context( + &self, + config: OneTimeIngestionJobConfig, + ) -> anyhow::Result { + const INSERT_QUERY: &str = formatcp!( + r"INSERT INTO `{table}` (`config`, `job_type`) VALUES (?, ?);", + table = INGESTION_JOB_TABLE_NAME, + ); + + let config_json = match &config { + OneTimeIngestionJobConfig::S3Prefix(prefix_config) => { + serde_json::to_string(prefix_config)? + } + OneTimeIngestionJobConfig::S3Keys(keys_config) => serde_json::to_string(keys_config)?, + }; + let result = sqlx::query(INSERT_QUERY) + .bind(config_json) + .bind(config.job_type()) + .execute(&self.db_pool) + .await?; + + Ok(self.create_one_time_ingestion_job_context_from_parts(result.last_insert_id(), config)) + } + + /// Retrieves the job type for the given ingestion job ID. + /// + /// # Returns + /// + /// The job type on success. + /// + /// # Errors + /// + /// Returns an error if: + /// + /// * [`anyhow::Error`] if the job is not found. + /// * Forwards [`sqlx::query::Query::fetch_one`]'s return values on failure. + pub async fn get_job_type( + &self, + job_id: IngestionJobId, + ) -> anyhow::Result { + const QUERY: &str = formatcp!( + r"SELECT `job_type` FROM `{table}` WHERE `id` = ?;", + table = INGESTION_JOB_TABLE_NAME, + ); + + let job_type: ClpIngestionJobType = sqlx::query_scalar(QUERY) + .bind(job_id) + .fetch_one(&self.db_pool) + .await + .map_err(|e| match e { + sqlx::Error::RowNotFound => { + anyhow::anyhow!("ingestion job with ID {job_id} not found") + } + other => other.into(), + })?; + + Ok(job_type) + } } /// A CLP-DB-backed implementation of ingestion job state. This state ingests object metadata into @@ -763,7 +1045,7 @@ impl ClpIngestionState { /// /// Returns an error if: /// - /// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. + /// * Forwards [`insert_object_metadata_rows`]'s return values on failure. /// /// # Panics /// @@ -775,52 +1057,15 @@ impl ClpIngestionState { tx: &mut sqlx::Transaction<'_, sqlx::MySql>, objects: Vec, ) -> anyhow::Result> { - const BASE_INGESTION_QUERY: &str = formatcp!( - r"INSERT INTO `{table}` (`bucket`, `key`, `size`, `ingestion_job_id`) VALUES ", - table = INGESTED_S3_OBJECT_METADATA_TABLE_NAME, - ); - - const CHUNK_SIZE: usize = 10000; - - assert!( - !objects.is_empty(), - "Cannot build S3 object metadata ingestion query with empty objects" - ); - - let mut buffer_entries = Vec::with_capacity(objects.len()); - - // Ingest object metadata - // NOTE: MySQL has a maximum placeholder limit of 65535. We need to batch the ingestion to - // avoid hitting this limit. If the number of placeholders per insert changes, we may need - // to adjust the chunk size accordingly. - for chunk in objects.chunks(CHUNK_SIZE) { - let query_string = format!( - "{}{}", - BASE_INGESTION_QUERY, - std::iter::repeat_n("(?, ?, ?, ?)", chunk.len()) - .collect::>() - .join(", ") - ); - - let mut query = sqlx::query(&query_string); - for object in chunk { - query = query - .bind(object.bucket.as_str()) - .bind(object.key.as_str()) - .bind(object.size) - .bind(self.job_id); - } - - let first_metadata_id: S3ObjectMetadataId = - query.execute(&mut **tx).await?.last_insert_id(); - for (next_metadata_id, object) in (first_metadata_id..).zip(chunk.iter()) { - buffer_entries.push(CompressionBufferEntry { - id: next_metadata_id, - size: object.size, - }); - } - } - + let metadata_ids = insert_object_metadata_rows(tx, self.job_id, &objects).await?; + let buffer_entries = metadata_ids + .into_iter() + .zip(objects.iter()) + .map(|(id, object)| CompressionBufferEntry { + id, + size: object.size, + }) + .collect(); Ok(buffer_entries) } @@ -1032,6 +1277,189 @@ impl SqsListenerState for ClpIngestionState { } } +/// A CLP-DB-backed state for one-time ingestion jobs. +#[derive(Clone)] +pub struct ClpOneTimeIngestionState { + job_id: IngestionJobId, + db_pool: MySqlPool, + aws_authentication: AwsAuthentication, + archive_output_config: ArchiveOutput, + base_config: BaseConfig, +} + +impl ClpOneTimeIngestionState { + /// Factory function. + /// + /// # Returns + /// + /// A newly created [`ClpOneTimeIngestionState`]. + #[must_use] + pub const fn new( + job_id: IngestionJobId, + db_pool: MySqlPool, + aws_authentication: AwsAuthentication, + archive_output_config: ArchiveOutput, + base_config: BaseConfig, + ) -> Self { + Self { + job_id, + db_pool, + aws_authentication, + archive_output_config, + base_config, + } + } + + /// # Returns + /// + /// The ID of the underlying ingestion job. + #[must_use] + pub const fn get_job_id(&self) -> IngestionJobId { + self.job_id + } +} + +#[async_trait] +impl IngestionJobState for ClpOneTimeIngestionState { + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`update_job_status`]'s return values on failure. + async fn start(&self) -> anyhow::Result<()> { + update_job_status( + self.db_pool.clone(), + self.job_id, + ClpIngestionJobStatus::Running, + None, + ) + .await + } + + /// # Errors + /// + /// Returns an error if: + /// + /// * Forwards [`update_job_status`]'s return values on failure. + async fn end(&self) -> anyhow::Result<()> { + update_job_status( + self.db_pool.clone(), + self.job_id, + ClpIngestionJobStatus::Finished, + None, + ) + .await + } + + async fn fail(&self, msg: String) { + match update_job_status( + self.db_pool.clone(), + self.job_id, + ClpIngestionJobStatus::Failed, + Some(&msg), + ) + .await + { + Ok(()) => {} + Err(err) => { + tracing::error!( + error = ? err, + status_msg = ? msg, + job_id = ? self.job_id, + "Failed to update job status to `failed`." + ); + } + } + } +} + +#[async_trait] +impl OneTimeIngestionState for ClpOneTimeIngestionState { + /// # Errors + /// + /// Returns an error if: + /// + /// * [`anyhow::Error`] if the underlying ingestion job is not in + /// [`ClpIngestionJobStatus::Running`] or [`ClpIngestionJobStatus::Paused`] status. + /// * Forwards [`sqlx::Pool::begin`]'s return values on failure. + /// * Forwards [`sqlx::query::Query::fetch_one`]'s return values on failure. + /// * Forwards [`insert_object_metadata_rows`]'s return values on failure. + /// * Forwards [`insert_compression_job_and_link_metadata`]'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 finalize(&self, objects: Vec) -> anyhow::Result { + const SELECT_STATUS_QUERY: &str = formatcp!( + r"SELECT `status` FROM `{table}` WHERE `id` = ? FOR UPDATE;", + table = INGESTION_JOB_TABLE_NAME, + ); + const UPDATE_STATUS_QUERY: &str = formatcp!( + r"UPDATE `{table}` SET `status` = ? WHERE `id` = ?;", + table = INGESTION_JOB_TABLE_NAME, + ); + + let mut tx = self.db_pool.begin().await?; + + let status: ClpIngestionJobStatus = sqlx::query_scalar(SELECT_STATUS_QUERY) + .bind(self.job_id) + .fetch_one(&mut *tx) + .await?; + match status { + ClpIngestionJobStatus::Paused => { + tx.commit().await?; + return Ok(FinalizeOutcome::Cancelled); + } + ClpIngestionJobStatus::Running => {} + other => { + return Err(anyhow::anyhow!( + "unexpected job status for finalization: {other}" + )); + } + } + + if objects.is_empty() { + sqlx::query(UPDATE_STATUS_QUERY) + .bind(ClpIngestionJobStatus::Finished) + .bind(self.job_id) + .execute(&mut *tx) + .await?; + tx.commit().await?; + return Ok(FinalizeOutcome::Finished); + } + + let object_metadata_ids = + insert_object_metadata_rows(&mut tx, self.job_id, &objects).await?; + let io_config = ClpIoConfig::from_ingested_s3_object_metadata( + self.aws_authentication.clone(), + &self.archive_output_config, + &self.base_config, + self.job_id, + object_metadata_ids.clone(), + ); + let compression_job_id = + insert_compression_job_and_link_metadata(&mut tx, &io_config, &object_metadata_ids) + .await?; + + sqlx::query(UPDATE_STATUS_QUERY) + .bind(ClpIngestionJobStatus::Finished) + .bind(self.job_id) + .execute(&mut *tx) + .await?; + tx.commit().await?; + + let compression_state = ClpCompressionState { + ingestion_job_id: self.job_id, + db_pool: self.db_pool.clone(), + }; + tokio::spawn(wait_for_compression_job_completion_and_update_metadata( + compression_state, + compression_job_id, + object_metadata_ids.len(), + )); + + Ok(FinalizeOutcome::Finished) + } +} + /// A CLP-DB-backed implementation of compression job state. /// /// This state manages the submission of compression jobs to CLP and the corresponding updates to @@ -1063,25 +1491,13 @@ impl ClpCompressionState { /// /// Returns an error if: /// - /// * [`anyhow::Error`] if the submitted compression job ID overflows. - /// * [`anyhow::Error`] if one or more object metadata rows fail to be updated in the DB. - /// * Forwards [`clp_rust_utils::serde::BrotliMsgpack::serialize`]'s return values on failure. - /// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. /// * Forwards [`sqlx::Pool::begin`]'s return values on failure. + /// * Forwards [`insert_compression_job_and_link_metadata`]'s return values on failure. /// * Forwards [`sqlx::Transaction::commit`]'s return values on failure. - /// - /// # Panics - /// - /// Panics if the length of a chunk cannot be converted to `u64`. pub async fn submit_for_compression( &self, io_config: ClpIoConfig, ) -> anyhow::Result { - const COMPRESSION_JOB_SUBMISSION_QUERY: &str = formatcp!( - r"INSERT INTO {table} (`clp_config`) VALUES (?)", - table = CLP_COMPRESSION_JOB_TABLE_NAME - ); - let object_metadata_ids: &[S3ObjectMetadataId] = match &io_config.input { InputConfig::S3ObjectMetadataInputConfig { config } => &config.s3_object_metadata_ids, InputConfig::S3InputConfig { .. } => { @@ -1090,54 +1506,11 @@ impl ClpCompressionState { }; let mut tx = self.db_pool.begin().await?; - - // Submit compression job - let result = sqlx::query(COMPRESSION_JOB_SUBMISSION_QUERY) - .bind(clp_rust_utils::serde::BrotliMsgpack::serialize(&io_config)?) - .execute(&mut *tx) - .await?; let compression_job_id = - CompressionJobId::try_from(result.last_insert_id()).map_err(|_| { - anyhow::anyhow!("The retrieved ID overflows: {}", result.last_insert_id()) - })?; - - // Update compression job ID for ingested objects. - // NOTE: We batch the update to avoid hitting the maximum placeholder limit of MySQL. The - // batch size is chosen to be 10000, which is conservative enough to avoid hitting the limit - // while also minimizing the number of batches for typical use cases. If the number of - // placeholders per update changes, we may need to adjust the batch size accordingly. - for chunk in object_metadata_ids.chunks(10000) { - let mut query_builder = sqlx::QueryBuilder::::new(formatcp!( - r"UPDATE `{table}` ", - table = INGESTED_S3_OBJECT_METADATA_TABLE_NAME, - )); - query_builder - .push("SET `compression_job_id` = ") - .push_bind(compression_job_id); - query_builder - .push(", `status` = ") - .push_bind(IngestedS3ObjectMetadataStatus::Submitted); - query_builder.push(" WHERE `id` IN ("); - let mut separated_ids = query_builder.separated(", "); - for id in chunk { - separated_ids.push_bind(id); - } - query_builder.push(")"); - query_builder - .push(" AND `status` = ") - .push_bind(IngestedS3ObjectMetadataStatus::Buffered); - - let result = query_builder.build().execute(&mut *tx).await?; - if result.rows_affected() - != u64::try_from(chunk.len()).expect("size conversion should always succeed") - { - return Err(anyhow::anyhow!( - "Failed to update compression job ID for some objects." - )); - } - } - + insert_compression_job_and_link_metadata(&mut tx, &io_config, object_metadata_ids) + .await?; tx.commit().await?; + Ok(compression_job_id) } @@ -1325,6 +1698,42 @@ impl MySqlEnumFormat for ClpIngestionJobStatus {} impl_sqlx_type!(ClpIngestionJobStatus => str); +/// Enum for CLP ingestion job types. +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + sqlx::Encode, + sqlx::Decode, + EnumIter, + AsRefStr, + Display, + EnumString, +)] +#[sqlx(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] +pub enum ClpIngestionJobType { + S3Scanner, + SqsListener, + S3Prefix, + S3Keys, +} + +impl ClpIngestionJobType { + /// Returns whether this job type runs as a single bounded task (not a long-running listener + /// or scanner loop). + #[must_use] + pub const fn is_one_time(self) -> bool { + matches!(self, Self::S3Prefix | Self::S3Keys) + } +} + +impl MySqlEnumFormat for ClpIngestionJobType {} + +impl_sqlx_type!(ClpIngestionJobType => str); + /// Enum for CLP ingestion S3 object metadata status. #[derive( Debug, @@ -1364,6 +1773,7 @@ fn ingestion_job_table_creation_query() -> String { CREATE TABLE IF NOT EXISTS `{table}` ( `id` BIGINT unsigned NOT NULL AUTO_INCREMENT, `config` TEXT NOT NULL, + `job_type` {job_type_enum} NOT NULL DEFAULT '{default_job_type}', `status` {status_enum} NOT NULL DEFAULT '{default_status}', `status_msg` TEXT NULL DEFAULT NULL, `num_files_compressed` BIGINT unsigned NOT NULL DEFAULT '0', @@ -1373,11 +1783,153 @@ CREATE TABLE IF NOT EXISTS `{table}` ( PRIMARY KEY (`id`) );", table = INGESTION_JOB_TABLE_NAME, + job_type_enum = ClpIngestionJobType::format_as_sql_enum(), + default_job_type = ClpIngestionJobType::S3Scanner, status_enum = ClpIngestionJobStatus::format_as_sql_enum(), default_status = ClpIngestionJobStatus::Requested, ) } +/// Inserts S3 object metadata rows into the database and returns the generated IDs. +/// +/// # Returns +/// +/// A vector of [`S3ObjectMetadataId`] values for the inserted rows on success. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. +/// +/// # Panics +/// +/// Panics if `objects` is empty, as it cannot build a valid ingestion query in that case. +async fn insert_object_metadata_rows( + tx: &mut sqlx::Transaction<'_, sqlx::MySql>, + job_id: IngestionJobId, + objects: &[ObjectMetadata], +) -> anyhow::Result> { + const BASE_INGESTION_QUERY: &str = formatcp!( + r"INSERT INTO `{table}` (`bucket`, `key`, `size`, `ingestion_job_id`) VALUES ", + table = INGESTED_S3_OBJECT_METADATA_TABLE_NAME, + ); + const CHUNK_SIZE: usize = 10000; + + assert!( + !objects.is_empty(), + "cannot build S3 object metadata ingestion query with empty objects" + ); + + let mut metadata_ids = Vec::with_capacity(objects.len()); + + // NOTE: MySQL has a maximum placeholder limit of 65535. We need to batch the ingestion to + // avoid hitting this limit. If the number of placeholders per insert changes, we may need + // to adjust the chunk size accordingly. + for chunk in objects.chunks(CHUNK_SIZE) { + let query_string = format!( + "{}{}", + BASE_INGESTION_QUERY, + std::iter::repeat_n("(?, ?, ?, ?)", chunk.len()) + .collect::>() + .join(", ") + ); + + let mut query = sqlx::query(&query_string); + for object in chunk { + query = query + .bind(object.bucket.as_str()) + .bind(object.key.as_str()) + .bind(object.size) + .bind(job_id); + } + + let first_metadata_id: S3ObjectMetadataId = + query.execute(&mut **tx).await?.last_insert_id(); + for next_metadata_id in first_metadata_id..(first_metadata_id + chunk.len() as u64) { + metadata_ids.push(next_metadata_id); + } + } + + Ok(metadata_ids) +} + +/// Inserts a compression job row and links the provided object metadata IDs to it. +/// +/// # Returns +/// +/// The ID of the inserted compression job on success. +/// +/// # Errors +/// +/// Returns an error if: +/// +/// * [`anyhow::Error`] if the submitted compression job ID overflows. +/// * [`anyhow::Error`] if one or more object metadata rows fail to be updated in the DB. +/// * Forwards [`clp_rust_utils::serde::BrotliMsgpack::serialize`]'s return values on failure. +/// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. +/// +/// # Panics +/// +/// Panics if the length of a chunk cannot be converted to `u64`. +async fn insert_compression_job_and_link_metadata( + tx: &mut sqlx::Transaction<'_, sqlx::MySql>, + io_config: &ClpIoConfig, + object_metadata_ids: &[S3ObjectMetadataId], +) -> anyhow::Result { + const COMPRESSION_JOB_SUBMISSION_QUERY: &str = formatcp!( + r"INSERT INTO {table} (`clp_config`) VALUES (?)", + table = CLP_COMPRESSION_JOB_TABLE_NAME + ); + const CHUNK_SIZE: usize = 10000; + + // Submit compression job + let result = sqlx::query(COMPRESSION_JOB_SUBMISSION_QUERY) + .bind(clp_rust_utils::serde::BrotliMsgpack::serialize(io_config)?) + .execute(&mut **tx) + .await?; + let compression_job_id = CompressionJobId::try_from(result.last_insert_id()) + .map_err(|_| anyhow::anyhow!("the retrieved ID overflows: {}", result.last_insert_id()))?; + + // Update compression job ID for ingested objects. + // NOTE: We batch the update to avoid hitting the maximum placeholder limit of MySQL. The + // batch size is chosen to be 10000, which is conservative enough to avoid hitting the limit + // while also minimizing the number of batches for typical use cases. If the number of + // placeholders per update changes, we may need to adjust the batch size accordingly. + for chunk in object_metadata_ids.chunks(CHUNK_SIZE) { + let mut query_builder = sqlx::QueryBuilder::::new(formatcp!( + r"UPDATE `{table}` ", + table = INGESTED_S3_OBJECT_METADATA_TABLE_NAME, + )); + query_builder + .push("SET `compression_job_id` = ") + .push_bind(compression_job_id); + query_builder + .push(", `status` = ") + .push_bind(IngestedS3ObjectMetadataStatus::Submitted); + query_builder.push(" WHERE `id` IN ("); + let mut separated_ids = query_builder.separated(", "); + for id in chunk { + separated_ids.push_bind(id); + } + query_builder.push(")"); + query_builder + .push(" AND `status` = ") + .push_bind(IngestedS3ObjectMetadataStatus::Buffered); + + let result = query_builder.build().execute(&mut **tx).await?; + if result.rows_affected() + != u64::try_from(chunk.len()).expect("size conversion should always succeed") + { + return Err(anyhow::anyhow!( + "failed to update compression job ID for some objects" + )); + } + } + + Ok(compression_job_id) +} + /// The query to create the table for S3 scanner job state tracking. #[must_use] fn ingestion_job_s3_scanner_state_table_creation_query() -> String { @@ -1517,6 +2069,14 @@ mod tests { ); } + #[test] + fn test_clp_ingestion_job_type_enum() { + assert_eq!( + ClpIngestionJobType::format_as_sql_enum(), + "ENUM('s3_scanner', 'sqs_listener', 's3_prefix', 's3_keys')" + ); + } + #[test] fn test_ingested_s3_object_metadata_status_enum() { assert_eq!( diff --git a/components/log-ingestor/src/routes.rs b/components/log-ingestor/src/routes.rs index 5293a78a68..bd0d8d5d11 100644 --- a/components/log-ingestor/src/routes.rs +++ b/components/log-ingestor/src/routes.rs @@ -7,11 +7,17 @@ use axum::{ response::IntoResponse, routing::get, }; -use clp_rust_utils::job_config::ingestion::s3::{ - S3IngestionJobConfig, - S3ScannerConfig, - SqsListenerConfig, +use clp_rust_utils::{ + job_config::ingestion::s3::{ + S3IngestionJobConfig, + S3KeysConfig, + S3PrefixConfig, + S3ScannerConfig, + SqsListenerConfig, + }, + s3::normalize_s3_object_urls, }; +use non_empty_string::NonEmptyString; use serde::Serialize; use tower_http::cors::{Any, CorsLayer}; use utoipa::{OpenApi, ToSchema}; @@ -22,6 +28,7 @@ use crate::{ ingestion_job_manager::{ Error as IngestionJobManagerError, IngestionJobManagerState, + OneTimeIngestionJobConfig, TerminalStatus, }, }; @@ -41,6 +48,8 @@ use crate::{ health, create_s3_scanner_job, create_sqs_listener_job, + create_s3_prefix_job, + create_s3_keys_job, terminate_ingestion_job, ) )] @@ -63,6 +72,8 @@ pub fn create_router() -> Result, serde_json::E .routes(routes!(health)) .routes(routes!(create_s3_scanner_job)) .routes(routes!(create_sqs_listener_job)) + .routes(routes!(create_s3_prefix_job)) + .routes(routes!(create_s3_keys_job)) .routes(routes!(terminate_ingestion_job)) .split_for_parts(); @@ -80,6 +91,9 @@ pub fn create_router() -> Result, serde_json::E enum Error { #[error("{0}")] IngestionJobManagerError(#[from] IngestionJobManagerError), + + #[error("{0}")] + InvalidS3ObjectUrls(String), } impl IntoResponse for Error { @@ -102,6 +116,7 @@ impl IntoResponse for Error { (axum::http::StatusCode::BAD_REQUEST, self.to_string()) } }, + Self::InvalidS3ObjectUrls(_) => (axum::http::StatusCode::BAD_REQUEST, self.to_string()), }; let body = ErrorResponse { error: error_message, @@ -131,6 +146,25 @@ struct ErrorResponse { error: String, } +#[derive(Clone, Debug, serde::Deserialize, ToSchema)] +struct CreateS3KeysRequest { + /// The explicit S3 object URLs to ingest. + #[schema(value_type = Vec)] + object_urls: Vec, + + /// The dataset to ingest into. Defaults to `None`. + #[schema(value_type = String, min_length = 1)] + dataset: Option, + + /// The optional key for extracting timestamps from object metadata. Defaults to `None`. + #[schema(value_type = String, min_length = 1)] + timestamp_key: Option, + + /// Whether to treat the ingested objects as unstructured logs. Defaults to `false`. + #[serde(default)] + unstructured: bool, +} + #[utoipa::path( get, path = "/health", @@ -243,6 +277,106 @@ async fn create_sqs_listener_job( Ok(Json(CreationResponse { id: job_id })) } +#[utoipa::path( + post, + path = "/s3/prefix", + tags = ["IngestionJob"], + description = "Creates a one-time ingestion job that scans the specified S3 bucket and key \ + prefix once, collects all objects, and submits a single compression job.\n\n\ + The job ingests all objects under the prefix in a single pass and transitions to a \ + terminal state upon completion.", + responses( + (status = OK, body = CreationResponse, description = "The ID of the created job."), + ( + status = INTERNAL_SERVER_ERROR, + body = ErrorResponse, + description = "Internal server failure." + ), + ( + status = BAD_REQUEST, + body = ErrorResponse, + description = "A region code is not provided when using the default AWS S3 endpoint." + ) + ) +)] +async fn create_s3_prefix_job( + State(ingestion_job_manager_state): State, + Json(config): Json, +) -> Result, Error> { + tracing::info!(config = ? config, "Create S3 prefix ingestion job."); + let one_time_ingestion_job_context = ingestion_job_manager_state + .register_one_time_ingestion_job(OneTimeIngestionJobConfig::S3Prefix(config)) + .await + .map_err(|err| { + tracing::error!(err = ? err, "Failed to register S3 prefix ingestion job."); + Error::IngestionJobManagerError(err) + })?; + let job_id = one_time_ingestion_job_context.get_job_id(); + ingestion_job_manager_state + .spawn_one_time_ingestion(one_time_ingestion_job_context) + .await; + tracing::info!(job_id = ? job_id, "Created S3 prefix ingestion job."); + Ok(Json(CreationResponse { id: job_id })) +} + +#[utoipa::path( + post, + path = "/s3/keys", + tags = ["IngestionJob"], + request_body = CreateS3KeysRequest, + description = "Creates a one-time ingestion job that ingests an explicit set of S3 object \ + URLs once by scanning their shared prefix and selecting only the requested objects.\n\n\ + All submitted URLs must share the same endpoint, region, bucket, and a non-empty common \ + key prefix. Duplicate object keys are rejected.", + responses( + (status = OK, body = CreationResponse, description = "The ID of the created job."), + ( + status = INTERNAL_SERVER_ERROR, + body = ErrorResponse, + description = "Internal server failure." + ), + ( + status = BAD_REQUEST, + body = ErrorResponse, + description = "The S3 URLs are invalid or a region code is not provided when using \ + the default AWS S3 endpoint." + ) + ) +)] +async fn create_s3_keys_job( + State(ingestion_job_manager_state): State, + Json(request): Json, +) -> Result, Error> { + tracing::info!(request = ? request, "Create S3 keys ingestion job."); + let normalized_urls = normalize_s3_object_urls(&request.object_urls) + .map_err(|err| Error::InvalidS3ObjectUrls(err.to_string()))?; + let config = S3KeysConfig { + base: clp_rust_utils::job_config::ingestion::s3::BaseConfig { + bucket_name: normalized_urls.bucket_name, + key_prefix: normalized_urls.common_key_prefix, + region: normalized_urls.region, + endpoint_url: normalized_urls.endpoint_url, + dataset: request.dataset, + timestamp_key: request.timestamp_key, + unstructured: request.unstructured, + }, + object_keys: normalized_urls.object_keys, + }; + let one_time_ingestion_job_context = ingestion_job_manager_state + .register_one_time_ingestion_job(OneTimeIngestionJobConfig::S3Keys(config)) + .await + .map_err(|err| { + tracing::error!(err = ? err, "Failed to register S3 keys ingestion job."); + Error::IngestionJobManagerError(err) + })?; + let job_id = one_time_ingestion_job_context.get_job_id(); + ingestion_job_manager_state + .spawn_one_time_ingestion(one_time_ingestion_job_context) + .await; + tracing::info!(job_id = ? job_id, "Created S3 keys ingestion job."); + Ok(Json(CreationResponse { id: job_id })) +} + #[utoipa::path( post, path = "/job/{job_id}/terminate", diff --git a/components/log-ingestor/tests/test_ingestion_job.rs b/components/log-ingestor/tests/test_ingestion_job.rs index 2d86ac4f8b..c09090853e 100644 --- a/components/log-ingestor/tests/test_ingestion_job.rs +++ b/components/log-ingestor/tests/test_ingestion_job.rs @@ -7,6 +7,8 @@ use clp_rust_utils::{ job_config::ingestion::s3::{ BaseConfig, BufferConfig, + S3KeysConfig, + S3PrefixConfig, S3ScannerConfig, SqsListenerConfig, ValidatedSqsListenerConfig, @@ -17,8 +19,12 @@ use clp_rust_utils::{ use log_ingestor::{ aws_client_manager::{S3ClientWrapper, SqsClientWrapper}, ingestion_job::{ + FinalizeOutcome, IngestionJobId, IngestionJobState, + OneTimeIngestionState, + S3KeysIngestion, + S3PrefixIngestion, S3Scanner, S3ScannerState, SqsListener, @@ -26,7 +32,7 @@ use log_ingestor::{ }, }; use non_empty_string::NonEmptyString; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, oneshot}; use uuid::Uuid; use super::{ @@ -119,6 +125,90 @@ impl S3ScannerState for S3ScannerTestState { } } +#[derive(Clone, Default)] +struct S3PrefixTestState { + finalize_sender: Arc>>>>, +} + +impl S3PrefixTestState { + fn new(finalize_sender: oneshot::Sender>) -> Self { + Self { + finalize_sender: Arc::new(Mutex::new(Some(finalize_sender))), + } + } +} + +#[async_trait] +impl IngestionJobState for S3PrefixTestState { + async fn start(&self) -> Result<()> { + Ok(()) + } + + async fn end(&self) -> Result<()> { + Ok(()) + } + + async fn fail(&self, msg: String) { + panic!("S3PrefixTestState::fail should be unreachable: {msg}"); + } +} + +#[async_trait] +impl OneTimeIngestionState for S3PrefixTestState { + async fn finalize(&self, objects: Vec) -> Result { + self.finalize_sender + .lock() + .await + .take() + .expect("finalize sender should only be used once") + .send(objects) + .expect("finalize receiver should remain alive until the assertion"); + Ok(FinalizeOutcome::Finished) + } +} + +#[derive(Clone, Default)] +struct S3KeysTestState { + finalize_sender: Arc>>>>, +} + +impl S3KeysTestState { + fn new(finalize_sender: oneshot::Sender>) -> Self { + Self { + finalize_sender: Arc::new(Mutex::new(Some(finalize_sender))), + } + } +} + +#[async_trait] +impl IngestionJobState for S3KeysTestState { + async fn start(&self) -> Result<()> { + Ok(()) + } + + async fn end(&self) -> Result<()> { + Ok(()) + } + + async fn fail(&self, msg: String) { + panic!("S3KeysTestState::fail should be unreachable: {msg}"); + } +} + +#[async_trait] +impl OneTimeIngestionState for S3KeysTestState { + async fn finalize(&self, objects: Vec) -> Result { + self.finalize_sender + .lock() + .await + .take() + .expect("finalize sender should only be used once") + .send(objects) + .expect("finalize receiver should remain alive until the assertion"); + Ok(FinalizeOutcome::Finished) + } +} + /// Uploads noise S3 objects that do not match any testing prefix. /// /// The keys are formatted as `{uuid}.log`, where `uuid` is a randomly generated v4 UUID. @@ -356,3 +446,162 @@ async fn test_s3_scanner() -> Result<()> { Ok(()) } + +#[tokio::test] +#[serial_test::serial] +#[ignore = "Requires LocalStack or AWS environment"] +async fn test_s3_prefix_ingestion() -> Result<()> { + let job_id = Uuid::new_v4().as_u64_pair().0; + let prefix = get_testing_prefix_as_non_empty_string(job_id); + + let aws_config = AwsConfig::from_env()?; + + let aws_auth = AwsAuthentication::Credentials { + credentials: AwsCredentials { + access_key_id: aws_config.access_key_id.clone(), + secret_access_key: aws_config.secret_access_key.clone(), + }, + }; + let s3_client = clp_rust_utils::s3::create_new_client( + aws_config.region.as_str(), + Some(&aws_config.endpoint), + &aws_auth, + ) + .await; + + let test_upload_handle = tokio::spawn(upload_test_objects( + s3_client.clone(), + aws_config.bucket_name.clone(), + prefix.clone(), + NUM_TEST_OBJECTS, + )); + let noise_upload_handle = tokio::spawn(upload_noise_objects( + s3_client.clone(), + aws_config.bucket_name.clone(), + NUM_NOISE_OBJECTS, + )); + + noise_upload_handle + .await + .context("Error while awaiting noise upload")??; + let mut created_objects = test_upload_handle + .await + .context("Error while awaiting test object upload")??; + + let (finalize_sender, finalize_receiver) = oneshot::channel(); + let state = S3PrefixTestState::new(finalize_sender); + let _job = S3PrefixIngestion::spawn( + job_id, + S3ClientWrapper::from(s3_client), + S3PrefixConfig { + base: BaseConfig { + region: Some(aws_config.region.clone()), + bucket_name: aws_config.bucket_name.clone(), + key_prefix: prefix, + endpoint_url: Some(aws_config.endpoint.clone()), + dataset: None, + timestamp_key: None, + unstructured: false, + }, + }, + state, + ); + + let mut received_objects = tokio::time::timeout( + Duration::from_secs(WAIT_FOR_INGESTED_OBJECTS_TIMEOUT_SEC), + finalize_receiver, + ) + .await + .expect("timed out while waiting for one-time ingestion finalization") + .expect("finalize sender should not be dropped before sending objects"); + created_objects.sort(); + received_objects.sort(); + assert_eq!(received_objects, created_objects); + + Ok(()) +} + +#[tokio::test] +#[serial_test::serial] +#[ignore = "Requires LocalStack or AWS environment"] +async fn test_s3_keys_ingestion() -> Result<()> { + let job_id = Uuid::new_v4().as_u64_pair().0; + let prefix = get_testing_prefix_as_non_empty_string(job_id); + + let aws_config = AwsConfig::from_env()?; + + let aws_auth = AwsAuthentication::Credentials { + credentials: AwsCredentials { + access_key_id: aws_config.access_key_id.clone(), + secret_access_key: aws_config.secret_access_key.clone(), + }, + }; + let s3_client = clp_rust_utils::s3::create_new_client( + aws_config.region.as_str(), + Some(&aws_config.endpoint), + &aws_auth, + ) + .await; + + let test_upload_handle = tokio::spawn(upload_test_objects( + s3_client.clone(), + aws_config.bucket_name.clone(), + prefix.clone(), + NUM_TEST_OBJECTS, + )); + let noise_upload_handle = tokio::spawn(upload_noise_objects( + s3_client.clone(), + aws_config.bucket_name.clone(), + NUM_NOISE_OBJECTS, + )); + + noise_upload_handle + .await + .context("Error while awaiting noise upload")??; + let created_objects = test_upload_handle + .await + .context("Error while awaiting test object upload")??; + + let requested_indices = [1_usize, 10, 50]; + let mut expected_objects: Vec = requested_indices + .iter() + .map(|idx| created_objects[*idx].clone()) + .collect(); + let object_keys = expected_objects + .iter() + .map(|object| object.key.clone()) + .collect(); + + let (finalize_sender, finalize_receiver) = oneshot::channel(); + let state = S3KeysTestState::new(finalize_sender); + let _job = S3KeysIngestion::spawn( + job_id, + S3ClientWrapper::from(s3_client), + S3KeysConfig { + base: BaseConfig { + region: Some(aws_config.region.clone()), + bucket_name: aws_config.bucket_name.clone(), + key_prefix: NonEmptyString::from_string(format!("{prefix}/0")), + endpoint_url: Some(aws_config.endpoint.clone()), + dataset: None, + timestamp_key: None, + unstructured: false, + }, + object_keys, + }, + state, + ); + + let mut received_objects = tokio::time::timeout( + Duration::from_secs(WAIT_FOR_INGESTED_OBJECTS_TIMEOUT_SEC), + finalize_receiver, + ) + .await + .expect("timed out while waiting for one-time S3 keys ingestion finalization") + .expect("finalize sender should not be dropped before sending objects"); + expected_objects.sort(); + received_objects.sort(); + assert_eq!(received_objects, expected_objects); + + Ok(()) +} diff --git a/docs/src/_static/generated/log-ingestor-openapi.json b/docs/src/_static/generated/log-ingestor-openapi.json index c2b1a03fa3..ebb6fe09ae 100644 --- a/docs/src/_static/generated/log-ingestor-openapi.json +++ b/docs/src/_static/generated/log-ingestor-openapi.json @@ -84,6 +84,108 @@ } } }, + "/s3/keys": { + "post": { + "tags": [ + "IngestionJob" + ], + "description": "Creates a one-time ingestion job that ingests an explicit set of S3 object URLs once by scanning their shared prefix and selecting only the requested objects.\n\nAll submitted URLs must share the same endpoint, region, bucket, and a non-empty common key prefix. Duplicate object keys are rejected.", + "operationId": "create_s3_keys_job", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CreateS3KeysRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "The ID of the created job.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CreationResponse" + } + } + } + }, + "400": { + "description": "The S3 URLs are invalid or a region code is not provided when using the default AWS S3 endpoint.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "500": { + "description": "Internal server failure.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + } + } + } + }, + "/s3/prefix": { + "post": { + "tags": [ + "IngestionJob" + ], + "description": "Creates a one-time ingestion job that scans the specified S3 bucket and key prefix once, collects all objects, and submits a single compression job.\n\nThe job ingests all objects under the prefix in a single pass and transitions to a terminal state upon completion.", + "operationId": "create_s3_prefix_job", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/S3PrefixConfig" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "The ID of the created job.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CreationResponse" + } + } + } + }, + "400": { + "description": "A region code is not provided when using the default AWS S3 endpoint.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "500": { + "description": "Internal server failure.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + } + } + } + }, "/s3_scanner": { "post": { "tags": [ @@ -279,6 +381,37 @@ } } }, + "CreateS3KeysRequest": { + "type": "object", + "required": [ + "object_urls", + "dataset", + "timestamp_key" + ], + "properties": { + "dataset": { + "type": "string", + "description": "The dataset to ingest into. Defaults to `None`.", + "minLength": 1 + }, + "object_urls": { + "type": "array", + "items": { + "type": "string" + }, + "description": "The explicit S3 object URLs to ingest." + }, + "timestamp_key": { + "type": "string", + "description": "The optional key for extracting timestamps from object metadata. Defaults to `None`.", + "minLength": 1 + }, + "unstructured": { + "type": "boolean", + "description": "Whether to treat the ingested objects as unstructured logs. Defaults to `false`." + } + } + }, "CreationResponse": { "type": "object", "required": [ @@ -303,6 +436,14 @@ } } }, + "S3PrefixConfig": { + "allOf": [ + { + "$ref": "#/components/schemas/BaseConfig" + } + ], + "description": "Configuration for a S3 prefix ingestion job." + }, "S3ScannerConfig": { "allOf": [ {