This is an automated email from the ASF dual-hosted git repository. szaszm pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 9f8dec516a4aacee35aff698aff0641f985cafd2 Author: Gabor Gyimesi <[email protected]> AuthorDate: Thu Apr 20 18:24:34 2023 +0200 MINIFICPP-2106 Add 'Use Path Style Access' property to PutS3Object Closes #1562 Signed-off-by: Marton Szasz <[email protected]> --- PROCESSORS.md | 3 ++- extensions/aws/processors/PutS3Object.cpp | 6 ++++++ extensions/aws/processors/PutS3Object.h | 5 ++++- extensions/aws/processors/S3ProcessorStaticDefinitions.cpp | 8 ++++++++ extensions/aws/s3/S3ClientRequestSender.cpp | 5 +++-- extensions/aws/s3/S3ClientRequestSender.h | 3 ++- extensions/aws/s3/S3RequestSender.h | 3 ++- extensions/aws/s3/S3Wrapper.cpp | 2 +- extensions/aws/s3/S3Wrapper.h | 1 + extensions/aws/tests/MockS3RequestSender.h | 9 ++++++++- extensions/aws/tests/PutS3ObjectTests.cpp | 9 +++++++++ 11 files changed, 46 insertions(+), 8 deletions(-) diff --git a/PROCESSORS.md b/PROCESSORS.md index 70de1ceed..1fe86736b 100644 --- a/PROCESSORS.md +++ b/PROCESSORS.md @@ -641,7 +641,7 @@ In the list below, the names of required properties appear in bold. Any other pr ### Description -ExecuteJavaClass runs NiFi processors given a provided system path +ExecuteJavaClass runs NiFi processors given a provided system path ### Properties @@ -2248,6 +2248,7 @@ In the list below, the names of required properties appear in bold. Any other pr | Read ACL User List | | [...] | Write ACL User List | | [...] | Canned ACL | | [...] +| **Use Path Style Access** | false | [...] ### Relationships diff --git a/extensions/aws/processors/PutS3Object.cpp b/extensions/aws/processors/PutS3Object.cpp index 22407b896..19b2bd693 100644 --- a/extensions/aws/processors/PutS3Object.cpp +++ b/extensions/aws/processors/PutS3Object.cpp @@ -73,6 +73,10 @@ void PutS3Object::onSchedule(const std::shared_ptr<core::ProcessContext> &contex } logger_->log_debug("PutS3Object: Server Side Encryption [%s]", server_side_encryption_); + if (auto use_path_style_access = context->getProperty<bool>(UsePathStyleAccess)) { + use_virtual_addressing_ = !*use_path_style_access; + } + fillUserMetadata(context); } @@ -152,6 +156,8 @@ std::optional<aws::s3::PutObjectRequestParameters> PutS3Object::buildPutS3Reques if (!setAccessControl(context, flow_file, params)) { return std::nullopt; } + + params.use_virtual_addressing = use_virtual_addressing_; return params; } diff --git a/extensions/aws/processors/PutS3Object.h b/extensions/aws/processors/PutS3Object.h index fd4c910ea..4763c6e2f 100644 --- a/extensions/aws/processors/PutS3Object.h +++ b/extensions/aws/processors/PutS3Object.h @@ -58,6 +58,7 @@ class PutS3Object : public S3Processor { static const core::Property ReadACLUserList; static const core::Property WriteACLUserList; static const core::Property CannedACL; + static const core::Property UsePathStyleAccess; static auto properties() { return minifi::utils::array_cat(S3Processor::properties(), std::array{ ObjectKey, @@ -68,7 +69,8 @@ class PutS3Object : public S3Processor { ReadPermissionUserList, ReadACLUserList, WriteACLUserList, - CannedACL + CannedACL, + UsePathStyleAccess }); } @@ -161,6 +163,7 @@ class PutS3Object : public S3Processor { std::map<std::string, std::string> user_metadata_map_; std::string storage_class_; std::string server_side_encryption_; + bool use_virtual_addressing_ = true; }; } // namespace org::apache::nifi::minifi::aws::processors diff --git a/extensions/aws/processors/S3ProcessorStaticDefinitions.cpp b/extensions/aws/processors/S3ProcessorStaticDefinitions.cpp index f3abed5d1..b79910b08 100644 --- a/extensions/aws/processors/S3ProcessorStaticDefinitions.cpp +++ b/extensions/aws/processors/S3ProcessorStaticDefinitions.cpp @@ -265,6 +265,14 @@ const core::Property PutS3Object::ServerSideEncryption( ->withAllowableValues<std::string>(PutS3Object::SERVER_SIDE_ENCRYPTIONS) ->withDescription("Specifies the algorithm used for server side encryption.") ->build()); +const core::Property PutS3Object::UsePathStyleAccess( + core::PropertyBuilder::createProperty("Use Path Style Access") + ->withDescription("Path-style access can be enforced by setting this property to true. Set it to true if your endpoint does not support " + "virtual-hosted-style requests, only path-style requests.") + ->withDefaultValue<bool>(false) + ->isRequired(true) + ->build()); + const core::Relationship PutS3Object::Success("success", "FlowFiles are routed to success relationship"); const core::Relationship PutS3Object::Failure("failure", "FlowFiles are routed to failure relationship"); diff --git a/extensions/aws/s3/S3ClientRequestSender.cpp b/extensions/aws/s3/S3ClientRequestSender.cpp index d24322e41..e95e57556 100644 --- a/extensions/aws/s3/S3ClientRequestSender.cpp +++ b/extensions/aws/s3/S3ClientRequestSender.cpp @@ -26,8 +26,9 @@ namespace org::apache::nifi::minifi::aws::s3 { std::optional<Aws::S3::Model::PutObjectResult> S3ClientRequestSender::sendPutObjectRequest( const Aws::S3::Model::PutObjectRequest& request, const Aws::Auth::AWSCredentials& credentials, - const Aws::Client::ClientConfiguration& client_config) { - Aws::S3::S3Client s3_client(credentials, client_config, Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, true); + const Aws::Client::ClientConfiguration& client_config, + bool use_virtual_addressing) { + Aws::S3::S3Client s3_client(credentials, client_config, Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, use_virtual_addressing); auto outcome = s3_client.PutObject(request); if (outcome.IsSuccess()) { diff --git a/extensions/aws/s3/S3ClientRequestSender.h b/extensions/aws/s3/S3ClientRequestSender.h index 769d7c6d1..904024b49 100644 --- a/extensions/aws/s3/S3ClientRequestSender.h +++ b/extensions/aws/s3/S3ClientRequestSender.h @@ -30,7 +30,8 @@ class S3ClientRequestSender : public S3RequestSender { std::optional<Aws::S3::Model::PutObjectResult> sendPutObjectRequest( const Aws::S3::Model::PutObjectRequest& request, const Aws::Auth::AWSCredentials& credentials, - const Aws::Client::ClientConfiguration& client_config) override; + const Aws::Client::ClientConfiguration& client_config, + bool use_virtual_addressing) override; bool sendDeleteObjectRequest( const Aws::S3::Model::DeleteObjectRequest& request, const Aws::Auth::AWSCredentials& credentials, diff --git a/extensions/aws/s3/S3RequestSender.h b/extensions/aws/s3/S3RequestSender.h index dd8f04375..c0cce8fee 100644 --- a/extensions/aws/s3/S3RequestSender.h +++ b/extensions/aws/s3/S3RequestSender.h @@ -62,7 +62,8 @@ class S3RequestSender { virtual std::optional<Aws::S3::Model::PutObjectResult> sendPutObjectRequest( const Aws::S3::Model::PutObjectRequest& request, const Aws::Auth::AWSCredentials& credentials, - const Aws::Client::ClientConfiguration& client_config) = 0; + const Aws::Client::ClientConfiguration& client_config, + bool use_virtual_addressing) = 0; virtual bool sendDeleteObjectRequest( const Aws::S3::Model::DeleteObjectRequest& request, const Aws::Auth::AWSCredentials& credentials, diff --git a/extensions/aws/s3/S3Wrapper.cpp b/extensions/aws/s3/S3Wrapper.cpp index dab76976f..ca3aa13e5 100644 --- a/extensions/aws/s3/S3Wrapper.cpp +++ b/extensions/aws/s3/S3Wrapper.cpp @@ -90,7 +90,7 @@ std::optional<PutObjectResult> S3Wrapper::putObject(const PutObjectRequestParame request.SetGrantWriteACP(put_object_params.write_acl_user_list); setCannedAcl(request, put_object_params.canned_acl); - auto aws_result = request_sender_->sendPutObjectRequest(request, put_object_params.credentials, put_object_params.client_config); + auto aws_result = request_sender_->sendPutObjectRequest(request, put_object_params.credentials, put_object_params.client_config, put_object_params.use_virtual_addressing); if (!aws_result) { return std::nullopt; } diff --git a/extensions/aws/s3/S3Wrapper.h b/extensions/aws/s3/S3Wrapper.h index 038d0d029..a622d5e72 100644 --- a/extensions/aws/s3/S3Wrapper.h +++ b/extensions/aws/s3/S3Wrapper.h @@ -126,6 +126,7 @@ struct PutObjectRequestParameters : public RequestParameters { std::string read_acl_user_list; std::string write_acl_user_list; std::string canned_acl; + bool use_virtual_addressing = true; }; struct DeleteObjectRequestParameters : public RequestParameters { diff --git a/extensions/aws/tests/MockS3RequestSender.h b/extensions/aws/tests/MockS3RequestSender.h index 6e4c4b6aa..19e3a0fb6 100644 --- a/extensions/aws/tests/MockS3RequestSender.h +++ b/extensions/aws/tests/MockS3RequestSender.h @@ -94,10 +94,12 @@ class MockS3RequestSender : public minifi::aws::s3::S3RequestSender { std::optional<Aws::S3::Model::PutObjectResult> sendPutObjectRequest( const Aws::S3::Model::PutObjectRequest& request, const Aws::Auth::AWSCredentials& credentials, - const Aws::Client::ClientConfiguration& client_config) override { + const Aws::Client::ClientConfiguration& client_config, + bool use_virtual_addressing) override { put_object_request = request; credentials_ = credentials; client_config_ = client_config; + use_virtual_addressing_ = use_virtual_addressing; Aws::S3::Model::PutObjectResult put_s3_result; if (!return_empty_result_) { @@ -250,6 +252,10 @@ class MockS3RequestSender : public minifi::aws::s3::S3RequestSender { return client_config_; } + bool getUseVirtualAddressing() const { + return use_virtual_addressing_; + } + std::string getPutObjectRequestBody() const { std::istreambuf_iterator<char> buf_it; return std::string(std::istreambuf_iterator<char>(*put_object_request.GetBody()), buf_it); @@ -291,4 +297,5 @@ class MockS3RequestSender : public minifi::aws::s3::S3RequestSender { bool is_listing_truncated_ = false; Aws::Auth::AWSCredentials credentials_; Aws::Client::ClientConfiguration client_config_; + bool use_virtual_addressing_ = true; }; diff --git a/extensions/aws/tests/PutS3ObjectTests.cpp b/extensions/aws/tests/PutS3ObjectTests.cpp index 884bfc7e7..66525c71c 100644 --- a/extensions/aws/tests/PutS3ObjectTests.cpp +++ b/extensions/aws/tests/PutS3ObjectTests.cpp @@ -136,6 +136,7 @@ TEST_CASE_METHOD(PutS3ObjectTestsFixture, "Check default client configuration", REQUIRE(mock_s3_request_sender_ptr->getClientConfig().proxyUserName.empty()); REQUIRE(mock_s3_request_sender_ptr->getClientConfig().proxyPassword.empty()); REQUIRE(mock_s3_request_sender_ptr->getPutObjectRequestBody() == INPUT_DATA); + REQUIRE(mock_s3_request_sender_ptr->getUseVirtualAddressing()); } TEST_CASE_METHOD(PutS3ObjectTestsFixture, "Check default client configuration with empty result", "[awsS3ClientConfig]") { @@ -171,6 +172,7 @@ TEST_CASE_METHOD(PutS3ObjectTestsFixture, "Set non-default client configuration" REQUIRE(mock_s3_request_sender_ptr->getClientConfig().connectTimeoutMs == 10000); REQUIRE(mock_s3_request_sender_ptr->getClientConfig().endpointOverride == "http://localhost:1234";); REQUIRE(mock_s3_request_sender_ptr->getPutObjectRequestBody() == INPUT_DATA); + REQUIRE(mock_s3_request_sender_ptr->getUseVirtualAddressing()); } TEST_CASE_METHOD(PutS3ObjectTestsFixture, "Test single user metadata", "[awsS3MetaData]") { @@ -218,4 +220,11 @@ TEST_CASE_METHOD(PutS3ObjectTestsFixture, "Test access control setting", "[awsS3 REQUIRE(mock_s3_request_sender_ptr->put_object_request.GetACL() == Aws::S3::Model::ObjectCannedACL::public_read_write); } +TEST_CASE_METHOD(PutS3ObjectTestsFixture, "Test path style access property", "[awsS3PathStyleAccess]") { + setRequiredProperties(); + plan->setProperty(s3_processor, "Use Path Style Access", "true"); + test_controller.runSession(plan, true); + REQUIRE(!mock_s3_request_sender_ptr->getUseVirtualAddressing()); +} + } // namespace
