You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2021/12/14 17:15:23 UTC
[nifi] 05/15: NIFI-9419 ParseCEF - Upgraded parcefone and supported empty extensions
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.15
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 25a273834ec76032e426a6fa3f12ba570925b7c9
Author: Pierre Villard <pi...@gmail.com>
AuthorDate: Mon Nov 29 18:45:54 2021 +0100
NIFI-9419 ParseCEF - Upgraded parcefone and supported empty extensions
- Upgraded com.fluenda:parcefone from 2.0.0 to 2.1.0
- Added Accept empty extensions property to ParseCEF
This closes #5555
Co-authored-by: David Handermann <ex...@apache.org>
Signed-off-by: David Handermann <ex...@apache.org>
---
.../apache/nifi/processors/standard/ParseCEF.java | 20 +++++++++++---
.../nifi/processors/standard/TestParseCEF.java | 32 ++++++++++++++++++++++
nifi-nar-bundles/nifi-standard-bundle/pom.xml | 2 +-
3 files changed, 49 insertions(+), 5 deletions(-)
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ParseCEF.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ParseCEF.java
index 4d8c7f6..de486a2 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ParseCEF.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ParseCEF.java
@@ -147,6 +147,16 @@ public class ParseCEF extends AbstractProcessor {
.allowableValues("true", "false")
.build();
+ public static final PropertyDescriptor ACCEPT_EMPTY_EXTENSIONS = new PropertyDescriptor.Builder()
+ .name("ACCEPT_EMPTY_EXTENSIONS")
+ .displayName("Accept empty extensions")
+ .description("If set to true, empty extensions will be accepted and will be associated to a null value.")
+ .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
+ .required(true)
+ .defaultValue("false")
+ .allowableValues("true", "false")
+ .build();
+
public static final PropertyDescriptor VALIDATE_DATA = new PropertyDescriptor.Builder()
.name("VALIDATE_DATA")
.displayName("Validate the CEF event")
@@ -200,6 +210,7 @@ public class ParseCEF extends AbstractProcessor {
properties.add(FIELDS_DESTINATION);
properties.add(APPEND_RAW_MESSAGE_TO_JSON);
properties.add(INCLUDE_CUSTOM_EXTENSIONS);
+ properties.add(ACCEPT_EMPTY_EXTENSIONS);
properties.add(VALIDATE_DATA);
properties.add(TIME_REPRESENTATION);
properties.add(DATETIME_REPRESENTATION);
@@ -262,12 +273,13 @@ public class ParseCEF extends AbstractProcessor {
// validator failed to identify an invalid Locale
final Locale parcefoneLocale = Locale.forLanguageTag(context.getProperty(DATETIME_REPRESENTATION).getValue());
final boolean validateData = context.getProperty(VALIDATE_DATA).asBoolean();
- event = parser.parse(buffer, validateData, parcefoneLocale);
+ final boolean acceptEmptyExtensions = context.getProperty(ACCEPT_EMPTY_EXTENSIONS).asBoolean();
+ event = parser.parse(buffer, validateData, acceptEmptyExtensions, parcefoneLocale);
} catch (Exception e) {
// This should never trigger but adding in here as a fencing mechanism to
// address possible ParCEFone bugs.
- getLogger().error("Parser returned unexpected Exception {} while processing {}; routing to failure", new Object[] {e, flowFile});
+ getLogger().error("CEF Parsing Failed: {}", flowFile, e);
session.transfer(flowFile, REL_FAILURE);
return;
}
@@ -339,7 +351,7 @@ public class ParseCEF extends AbstractProcessor {
session.transfer(flowFile, REL_SUCCESS);
} catch (CEFHandlingException e) {
// The flowfile has failed parsing & validation, routing to failure and committing
- getLogger().error("Failed to parse {} as a CEF message due to {}; routing to failure", new Object[] {flowFile, e});
+ getLogger().error("Reading CEF Event Failed: {}", flowFile, e);
// Create a provenance event recording the routing to failure
session.getProvenanceReporter().route(flowFile, REL_FAILURE);
session.transfer(flowFile, REL_FAILURE);
@@ -379,6 +391,7 @@ public class ParseCEF extends AbstractProcessor {
return new ValidationResult.Builder().subject(subject).input(input).valid(false)
.explanation(subject + " cannot be empty").build();
}
+
final Locale testLocale = Locale.forLanguageTag(input);
final Locale[] availableLocales = Locale.getAvailableLocales();
@@ -389,7 +402,6 @@ public class ParseCEF extends AbstractProcessor {
.explanation(input + " is not a valid locale format.").build();
} else {
return new ValidationResult.Builder().subject(subject).input(input).valid(true).build();
-
}
}
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestParseCEF.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestParseCEF.java
index 94c61ca..9ec2e87 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestParseCEF.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestParseCEF.java
@@ -19,6 +19,7 @@ package org.apache.nifi.processors.standard;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
@@ -343,6 +344,37 @@ public class TestParseCEF {
}
@Test
+ public void testAcceptEmptyExtensions() throws Exception {
+ String sample3 = "CEF:0|TestVendor|TestProduct|TestVersion|TestEventClassID|TestName|Low|" +
+ "rt=Feb 09 2015 00:27:43 UTC cn3Label=Test Long cn3= " +
+ "cfp1=1.234 cfp1Label=Test FP Number smac=00:00:0c:07:ac:00 " +
+ "c6a3=2001:cdba::3257:9652 c6a3Label=Test IPv6 cs1Label=Test String cs1=test test test chocolate " +
+ "destinationTranslatedAddress=123.123.123.123 " +
+ "deviceCustomDate1=Feb 06 2015 13:27:43 " +
+ "dpt= agt= dlat=";
+
+ final TestRunner runner = TestRunners.newTestRunner(new ParseCEF());
+ runner.setProperty(ParseCEF.FIELDS_DESTINATION, ParseCEF.DESTINATION_CONTENT);
+ runner.setProperty(ParseCEF.TIME_REPRESENTATION, ParseCEF.UTC);
+ runner.setProperty(ParseCEF.INCLUDE_CUSTOM_EXTENSIONS, "true");
+ runner.setProperty(ParseCEF.ACCEPT_EMPTY_EXTENSIONS, "true");
+ runner.setProperty(ParseCEF.VALIDATE_DATA, "false");
+ runner.enqueue(sample3.getBytes());
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(ParseCEF.REL_SUCCESS, 1);
+ final MockFlowFile mff = runner.getFlowFilesForRelationship(ParseCEF.REL_SUCCESS).get(0);
+
+ byte [] rawJson = mff.toByteArray();
+
+ JsonNode results = new ObjectMapper().readTree(rawJson);
+
+ JsonNode extensions = results.get("extension");
+ Assert.assertTrue(extensions.has("cn3"));
+ Assert.assertTrue(extensions.get("cn3").isNull());
+ }
+
+ @Test
public void testDataValidation() throws Exception {
String invalidEvent = sample1 + " proto=ICMP"; // according to the standard, proto can be either tcp or udp.
diff --git a/nifi-nar-bundles/nifi-standard-bundle/pom.xml b/nifi-nar-bundles/nifi-standard-bundle/pom.xml
index f69c089..a3aad04 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/pom.xml
+++ b/nifi-nar-bundles/nifi-standard-bundle/pom.xml
@@ -316,7 +316,7 @@
<dependency>
<groupId>com.fluenda</groupId>
<artifactId>parcefone</artifactId>
- <version>2.0.0</version>
+ <version>2.1.0</version>
</dependency>
<dependency>
<groupId>com.github.wnameless.json</groupId>