Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 3 additions & 0 deletions components/api-server/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,9 @@ impl From<clp_rust_utils::Error> for ClientError {
clp_rust_utils::Error::MsgpackEncode(_) | clp_rust_utils::Error::SerdeYaml(_) => {
Self::MalformedData
}
clp_rust_utils::Error::UnsupportedS3Endpoint(_) => {
Self::InvalidInput(value.to_string())
}
clp_rust_utils::Error::Io(error) => error.into(),
clp_rust_utils::Error::Sqlx(error) => error.into(),
clp_rust_utils::Error::TelemetryExporterBuildError(error) => Self::Telemetry(error),
Expand Down
1 change: 1 addition & 0 deletions components/clp-rust-utils/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ aws-config = "1.8.12"
aws-sdk-s3 = "1.121.0"
aws-sdk-sqs = "1.92.0"
brotli = "8.0.2"
const_format = "0.2.35"
non-empty-string = { version = "0.2.6", features = ["serde"] }
num_enum = "0.7.5"
opentelemetry = { version = "0.32.0" }
Expand Down
9 changes: 0 additions & 9 deletions components/clp-rust-utils/src/clp_config/package.rs
Original file line number Diff line number Diff line change
@@ -1,15 +1,6 @@
use std::sync::LazyLock;

use non_empty_string::NonEmptyString;

use crate::types::non_empty_string::ExpectedNonEmpty;

pub mod config;
pub mod credentials;

/// Mirror of configuration paths in `clp_py_utils.clp_config`.
pub const DEFAULT_CONFIG_FILE_PATH: &str = "etc/clp-config.yaml";
pub const DEFAULT_CREDENTIALS_FILE_PATH: &str = "etc/credentials.yaml";

pub static DEFAULT_DATASET_NAME: LazyLock<NonEmptyString> =
LazyLock::new(|| NonEmptyString::from_static_str("default"));
282 changes: 274 additions & 8 deletions components/clp-rust-utils/src/clp_config/package/config.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,11 @@
use std::path::{Path, PathBuf};

use serde::Deserialize;

use crate::clp_config::{AwsAuthentication, S3Config};
use crate::{
clp_config::{AwsAuthentication, S3Config},
dataset::resolve_dataset_name,
};

/// Mirror of `clp_py_utils.clp_config.ClpConfig`.
///
Expand Down Expand Up @@ -44,13 +49,55 @@ impl Default for Config {
}

/// Configuration for the Spider task executor.
///
/// # NOTE
///
/// This type is a scaffolding config definition.
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq)]
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
#[serde(default)]
pub struct SpiderTaskExecutorConfig {}
pub struct SpiderTaskExecutorConfig {
pub package: Package,
pub archive_output: ArchiveOutput,
pub tmp_directory: String,
pub database: Database,
}

impl SpiderTaskExecutorConfig {
/// Resolves `tmp_directory` against `clp_home`.
///
/// # Returns
///
/// `tmp_directory` unchanged if it is already absolute; otherwise, it is joined with
/// `clp_home`.
#[must_use]
pub fn abs_tmp_directory(&self, clp_home: &Path) -> PathBuf {
make_config_path_absolute(clp_home, &self.tmp_directory)
}

/// Resolves the archive-output storage's local directory against `clp_home`.
///
/// # Returns
///
/// The storage's local directory (`directory` for `Fs`, `staging_directory` for `S3`) unchanged
/// if it is already absolute; otherwise, it is joined with `clp_home`.
#[must_use]
pub fn abs_archive_output_staging(&self, clp_home: &Path) -> PathBuf {
let directory = match &self.archive_output.storage {
ArchiveOutputStorage::Fs { directory } => directory,
ArchiveOutputStorage::S3 {
staging_directory, ..
} => staging_directory,
};
make_config_path_absolute(clp_home, directory)
}
}

impl Default for SpiderTaskExecutorConfig {
fn default() -> Self {
Self {
package: Package::default(),
database: Database::default(),
archive_output: ArchiveOutput::default(),
tmp_directory: "var/tmp".to_owned(),
}
}
}

