szaszm commented on a change in pull request #1195: URL: https://github.com/apache/nifi-minifi-cpp/pull/1195#discussion_r751241777
########## File path: extensions/azure/processors/DeleteAzureDataLakeStorage.cpp ########## @@ -0,0 +1,84 @@ +/** + * @file DeleteAzureDataLakeStorage.cpp + * DeleteAzureDataLakeStorage class implementation + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "DeleteAzureDataLakeStorage.h" + +#include "utils/ProcessorConfigUtils.h" +#include "utils/gsl.h" +#include "core/Resource.h" + +namespace org::apache::nifi::minifi::azure::processors { + +const core::Relationship DeleteAzureDataLakeStorage::Success("success", "If file deletion from Azure storage succeeds the flowfile is transferred to this relationship"); +const core::Relationship DeleteAzureDataLakeStorage::Failure("failure", "If file deletion from Azure storage fails the flowfile is transferred to this relationship"); + +void DeleteAzureDataLakeStorage::initialize() { + // Set the supported properties + setSupportedProperties({ + AzureStorageCredentialsService, + FilesystemName, + DirectoryName, + FileName + }); + + // Set the supported relationships + setSupportedRelationships({ + Success, + Failure + }); +} + +std::optional<storage::DeleteAzureDataLakeStorageParameters> DeleteAzureDataLakeStorage::buildDeleteParameters( + const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::FlowFile>& flow_file) { + storage::DeleteAzureDataLakeStorageParameters params; + if (!setCommonParameters(params, context, flow_file)) { + return std::nullopt; + } + + return params; +} + +void DeleteAzureDataLakeStorage::onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) { Review comment: ```suggestion void DeleteAzureDataLakeStorage::onTrigger(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSession>& session) { gsl_Expects(context && session); ``` ########## File path: extensions/azure/processors/AzureDataLakeStorageProcessorBase.cpp ########## @@ -0,0 +1,80 @@ +/** + * @file AzureDataLakeStorageProcessorBase.cpp + * AzureDataLakeStorageProcessorBase class implementation + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "AzureDataLakeStorageProcessorBase.h" + +#include "utils/ProcessorConfigUtils.h" +#include "controllerservices/AzureStorageCredentialsService.h" + +namespace org::apache::nifi::minifi::azure::processors { + +const core::Property AzureDataLakeStorageProcessorBase::FilesystemName( + core::PropertyBuilder::createProperty("Filesystem Name") + ->withDescription("Name of the Azure Storage File System. It is assumed to be already existing.") + ->supportsExpressionLanguage(true) + ->isRequired(true) + ->build()); +const core::Property AzureDataLakeStorageProcessorBase::DirectoryName( + core::PropertyBuilder::createProperty("Directory Name") + ->withDescription("Name of the Azure Storage Directory. The Directory Name cannot contain a leading '/'. " + "If left empty it designates the root directory. The directory will be created if not already existing.") + ->supportsExpressionLanguage(true) + ->build()); +const core::Property AzureDataLakeStorageProcessorBase::FileName( + core::PropertyBuilder::createProperty("File Name") + ->withDescription("The filename in Azure Storage. If left empty the filename attribute will be used by default.") + ->supportsExpressionLanguage(true) + ->build()); + +void AzureDataLakeStorageProcessorBase::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) { + std::optional<storage::AzureStorageCredentials> credentials; + std::tie(std::ignore, credentials) = getCredentialsFromControllerService(context); + if (!credentials) { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials Service property missing or invalid"); + } + + if (!credentials->isValid()) { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials Service properties are not set or invalid"); + } + + credentials_ = *credentials; +} + +bool AzureDataLakeStorageProcessorBase::setCommonParameters( + storage::AzureDataLakeStorageParameters& params, const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::FlowFile>& flow_file) { + params.credentials = credentials_; + + if (!context->getProperty(FilesystemName, params.file_system_name, flow_file) || params.file_system_name.empty()) { + logger_->log_error("Filesystem Name '%s' is invalid or empty!", params.file_system_name); + return false; + } + + context->getProperty(DirectoryName, params.directory_name, flow_file); + + context->getProperty(FileName, params.filename, flow_file); Review comment: `context` could be a const reference to ProcessContext. ```suggestion bool AzureDataLakeStorageProcessorBase::setCommonParameters(storage::AzureDataLakeStorageParameters& params, const core::ProcessContext& context, const std::shared_ptr<core::FlowFile>& flow_file) { params.credentials = credentials_; if (!context->getProperty(FilesystemName, params.file_system_name, flow_file) || params.file_system_name.empty()) { logger_->log_error("Filesystem Name '%s' is invalid or empty!", params.file_system_name); return false; } context->getProperty(DirectoryName, params.directory_name, flow_file); context->getProperty(FileName, params.filename, flow_file); ``` ########## File path: extensions/azure/processors/AzureDataLakeStorageProcessorBase.cpp ########## @@ -0,0 +1,80 @@ +/** + * @file AzureDataLakeStorageProcessorBase.cpp + * AzureDataLakeStorageProcessorBase class implementation + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "AzureDataLakeStorageProcessorBase.h" + +#include "utils/ProcessorConfigUtils.h" +#include "controllerservices/AzureStorageCredentialsService.h" + +namespace org::apache::nifi::minifi::azure::processors { + +const core::Property AzureDataLakeStorageProcessorBase::FilesystemName( + core::PropertyBuilder::createProperty("Filesystem Name") + ->withDescription("Name of the Azure Storage File System. It is assumed to be already existing.") + ->supportsExpressionLanguage(true) + ->isRequired(true) + ->build()); +const core::Property AzureDataLakeStorageProcessorBase::DirectoryName( + core::PropertyBuilder::createProperty("Directory Name") + ->withDescription("Name of the Azure Storage Directory. The Directory Name cannot contain a leading '/'. " + "If left empty it designates the root directory. The directory will be created if not already existing.") + ->supportsExpressionLanguage(true) + ->build()); +const core::Property AzureDataLakeStorageProcessorBase::FileName( + core::PropertyBuilder::createProperty("File Name") + ->withDescription("The filename in Azure Storage. If left empty the filename attribute will be used by default.") + ->supportsExpressionLanguage(true) + ->build()); + +void AzureDataLakeStorageProcessorBase::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) { Review comment: ```suggestion void AzureDataLakeStorageProcessorBase::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) { gsl_Expects(context); ``` ########## File path: PROCESSORS.md ########## @@ -1278,10 +1303,10 @@ In the list below, the names of required properties appear in bold. Any other pr | Name | Default Value | Allowable Values | Description | | - | - | - | - | |**Azure Storage Credentials Service**|||Name of the Azure Storage Credentials Service used to retrieve the connection string from.| +|**Conflict Resolution Strategy**|fail|fail<br/>replace<br/>ignore|Indicates what should happen when a file with the same name already exists in the output directory.| Review comment: Can't we just delete the file if it exists? I'm assuming a conflict of delete would be someone else changing the file between our listing and delete command. ########## File path: extensions/azure/processors/AzureDataLakeStorageProcessorBase.cpp ########## @@ -0,0 +1,80 @@ +/** + * @file AzureDataLakeStorageProcessorBase.cpp + * AzureDataLakeStorageProcessorBase class implementation + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "AzureDataLakeStorageProcessorBase.h" + +#include "utils/ProcessorConfigUtils.h" +#include "controllerservices/AzureStorageCredentialsService.h" + +namespace org::apache::nifi::minifi::azure::processors { + +const core::Property AzureDataLakeStorageProcessorBase::FilesystemName( + core::PropertyBuilder::createProperty("Filesystem Name") + ->withDescription("Name of the Azure Storage File System. It is assumed to be already existing.") + ->supportsExpressionLanguage(true) + ->isRequired(true) + ->build()); +const core::Property AzureDataLakeStorageProcessorBase::DirectoryName( + core::PropertyBuilder::createProperty("Directory Name") + ->withDescription("Name of the Azure Storage Directory. The Directory Name cannot contain a leading '/'. " + "If left empty it designates the root directory. The directory will be created if not already existing.") + ->supportsExpressionLanguage(true) + ->build()); +const core::Property AzureDataLakeStorageProcessorBase::FileName( + core::PropertyBuilder::createProperty("File Name") + ->withDescription("The filename in Azure Storage. If left empty the filename attribute will be used by default.") + ->supportsExpressionLanguage(true) + ->build()); + +void AzureDataLakeStorageProcessorBase::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) { + std::optional<storage::AzureStorageCredentials> credentials; + std::tie(std::ignore, credentials) = getCredentialsFromControllerService(context); + if (!credentials) { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials Service property missing or invalid"); + } + + if (!credentials->isValid()) { + throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials Service properties are not set or invalid"); + } Review comment: Isn't this basically the same error message? Couldn't we unify these branches? ```suggestion if (!credentials || !credentials->isValid()) { throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials Service property is missing or invalid"); } ``` ########## File path: extensions/azure/processors/PutAzureDataLakeStorage.cpp ########## @@ -71,18 +55,8 @@ void PutAzureDataLakeStorage::initialize() { }); } -void PutAzureDataLakeStorage::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& /*sessionFactory*/) { - std::optional<storage::AzureStorageCredentials> credentials; - std::tie(std::ignore, credentials) = getCredentialsFromControllerService(context); - if (!credentials) { - throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials Service property missing or invalid"); - } - - if (!credentials->isValid()) { - throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Azure Storage Credentials Service properties are not set or invalid"); - } - - credentials_ = *credentials; +void PutAzureDataLakeStorage::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& sessionFactory) { + AzureDataLakeStorageProcessorBase::onSchedule(context, sessionFactory); Review comment: ```suggestion void PutAzureDataLakeStorage::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& sessionFactory) { gsl_Expects(context && sessionFactory); AzureDataLakeStorageProcessorBase::onSchedule(context, sessionFactory); ``` ########## File path: libminifi/test/azure-tests/AzureDataLakeStorageTestsFixture.h ########## @@ -0,0 +1,123 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <utility> +#include <vector> +#include <memory> +#include <string> + +#include "MockDataLakeStorageClient.h" +#include "../TestBase.h" +#include "utils/TestUtils.h" +#include "utils/IntegrationTestUtils.h" +#include "core/Processor.h" +#include "processors/GetFile.h" +#include "processors/PutFile.h" +#include "processors/LogAttribute.h" +#include "processors/UpdateAttribute.h" +#include "utils/file/FileUtils.h" +#include "controllerservices/AzureStorageCredentialsService.h" + +const std::string FILESYSTEM_NAME = "testfilesystem"; +const std::string DIRECTORY_NAME = "testdir"; +const std::string FILE_NAME = "testfile.txt"; +const std::string CONNECTION_STRING = "test-connectionstring"; +const std::string TEST_DATA = "data123"; +const std::string GETFILE_FILE_NAME = "input_data.log"; + +template<typename AzureDataLakeStorageProcessor> +class AzureDataLakeStorageTestsFixture { + public: + AzureDataLakeStorageTestsFixture() { + LogTestController::getInstance().setDebug<TestPlan>(); + LogTestController::getInstance().setDebug<minifi::core::Processor>(); + LogTestController::getInstance().setTrace<minifi::core::ProcessSession>(); + LogTestController::getInstance().setTrace<minifi::processors::GetFile>(); + LogTestController::getInstance().setTrace<minifi::processors::PutFile>(); + LogTestController::getInstance().setDebug<minifi::processors::UpdateAttribute>(); + LogTestController::getInstance().setDebug<minifi::processors::LogAttribute>(); + LogTestController::getInstance().setTrace<AzureDataLakeStorageProcessor>(); + + // Build MiNiFi processing graph + plan_ = test_controller_.createPlan(); + auto mock_data_lake_storage_client = std::make_unique<MockDataLakeStorageClient>(); + mock_data_lake_storage_client_ptr_ = mock_data_lake_storage_client.get(); + azure_data_lake_storage_ = std::shared_ptr<AzureDataLakeStorageProcessor>( + new AzureDataLakeStorageProcessor("AzureDataLakeStorageProcessor", utils::Identifier(), std::move(mock_data_lake_storage_client))); + auto input_dir = test_controller_.createTempDirectory(); + utils::putFileToDir(input_dir, GETFILE_FILE_NAME, TEST_DATA); + + get_file_ = plan_->addProcessor("GetFile", "GetFile"); + plan_->setProperty(get_file_, minifi::processors::GetFile::Directory.getName(), input_dir); + plan_->setProperty(get_file_, minifi::processors::GetFile::KeepSourceFile.getName(), "false"); + + update_attribute_ = plan_->addProcessor("UpdateAttribute", "UpdateAttribute", { {"success", "d"} }, true); + plan_->addProcessor(azure_data_lake_storage_, "AzureDataLakeStorageProcessor", { {"success", "d"}, {"failure", "d"} }, true); + auto logattribute = plan_->addProcessor("LogAttribute", "LogAttribute", { {"success", "d"} }, true); + logattribute->setAutoTerminatedRelationships({{"success", "d"}}); + + putfile_ = plan_->addProcessor("PutFile", "PutFile", { {"success", "d"} }, false); + plan_->addConnection(azure_data_lake_storage_, {"failure", "d"}, putfile_); + putfile_->setAutoTerminatedRelationships({{"success", "d"}, {"failure", "d"}}); + output_dir_ = test_controller_.createTempDirectory(); + plan_->setProperty(putfile_, org::apache::nifi::minifi::processors::PutFile::Directory.getName(), output_dir_); + + azure_storage_cred_service_ = plan_->addController("AzureStorageCredentialsService", "AzureStorageCredentialsService"); + setDefaultProperties(); + } + + std::vector<std::string> getFailedFlowFileContents() { + std::vector<std::string> file_contents; + + auto lambda = [&file_contents](const std::string& path, const std::string& filename) -> bool { + std::ifstream is(path + utils::file::FileUtils::get_separator() + filename, std::ifstream::binary); + std::string file_content((std::istreambuf_iterator<char>(is)), std::istreambuf_iterator<char>()); + file_contents.push_back(file_content); Review comment: Consider inlining `file_content` to avoid copying the string. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
