You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by fg...@apache.org on 2021/08/04 16:29:30 UTC
[nifi-minifi-cpp] 01/03: MINIFICPP-1522 Log invalid attribute in
case of YAML parse failure
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 e1433c1b4e04aaaa1bac46d338631e7689549a0c
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Jul 26 14:07:24 2021 +0200
MINIFICPP-1522 Log invalid attribute in case of YAML parse failure
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
This closes #1143
---
.../tests/unit/YamlConfigurationTests.cpp | 327 +++++++++++----------
libminifi/src/core/yaml/YamlConfiguration.cpp | 10 +-
2 files changed, 186 insertions(+), 151 deletions(-)
diff --git a/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp b/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp
index b4b8f6b..8121bda 100644
--- a/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp
+++ b/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp
@@ -18,6 +18,7 @@
#include <map>
#include <memory>
+#include <chrono>
#include "core/repository/VolatileContentRepository.h"
#include "core/ProcessGroup.h"
#include "core/RepositoryFactory.h"
@@ -26,6 +27,8 @@
#include "TestBase.h"
#include "utils/TestUtils.h"
+using namespace std::chrono_literals; // NOLINT using namespace directive is required for literals
+
TEST_CASE("Test YAML Config Processing", "[YamlConfiguration]") {
TestController test_controller;
@@ -37,160 +40,184 @@ TEST_CASE("Test YAML Config Processing", "[YamlConfiguration]") {
core::YamlConfiguration yamlConfig(testProvRepo, testFlowFileRepo, content_repo, streamFactory, configuration);
SECTION("loading YAML without optional component IDs works") {
- static const std::string CONFIG_YAML_WITHOUT_IDS = ""
- "MiNiFi Config Version: 1\n"
- "Flow Controller:\n"
- " name: MiNiFi Flow\n"
- " comment:\n"
- "\n"
- "Core Properties:\n"
- " flow controller graceful shutdown period: 10 sec\n"
- " flow service write delay interval: 500 ms\n"
- " administrative yield duration: 30 sec\n"
- " bored yield duration: 10 millis\n"
- "\n"
- "FlowFile Repository:\n"
- " partitions: 256\n"
- " checkpoint interval: 2 mins\n"
- " always sync: false\n"
- " Swap:\n"
- " threshold: 20000\n"
- " in period: 5 sec\n"
- " in threads: 1\n"
- " out period: 5 sec\n"
- " out threads: 4\n"
- "\n"
- "Provenance Repository:\n"
- " provenance rollover time: 1 min\n"
- "\n"
- "Content Repository:\n"
- " content claim max appendable size: 10 MB\n"
- " content claim max flow files: 100\n"
- " always sync: false\n"
- "\n"
- "Component Status Repository:\n"
- " buffer size: 1440\n"
- " snapshot frequency: 1 min\n"
- "\n"
- "Security Properties:\n"
- " keystore: /tmp/ssl/localhost-ks.jks\n"
- " keystore type: JKS\n"
- " keystore password: localtest\n"
- " key password: localtest\n"
- " truststore: /tmp/ssl/localhost-ts.jks\n"
- " truststore type: JKS\n"
- " truststore password: localtest\n"
- " ssl protocol: TLS\n"
- " Sensitive Props:\n"
- " key:\n"
- " algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL\n"
- " provider: BC\n"
- "\n"
- "Processors:\n"
- " - name: TailFile\n"
- " class: org.apache.nifi.processors.standard.TailFile\n"
- " max concurrent tasks: 1\n"
- " scheduling strategy: TIMER_DRIVEN\n"
- " scheduling period: 1 sec\n"
- " penalization period: 30 sec\n"
- " yield period: 1 sec\n"
- " run duration nanos: 0\n"
- " auto-terminated relationships list:\n"
- " Properties:\n"
- " File to Tail: logs/minifi-app.log\n"
- " Rolling Filename Pattern: minifi-app*\n"
- " Initial Start Position: Beginning of File\n"
- "\n"
- "Connections:\n"
- " - name: TailToS2S\n"
- " source name: TailFile\n"
- " source relationship name: success\n"
- " destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61\n"
- " max work queue size: 0\n"
- " max work queue data size: 1 MB\n"
- " flowfile expiration: 60 sec\n"
- " queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer\n"
- "\n"
- "Remote Processing Groups:\n"
- " - name: NiFi Flow\n"
- " comment:\n"
- " url: https://localhost:8090/nifi\n"
- " timeout: 30 secs\n"
- " yield period: 10 sec\n"
- " Input Ports:\n"
- " - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61\n"
- " name: tailed log\n"
- " comments:\n"
- " max concurrent tasks: 1\n"
- " use compression: false\n"
- "\n"
- "Provenance Reporting:\n"
- " comment:\n"
- " scheduling strategy: TIMER_DRIVEN\n"
- " scheduling period: 30 sec\n"
- " host: localhost\n"
- " port name: provenance\n"
- " port: 8090\n"
- " port uuid: 2f389b8d-83f2-48d3-b465-048f28a1cb56\n"
- " url: https://localhost:8090/\n"
- " originating url: http://${hostname(true)}:8081/nifi\n"
- " use compression: true\n"
- " timeout: 30 secs\n"
- " batch size: 1000";
-
- std::istringstream configYamlStream(CONFIG_YAML_WITHOUT_IDS);
- std::unique_ptr<core::ProcessGroup> rootFlowConfig = yamlConfig.getYamlRoot(configYamlStream);
+ static const std::string CONFIG_YAML_WITHOUT_IDS =
+ R"(
+MiNiFi Config Version: 1
+Flow Controller:
+ name: MiNiFi Flow
+ comment:
- REQUIRE(rootFlowConfig);
- REQUIRE(rootFlowConfig->findProcessorByName("TailFile"));
- utils::Identifier uuid = rootFlowConfig->findProcessorByName("TailFile")->getUUID();
- REQUIRE(uuid);
- REQUIRE(!rootFlowConfig->findProcessorByName("TailFile")->getUUIDStr().empty());
- REQUIRE(1 == rootFlowConfig->findProcessorByName("TailFile")->getMaxConcurrentTasks());
- REQUIRE(
- core::SchedulingStrategy::TIMER_DRIVEN == rootFlowConfig->findProcessorByName("TailFile")->getSchedulingStrategy());
- REQUIRE(1 == rootFlowConfig->findProcessorByName("TailFile")->getMaxConcurrentTasks());
- REQUIRE(1 * 1000 * 1000 * 1000 == rootFlowConfig->findProcessorByName("TailFile")->getSchedulingPeriodNano());
- REQUIRE(std::chrono::seconds(30) == rootFlowConfig->findProcessorByName("TailFile")->getPenalizationPeriod());
- REQUIRE(1 * 1000 == rootFlowConfig->findProcessorByName("TailFile")->getYieldPeriodMsec());
- REQUIRE(0 == rootFlowConfig->findProcessorByName("TailFile")->getRunDurationNano());
+Core Properties:
+ flow controller graceful shutdown period: 10 sec
+ flow service write delay interval: 500 ms
+ administrative yield duration: 30 sec
+ bored yield duration: 10 millis
- std::map<std::string, std::shared_ptr<minifi::Connection>> connectionMap;
- rootFlowConfig->getConnections(connectionMap);
- REQUIRE(2 == connectionMap.size());
- // This is a map of UUID->Connection, and we don't know UUID, so just going to loop over it
- for (auto it : connectionMap) {
- REQUIRE(it.second);
- REQUIRE(!it.second->getUUIDStr().empty());
- REQUIRE(it.second->getDestination());
- REQUIRE(it.second->getSource());
- REQUIRE(60000 == it.second->getFlowExpirationDuration());
+FlowFile Repository:
+ partitions: 256
+ checkpoint interval: 2 mins
+ always sync: false
+ Swap:
+ threshold: 20000
+ in period: 5 sec
+ in threads: 1
+ out period: 5 sec
+ out threads: 4
+
+Provenance Repository:
+ provenance rollover time: 1 min
+
+Content Repository:
+ content claim max appendable size: 10 MB
+ content claim max flow files: 100
+ always sync: false
+
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+
+Security Properties:
+ keystore: /tmp/ssl/localhost-ks.jks
+ keystore type: JKS
+ keystore password: localtest
+ key password: localtest
+ truststore: /tmp/ssl/localhost-ts.jks
+ truststore type: JKS
+ truststore password: localtest
+ ssl protocol: TLS
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+
+Processors:
+ - name: TailFile
+ class: org.apache.nifi.processors.standard.TailFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ File to Tail: logs/minifi-app.log
+ Rolling Filename Pattern: minifi-app*
+ Initial Start Position: Beginning of File
+
+Connections:
+ - name: TailToS2S
+ source name: TailFile
+ source relationship name: success
+ destination name: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ queue prioritizer class: org.apache.nifi.prioritizer.NewestFlowFileFirstPrioritizer
+
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - id: 8644cbcc-a45c-40e0-964d-5e536e2ada61
+ name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+
+Provenance Reporting:
+ comment:
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 30 sec
+ host: localhost
+ port name: provenance
+ port: 8090
+ port uuid: 2f389b8d-83f2-48d3-b465-048f28a1cb56
+ url: https://localhost:8090/
+ originating url: http://${hostname(true)}:8081/nifi
+ use compression: true
+ timeout: 30 secs
+ batch size: 1000;
+ )";
+
+ std::istringstream configYamlStream(CONFIG_YAML_WITHOUT_IDS);
+ std::unique_ptr<core::ProcessGroup> rootFlowConfig = yamlConfig.getYamlRoot(configYamlStream);
+
+ REQUIRE(rootFlowConfig);
+ REQUIRE(rootFlowConfig->findProcessorByName("TailFile"));
+ utils::Identifier uuid = rootFlowConfig->findProcessorByName("TailFile")->getUUID();
+ REQUIRE(uuid);
+ REQUIRE(!rootFlowConfig->findProcessorByName("TailFile")->getUUIDStr().empty());
+ REQUIRE(1 == rootFlowConfig->findProcessorByName("TailFile")->getMaxConcurrentTasks());
+ REQUIRE(core::SchedulingStrategy::TIMER_DRIVEN == rootFlowConfig->findProcessorByName("TailFile")->getSchedulingStrategy());
+ REQUIRE(1 * 1000 * 1000 * 1000 == rootFlowConfig->findProcessorByName("TailFile")->getSchedulingPeriodNano());
+ REQUIRE(30s == rootFlowConfig->findProcessorByName("TailFile")->getPenalizationPeriod());
+ REQUIRE(1 * 1000 == rootFlowConfig->findProcessorByName("TailFile")->getYieldPeriodMsec());
+ REQUIRE(0 == rootFlowConfig->findProcessorByName("TailFile")->getRunDurationNano());
+
+ std::map<std::string, std::shared_ptr<minifi::Connection>> connectionMap;
+ rootFlowConfig->getConnections(connectionMap);
+ REQUIRE(2 == connectionMap.size());
+ // This is a map of UUID->Connection, and we don't know UUID, so just going to loop over it
+ for (auto it : connectionMap) {
+ REQUIRE(it.second);
+ REQUIRE(!it.second->getUUIDStr().empty());
+ REQUIRE(it.second->getDestination());
+ REQUIRE(it.second->getSource());
+ REQUIRE(60000 == it.second->getFlowExpirationDuration());
+ }
}
-}
SECTION("missing required field in YAML throws exception") {
- static const std::string CONFIG_YAML_NO_RPG_PORT_ID = ""
- "MiNiFi Config Version: 1\n"
- "Flow Controller:\n"
- " name: MiNiFi Flow\n"
- "Processors: []\n"
- "Connections: []\n"
- "Remote Processing Groups:\n"
- " - name: NiFi Flow\n"
- " comment:\n"
- " url: https://localhost:8090/nifi\n"
- " timeout: 30 secs\n"
- " yield period: 10 sec\n"
- " Input Ports:\n"
- " - name: tailed log\n"
- " comments:\n"
- " max concurrent tasks: 1\n"
- " use compression: false\n"
- "\n";
-
- std::istringstream configYamlStream(CONFIG_YAML_NO_RPG_PORT_ID);
- REQUIRE_THROWS_AS(yamlConfig.getYamlRoot(configYamlStream), std::invalid_argument);
-}
+ static const std::string CONFIG_YAML_NO_RPG_PORT_ID =
+ R"(
+MiNiFi Config Version: 1
+Flow Controller:
+ name: MiNiFi Flow
+Processors: []
+Connections: []
+Remote Processing Groups:
+ - name: NiFi Flow
+ comment:
+ url: https://localhost:8090/nifi
+ timeout: 30 secs
+ yield period: 10 sec
+ Input Ports:
+ - name: tailed log
+ comments:
+ max concurrent tasks: 1
+ use compression: false
+ )";
+
+ std::istringstream configYamlStream(CONFIG_YAML_NO_RPG_PORT_ID);
+ REQUIRE_THROWS_AS(yamlConfig.getYamlRoot(configYamlStream), std::invalid_argument);
+ }
+
+ SECTION("Validated YAML property failure throws exception and logs invalid attribute name") {
+ static const std::string CONFIG_YAML_EMPTY_RETRY_ATTRIBUTE =
+ R"(
+Flow Controller:
+ name: MiNiFi Flow
+Processors:
+ - name: RetryFlowFile
+ class: org.apache.nifi.processors.standard.RetryFlowFile
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ auto-terminated relationships list:
+ Properties:
+ Retry Attribute: ""
+Connections: []
+Remote Processing Groups: []
+Provenance Reporting:
+ )";
+
+ std::istringstream configYamlStream(CONFIG_YAML_EMPTY_RETRY_ATTRIBUTE);
+ REQUIRE_THROWS_AS(yamlConfig.getYamlRoot(configYamlStream), utils::internal::InvalidValueException);
+ REQUIRE(LogTestController::getInstance().contains("Invalid value was set for property 'Retry Attribute' creating component 'RetryFlowFile'"));
+ }
}
TEST_CASE("Test YAML v3 Invalid Type", "[YamlConfiguration3]") {
@@ -452,7 +479,7 @@ NiFi Properties Overrides: {}
REQUIRE(core::SchedulingStrategy::TIMER_DRIVEN == rootFlowConfig->findProcessorByName("TailFile")->getSchedulingStrategy());
REQUIRE(1 == rootFlowConfig->findProcessorByName("TailFile")->getMaxConcurrentTasks());
REQUIRE(1 * 1000 * 1000 * 1000 == rootFlowConfig->findProcessorByName("TailFile")->getSchedulingPeriodNano());
- REQUIRE(std::chrono::seconds(30) == rootFlowConfig->findProcessorByName("TailFile")->getPenalizationPeriod());
+ REQUIRE(30s == rootFlowConfig->findProcessorByName("TailFile")->getPenalizationPeriod());
REQUIRE(1 * 1000 == rootFlowConfig->findProcessorByName("TailFile")->getYieldPeriodMsec());
REQUIRE(0 == rootFlowConfig->findProcessorByName("TailFile")->getRunDurationNano());
diff --git a/libminifi/src/core/yaml/YamlConfiguration.cpp b/libminifi/src/core/yaml/YamlConfiguration.cpp
index d5d2e6e..1a4850b 100644
--- a/libminifi/src/core/yaml/YamlConfiguration.cpp
+++ b/libminifi/src/core/yaml/YamlConfiguration.cpp
@@ -764,8 +764,16 @@ void YamlConfiguration::parseSingleProperty(const std::string& propertyName, con
core::Property myProp(propertyName, "", "");
processor->getProperty(propertyName, myProp);
const PropertyValue coercedValue = getValidatedProcessorPropertyForDefaultTypeInfo(myProp, propertyValueNode);
+ bool property_set = false;
+ try {
+ property_set = processor->setProperty(myProp, coercedValue);
+ } catch(const utils::internal::InvalidValueException&) {
+ auto component = std::dynamic_pointer_cast<core::CoreComponent>(processor);
+ logger_->log_error("Invalid value was set for property '%s' creating component '%s'", propertyName, component->getName());
+ throw;
+ }
const std::string rawValueString = propertyValueNode.as<std::string>();
- if (!processor->setProperty(myProp, coercedValue)) {
+ if (!property_set) {
std::shared_ptr<core::Connectable> proc = std::dynamic_pointer_cast<core::Connectable>(processor);
if (proc) {
logger_->log_warn("Received property %s with value %s but is not one of the properties for %s. Attempting to add as dynamic property.", propertyName, rawValueString, proc->getName());