/// Database names for CLP components.
///
Expand Down Expand Up @@ -83,20 +130,28 @@ impl Default for ClpDbNames {
/// * This type is partially defined: unused fields are omitted and discarded through
/// deserialization.
/// * The default values must be kept in sync with the Python definition.
/// * `table_prefix` is a fixed constant mirroring `CLP_METADATA_TABLE_PREFIX` (`"clp_"`) and is
/// excluded from (de)serialization.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
#[serde(default)]
pub struct Database {
pub host: String,
pub port: u16,
pub names: ClpDbNames,

#[serde(skip)]
pub table_prefix: String,
}

impl Default for Database {
fn default() -> Self {
/// Mirror of `clp_py_utils.clp_config.CLP_METADATA_TABLE_PREFIX`.
const CLP_METADATA_TABLE_PREFIX: &str = "clp_";
Self {
host: "localhost".to_owned(),
port: 3306,
names: ClpDbNames::default(),
table_prefix: CLP_METADATA_TABLE_PREFIX.to_owned(),
}
}
}
Expand Down Expand Up @@ -260,16 +315,38 @@ impl Default for LogIngestor {
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
#[serde(default)]
pub struct ArchiveOutput {
pub storage: ArchiveOutputStorage,
pub target_archive_size: u64,
pub target_dictionaries_size: u64,
pub target_encoded_file_size: u64,
pub target_segment_size: u64,
pub compression_level: u8,
}

impl ArchiveOutput {
/// Derives the archive storage directory for a dataset.
///
/// # Returns
///
/// The dataset's storage base (`s3_config.key_prefix` for S3, `directory` for `Fs`) joined with
/// `dataset`, where a `None` dataset resolves to `default`.
#[must_use]
pub fn dataset_archive_storage_directory(&self, dataset: Option<&str>) -> String {
let base = match &self.storage {
ArchiveOutputStorage::Fs { directory } => directory.as_str(),
ArchiveOutputStorage::S3 { s3_config, .. } => s3_config.key_prefix.as_str(),
};
Path::new(base)
.join(resolve_dataset_name(dataset))
.to_string_lossy()
.into_owned()
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}

impl Default for ArchiveOutput {
fn default() -> Self {
Self {
storage: ArchiveOutputStorage::default(),
target_archive_size: 256 * 1024 * 1024,
target_dictionaries_size: 32 * 1024 * 1024,
target_encoded_file_size: 256 * 1024 * 1024,
Expand All @@ -279,6 +356,29 @@ impl Default for ArchiveOutput {
}
}

/// Mirror of `clp_py_utils.clp_config.ArchiveFsStorage` | `ArchiveS3Storage`.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
#[serde(tag = "type")]
pub enum ArchiveOutputStorage {
#[serde(rename = "fs")]
Fs { directory: String },

#[serde(rename = "s3")]
S3 {
#[serde(default = "default_archive_staging_directory")]
staging_directory: String,
s3_config: S3Config,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@coderabbitai Open an issue:

pub enum AwsAuthentication {

Support credential-free workload in Rust.
Long-term speaking, AWS credentials should never be serialized into ClpIoConfig, persisted in compression job records, or forwarded through Spider task inputs. Instead, the coordinator and scheduler should pass only non-sensitive S3 location metadata, such as the endpoint, region, bucket, and object key. Each task executor should retrieve credentials locally using the AWS default credential provider chain, a mounted AWS profile, workload identity, or environment variables injected at runtime.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

repo="y-scope/clp"
title="Support credential-free S3 workloads in Rust"

body_file="$(mktemp)"
cat >"$body_file" <<'EOF'
## Summary

Support credential-free S3 workloads in the Rust components.

AWS credentials must not be serialized into `ClpIoConfig`, persisted in compression-job records, or forwarded through Spider task inputs. The coordinator and scheduler should instead carry only non-sensitive S3 location metadata (endpoint, region, bucket, and object key). Each task executor must obtain AWS credentials locally at runtime through the AWS default credential provider chain, such as a mounted AWS profile, workload identity, or environment variables.

## Rationale

Keeping credentials out of configuration objects and task payloads reduces the risk of credential disclosure through persistence, logs, job records, and cross-component transport. It also enables workload-identity-based deployments without embedding static credentials.

## Affected areas

- `components/clp-rust-utils/src/clp_config/s3_config.rs`
- Rust `ClpIoConfig` modelling and serialization/deserialization
- Compression job record creation and persistence
- Coordinator/scheduler-to-Spider task input contracts
- Rust S3 client construction and task-executor authentication resolution

## Required changes

1. Define credential-free S3 location/task metadata that contains only endpoint, region, bucket, and object key (plus other non-sensitive fields required for S3 access).
2. Remove or prevent credentials from being represented in serialized `ClpIoConfig`, compression-job records, and Spider task inputs.
3. Update Rust S3 client/task-executor setup to resolve credentials locally through the AWS default credential provider chain.
4. Preserve support for runtime credential sources including mounted profiles, workload identity, and environment-injected credentials.
5. Add tests demonstrating that serialized/persisted/forwarded payloads do not contain access keys or secret keys, and that credential-free metadata can be used by a task executor.

## Acceptance criteria

- [ ] No AWS access key ID, secret access key, session token, or equivalent credential material is serialized into `ClpIoConfig`.
- [ ] Compression job records and Spider task inputs include only non-sensitive S3 location metadata.
- [ ] A Rust task executor can create/use an S3 client using the AWS default credential provider chain without credentials in its input payload.
- [ ] Tests cover credential exclusion and successful credential-free configuration/task handling.

Requested by `@Bill-hbrhbr`.

Backlinks:
- PR `#2406`: https://github.com/y-scope/clp/pull/2406
- Review comment: https://github.com/y-scope/clp/pull/2406#discussion_r3661977290
EOF

issue_url="$(gh issue create \
  --repo "$repo" \
  --title "$title" \
  --body-file "$body_file" \
  --assignee "Bill-hbrhbr")"
rm -f "$body_file"

printf '%s\n' "$issue_url"

Length of output: 3107


@Bill-hbrhbr, opened and assigned #2432: Support credential-free S3 workloads in Rust.

It captures the requirement to keep credentials out of ClpIoConfig, job records, and Spider task inputs, while resolving them locally through the AWS default credential provider chain.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To be fair: I think this is out of the scope for "Rust" components. This is a package-level design problem, which we already have a solution for through a new component called credential-manager. This component has been completed in a private branch, but we haven't had time to open-source it.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems like the humans are having a chat. I'll hop back into my burrow for now. If you need me again, just tag @coderabbitai in a new comment, and I'll come hopping out!

},
}

impl Default for ArchiveOutputStorage {
fn default() -> Self {
Self::Fs {
directory: "var/data/archives".to_owned(),
}
}
}

/// Mirror of `clp_py_utils.clp_config.S3IngestionConfig`.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq)]
pub struct S3Ingestion {
Expand Down Expand Up @@ -337,9 +437,32 @@ impl Default for Telemetry {
}
}

/// # Returns
///
/// `path` unchanged if it is already absolute, otherwise joined with `root`.
fn make_config_path_absolute(root: &Path, path: &str) -> PathBuf {
if Path::new(path).is_absolute() {
PathBuf::from(path)
} else {
root.join(path)
}
}

fn default_archive_staging_directory() -> String {
"var/data/staged-archives".to_owned()
}

#[cfg(test)]
mod tests {
use super::LogsInput;
use std::path::Path;

use super::{
ArchiveOutput,
ArchiveOutputStorage,
Database,
LogsInput,
SpiderTaskExecutorConfig,
};

#[test]
fn deserialize_logs_input_s3_config() {
Expand Down Expand Up @@ -418,4 +541,147 @@ mod tests {
LogsInput::S3 { .. } => panic!("Expected Fs"),
}
}

#[test]
fn dataset_archive_storage_directory_fs() {
let archive_output = ArchiveOutput {
storage: ArchiveOutputStorage::Fs {
directory: "/var/data/archives".to_owned(),
},
..ArchiveOutput::default()
};

assert_eq!(
archive_output.dataset_archive_storage_directory(Some("mydataset")),
"/var/data/archives/mydataset"
);
assert_eq!(
archive_output.dataset_archive_storage_directory(None),
"/var/data/archives/default"
);
}

#[test]
fn dataset_archive_storage_directory_s3() {
use non_empty_string::NonEmptyString;

use crate::clp_config::{AwsAuthentication, S3Config};

let archive_output = ArchiveOutput {
storage: ArchiveOutputStorage::S3 {
staging_directory: "var/data/staged-archives".to_owned(),
s3_config: S3Config {
bucket: NonEmptyString::try_from("bucket".to_string())
.expect("bucket is non-empty"),
region_code: None,
key_prefix: NonEmptyString::try_from("prefix".to_string())
.expect("key prefix is non-empty"),
endpoint_url: None,
aws_authentication: AwsAuthentication::Default,
},
},
..ArchiveOutput::default()
};

assert_eq!(
archive_output.dataset_archive_storage_directory(Some("mydataset")),
"prefix/mydataset"
);
assert_eq!(
archive_output.dataset_archive_storage_directory(None),
"prefix/default"
);
}

#[test]
fn deserialize_database_ignores_provided_table_prefix() {
let database_json = serde_json::json!({
"host": "h",
"port": 3306,
"names": {
"clp": "clp-db",
"spider": "spider-db",
},
"table_prefix": "custom_"
});

let db = serde_json::from_str::<Database>(database_json.to_string().as_str())
.expect("failed to deserialize `Database` from JSON");

assert_eq!(db.table_prefix, "clp_");
}

#[test]
fn abs_tmp_directory_joins_relative_path() {
let config = SpiderTaskExecutorConfig {
tmp_directory: "var/tmp".to_owned(),
..SpiderTaskExecutorConfig::default()
};

assert_eq!(
config.abs_tmp_directory(Path::new("/opt/clp")),
Path::new("/opt/clp/var/tmp")
);
}

#[test]
fn abs_tmp_directory_leaves_absolute_path_unchanged() {
let config = SpiderTaskExecutorConfig {
tmp_directory: "/abs/tmp".to_owned(),
..SpiderTaskExecutorConfig::default()
};

assert_eq!(
config.abs_tmp_directory(Path::new("/opt/clp")),
Path::new("/abs/tmp")
);
}

#[test]
fn abs_archive_output_staging_joins_relative_s3_path() {
let config = s3_config_with_staging_directory("var/staged-archives");

assert_eq!(
config.abs_archive_output_staging(Path::new("/opt/clp")),
Path::new("/opt/clp/var/staged-archives")
);
}

#[test]
fn abs_archive_output_staging_leaves_absolute_s3_path_unchanged() {
let config = s3_config_with_staging_directory("/abs/staged-archives");

assert_eq!(
config.abs_archive_output_staging(Path::new("/opt/clp")),
Path::new("/abs/staged-archives")
);
}

/// # Returns
///
/// A [`SpiderTaskExecutorConfig`] whose archive output is S3-backed with `staging_directory`.
fn s3_config_with_staging_directory(staging_directory: &str) -> SpiderTaskExecutorConfig {
use non_empty_string::NonEmptyString;

use crate::clp_config::{AwsAuthentication, S3Config};

SpiderTaskExecutorConfig {
archive_output: ArchiveOutput {
storage: ArchiveOutputStorage::S3 {
staging_directory: staging_directory.to_owned(),
s3_config: S3Config {
bucket: NonEmptyString::try_from("bucket".to_string())
.expect("bucket is non-empty"),
region_code: None,
key_prefix: NonEmptyString::try_from("prefix/".to_string())
.expect("key prefix is non-empty"),
endpoint_url: None,
aws_authentication: AwsAuthentication::Default,
},
},
..ArchiveOutput::default()
},
..SpiderTaskExecutorConfig::default()
}
}
}
Loading
Loading