This is an automated email from the ASF dual-hosted git repository. fgerlits pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 6e4bd96fb31a4dc042144ddb72ae97b2f7c4df98 Author: Gabor Gyimesi <[email protected]> AuthorDate: Thu May 9 17:20:50 2024 +0200 MINIFICPP-2367 Add support for non-sensitive parameters Signed-off-by: Ferenc Gerlits <[email protected]> This closes #1792 --- CONFIGURE.md | 96 ++++ .../tests/unit/FlowJsonTests.cpp | 638 ++++++++++++++++++++- .../tests/unit/YamlConfigurationTests.cpp | 598 ++++++++++++++++++- libminifi/include/Exception.h | 3 +- libminifi/include/core/ConfigurableComponent.h | 2 + libminifi/include/core/FlowConfiguration.h | 2 + libminifi/include/core/ParameterContext.h | 62 ++ libminifi/include/core/ParameterTokenParser.h | 135 +++++ libminifi/include/core/ProcessGroup.h | 6 + libminifi/include/core/ProcessorConfig.h | 1 + libminifi/include/core/flow/FlowSchema.h | 6 + .../include/core/flow/StructuredConfiguration.h | 14 +- libminifi/src/core/ConfigurableComponent.cpp | 22 + libminifi/src/core/ParameterContext.cpp | 37 ++ libminifi/src/core/ParameterTokenParser.cpp | 105 ++++ libminifi/src/core/ProcessGroup.cpp | 8 + libminifi/src/core/flow/FlowSchema.cpp | 16 +- .../src/core/flow/StructuredConfiguration.cpp | 112 +++- libminifi/test/unit/ParameterTokenParserTest.cpp | 144 +++++ 19 files changed, 1969 insertions(+), 38 deletions(-) diff --git a/CONFIGURE.md b/CONFIGURE.md index d37dd7445..c456a6690 100644 --- a/CONFIGURE.md +++ b/CONFIGURE.md @@ -19,6 +19,7 @@ - [Table of Contents](#table-of-contents) - [Configuring](#configuring) + - [Parameter Contexts](#parameter-contexts) - [Configuring flow configuration format](#configuring-flow-configuration-format) - [Scheduling strategies](#scheduling-strategies) - [Configuring encryption for flow configuration](#configuring-encryption-for-flow-configuration) @@ -41,6 +42,7 @@ - [Configuring Repository storage locations](#configuring-repository-storage-locations) - [Configuring compression for rocksdb database](#configuring-compression-for-rocksdb-database) - [Configuring compaction for rocksdb database](#configuring-compaction-for-rocksdb-database) + - [Global RocksDB options](#global-rocksdb-options) - [Shared database](#shared-database) - [Configuring Repository encryption](#configuring-repository-encryption) - [Mixing encryption with shared backend](#mixing-encryption-with-shared-backend) @@ -57,6 +59,7 @@ - [Network Prioritizer Controller Service](#network-prioritizer-controller-service) - [Disk space watchdog](#disk-space-watchdog) - [Extension configuration](#extension-configuration) + - [Python processors](#python-processors) - [Log configuration](#log-configuration) - [Log appenders](#log-appenders) - [Log levels](#log-levels) @@ -120,8 +123,101 @@ It's recommended to create your configuration in YAML format or configure the ag max concurrent tasks: 1 Properties: +Besides YAML configuration format, MiNiFi C++ also supports JSON configuration. To see different uses cases in both formats, please refer to the [examples page](examples/README.md) for flow config examples. + **NOTE:** Make sure to specify id for each component (Processor, Connection, Controller, RPG etc.) to make sure that Apache MiNiFi C++ can reload the state after a process restart. The id should be unique in the flow configuration. +### Parameter Contexts + +Processor properties in flow configurations can be parameterized using parameters defined in parameter contexts. Flow configurations can define parameter contexts that define parameter-value pairs to be reused in the flow configuration as per the following rules: + - Parameters within a process group can only be used if the parameter context is assigned to that process group. + - Each process group can be assigned only one parameter context. + - Parameter contexts can be assigned to multiple process groups. + - The assigned parameter context has the scope of the process group, but not its child process groups. + - The parameters can be used in the flow configuration by using the `#{parameterName}` syntax for processor properties. + - Only alpha-numeric characters (a-z, A-Z, 0-9), hyphens ( - ), underscores ( _ ), periods ( . ), and spaces are allowed in parameter name. + - `#` character can be used to escape the parameter syntax. E.g. if the `parameterName` parameter's value is `xxx` then `#{parameterName}` will be replaced with `xxx`, `##{parameterName}` will be replaced with `#{parameterName}`, and `#####{parameterName}` will be replaced with `##xxx`. + - Parameters can only be used in values of non-sensitive processor properties. + +An example for using parameters in a JSON configuration file: + +```json +{ + "parameterContexts": [ + { + "identifier": "804e6b47-ea22-45cd-a472-545801db98e6", + "name": "root-process-group-context", + "description": "Root process group parameter context", + "parameters": [ + { + "name": "tail_base_dir", + "description": "Base dir of tailed files", + "value": "/tmp/tail/file/path" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [ + { + "name": "Tail test_file1.log", + "identifier": "83b58f9f-e661-4634-96fb-0e82b92becdf", + "type": "org.apache.nifi.minifi.processors.TailFile", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "1000 ms", + "properties": { + "File to Tail": "#{tail_base_dir}/test_file1.log" + } + }, + { + "name": "Tail test_file2.log", + "identifier": "8a772a10-7c34-48e7-b152-b1a32c5db83e", + "type": "org.apache.nifi.minifi.processors.TailFile", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "1000 ms", + "properties": { + "File to Tail": "#{tail_base_dir}/test_file2.log" + } + } + ], + "parameterContextName": "root-process-group-context" + } +} +``` + +An example for using parameters in a YAML configuration file: + +```yaml + MiNiFi Config Version: 3 + Flow Controller: + name: MiNiFi Flow + Parameter Contexts: + - id: 804e6b47-ea22-45cd-a472-545801db98e6 + name: root-process-group-context + description: Root process group parameter context + Parameters: + - name: tail_base_dir + description: 'Base dir of tailed files' + value: /tmp/tail/file/path + Processors: + - name: Tail test_file1.log + id: 83b58f9f-e661-4634-96fb-0e82b92becdf + class: org.apache.nifi.minifi.processors.TailFile + scheduling strategy: TIMER_DRIVEN + scheduling period: 1000 ms + Properties: + File to Tail: "#{tail_base_dir}/test_file1.log" + - name: Tail test_file2.log + id: 8a772a10-7c34-48e7-b152-b1a32c5db83e + class: org.apache.nifi.minifi.processors.TailFile + scheduling strategy: TIMER_DRIVEN + scheduling period: 1000 ms + Properties: + File to Tail: "#{tail_base_dir}/test_file2.log" + Parameter Context Name: root-process-group-context +``` + ### Configuring flow configuration format MiNiFi supports YAML and JSON configuration formats. The desired configuration format can be set in the minifi.properties file, but it is automatically identified by default. The default value is `adaptiveconfiguration`, but we can force to use YAML with the `yamlconfiguration` value. diff --git a/extensions/standard-processors/tests/unit/FlowJsonTests.cpp b/extensions/standard-processors/tests/unit/FlowJsonTests.cpp index 03c8acea8..ce06af515 100644 --- a/extensions/standard-processors/tests/unit/FlowJsonTests.cpp +++ b/extensions/standard-processors/tests/unit/FlowJsonTests.cpp @@ -27,9 +27,12 @@ #include "utils/StringUtils.h" #include "unit/ConfigurationTestController.h" #include "Funnel.h" +#include "core/Resource.h" using namespace std::literals::chrono_literals; +namespace org::apache::nifi::minifi::test { + TEST_CASE("NiFi flow json format is correctly parsed") { ConfigurationTestController test_controller; @@ -38,6 +41,25 @@ TEST_CASE("NiFi flow json format is correctly parsed") { static const std::string CONFIG_JSON = R"( { + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "file_size", + "description": "", + "value": "10 B" + }, + { + "name": "batch_size", + "description": "", + "value": "12" + } + ] + } + ], "rootGroup": { "name": "MiNiFi Flow", "processors": [{ @@ -52,8 +74,10 @@ TEST_CASE("NiFi flow json format is correctly parsed") { "runDurationMillis": 12, "autoTerminatedRelationships": ["one", "two"], "properties": { - "File Size": "10 B", - "Batch Size": 12 + "File Size": "#{file_size}", + "Batch Size": "#{batch_size}", + "Data Format": "Text", + "Unique FlowFiles": false } }], "funnels": [{ @@ -99,7 +123,8 @@ TEST_CASE("NiFi flow json format is correctly parsed") { "targetId": "00000000-0000-0000-0000-000000000005", "concurrentlySchedulableTaskCount": 7 }] - }] + }], + "parameterContextName": "my-context" } })"; @@ -158,3 +183,610 @@ TEST_CASE("NiFi flow json format is correctly parsed") { REQUIRE(connection2->getDestination() == port); REQUIRE(connection2->getRelationships() == (std::set<core::Relationship>{{"success", ""}})); } + +TEST_CASE("Parameters from different parameter contexts should not be replaced") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "file_size", + "description": "", + "value": "10 B" + } + ] + }, + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376789", + "name": "other-context", + "description": "my other context", + "parameters": [ + { + "name": "batch_size", + "description": "", + "value": "12" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyGenFF", + "type": "org.apache.nifi.processors.standard.GenerateFlowFile", + "concurrentlySchedulableTaskCount": 15, + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "penaltyDuration": "12 sec", + "yieldDuration": "4 sec", + "runDurationMillis": 12, + "autoTerminatedRelationships": ["one", "two"], + "properties": { + "File Size": "#{file_size}", + "Batch Size": "#{batch_size}" + } + }], + "parameterContextName": "my-context" + } +})"; + + REQUIRE_THROWS_WITH(config.getRootFromPayload(CONFIG_JSON), "Parameter Operation: Parameter 'batch_size' not found"); +} + +TEST_CASE("Cannot use the same parameter context name twice") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "file_size", + "description": "", + "value": "10 B" + } + ] + }, + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376789", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "batch_size", + "description": "", + "value": "12" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [], + "parameterContextName": "my-context" + } +})"; + + REQUIRE_THROWS_WITH(config.getRootFromPayload(CONFIG_JSON), "Parameter context name 'my-context' already exists, parameter context names must be unique!"); +} + +TEST_CASE("Cannot use the same parameter name within a parameter context twice") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "file_size", + "description": "", + "value": "10 B" + }, + { + "name": "file_size", + "description": "", + "value": "12 B" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [], + "parameterContextName": "my-context" + } +})"; + + REQUIRE_THROWS_WITH(config.getRootFromPayload(CONFIG_JSON), "Parameter Operation: Parameter name 'file_size' already exists, parameter names must be unique within a parameter context!"); +} + +class DummyFlowJsonProcessor : public core::Processor { + public: + using core::Processor::Processor; + + static constexpr const char* Description = "A processor that does nothing."; + static constexpr auto SimpleProperty = core::PropertyDefinitionBuilder<>::createProperty("Simple Property") + .withDescription("Just a simple string property") + .build(); + static constexpr auto SensitiveProperty = core::PropertyDefinitionBuilder<>::createProperty("Sensitive Property") + .withDescription("Sensitive property") + .isSensitive(true) + .build(); + static constexpr auto Properties = std::array<core::PropertyReference, 2>{SimpleProperty, SensitiveProperty}; + static constexpr auto Relationships = std::array<core::RelationshipDefinition, 0>{}; + static constexpr bool SupportsDynamicProperties = true; + static constexpr bool SupportsDynamicRelationships = true; + static constexpr core::annotation::Input InputRequirement = core::annotation::Input::INPUT_ALLOWED; + static constexpr bool IsSingleThreaded = false; + ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS + + void initialize() override { setSupportedProperties(Properties); } +}; + +REGISTER_RESOURCE(DummyFlowJsonProcessor, Processor); + +TEST_CASE("Cannot use non-sensitive parameter in sensitive property") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "my_value", + "description": "", + "value": "value1" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyGenFF", + "type": "org.apache.nifi.processors.DummyFlowJsonProcessor", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "properties": { + "Simple Property": "simple", + "Sensitive Property": "#{my_value}" + } + }], + "parameterContextName": "my-context" + } +})"; + + REQUIRE_THROWS_WITH(config.getRootFromPayload(CONFIG_JSON), "Parameter Operation: Non-sensitive parameter 'my_value' cannot be referenced in a sensitive property"); +} + +TEST_CASE("Cannot use non-sensitive parameter in sensitive property value sequence") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "my_value", + "description": "", + "value": "value1" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyGenFF", + "type": "org.apache.nifi.processors.DummyFlowJsonProcessor", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "properties": { + "Simple Property": "simple", + "Sensitive Property": [ + {"value": "value1"}, + {"value": "#{my_value}"} + ] + } + }], + "parameterContextName": "my-context" + } +})"; + + REQUIRE_THROWS_WITH(config.getRootFromPayload(CONFIG_JSON), "Parameter Operation: Non-sensitive parameter 'my_value' cannot be referenced in a sensitive property"); +} + +TEST_CASE("Parameters can be used in nested process groups") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "batch_size", + "description": "", + "value": "12" + } + ] + }, + { + "identifier": "123e10b7-8e00-3188-9a27-476cca376456", + "name": "sub-context", + "description": "my sub context", + "parameters": [ + { + "name": "file_size", + "description": "", + "value": "10 B" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyGenFF", + "type": "org.apache.nifi.processors.standard.GenerateFlowFile", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "autoTerminatedRelationships": ["success"], + "properties": { + "File Size": "1 MB", + "Batch Size": "#{batch_size}", + "Data Format": "Text", + "Unique FlowFiles": false + } + }], + "funnels": [], + "connections": [], + "remoteProcessGroups": [], + "parameterContextName": "my-context", + "processGroups": [ + { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000002", + "name": "SubGenFF", + "type": "org.apache.nifi.processors.standard.GenerateFlowFile", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "autoTerminatedRelationships": ["success"], + "properties": { + "File Size": "#{file_size}", + "Batch Size": 1, + "Data Format": "Text", + "Unique FlowFiles": false + } + }], + "funnels": [], + "connections": [], + "remoteProcessGroups": [], + "parameterContextName": "sub-context" + } + ] + } +})"; + + std::unique_ptr<core::ProcessGroup> flow = config.getRootFromPayload(CONFIG_JSON); + REQUIRE(flow); + + auto* proc = flow->findProcessorByName("MyGenFF"); + REQUIRE(proc); + CHECK(proc->getProperty("File Size") == "1 MB"); + CHECK(proc->getProperty("Batch Size") == "12"); + auto* subproc = flow->findProcessorByName("SubGenFF"); + REQUIRE(subproc); + CHECK(subproc->getProperty("File Size") == "10 B"); + CHECK(subproc->getProperty("Batch Size") == "1"); +} + +TEST_CASE("Subprocessgroups cannot inherit parameters from parent processgroup") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "batch_size", + "description": "", + "value": "12" + } + ] + }, + { + "identifier": "123e10b7-8e00-3188-9a27-476cca376456", + "name": "sub-context", + "description": "my sub context", + "parameters": [ + { + "name": "file_size", + "description": "", + "value": "10 B" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyGenFF", + "type": "org.apache.nifi.processors.standard.GenerateFlowFile", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "autoTerminatedRelationships": ["success"], + "properties": { + "File Size": "1 MB", + "Batch Size": "#{batch_size}", + "Data Format": "Text", + "Unique FlowFiles": false + } + }], + "funnels": [], + "connections": [], + "remoteProcessGroups": [], + "parameterContextName": "my-context", + "processGroups": [ + { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000002", + "name": "SubGenFF", + "type": "org.apache.nifi.processors.standard.GenerateFlowFile", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "autoTerminatedRelationships": ["success"], + "properties": { + "File Size": "#{file_size}", + "Batch Size": "#{batch_size}", + "Data Format": "Text", + "Unique FlowFiles": false + } + }], + "funnels": [], + "connections": [], + "remoteProcessGroups": [], + "parameterContextName": "sub-context" + } + ] + } +})"; + + REQUIRE_THROWS_WITH(config.getRootFromPayload(CONFIG_JSON), "Parameter Operation: Parameter 'batch_size' not found"); +} + +TEST_CASE("Cannot use parameters if no parameter context is defined") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyGenFF", + "type": "org.apache.nifi.processors.DummyFlowJsonProcessor", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "properties": { + "Simple Property": "#{my_value}" + } + }] + } +})"; + + REQUIRE_THROWS_WITH(config.getRootFromPayload(CONFIG_JSON), "Parameter Operation: Property references a parameter in its value, but no parameter context was provided."); +} + +TEST_CASE("Cannot use parameters in property value sequences if no parameter context is defined") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyGenFF", + "type": "org.apache.nifi.processors.DummyFlowJsonProcessor", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "properties": { + "Simple Property": [ + {"value": "#{first_value}"}, + {"value": "#{second_value}"} + ] + } + }] + } +})"; + + REQUIRE_THROWS_WITH(config.getRootFromPayload(CONFIG_JSON), "Parameter Operation: Property references a parameter in its value, but no parameter context was provided."); +} + +TEST_CASE("Property value sequences can use parameters") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "first_value", + "description": "", + "value": "value1" + }, + { + "name": "second_value", + "description": "", + "value": "value2" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyProcessor", + "type": "org.apache.nifi.processors.DummyFlowJsonProcessor", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "properties": { + "Simple Property": [ + {"value": "#{first_value}"}, + {"value": "#{second_value}"} + ] + } + }], + "parameterContextName": "my-context" + } +})"; + + std::unique_ptr<core::ProcessGroup> flow = config.getRootFromPayload(CONFIG_JSON); + REQUIRE(flow); + + auto* proc = flow->findProcessorByName("MyProcessor"); + REQUIRE(proc); + core::Property property("Simple Property", ""); + proc->getProperty("Simple Property", property); + auto values = property.getValues(); + REQUIRE(values.size() == 2); + CHECK(values[0] == "value1"); + CHECK(values[1] == "value2"); +} + +TEST_CASE("Dynamic properties can use parameters") { + ConfigurationTestController test_controller; + + core::flow::AdaptiveConfiguration config(test_controller.getContext()); + + static const std::string CONFIG_JSON = + R"( +{ + "parameterContexts": [ + { + "identifier": "721e10b7-8e00-3188-9a27-476cca376978", + "name": "my-context", + "description": "my parameter context", + "parameters": [ + { + "name": "first_value", + "description": "", + "value": "value1" + }, + { + "name": "second_value", + "description": "", + "value": "value2" + } + ] + } + ], + "rootGroup": { + "name": "MiNiFi Flow", + "processors": [{ + "identifier": "00000000-0000-0000-0000-000000000001", + "name": "MyProcessor", + "type": "org.apache.nifi.processors.DummyFlowJsonProcessor", + "schedulingStrategy": "TIMER_DRIVEN", + "schedulingPeriod": "3 sec", + "properties": { + "My Dynamic Property Sequence": [ + {"value": "#{first_value}"}, + {"value": "#{second_value}"} + ], + "My Dynamic Property": "#{first_value}" + } + }], + "parameterContextName": "my-context" + } +})"; + + std::unique_ptr<core::ProcessGroup> flow = config.getRootFromPayload(CONFIG_JSON); + REQUIRE(flow); + + auto* proc = flow->findProcessorByName("MyProcessor"); + REQUIRE(proc); + core::Property property("My Dynamic Property Sequence", ""); + proc->getDynamicProperty("My Dynamic Property Sequence", property); + auto values = property.getValues(); + REQUIRE(values.size() == 2); + CHECK(values[0] == "value1"); + CHECK(values[1] == "value2"); + std::string value; + REQUIRE(proc->getDynamicProperty("My Dynamic Property", value)); + CHECK(value == "value1"); +} + +} // namespace org::apache::nifi::minifi::test diff --git a/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp b/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp index caa0002f3..f7d9b9ba6 100644 --- a/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp +++ b/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp @@ -29,9 +29,12 @@ #include "utils/StringUtils.h" #include "unit/ConfigurationTestController.h" #include "unit/TestUtils.h" +#include "core/Resource.h" using namespace std::literals::chrono_literals; +namespace org::apache::nifi::minifi::test { + TEST_CASE("Test YAML Config Processing", "[YamlConfiguration]") { ConfigurationTestController test_controller; @@ -145,7 +148,7 @@ Provenance Reporting: REQUIRE(rootFlowConfig); REQUIRE(rootFlowConfig->findProcessorByName("TailFile")); - utils::Identifier uuid = rootFlowConfig->findProcessorByName("TailFile")->getUUID(); + minifi::utils::Identifier uuid = rootFlowConfig->findProcessorByName("TailFile")->getUUID(); REQUIRE(uuid); REQUIRE(!rootFlowConfig->findProcessorByName("TailFile")->getUUIDStr().empty()); REQUIRE(1 == rootFlowConfig->findProcessorByName("TailFile")->getMaxConcurrentTasks()); @@ -210,7 +213,7 @@ Remote Processing Groups: [] Provenance Reporting: )"; - REQUIRE_THROWS_AS(yamlConfig.getRootFromPayload(CONFIG_YAML_EMPTY_RETRY_ATTRIBUTE), utils::internal::InvalidValueException); + REQUIRE_THROWS_AS(yamlConfig.getRootFromPayload(CONFIG_YAML_EMPTY_RETRY_ATTRIBUTE), minifi::utils::internal::InvalidValueException); REQUIRE(LogTestController::getInstance().contains("Invalid value was set for property 'Retry Attribute' creating component 'RetryFlowFile'")); } } @@ -455,7 +458,7 @@ NiFi Properties Overrides: {} REQUIRE(rootFlowConfig); REQUIRE(rootFlowConfig->findProcessorByName("TailFile")); - utils::Identifier uuid = rootFlowConfig->findProcessorByName("TailFile")->getUUID(); + minifi::utils::Identifier uuid = rootFlowConfig->findProcessorByName("TailFile")->getUUID(); REQUIRE(uuid); REQUIRE(!rootFlowConfig->findProcessorByName("TailFile")->getUUIDStr().empty()); REQUIRE(1 == rootFlowConfig->findProcessorByName("TailFile")->getMaxConcurrentTasks()); @@ -497,7 +500,7 @@ Processors: REQUIRE(rootFlowConfig); REQUIRE(rootFlowConfig->findProcessorByName("GenerateFlowFile")); - const utils::Identifier uuid = rootFlowConfig->findProcessorByName("GenerateFlowFile")->getUUID(); + const minifi::utils::Identifier uuid = rootFlowConfig->findProcessorByName("GenerateFlowFile")->getUUID(); REQUIRE(uuid); REQUIRE(!rootFlowConfig->findProcessorByName("GenerateFlowFile")->getUUIDStr().empty()); @@ -526,7 +529,7 @@ Processors: REQUIRE(rootFlowConfig); REQUIRE(rootFlowConfig->findProcessorByName("GetFile")); - utils::Identifier uuid = rootFlowConfig->findProcessorByName("GetFile")->getUUID(); + minifi::utils::Identifier uuid = rootFlowConfig->findProcessorByName("GetFile")->getUUID(); REQUIRE(uuid); REQUIRE(!rootFlowConfig->findProcessorByName("GetFile")->getUUIDStr().empty()); } catch (const std::exception &e) { @@ -557,7 +560,7 @@ Processors: REQUIRE(rootFlowConfig); REQUIRE(rootFlowConfig->findProcessorByName("XYZ")); - utils::Identifier uuid = rootFlowConfig->findProcessorByName("XYZ")->getUUID(); + minifi::utils::Identifier uuid = rootFlowConfig->findProcessorByName("XYZ")->getUUID(); REQUIRE(uuid); REQUIRE(!rootFlowConfig->findProcessorByName("XYZ")->getUUIDStr().empty()); } @@ -720,7 +723,7 @@ Remote Process Groups: [] REQUIRE(rootFlowConfig); REQUIRE(rootFlowConfig->findProcessorByName("GenerateFlowFile1")); REQUIRE(rootFlowConfig->findProcessorByName("GenerateFlowFile2")); - REQUIRE(rootFlowConfig->findProcessorById(utils::Identifier::parse("01a2f910-7050-41c1-8528-942764e7591d").value())); + REQUIRE(rootFlowConfig->findProcessorById(minifi::utils::Identifier::parse("01a2f910-7050-41c1-8528-942764e7591d").value())); std::map<std::string, minifi::Connection*> connectionMap; rootFlowConfig->getConnections(connectionMap); @@ -783,7 +786,7 @@ TEST_CASE("Test UUID duplication checks", "[YamlConfiguration]") { class: SSLContextService )"; - utils::string::replaceAll(config_yaml, std::string("00000000-0000-0000-0000-00000000000") + i, "99999999-9999-9999-9999-999999999999"); + minifi::utils::string::replaceAll(config_yaml, std::string("00000000-0000-0000-0000-00000000000") + i, "99999999-9999-9999-9999-999999999999"); REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(config_yaml), "General Operation: UUID 99999999-9999-9999-9999-999999999999 is duplicated in the flow configuration"); } } @@ -1072,3 +1075,582 @@ TEST_CASE("Test serialization", "[YamlConfiguration]") { const std::string serialized_flow_definition_masked = std::regex_replace(serialized_flow_definition, std::regex{"enc\\{.*\\}"}, "enc{...}"); CHECK(serialized_flow_definition_masked == TEST_FLOW_WITH_SENSITIVE_PROPERTIES_ENCRYPTED); } + +TEST_CASE("Yaml configuration can use parameter contexts", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: Simple TailFile +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: lookup.frequency + description: '' + value: 12 min + - name: batch_size + description: '' + value: 12 +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: TailFile + class: org.apache.nifi.processors.standard.TailFile + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Batch Size: "#{batch_size}" + File to Tail: ./logs/minifi-app.log + Initial Start Position: Beginning of File + tail-mode: Single file + Lookup frequency: "#{lookup.frequency}" +Controller Services: [] +Process Groups: [] +Input Ports: [] +Output Ports: [] +Funnels: [] +Connections: [] +Parameter Context Name: my-context +NiFi Properties Overrides: {} + )"; + + std::unique_ptr<core::ProcessGroup> flow = yaml_config.getRootFromPayload(TEST_CONFIG_YAML); + REQUIRE(flow); + auto* proc = flow->findProcessorByName("TailFile"); + REQUIRE(proc); + REQUIRE(proc->getProperty("Batch Size") == "12"); + REQUIRE(proc->getProperty("Lookup frequency") == "12 min"); +} + +TEST_CASE("Yaml config should not replace parameter from different parameter context", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: Simple TailFile +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: lookup.frequency + description: '' + value: 12 min + - id: 123e10b7-8e00-3188-9a27-476cca376978 + name: other-context + description: my other context + Parameters: + - name: batch_size + description: '' + value: 1 +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: TailFile + class: org.apache.nifi.processors.standard.TailFile + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Batch Size: "#{batch_size}" + File to Tail: ./logs/minifi-app.log + Initial Start Position: Beginning of File + tail-mode: Single file + Lookup frequency: "#{lookup.frequency}" +Controller Services: [] +Process Groups: [] +Input Ports: [] +Output Ports: [] +Funnels: [] +Connections: [] +Parameter Context Name: my-context +NiFi Properties Overrides: {} + )"; + + REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(TEST_CONFIG_YAML), "Parameter Operation: Parameter 'batch_size' not found"); +} + +TEST_CASE("Cannot use the same parameter context name twice", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: Simple TailFile +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: lookup.frequency + description: '' + value: 12 min + - id: 123e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: batch_size + description: '' + value: 1 +Processors: [] +Controller Services: [] +Process Groups: [] +Input Ports: [] +Output Ports: [] +Funnels: [] +Connections: [] +Parameter Context Name: my-context +NiFi Properties Overrides: {} + )"; + + REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(TEST_CONFIG_YAML), "Parameter context name 'my-context' already exists, parameter context names must be unique!"); +} + +TEST_CASE("Cannot use the same parameter name within a parameter context twice", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: Simple TailFile +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: lookup.frequency + description: '' + value: 12 min + - name: lookup.frequency + description: '' + value: 1 min +Processors: [] +Controller Services: [] +Process Groups: [] +Input Ports: [] +Output Ports: [] +Funnels: [] +Connections: [] +Parameter Context Name: my-context +NiFi Properties Overrides: {} + )"; + + REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(TEST_CONFIG_YAML), + "Parameter Operation: Parameter name 'lookup.frequency' already exists, parameter names must be unique within a parameter context!"); +} + +class DummyFlowYamlProcessor : public core::Processor { + public: + using core::Processor::Processor; + + static constexpr const char* Description = "A processor that does nothing."; + static constexpr auto SimpleProperty = core::PropertyDefinitionBuilder<>::createProperty("Simple Property") + .withDescription("Just a simple string property") + .build(); + static constexpr auto SensitiveProperty = core::PropertyDefinitionBuilder<>::createProperty("Sensitive Property") + .withDescription("Sensitive property") + .isSensitive(true) + .build(); + static constexpr auto Properties = std::array<core::PropertyReference, 2>{SimpleProperty, SensitiveProperty}; + static constexpr auto Relationships = std::array<core::RelationshipDefinition, 0>{}; + static constexpr bool SupportsDynamicProperties = true; + static constexpr bool SupportsDynamicRelationships = true; + static constexpr core::annotation::Input InputRequirement = core::annotation::Input::INPUT_ALLOWED; + static constexpr bool IsSingleThreaded = false; + ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS + + void initialize() override { setSupportedProperties(Properties); } +}; + +REGISTER_RESOURCE(DummyFlowYamlProcessor, Processor); + +TEST_CASE("Cannot use non-sensitive parameter in sensitive property", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: flowconfig +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: my_value + description: '' + value: value1 +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: TailFile + class: org.apache.nifi.processors.DummyFlowYamlProcessor + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Simple Property: simple + Sensitive Property: "#{my_value}" +Controller Services: [] +Process Groups: [] +Input Ports: [] +Output Ports: [] +Funnels: [] +Connections: [] +Parameter Context Name: my-context +NiFi Properties Overrides: {} + )"; + + REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(TEST_CONFIG_YAML), "Parameter Operation: Non-sensitive parameter 'my_value' cannot be referenced in a sensitive property"); +} + +TEST_CASE("Cannot use non-sensitive parameter in sensitive property value sequence", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: flowconfig +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: my_value + description: '' + value: value1 +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: TailFile + class: org.apache.nifi.processors.DummyFlowYamlProcessor + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Simple Property: simple + Sensitive Property: + - value: first value + - value: "#{my_value}" +Controller Services: [] +Process Groups: [] +Input Ports: [] +Output Ports: [] +Funnels: [] +Connections: [] +Parameter Context Name: my-context +NiFi Properties Overrides: {} + )"; + + REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(TEST_CONFIG_YAML), "Parameter Operation: Non-sensitive parameter 'my_value' cannot be referenced in a sensitive property"); +} + +TEST_CASE("Parameters can be used in nested process groups", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: Simple TailFile +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: lookup.frequency + description: '' + value: 12 min + - id: 123e10b7-8e00-3188-9a27-476cca376456 + name: sub-context + description: my sub context + Parameters: + - name: batch_size + description: '' + value: 12 +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: TailFile + class: org.apache.nifi.processors.standard.TailFile + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Batch Size: 1 + File to Tail: ./logs/minifi-app.log + Initial Start Position: Beginning of File + tail-mode: Single file + Lookup frequency: "#{lookup.frequency}" +Controller Services: [] +Input Ports: [] +Output Ports: [] +Funnels: [] +Connections: [] +Parameter Context Name: my-context +Process Groups: + - id: 2a3aaf32-8574-4fa7-b720-84001f8dde43 + name: Sub process group + Processors: + - id: 12304f28-0158-1000-0000-000000000000 + name: SubTailFile + class: org.apache.nifi.processors.standard.TailFile + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Batch Size: "#{batch_size}" + File to Tail: ./logs/minifi-app.log + Initial Start Position: Beginning of File + tail-mode: Single file + Lookup frequency: 1 sec + Parameter Context Name: sub-context + )"; + + std::unique_ptr<core::ProcessGroup> flow = yaml_config.getRootFromPayload(TEST_CONFIG_YAML); + REQUIRE(flow); + auto* proc = flow->findProcessorByName("TailFile"); + REQUIRE(proc); + CHECK(proc->getProperty("Batch Size") == "1"); + CHECK(proc->getProperty("Lookup frequency") == "12 min"); + auto* subproc = flow->findProcessorByName("SubTailFile"); + REQUIRE(subproc); + CHECK(subproc->getProperty("Batch Size") == "12"); + CHECK(subproc->getProperty("Lookup frequency") == "1 sec"); +} + +TEST_CASE("Subprocessgroups cannot inherit parameters from parent processgroup", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: Simple TailFile +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: lookup.frequency + description: '' + value: 12 min + - id: 123e10b7-8e00-3188-9a27-476cca376456 + name: sub-context + description: my sub context + Parameters: + - name: batch_size + description: '' + value: 12 +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: TailFile + class: org.apache.nifi.processors.standard.TailFile + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Batch Size: 1 + File to Tail: ./logs/minifi-app.log + Initial Start Position: Beginning of File + tail-mode: Single file + Lookup frequency: "#{lookup.frequency}" +Controller Services: [] +Input Ports: [] +Output Ports: [] +Funnels: [] +Connections: [] +Parameter Context Name: my-context +Process Groups: + - id: 2a3aaf32-8574-4fa7-b720-84001f8dde43 + name: Sub process group + Processors: + - id: 12304f28-0158-1000-0000-000000000000 + name: SubTailFile + class: org.apache.nifi.processors.standard.TailFile + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Batch Size: "#{batch_size}" + File to Tail: ./logs/minifi-app.log + Initial Start Position: Beginning of File + tail-mode: Single file + Lookup frequency: "#{lookup.frequency}" + Parameter Context Name: sub-context + )"; + + REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(TEST_CONFIG_YAML), "Parameter Operation: Parameter 'lookup.frequency' not found"); +} + +TEST_CASE("Cannot use parameters if no parameter context is defined", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: flowconfig +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: TailFile + class: org.apache.nifi.processors.DummyFlowYamlProcessor + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Simple Property: "#{my_value}" + )"; + + REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(TEST_CONFIG_YAML), "Parameter Operation: Property references a parameter in its value, but no parameter context was provided."); +} + +TEST_CASE("Cannot use parameters in property value sequences if no parameter context is defined", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: flowconfig +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: TailFile + class: org.apache.nifi.processors.DummyFlowYamlProcessor + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Simple Property: + - value: "#{first_value}" + - value: "#{second_value}" + )"; + + REQUIRE_THROWS_WITH(yaml_config.getRootFromPayload(TEST_CONFIG_YAML), "Parameter Operation: Property references a parameter in its value, but no parameter context was provided."); +} + +TEST_CASE("Property value sequences can use parameters", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: flow +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: first_value + description: '' + value: value1 + - name: second_value + description: '' + value: value2 +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: DummyProcessor + class: org.apache.nifi.processors.DummyFlowYamlProcessor + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + Simple Property: + - value: "#{first_value}" + - value: "#{second_value}" +Parameter Context Name: my-context + )"; + + std::unique_ptr<core::ProcessGroup> flow = yaml_config.getRootFromPayload(TEST_CONFIG_YAML); + REQUIRE(flow); + auto* proc = flow->findProcessorByName("DummyProcessor"); + REQUIRE(proc); + core::Property property("Simple Property", ""); + proc->getProperty("Simple Property", property); + auto values = property.getValues(); + REQUIRE(values.size() == 2); + CHECK(values[0] == "value1"); + CHECK(values[1] == "value2"); +} + +TEST_CASE("Dynamic properties can use parameters", "[YamlConfiguration]") { + ConfigurationTestController test_controller; + core::YamlConfiguration yaml_config(test_controller.getContext()); + + static const std::string TEST_CONFIG_YAML = + R"( +MiNiFi Config Version: 3 +Flow Controller: + name: flow +Parameter Contexts: + - id: 721e10b7-8e00-3188-9a27-476cca376978 + name: my-context + description: my parameter context + Parameters: + - name: first_value + description: '' + value: value1 + - name: second_value + description: '' + value: value2 +Processors: +- id: b0c04f28-0158-1000-0000-000000000000 + name: DummyProcessor + class: org.apache.nifi.processors.DummyFlowYamlProcessor + max concurrent tasks: 1 + scheduling strategy: TIMER_DRIVEN + scheduling period: 1 sec + auto-terminated relationships list: [success] + Properties: + My Dynamic Property Sequence: + - value: "#{first_value}" + - value: "#{second_value}" + My Dynamic Property: "#{first_value}" +Parameter Context Name: my-context + )"; + + std::unique_ptr<core::ProcessGroup> flow = yaml_config.getRootFromPayload(TEST_CONFIG_YAML); + REQUIRE(flow); + + auto* proc = flow->findProcessorByName("DummyProcessor"); + REQUIRE(proc); + core::Property property("My Dynamic Property Sequence", ""); + proc->getDynamicProperty("My Dynamic Property Sequence", property); + auto values = property.getValues(); + REQUIRE(values.size() == 2); + CHECK(values[0] == "value1"); + CHECK(values[1] == "value2"); + std::string value; + REQUIRE(proc->getDynamicProperty("My Dynamic Property", value)); + CHECK(value == "value1"); +} + +} // namespace org::apache::nifi::minifi::test diff --git a/libminifi/include/Exception.h b/libminifi/include/Exception.h index 1a0759c0f..dff108d84 100644 --- a/libminifi/include/Exception.h +++ b/libminifi/include/Exception.h @@ -42,11 +42,12 @@ enum ExceptionType { GENERAL_EXCEPTION, REGEX_EXCEPTION, REPOSITORY_EXCEPTION, + PARAMETER_EXCEPTION, MAX_EXCEPTION }; static const char *ExceptionStr[MAX_EXCEPTION] = { "File Operation", "Flow File Operation", "Processor Operation", "Process Session Operation", "Process Schedule Operation", "Site2Site Protocol", - "General Operation", "Regex Operation", "Repository Operation" }; + "General Operation", "Regex Operation", "Repository Operation", "Parameter Operation"}; inline const char *ExceptionTypeToString(ExceptionType type) { if (type < MAX_EXCEPTION) diff --git a/libminifi/include/core/ConfigurableComponent.h b/libminifi/include/core/ConfigurableComponent.h index 41fa10b04..56e518fcd 100644 --- a/libminifi/include/core/ConfigurableComponent.h +++ b/libminifi/include/core/ConfigurableComponent.h @@ -136,6 +136,8 @@ class ConfigurableComponent { */ bool getDynamicProperty(const std::string& name, std::string &value) const; + bool getDynamicProperty(const std::string& name, core::Property &item) const; + /** * Sets the value of a new dynamic property. * diff --git a/libminifi/include/core/FlowConfiguration.h b/libminifi/include/core/FlowConfiguration.h index 1841c8bf4..33f659948 100644 --- a/libminifi/include/core/FlowConfiguration.h +++ b/libminifi/include/core/FlowConfiguration.h @@ -41,6 +41,7 @@ #include "core/state/nodes/FlowInformation.h" #include "utils/file/FileSystem.h" #include "utils/ChecksumCalculator.h" +#include "ParameterContext.h" namespace org::apache::nifi::minifi::core { @@ -134,6 +135,7 @@ class FlowConfiguration : public CoreComponent { utils::ChecksumCalculator& getChecksumCalculator() { return checksum_calculator_; } protected: + std::unordered_map<std::string, gsl::not_null<std::unique_ptr<ParameterContext>>> parameter_contexts_; std::optional<std::filesystem::path> config_path_; std::shared_ptr<core::Repository> flow_file_repo_; std::shared_ptr<core::ContentRepository> content_repo_; diff --git a/libminifi/include/core/ParameterContext.h b/libminifi/include/core/ParameterContext.h new file mode 100644 index 000000000..c6f61741c --- /dev/null +++ b/libminifi/include/core/ParameterContext.h @@ -0,0 +1,62 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include <string> +#include <unordered_set> +#include <optional> + +#include "Core.h" +#include "Exception.h" + +namespace org::apache::nifi::minifi::core { + +class ParameterException : public Exception { + public: + explicit ParameterException(const std::string& message) : Exception(ExceptionType::PARAMETER_EXCEPTION, message) {} + explicit ParameterException(const char* message) : Exception(ExceptionType::PARAMETER_EXCEPTION, message) {} +}; + +struct Parameter { + std::string name; + std::string description; + std::string value; +}; + +class ParameterContext : public CoreComponent { + public: + using CoreComponent::CoreComponent; + + void setDescription(const std::string &description) { + description_ = description; + } + + std::string getDescription() const { + return description_; + } + + void addParameter(const Parameter ¶meter); + std::optional<Parameter> getParameter(const std::string &name) const; + + private: + std::string description_; + std::unordered_map<std::string, Parameter> parameters_; +}; + +} // namespace org::apache::nifi::minifi::core + diff --git a/libminifi/include/core/ParameterTokenParser.h b/libminifi/include/core/ParameterTokenParser.h new file mode 100644 index 000000000..97960485e --- /dev/null +++ b/libminifi/include/core/ParameterTokenParser.h @@ -0,0 +1,135 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include <utility> +#include <string> +#include <unordered_map> +#include <memory> +#include <optional> + +#include "ParameterContext.h" +#include "Exception.h" + +namespace org::apache::nifi::minifi::core { + +class ParameterToken { + public: + enum class ParameterTokenType { + Escaped, + Replaceable + }; + + ParameterToken(uint32_t start, uint32_t size) : start_(start), size_(size) { + } + + virtual ~ParameterToken() = default; + + uint32_t getStart() const { + return start_; + } + + uint32_t getSize() const { + return size_; + } + + virtual ParameterTokenType getType() const = 0; + virtual std::optional<std::string> getName() const { + return std::nullopt; + } + + virtual std::optional<std::string> getValue() const { + return std::nullopt; + } + + virtual uint32_t getAdditionalHashmarks() const { + return 0; + } + + private: + std::string name_; + uint32_t start_; + uint32_t size_; +}; + +class ReplaceableToken : public ParameterToken { + public: + ReplaceableToken(std::string name, uint32_t additional_hashmarks, uint32_t start, uint32_t size) : ParameterToken(start, size), name_(std::move(name)), additional_hashmarks_(additional_hashmarks) { + } + + std::optional<std::string> getName() const override { + return name_; + } + + ParameterTokenType getType() const override { + return ParameterTokenType::Replaceable; + } + + uint32_t getAdditionalHashmarks() const override { + return additional_hashmarks_; + } + + private: + std::string name_; + uint32_t additional_hashmarks_; +}; + +class EscapedToken : public ParameterToken { + public: + EscapedToken(uint32_t start, uint32_t size, std::string replaced_value) : ParameterToken(start, size), replaced_value_(std::move(replaced_value)) { + }; + + ParameterTokenType getType() const override { + return ParameterTokenType::Escaped; + } + + std::optional<std::string> getValue() const override { + return replaced_value_; + } + + private: + std::string replaced_value_; +}; + +class ParameterTokenParser { + public: + enum class ParseState { + OutsideToken, + InHashMark, + InToken + }; + + explicit ParameterTokenParser(std::string input) : input_(std::move(input)) { + parse(); + } + + const std::vector<std::unique_ptr<ParameterToken>>& getTokens() const { + return tokens_; + } + + std::string replaceParameters(ParameterContext* parameter_context, bool is_sensitive) const; + + private: + void parse(); + + std::string input_; + std::vector<std::unique_ptr<ParameterToken>> tokens_; +}; + +} // namespace org::apache::nifi::minifi::core + diff --git a/libminifi/include/core/ProcessGroup.h b/libminifi/include/core/ProcessGroup.h index 85a1eafaa..3acca6741 100644 --- a/libminifi/include/core/ProcessGroup.h +++ b/libminifi/include/core/ProcessGroup.h @@ -42,6 +42,7 @@ #include "http/BaseHTTPClient.h" #include "utils/CallBackTimer.h" #include "range/v3/algorithm/find_if.hpp" +#include "core/ParameterContext.h" struct ProcessGroupTestAccessor; @@ -217,6 +218,9 @@ class ProcessGroup : public CoreComponent { void verify() const; + void setParameterContext(ParameterContext* parameter_context); + ParameterContext* getParameterContext() const; + protected: void startProcessingProcessors(TimerDrivenSchedulingAgent& timeScheduler, EventDrivenSchedulingAgent& eventScheduler, CronDrivenSchedulingAgent& cronScheduler); @@ -251,6 +255,8 @@ class ProcessGroup : public CoreComponent { core::controller::ControllerServiceNodeMap controller_service_map_; + ParameterContext* parameter_context_ = nullptr; + private: static Port* findPortById(const std::set<Port*>& ports, const utils::Identifier& uuid); diff --git a/libminifi/include/core/ProcessorConfig.h b/libminifi/include/core/ProcessorConfig.h index d454a3f69..e4706204c 100644 --- a/libminifi/include/core/ProcessorConfig.h +++ b/libminifi/include/core/ProcessorConfig.h @@ -50,6 +50,7 @@ struct ProcessorConfig { std::string runDurationNanos; std::vector<std::string> autoTerminatedRelationships; std::vector<core::Property> properties; + std::string parameterContextName; }; } // namespace core diff --git a/libminifi/include/core/flow/FlowSchema.h b/libminifi/include/core/flow/FlowSchema.h index 773550ddc..74b6993e0 100644 --- a/libminifi/include/core/flow/FlowSchema.h +++ b/libminifi/include/core/flow/FlowSchema.h @@ -80,6 +80,12 @@ struct FlowSchema { Keys rpg_port_properties; Keys rpg_port_target_id; + Keys parameter_contexts; + Keys parameters; + Keys description; + Keys value; + Keys parameter_context_name; + static FlowSchema getDefault(); static FlowSchema getNiFiFlowJson(); }; diff --git a/libminifi/include/core/flow/StructuredConfiguration.h b/libminifi/include/core/flow/StructuredConfiguration.h index bfce44f77..c87d3af68 100644 --- a/libminifi/include/core/flow/StructuredConfiguration.h +++ b/libminifi/include/core/flow/StructuredConfiguration.h @@ -71,6 +71,7 @@ class StructuredConfiguration : public FlowConfiguration { std::unique_ptr<core::ProcessGroup> createProcessGroup(const Node& node, bool is_root = false); std::unique_ptr<core::ProcessGroup> parseProcessGroup(const Node& header_node, const Node& node, bool is_root = false); + /** * Parses processors from its corresponding config node and adds * them to a parent ProcessGroup. The processors_node argument must point @@ -107,6 +108,7 @@ class StructuredConfiguration : public FlowConfiguration { */ std::unique_ptr<core::ProcessGroup> parseRootProcessGroup(const Node& root_flow_node); + void parseParameterContexts(const Node& parameter_contexts_node); void parseControllerServices(const Node& controller_services_node); /** @@ -149,7 +151,7 @@ class StructuredConfiguration : public FlowConfiguration { * @param properties_node the Node containing the properties * @param processor the Processor to which to add the resulting properties */ - void parsePropertiesNode(const Node& properties_node, core::ConfigurableComponent& component, const std::string& component_name); + void parsePropertiesNode(const Node& properties_node, core::ConfigurableComponent& component, const std::string& component_name, ParameterContext* parameter_context); /** * Parses the Funnels section of a configuration. @@ -173,6 +175,8 @@ class StructuredConfiguration : public FlowConfiguration { */ void parsePorts(const flow::Node& node, core::ProcessGroup* parent, PortType port_type); + void parseParameterContext(const flow::Node& node, core::ProcessGroup& parent); + /** * A helper function for parsing or generating optional id fields. * @@ -216,10 +220,10 @@ class StructuredConfiguration : public FlowConfiguration { std::shared_ptr<logging::Logger> logger_; private: - PropertyValue getValidatedProcessorPropertyForDefaultTypeInfo(const core::Property& property_from_processor, const Node& property_value_node); - void parsePropertyValueSequence(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& component); - void parseSingleProperty(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& processor); - void parsePropertyNodeElement(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& processor); + PropertyValue getValidatedProcessorPropertyForDefaultTypeInfo(const core::Property& property_from_processor, const Node& property_value_node, ParameterContext* parameter_context); + void parsePropertyValueSequence(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& component, ParameterContext* parameter_context); + void parseSingleProperty(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& processor, ParameterContext* parameter_context); + void parsePropertyNodeElement(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& processor, ParameterContext* parameter_context); void addNewId(const std::string& uuid); /** diff --git a/libminifi/src/core/ConfigurableComponent.cpp b/libminifi/src/core/ConfigurableComponent.cpp index 00fdbb241..2769472bc 100644 --- a/libminifi/src/core/ConfigurableComponent.cpp +++ b/libminifi/src/core/ConfigurableComponent.cpp @@ -245,6 +245,27 @@ bool ConfigurableComponent::getDynamicProperty(const std::string& name, std::str } } +bool ConfigurableComponent::getDynamicProperty(const std::string& name, core::Property &item) const { + std::lock_guard<std::mutex> lock(configuration_mutex_); + + auto &&it = dynamic_properties_.find(name); + if (it != dynamic_properties_.end()) { + item = it->second; + if (item.getValue().getValue() == nullptr) { + // empty property value + if (item.getRequired()) { + logger_->log_error("Component {} required dynamic property {} is empty", name, item.getName()); + throw std::runtime_error("Required dynamic property is empty: " + item.getName()); + } + logger_->log_debug("Component {} dynamic property name {}, empty value", name, item.getName()); + return false; + } + return true; + } else { + return false; + } +} + bool ConfigurableComponent::createDynamicProperty(const std::string &name, const std::string &value) { if (!supportsDynamicProperties()) { logger_->log_debug("Attempted to create dynamic property {}, but this component does not support creation." @@ -255,6 +276,7 @@ bool ConfigurableComponent::createDynamicProperty(const std::string &name, const static const std::string DEFAULT_DYNAMIC_PROPERTY_DESC = "Dynamic Property"; Property new_property(name, DEFAULT_DYNAMIC_PROPERTY_DESC, value, false, { }, { }); + new_property.setValue(value); new_property.setSupportsExpressionLanguage(true); logger_->log_info("Processor {} dynamic property '{}' value '{}'", name.c_str(), new_property.getName().c_str(), value.c_str()); dynamic_properties_[new_property.getName()] = new_property; diff --git a/libminifi/src/core/ParameterContext.cpp b/libminifi/src/core/ParameterContext.cpp new file mode 100644 index 000000000..b4501930f --- /dev/null +++ b/libminifi/src/core/ParameterContext.cpp @@ -0,0 +1,37 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "core/ParameterContext.h" + +namespace org::apache::nifi::minifi::core { + +void ParameterContext::addParameter(const Parameter ¶meter) { + if (parameters_.find(parameter.name) != parameters_.end()) { + throw ParameterException("Parameter name '" + parameter.name + "' already exists, parameter names must be unique within a parameter context!"); + } + parameters_.emplace(parameter.name, parameter); +} + +std::optional<Parameter> ParameterContext::getParameter(const std::string &name) const { + auto it = parameters_.find(name); + if (it != parameters_.end()) { + return it->second; + } + return std::nullopt; +} + +} // namespace org::apache::nifi::minifi::core diff --git a/libminifi/src/core/ParameterTokenParser.cpp b/libminifi/src/core/ParameterTokenParser.cpp new file mode 100644 index 000000000..c9d330a5e --- /dev/null +++ b/libminifi/src/core/ParameterTokenParser.cpp @@ -0,0 +1,105 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "core/ParameterTokenParser.h" + +#include <stdexcept> +#include "utils/RegexUtils.h" + +namespace org::apache::nifi::minifi::core { + +void ParameterTokenParser::parse() { + minifi::utils::Regex expr("[-a-zA-Z0-9_\\. ]+"); + ParseState state = ParseState::OutsideToken; + uint32_t token_start = 0; + uint32_t hashmark_length = 0; + for (uint32_t i = 0; i < input_.size(); ++i) { + if (input_[i] == '#') { + if (state == ParseState::OutsideToken) { + state = ParseState::InHashMark; + } + if (state != ParseState::InToken) { + ++hashmark_length; + } + } else if (input_[i] == '{') { + if (state == ParseState::InHashMark) { + token_start = i - hashmark_length; + state = ParseState::InToken; + } + } else if (input_[i] == '}') { + if (state == ParseState::InToken) { + state = ParseState::OutsideToken; + gsl_Assert(hashmark_length > 0); + if (hashmark_length % 2 == 0) { + tokens_.push_back(std::make_unique<EscapedToken>(token_start, i - token_start + 1, input_.substr(token_start + (hashmark_length / 2), i - token_start + 1 - (hashmark_length / 2)))); + } else { + auto token_name = input_.substr(token_start + hashmark_length + 1, i - token_start - hashmark_length - 1); + if (token_name.empty() || !minifi::utils::regexMatch(token_name, expr)) { + throw ParameterException("Invalid token name: '" + token_name + "'. " + "Only alpha-numeric characters (a-z, A-Z, 0-9), hyphens ( - ), underscores ( _ ), periods ( . ), and spaces are allowed in token name."); + } + tokens_.push_back(std::make_unique<ReplaceableToken>(token_name, (hashmark_length - 1) / 2, token_start, i - token_start + 1)); + token_start = 0; + } + } else { + state = ParseState::OutsideToken; + } + hashmark_length = 0; + } else { + if (state != ParseState::InToken) { + state = ParseState::OutsideToken; + hashmark_length = 0; + } + } + } +} + +std::string ParameterTokenParser::replaceParameters(ParameterContext* parameter_context, bool is_sensitive) const { + if (tokens_.empty()) { + return input_; + } + std::string result; + uint32_t last_end = 0; + for (const auto& token : tokens_) { + result.append(input_.substr(last_end, token->getStart() - last_end)); + if (token->getType() == ParameterToken::ParameterTokenType::Escaped) { + result.append(token->getValue().value()); + last_end = token->getStart() + token->getSize(); + continue; + } + + if (!parameter_context) { + throw ParameterException("Property references a parameter in its value, but no parameter context was provided."); + } + + gsl_Assert(token->getName().has_value()); + auto parameter = parameter_context->getParameter(token->getName().value()); + if (!parameter.has_value()) { + throw ParameterException("Parameter '" + token->getName().value() + "' not found"); + } + if (is_sensitive) { + throw ParameterException("Non-sensitive parameter '" + parameter->name + "' cannot be referenced in a sensitive property"); + } + result.append(std::string(token->getAdditionalHashmarks(), '#') + parameter->value); + last_end = token->getStart() + token->getSize(); + } + result.append(input_.substr(last_end)); + return result; +} + +} // namespace org::apache::nifi::minifi::core + diff --git a/libminifi/src/core/ProcessGroup.cpp b/libminifi/src/core/ProcessGroup.cpp index bca53be6b..7f8fe1d27 100644 --- a/libminifi/src/core/ProcessGroup.cpp +++ b/libminifi/src/core/ProcessGroup.cpp @@ -448,4 +448,12 @@ void ProcessGroup::verify() const { } } +void ProcessGroup::setParameterContext(ParameterContext* parameter_context) { + parameter_context_ = parameter_context; +} + +ParameterContext* ProcessGroup::getParameterContext() const { + return parameter_context_; +} + } // namespace org::apache::nifi::minifi::core diff --git a/libminifi/src/core/flow/FlowSchema.cpp b/libminifi/src/core/flow/FlowSchema.cpp index c07e548e7..d2c3f4523 100644 --- a/libminifi/src/core/flow/FlowSchema.cpp +++ b/libminifi/src/core/flow/FlowSchema.cpp @@ -75,7 +75,13 @@ FlowSchema FlowSchema::getDefault() { .rpg_input_ports = {"Input Ports"}, .rpg_output_ports = {"Output Ports"}, .rpg_port_properties = {"Properties"}, - .rpg_port_target_id = {} + .rpg_port_target_id = {}, + + .parameter_contexts = {"Parameter Contexts"}, + .parameters = {"Parameters"}, + .description = {"description"}, + .value = {"value"}, + .parameter_context_name = {"Parameter Context Name"} }; } @@ -135,7 +141,13 @@ FlowSchema FlowSchema::getNiFiFlowJson() { .rpg_input_ports = {"inputPorts"}, .rpg_output_ports = {"outputPorts"}, .rpg_port_properties = {}, - .rpg_port_target_id = {"targetId"} + .rpg_port_target_id = {"targetId"}, + + .parameter_contexts = {"parameterContexts"}, + .parameters = {"parameters"}, + .description = {"description"}, + .value = {"value"}, + .parameter_context_name = {"parameterContextName"} }; } diff --git a/libminifi/src/core/flow/StructuredConfiguration.cpp b/libminifi/src/core/flow/StructuredConfiguration.cpp index 4443ed34a..1a6f36532 100644 --- a/libminifi/src/core/flow/StructuredConfiguration.cpp +++ b/libminifi/src/core/flow/StructuredConfiguration.cpp @@ -28,6 +28,8 @@ #include "utils/TimeUtil.h" #include "utils/RegexUtils.h" #include "Funnel.h" +#include "core/ParameterContext.h" +#include "core/ParameterTokenParser.h" namespace org::apache::nifi::minifi::core::flow { @@ -94,7 +96,9 @@ std::unique_ptr<core::ProcessGroup> StructuredConfiguration::parseProcessGroup(c Node outputPortsNode = node[schema_.output_ports]; Node remoteProcessingGroupsNode = node[schema_.remote_process_group]; Node childProcessGroupNodeSeq = node[schema_.process_groups]; + Node parameterContextNameNode = node[schema_.parameter_context_name]; + parseParameterContext(parameterContextNameNode, *group); parseProcessorNode(processorsNode, group.get()); parseRemoteProcessGroup(remoteProcessingGroupsNode, group.get()); parseFunnels(funnelsNode, group.get()); @@ -116,9 +120,11 @@ std::unique_ptr<core::ProcessGroup> StructuredConfiguration::getRootFrom(const N try { schema_ = std::move(schema); uuids_.clear(); + Node parameterContextsNode = root_node[schema_.parameter_contexts]; Node controllerServiceNode = root_node[schema_.root_group][schema_.controller_services]; Node provenanceReportNode = root_node[schema_.provenance_reporting]; + parseParameterContexts(parameterContextsNode); parseControllerServices(controllerServiceNode); // Create the root process group std::unique_ptr<core::ProcessGroup> root = parseRootProcessGroup(root_node); @@ -139,6 +145,36 @@ std::unique_ptr<core::ProcessGroup> StructuredConfiguration::getRootFrom(const N } } +void StructuredConfiguration::parseParameterContexts(const Node& parameter_contexts_node) { + if (!parameter_contexts_node || !parameter_contexts_node.isSequence()) { + return; + } + for (const auto& parameter_context_node : parameter_contexts_node) { + checkRequiredField(parameter_context_node, schema_.name); + + auto name = parameter_context_node[schema_.name].getString().value(); + if (parameter_contexts_.find(name) != parameter_contexts_.end()) { + throw std::invalid_argument("Parameter context name '" + name + "' already exists, parameter context names must be unique!"); + } + auto id = getRequiredIdField(parameter_context_node); + + utils::Identifier uuid; + uuid = id; + auto parameter_context = std::make_unique<ParameterContext>(name, uuid); + parameter_context->setDescription(getOptionalField(parameter_context_node, schema_.description, "")); + for (const auto& parameter_node : parameter_context_node[schema_.parameters]) { + checkRequiredField(parameter_node, schema_.name); + checkRequiredField(parameter_node, schema_.value); + auto parameter_name = parameter_node[schema_.name].getString().value(); + auto parameter_value = parameter_node[schema_.value].getString().value(); + auto parameter_description = getOptionalField(parameter_node, schema_.description, ""); + parameter_context->addParameter(Parameter{parameter_name, parameter_description, parameter_value}); + } + + parameter_contexts_.emplace(name, gsl::make_not_null(std::move(parameter_context))); + } +} + void StructuredConfiguration::parseProcessorNode(const Node& processors_node, core::ProcessGroup* parentGroup) { int64_t runDurationNanos = -1; utils::Identifier uuid; @@ -230,7 +266,7 @@ void StructuredConfiguration::parseProcessorNode(const Node& processors_node, co // handle processor properties if (Node propertiesNode = procNode[schema_.processor_properties]) { - parsePropertiesNode(propertiesNode, *processor, procCfg.name); + parsePropertiesNode(propertiesNode, *processor, procCfg.name, parentGroup->getParameterContext()); } // Take care of scheduling @@ -505,9 +541,9 @@ void StructuredConfiguration::parseControllerServices(const Node& controller_ser controller_service_node->initialize(); if (Node propertiesNode = service_node[schema_.controller_service_properties]) { // we should propagate properties to the node and to the implementation - parsePropertiesNode(propertiesNode, *controller_service_node, name); + parsePropertiesNode(propertiesNode, *controller_service_node, name, nullptr); if (auto controllerServiceImpl = controller_service_node->getControllerServiceImplementation(); controllerServiceImpl) { - parsePropertiesNode(propertiesNode, *controllerServiceImpl, name); + parsePropertiesNode(propertiesNode, *controllerServiceImpl, name, nullptr); } } } else { @@ -599,9 +635,9 @@ void StructuredConfiguration::parseRPGPort(const Node& port_node, core::ProcessG // handle port properties if (Node propertiesNode = port_node[schema_.rpg_port_properties]) { - parsePropertiesNode(propertiesNode, *port, nameStr); + parsePropertiesNode(propertiesNode, *port, nameStr, nullptr); } else { - parsePropertyNodeElement(std::string(minifi::RemoteProcessorGroupPort::portUUID.name), port_node[schema_.rpg_port_target_id], *port); + parsePropertyNodeElement(std::string(minifi::RemoteProcessorGroupPort::portUUID.name), port_node[schema_.rpg_port_target_id], *port, nullptr); validateComponentProperties(*port, nameStr, port_node.getPath()); } @@ -621,7 +657,8 @@ void StructuredConfiguration::parseRPGPort(const Node& port_node, core::ProcessG } } -void StructuredConfiguration::parsePropertyValueSequence(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& component) { +void StructuredConfiguration::parsePropertyValueSequence(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& component, + ParameterContext* parameter_context) { core::Property myProp(property_name, "", ""); component.getProperty(property_name, myProp); @@ -632,13 +669,22 @@ void StructuredConfiguration::parsePropertyValueSequence(const std::string& prop if (myProp.isSensitive()) { rawValueString = utils::crypto::property_encryption::decrypt(rawValueString, sensitive_properties_encryptor_); } + + try { + core::ParameterTokenParser token_parser(rawValueString); + rawValueString = token_parser.replaceParameters(parameter_context, myProp.isSensitive()); + } catch (const ParameterException& e) { + logger_->log_error("Error while substituting parameters in property '{}': {}", property_name, e.what()); + throw; + } + logger_->log_debug("Found property {}", property_name); if (!component.updateProperty(property_name, rawValueString)) { auto proc = dynamic_cast<core::Connectable*>(&component); if (proc) { logger_->log_warn("Received property {} with value {} but is not one of the properties for {}. Attempting to add as dynamic property.", property_name, rawValueString, proc->getName()); - if (!component.setDynamicProperty(property_name, rawValueString)) { + if (!component.updateDynamicProperty(property_name, rawValueString)) { logger_->log_warn("Unable to set the dynamic property {}", property_name); } else { logger_->log_warn("Dynamic property {} has been set", property_name); @@ -649,7 +695,8 @@ void StructuredConfiguration::parsePropertyValueSequence(const std::string& prop } } -PropertyValue StructuredConfiguration::getValidatedProcessorPropertyForDefaultTypeInfo(const core::Property& property_from_processor, const Node& property_value_node) { +PropertyValue StructuredConfiguration::getValidatedProcessorPropertyForDefaultTypeInfo(const core::Property& property_from_processor, const Node& property_value_node, + ParameterContext* parameter_context) { using state::response::Value; PropertyValue defaultValue; defaultValue = property_from_processor.getDefaultValue(); @@ -667,15 +714,24 @@ PropertyValue StructuredConfiguration::getValidatedProcessorPropertyForDefaultTy coercedValue = gsl::narrow<int>(int64_val.value()); } else if (defaultType == Value::BOOL_TYPE && property_value_node.getBool()) { coercedValue = property_value_node.getBool().value(); - } else if (property_from_processor.isSensitive()) { - coercedValue = utils::crypto::property_encryption::decrypt(property_value_node.getScalarAsString().value(), sensitive_properties_encryptor_); } else { - coercedValue = property_value_node.getScalarAsString().value(); + std::string property_value_string; + if (property_from_processor.isSensitive()) { + property_value_string = utils::crypto::property_encryption::decrypt(property_value_node.getScalarAsString().value(), sensitive_properties_encryptor_); + } else { + property_value_string = property_value_node.getScalarAsString().value(); + } + core::ParameterTokenParser token_parser(property_value_string); + property_value_string = token_parser.replaceParameters(parameter_context, property_from_processor.isSensitive()); + coercedValue = property_value_string; } return coercedValue; } catch (const utils::crypto::EncryptionError& e) { logger_->log_error("Fetching property failed with a decryption error: {}", e.what()); throw; + } catch (const ParameterException& e) { + logger_->log_error("Error while substituting parameters in property '{}': {}", property_from_processor.getName(), e.what()); + throw; } catch (const std::exception& e) { logger_->log_error("Fetching property failed with an exception of {}", e.what()); logger_->log_error("Invalid conversion for field {}. Value {}", property_from_processor.getName(), property_value_node.getDebugString()); @@ -685,11 +741,12 @@ PropertyValue StructuredConfiguration::getValidatedProcessorPropertyForDefaultTy return defaultValue; } -void StructuredConfiguration::parseSingleProperty(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& processor) { +void StructuredConfiguration::parseSingleProperty(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& processor, + ParameterContext* parameter_context) { core::Property myProp(property_name, "", ""); processor.getProperty(property_name, myProp); - const PropertyValue coercedValue = getValidatedProcessorPropertyForDefaultTypeInfo(myProp, property_value_node); + PropertyValue coercedValue = getValidatedProcessorPropertyForDefaultTypeInfo(myProp, property_value_node, parameter_context); bool property_set = false; try { @@ -704,7 +761,7 @@ void StructuredConfiguration::parseSingleProperty(const std::string& property_na throw; } if (!property_set) { - const auto rawValueString = property_value_node.getScalarAsString().value(); + const auto rawValueString = coercedValue.getValue()->getStringValue(); auto proc = dynamic_cast<core::Connectable*>(&processor); if (proc) { logger_->log_warn("Received property {} but is not one of the properties for {}. Attempting to add as dynamic property.", property_name, proc->getName()); @@ -719,25 +776,27 @@ void StructuredConfiguration::parseSingleProperty(const std::string& property_na } } -void StructuredConfiguration::parsePropertyNodeElement(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& processor) { +void StructuredConfiguration::parsePropertyNodeElement(const std::string& property_name, const Node& property_value_node, core::ConfigurableComponent& processor, + ParameterContext* parameter_context) { logger_->log_trace("Encountered {}", property_name); if (!property_value_node || property_value_node.isNull()) { return; } if (property_value_node.isSequence()) { - parsePropertyValueSequence(property_name, property_value_node, processor); + parsePropertyValueSequence(property_name, property_value_node, processor, parameter_context); } else { - parseSingleProperty(property_name, property_value_node, processor); + parseSingleProperty(property_name, property_value_node, processor, parameter_context); } } -void StructuredConfiguration::parsePropertiesNode(const Node& properties_node, core::ConfigurableComponent& component, const std::string& component_name) { +void StructuredConfiguration::parsePropertiesNode(const Node& properties_node, core::ConfigurableComponent& component, const std::string& component_name, + ParameterContext* parameter_context) { // Treat generically as a node so we can perform inspection on entries to ensure they are populated logger_->log_trace("Entered {}", component_name); for (const auto& property_node : properties_node) { const auto propertyName = property_node.first.getString().value(); const Node propertyValueNode = property_node.second; - parsePropertyNodeElement(propertyName, propertyValueNode, component); + parsePropertyNodeElement(propertyName, propertyValueNode, component, parameter_context); } validateComponentProperties(component, component_name, properties_node.getPath()); @@ -799,6 +858,21 @@ void StructuredConfiguration::parsePorts(const flow::Node& node, core::ProcessGr } } +void StructuredConfiguration::parseParameterContext(const flow::Node& node, core::ProcessGroup& parent) { + if (!node) { + return; + } + + auto parameter_context_name = node.getString().value(); + if (parameter_context_name.empty()) { + return; + } + + if (parameter_contexts_.find(parameter_context_name) != parameter_contexts_.end()) { + parent.setParameterContext(parameter_contexts_.at(parameter_context_name).get()); + } +} + void StructuredConfiguration::validateComponentProperties(ConfigurableComponent& component, const std::string &component_name, const std::string §ion) const { const auto &component_properties = component.getProperties(); diff --git a/libminifi/test/unit/ParameterTokenParserTest.cpp b/libminifi/test/unit/ParameterTokenParserTest.cpp new file mode 100644 index 000000000..917838d42 --- /dev/null +++ b/libminifi/test/unit/ParameterTokenParserTest.cpp @@ -0,0 +1,144 @@ +/** + * + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "unit/TestBase.h" +#include "unit/Catch.h" +#include "core/ParameterTokenParser.h" + +namespace org::apache::nifi::minifi::test { + +TEST_CASE("Empty string has zero parameters") { + core::ParameterTokenParser parser(""); + REQUIRE(parser.getTokens().empty()); +} + +TEST_CASE("Parse a single token") { + core::ParameterTokenParser parser("#{token.1}"); + auto& tokens = parser.getTokens(); + REQUIRE(tokens.size() == 1); + CHECK(tokens.at(0)->getName().value() == "token.1"); + CHECK(tokens.at(0)->getStart() == 0); + CHECK(tokens.at(0)->getSize() == 10); +} + +TEST_CASE("Parse multiple tokens") { + core::ParameterTokenParser parser("#{token1} #{token-2}"); + auto& tokens = parser.getTokens(); + REQUIRE(tokens.size() == 2); + CHECK(tokens.at(0)->getName().value() == "token1"); + CHECK(tokens.at(0)->getStart() == 0); + CHECK(tokens.at(0)->getSize() == 9); + CHECK(tokens.at(1)->getName().value() == "token-2"); + CHECK(tokens.at(1)->getStart() == 10); + CHECK(tokens.at(1)->getSize() == 10); +} + +TEST_CASE("Parse the same token multiple times") { + core::ParameterTokenParser parser("#{token1} #{token-2} #{token1}"); + auto& tokens = parser.getTokens(); + REQUIRE(tokens.size() == 3); + CHECK(tokens.at(0)->getName().value() == "token1"); + CHECK(tokens.at(0)->getStart() == 0); + CHECK(tokens.at(0)->getSize() == 9); + CHECK(tokens.at(1)->getName().value() == "token-2"); + CHECK(tokens.at(1)->getStart() == 10); + CHECK(tokens.at(1)->getSize() == 10); + CHECK(tokens.at(2)->getName().value() == "token1"); + CHECK(tokens.at(2)->getStart() == 21); + CHECK(tokens.at(2)->getSize() == 9); +} + +TEST_CASE("Tokens can be escaped") { + core::ParameterTokenParser parser("## ##{token1} #{token-2} ###{token_3}# ## ##not_a_token"); + auto& tokens = parser.getTokens(); + REQUIRE(tokens.size() == 3); + CHECK(tokens.at(0)->getValue().value() == "#{token1}"); + CHECK(tokens.at(0)->getStart() == 3); + CHECK(tokens.at(0)->getSize() == 10); + CHECK(tokens.at(1)->getName().value() == "token-2"); + CHECK(tokens.at(1)->getStart() == 14); + CHECK(tokens.at(1)->getSize() == 10); + CHECK(tokens.at(2)->getName().value() == "token_3"); + CHECK(tokens.at(2)->getStart() == 25); + CHECK(tokens.at(2)->getSize() == 12); +} + +TEST_CASE("Unfinished token is not a token") { + core::ParameterTokenParser parser("this is #{_token_ 1} and #{token-2 not finished"); + auto& tokens = parser.getTokens(); + REQUIRE(tokens.size() == 1); + CHECK(tokens.at(0)->getName().value() == "_token_ 1"); + CHECK(tokens.at(0)->getStart() == 8); + CHECK(tokens.at(0)->getSize() == 12); +} + +TEST_CASE("Test invalid token names") { + auto create_error_message = [](const std::string& invalid_name){ + return "Parameter Operation: Invalid token name: '" + invalid_name + + "'. Only alpha-numeric characters (a-z, A-Z, 0-9), hyphens ( - ), underscores ( _ ), periods ( . ), and spaces are allowed in token name."; + }; + CHECK_THROWS_WITH(core::ParameterTokenParser("#{}"), create_error_message("")); + CHECK_THROWS_WITH(core::ParameterTokenParser("#{#}"), create_error_message("#")); + CHECK_THROWS_WITH(core::ParameterTokenParser("#{[]}"), create_error_message("[]")); + CHECK_THROWS_WITH(core::ParameterTokenParser("#{a{}"), create_error_message("a{")); + CHECK_THROWS_WITH(core::ParameterTokenParser("#{$$}"), create_error_message("$$")); +} + +TEST_CASE("Test token replacement") { + core::ParameterTokenParser parser("## What is #{what}, baby don't hurt #{who}, don't hurt #{who}, no more ##"); + core::ParameterContext context("test_context"); + context.addParameter(core::Parameter{"what", "", "love"}); + context.addParameter(core::Parameter{"who", "", "me"}); + REQUIRE(parser.replaceParameters(&context, false) == "## What is love, baby don't hurt me, don't hurt me, no more ##"); +} + +TEST_CASE("Test replacement with escaped tokens") { + core::ParameterTokenParser parser("### What is #####{what}, baby don't hurt ###{who}, don't hurt ###{who}, no ####{more} ##{"); + REQUIRE(parser.getTokens().size() == 4); + core::ParameterContext context("test_context"); + context.addParameter(core::Parameter{"what", "", "love"}); + context.addParameter(core::Parameter{"who", "", "me"}); + REQUIRE(parser.replaceParameters(&context, false) == "### What is ##love, baby don't hurt #me, don't hurt #me, no ##{more} ##{"); +} + +TEST_CASE("Test replacement with missing token in context") { + core::ParameterTokenParser parser("What is #{what}, baby don't hurt #{who}, don't hurt #{who}, no more"); + core::ParameterContext context("test_context"); + context.addParameter(core::Parameter{"what", "", "love"}); + REQUIRE_THROWS_WITH(parser.replaceParameters(&context, false), "Parameter Operation: Parameter 'who' not found"); +} + +TEST_CASE("Sensitive property parameter replacement is not supported") { + core::ParameterTokenParser parser("What is #{what}, baby don't hurt #{who}, don't hurt #{who}, no more"); + core::ParameterContext context("test_context"); + context.addParameter(core::Parameter{"what", "", "love"}); + context.addParameter(core::Parameter{"who", "", "me"}); + REQUIRE_THROWS_WITH(parser.replaceParameters(&context, true), "Parameter Operation: Non-sensitive parameter 'what' cannot be referenced in a sensitive property"); +} + +TEST_CASE("Parameter context is not provided when parameter is referenced") { + core::ParameterTokenParser parser("What is #{what}, baby don't hurt #{who}, don't hurt #{who}, no more"); + REQUIRE_THROWS_WITH(parser.replaceParameters(nullptr, false), "Parameter Operation: Property references a parameter in its value, but no parameter context was provided."); +} + +TEST_CASE("Replace only escaped tokens") { + core::ParameterTokenParser parser("No ##{parameters} are ####{present}"); + REQUIRE(parser.replaceParameters(nullptr, false) == "No #{parameters} are ##{present}"); + REQUIRE(parser.replaceParameters(nullptr, true) == "No #{parameters} are ##{present}"); +} + +} // namespace org::apache::nifi::minifi::test
