diff --git a/.github/references/ubuntu_22_04_clang_arm_manifest.json b/.github/references/ubuntu_22_04_clang_arm_manifest.json index 5f72378a92..d4139e9fbc 100644 --- a/.github/references/ubuntu_22_04_clang_arm_manifest.json +++ b/.github/references/ubuntu_22_04_clang_arm_manifest.json @@ -4144,6 +4144,89 @@ "version": "1.0.0" } }, +{ + "bundles": { + "componentManifest": { + "processors": [ + { + "propertyDescriptors": { + "Max Batch Size": { + "name": "Max Batch Size", + "description": "The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once.", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "false", + "supportedRelationships": [ + { + "name": "enrichment", + "description": "A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + }, + { + "name": "original", + "description": "The incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + } + ], + "typeDescription": "Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and then be routed to the 'enrichment' relationship.", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "org.apache.nifi.minifi.enrichment.ForkEnrichment" + }, + { + "propertyDescriptors": { + "Max Batch Size": { + "name": "Max Batch Size", + "description": "The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once.", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Timeout": { + "name": "Timeout", + "description": "Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the first FlowFile will be routed to the 'timeout' relationship.", + "validator": "TIME_PERIOD_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "true", + "supportedRelationships": [ + { + "name": "invalid", + "description": "Any FlowFiles without the requisite attributes will be routed here" + }, + { + "name": "joined", + "description": "The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship" + }, + { + "name": "original", + "description": "Both of the incoming FlowFiles ('original' and 'enrichment') will be routed to this Relationship. I.e., this is the 'original' version of both of these FlowFiles." + }, + { + "name": "timeout", + "description": "If one of the incoming FlowFiles (i.e., the 'original' FlowFile or the 'enrichment' FlowFile) arrives to this Processor but the other does not arrive within the configured Timeout period, the FlowFile that did arrive is routed to this relationship." + } + ], + "typeDescription": "Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes from both of them (prioritizing Enrichment's).", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "org.apache.nifi.minifi.enrichment.JoinEnrichmentAttributes" + } + ] + }, + "group": "org.apache.nifi.minifi", + "artifact": "minifi-enrichment", + "version": "1.0.0" + } +}, { "bundles": { "componentManifest": { @@ -12104,6 +12187,89 @@ "version": "1.0.0" } }, +{ + "bundles": { + "componentManifest": { + "processors": [ + { + "propertyDescriptors": { + "Batch Size": { + "name": "Batch Size", + "description": "The maximum number of FlowFiles to fork in each trigger", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "false", + "supportedRelationships": [ + { + "name": "enrichment", + "description": "A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + }, + { + "name": "original", + "description": "The incoming FlowFile will be routed to this relationship, after adding appropriate attributes." + } + ], + "typeDescription": "Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and then be routed to the 'enrichment' relationship.", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "minifi_enrichment.processors.fork_enrichment.ForkEnrichment" + }, + { + "propertyDescriptors": { + "Batch Size": { + "name": "Batch Size", + "description": "The maximum number of FlowFiles to process in each trigger", + "validator": "NON_NEGATIVE_INTEGER_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + }, + "Timeout": { + "name": "Timeout", + "description": "Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the first FlowFile will be routed to the 'timeout' relationship.", + "validator": "TIME_PERIOD_VALIDATOR", + "required": "false", + "sensitive": "false", + "expressionLanguageScope": "NONE" + } + }, + "inputRequirement": "INPUT_REQUIRED", + "isSingleThreaded": "true", + "supportedRelationships": [ + { + "name": "invalid", + "description": "Any FlowFiles without the requisite attributes will be routed here" + }, + { + "name": "joined", + "description": "The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship" + }, + { + "name": "original", + "description": "Both of the incoming FlowFiles ('original' and 'enrichment') will be routed to this Relationship. I.e., this is the 'original' version of both of these FlowFiles." + }, + { + "name": "timeout", + "description": "If one of the incoming FlowFiles (i.e., the 'original' FlowFile or the 'enrichment' FlowFile) arrives to this Processor but the other does not arrive within the configured Timeout period, the FlowFile that did arrive is routed to this relationship." + } + ], + "typeDescription": "Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes from both of them (prioritizing Enrichment's).", + "supportsDynamicRelationships": "false", + "supportsDynamicProperties": "false", + "type": "minifi_enrichment.processors.join_enrichment_attributes.JoinEnrichmentAttributes" + } + ] + }, + "group": "org.apache.nifi.minifi.rust.enrichment", + "artifact": "minifi_enrichment", + "version": "0.1.0" + } +}, { "bundles": { "componentManifest": { diff --git a/CONTROLLERS.md b/CONTROLLERS.md index 28fce43e1e..c3ac2fccf5 100644 --- a/CONTROLLERS.md +++ b/CONTROLLERS.md @@ -393,4 +393,3 @@ In the list below, the names of required properties appear in bold. Any other pr | **Pretty Print XML** | false | true
false | Specifies whether or not the XML should be pretty printed | | **Name of Record Tag** | | | Specifies the name of the XML record tag wrapping the record fields. | | **Name of Root Tag** | | | Specifies the name of the XML root tag wrapping the record set. | - diff --git a/PROCESSORS.md b/PROCESSORS.md index 4c7705b2a6..68487069c5 100644 --- a/PROCESSORS.md +++ b/PROCESSORS.md @@ -46,12 +46,14 @@ limitations under the License. - [FetchSFTP](#FetchSFTP) - [FetchSmb](#FetchSmb) - [FocusArchiveEntry](#FocusArchiveEntry) +- [ForkEnrichment](#ForkEnrichment) - [GenerateFlowFile](#GenerateFlowFile) - [GetCouchbaseKey](#GetCouchbaseKey) - [GetFile](#GetFile) - [GetTCP](#GetTCP) - [HashContent](#HashContent) - [InvokeHTTP](#InvokeHTTP) +- [JoinEnrichmentAttributes](#JoinEnrichmentAttributes) - [JoltTransformJSON](#JoltTransformJSON) - [ListAzureBlobStorage](#ListAzureBlobStorage) - [ListAzureDataLakeStorage](#ListAzureDataLakeStorage) @@ -1067,6 +1069,35 @@ In the list below, the names of required properties appear in bold. Any other pr | success | success operational on the flow record | +## ForkEnrichment + +### Description + +Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and then be routed to the 'enrichment' relationship. + +### Properties + +In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. + +| Name | Default Value | Allowable Values | Description | +|----------------|---------------|------------------|-----------------------------------------------------------------------------------------------------------| +| Max Batch Size | | | The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once. | + +### Relationships + +| Name | Description | +|------------|------------------------------------------------------------------------------------------------------------| +| enrichment | A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes. | +| original | The incoming FlowFile will be routed to this relationship, after adding appropriate attributes. | + +### Output Attributes + +| Attribute | Relationship | Description | +|---------------------|----------------------|---------------------------------------------------------------------------------------------------| +| enrichment.role | enrichment, original | The role to use for enrichment. This will either be ORIGINAL or ENRICHMENT. | +| enrichment.group.id | enrichment, original | The Group ID to use in order to correlate the 'original' FlowFile with the 'enrichment' FlowFile. | + + ## GenerateFlowFile ### Description @@ -1277,6 +1308,31 @@ In the list below, the names of required properties appear in bold. Any other pr | invokehttp.tx.id | success, response, retry, no retry | The transaction ID that is returned after reading the response | +## JoinEnrichmentAttributes + +### Description + +Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes from both of them (prioritizing Enrichment's). + +### Properties + +In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language. + +| Name | Default Value | Allowable Values | Description | +|----------------|---------------|------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Timeout | | | Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the first FlowFile will be routed to the 'timeout' relationship. | +| Max Batch Size | | | The maximum number of flow files to process at a time. If unset, all FlowFiles will be processed at once. | + +### Relationships + +| Name | Description | +|----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| invalid | Any FlowFiles without the requisite attributes will be routed here | +| joined | The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship | +| original | Both of the incoming FlowFiles ('original' and 'enrichment') will be routed to this Relationship. I.e., this is the 'original' version of both of these FlowFiles. | +| timeout | If one of the incoming FlowFiles (i.e., the 'original' FlowFile or the 'enrichment' FlowFile) arrives to this Processor but the other does not arrive within the configured Timeout period, the FlowFile that did arrive is routed to this relationship. | + + ## JoltTransformJSON ### Description diff --git a/cmake/MiNiFiOptions.cmake b/cmake/MiNiFiOptions.cmake index 7a898a7658..80059106c6 100644 --- a/cmake/MiNiFiOptions.cmake +++ b/cmake/MiNiFiOptions.cmake @@ -120,6 +120,7 @@ add_minifi_option(ENABLE_CONTROLLER "Enables the build of MiNiFi controller bina add_minifi_option(ENABLE_LLAMACPP "Enables llama.cpp support." ON) add_minifi_option(ENABLE_OPC "Instructs the build system to enable the OPC extension" ON) add_minifi_option(MINIFI_RUST "Enables the build of rust based extensions." OFF) +add_minifi_dependent_option(MINIFI_EXTENSION_ENRICHMENT "Enables the build of enrichment (rust) extension." ON "MINIFI_RUST" OFF) add_minifi_option(MINIFI_LMDB "Enables the LMDB extension." OFF) set_minifi_cache_variable(CUSTOM_MALLOC OFF "Overwrite malloc implementation.") diff --git a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h b/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h index c2a5fb4244..0c953f80ca 100644 --- a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h +++ b/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h @@ -39,4 +39,10 @@ struct EnsureMovedFromDeleter { using FlowFile = std::unique_ptr; +struct NoopStashedFlowFileDeleter { + void operator()(minifi_stashed_flow_file*) const noexcept {} +}; + +using StashedFlowFile = std::unique_ptr; + } // namespace org::apache::nifi::minifi::api::core diff --git a/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h b/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h index 7212046db2..f8bd900816 100644 --- a/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h +++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h @@ -38,12 +38,15 @@ class ProcessSession { ProcessSession& operator=(const ProcessSession&) = delete; ProcessSession& operator=(ProcessSession&&) = delete; + virtual FlowFile clone(const FlowFile& flow_file) = 0; virtual FlowFile create(const FlowFile* parent = nullptr) = 0; virtual FlowFile get() = 0; virtual void penalize(FlowFile& ff) = 0; virtual void transfer(FlowFile ff, const minifi::core::Relationship& relationship) = 0; virtual void remove(FlowFile ff) = 0; + virtual StashedFlowFile stash(FlowFile ff) = 0; + virtual FlowFile unstash(StashedFlowFile stashed_flow_file) = 0; virtual void write(FlowFile& flow, const io::OutputStreamCallback& callback) = 0; virtual void read(FlowFile& flow, const io::InputStreamCallback& callback) = 0; @@ -64,10 +67,13 @@ class CffiProcessSession : public ProcessSession { explicit CffiProcessSession(minifi_process_session* impl): impl_(impl) {} FlowFile create(const FlowFile* parent = nullptr) override; + FlowFile clone(const FlowFile& flow_file) override; FlowFile get() override; void penalize(FlowFile& ff) override; void transfer(FlowFile ff, const minifi::core::Relationship& relationship) override; void remove(FlowFile ff) override; + StashedFlowFile stash(FlowFile ff) override; + FlowFile unstash(StashedFlowFile stashed_flow_file) override; void write(FlowFile& flow, const io::OutputStreamCallback& callback) override; void read(FlowFile& flow, const io::InputStreamCallback& callback) override; diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h index ad3aeb1099..e06f9f5395 100644 --- a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h +++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h @@ -45,6 +45,10 @@ struct minifi_flow_file { std::string id; }; +struct minifi_stashed_flow_file { + std::unique_ptr flow_file; +}; + namespace org::apache::nifi::minifi::mock { class MockProcessSession : public api::core::ProcessSession { @@ -57,9 +61,12 @@ class MockProcessSession : public api::core::ProcessSession { api::core::FlowFile create(const api::core::FlowFile* parent) override; api::core::FlowFile get() override; + api::core::FlowFile clone(const api::core::FlowFile& flow_file) override; void penalize(api::core::FlowFile& ff) override; void transfer(api::core::FlowFile ff, const minifi::core::Relationship& relationship) override; void remove(api::core::FlowFile ff) override; + api::core::StashedFlowFile stash(api::core::FlowFile ff) override; + api::core::FlowFile unstash(api::core::StashedFlowFile stashed_flow_file) override; void write(api::core::FlowFile& ff, const io::OutputStreamCallback& callback) override; void read(api::core::FlowFile& ff, const io::InputStreamCallback& callback) override; void setAttribute(api::core::FlowFile& ff, std::string_view key, std::string value) override; diff --git a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp index 6af5178e5e..30b53046f8 100644 --- a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp +++ b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp @@ -33,6 +33,11 @@ api::core::FlowFile MockProcessSession::get() { return api::core::FlowFile{ff.release()}; } +api::core::FlowFile MockProcessSession::clone(const api::core::FlowFile& flow_file) { + return api::core::FlowFile{new minifi_flow_file(*flow_file.get())}; +} + + void MockProcessSession::penalize(api::core::FlowFile& ff) { ff->is_penalized = true; } @@ -45,6 +50,15 @@ void MockProcessSession::remove(api::core::FlowFile ff) { removed_flow_files_.push_back(std::unique_ptr(ff.release())); } +api::core::StashedFlowFile MockProcessSession::stash(api::core::FlowFile ff) { + return api::core::StashedFlowFile{new minifi_stashed_flow_file{std::unique_ptr(ff.release())}}; +} + +api::core::FlowFile MockProcessSession::unstash(api::core::StashedFlowFile stashed_flow_file) { + const std::unique_ptr owned{stashed_flow_file.release()}; + return api::core::FlowFile{owned->flow_file.release()}; +} + void MockProcessSession::write(api::core::FlowFile& ff, const io::OutputStreamCallback& callback) { const auto stream = std::make_shared(ff->content); callback(stream); diff --git a/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp index 5647bfa118..e244ba51c9 100644 --- a/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp +++ b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp @@ -17,6 +17,7 @@ #include +#include "MockProcessSession.h" #include "minifi-api.h" extern "C" { @@ -75,6 +76,14 @@ minifi_status minifi_process_session_remove(minifi_process_session*, MINIFI_OWNE throw std::runtime_error("Not implemented"); } +minifi_status minifi_process_session_stash(minifi_process_session*, MINIFI_OWNED minifi_flow_file*, minifi_stashed_flow_file**) { + throw std::runtime_error("Not implemented"); +} + +minifi_status minifi_process_session_unstash(minifi_process_session*, MINIFI_OWNED minifi_stashed_flow_file*, minifi_flow_file**) { + throw std::runtime_error("Not implemented"); +} + minifi_status minifi_process_session_read(minifi_process_session*, minifi_flow_file*, int64_t (*)(void* user_ctx, minifi_input_stream*), void*) { throw std::runtime_error("Not implemented"); } @@ -146,4 +155,8 @@ minifi_status minifi_process_context_report_metrics(minifi_process_context*, siz throw std::runtime_error("Not implemented"); } +minifi_flow_file* minifi_process_session_clone(minifi_process_session*, minifi_flow_file*) { + throw std::runtime_error("Not implemented"); +} + } // extern "C" diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp index 41b8be081c..0f232e5c83 100644 --- a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp +++ b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp @@ -77,6 +77,10 @@ FlowFile CffiProcessSession::create(const FlowFile* parent) { return FlowFile{minifi_process_session_create(impl_, parent ? parent->get() : nullptr)}; } +FlowFile CffiProcessSession::clone(const FlowFile& flow_file) { + return FlowFile{minifi_process_session_clone(impl_, flow_file.get())}; +} + void CffiProcessSession::penalize(FlowFile& ff) { if (MINIFI_STATUS_SUCCESS != minifi_process_session_penalize(impl_, ff.get())) { throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to penalize flowfile"); @@ -96,6 +100,22 @@ void CffiProcessSession::remove(FlowFile ff) { } } +StashedFlowFile CffiProcessSession::stash(FlowFile ff) { + minifi_stashed_flow_file* stashed = nullptr; + if (MINIFI_STATUS_SUCCESS != minifi_process_session_stash(impl_, ff.release(), &stashed)) { + throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to stash flowfile"); + } + return StashedFlowFile{stashed}; +} + +FlowFile CffiProcessSession::unstash(StashedFlowFile stashed_flow_file) { + minifi_flow_file* ff = nullptr; + if (MINIFI_STATUS_SUCCESS != minifi_process_session_unstash(impl_, stashed_flow_file.release(), &ff)) { + throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to unstash flowfile"); + } + return FlowFile{ff}; +} + void CffiProcessSession::write(FlowFile& flow_file, const io::OutputStreamCallback& callback) { const auto status = minifi_process_session_write( impl_, diff --git a/extensions/libarchive/BinFiles.cpp b/extensions/libarchive/BinFiles.cpp index 584eff4335..835282ce4b 100644 --- a/extensions/libarchive/BinFiles.cpp +++ b/extensions/libarchive/BinFiles.cpp @@ -215,6 +215,7 @@ bool BinFiles::resurrectFlowFiles(core::ProcessSession &session) { for (auto &file : flow_files) { std::string group_id = getGroupId(file); if (!binManager_.offer(group_id, file)) { + session.add(file); session.transfer(file, Failure); had_failure = true; } diff --git a/extensions/libarchive/tests/MergeFileTests.cpp b/extensions/libarchive/tests/MergeFileTests.cpp index 2ab3fc53a7..9de83dc1f4 100644 --- a/extensions/libarchive/tests/MergeFileTests.cpp +++ b/extensions/libarchive/tests/MergeFileTests.cpp @@ -32,6 +32,7 @@ #include "core/ProcessSession.h" #include "core/ProcessSessionFactory.h" #include "FlowController.h" +#include "FlowFileRecord.h" #include "Connection.h" #include "../../include/core/FlowFile.h" #include "MergeContent.h" @@ -924,3 +925,19 @@ TEST_CASE("Empty MergeContent doesnt yield when processing readybins") { auto second_trigger_results = controller.trigger(); CHECK_FALSE(merge_content->isYield()); } + +TEST_CASE("MergeContent routes a restored flow file rejected by its bin to failure") { + minifi::test::SingleProcessorTestController controller{minifi::test::utils::make_processor("mergeContent")}; + const auto merge_content = controller.getProcessor(); + REQUIRE(controller.plan->setProperty(merge_content, minifi::processors::MergeContent::MergeStrategy, + minifi::processors::merge_content_options::MERGE_STRATEGY_DEFRAGMENT)); + + auto restored_flow_file = std::make_shared(); + restored_flow_file->addAttribute(minifi::processors::BinFiles::FRAGMENT_ID_ATTRIBUTE, "group-1"); + restored_flow_file->addAttribute(minifi::processors::BinFiles::FRAGMENT_COUNT_ATTRIBUTE, "0"); + merge_content->restore(restored_flow_file); + + minifi::test::ProcessorTriggerResult results; + REQUIRE_NOTHROW(results = controller.trigger()); + CHECK(results.at(minifi::processors::MergeContent::Failure).size() == 1); +} diff --git a/extensions/standard-processors/processors/DefragmentText.cpp b/extensions/standard-processors/processors/DefragmentText.cpp index a66eea44f9..1f93ac60c6 100644 --- a/extensions/standard-processors/processors/DefragmentText.cpp +++ b/extensions/standard-processors/processors/DefragmentText.cpp @@ -59,8 +59,10 @@ void DefragmentText::onSchedule(core::ProcessContext& context, core::ProcessSess void DefragmentText::onTrigger(core::ProcessContext&, core::ProcessSession& session) { auto flowFiles = flow_file_store_.getNewFlowFiles(); for (auto& file : flowFiles) { - if (file) + if (file) { + session.add(file); processNextFragment(session, gsl::not_null(file)); + } } { std::shared_ptr original_flow_file = session.get(); diff --git a/extensions/standard-processors/tests/unit/DefragmentTextTests.cpp b/extensions/standard-processors/tests/unit/DefragmentTextTests.cpp index 017c2d1b9a..e25692a7b0 100644 --- a/extensions/standard-processors/tests/unit/DefragmentTextTests.cpp +++ b/extensions/standard-processors/tests/unit/DefragmentTextTests.cpp @@ -23,6 +23,7 @@ #include "unit/ReadFromFlowFileTestProcessor.h" #include "UpdateAttribute.h" #include "DefragmentText.h" +#include "FlowFileRecord.h" #include "TextFragmentUtils.h" #include "serialization/PayloadSerializer.h" #include "serialization/FlowFileSerializer.h" @@ -425,3 +426,49 @@ TEST_CASE("DefragmentText with offset attributes", "[defragmenttextoffsetattribu CHECK(read_from_success_relationship.get().readFlowFileWithContent("%cat,octopus")); } } + +TEST_CASE("DefragmentText routes a restored out-of-order fragment to failure", "[defragmenttextrestore]") { + TestController testController; + auto plan = testController.createPlan(); + auto input_1 = plan->addProcessor("input_1"); + auto defrag_text_flow_files = plan->addProcessor("defrag_text_flow_files"); + auto read_from_failure_relationship = plan->addProcessor("read_from_failure_relationship"); + auto read_from_success_relationship = plan->addProcessor("read_from_success_relationship"); + + plan->addConnection(input_1, FragmentGenerator::Success, defrag_text_flow_files); + plan->addConnection(defrag_text_flow_files, DefragmentText::Failure, read_from_failure_relationship); + plan->addConnection(defrag_text_flow_files, DefragmentText::Success, read_from_success_relationship); + + read_from_failure_relationship.get().disableClearOnTrigger(); + read_from_success_relationship.get().disableClearOnTrigger(); + read_from_failure_relationship->setAutoTerminatedRelationships(std::array{ReadFromFlowFileTestProcessor::Success}); + read_from_success_relationship->setAutoTerminatedRelationships(std::array{ReadFromFlowFileTestProcessor::Success}); + plan->setProperty(defrag_text_flow_files, DefragmentText::Pattern, "%"); + input_1.get().setBaseNameAttribute("input_1"); + input_1.get().setPostNameAttribute("log"); + input_1.get().setAbsolutePathAttribute("/tmp/input/input_1.log"); + + // First trigger: "foo" is flushed to success and "%bar" is left in the buffer, so the buffer + // is non-empty and expects the next fragment to start at offset 7. + input_1.get().setFragments({"foo%bar"}); + testController.runSession(plan); + plan->reset(); + REQUIRE(read_from_success_relationship.get().numberOfFlowFilesRead() == 1); + + // A fragment recovered from the flow file repository after a restart is handed to the processor + // via restore(), so it never passed through a session. Its offset does not continue the buffered + // fragment, which sends it down the transfer-to-failure path in processNextFragment(). + auto restored_fragment = std::make_shared(); + restored_fragment->addAttribute(core::SpecialFlowAttribute::ABSOLUTE_PATH, "/tmp/input/input_1.log"); + restored_fragment->addAttribute(textfragmentutils::BASE_NAME_ATTRIBUTE, "input_1"); + restored_fragment->addAttribute(textfragmentutils::POST_NAME_ATTRIBUTE, "log"); + restored_fragment->addAttribute(textfragmentutils::OFFSET_ATTRIBUTE, "9999"); + defrag_text_flow_files->restore(restored_fragment); + + REQUIRE_NOTHROW(testController.runSession(plan)); + + // Both the flushed buffer and the out-of-order fragment end up on failure. + CHECK(read_from_failure_relationship.get().numberOfFlowFilesRead() == 2); + CHECK(read_from_failure_relationship.get().readFlowFileWithContent("%bar")); + CHECK(read_from_success_relationship.get().numberOfFlowFilesRead() == 1); +} diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h index 72494dd1b7..f88402d8d3 100644 --- a/libminifi/include/core/ProcessSession.h +++ b/libminifi/include/core/ProcessSession.h @@ -79,6 +79,9 @@ class ProcessSessionImpl : public ReferenceContainerImpl, public virtual Process void remove(const std::shared_ptr &flow) override; + StashedFlowFile stash(std::shared_ptr flow_file) override; + std::shared_ptr unstash(StashedFlowFile stashed_flow_file) override; + std::shared_ptr getFlowFileContentStream(const core::FlowFile& flow_file) override; int64_t read(const std::shared_ptr& flow_file, const io::InputStreamCallback& callback) override; diff --git a/libminifi/include/core/Processor.h b/libminifi/include/core/Processor.h index 4f48c09383..9e47a61189 100644 --- a/libminifi/include/core/Processor.h +++ b/libminifi/include/core/Processor.h @@ -123,6 +123,11 @@ class Processor : public ConnectableImpl, public ConfigurableComponentImpl, publ void setLoggerCallback(const std::function& callback); void restore(const std::shared_ptr& file) override; + // Stashing keeps a flow file owned by this processor across onTrigger invocations (used by the + // stable C API's stash/unstash). The owning shared_ptr lives here; sessions hand it in and out. + void stashFlowFile(std::shared_ptr flow_file); + std::shared_ptr unstashFlowFile(const FlowFile* flow_file); + static constexpr auto DynamicProperties = std::array{}; static constexpr auto OutputAttributes = std::array{}; @@ -159,6 +164,9 @@ class Processor : public ConnectableImpl, public ConfigurableComponentImpl, publ mutable std::mutex mutex_; std::atomic yield_expiration_{}; + // Flow files stashed with this processor (owning), keyed by identity via unstashFlowFile. + std::vector> stashed_flow_files_; + // must hold the graphMutex void updateReachability(const std::lock_guard& graph_lock, bool force = false); diff --git a/libminifi/include/utils/minifi-api-utils.h b/libminifi/include/utils/minifi-api-utils.h index f98b608131..54b77bfd75 100644 --- a/libminifi/include/utils/minifi-api-utils.h +++ b/libminifi/include/utils/minifi-api-utils.h @@ -73,6 +73,9 @@ MINIFI_API_MAP(minifi_extension, minifi::core::extension::Extension); MINIFI_API_MAP(minifi_extension_context, minifi::core::extension::Extension::Context); MINIFI_API_MAP(minifi_logger, std::shared_ptr); MINIFI_API_MAP(minifi_flow_file, std::shared_ptr); +// A stashed flow file handle is an opaque, non-owning identity token: the raw FlowFile whose owning +// shared_ptr lives in the processor's stash. See core::StashedFlowFile / Processor::stashFlowFile. +MINIFI_API_MAP(minifi_stashed_flow_file, minifi::core::FlowFile); #undef MINIFI_API_MAP diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp index 4352f7b41c..c1ca7356ef 100644 --- a/libminifi/src/core/ProcessSession.cpp +++ b/libminifi/src/core/ProcessSession.cpp @@ -209,6 +209,35 @@ void ProcessSessionImpl::remove(const std::shared_ptr &flow) { provenance_report_->drop(*flow, reason); } +StashedFlowFile ProcessSessionImpl::stash(std::shared_ptr flow_file) { + logger_->log_trace("Stashing flow file with UUID: {}", flow_file->getUUIDStr()); + const auto uuid = flow_file->getUUID(); + // The session forgets this flow file: on commit it is neither routed, persisted, nor deleted. + updated_flowfiles_.erase(uuid); + updated_relationships_.erase(uuid); + added_flowfiles_.erase(uuid); + std::erase_if(deleted_flowfiles_, [&uuid](const auto& ff) { return ff->getUUID() == uuid; }); + + core::FlowFile* identity = flow_file.get(); + process_context_->getProcessor().stashFlowFile(std::move(flow_file)); + return StashedFlowFile{identity}; +} + +std::shared_ptr ProcessSessionImpl::unstash(StashedFlowFile stashed_flow_file) { + auto record = process_context_->getProcessor().unstashFlowFile(stashed_flow_file.flow_file); + if (!record) { + return nullptr; + } + logger_->log_trace("Unstashing flow file with UUID: {}", record->getUUIDStr()); + const utils::Identifier uuid = record->getUUID(); + if (updated_flowfiles_.contains(uuid)) { + throw Exception(ExceptionType::PROCESSOR_EXCEPTION, "Mustn't unstash a file that was provided by this session"); + } + added_flowfiles_[uuid].flow_file = record; + record->setDeleted(false); + return record; +} + void ProcessSessionImpl::putAttribute(core::FlowFile& flow_file, std::string_view key, const std::string& value) { flow_file.setAttribute(key, value); std::string details = fmt::format("{} modify flow record {} attribute {}:{}", process_context_->getProcessor().getName(), flow_file.getUUIDStr(), key, value); @@ -233,8 +262,11 @@ void ProcessSessionImpl::transfer(const std::shared_ptr& flow, c utils::Identifier uuid = flow->getUUID(); if (auto it = added_flowfiles_.find(uuid); it != added_flowfiles_.end()) { it->second.rel = &*relationships_.insert(relationship).first; - } else { + } else if (updated_flowfiles_.contains(uuid)) { updated_relationships_[uuid] = &*relationships_.insert(relationship).first; + } else { + throw Exception(PROCESS_SESSION_EXCEPTION, + "Cannot transfer flow file " + flow->getUUIDStr() + " that was not obtained from or added to this session"); } flow->setDeleted(false); } diff --git a/libminifi/src/core/Processor.cpp b/libminifi/src/core/Processor.cpp index b20004a092..59675699f0 100644 --- a/libminifi/src/core/Processor.cpp +++ b/libminifi/src/core/Processor.cpp @@ -543,6 +543,23 @@ void Processor::restore(const std::shared_ptr& file) { impl_->restore(file); } +void Processor::stashFlowFile(std::shared_ptr flow_file) { + const std::lock_guard lock(mutex_); + stashed_flow_files_.push_back(std::move(flow_file)); +} + +std::shared_ptr Processor::unstashFlowFile(const FlowFile* flow_file) { + const std::lock_guard lock(mutex_); + for (auto it = stashed_flow_files_.begin(); it != stashed_flow_files_.end(); ++it) { + if (it->get() == flow_file) { + auto result = std::move(*it); + stashed_flow_files_.erase(it); + return result; + } + } + return nullptr; +} + const std::unordered_map>& Processor::reachable_processors() const { return reachable_processors_; } diff --git a/libminifi/src/core/extension/ApiVersion.cpp b/libminifi/src/core/extension/ApiVersion.cpp index e7930fc116..d9d6a91e2f 100644 --- a/libminifi/src/core/extension/ApiVersion.cpp +++ b/libminifi/src/core/extension/ApiVersion.cpp @@ -22,7 +22,7 @@ namespace org::apache::nifi::minifi::core::extension { static uint32_t agent_api_version{MINIFI_API_VERSION}; -static uint32_t min_supported_api_version{MINIFI_API_VERSION}; +static uint32_t min_supported_api_version = 2; // Since version 2 everything was additive only uint32_t getAgentApiVersion() { return agent_api_version; diff --git a/libminifi/src/minifi-api.cpp b/libminifi/src/minifi-api.cpp index 0fae27f2b1..c7a3b6eee8 100644 --- a/libminifi/src/minifi-api.cpp +++ b/libminifi/src/minifi-api.cpp @@ -442,6 +442,15 @@ MINIFI_OWNED minifi_flow_file* minifi_process_session_create(minifi_process_sess return nullptr; } +MINIFI_OWNED minifi_flow_file* minifi_process_session_clone(minifi_process_session* session, minifi_flow_file* flow_file) { + gsl_Assert(session); + gsl_Assert(flow_file); + if (const auto ff = toCpp(session)->clone(*toRawFlowFile(flow_file))) { + return toC(new std::shared_ptr(ff)); // NOLINT(cppcoreguidelines-owning-memory) + } + return nullptr; +} + minifi_status minifi_process_session_penalize(minifi_process_session* session, minifi_flow_file* flowfile) { gsl_Assert(session); gsl_Assert(flowfile); @@ -479,6 +488,36 @@ minifi_status minifi_process_session_remove(minifi_process_session* session, MIN } } +enum minifi_status minifi_process_session_stash(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flow_file, struct minifi_stashed_flow_file** stashed_flow_file) { + gsl_Assert(session); + gsl_Assert(flow_file); + gsl_Assert(stashed_flow_file); + const std::unique_ptr> owned{toCpp(flow_file)}; + try { + *stashed_flow_file = toC(toCpp(session)->stash(std::move(*owned)).flow_file); + return MINIFI_STATUS_SUCCESS; + } catch (...) { + return MINIFI_STATUS_UNKNOWN_ERROR; + } +} + +minifi_status minifi_process_session_unstash(struct minifi_process_session* session, MINIFI_OWNED struct minifi_stashed_flow_file* stashed_flow_file, struct minifi_flow_file** flow_file) { + gsl_Assert(session); + gsl_Assert(flow_file); + gsl_Assert(stashed_flow_file); + try { + auto unstashed = toCpp(session)->unstash(minifi::core::StashedFlowFile{toCpp(stashed_flow_file)}); + if (!unstashed) { + return MINIFI_STATUS_UNKNOWN_ERROR; + } + *flow_file = toC(new std::shared_ptr(std::move(unstashed))); // NOLINT(cppcoreguidelines-owning-memory) + return MINIFI_STATUS_SUCCESS; + } + catch (...) { + return MINIFI_STATUS_UNKNOWN_ERROR; + } +} + minifi_status minifi_process_session_read(minifi_process_session* session, minifi_flow_file* flowfile, int64_t(*cb)(void* user_ctx, minifi_input_stream*), void* user_ctx) { gsl_Assert(session); gsl_Assert(flowfile); diff --git a/libminifi/test/unit/ProcessSessionTests.cpp b/libminifi/test/unit/ProcessSessionTests.cpp index 8c7e99bd61..0bbfc4f0dd 100644 --- a/libminifi/test/unit/ProcessSessionTests.cpp +++ b/libminifi/test/unit/ProcessSessionTests.cpp @@ -166,3 +166,35 @@ TEST_CASE("Stable C API does not leak the flow file handle", "[minifi-api][flowf // FlowFile alive and this assertion fails. REQUIRE(weak_flow_file.expired()); } + +TEST_CASE("Stable C API: stash keeps a flow file across a commit and unstash brings it back", "[minifi-api][flowfilehandle]") { + Fixture fixture; + auto& process_session = fixture.processSession(); + auto* c_session = minifi::utils::toC(&process_session); + + auto created = process_session.create(); + process_session.transfer(created, Success); + process_session.commit(); + + std::weak_ptr weak_flow_file = created; + created.reset(); + + auto* handle = minifi_process_session_get(c_session); + REQUIRE(handle != nullptr); + + // Stash detaches the flow file from the session; the returned stashed handle keeps it alive + // across the commit (it is neither routed nor deleted). + minifi_stashed_flow_file* stashed = nullptr; + REQUIRE(minifi_process_session_stash(c_session, handle, &stashed) == MINIFI_STATUS_SUCCESS); + REQUIRE(stashed != nullptr); + REQUIRE_NOTHROW(process_session.commit()); + REQUIRE_FALSE(weak_flow_file.expired()); + + // Unstash hands it back as a MINIFI_OWNED handle; removing it and committing frees it for good. + minifi_flow_file* unstashed = nullptr; + REQUIRE(minifi_process_session_unstash(c_session, stashed, &unstashed) == MINIFI_STATUS_SUCCESS); + REQUIRE(unstashed != nullptr); + REQUIRE(minifi_process_session_remove(c_session, unstashed) == MINIFI_STATUS_SUCCESS); + process_session.commit(); + REQUIRE(weak_flow_file.expired()); +} diff --git a/minifi-api/include/minifi-api.h b/minifi-api/include/minifi-api.h index f9d1fe078b..0cf11cc98b 100644 --- a/minifi-api/include/minifi-api.h +++ b/minifi-api/include/minifi-api.h @@ -43,7 +43,7 @@ extern "C" { #define MINIFI_PROXY_CONFIGURATION_SERVICE_INTERFACE_PROPERTY_TYPE "org.apache.nifi.minifi.controllers.ProxyConfigurationServiceInterface" enum : uint32_t { - MINIFI_API_VERSION = 2 + MINIFI_API_VERSION = 3 }; enum minifi_io_status : int64_t { @@ -100,6 +100,8 @@ struct minifi_output_stream; struct minifi_extension; struct minifi_extension_context; +struct minifi_stashed_flow_file; // MINIFI_API_VERSION >= 3 + enum minifi_status : uint32_t { MINIFI_STATUS_SUCCESS = 0, MINIFI_STATUS_UNKNOWN_ERROR = 1, @@ -243,12 +245,33 @@ bool minifi_logger_should_log(struct minifi_logger*, enum minifi_log_level); MINIFI_OWNED struct minifi_flow_file* minifi_process_session_get(struct minifi_process_session*); MINIFI_OWNED struct minifi_flow_file* minifi_process_session_create(struct minifi_process_session* session, MINIFI_NULLABLE struct minifi_flow_file* parent_flowfile); +MINIFI_OWNED struct minifi_flow_file* minifi_process_session_clone(struct minifi_process_session* session, + struct minifi_flow_file* flow_file); enum minifi_status minifi_process_session_penalize(struct minifi_process_session* session, struct minifi_flow_file* flowfile); enum minifi_status minifi_process_session_transfer(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flowfile, struct minifi_string_view relationship_name); enum minifi_status minifi_process_session_remove(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flowfile); +// stash/unstash let an extension keep a flow file across onTrigger invocations without the legacy +// self-relationship trick. The flow file's record/content stay in the repositories throughout; what +// moves is where it is owned. +// +// stash: takes a MINIFI_OWNED flow file (e.g. one from get()) out of the current session - on commit +// it is neither routed, persisted, nor deleted - and hands its ownership to the processor, which +// outlives the session. It writes an opaque handle to *stashed_flow_file identifying the stashed +// flow file. The handle is a lightweight token, not a new allocation: there is nothing to free. +// unstash: hands a stashed flow file back to a (later) session. It writes a MINIFI_OWNED flow file to +// *flow_file, which is owned by that session again and must be transferred or removed as usual. +// The stashed_flow_file token is consumed and must not be used again. +// +// Because the processor owns stashed flow files, there is no explicit free: a token that is never +// unstashed simply leaves its flow file in the processor, which reclaims it when the processor is +// torn down (the record/content stay in the repositories and are recovered on next startup). +enum minifi_status minifi_process_session_stash(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flow_file, struct minifi_stashed_flow_file** stashed_flow_file); +enum minifi_status minifi_process_session_unstash(struct minifi_process_session* session, MINIFI_OWNED struct minifi_stashed_flow_file* stashed_flow_file, struct minifi_flow_file** flow_file); + + enum minifi_status minifi_process_session_read(struct minifi_process_session*, struct minifi_flow_file*, int64_t (*cb)(void* user_ctx, struct minifi_input_stream*), void* user_ctx); enum minifi_status minifi_process_session_write(struct minifi_process_session*, struct minifi_flow_file*, diff --git a/minifi-api/include/minifi-cpp/core/FlowFile.h b/minifi-api/include/minifi-cpp/core/FlowFile.h index 951829b45f..8a026478c5 100644 --- a/minifi-api/include/minifi-cpp/core/FlowFile.h +++ b/minifi-api/include/minifi-cpp/core/FlowFile.h @@ -81,4 +81,11 @@ class FlowFile : public virtual CoreComponent, public virtual ReferenceContainer static std::shared_ptr create(); }; +// A non-owning token identifying a flow file stashed with a processor (see +// ProcessSession::stash / Processor::stashFlowFile). The owning shared_ptr lives in the processor; +// this only records the flow file's identity so unstash() can retrieve it. +struct StashedFlowFile { + core::FlowFile* flow_file; +}; + } // namespace org::apache::nifi::minifi::core diff --git a/minifi-api/include/minifi-cpp/core/ProcessSession.h b/minifi-api/include/minifi-cpp/core/ProcessSession.h index 5b77d84229..3820e7a0f1 100644 --- a/minifi-api/include/minifi-cpp/core/ProcessSession.h +++ b/minifi-api/include/minifi-cpp/core/ProcessSession.h @@ -83,6 +83,9 @@ class ProcessSession : public virtual ReferenceContainer { // Penalize the flow virtual void penalize(const std::shared_ptr &flow) = 0; + virtual StashedFlowFile stash(std::shared_ptr flow_file) = 0; + virtual std::shared_ptr unstash(StashedFlowFile stashed_flow_file) = 0; + virtual bool outgoingConnectionsFull(const std::string& relationship) = 0; /** diff --git a/minifi-api/minifi-api.def b/minifi-api/minifi-api.def index a0a0b74add..2670fe8e75 100644 --- a/minifi-api/minifi-api.def +++ b/minifi-api/minifi-api.def @@ -14,6 +14,7 @@ EXPORTS minifi_process_context_get_ssl_data_from_property minifi_process_context_report_metrics minifi_process_context_set_trigger_when_empty + minifi_process_session_clone minifi_process_session_create minifi_process_session_get minifi_process_session_get_flow_file_attribute @@ -24,7 +25,9 @@ EXPORTS minifi_process_session_read minifi_process_session_remove minifi_process_session_set_flow_file_attribute + minifi_process_session_stash minifi_process_session_transfer + minifi_process_session_unstash minifi_process_session_write minifi_register_controller_service minifi_register_extension diff --git a/minifi_rust/CMakeLists.txt b/minifi_rust/CMakeLists.txt index 0ecc1441de..3182e491bf 100644 --- a/minifi_rust/CMakeLists.txt +++ b/minifi_rust/CMakeLists.txt @@ -39,6 +39,10 @@ if (NOT ENABLE_TEST_PROCESSORS) set_target_properties(cargo-build_minifi_rs_playground PROPERTIES EXCLUDE_FROM_ALL TRUE) endif() +if (NOT MINIFI_EXTENSION_ENRICHMENT) + set_target_properties(cargo-build_minifi_enrichment PROPERTIES EXCLUDE_FROM_ALL TRUE) +endif() + include(CTest) find_program(CARGO_NEXTEST_EXECUTABLE cargo-nextest) diff --git a/minifi_rust/extensions/minifi_enrichment/.cargo/config.toml b/minifi_rust/extensions/minifi_enrichment/.cargo/config.toml new file mode 100644 index 0000000000..cb8c02ddc4 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/.cargo/config.toml @@ -0,0 +1,5 @@ +[target.aarch64-apple-darwin] +rustflags = ["-C", "link-arg=-undefined", "-C", "link-arg=dynamic_lookup"] + +[target.x86_64-apple-darwin] +rustflags = ["-C", "link-arg=-undefined", "-C", "link-arg=dynamic_lookup"] diff --git a/minifi_rust/extensions/minifi_enrichment/.gitignore b/minifi_rust/extensions/minifi_enrichment/.gitignore new file mode 100644 index 0000000000..32c9b263bd --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/.gitignore @@ -0,0 +1,7 @@ +target +output +features/.venv +features/output +integration-test/features/.venv +integration-test/features/linux_so +integration-test/.venv diff --git a/minifi_rust/extensions/minifi_enrichment/Cargo.toml b/minifi_rust/extensions/minifi_enrichment/Cargo.toml new file mode 100644 index 0000000000..71f132fd7f --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "minifi_enrichment" +version = "0.1.0" +edition = "2024" + +[lib] +crate-type = ["cdylib"] + +[dependencies] +minifi_native = { path = "../../minifi_native" } +strum = "0.28.0" +walkdir = "2.5.0" +rand = "0.10.0" +hex = "0.4.3" +strum_macros = "0.28.0" +lipsum = "0.9.1" +uuid = { version = "1.24.0", features = ["v4"] } + +[dev-dependencies] +tempfile = "3.22.0" +filetime = "0.2.26" diff --git a/minifi_rust/extensions/minifi_enrichment/features/enrichment.feature b/minifi_rust/extensions/minifi_enrichment/features/enrichment.feature new file mode 100644 index 0000000000..1401abc46a --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/features/enrichment.feature @@ -0,0 +1,47 @@ +# 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. + +@MINIFI_EXTENSION_ENRICHMENT @SUPPORTS_WINDOWS +Feature: ForkEnrichment and JoinEnrichmentAttributes + + Scenario: Merges correctly + Given a GenerateFlowFile processor with the "Custom Text" property set to "original_${literal("content")}" + And the scheduling period of the GenerateFlowFile processor is set to "1 hour" + And the "Data Format" property of the GenerateFlowFile processor is set to "Text" + And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false" + + And a ForkEnrichment processor + And a JoinEnrichmentAttributes processor + + And a ReplaceText processor with the "Evaluation Mode" property set to "Entire text" + And the "Replacement Strategy" property of the ReplaceText processor is set to "Always Replace" + And the "Replacement Value" property of the ReplaceText processor is set to "replaced_content" + + And an UpdateAttribute processor with the "extra_prop" property set to "foo" + + And a LogAttribute processor with the "Log Payload" property set to "true" + + And the "success" relationship of the GenerateFlowFile processor is connected to the ForkEnrichment + And the "original" relationship of the ForkEnrichment processor is connected to the JoinEnrichmentAttributes + And the "enrichment" relationship of the ForkEnrichment processor is connected to the ReplaceText + And the "success" relationship of the ReplaceText processor is connected to the UpdateAttribute + And the "success" relationship of the UpdateAttribute processor is connected to the JoinEnrichmentAttributes + And the "joined" relationship of the JoinEnrichmentAttributes processor is connected to the LogAttribute + And JoinEnrichmentAttributes's original relationship is auto-terminated + And LogAttribute's success relationship is auto-terminated + When the MiNiFi instance starts up + Then the Minifi logs contain the following message: "key:enrichment.role value:JOINED" in less than 10 seconds + And the Minifi logs contain the following message: "key:extra_prop value:foo" in less than 1 second + And the Minifi logs contain the following message: "original_content" in less than 1 second diff --git a/minifi_rust/extensions/minifi_enrichment/features/enrichment_restart.feature b/minifi_rust/extensions/minifi_enrichment/features/enrichment_restart.feature new file mode 100644 index 0000000000..36d3291013 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/features/enrichment_restart.feature @@ -0,0 +1,71 @@ +# 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. + +@MINIFI_EXTENSION_ENRICHMENT +Feature: JoinEnrichmentAttributes recovers released FlowFiles after an agent restart + + # JoinEnrichmentAttributes releases the first half of a pair and holds it in memory until the + # other half arrives. The released FlowFile is not routed, persisted-as-moved, nor deleted on + # commit, so its record stays in the (default, RocksDB-backed) flow file repository. When the + # agent is restarted the held FlowFile must be recovered from disk and re-enqueued, so that the + # join still completes once its pair finally arrives. + # + # The two halves are produced independently with GetFile + UpdateAttribute (instead of + # ForkEnrichment) so their arrival can be separated across the restart: the ORIGINAL is present + # at startup, and its ENRICHMENT pair is only delivered after the restart. + + Scenario: A released FlowFile held across an agent restart is recovered and joined with its late-arriving pair + Given a GetFile processor with the name "GetOriginal" and the "Input Directory" property set to "/tmp/original_input" + And the scheduling period of the GetOriginal processor is set to "100 ms" + And a UpdateAttribute processor with the name "TagOriginal" and the "enrichment.role" property set to "ORIGINAL" + And the "enrichment.group.id" property of the TagOriginal processor is set to "group-1" + + And a GetFile processor with the name "GetEnrichment" and the "Input Directory" property set to "/tmp/enrichment_input" + And the scheduling period of the GetEnrichment processor is set to "100 ms" + And a UpdateAttribute processor with the name "TagEnrichment" and the "enrichment.role" property set to "ENRICHMENT" + And the "enrichment.group.id" property of the TagEnrichment processor is set to "group-1" + + And a JoinEnrichmentAttributes processor + And a PutFile processor with the "Directory" property set to "/tmp/output" + + And the "success" relationship of the GetOriginal processor is connected to the TagOriginal + And the "success" relationship of the TagOriginal processor is connected to the JoinEnrichmentAttributes + And the "success" relationship of the GetEnrichment processor is connected to the TagEnrichment + And the "success" relationship of the TagEnrichment processor is connected to the JoinEnrichmentAttributes + And the "joined" relationship of the JoinEnrichmentAttributes processor is connected to the PutFile + + And JoinEnrichmentAttributes's original relationship is auto-terminated + And JoinEnrichmentAttributes's invalid relationship is auto-terminated + And JoinEnrichmentAttributes's timeout relationship is auto-terminated + And PutFile's success relationship is auto-terminated + And PutFile's failure relationship is auto-terminated + + # Only the ORIGINAL half exists at startup; the ENRICHMENT half arrives after the restart. + And a directory at "/tmp/original_input" has a file with the content "original_content" + + When the MiNiFi instance starts up + # JoinEnrichmentAttributes gets the ORIGINAL, releases it and holds it waiting for its pair, + # so nothing is joined yet. + Then no files are placed in the "/tmp/output" directory in 5 seconds of running time + + # Graceful stop destroys JoinEnrichmentAttributes while it still holds the released ORIGINAL, + # then restart brings the agent back with the persistent repositories intact. + When MiNiFi is stopped + And MiNiFi is restarted + + # The released ORIGINAL is recovered from the flow file repository and re-enqueued. Once its + # ENRICHMENT pair arrives, the join produces a single FlowFile with the ORIGINAL's content. + And a file with the content "enrichment_content" is placed in "/tmp/enrichment_input" + Then a single file with the content "original_content" is placed in the "/tmp/output" directory in less than 60 seconds diff --git a/minifi_rust/extensions/minifi_enrichment/features/environment.py b/minifi_rust/extensions/minifi_enrichment/features/environment.py new file mode 100644 index 0000000000..7e198c4acc --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/features/environment.py @@ -0,0 +1,40 @@ +# 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 +# +# https://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. + +import os + +from minifi_behave.core.hooks import ( + add_extension_to_minifi_container, + common_after_scenario, + common_before_scenario, +) + + +def before_all(context): + dir_path = os.path.dirname(os.path.realpath(__file__)) + build_path = os.path.normpath(os.path.join(dir_path, "../../../target/release/")) + deps_build_path = os.path.normpath(os.path.join(dir_path, "../../../target/release/deps/")) + add_extension_to_minifi_container("minifi_enrichment", [build_path, deps_build_path], context) + + +def before_scenario(context, scenario): + context.minifi_container_image = "apacheminificpp:minifi_enrichment" + common_before_scenario(context, scenario) + + +def after_scenario(context, scenario): + common_after_scenario(context, scenario) diff --git a/minifi_rust/extensions/minifi_enrichment/features/steps/steps.py b/minifi_rust/extensions/minifi_enrichment/features/steps/steps.py new file mode 100644 index 0000000000..ff5817598b --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/features/steps/steps.py @@ -0,0 +1,21 @@ +# 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. + +from minifi_behave.steps import ( + checking_steps, # noqa: F401 + configuration_steps, # noqa: F401 + core_steps, # noqa: F401 + flow_building_steps, # noqa: F401 +) diff --git a/minifi_rust/extensions/minifi_enrichment/src/lib.rs b/minifi_rust/extensions/minifi_enrichment/src/lib.rs new file mode 100644 index 0000000000..850e662ac4 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/lib.rs @@ -0,0 +1,31 @@ +// 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 +// +// https://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. + +mod processors; + +use crate::processors::fork_enrichment::ForkEnrichment; +use crate::processors::join_enrichment_attributes::JoinEnrichmentAttributes; +use minifi_native::{ComplexProcessorType, MultiThreaded, SingleThreaded}; + +minifi_native::declare_minifi_extension!( + group_name: "org.apache.nifi.minifi.rust.enrichment", + processors: [ + (ComplexProcessorType, MultiThreaded, ForkEnrichment), + (ComplexProcessorType, SingleThreaded, JoinEnrichmentAttributes), + ], + controllers: [] +); diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/attributes.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/attributes.rs new file mode 100644 index 0000000000..bc5129aae1 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/attributes.rs @@ -0,0 +1,56 @@ +// 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 +// +// https://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. + +use minifi_native::OutputAttribute; +use strum_macros::{EnumString, IntoStaticStr}; + +const ENRICHMENT_ROLE: &str = "enrichment.role"; +const ENRICHMENT_GROUP_ID: &str = "enrichment.group.id"; + +/// The roles ForkEnrichment writes and JoinEnrichmentAttributes pairs on. +/// +/// `EnumString` parses the attribute back; `IntoStaticStr` writes it, so the wire strings +/// (`ORIGINAL` / `ENRICHMENT`) have exactly one definition. +#[derive(Debug, Clone, Copy, PartialEq, EnumString, IntoStaticStr)] +#[strum(serialize_all = "UPPERCASE", const_into_str)] +pub(crate) enum Role { + Original, + Enrichment, +} + +/// The role a joined FlowFile gets. Deliberately not a [`Role`] variant: `Role` is the set of roles +/// this processor *pairs* on, and a `JOINED` flow file fed back in must not parse as a pairable +/// half - it should fall through to the `invalid` relationship. +pub(crate) const JOINED_ROLE: &str = "JOINED"; + +pub(crate) const FORK_ROLE_ATTR: OutputAttribute = OutputAttribute { + name: ENRICHMENT_ROLE, + relationships: &["enrichment", "original"], + description: "The role to use for enrichment. This will either be ORIGINAL or ENRICHMENT.", +}; + +pub(crate) const GROUP_ID_ATTR: OutputAttribute = OutputAttribute { + name: ENRICHMENT_GROUP_ID, + relationships: &["enrichment", "original"], + description: "The Group ID to use in order to correlate the 'original' FlowFile with the 'enrichment' FlowFile.", +}; + +pub(crate) const JOIN_ROLE_ATTR: OutputAttribute = OutputAttribute { + name: ENRICHMENT_ROLE, + relationships: &["joined"], + description: "JOINED", +}; diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment.rs new file mode 100644 index 0000000000..eb65771d56 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment.rs @@ -0,0 +1,151 @@ +// 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 +// +// https://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. + +mod fork_enrichment_def; + +use crate::processors::attributes::{FORK_ROLE_ATTR, GROUP_ID_ATTR, Role}; +use crate::processors::fork_enrichment::fork_enrichment_def::{ENRICHMENT, ORIGINAL}; +use minifi_native::macros::ComponentIdentifier; +use minifi_native::{ + GetProperty, Logger, MinifiError, OnTriggerResult, ProcessContext, ProcessSession, Schedule, + Trigger, +}; +use uuid::Uuid; + +#[derive(ComponentIdentifier)] +pub(crate) struct ForkEnrichment {} + +impl Schedule for ForkEnrichment { + fn schedule( + _context: &Ctx, + _logger: &L, + ) -> Result + where + Self: Sized, + { + Ok(Self {}) + } +} + +impl Trigger for ForkEnrichment { + fn trigger( + &self, + _context: &mut Context, + session: &mut Session, + _logger: &Lggr, + ) -> Result + where + Context: ProcessContext, + Session: ProcessSession, + Lggr: Logger, + { + let Some(mut original) = session.get() else { + return Ok(OnTriggerResult::Yield); + }; + let mut enrichment = session.clone_ff(&original)?; + session.set_attribute(&mut original, FORK_ROLE_ATTR.name, Role::Original.into())?; + session.set_attribute( + &mut enrichment, + FORK_ROLE_ATTR.name, + Role::Enrichment.into(), + )?; + + let group_id = Uuid::new_v4().to_string(); + session.set_attribute(&mut original, GROUP_ID_ATTR.name, &group_id)?; + session.set_attribute(&mut enrichment, GROUP_ID_ATTR.name, &group_id)?; + + session.transfer(original, ORIGINAL.name)?; + session.transfer(enrichment, ENRICHMENT.name)?; + + Ok(OnTriggerResult::Ok) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use minifi_native::{ + ComponentIdentifier, MockFlowFile, MockLogger, MockProcessContext, MockProcessSession, + }; + + #[test] + fn component_id() { + assert_eq!( + ForkEnrichment::CLASS_NAME, + "minifi_enrichment::processors::fork_enrichment::ForkEnrichment" + ); + assert_eq!(ForkEnrichment::GROUP_NAME, "minifi_enrichment"); + } + + #[test] + fn forks_into_original_and_enrichment_sharing_a_group_id() { + let logger = MockLogger::new(); + let mut context = MockProcessContext::new(); + let processor = ForkEnrichment::schedule(&context, &logger).unwrap(); + + let mut session = MockProcessSession::new(); + session + .input_flow_files + .push(MockFlowFile::with_content(b"hello")); + + processor + .trigger(&mut context, &mut session, &logger) + .expect("trigger should succeed"); + + let transferred = session.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 2); + + let original = transferred + .iter() + .find(|t| t.relationship == ORIGINAL.name) + .expect("an original FlowFile should be transferred"); + let enrichment = transferred + .iter() + .find(|t| t.relationship == ENRICHMENT.name) + .expect("an enrichment FlowFile should be transferred"); + + assert_eq!( + original + .flow_file + .attributes + .get(FORK_ROLE_ATTR.name) + .unwrap(), + "ORIGINAL" + ); + assert_eq!( + enrichment + .flow_file + .attributes + .get(FORK_ROLE_ATTR.name) + .unwrap(), + "ENRICHMENT" + ); + + let orig_group = original + .flow_file + .attributes + .get(GROUP_ID_ATTR.name) + .unwrap(); + let enr_group = enrichment + .flow_file + .attributes + .get(GROUP_ID_ATTR.name) + .unwrap(); + assert_eq!(orig_group, enr_group); + assert!(!orig_group.is_empty()); + } +} diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment/fork_enrichment_def.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment/fork_enrichment_def.rs new file mode 100644 index 0000000000..18b2508c08 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/fork_enrichment/fork_enrichment_def.rs @@ -0,0 +1,44 @@ +// 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 +// +// https://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. + +use crate::processors::attributes::{FORK_ROLE_ATTR, GROUP_ID_ATTR}; +use crate::processors::fork_enrichment::ForkEnrichment; +use minifi_native::{ + OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, PropertyDefinition, + Relationship, property_definitions, +}; + +pub(crate) const ORIGINAL: Relationship = Relationship { + name: "original", + description: "The incoming FlowFile will be routed to this relationship, after adding appropriate attributes.", +}; + +pub(crate) const ENRICHMENT: Relationship = Relationship { + name: "enrichment", + description: "A clone of the incoming FlowFile will be routed to this relationship, after adding appropriate attributes.", +}; + +impl ProcessorDefinition for ForkEnrichment { + const DESCRIPTION: &'static str = "Used in conjunction with the JoinEnrichmentAttributes processor, this processor is responsible for adding the attributes that are necessary for the JoinEnrichmentAttributes processor to perform its function. Each incoming FlowFile will be cloned. The original FlowFile will have appropriate attributes added and then be transferred to the 'original' relationship. The clone will have appropriate attributes added and then be routed to the 'enrichment' relationship."; + const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required; + const SUPPORTS_DYNAMIC_PROPERTIES: bool = false; + const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; + const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[FORK_ROLE_ATTR, GROUP_ID_ATTR]; + const RELATIONSHIPS: &'static [Relationship] = &[ORIGINAL, ENRICHMENT]; + + const PROPERTIES: &[PropertyDefinition] = property_definitions![]; +} diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes.rs new file mode 100644 index 0000000000..b8a5badd05 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes.rs @@ -0,0 +1,383 @@ +// 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 +// +// https://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. + +use std::time::{Duration, Instant}; + +use minifi_native::macros::ComponentIdentifier; +use minifi_native::{ + FlowFileStore, Logger, MinifiError, MutTrigger, OnTriggerResult, ProcessContext, + ProcessSession, Schedule, ScheduleContext, warn, +}; + +use crate::processors::attributes::{ + FORK_ROLE_ATTR, GROUP_ID_ATTR, JOIN_ROLE_ATTR, JOINED_ROLE, Role, +}; +use crate::processors::join_enrichment_attributes::join_enrichment_attributes_def::{ + BATCH_SIZE, INVALID, JOINED, ORIGINAL, TIMEOUT_PROP, TIMEOUT_REL, +}; + +mod join_enrichment_attributes_def; + +#[derive(ComponentIdentifier)] +pub(crate) struct JoinEnrichmentAttributes { + batch_size: usize, + timeout: Option, + pending: FlowFileStore, +} + +fn get_role( + session: &Session, + flow_file: &Session::FlowFile, +) -> Option { + session + .get_attribute(flow_file, FORK_ROLE_ATTR.name)? + .parse() + .ok() +} + +fn get_role_and_group_id( + session: &Session, + flow_file: &Session::FlowFile, +) -> Option<(Role, String)> { + let role = get_role(session, flow_file)?; + let group_id = session + .get_required_attribute(flow_file, GROUP_ID_ATTR.name) + .ok()?; + Some((role, group_id)) +} + +fn join( + session: &mut Session, + original_ff: Session::FlowFile, + enrichment_ff: Session::FlowFile, +) -> Result<(), MinifiError> { + let mut joined_ff = session.clone_ff(&original_ff)?; + + let mut enrichment_attrs: Vec<(String, String)> = Vec::new(); + session.for_each_attribute(&enrichment_ff, |key, value| { + enrichment_attrs.push((key.to_string(), value.to_string())); + }); + for (key, value) in &enrichment_attrs { + session.set_attribute(&mut joined_ff, key, value)?; + } + + session.set_attribute(&mut joined_ff, JOIN_ROLE_ATTR.name, JOINED_ROLE)?; + + session.transfer(original_ff, ORIGINAL.name)?; + session.transfer(enrichment_ff, ORIGINAL.name)?; + session.transfer(joined_ff, JOINED.name)?; + Ok(()) +} + +impl JoinEnrichmentAttributes { + fn handle_flow_file( + &mut self, + incoming: Session::FlowFile, + session: &mut Session, + role: Role, + group_id: String, + ) -> Result<(), MinifiError> + where + Session: ProcessSession, + { + let Some(pending) = self.pending.take(session, &group_id)? else { + return self.pending.store(session, group_id, incoming); + }; + + match (role, get_role(session, &pending)) { + (Role::Original, Some(Role::Enrichment)) => join(session, incoming, pending), + (Role::Enrichment, Some(Role::Original)) => join(session, pending, incoming), + _ => { + session.transfer(incoming, INVALID.name)?; + session.transfer(pending, INVALID.name)?; + Ok(()) + } + } + } +} + +impl Schedule for JoinEnrichmentAttributes { + fn schedule( + context: &Ctx, + _logger: &L, + ) -> Result + where + Self: Sized, + { + let timeout = context + .get_property(&TIMEOUT_PROP)? + .filter(|d| !d.is_zero()); + + if timeout.is_some() { + context.set_trigger_when_empty(true)?; + } + + Ok(Self { + batch_size: context.get_property(&BATCH_SIZE)?.unwrap_or(usize::MAX), + timeout, + pending: FlowFileStore::new(), + }) + } +} + +impl JoinEnrichmentAttributes { + fn on_trigger( + &mut self, + session: &mut Session, + logger: &Lggr, + now: Instant, + ) -> Result + where + Session: ProcessSession, + Lggr: Logger, + { + for _ in 0..self.batch_size { + let Some(flow_file) = session.get() else { + break; + }; + + let Some((role, group_id)) = get_role_and_group_id(session, &flow_file) else { + warn!(logger, "Missing required attribute"); + session.transfer(flow_file, INVALID.name)?; + continue; + }; + + self.handle_flow_file(flow_file, session, role, group_id)?; + } + + if let Some(timeout) = self.timeout { + for timed_out in self.pending.drain_expired(session, now, timeout)? { + session.transfer(timed_out, TIMEOUT_REL.name)?; + } + } + + Ok(OnTriggerResult::Ok) + } +} + +impl MutTrigger for JoinEnrichmentAttributes { + fn trigger( + &mut self, + _context: &mut Context, + session: &mut Session, + logger: &Lggr, + ) -> Result + where + Context: ProcessContext, + Session: ProcessSession, + Lggr: Logger, + { + self.on_trigger(session, logger, Instant::now()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use minifi_native::{MockFlowFile, MockLogger, MockProcessContext, MockProcessSession}; + + fn flow_file(role: &str, group_id: &str, content: &[u8]) -> MockFlowFile { + let mut ff = MockFlowFile::with_content(content); + ff.attributes + .insert(FORK_ROLE_ATTR.name.to_string(), role.to_string()); + ff.attributes + .insert(GROUP_ID_ATTR.name.to_string(), group_id.to_string()); + ff + } + + fn scheduled() -> (JoinEnrichmentAttributes, MockProcessContext, MockLogger) { + let logger = MockLogger::new(); + let context = MockProcessContext::new(); + let processor = JoinEnrichmentAttributes::schedule(&context, &logger).unwrap(); + (processor, context, logger) + } + + #[test] + fn a_configured_timeout_requests_triggering_on_an_empty_input() { + let logger = MockLogger::new(); + let mut context = MockProcessContext::new(); + context.properties.insert(TIMEOUT_PROP.name(), "60 s"); + + let processor = JoinEnrichmentAttributes::schedule(&context, &logger).unwrap(); + + assert!(context.trigger_when_empty.get()); + assert_eq!(processor.timeout, Some(Duration::from_secs(60))); + } + + #[test] + fn without_a_timeout_we_do_not_ask_to_be_triggered_on_an_empty_input() { + let (processor, context, _logger) = scheduled(); + assert!(!context.trigger_when_empty.get()); + assert_eq!(processor.timeout, None); + } + + fn with_timeout(timeout: Duration) -> JoinEnrichmentAttributes { + JoinEnrichmentAttributes { + batch_size: usize::MAX, + timeout: Some(timeout), + pending: FlowFileStore::new(), + } + } + + #[test] + fn joins_original_and_enrichment_arriving_on_separate_triggers() { + let (mut processor, mut context, logger) = scheduled(); + + // Trigger 1: only the original arrives -> stored, nothing transferred. + let mut session1 = MockProcessSession::new(); + session1 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"original-content")); + processor + .trigger(&mut context, &mut session1, &logger) + .unwrap(); + assert_eq!(session1.num_of_transferred_flow_files(), 0); + + // Trigger 2: the enrichment arrives (with an extra attribute) -> join. + let mut enrichment = flow_file("ENRICHMENT", "g1", b"enrichment-content"); + enrichment + .attributes + .insert("enriched".to_string(), "yes".to_string()); + let mut session2 = MockProcessSession::new(); + session2.input_flow_files.push(enrichment); + processor + .trigger(&mut context, &mut session2, &logger) + .unwrap(); + + let transferred = session2.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 3); + assert_eq!( + transferred + .iter() + .filter(|t| t.relationship == ORIGINAL.name) + .count(), + 2 + ); + + let joined = transferred + .iter() + .find(|t| t.relationship == JOINED.name) + .expect("a joined FlowFile should be transferred"); + // Original's content is preserved. + assert!(joined.flow_file.content_eq("original-content")); + // Enrichment's attributes are merged in, and the role is set to JOINED. + assert_eq!(joined.flow_file.attributes.get("enriched").unwrap(), "yes"); + assert_eq!( + joined + .flow_file + .attributes + .get(JOIN_ROLE_ATTR.name) + .unwrap(), + "JOINED" + ); + } + + #[test] + fn missing_required_attribute_routes_to_invalid() { + let (mut processor, mut context, logger) = scheduled(); + + let mut session = MockProcessSession::new(); + session + .input_flow_files + .push(MockFlowFile::with_content(b"no attributes")); + processor + .trigger(&mut context, &mut session, &logger) + .unwrap(); + + let transferred = session.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 1); + assert_eq!(transferred[0].relationship, INVALID.name); + } + + #[test] + fn duplicate_role_for_a_group_routes_both_to_invalid() { + let (mut processor, mut context, logger) = scheduled(); + + // Trigger 1: first original stored. + let mut session1 = MockProcessSession::new(); + session1 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"first")); + processor + .trigger(&mut context, &mut session1, &logger) + .unwrap(); + assert_eq!(session1.num_of_transferred_flow_files(), 0); + + // Trigger 2: a second original for the same group -> both invalid. + let mut session2 = MockProcessSession::new(); + session2 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"second")); + processor + .trigger(&mut context, &mut session2, &logger) + .unwrap(); + + let transferred = session2.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 2); + assert!(transferred.iter().all(|t| t.relationship == INVALID.name)); + } + + #[test] + fn unpaired_half_is_routed_to_timeout_after_the_timeout_elapses() { + let logger = MockLogger::new(); + let mut processor = with_timeout(Duration::from_secs(60)); + let start = Instant::now(); + + // Trigger 1: the original arrives and is stored; nothing is emitted yet. + let mut session1 = MockProcessSession::new(); + session1 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"lonely")); + processor.on_trigger(&mut session1, &logger, start).unwrap(); + assert_eq!(session1.num_of_transferred_flow_files(), 0); + + // Trigger 2, long past the timeout, with no pair: the stored half is + // handed back and routed to `timeout`. + let mut session2 = MockProcessSession::new(); + processor + .on_trigger(&mut session2, &logger, start + Duration::from_secs(120)) + .unwrap(); + + let transferred = session2.transferred_flow_files.borrow(); + assert_eq!(transferred.len(), 1); + assert_eq!(transferred[0].relationship, TIMEOUT_REL.name); + assert!(transferred[0].flow_file.content_eq("lonely")); + assert!(processor.pending.is_empty()); + } + + #[test] + fn half_within_the_timeout_window_is_kept() { + let logger = MockLogger::new(); + let mut processor = with_timeout(Duration::from_secs(60)); + let start = Instant::now(); + + let mut session1 = MockProcessSession::new(); + session1 + .input_flow_files + .push(flow_file("ORIGINAL", "g1", b"waiting")); + processor.on_trigger(&mut session1, &logger, start).unwrap(); + + // A later trigger still inside the timeout window: nothing is emitted and + // the half is still pending its pair. + let mut session2 = MockProcessSession::new(); + processor + .on_trigger(&mut session2, &logger, start + Duration::from_secs(1)) + .unwrap(); + assert_eq!(session2.num_of_transferred_flow_files(), 0); + assert_eq!(processor.pending.len(), 1); + } +} diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes/join_enrichment_attributes_def.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes/join_enrichment_attributes_def.rs new file mode 100644 index 0000000000..add4aa6c57 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/join_enrichment_attributes/join_enrichment_attributes_def.rs @@ -0,0 +1,65 @@ +// 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 +// +// https://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. + +use crate::processors::attributes::JOIN_ROLE_ATTR; +use crate::processors::join_enrichment_attributes::JoinEnrichmentAttributes; +use minifi_native::{ + OutputAttribute, ProcessorDefinition, ProcessorInputRequirement, Property, PropertyDefinition, + Relationship, property_definitions, +}; +use std::time::Duration; + +pub(crate) const INVALID: Relationship = Relationship { + name: "invalid", + description: "Any FlowFiles without the requisite attributes will be routed here", +}; + +pub(crate) const JOINED: Relationship = Relationship { + name: "joined", + description: "The resultant FlowFile with Records joined together from both the original and enrichment FlowFiles will be routed to this relationship", +}; + +pub(crate) const ORIGINAL: Relationship = Relationship { + name: "original", + description: "Both of the incoming FlowFiles ('original' and 'enrichment') will be routed to this Relationship. I.e., this is the 'original' version of both of these FlowFiles.", +}; + +pub(crate) const TIMEOUT_REL: Relationship = Relationship { + name: "timeout", + description: "If one of the incoming FlowFiles (i.e., the 'original' FlowFile or the 'enrichment' FlowFile) arrives to this Processor but the other does not arrive within the configured Timeout period, the FlowFile that did arrive is routed to this relationship.", +}; + +pub(crate) const BATCH_SIZE: Property> = Property::new( + "Batch Size", + "The maximum number of FlowFiles to process in each trigger", +); + +pub(crate) const TIMEOUT_PROP: Property> = Property::new( + "Timeout", + "Specifies the maximum amount of time to wait for the second FlowFile once the first arrives at the processor, after which point the first FlowFile will be routed to the 'timeout' relationship.", +); + +impl ProcessorDefinition for JoinEnrichmentAttributes { + const DESCRIPTION: &'static str = "Rejoins the forked FlowFiles coming from ForkEnrichment processor, the resulting FlowFile will have the Original's content and all attributes from both of them (prioritizing Enrichment's)."; + const INPUT_REQUIREMENT: ProcessorInputRequirement = ProcessorInputRequirement::Required; + const SUPPORTS_DYNAMIC_PROPERTIES: bool = false; + const SUPPORTS_DYNAMIC_RELATIONSHIPS: bool = false; + const OUTPUT_ATTRIBUTES: &'static [OutputAttribute] = &[JOIN_ROLE_ATTR]; + const RELATIONSHIPS: &'static [Relationship] = &[INVALID, JOINED, ORIGINAL, TIMEOUT_REL]; + + const PROPERTIES: &[PropertyDefinition] = property_definitions![BATCH_SIZE, TIMEOUT_PROP]; +} diff --git a/minifi_rust/extensions/minifi_enrichment/src/processors/mod.rs b/minifi_rust/extensions/minifi_enrichment/src/processors/mod.rs new file mode 100644 index 0000000000..c55651fe43 --- /dev/null +++ b/minifi_rust/extensions/minifi_enrichment/src/processors/mod.rs @@ -0,0 +1,20 @@ +// 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 +// +// https://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. + +pub(crate) mod attributes; +pub(crate) mod fork_enrichment; +pub(crate) mod join_enrichment_attributes; diff --git a/minifi_rust/minifi_native/src/api.rs b/minifi_rust/minifi_native/src/api.rs index 49f3d88d51..46c202c937 100644 --- a/minifi_rust/minifi_native/src/api.rs +++ b/minifi_rust/minifi_native/src/api.rs @@ -20,6 +20,7 @@ pub(crate) mod component_definition_traits; pub(crate) mod controller_service; pub(crate) mod errors; mod flow_file; +pub(crate) mod flow_file_store; pub(crate) mod logger; mod process_context; pub(crate) mod process_session; @@ -32,8 +33,9 @@ pub(crate) mod raw_processor; pub(crate) mod relationship; pub use flow_file::{FlowFile, GetId}; +pub use flow_file_store::FlowFileStore; pub use logger::{LogLevel, Logger}; -pub use process_context::ProcessContext; +pub use process_context::{ProcessContext, ScheduleContext}; pub use process_session::{InputStream, OutputStream, ProcessSession}; pub use raw_controller_service::RawControllerService; pub use raw_processor::{OnTriggerResult, ProcessorInputRequirement, RawProcessor, ThreadingModel}; diff --git a/minifi_rust/minifi_native/src/api/flow_file_store.rs b/minifi_rust/minifi_native/src/api/flow_file_store.rs new file mode 100644 index 0000000000..87646bf55e --- /dev/null +++ b/minifi_rust/minifi_native/src/api/flow_file_store.rs @@ -0,0 +1,200 @@ +// 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 +// +// https://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. + +use std::any::Any; +use std::collections::HashMap; +use std::hash::Hash; +use std::time::{Duration, Instant}; + +use crate::MinifiError; +use crate::api::ProcessSession; + +struct StoredEntry { + flow_file: Box, + stored_at: Instant, +} + +pub struct FlowFileStore { + stored: HashMap, +} + +impl FlowFileStore { + pub fn new() -> Self { + Self { + stored: HashMap::new(), + } + } + + pub fn store( + &mut self, + session: &mut S, + key: K, + flow_file: S::FlowFile, + ) -> Result<(), MinifiError> + where + S: ProcessSession, + { + let stashed = session.stash(flow_file)?; + self.stored.insert( + key, + StoredEntry { + flow_file: Box::new(stashed), + stored_at: Instant::now(), + }, + ); + Ok(()) + } + + pub fn take(&mut self, session: &mut S, key: &K) -> Result, MinifiError> + where + S: ProcessSession, + { + let Some(entry) = self.stored.remove(key) else { + return Ok(None); + }; + Ok(Some(Self::add_back(session, entry)?)) + } + + pub fn drain_expired( + &mut self, + session: &mut S, + now: Instant, + max_age: Duration, + ) -> Result, MinifiError> + where + S: ProcessSession, + K: Clone, + { + let expired_keys: Vec = self + .stored + .iter() + .filter(|(_, entry)| now.saturating_duration_since(entry.stored_at) > max_age) + .map(|(key, _)| key.clone()) + .collect(); + + let mut expired = Vec::with_capacity(expired_keys.len()); + for key in expired_keys { + let entry = self.stored.remove(&key).expect("key came from the map"); + expired.push(Self::add_back(session, entry)?); + } + Ok(expired) + } + + fn add_back(session: &mut S, entry: StoredEntry) -> Result + where + S: ProcessSession, + { + // The entry is type-erased because a processor struct cannot name the session's + // `StashedFlowFile`: `Session` is a generic on `trigger`, not on the processor, and the same + // processor is driven by both the CFFI and mock sessions. Mixing session types on one store + // is a programming error, but it is reported rather than panicked on - a panic here would + // unwind across the C ABI boundary. + let Ok(stashed) = entry.flow_file.downcast::() else { + return Err(MinifiError::CustomError( + "stashed FlowFile was stored by a different session type".into(), + )); + }; + session.unstash(*stashed) + } + + pub fn contains(&self, key: &K) -> bool { + self.stored.contains_key(key) + } + + pub fn len(&self) -> usize { + self.stored.len() + } + + pub fn is_empty(&self) -> bool { + self.stored.is_empty() + } +} + +impl Default for FlowFileStore { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::MockFlowFile; + use crate::mock::MockProcessSession; + + #[test] + fn store_then_take_round_trips_a_flow_file() { + let mut session = MockProcessSession::new(); + let mut store: FlowFileStore = FlowFileStore::new(); + + let mut ff = MockFlowFile::new(); + ff.attributes + .insert("enrichment.group.id".to_string(), "g1".to_string()); + let id = ff.id.clone(); + + store + .store(&mut session, "g1".to_string(), ff) + .expect("store should succeed"); + assert_eq!(store.len(), 1); + assert!(store.contains(&"g1".to_string())); + + let taken = store + .take(&mut session, &"g1".to_string()) + .expect("take should succeed") + .expect("a flow file should be stored under g1"); + assert_eq!(taken.id, id); + assert!(store.is_empty()); + } + + #[test] + fn take_missing_key_returns_none() { + let mut session = MockProcessSession::new(); + let mut store: FlowFileStore = FlowFileStore::new(); + + let result = store + .take(&mut session, &"missing".to_string()) + .expect("take should not error"); + assert!(result.is_none()); + } + + #[test] + fn drain_expired_only_returns_entries_older_than_max_age() { + let mut session = MockProcessSession::new(); + let mut store: FlowFileStore = FlowFileStore::new(); + + store + .store(&mut session, "g1".to_string(), MockFlowFile::new()) + .expect("store should succeed"); + + let max_age = Duration::from_secs(60); + // `now` right after storing: nothing has aged past max_age. + let just_now = Instant::now(); + let drained = store + .drain_expired(&mut session, just_now, max_age) + .expect("drain should succeed"); + assert!(drained.is_empty()); + assert_eq!(store.len(), 1); + + // `now` far in the future: the entry is older than max_age and comes back. + let future = just_now + Duration::from_secs(120); + let drained = store + .drain_expired(&mut session, future, max_age) + .expect("drain should succeed"); + assert_eq!(drained.len(), 1); + assert!(store.is_empty()); + } +} diff --git a/minifi_rust/minifi_native/src/api/process_context.rs b/minifi_rust/minifi_native/src/api/process_context.rs index 4071dc9c8c..d70093285f 100644 --- a/minifi_rust/minifi_native/src/api/process_context.rs +++ b/minifi_rust/minifi_native/src/api/process_context.rs @@ -56,6 +56,10 @@ pub trait ProcessContext { fn report_metrics(&self, metrics: Vec<(String, f64)>) -> Result<(), MinifiError>; } +pub trait ScheduleContext: GetProperty { + fn set_trigger_when_empty(&self, trigger_when_empty: bool) -> Result<(), MinifiError>; +} + impl GetProperty for S where S: ProcessContext, diff --git a/minifi_rust/minifi_native/src/api/process_session.rs b/minifi_rust/minifi_native/src/api/process_session.rs index eb121d3bfc..383eb81090 100644 --- a/minifi_rust/minifi_native/src/api/process_session.rs +++ b/minifi_rust/minifi_native/src/api/process_session.rs @@ -31,11 +31,24 @@ pub enum IoState { pub trait ProcessSession { type FlowFile: FlowFile; + /// A handle to a flow file stashed with the session, storable across triggers. + /// + /// Produced by [`stash`](Self::stash) and consumed by [`unstash`](Self::unstash). + /// Unlike [`Self::FlowFile`] (which is bound to the session), this type is + /// `'static` so it can be held between `onTrigger` invocations - see + /// [`FlowFileStore`](crate::FlowFileStore). + type StashedFlowFile: 'static + Send; + fn create(&mut self) -> Result; fn get(&mut self) -> Option; + fn clone_ff(&mut self, flow_file: &Self::FlowFile) -> Result; fn transfer(&self, flow_file: Self::FlowFile, relationship: &str) -> Result<(), MinifiError>; fn remove(&mut self, flow_file: Self::FlowFile) -> Result<(), MinifiError>; + fn stash(&mut self, flow_file: Self::FlowFile) -> Result; + + fn unstash(&mut self, stashed: Self::StashedFlowFile) -> Result; + fn set_attribute( &self, flow_file: &mut Self::FlowFile, @@ -66,4 +79,15 @@ pub trait ProcessSession { F: FnOnce(&mut dyn InputStream) -> Result; fn get_flow_file_id(&self, flow_file: &Self::FlowFile) -> Result; + + fn get_required_attribute( + &self, + flow_file: &Self::FlowFile, + attr_key: &str, + ) -> Result { + self.get_attribute(flow_file, attr_key) + .ok_or(MinifiError::MissingRequiredAttribute( + attr_key.to_owned().into(), + )) + } } diff --git a/minifi_rust/minifi_native/src/api/processor.rs b/minifi_rust/minifi_native/src/api/processor.rs index 8ac4bfb12f..67a29a1877 100644 --- a/minifi_rust/minifi_native/src/api/processor.rs +++ b/minifi_rust/minifi_native/src/api/processor.rs @@ -16,11 +16,11 @@ // under the License. use crate::api::{RawProcessor, ThreadingModel}; -use crate::{GetProperty, LogLevel, Logger, MinifiError, ProcessContext}; +use crate::{LogLevel, Logger, MinifiError, ProcessContext, ScheduleContext}; use std::marker::PhantomData; pub trait Schedule { - fn schedule( + fn schedule( context: &Ctx, logger: &L, ) -> Result @@ -64,7 +64,10 @@ where self.logger.log(log_level, args); } - fn schedule(&mut self, context: &P) -> Result<(), MinifiError> { + fn schedule( + &mut self, + context: &P, + ) -> Result<(), MinifiError> { self.scheduled_impl = Some(Impl::schedule(context, &self.logger)?); Ok(()) } diff --git a/minifi_rust/minifi_native/src/api/raw_processor.rs b/minifi_rust/minifi_native/src/api/raw_processor.rs index 51a4a05504..f91df33091 100644 --- a/minifi_rust/minifi_native/src/api/raw_processor.rs +++ b/minifi_rust/minifi_native/src/api/raw_processor.rs @@ -16,7 +16,7 @@ // under the License. use crate::api::errors::MinifiError; -use crate::{LogLevel, Logger, ProcessContext, ProcessSession}; +use crate::{LogLevel, Logger, ProcessContext, ProcessSession, ScheduleContext}; pub enum ProcessorInputRequirement { Required, @@ -37,7 +37,10 @@ pub trait RawProcessor: Sized { fn new(logger: Self::LoggerType) -> Self; fn log(&self, log_level: LogLevel, args: std::fmt::Arguments); - fn schedule(&mut self, context: &P) -> Result<(), MinifiError>; + fn schedule( + &mut self, + context: &P, + ) -> Result<(), MinifiError>; fn unschedule(&mut self); } diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs index 580c04626c..9b1a9b012e 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs +++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_flow_file.rs @@ -16,7 +16,7 @@ // under the License. use crate::api::FlowFile; -use minifi_native_sys::minifi_flow_file; +use minifi_native_sys::{minifi_flow_file, minifi_stashed_flow_file}; pub struct CffiFlowFile<'a> { ptr: *mut minifi_flow_file, @@ -37,3 +37,33 @@ impl CffiFlowFile<'_> { } impl FlowFile for CffiFlowFile<'_> {} + +/// A handle to a flow file that has been stashed with a session and can be kept +/// across `onTrigger` invocations (see [`ProcessSession::stash`]). +/// +/// Unlike [`CffiFlowFile`] it carries no session lifetime, so it is `'static`. +/// The stashed flow file itself is owned by the processor/session on the minifi +/// side; this is only a token used to [`unstash`](crate::ProcessSession::unstash) +/// it again. There is deliberately no explicit free: a token dropped without +/// being unstashed simply leaves the stashed flow file in the processor, which +/// minifi reclaims when the processor is torn down. +/// +/// [`ProcessSession::stash`]: crate::ProcessSession::stash +pub struct CffiStashedFlowFile { + ptr: *mut minifi_stashed_flow_file, +} + +// SAFETY: the handle is only ever touched from the single processor instance +// that stashed it; the stashed flow file lives on the minifi side, not behind +// this pointer. +unsafe impl Send for CffiStashedFlowFile {} + +impl CffiStashedFlowFile { + pub(crate) fn new(ptr: *mut minifi_stashed_flow_file) -> Self { + Self { ptr } + } + + pub(crate) fn get_ptr(&self) -> *mut minifi_stashed_flow_file { + self.ptr + } +} diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_context.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_context.rs index d7545a7662..d1ef824849 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_context.rs +++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_context.rs @@ -17,9 +17,9 @@ use super::c_ffi_flow_file::CffiFlowFile; use super::c_ffi_primitives::StringView; -use crate::api::ProcessContext; use crate::api::controller_service::ControllerService; use crate::api::property::PropertySchema; +use crate::api::{ProcessContext, ScheduleContext}; use crate::c_ffi::{CffiLogger, StaticStrAsMinifiCStr}; use crate::{ ComponentIdentifier, ControllerServiceApi, EnableControllerService, MinifiError, Property, @@ -216,3 +216,19 @@ impl<'a> ProcessContext for CffiProcessContext<'a> { Ok(()) } } + +impl ScheduleContext for CffiProcessContext<'_> { + fn set_trigger_when_empty(&self, trigger_when_empty: bool) -> Result<(), MinifiError> { + unsafe { + let err_code = + minifi_process_context_set_trigger_when_empty(self.ptr, trigger_when_empty); + if err_code != minifi_status_MINIFI_STATUS_SUCCESS { + return Err(MinifiError::StatusError(( + "set_trigger_when_empty".into(), + NonZeroU32::new_unchecked(err_code), + ))); + } + } + Ok(()) + } +} diff --git a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs index 9bed92bd6d..c1b4e2b697 100644 --- a/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs +++ b/minifi_rust/minifi_native/src/c_ffi/c_ffi_process_session.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use super::c_ffi_flow_file::CffiFlowFile; +use super::c_ffi_flow_file::{CffiFlowFile, CffiStashedFlowFile}; use crate::MinifiError; use crate::api::process_session::{IoState, OutputStream}; use crate::api::{InputStream, ProcessSession}; @@ -24,12 +24,14 @@ use crate::c_ffi::c_ffi_streams::{CffiInputStream, CffiOutputStream}; use minifi_native_sys::{ minifi_input_stream, minifi_input_stream_read, minifi_input_stream_size, minifi_io_status_MINIFI_IO_CANCEL, minifi_io_status_MINIFI_IO_ERROR, minifi_output_stream, - minifi_output_stream_write, minifi_process_session, minifi_process_session_create, - minifi_process_session_get, minifi_process_session_get_flow_file_attribute, + minifi_output_stream_write, minifi_process_session, minifi_process_session_clone, + minifi_process_session_create, minifi_process_session_get, + minifi_process_session_get_flow_file_attribute, minifi_process_session_get_flow_file_attributes, minifi_process_session_get_flow_file_id, minifi_process_session_read, minifi_process_session_remove, - minifi_process_session_set_flow_file_attribute, minifi_process_session_transfer, - minifi_process_session_write, minifi_status_MINIFI_STATUS_SUCCESS, minifi_string_view, + minifi_process_session_set_flow_file_attribute, minifi_process_session_stash, + minifi_process_session_transfer, minifi_process_session_unstash, minifi_process_session_write, + minifi_stashed_flow_file, minifi_status_MINIFI_STATUS_SUCCESS, minifi_string_view, }; use std::ffi::{CString, c_void}; use std::io::Read; @@ -163,6 +165,7 @@ impl<'a> CffiProcessSession<'a> { impl<'a> ProcessSession for CffiProcessSession<'a> { type FlowFile = CffiFlowFile<'a>; // FlowFile shouldn't outlive the Session + type StashedFlowFile = CffiStashedFlowFile; // detached token, storable across triggers fn create(&mut self) -> Result { let ff_ptr = unsafe { minifi_process_session_create(self.ptr, std::ptr::null_mut()) }; @@ -182,6 +185,15 @@ impl<'a> ProcessSession for CffiProcessSession<'a> { } } + fn clone_ff(&mut self, flow_file: &Self::FlowFile) -> Result { + let ff_ptr = unsafe { minifi_process_session_clone(self.ptr, flow_file.get_ptr()) }; + if ff_ptr.is_null() { + Err(MinifiError::UnknownError) + } else { + Ok(CffiFlowFile::new(ff_ptr)) + } + } + fn transfer(&self, flow_file: Self::FlowFile, relationship: &str) -> Result<(), MinifiError> { let c_relationship = CString::new(relationship)?; unsafe { @@ -216,6 +228,34 @@ impl<'a> ProcessSession for CffiProcessSession<'a> { } } + fn stash(&mut self, flow_file: Self::FlowFile) -> Result { + let mut stashed: *mut minifi_stashed_flow_file = std::ptr::null_mut(); + unsafe { + match minifi_process_session_stash(self.ptr, flow_file.get_ptr(), &mut stashed) { + #[allow(non_upper_case_globals)] + minifi_status_MINIFI_STATUS_SUCCESS => Ok(CffiStashedFlowFile::new(stashed)), + err_code => Err(MinifiError::StatusError(( + "minifi_process_session_stash".into(), + NonZeroU32::new_unchecked(err_code), // SAFETY we checked against 0 (minifi_status_MINIFI_STATUS_SUCCESS) + ))), + } + } + } + + fn unstash(&mut self, stashed: Self::StashedFlowFile) -> Result { + let mut flow_file: *mut minifi_native_sys::minifi_flow_file = std::ptr::null_mut(); + unsafe { + match minifi_process_session_unstash(self.ptr, stashed.get_ptr(), &mut flow_file) { + #[allow(non_upper_case_globals)] + minifi_status_MINIFI_STATUS_SUCCESS => Ok(CffiFlowFile::new(flow_file)), + err_code => Err(MinifiError::StatusError(( + "minifi_process_session_unstash".into(), + NonZeroU32::new_unchecked(err_code), // SAFETY we checked against 0 (minifi_status_MINIFI_STATUS_SUCCESS) + ))), + } + } + } + fn set_attribute( &self, flow_file: &mut Self::FlowFile, diff --git a/minifi_rust/minifi_native/src/lib.rs b/minifi_rust/minifi_native/src/lib.rs index bbef73995e..6dc791c3c4 100644 --- a/minifi_rust/minifi_native/src/lib.rs +++ b/minifi_rust/minifi_native/src/lib.rs @@ -58,9 +58,9 @@ pub use api::process_session::IoState; pub use api::attribute::{GetAttribute, OutputAttribute}; pub use api::{ - DataSize, FlowFile, GetId, InputStream, NonBlankPath, OnTriggerResult, OutputStream, - ProcessContext, ProcessSession, ProcessorInputRequirement, PropertyConstraints, PropertyType, - Relationship, StandardPropertyValidator, + DataSize, FlowFile, FlowFileStore, GetId, InputStream, NonBlankPath, OnTriggerResult, + OutputStream, ProcessContext, ProcessSession, ProcessorInputRequirement, PropertyConstraints, + PropertyType, Relationship, ScheduleContext, StandardPropertyValidator, }; pub use minifi_native_macros as macros; diff --git a/minifi_rust/minifi_native/src/mock/mock_flow_file.rs b/minifi_rust/minifi_native/src/mock/mock_flow_file.rs index c998335a24..8abd11de22 100644 --- a/minifi_rust/minifi_native/src/mock/mock_flow_file.rs +++ b/minifi_rust/minifi_native/src/mock/mock_flow_file.rs @@ -57,6 +57,16 @@ impl MockFlowFile { } } + /// A deep copy with the same content and attributes but a fresh id, + /// mirroring `ProcessSession::clone_ff`. + pub fn deep_clone(&self) -> MockFlowFile { + Self { + content: RefCell::new(self.content.borrow().clone()), + attributes: self.attributes.clone(), + id: next_mock_flow_file_id(), + } + } + pub fn content_len(&self) -> usize { self.content.borrow().len() } diff --git a/minifi_rust/minifi_native/src/mock/mock_process_context.rs b/minifi_rust/minifi_native/src/mock/mock_process_context.rs index 580d7ceb8f..90b08a7149 100644 --- a/minifi_rust/minifi_native/src/mock/mock_process_context.rs +++ b/minifi_rust/minifi_native/src/mock/mock_process_context.rs @@ -16,12 +16,13 @@ // under the License. use crate::api::property::PropertySchema; -use crate::api::{GetId, ProcessContext, RawControllerService}; +use crate::api::{GetId, ProcessContext, RawControllerService, ScheduleContext}; use crate::{ ComponentIdentifier, ControllerServiceApi, EnableControllerService, GetAttribute, MinifiError, MockFlowFile, Property, }; use std::any::Any; +use std::cell::Cell; use std::collections::HashMap; pub struct MockPropertyMap { @@ -74,6 +75,7 @@ pub struct MockProcessContext { pub properties: MockPropertyMap, pub controller_services: HashMap>, pub attributes: HashMap, + pub trigger_when_empty: Cell, } impl ProcessContext for MockProcessContext { @@ -143,6 +145,13 @@ impl ProcessContext for MockProcessContext { } } +impl ScheduleContext for MockProcessContext { + fn set_trigger_when_empty(&self, trigger_when_empty: bool) -> Result<(), MinifiError> { + self.trigger_when_empty.set(trigger_when_empty); + Ok(()) + } +} + impl Default for MockProcessContext { fn default() -> Self { Self::new() @@ -155,6 +164,7 @@ impl MockProcessContext { properties: MockPropertyMap::new(), controller_services: HashMap::new(), attributes: HashMap::new(), + trigger_when_empty: Cell::new(false), } } } diff --git a/minifi_rust/minifi_native/src/mock/mock_process_session.rs b/minifi_rust/minifi_native/src/mock/mock_process_session.rs index b43f582cf4..29ac4d3943 100644 --- a/minifi_rust/minifi_native/src/mock/mock_process_session.rs +++ b/minifi_rust/minifi_native/src/mock/mock_process_session.rs @@ -34,6 +34,7 @@ pub struct MockProcessSession { impl ProcessSession for MockProcessSession { type FlowFile = MockFlowFile; + type StashedFlowFile = MockFlowFile; fn create(&mut self) -> Result { Ok(Self::FlowFile::new()) @@ -41,6 +42,11 @@ impl ProcessSession for MockProcessSession { fn get(&mut self) -> Option { self.input_flow_files.pop() } + + fn clone_ff(&mut self, flow_file: &Self::FlowFile) -> Result { + Ok(flow_file.deep_clone()) + } + fn transfer(&self, flow_file: Self::FlowFile, relationship: &str) -> Result<(), MinifiError> { self.transferred_flow_files .borrow_mut() @@ -55,6 +61,14 @@ impl ProcessSession for MockProcessSession { Ok(()) } + fn stash(&mut self, flow_file: Self::FlowFile) -> Result { + Ok(flow_file) + } + + fn unstash(&mut self, stashed: Self::StashedFlowFile) -> Result { + Ok(stashed) + } + fn set_attribute( &self, flow_file: &mut Self::FlowFile,