Merge branch 'main' into always-pull-presto

Author: junhaoliao

components/api-server/src/client.rs

@@ -3,12 +3,9 @@ use std::pin::Pin;
 use async_stream::stream;
 use clp_rust_utils::{
     aws::AWS_DEFAULT_REGION,
-    clp_config::{
-        AwsAuthentication,
-        package::{
-            config::{Config, StorageEngine, StreamOutputStorage},
-            credentials::Credentials,
-        },
+    clp_config::package::{
+        config::{Config, StorageEngine, StreamOutputStorage},
+        credentials::Credentials,
     },
     database::mysql::create_clp_db_mysql_pool,
     job_config::{QUERY_JOBS_TABLE_NAME, QueryJobStatus, QueryJobType, SearchJobConfig},
@@ -374,8 +371,6 @@ impl Client {
             unreachable!();
         };
 
-        let AwsAuthentication::Credentials { credentials } = &s3_config.aws_authentication;
-
         let s3_config = s3_config.clone();
         if s3_config.region_code.is_none() && s3_config.endpoint_url.is_none() {
             return Err(ClientError::Aws {
@@ -384,15 +379,14 @@ impl Client {
             });
         }
 
-        let credentials = credentials.clone();
+        let region_str = s3_config
+            .region_code
+            .as_ref()
+            .map_or(AWS_DEFAULT_REGION, non_empty_string::NonEmptyString::as_str);
         let s3_client = clp_rust_utils::s3::create_new_client(
-            credentials.access_key_id.as_str(),
-            credentials.secret_access_key.as_str(),
-            s3_config
-                .region_code
-                .as_ref()
-                .map_or(AWS_DEFAULT_REGION, non_empty_string::NonEmptyString::as_str),
+            region_str,
             s3_config.endpoint_url.as_ref(),
+            &s3_config.aws_authentication,
         )
         .await;
 

components/clp-py-utils/clp_py_utils/clp_config.py

@@ -165,10 +165,10 @@ class StorageType(LowercaseStrEnum):
 
 
 class AwsAuthType(LowercaseStrEnum):
+    default = auto()
     credentials = auto()
     profile = auto()
     env_vars = auto()
-    ec2 = auto()
 
 
 AwsAuthTypeStr = Annotated[AwsAuthType, StrEnumSerializer]
@@ -588,7 +588,7 @@ def validate_authentication(cls, data):
             raise ValueError(f"profile must be set when type is '{auth_enum}.'")
         if AwsAuthType.credentials == auth_enum and not credentials:
             raise ValueError(f"credentials must be set when type is '{auth_enum}.'")
-        if auth_enum in [AwsAuthType.ec2, AwsAuthType.env_vars] and (profile or credentials):
+        if auth_enum in [AwsAuthType.default, AwsAuthType.env_vars] and (profile or credentials):
             raise ValueError(f"profile and credentials must not be set when type is '{auth_enum}.'")
         return data
 
@@ -951,7 +951,6 @@ def validate_tmp_dir(self, use_host_mount: bool = False):
             raise ValueError(f"tmp_directory is invalid: {ex}")
 
     def validate_aws_config_dir(self, use_host_mount: bool = False):
-        profile_auth_used = False
         auth_configs = []
 
         if StorageType.S3 == self.logs_input.type:
@@ -961,16 +960,23 @@ def validate_aws_config_dir(self, use_host_mount: bool = False):
         if StorageType.S3 == self.stream_output.storage.type:
             auth_configs.append(self.stream_output.storage.s3_config.aws_authentication)
 
-        for auth in auth_configs:
-            if AwsAuthType.profile == auth.type:
-                profile_auth_used = True
-                break
+        auth_types_used = {auth.type for auth in auth_configs}
+        default_auth_used = AwsAuthType.default in auth_types_used
+        profile_auth_used = AwsAuthType.profile in auth_types_used
+        config_dir_allowed = profile_auth_used or default_auth_used
 
         if profile_auth_used:
             if self.aws_config_directory is None:
                 raise ValueError(
                     "aws_config_directory must be set when using profile authentication"
                 )
+
+        if self.aws_config_directory is not None:
+            if not config_dir_allowed:
+                raise ValueError(
+                    "aws_config_directory is only supported with 'profile' or 'default'"
+                    " authentication"
+                )
             resolved_aws_config_dir = (
                 resolve_host_path_in_container(self.aws_config_directory)
                 if use_host_mount
@@ -980,10 +986,6 @@ def validate_aws_config_dir(self, use_host_mount: bool = False):
                 raise ValueError(
                     f"aws_config_directory does not exist: '{self.aws_config_directory}'"
                 )
-        if not profile_auth_used and self.aws_config_directory is not None:
-            raise ValueError(
-                "aws_config_directory should not be set when profile authentication is not used"
-            )
 
     def validate_api_server(self):
         if StorageEngine.CLP == self.package.storage_engine and self.api_server is not None:

components/clp-py-utils/clp_py_utils/s3_utils.py

@@ -90,10 +90,10 @@ def get_credential_env_vars(auth: AwsAuthentication) -> dict[str, str]:
         if aws_credentials is None:
             raise ValueError(f"Failed to authenticate with profile {auth.profile}")
 
-    elif AwsAuthType.ec2 == auth.type:
+    elif AwsAuthType.default == auth.type:
         aws_credentials = _get_session_credentials()
         if aws_credentials is None:
-            raise ValueError("Failed to authenticate with EC2 metadata.")
+            raise ValueError("Failed to authenticate with the default credential provider chain.")
     else:
         raise ValueError(f"Unsupported authentication type: {auth.type}")
 
@@ -206,7 +206,7 @@ def _create_s3_client(
             region_name=region_code,
             aws_session_token=credentials.session_token,
         )
-    elif AwsAuthType.env_vars == s3_auth.type or AwsAuthType.ec2 == s3_auth.type:
+    elif AwsAuthType.env_vars == s3_auth.type or AwsAuthType.default == s3_auth.type:
         # Use default session which will use environment variables or instance role
         aws_session = boto3.Session(region_name=region_code)
     else:

components/clp-rust-utils/src/clp_config/package/config.rs

@@ -345,11 +345,38 @@ mod tests {
                     assert_eq!(credentials.access_key_id, ACCESS_KEY_ID);
                     assert_eq!(credentials.secret_access_key, SECRET_ACCESS_KEY);
                 }
+                crate::clp_config::AwsAuthentication::Default => {
+                    panic!("Expected credentials, got `default`")
+                }
             },
             LogsInput::Fs { .. } => panic!("Expected S3"),
         }
     }
 
+    #[test]
+    fn deserialize_logs_input_s3_default_config() {
+        let logs_input_config_json = serde_json::json!({
+            "type": "s3",
+            "aws_authentication": {
+                "type": "default",
+            }
+        });
+
+        let deserialized =
+            serde_json::from_str::<LogsInput>(logs_input_config_json.to_string().as_str())
+                .expect("failed to deserialize `LogsInput` from JSON");
+
+        match deserialized {
+            LogsInput::S3 { config } => {
+                assert_eq!(
+                    config.aws_authentication,
+                    crate::clp_config::AwsAuthentication::Default
+                );
+            }
+            LogsInput::Fs { .. } => panic!("Expected S3"),
+        }
+    }
+
     #[test]
     fn deserialize_logs_input_fs_config() {
         const DIRECTORY: &str = "/var/logs";

components/clp-rust-utils/src/clp_config/s3_config.rs

@@ -15,6 +15,10 @@ pub struct S3Config {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "type")]
 pub enum AwsAuthentication {
+    /// Uses the default AWS SDK credential provider chain.
+    #[serde(rename = "default")]
+    Default,
+
     #[serde(rename = "credentials")]
     Credentials { credentials: AwsCredentials },
 }

components/clp-rust-utils/src/s3/client.rs

@@ -5,8 +5,14 @@ use aws_sdk_s3::{
 };
 use non_empty_string::NonEmptyString;
 
+use crate::clp_config::AwsAuthentication;
+
 /// Creates a new S3 client.
 ///
+/// When `aws_authentication` is [`AwsAuthentication::Credentials`], the client uses the given
+/// access key pair. When [`AwsAuthentication::Default`], the client uses the default AWS SDK
+/// credential provider chain.
+///
 /// # Notes
 ///
 /// * The client is configured using the latest AWS SDK behavior version.
@@ -17,25 +23,23 @@ use non_empty_string::NonEmptyString;
 /// A newly created S3 client.
 #[must_use]
 pub async fn create_new_client(
-    access_key_id: &str,
-    secret_access_key: &str,
     region_id: &str,
     endpoint: Option<&NonEmptyString>,
+    aws_authentication: &AwsAuthentication,
 ) -> Client {
-    let credentials = Credentials::new(
-        access_key_id,
-        secret_access_key,
-        None,
-        None,
-        "clp-credentials-provider",
-    );
-    let base_config = aws_config::defaults(BehaviorVersion::latest())
-        .credentials_provider(credentials)
-        .region(Region::new(region_id.to_string()))
-        .load()
-        .await;
+    let mut config_defaults =
+        aws_config::defaults(BehaviorVersion::latest()).region(Region::new(region_id.to_string()));
+    if let AwsAuthentication::Credentials { credentials } = aws_authentication {
+        config_defaults = config_defaults.credentials_provider(Credentials::new(
+            credentials.access_key_id.as_str(),
+            credentials.secret_access_key.as_str(),
+            None,
+            None,
+            "clp-credentials-provider",
+        ));
+    }
+    let base_config = config_defaults.load().await;
     let mut config_builder = Builder::from(&base_config).force_path_style(true);
     config_builder.set_endpoint_url(endpoint.map(std::string::ToString::to_string));
-    let config = config_builder.build();
-    Client::from_conf(config)
+    Client::from_conf(config_builder.build())
 }

components/clp-rust-utils/src/sqs/client.rs

@@ -5,31 +5,37 @@ use aws_sdk_sqs::{
 };
 use non_empty_string::NonEmptyString;
 
+use crate::clp_config::AwsAuthentication;
+
 /// Creates a new SQS client.
+///
+/// When `aws_authentication` is [`AwsAuthentication::Credentials`], the client uses the given
+/// access key pair. When [`AwsAuthentication::Default`], the client uses the default AWS SDK
+/// credential provider chain.
+///
 /// The client is configured using the latest AWS SDK behavior version.
 ///
 /// # Returns
 ///
 /// A newly created SQS client.
 #[must_use]
 pub async fn create_new_client(
-    access_key_id: &str,
-    secret_access_key: &str,
     region_id: &str,
     endpoint: Option<&NonEmptyString>,
+    aws_authentication: &AwsAuthentication,
 ) -> Client {
-    let credentials = Credentials::new(
-        access_key_id,
-        secret_access_key,
-        None,
-        None,
-        "clp-credentials-provider",
-    );
-    let base_config = aws_config::defaults(BehaviorVersion::latest())
-        .credentials_provider(credentials)
-        .region(Region::new(region_id.to_string()))
-        .load()
-        .await;
+    let mut config_defaults =
+        aws_config::defaults(BehaviorVersion::latest()).region(Region::new(region_id.to_string()));
+    if let AwsAuthentication::Credentials { credentials } = aws_authentication {
+        config_defaults = config_defaults.credentials_provider(Credentials::new(
+            credentials.access_key_id.as_str(),
+            credentials.secret_access_key.as_str(),
+            None,
+            None,
+            "clp-credentials-provider",
+        ));
+    }
+    let base_config = config_defaults.load().await;
     let mut config_builder = Builder::from(&base_config);
     config_builder.set_endpoint_url(endpoint.map(std::string::ToString::to_string));
     Client::from_conf(config_builder.build())

components/core/CMakeLists.txt

@@ -361,6 +361,10 @@ set(SOURCE_FILES_clp_s_unitTest
     src/clp_s/FileWriter.hpp
     src/clp_s/FloatFormatEncoding.cpp
     src/clp_s/FloatFormatEncoding.hpp
+    src/clp_s/ffi/sfa/ClpArchiveReader.cpp
+    src/clp_s/ffi/sfa/ClpArchiveReader.hpp
+    src/clp_s/ffi/sfa/SfaErrorCode.cpp
+    src/clp_s/ffi/sfa/SfaErrorCode.hpp
     src/clp_s/InputConfig.cpp
     src/clp_s/InputConfig.hpp
     src/clp_s/JsonConstructor.cpp
@@ -500,8 +504,8 @@ set(SOURCE_FILES_unitTest
         src/clp/ffi/ir_stream/decoding_methods.inc
         src/clp/ffi/ir_stream/encoding_methods.cpp
         src/clp/ffi/ir_stream/encoding_methods.hpp
-        src/clp/ffi/ir_stream/IrErrorCode.cpp
-        src/clp/ffi/ir_stream/IrErrorCode.hpp
+        src/clp/ffi/ir_stream/IrDeserializationError.cpp
+        src/clp/ffi/ir_stream/IrDeserializationError.hpp
         src/clp/ffi/ir_stream/IrSerializationError.cpp
         src/clp/ffi/ir_stream/IrSerializationError.hpp
         src/clp/ffi/ir_stream/IrUnitHandlerReq.hpp
@@ -714,6 +718,7 @@ set(SOURCE_FILES_unitTest
         tests/test-BufferedReader.cpp
         tests/test-clp_s-delta-encode-log-order.cpp
         tests/test-clp_s-end_to_end.cpp
+        tests/test-clp_s-ffi_sfa_reader.cpp
         tests/test-clp_s-range_index.cpp
         tests/test-clp_s-search.cpp
         tests/test-EncodedVariableInterpreter.cpp

components/core/src/clp/clg/CMakeLists.txt

@@ -21,6 +21,8 @@ set(
         ../ffi/ir_stream/decoding_methods.cpp
         ../ffi/ir_stream/decoding_methods.hpp
         ../ffi/ir_stream/decoding_methods.inc
+        ../ffi/ir_stream/IrDeserializationError.cpp
+        ../ffi/ir_stream/IrDeserializationError.hpp
         ../ffi/ir_stream/IrSerializationError.cpp
         ../ffi/ir_stream/IrSerializationError.hpp
         ../ffi/StringBlob.hpp

components/core/src/clp/clo/CMakeLists.txt

@@ -27,6 +27,8 @@ set(
         ../ffi/ir_stream/decoding_methods.inc
         ../ffi/ir_stream/encoding_methods.cpp
         ../ffi/ir_stream/encoding_methods.hpp
+        ../ffi/ir_stream/IrDeserializationError.cpp
+        ../ffi/ir_stream/IrDeserializationError.hpp
         ../ffi/ir_stream/IrSerializationError.cpp
         ../ffi/ir_stream/IrSerializationError.hpp
         ../ffi/ir_stream/utils.cpp