Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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.

2 changes: 2 additions & 0 deletions components/clp-rust-utils/src/clp_config/s3_config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -29,4 +29,6 @@ pub enum AwsAuthentication {
pub struct AwsCredentials {
pub access_key_id: String,
pub secret_access_key: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub session_token: Option<String>,
}
2 changes: 1 addition & 1 deletion components/clp-rust-utils/src/s3/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ pub async fn create_new_client(
config_defaults = config_defaults.credentials_provider(Credentials::new(
credentials.access_key_id.as_str(),
credentials.secret_access_key.as_str(),
None,
credentials.session_token.clone(),
None,
"clp-credentials-provider",
));
Expand Down
2 changes: 1 addition & 1 deletion components/clp-rust-utils/src/sqs/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ pub async fn create_new_client(
config_defaults = config_defaults.credentials_provider(Credentials::new(
credentials.access_key_id.as_str(),
credentials.secret_access_key.as_str(),
None,
credentials.session_token.clone(),
None,
"clp-credentials-provider",
));
Expand Down
1 change: 1 addition & 0 deletions components/clp-rust-utils/tests/clp_config_test.rs
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ fn test_clp_io_config_serialization() {
credentials: AwsCredentials {
access_key_id: "ACCESS_KEY_ID".into(),
secret_access_key: "SECRET_ACCESS_KEY".into(),
session_token: None,
},
},
};
Expand Down
1 change: 1 addition & 0 deletions components/clp-tdl-package/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ crate-type = ["cdylib", "rlib"]

[dependencies]
anyhow = "1.0.100"
aws-config = "1.8.12"
aws-sdk-s3 = "1.121.0"
clp-rust-utils = { path = "../clp-rust-utils" }
non-empty-string = { version = "0.2.6", features = ["serde"] }
Expand Down
92 changes: 70 additions & 22 deletions components/clp-tdl-package/src/task/compression/compress.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@ use std::process::Command;
use std::process::Stdio;

use anyhow::Context;
use aws_config::BehaviorVersion;
use aws_sdk_s3::config::Credentials;
use aws_sdk_s3::config::ProvideCredentials;
use clp_rust_utils::aws::AWS_DEFAULT_REGION;
use clp_rust_utils::clp_config::AwsAuthentication;
use clp_rust_utils::clp_config::S3Config;
Expand Down Expand Up @@ -46,6 +49,7 @@ use crate::common::runtime;
/// * A spawned archive finisher panics.
/// * Forwards [`build_s3_logs_list`]'s return values on failure.
/// * Forwards [`std::fs::write`]'s return values on failure.
/// * Forwards [`s3_credential_env`]'s return values on failure.
/// * Forwards [`extract_s3_output_config`]'s return values on failure.
/// * Forwards [`prepare_clp_s_input`]'s return values on failure.
/// * Forwards [`run_clp_s`]'s return values on failure.
Expand Down Expand Up @@ -84,10 +88,16 @@ pub(super) fn compress(
})?;
tmp_file_deleter.add(list_path.clone());

let runtime = runtime();
let S3InputSource {
aws_authentication, ..
region_code,
aws_authentication,
..
} = input_source;
let credential_env = s3_credential_env(&aws_authentication);
let input_region = region_code
.as_ref()
.map_or(AWS_DEFAULT_REGION, NonEmptyString::as_str);
let credential_env = s3_credential_env(&runtime, input_region, &aws_authentication)?;
Comment thread
20001020ycx marked this conversation as resolved.
Resolved

let s3_config = extract_s3_output_config(config)?;
let archive_dir = config
Expand All @@ -107,7 +117,6 @@ pub(super) fn compress(
&mut tmp_file_deleter,
)?;

let runtime = runtime();
let client = build_s3_client(&runtime, s3_config);
let bucket = s3_config.bucket.to_string();

Expand Down Expand Up @@ -341,26 +350,66 @@ fn build_s3_logs_list(input_source: &S3InputSource) -> anyhow::Result<String> {
///
/// # Returns
///
/// * The env-var name-value pairs for [`AwsAuthentication::Credentials`].
/// * An empty vector for [`AwsAuthentication::Default`] (which assumes credentials are already in
/// the ambient env).
fn s3_credential_env(auth: &AwsAuthentication) -> Vec<(&'static str, String)> {
/// The env-var name-value pairs containing the resolved access key, secret key, and optional
/// session token.
Comment thread
20001020ycx marked this conversation as resolved.
Resolved
///
/// # Errors
///
/// Returns an error if:
///
/// * The default AWS SDK credential provider chain has no provider.
/// * The default AWS SDK credential provider chain fails to resolve credentials.
Comment thread
20001020ycx marked this conversation as resolved.
Resolved
fn s3_credential_env(
runtime: &tokio::runtime::Handle,
region: &str,
auth: &AwsAuthentication,
) -> anyhow::Result<Vec<(&'static str, String)>> {
/// The env var holding the AWS access key ID.
const AWS_ACCESS_KEY_ID_ENV_VAR: &str = "AWS_ACCESS_KEY_ID";

/// The env var holding the AWS secret access key.
const AWS_SECRET_ACCESS_KEY_ENV_VAR: &str = "AWS_SECRET_ACCESS_KEY";

match auth {
AwsAuthentication::Credentials { credentials } => vec![
(AWS_ACCESS_KEY_ID_ENV_VAR, credentials.access_key_id.clone()),
(
AWS_SECRET_ACCESS_KEY_ENV_VAR,
credentials.secret_access_key.clone(),
),
],
AwsAuthentication::Default => Vec::new(),
/// The env var holding the AWS session token.
const AWS_SESSION_TOKEN_ENV_VAR: &str = "AWS_SESSION_TOKEN";

let credentials = match auth {
AwsAuthentication::Credentials { credentials } => Credentials::new(
credentials.access_key_id.clone(),
credentials.secret_access_key.clone(),
credentials.session_token.clone(),
None,
"clp-credentials-provider",
),
AwsAuthentication::Default => {
let sdk_config = runtime.block_on(
aws_config::defaults(BehaviorVersion::latest())
.region(aws_sdk_s3::config::Region::new(region.to_string()))
.load(),
);
let provider = sdk_config
.credentials_provider()
.context("default AWS SDK credential provider is unavailable")?;
runtime
.block_on(provider.provide_credentials())
.context("failed to resolve credentials from the default AWS SDK provider chain")?
}
};

let mut env = vec![
(
AWS_ACCESS_KEY_ID_ENV_VAR,
credentials.access_key_id().to_string(),
),
(
AWS_SECRET_ACCESS_KEY_ENV_VAR,
credentials.secret_access_key().to_string(),
),
];
if let Some(session_token) = credentials.session_token() {
env.push((AWS_SESSION_TOKEN_ENV_VAR, session_token.to_string()));
Comment thread
20001020ycx marked this conversation as resolved.
Resolved
}
Ok(env)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}

/// Parses a single clp-s `--print-archive-stats` stdout line into an [`ArchiveMetadata`].
Expand Down Expand Up @@ -897,25 +946,24 @@ mod tests {
Ok(())
}

#[test]
fn s3_credential_env_default() {
assert_eq!(s3_credential_env(&AwsAuthentication::Default), Vec::new());
}

#[test]
fn s3_credential_env_credentials() {
let runtime = tokio::runtime::Runtime::new().expect("failed to create Tokio runtime");
let auth = AwsAuthentication::Credentials {
credentials: AwsCredentials {
access_key_id: "the-access-key".to_string(),
secret_access_key: "the-secret-key".to_string(),
session_token: Some("the-session-token".to_string()),
},
};

assert_eq!(
s3_credential_env(&auth),
s3_credential_env(runtime.handle(), "us-east-1", &auth)
.expect("failed to resolve credentials"),
vec![
("AWS_ACCESS_KEY_ID", "the-access-key".to_string()),
("AWS_SECRET_ACCESS_KEY", "the-secret-key".to_string()),
("AWS_SESSION_TOKEN", "the-session-token".to_string()),
]
);
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
Expand Down
3 changes: 3 additions & 0 deletions components/log-ingestor/tests/test_ingestion_job.rs
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,7 @@ async fn run_sqs_listener_test(
credentials: AwsCredentials {
access_key_id: aws_config.access_key_id.clone(),
secret_access_key: aws_config.secret_access_key.clone(),
session_token: None,
},
};
let sqs_client = clp_rust_utils::sqs::create_new_client(
Expand Down Expand Up @@ -290,6 +291,7 @@ async fn test_s3_scanner() -> Result<()> {
credentials: AwsCredentials {
access_key_id: aws_config.access_key_id.clone(),
secret_access_key: aws_config.secret_access_key.clone(),
session_token: None,
},
};
let s3_client = clp_rust_utils::s3::create_new_client(
Expand Down Expand Up @@ -368,6 +370,7 @@ async fn test_s3_scanner_single_object() -> Result<()> {
credentials: AwsCredentials {
access_key_id: aws_config.access_key_id.clone(),
secret_access_key: aws_config.secret_access_key.clone(),
session_token: None,
},
};
let s3_client = clp_rust_utils::s3::create_new_client(
Expand Down
1 change: 1 addition & 0 deletions components/log-ingestor/tests/test_scan.rs
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ async fn create_test_s3_client(aws_config: &AwsConfig) -> aws_sdk_s3::Client {
credentials: AwsCredentials {
access_key_id: aws_config.access_key_id.clone(),
secret_access_key: aws_config.secret_access_key.clone(),
session_token: None,
},
};

Expand Down
Loading