feat(log-ingestor): Add one-time S3 keys ingestion.

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"

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<S3ObjectMetadataId>,
+    ) -> 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,
+        }
+    }
+}

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

@@ -176,6 +176,24 @@ pub mod s3 {
         pub start_after: Option<NonEmptyString>,
     }
 
+    /// 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<String>)]
+        pub object_keys: Vec<NonEmptyString>,
+    }
+
     /// Configuration for buffer behavior.
     #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
     #[serde(default)]

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;

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<Regex> = LazyLock::new(|| {
+    Regex::new(concat!(
+        r"^(?P<scheme>https?)://",
+        r"(?P<bucket_name>[a-z0-9.-]+)\.",
+        r"(?P<s3>s3)\.",
+        r"(?:(?P<region_code>[a-z0-9\-]+)\.)?",
+        r"(?P<endpoint>[A-Za-z0-9.-]+(?::[0-9]+)?)",
+        r"/(?P<key_prefix>[^?]+)(?:\?.*)?$"
+    ))
+    .expect("host-style S3 URL regex should be valid")
+});
+static PATH_STYLE_URL_REGEX: LazyLock<Regex> = LazyLock::new(|| {
+    Regex::new(concat!(
+        r"^(?P<scheme>https?)://",
+        r"(?:(?P<s3>s3)\.(?:(?P<region_code>[a-z0-9\-]+)\.)?)?",
+        r"(?P<endpoint>[A-Za-z0-9.-]+(?::[0-9]+)?)",
+        r"/(?P<bucket_name>[a-z0-9.-]+)",
+        r"/(?P<key_prefix>[^?]+)(?:\?.*)?$"
+    ))
+    .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<NonEmptyString>,
+    pub region: Option<NonEmptyString>,
+    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<NonEmptyString>,
+    pub region: Option<NonEmptyString>,
+    pub bucket_name: NonEmptyString,
+    pub common_key_prefix: NonEmptyString,
+    pub object_keys: Vec<NonEmptyString>,
+}
+
+/// 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<ParsedS3Url> {
+    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<NormalizedS3ObjectUrls> {
+    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<NonEmptyString> = 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<NonEmptyString> {
+    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)
+}