Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-9341 Added CEF RecordReader 

This closes #5530

Signed-off-by: David Handermann <exceptionfactory@apache.org>
This commit is contained in:
Bence Simon 2021-11-17 17:45:12 +01:00 committed by exceptionfactory
parent ed5e90fb81
commit 49978cdd91
No known key found for this signature in database
GPG Key ID: 29B6A52D2AAE8DBA
35 changed files with 2586 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestParseCEF.java
View File

@ -346,7 +346,7 @@ 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= " +
"rt=Feb 09 2015 00:27:43 UTC cn3Label= 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 " +
@ -372,6 +372,9 @@ public class TestParseCEF {
JsonNode extensions = results.get("extension");
Assert.assertTrue(extensions.has("cn3"));
Assert.assertTrue(extensions.get("cn3").isNull());
Assert.assertTrue(extensions.has("cn3Label"));
Assert.assertTrue(extensions.get("cn3Label").asText().isEmpty());
}
@Test

37
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/pom.xml
View File

@ -119,11 +119,31 @@
<artifactId>caffeine</artifactId>
<version>2.8.1</version>
</dependency>
<dependency>
<groupId>javax.validation</groupId>
<artifactId>validation-api</artifactId>
<version>2.0.1.Final</version>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>27.0.1-jre</version>
</dependency>
<dependency>
<groupId>com.fluenda</groupId>
<artifactId>parcefone</artifactId>
<version>2.1.0</version>
</dependency>
<dependency>
<groupId>org.apache.bval</groupId>
<artifactId>bval-jsr</artifactId>
<version>2.0.5</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-mock-record-utils</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
@ -143,6 +163,23 @@
<exclude>src/test/resources/avro/multiple-types.avsc</exclude>
<exclude>src/test/resources/avro/simple.avsc</exclude>
<exclude>src/test/resources/avro/recursive.avsc</exclude>
<exclude>src/test/resources/cef/empty-row.txt</exclude>
<exclude>src/test/resources/cef/misformatted-row.txt</exclude>
<exclude>src/test/resources/cef/multiple-rows.txt</exclude>
<exclude>src/test/resources/cef/multiple-rows-decreasing-number-of-extensions.txt</exclude>
<exclude>src/test/resources/cef/multiple-rows-increasing-number-of-extensions.txt</exclude>
<exclude>src/test/resources/cef/multiple-rows-starting-with-empty-row.txt</exclude>
<exclude>src/test/resources/cef/multiple-rows-with-different-custom-types.txt</exclude>
<exclude>src/test/resources/cef/multiple-rows-with-empty-rows.txt</exclude>
<exclude>src/test/resources/cef/single-row-header-fields-only.txt</exclude>
<exclude>src/test/resources/cef/single-row-with-custom-extensions.txt</exclude>
<exclude>src/test/resources/cef/single-row-with-empty-extension.txt</exclude>
<exclude>src/test/resources/cef/single-row-with-empty-custom-extensions.txt</exclude>
<exclude>src/test/resources/cef/single-row-with-extensions.txt</exclude>
<exclude>src/test/resources/cef/single-row-with-incorrect-custom-extensions.txt</exclude>
<exclude>src/test/resources/cef/single-row-with-incorrect-header-field.txt</exclude>
<exclude>src/test/resources/csv/extra-white-space.csv</exclude>
<exclude>src/test/resources/csv/multi-bank-account.csv</exclude>
<exclude>src/test/resources/csv/single-bank-account.csv</exclude>

38
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/cef/CEFCustomExtensionTypeResolver.java Normal file
View File

@ -0,0 +1,38 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.util.SchemaInferenceUtil;
/**
* Provides strategy for resolving data type for custom extensions.
*/
interface CEFCustomExtensionTypeResolver {
/**
* @param value String representation of the field value.
*
* @return The resolved data type matches the given value most, based on the implemented strategy.
*/
DataType resolve(String value);
CEFCustomExtensionTypeResolver STRING_RESOLVER = value -> RecordFieldType.STRING.getDataType();
CEFCustomExtensionTypeResolver SIMPLE_RESOLVER = value -> SchemaInferenceUtil.getDataType(value);
CEFCustomExtensionTypeResolver SKIPPING_RESOLVER = value -> null;
}

241
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/cef/CEFReader.java Normal file
View File

@ -0,0 +1,241 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import com.fluenda.parcefone.parser.CEFParser;
import org.apache.bval.jsr.ApacheValidationProvider;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.Validator;
import org.apache.nifi.context.PropertyContext;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.schema.access.SchemaAccessStrategy;
import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.schema.inference.SchemaInferenceUtil;
import org.apache.nifi.schemaregistry.services.SchemaRegistry;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.serialization.SchemaRegistryService;
import org.apache.nifi.serialization.record.RecordSchema;
import javax.validation.Validation;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.nifi.schema.access.SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY;
import static org.apache.nifi.schema.inference.SchemaInferenceUtil.SCHEMA_CACHE;
@Tags({"cef", "record", "reader", "parser"})
@CapabilityDescription("Parses CEF (Common Event Format) events, returning each row as a record. "
+ "This reader allows for inferring a schema based on the first event in the FlowFile or providing an explicit schema for interpreting the values.")
public final class CEFReader extends SchemaRegistryService implements RecordReaderFactory {
static final AllowableValue HEADERS_ONLY = new AllowableValue("headers-only", "Headers only", "Includes only CEF header fields into the inferred schema.");
static final AllowableValue HEADERS_AND_EXTENSIONS = new AllowableValue("headers-and-extensions", "Headers and extensions",
"Includes the CEF header and extension fields to the schema, but not the custom extensions.");
static final AllowableValue CUSTOM_EXTENSIONS_AS_STRINGS = new AllowableValue("custom-extensions-as-string", "With custom extensions as strings",
"Includes all fields into the inferred schema, involving custom extension fields as string values.");
static final AllowableValue CUSTOM_EXTENSIONS_INFERRED = new AllowableValue("custom-extensions-inferred", "With custom extensions inferred",
"Includes all fields into the inferred schema, involving custom extension fields with inferred data types. " +
"The inference works based on the values in the FlowFile. In some scenarios this might result unsatisfiable behaviour. " +
"In these cases it is suggested to use \"" + CUSTOM_EXTENSIONS_AS_STRINGS.getDisplayName() + "\" Inference Strategy or predefined schema.");
static final PropertyDescriptor INFERENCE_STRATEGY = new PropertyDescriptor.Builder()
.name("inference-strategy")
.displayName("Inference Strategy")
.description("Defines the set of fields should be included in the schema and the way the fields are being interpreted.")
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.required(true)
.allowableValues(HEADERS_ONLY, HEADERS_AND_EXTENSIONS, CUSTOM_EXTENSIONS_AS_STRINGS, CUSTOM_EXTENSIONS_INFERRED)
.defaultValue(CUSTOM_EXTENSIONS_INFERRED.getValue())
.dependsOn(SCHEMA_ACCESS_STRATEGY, SchemaInferenceUtil.INFER_SCHEMA)
.build();
static final PropertyDescriptor RAW_FIELD = new PropertyDescriptor.Builder()
.name("raw-message-field")
.displayName("Raw Message Field")
.description("If set the raw message will be added to the record using the property value as field name. This is not the same as the \"rawEvent\" extension field!")
.addValidator(new ValidateRawField())
.required(false)
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
.build();
static final PropertyDescriptor INVALID_FIELD = new PropertyDescriptor.Builder()
.name("invalid-message-field")
.displayName("Invalid Field")
.description("Used when a line in the FlowFile cannot be parsed by the CEF parser. " +
"If set, instead of failing to process the FlowFile, a record is being added with one field. " +
"This record contains one field with the name specified by the property and the raw message as value.")
.addValidator(new ValidateRawField())
.required(false)
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
.build();
static final PropertyDescriptor DATETIME_REPRESENTATION = new PropertyDescriptor.Builder()
.name("datetime-representation")
.displayName("DateTime Locale")
.description("The IETF BCP 47 representation of the Locale to be used when parsing date " +
"fields with long or short month names (e.g. may <en-US> vs. mai. <fr-FR>. The default" +
"value is generally safe. Only change if having issues parsing CEF messages")
.required(true)
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
.addValidator(new ValidateLocale())
.defaultValue("en-US")
.build();
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();
private final javax.validation.Validator validator = Validation.byProvider(ApacheValidationProvider.class).configure().buildValidatorFactory().getValidator();
private final CEFParser parser = new CEFParser(validator);
private volatile String rawMessageField;
private volatile String invalidField;
private volatile Locale parcefoneLocale;
private volatile boolean includeCustomExtensions;
private volatile boolean acceptEmptyExtensions;
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
final List<PropertyDescriptor> properties = new ArrayList<>(super.getSupportedPropertyDescriptors());
properties.add(RAW_FIELD);
properties.add(INVALID_FIELD);
properties.add(DATETIME_REPRESENTATION);
properties.add(INFERENCE_STRATEGY);
properties.add(new PropertyDescriptor.Builder()
.fromPropertyDescriptor(SCHEMA_CACHE)
.dependsOn(SCHEMA_ACCESS_STRATEGY, SchemaInferenceUtil.INFER_SCHEMA)
.build());
properties.add(ACCEPT_EMPTY_EXTENSIONS);
return properties;
}
@Override
protected List<AllowableValue> getSchemaAccessStrategyValues() {
final List<AllowableValue> allowableValues = new ArrayList<>(super.getSchemaAccessStrategyValues());
allowableValues.add(SchemaInferenceUtil.INFER_SCHEMA);
return allowableValues;
}
@Override
protected AllowableValue getDefaultSchemaAccessStrategy() {
return SchemaInferenceUtil.INFER_SCHEMA;
}
@Override
protected SchemaAccessStrategy getSchemaAccessStrategy(final String strategy, final SchemaRegistry schemaRegistry, final PropertyContext context) {
if (strategy.equals(SchemaInferenceUtil.INFER_SCHEMA.getValue())) {
final String inferenceStrategy = context.getProperty(INFERENCE_STRATEGY).getValue();
final CEFSchemaInferenceBuilder builder = new CEFSchemaInferenceBuilder();
if (inferenceStrategy.equals(HEADERS_AND_EXTENSIONS.getValue())) {
builder.withExtensions();
} else if (inferenceStrategy.equals(CUSTOM_EXTENSIONS_AS_STRINGS.getValue())) {
builder.withCustomExtensions(CEFCustomExtensionTypeResolver.STRING_RESOLVER);
} else if (inferenceStrategy.equals(CUSTOM_EXTENSIONS_INFERRED.getValue())) {
builder.withCustomExtensions(CEFCustomExtensionTypeResolver.SIMPLE_RESOLVER);
}
if (rawMessageField != null) {
builder.withRawMessage(rawMessageField);
}
if (invalidField != null) {
builder.withInvalidField(invalidField);
}
final boolean failFast = invalidField == null || invalidField.isEmpty();
final CEFSchemaInference inference = builder.build();
return SchemaInferenceUtil.getSchemaAccessStrategy(
strategy,
context,
getLogger(),
(variables, in) -> new CEFRecordSource(in, parser, parcefoneLocale, acceptEmptyExtensions, failFast),
() -> inference,
() -> super.getSchemaAccessStrategy(strategy, schemaRegistry, context));
}
return super.getSchemaAccessStrategy(strategy, schemaRegistry, context);
}
@OnEnabled
public void onEnabled(final ConfigurationContext context) {
rawMessageField = context.getProperty(RAW_FIELD).evaluateAttributeExpressions().getValue();
invalidField = context.getProperty(INVALID_FIELD).evaluateAttributeExpressions().getValue();
parcefoneLocale = Locale.forLanguageTag(context.getProperty(DATETIME_REPRESENTATION).evaluateAttributeExpressions().getValue());
final String inferenceStrategy = context.getProperty(INFERENCE_STRATEGY).getValue();
final boolean inferenceNeedsCustomExtensions = !inferenceStrategy.equals(HEADERS_ONLY.getValue()) && !inferenceStrategy.equals(HEADERS_AND_EXTENSIONS.getValue());
final boolean isInferSchema = context.getProperty(SCHEMA_ACCESS_STRATEGY).getValue().equals(SchemaInferenceUtil.INFER_SCHEMA.getValue());
includeCustomExtensions = !isInferSchema || inferenceNeedsCustomExtensions;
acceptEmptyExtensions = context.getProperty(ACCEPT_EMPTY_EXTENSIONS).asBoolean();
}
@Override
public RecordReader createRecordReader(
final Map<String, String> variables, final InputStream in, final long inputLength, final ComponentLog logger
) throws MalformedRecordException, IOException, SchemaNotFoundException {
final RecordSchema schema = getSchema(variables, in, null);
return new CEFRecordReader(in, schema, parser, logger, parcefoneLocale, rawMessageField, invalidField, includeCustomExtensions, acceptEmptyExtensions);
}
private static class ValidateRawField implements Validator {
private final Set<String> headerFields = CEFSchemaUtil.getHeaderFields().stream().map(r -> r.getFieldName()).collect(Collectors.toSet());
private final Set<String> extensionFields = CEFSchemaUtil.getExtensionTypeMapping().keySet().stream().flatMap(fields -> fields.stream()).collect(Collectors.toSet());
@Override
public ValidationResult validate(final String subject, final String input, final ValidationContext context) {
if (headerFields.contains(input)) {
return new ValidationResult.Builder().subject(subject).input(input).valid(false)
.explanation(input + " is one of the CEF header fields.").build();
}
if (extensionFields.contains(input)) {
return new ValidationResult.Builder().subject(subject).input(input).valid(false)
.explanation(input + " is one of the CEF extension fields.").build();
}
// Field names are not part of the specified CEF field names are accepted, just like null or empty value
return new ValidationResult.Builder().subject(subject).input(input).valid(true).build();
}
}
}

215
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/cef/CEFRecordReader.java Normal file
View File

@ -0,0 +1,215 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import com.fluenda.parcefone.event.CEFHandlingException;
import com.fluenda.parcefone.event.CommonEvent;
import com.fluenda.parcefone.parser.CEFParser;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.util.DataTypeUtils;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.InetAddress;
import java.text.DateFormat;
import java.util.Collections;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import java.util.function.Supplier;
/**
* CEF (Common Event Format) based implementation for {@code org.apache.nifi.serialization.RecordReader}. The implementation
* builds on top of the {@code com.fluenda.parcefone.parser.CEFParser} which is responsible to deal with the textual representation
* of the events. This implementation intends to fill the gap between the CEF parser's {@code com.fluenda.parcefone.event.CommonEvent}
* data objects and the NiFi's internal Record representation.
*/
final class CEFRecordReader implements RecordReader {
/**
* Currently these are not used but set the way it follows the expected CEF format.
*/
private static final String DATE_FORMAT = "MMM dd yyyy";
private static final String TIME_FORMAT = "HH:mm:ss";
private static final String DATETIME_FORMAT = DATE_FORMAT + " " + TIME_FORMAT;
private static final Supplier<DateFormat> DATE_FORMAT_SUPPLIER = () -> DataTypeUtils.getDateFormat(DATE_FORMAT);
private static final Supplier<DateFormat> TIME_FORMAT_SUPPLIER = () -> DataTypeUtils.getDateFormat(TIME_FORMAT);
private static final Supplier<DateFormat> DATETIME_FORMAT_SUPPLIER = () -> DataTypeUtils.getDateFormat(DATETIME_FORMAT);
private final RecordSchema schema;
private final BufferedReader reader;
private final CEFParser parser;
private final ComponentLog logger;
private final Locale locale;
private final String rawMessageField;
private final String invalidField;
/**
* It would not cause any functional drawback to acquire the custom extensions all the time, but there are some cases
* (inferred schema when custom extension fields are not included) where we can be sure about that these fields are not
* required. For better performance, in these cases we do not work with these fields.
*/
private final boolean includeCustomExtensions;
private final boolean acceptEmptyExtensions;
CEFRecordReader(
final InputStream inputStream,
final RecordSchema recordSchema,
final CEFParser parser,
final ComponentLog logger,
final Locale locale,
final String rawMessageField,
final String invalidField,
final boolean includeCustomExtensions,
final boolean acceptEmptyExtensions
) {
this.reader = new BufferedReader(new InputStreamReader(inputStream));
this.schema = recordSchema;
this.parser = parser;
this.logger = logger;
this.locale = locale;
this.rawMessageField = rawMessageField;
this.invalidField = invalidField;
this.includeCustomExtensions = includeCustomExtensions;
this.acceptEmptyExtensions = acceptEmptyExtensions;
}
@Override
public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException {
final String line = nextLine();
if (line == null) {
return null;
}
final CommonEvent event = parser.parse(line, true, acceptEmptyExtensions, locale);
if (event == null) {
logger.debug("Event parsing resulted no event");
if (invalidField != null && !invalidField.isEmpty()) {
return new MapRecord(schema, Collections.singletonMap(invalidField, line));
} else {
throw new MalformedRecordException("The following line could not be parsed by the CEF parser: " + line);
}
}
final Map<String, Object> values = new HashMap<>();
try {
event.getHeader().entrySet().forEach(field -> values.put(field.getKey(), convertValue(field.getKey(), field.getValue(), coerceTypes)));
event.getExtension(true, includeCustomExtensions).entrySet().forEach(field -> values.put(field.getKey(), convertValue(field.getKey() ,field.getValue(), coerceTypes)));
for (final String fieldName : schema.getFieldNames()) {
if (!values.containsKey(fieldName)) {
values.put(fieldName, null);
}
}
} catch (final CEFHandlingException e) {
throw new MalformedRecordException("Error during extracting information from CEF event", e);
}
if (rawMessageField != null) {
values.put(rawMessageField, line);
}
return new MapRecord(schema, values, true, dropUnknownFields);
}
private String nextLine() throws IOException {
String line;
while((line = reader.readLine()) != null) {
if (!line.isEmpty()) {
break;
}
}
return line;
}
private Object convertValue(final String fieldName, final Object fieldValue, final boolean coerceType) {
final DataType dataType = schema.getDataType(fieldName).orElse(RecordFieldType.STRING.getDataType());
return coerceType
? convert(fieldValue, dataType, fieldName)
: convertSimpleIfPossible(fieldValue, dataType, fieldName);
}
private Object convert(final Object value, final DataType dataType, final String fieldName) {
return DataTypeUtils.convertType(prepareValue(value), dataType, DATE_FORMAT_SUPPLIER, TIME_FORMAT_SUPPLIER, DATETIME_FORMAT_SUPPLIER, fieldName);
}
private Object convertSimpleIfPossible(final Object value, final DataType dataType, final String fieldName) {
if (dataType == null || value == null) {
return value;
}
final Object preparedValue = prepareValue(value);
switch (dataType.getFieldType()) {
case STRING:
return preparedValue;
case BOOLEAN:
case INT:
case LONG:
case FLOAT:
case DOUBLE:
case DECIMAL:
if (DataTypeUtils.isCompatibleDataType(preparedValue, dataType)) {
return DataTypeUtils.convertType(preparedValue, dataType, DATE_FORMAT_SUPPLIER, TIME_FORMAT_SUPPLIER, DATETIME_FORMAT_SUPPLIER, fieldName);
}
break;
case TIMESTAMP:
if (DataTypeUtils.isTimestampTypeCompatible(preparedValue, DATETIME_FORMAT)) {
return DataTypeUtils.convertType(preparedValue, dataType, DATE_FORMAT_SUPPLIER, TIME_FORMAT_SUPPLIER, DATETIME_FORMAT_SUPPLIER, fieldName);
}
break;
}
return value;
}
private Object prepareValue(final Object value) {
// InetAddress event fields are mapped as string values internally. Inet4Address and Inet6Address fields are handled by this too.
if (value instanceof InetAddress) {
return ((InetAddress) value).getHostAddress();
}
return value;
}
@Override
public RecordSchema getSchema() throws MalformedRecordException {
return schema;
}
@Override
public void close() throws IOException {
reader.close();
}
}

71
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/cef/CEFRecordSource.java Normal file
View File

@ -0,0 +1,71 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.nifi.cef;
import com.fluenda.parcefone.event.CommonEvent;
import com.fluenda.parcefone.parser.CEFParser;
import org.apache.nifi.schema.inference.RecordSource;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.Locale;
final class CEFRecordSource implements RecordSource<CommonEvent> {
private final CEFParser parser;
private final BufferedReader reader;
private final Locale locale;
private final boolean acceptEmptyExtensions;
private final boolean failFast;
CEFRecordSource(final InputStream in, final CEFParser parser, final Locale locale, final boolean acceptEmptyExtensions, final boolean failFast) {
this.parser = parser;
this.reader = new BufferedReader(new InputStreamReader(in));
this.locale = locale;
this.acceptEmptyExtensions = acceptEmptyExtensions;
this.failFast = failFast;
}
@Override
public CommonEvent next() throws IOException {
final String line = nextLine();
if (line == null) {
return null;
}
final CommonEvent event = parser.parse(line, true, acceptEmptyExtensions, locale);
if (event == null && failFast) {
throw new IOException("Could not parse event");
}
return event;
}
private String nextLine() throws IOException {
String line;
while((line = reader.readLine()) != null) {
if (!line.isEmpty()) {
break;
}
}
return line;
}
}

119
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/cef/CEFSchemaInference.java Normal file
View File

@ -0,0 +1,119 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import com.fluenda.parcefone.event.CEFHandlingException;
import com.fluenda.parcefone.event.CommonEvent;
import org.apache.nifi.schema.inference.FieldTypeInference;
import org.apache.nifi.schema.inference.RecordSource;
import org.apache.nifi.schema.inference.SchemaInferenceEngine;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
final class CEFSchemaInference implements SchemaInferenceEngine<CommonEvent> {
private final boolean includeExtensions;
private final boolean includeCustomExtensions;
private final CEFCustomExtensionTypeResolver typeResolver;
private final String rawMessageField;
private final String invalidField;
CEFSchemaInference(
final boolean includeExtensions,
final boolean includeCustomExtensions,
final CEFCustomExtensionTypeResolver typeResolver,
final String rawMessageField,
final String invalidField) {
this.includeExtensions = includeExtensions;
this.includeCustomExtensions = includeCustomExtensions;
this.typeResolver = typeResolver;
this.rawMessageField = rawMessageField;
this.invalidField = invalidField;
}
@Override
public RecordSchema inferSchema(final RecordSource<CommonEvent> recordSource) throws IOException {
try {
CommonEvent event;
// Header fields are always part of the schema
final List<RecordField> fields = new ArrayList<>(CEFSchemaUtil.getHeaderFields());
final Set<String> extensionFields = new HashSet<>();
final Map<String, FieldTypeInference> customExtensionTypes = new LinkedHashMap<>();
// Even though we need to process this only when we want extensions, the records are being read in order to check if the are valid
while ((event = recordSource.next()) != null) {
if (includeExtensions) {
for (final Map.Entry<String, Object> field : event.getExtension(true, includeCustomExtensions).entrySet()) {
final Optional<RecordField> extensionField = getExtensionField(field.getKey());
if (extensionField.isPresent()) {
// We add extension field to the schema only at the first time
if (!extensionFields.contains(field.getKey())) {
extensionFields.add(field.getKey());
fields.add(extensionField.get());
}
} else if (includeCustomExtensions) {
// CommonEvent will always return custom extensions as String values
final FieldTypeInference typeInference = customExtensionTypes.computeIfAbsent(field.getKey(), key -> new FieldTypeInference());
typeInference.addPossibleDataType(typeResolver.resolve((String) field.getValue()));
}
}
}
}
final List<RecordField> customExtensionFields = new ArrayList<>(customExtensionTypes.size());
customExtensionTypes.forEach((fieldName, type) -> customExtensionFields.add(new RecordField(fieldName, type.toDataType(), true)));
fields.addAll(customExtensionFields);
if (rawMessageField != null) {
fields.add(new RecordField(rawMessageField, RecordFieldType.STRING.getDataType()));
}
if (invalidField != null) {
fields.add(new RecordField(invalidField, RecordFieldType.STRING.getDataType()));
}
return new SimpleRecordSchema(fields);
} catch (final CEFHandlingException e) {
throw new IOException(e);
}
}
private Optional<RecordField> getExtensionField(final String fieldName) {
for (final Map.Entry<Set<String>, DataType> entry : CEFSchemaUtil.getExtensionTypeMapping().entrySet()) {
if (entry.getKey().contains(fieldName)) {
return Optional.of(new RecordField(fieldName, entry.getValue()));
}
}
return Optional.empty();
}
}

51
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/cef/CEFSchemaInferenceBuilder.java Normal file
View File

@ -0,0 +1,51 @@
/*
* 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.
*/
package org.apache.nifi.cef;
final class CEFSchemaInferenceBuilder {
private String rawMessageField = null;
private String invalidField = null;
private boolean includeExtensions = false;
private boolean includeCustomExtensions = false;
private CEFCustomExtensionTypeResolver dataTypeResolver = CEFCustomExtensionTypeResolver.SKIPPING_RESOLVER;
CEFSchemaInferenceBuilder withRawMessage(final String rawMessageField) {
this.rawMessageField = rawMessageField;
return this;
}
CEFSchemaInferenceBuilder withInvalidField(final String invalidField) {
this.invalidField = invalidField;
return this;
}
CEFSchemaInferenceBuilder withExtensions() {
this.includeExtensions = true;
return this;
}
CEFSchemaInferenceBuilder withCustomExtensions(final CEFCustomExtensionTypeResolver dataTypeResolver) {
withExtensions();
this.includeCustomExtensions = true;
this.dataTypeResolver = dataTypeResolver;
return this;
}
CEFSchemaInference build() {
return new CEFSchemaInference(includeExtensions, includeCustomExtensions, dataTypeResolver, rawMessageField, invalidField);
}
}

100
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/cef/CEFSchemaUtil.java Normal file
View File

@ -0,0 +1,100 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
final class CEFSchemaUtil {
private static final List<RecordField> HEADER_FIELDS = new ArrayList<>();
static {
// Reference states that severity might be represented by integer values (0-10) and string values (like High) too
HEADER_FIELDS.add(new RecordField("version", RecordFieldType.INT.getDataType()));
HEADER_FIELDS.add(new RecordField("deviceVendor", RecordFieldType.STRING.getDataType()));
HEADER_FIELDS.add(new RecordField("deviceProduct", RecordFieldType.STRING.getDataType()));
HEADER_FIELDS.add(new RecordField("deviceVersion", RecordFieldType.STRING.getDataType()));
HEADER_FIELDS.add(new RecordField("deviceEventClassId", RecordFieldType.STRING.getDataType()));
HEADER_FIELDS.add(new RecordField("name", RecordFieldType.STRING.getDataType()));
HEADER_FIELDS.add(new RecordField("severity", RecordFieldType.STRING.getDataType()));
}
// Fields known by the CEF Extension Dictionary in version 23
private static final Set<String> EXTENSIONS_STRING = new HashSet<>(Arrays.asList("act", "app", "c6a1Label", "c6a2Label", "c6a3Label",
"c6a4Label", "cfp1Label", "cfp2Label", "cfp3Label", "cfp4Label", "cn1Label", "cn2Label", "cn3Label", "cs1", "cs1Label",
"cs2", "cs2Label", "cs3", "cs3Label", "cs4", "cs4Label", "cs5", "cs5Label", "cs6", "cs6Label", "destinationDnsDomain",
"destinationServiceName", "destinationTranslatedAddress", "destinationTranslatedPort", "deviceCustomDate1Label", "deviceCustomDate2Label",
"deviceDnsDomain", "deviceExternalId", "deviceFacility", "deviceInboundInterface", "deviceNtDomain", "deviceOutboundInterface",
"deviceProcessName", "dhost", "dntdom", "dpriv", "dproc", "dtz", "duid", "duser", "dvchost", "externalId", "fileHash", "field", "filePath",
"filePermission", "fileType", "flexDate1Label", "flexNumber1Label", "flexNumber2Label", "flexString1", "flexString1Label",
"flexString2", "flexString2Label", "fname", "msg", "oldFileHash", "oldField", "oldFileName", "oldFilePath", "oldFilePermission",
"oldFileType", "outcome", "proto", "reason", "request", "requestClientApplication", "requestContext", "requestCookies", "requestMethod",
"shost", "sntdom", "sourceDnsDomain", "sourceServiceName", "spriv", "sproc", "suid", "suser", "agentDnsDomain", "agentNtDomain",
"agentTranslatedZoneExternalID", "agentTranslatedZoneURI", "ahost", "aid", "at", "atz", "av", "cat", "customerExternalID", "customerURI",
"destinationTranslatedZoneExternalID", "destinationTranslatedZoneURI", "destinationZoneExternalID", "destinationZoneURI",
"deviceTranslatedZoneExternalID", "deviceTranslatedZoneURI", "deviceZoneExternalID", "deviceZoneURI", "rawEvent", "sourceTranslatedZoneExternalID",
"sourceTranslatedZoneURI", "sourceZoneExternalID", "sourceZoneURI"));
private static final Set<String> EXTENSIONS_INTEGER = new HashSet<>(Arrays.asList("cnt", "deviceDirection", "dpid", "dpt", "dvcpid", "fsize", "in",
"oldFileSize", "out", "sourceTranslatedPort", "spid", "spt", "type"));
private static final Set<String> EXTENSIONS_LONG = new HashSet<>(Arrays.asList("cn1", "cn2", "cn3", "flexNumber1", "flexNumber2", "eventId"));
private static final Set<String> EXTENSIONS_FLOAT = new HashSet<>(Arrays.asList("cfp1", "cfp2", "cfp3", "cfp4"));
private static final Set<String> EXTENSIONS_DOUBLE = new HashSet<>(Arrays.asList("dlat", "dlong", "slat", "slong"));
private static final Set<String> EXTENSIONS_INET_ADDRESS = new HashSet<>(Arrays.asList("agt"));
private static final Set<String> EXTENSIONS_IPV4_ADDRESS = new HashSet<>(Arrays.asList("deviceTranslatedAddress", "dst", "dvc", "sourceTranslatedAddress", "src",
"agentTranslatedAddress"));
private static final Set<String> EXTENSIONS_IPV6_ADDRESS = new HashSet<>(Arrays.asList("c6a1", "c6a2", "c6a3", "c6a4"));
private static final Set<String> EXTENSIONS_MAC_ADDRESS = new HashSet<>(Arrays.asList("dmac", "dvcmac", "smac", "amac"));
private static final Set<String> EXTENSIONS_TIMESTAMP = new HashSet<>(Arrays.asList("deviceCustomDate1", "deviceCustomDate2", "end", "fileCreatedTime",
"fileModificationTime", "flexDate1", "oldFileCreateTime", "oldFileModificationTime", "rt", "start", "art"));
private static final Map<Set<String>, DataType> EXTENSION_TYPE_MAPPING = new HashMap<>();
static {
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_STRING, RecordFieldType.STRING.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_INTEGER, RecordFieldType.INT.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_LONG, RecordFieldType.LONG.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_FLOAT, RecordFieldType.FLOAT.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_DOUBLE, RecordFieldType.DOUBLE.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_INET_ADDRESS, RecordFieldType.STRING.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_IPV4_ADDRESS, RecordFieldType.STRING.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_IPV6_ADDRESS, RecordFieldType.STRING.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_MAC_ADDRESS, RecordFieldType.STRING.getDataType());
EXTENSION_TYPE_MAPPING.put(EXTENSIONS_TIMESTAMP, RecordFieldType.TIMESTAMP.getDataType());
}
static List<RecordField> getHeaderFields() {
return HEADER_FIELDS;
}
static Map<Set<String>, DataType> getExtensionTypeMapping() {
return EXTENSION_TYPE_MAPPING;
}
private CEFSchemaUtil() {
// Util class, not to be instantiated
}
}

49
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/cef/ValidateLocale.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.Validator;
import java.util.Locale;
/**
* This class is identical to {@code org.apache.nifi.processors.standard.ParseCEF.ValidateLocale}.
*/
public class ValidateLocale implements Validator {
@Override
public ValidationResult validate(String subject, String input, ValidationContext context) {
if (null == input || input.isEmpty()) {
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();
// Check if the provided Locale is valid by checking against the first value of the array (i.e. "null" locale)
if (availableLocales[0].equals(testLocale)) {
// Locale matches the "null" locale so it is treated as invalid
return new ValidationResult.Builder().subject(subject).input(input).valid(false)
.explanation(input + " is not a valid locale format.").build();
} else {
return new ValidationResult.Builder().subject(subject).input(input).valid(true).build();
}
}
}

55
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/csv/CSVSchemaInference.java
View File

@ -17,7 +17,6 @@
package org.apache.nifi.csv;
import org.apache.commons.csv.CSVRecord;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.nifi.schema.inference.FieldTypeInference;
import org.apache.nifi.schema.inference.RecordSource;
import org.apache.nifi.schema.inference.SchemaInferenceEngine;
@ -25,15 +24,14 @@ import org.apache.nifi.schema.inference.TimeValueInference;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.util.SchemaInferenceUtil;
import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
public class CSVSchemaInference implements SchemaInferenceEngine<CSVRecordAndFieldNames> {
@ -43,7 +41,6 @@ public class CSVSchemaInference implements SchemaInferenceEngine<CSVRecordAndFie
this.timeValueInference = timeValueInference;
}
@Override
public RecordSchema inferSchema(final RecordSource<CSVRecordAndFieldNames> recordSource) throws IOException {
final Map<String, FieldTypeInference> typeMap = new LinkedHashMap<>();
@ -67,7 +64,6 @@ public class CSVSchemaInference implements SchemaInferenceEngine<CSVRecordAndFie
return createSchema(typeMap);
}
private void inferSchema(final CSVRecordAndFieldNames recordAndFieldNames, final Map<String, FieldTypeInference> typeMap) {
final CSVRecord csvRecord = recordAndFieldNames.getRecord();
for (final String fieldName : recordAndFieldNames.getFieldNames()) {
@ -78,7 +74,7 @@ public class CSVSchemaInference implements SchemaInferenceEngine<CSVRecordAndFie
final FieldTypeInference typeInference = typeMap.computeIfAbsent(fieldName, key -> new FieldTypeInference());
final String trimmed = trim(value);
final DataType dataType = getDataType(trimmed);
final DataType dataType = SchemaInferenceUtil.getDataType(trimmed, timeValueInference);
typeInference.addPossibleDataType(dataType);
}
}
@ -87,56 +83,9 @@ public class CSVSchemaInference implements SchemaInferenceEngine<CSVRecordAndFie
return (value.length() > 1) && value.startsWith("\"") && value.endsWith("\"") ? value.substring(1, value.length() - 1) : value;
}
private DataType getDataType(final String value) {
if (value == null || value.isEmpty()) {
return null;
}
if (NumberUtils.isParsable(value)) {
if (value.contains(".")) {
try {
final double doubleValue = Double.parseDouble(value);
if (doubleValue == Double.POSITIVE_INFINITY || doubleValue == Double.NEGATIVE_INFINITY) {
return RecordFieldType.DECIMAL.getDecimalDataType(value.length() - 1, value.length() - 1 - value.indexOf("."));
}
if (doubleValue > Float.MAX_VALUE || doubleValue < Float.MIN_VALUE) {
return RecordFieldType.DOUBLE.getDataType();
}
return RecordFieldType.FLOAT.getDataType();
} catch (final NumberFormatException nfe) {
return RecordFieldType.STRING.getDataType();
}
}
try {
final long longValue = Long.parseLong(value);
if (longValue > Integer.MAX_VALUE || longValue < Integer.MIN_VALUE) {
return RecordFieldType.LONG.getDataType();
}
return RecordFieldType.INT.getDataType();
} catch (final NumberFormatException nfe) {
return RecordFieldType.STRING.getDataType();
}
}
if (value.equalsIgnoreCase("true") || value.equalsIgnoreCase("false")) {
return RecordFieldType.BOOLEAN.getDataType();
}
final Optional<DataType> timeDataType = timeValueInference.getDataType(value);
return timeDataType.orElse(RecordFieldType.STRING.getDataType());
}
private RecordSchema createSchema(final Map<String, FieldTypeInference> inferences) {
final List<RecordField> recordFields = new ArrayList<>(inferences.size());
inferences.forEach((fieldName, type) -> recordFields.add(new RecordField(fieldName, type.toDataType(), true)));
return new SimpleRecordSchema(recordFields);
}
}

86
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/util/SchemaInferenceUtil.java Normal file
View File

@ -0,0 +1,86 @@
/*
* 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.
*/
package org.apache.nifi.util;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.nifi.schema.inference.TimeValueInference;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.RecordFieldType;
import java.util.Optional;
public final class SchemaInferenceUtil {
private SchemaInferenceUtil() {
// Not intended to be instantiated
}
public static DataType getDataType(final String value) {
return getDataType(value, Optional.empty());
}
public static DataType getDataType(final String value, final TimeValueInference timeValueInference) {
return getDataType(value, Optional.of(timeValueInference));
}
private static DataType getDataType(final String value, final Optional<TimeValueInference> timeValueInference) {
if (value == null || value.isEmpty()) {
return null;
}
if (NumberUtils.isParsable(value)) {
if (value.contains(".")) {
try {
final double doubleValue = Double.parseDouble(value);
if (doubleValue == Double.POSITIVE_INFINITY || doubleValue == Double.NEGATIVE_INFINITY) {
return RecordFieldType.DECIMAL.getDecimalDataType(value.length() - 1, value.length() - 1 - value.indexOf("."));
}
if (doubleValue > Float.MAX_VALUE || doubleValue < Float.MIN_VALUE) {
return RecordFieldType.DOUBLE.getDataType();
}
return RecordFieldType.FLOAT.getDataType();
} catch (final NumberFormatException nfe) {
return RecordFieldType.STRING.getDataType();
}
}
try {
final long longValue = Long.parseLong(value);
if (longValue > Integer.MAX_VALUE || longValue < Integer.MIN_VALUE) {
return RecordFieldType.LONG.getDataType();
}
return RecordFieldType.INT.getDataType();
} catch (final NumberFormatException nfe) {
return RecordFieldType.STRING.getDataType();
}
}
if (value.equalsIgnoreCase("true") || value.equalsIgnoreCase("false")) {
return RecordFieldType.BOOLEAN.getDataType();
}
if (timeValueInference.isPresent()) {
final Optional<DataType> timeDataType = timeValueInference.get().getDataType(value);
return timeDataType.orElse(RecordFieldType.STRING.getDataType());
} else {
return RecordFieldType.STRING.getDataType();
}
}
}

47
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/xml/inference/XmlSchemaInference.java
View File

@ -16,15 +16,13 @@
*/
package org.apache.nifi.xml.inference;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.nifi.schema.inference.HierarchicalSchemaInference;
import org.apache.nifi.schema.inference.TimeValueInference;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.util.SchemaInferenceUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Optional;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
@ -49,50 +47,9 @@ public class XmlSchemaInference extends HierarchicalSchemaInference<XmlNode> {
}
public DataType inferTextualDataType(final String text) {
if (text == null || text.isEmpty()) {
return null;
return SchemaInferenceUtil.getDataType(text, timeValueInference);
}
if (NumberUtils.isParsable(text)) {
if (text.contains(".")) {
try {
final double doubleValue = Double.parseDouble(text);
if (doubleValue == Double.POSITIVE_INFINITY || doubleValue == Double.NEGATIVE_INFINITY) {
return RecordFieldType.DECIMAL.getDecimalDataType(text.length() - 1, text.length() - 1 - text.indexOf("."));
}
if (doubleValue > Float.MAX_VALUE || doubleValue < Float.MIN_VALUE) {
return RecordFieldType.DOUBLE.getDataType();
}
return RecordFieldType.FLOAT.getDataType();
} catch (final NumberFormatException nfe) {
return RecordFieldType.STRING.getDataType();
}
}
try {
final long longValue = Long.parseLong(text);
if (longValue > Integer.MAX_VALUE || longValue < Integer.MIN_VALUE) {
return RecordFieldType.LONG.getDataType();
}
return RecordFieldType.INT.getDataType();
} catch (final NumberFormatException nfe) {
return RecordFieldType.STRING.getDataType();
}
}
if (text.equalsIgnoreCase("true") || text.equalsIgnoreCase("false")) {
return RecordFieldType.BOOLEAN.getDataType();
}
final Optional<DataType> timeDataType = timeValueInference.getDataType(text);
return timeDataType.orElse(RecordFieldType.STRING.getDataType());
}
@Override
protected boolean isObject(final XmlNode value) {
return value.getNodeType() == XmlNodeType.CONTAINER;

2
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService
View File

@ -26,6 +26,8 @@ org.apache.nifi.lookup.ReaderLookup
org.apache.nifi.csv.CSVReader
org.apache.nifi.csv.CSVRecordSetWriter
org.apache.nifi.cef.CEFReader
org.apache.nifi.grok.GrokReader
org.apache.nifi.text.FreeFormTextRecordSetWriter

199
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/resources/docs/org.apache.nifi.cef.CEFReader/additionalDetails.html Normal file
View File

@ -0,0 +1,199 @@
<!DOCTYPE html>
<html lang="en">
<!--
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.
-->
<head>
<meta charset="utf-8"/>
<title>CEFReader</title>
<link rel="stylesheet" href="../../../../../css/component-usage.css" type="text/css"/>
</head>
<body>
<p>
The CEFReader Controller Service serves as a mean to read and interpret CEF messages. Contrary to the ParseCEF Processor the
CEFReader is intended to let the users access the whole range of record processing tools of NiFi, making it easy to work with a
wide amount of incoming CEF messages.
</p>
<p>
The reader supports CEF Version 23. The expected format and the extension fields known by the Extension Dictionary are defined by
the description of the ArcSight Common Event Format. The reader allows to work with Syslog prefixes and custom extensions. A couple
of CEF message examples the reader can work with:
</p>
<code><pre>
CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789 loginsequence=123
</pre></code>
<h3>Raw message</h3>
<p>
It is possible to preserve the original message in the produced record. This comes in handy when the message contains a Syslog
prefix which is not part of the Record instance. In order to preserve the raw message, the "Raw Message Field" property must be set.
The reader will use the value of this property as field name and will add the raw message as custom extension field. The value of the
"Raw Message Field" must differ from the header fields and the extension fields known by the CEF Extension Dictionary. If the property is
empty, the raw message will not be added.
</p>
<p>
When using predefined schema, the field defined by the "Raw Message Field" must appear in it as a STRING record field. In case of the
schema is inferred, the field will be automatically added as an additional custom extension, regardless of the Inference Strategy.
</p>
<h2>Schemas and Type Coercion</h2>
<p>
When a record is parsed from incoming data, it is separated into fields. Each of these fields is then looked up against the
configured schema (by field name) in order to determine what the type of the data should be. If the field is not present in
the schema, that field is omitted from the Record. If the field is found in the schema, the data type of the received data
is compared against the data type specified in the schema. If the types match, the value of that field is used as-is. If the
schema indicates that the field should be of a different type, then the Controller Service will attempt to coerce the data
into the type specified by the schema. If the field cannot be coerced into the specified type, an Exception will be thrown.
</p>
<p>
The following rules apply when attempting to coerce a field value from one data type to another:
</p>
<ul>
<li>Any data type can be coerced into a String type.</li>
<li>Any numeric data type (Byte, Short, Int, Long, Float, Double) can be coerced into any other numeric data type.</li>
<li>Any numeric value can be coerced into a Date, Time, or Timestamp type, by assuming that the Long value is the number of
milliseconds since epoch (Midnight GMT, January 1, 1970).</li>
<li>A String value can be coerced into a Date, Time, or Timestamp type, if its format matches the configured "Date Format," "Time Format,"
or "Timestamp Format."</li>
<li>A String value can be coerced into a numeric value if the value is of the appropriate type. For example, the String value
<code>8</code> can be coerced into any numeric type. However, the String value <code>8.2</code> can be coerced into a Double or Float
type but not an Integer.</li>
<li>A String value of "true" or "false" (regardless of case) can be coerced into a Boolean value.</li>
<li>A String value that is not empty can be coerced into a Char type. If the String contains more than 1 character, the first character is used
and the rest of the characters are ignored.</li>
<li>Any "date/time" type (Date, Time, Timestamp) can be coerced into any other "date/time" type.</li>
<li>Any "date/time" type can be coerced into a Long type, representing the number of milliseconds since epoch (Midnight GMT, January 1, 1970).</li>
<li>Any "date/time" type can be coerced into a String. The format of the String is whatever DateFormat is configured for the corresponding
property (Date Format, Time Format, Timestamp Format property).</li>
</ul>
<p>
If none of the above rules apply when attempting to coerce a value from one data type to another, the coercion will fail and an Exception
will be thrown.
</p>
<h2>Schema inference</h2>
<p>
While NiFi's Record API does require that each Record have a schema, it is often convenient to infer the schema based on the values in the data,
rather than having to manually create a schema. This is accomplished by selecting a value of "Infer Schema" for the "Schema Access Strategy" property.
When using this strategy, the Reader will determine the schema by first parsing all data in the FlowFile, keeping track of all fields that it has encountered
and the type of each field. Once all data has been parsed, a schema is formed that encompasses all fields that have been encountered.
</p>
<p>
A common concern when inferring schemas is how to handle the condition of two values that have different types. For example, a custom extension field might
have a Float value in one record and String in an other. In these cases, the inferred will contain a CHOICE data type with FLOAT and STRING options. Records will
be allowed to have either value for the particular field.
</p>
<p>
CEF format comes with specification not only to the message format but also has directives for the content. Because of this, the data type of some
fields are not determined by the actual value(s) in the FlowFile but by the CEF format. This includes header fields, which always have to appear and
comply to the data types defined in the CEF format. Also, extension fields from the Extension Dictionary might or might not appear in the generated
schema based on the FlowFile content but in case an extension field is added it's data type is bound by the CEF format. Custom extensions have no similar
restrictions, their presence in the schema is completely depending on the FlowFile content.
</p>
<p>
Schema inference in CEFReader supports multiple possible strategies for convenience. These strategies determine which fields should be included to the schema
from the incoming CEF messages. With this, one might filter out custom extensions for example. It is important to mention that this will have serious effect on
every further steps in the record procession. For example using an Inference Strategy which omits fields together with ConvertRecord Processor will result Records
with only the part of the original fields.
</p>
<h3>Headers only</h3>
<p>
Using this strategy will result a schema which contains only the header fields from the incoming message. All other fields (standard of custom
extensions) will be ignored. The type of these fields are defined by the CEF format and regardless of the content of the message used as a template,
their data type is also defined by the format.
</p>
<h3>Headers and extensions</h3>
<p>
Additionally to the header fields, this strategy will include standard extensions from the messages in the FlowFile. This means, not all
standard extensions will be part of the outgoing Record but the ones the Schema Inference found in the incoming messages. The data type of these Record
fields are determined by the CEF format, ignoring the actual value in the observed field.
</p>
<h3>With custom extensions inferred</h3>
<p>
While the type of the header and standard extension fields are bound by the CEF format, it is possible to add further fields to the message called
"custom extensions". These fields are not part of the "Extension Dictionary", thus their data type is not predefined. Using "With custom extensions inferred"
Inference Strategy, the CEFReader tries to determine the possible data type for these custom extension fields based on their value.
</p>
<h3>With custom extensions as strings</h3>
<p>
In some cases it is undesirable to let the Reader determine the type of the custom extensions. For convenience CEFReader provides an Inference Strategy which
regardless of their value, consider custom extension fields as String data. Otherwise this strategy behaves like the "With custom extensions inferred".
</p>
<h2>Caching of Inferred Schemas</h2>
<p>
This Record Reader requires that if a schema is to be inferred, that all records be read in order to ensure that the schema that gets inferred is applicable for all
records in the FlowFile. However, this can become expensive, especially if the data undergoes many different transformations. To alleviate the cost of inferring schemas,
the Record Reader can be configured with a "Schema Inference Cache" by populating the property with that name. This is a Controller Service that can be shared by Record
Readers and Record Writers.
</p>
<p>
Whenever a Record Writer is used to write data, if it is configured with a "Schema Cache," it will also add the schema to the Schema Cache. This will result in an
identifier for that schema being added as an attribute to the FlowFile.
</p>
<p>
Whenever a Record Reader is used to read data, if it is configured with a "Schema Inference Cache", it will first look for a "schema.cache.identifier" attribute on the FlowFile.
If the attribute exists, it will use the value of that attribute to lookup the schema in the schema cache. If it is able to find a schema in the cache with that identifier,
then it will use that schema instead of reading, parsing, and analyzing the data to infer the schema. If the attribute is not available on the FlowFile, or if the attribute is
available but the cache does not have a schema with that identifier, then the Record Reader will proceed to infer the schema as described above.
</p>
<p>
The end result is that users are able to chain together many different Processors to operate on Record-oriented data. Typically, only the first such Processor in the chain will
incur the "penalty" of inferring the schema. For all other Processors in the chain, the Record Reader is able to simply lookup the schema in the Schema Cache by identifier.
This allows the Record Reader to infer a schema accurately, since it is inferred based on all data in the FlowFile, and still allows this to happen efficiently since the schema
will typically only be inferred once, regardless of how many Processors handle the data.
</p>
<h2>Handling invalid events</h2>
<p>
An event is considered invalid if it is malformed in a way that the underlying CEF parser cannot read it properly. CEFReader has two ways to deal with malformed events determined by
the usage of the property "Invalid Field". If the property is not set, the reading will fail at the time of reading the first invalid event. If the property is set, a product of the read
will be a record with single field. The field is named based on the property and the value of the field will be the original event text. By default, the "Invalid Field" property is not set.
</p>
<p>
When the "Invalid Field" property is set, the read records might contain both records representing well formed CEF events and malformed ones as well. As of this, further steps might be needed
in order to separate these before further processing.
</p>
</body>
</html>

335
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/cef/TestCEFReader.java Normal file
View File

@ -0,0 +1,335 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.schema.access.SchemaAccessUtils;
import org.apache.nifi.schema.inference.SchemaInferenceUtil;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.RecordReaderFactory;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.MockSchemaRegistry;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
public class TestCEFReader {
private TestRunner runner;
private TestCEFProcessor processor;
private CEFReader reader;
@BeforeEach
public void setUp() {
runner = null;
processor = null;
reader = null;
}
@Test
public void testValidatingReaderWhenRawFieldValueIsInvalid() throws Exception {
setUpReader();
setRawField("dst"); // Invalid because there is an extension with the same name
setSchemaIsInferred(CEFReader.HEADERS_ONLY);
assertReaderIsInvalid();
}
@Test
public void testReadingSingleRowWithHeaderFields() throws Exception {
setUpReader();
setSchemaIsInferred(CEFReader.HEADERS_ONLY);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
assertNumberOfResults(1);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingSingleRowWithHeaderFieldsAndRaw() throws Exception {
setUpReader();
setRawField(TestCEFUtil.RAW_FIELD);
setSchemaIsInferred(CEFReader.HEADERS_ONLY);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
assertNumberOfResults(1);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, Collections.singletonMap(TestCEFUtil.RAW_FIELD, TestCEFUtil.RAW_VALUE));
}
@Test
public void testReadingSingleRowWithExtensionFieldsWhenSchemaIsHeadersOnly() throws Exception {
setUpReader();
setSchemaIsInferred(CEFReader.HEADERS_ONLY);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
assertNumberOfResults(1);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingSingleRowWithCustomExtensionFieldsAsStrings() throws Exception {
setUpReader();
setSchemaIsInferred(CEFReader.CUSTOM_EXTENSIONS_AS_STRINGS);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
assertNumberOfResults(1);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES, Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, "123"));
}
@Test
public void testMisformattedRowsWithoutInvalidFieldIsBeingSet() throws Exception {
setUpReader();
setSchemaIsInferred(CEFReader.CUSTOM_EXTENSIONS_INFERRED);
enableReader();
triggerProcessorWithError(TestCEFUtil.INPUT_MISFORMATTED_ROW);
}
@Test
public void testMisformattedRowsWithInvalidFieldIsSet() throws Exception {
setUpReader();
setInvalidField();
setSchemaIsInferred(CEFReader.CUSTOM_EXTENSIONS_INFERRED);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_MISFORMATTED_ROW);
assertNumberOfResults(3);
assertFieldIsSet(1, "invalid", "Oct 12 04:16:11 localhost CEF:0|nxlog.org|nxlog|2.7.1243|");
}
@Test
public void testReadingSingleRowWithCustomExtensionFields() throws Exception {
setUpReader();
setSchemaIsInferred(CEFReader.CUSTOM_EXTENSIONS_INFERRED);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
assertNumberOfResults(1);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES, Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, 123));
}
@Test
public void testReadingMultipleRows() throws Exception {
setUpReader();
setSchemaIsInferred(CEFReader.HEADERS_ONLY);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_MULTIPLE_IDENTICAL_ROWS);
assertNumberOfResults(3);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingMultipleRowsWithEmptyRows() throws Exception {
setUpReader();
setSchemaIsInferred(CEFReader.HEADERS_ONLY);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_MULTIPLE_ROWS_WITH_EMPTY_ROWS);
assertNumberOfResults(3);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingMultipleRowsStartingWithEmptyRow() throws Exception {
setUpReader();
setSchemaIsInferred(CEFReader.HEADERS_ONLY);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_MULTIPLE_ROWS_STARTING_WITH_EMPTY_ROW);
assertNumberOfResults(3);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testWithPredefinedSchema() throws Exception {
setUpReader();
setSchema(CEFSchemaUtil.getHeaderFields());
enableReader();
triggerProcessor(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
assertNumberOfResults(1);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingSingleRowWithEmptyExtensionFields() throws Exception {
setUpReader();
setAcceptEmptyExtensions();
setSchemaIsInferred(CEFReader.CUSTOM_EXTENSIONS_INFERRED);
enableReader();
triggerProcessor(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EMPTY_CUSTOM_EXTENSIONS);
assertNumberOfResults(1);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES, Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, ""));
}
private void setUpReader() throws InitializationException {
processor = new TestCEFProcessor();
runner = TestRunners.newTestRunner(processor);
reader = new CEFReader();
runner.addControllerService("reader", reader);
runner.setProperty(TestCEFProcessor.READER, "reader");
}
private void setAcceptEmptyExtensions() {
runner.setProperty(reader, CEFReader.ACCEPT_EMPTY_EXTENSIONS, "true");
}
private void setSchemaIsInferred(final AllowableValue value) {
runner.setProperty(reader, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY, SchemaInferenceUtil.INFER_SCHEMA);
runner.setProperty(reader, CEFReader.INFERENCE_STRATEGY, value);
}
private void setInvalidField() {
runner.setProperty(reader, CEFReader.INVALID_FIELD, "invalid");
}
private void setSchema(List<RecordField> fields) throws InitializationException {
final MockSchemaRegistry registry = new MockSchemaRegistry();
registry.addSchema("predefinedSchema", new SimpleRecordSchema(fields));
runner.addControllerService("registry", registry);
runner.setProperty(reader, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY, SchemaAccessUtils.SCHEMA_NAME_PROPERTY);
runner.setProperty(reader, SchemaAccessUtils.SCHEMA_NAME, "predefinedSchema");
runner.setProperty(reader, SchemaAccessUtils.SCHEMA_REGISTRY, "registry");
runner.enableControllerService(registry);
}
private void setRawField(final String value) {
runner.setProperty(reader, CEFReader.RAW_FIELD, value);
}
private void enableReader() {
runner.assertValid(reader);
runner.enableControllerService(reader);
}
private void triggerProcessor(final String input) throws FileNotFoundException {
runner.enqueue(new FileInputStream(input));
runner.run();
runner.assertAllFlowFilesTransferred(TestCEFProcessor.SUCCESS);
}
private void triggerProcessorWithError(final String input) throws FileNotFoundException {
runner.enqueue(new FileInputStream(input));
final AssertionError exception = Assertions.assertThrows(AssertionError.class, () -> runner.run());
Assertions.assertTrue(exception.getCause() instanceof RuntimeException);
Assertions.assertTrue(exception.getCause().getCause() instanceof IOException);
}
private void assertNumberOfResults(final int numberOfResults) {
Assertions.assertEquals(numberOfResults, processor.getRecords().size());
}
private void assertReaderIsInvalid() {
runner.assertNotValid(reader);
}
private void assertFieldIsSet(final int number, final String name, final String value) {
Assertions.assertEquals(value, processor.getRecords().get(number).getValue(name));
}
@SafeVarargs
private void assertFieldsAre(final Map<String, Object>... fieldGroups) {
final Map<String, Object> expectedFields = new HashMap<>();
Arrays.stream(fieldGroups).forEach(fieldGroup -> expectedFields.putAll(fieldGroup));
processor.getRecords().forEach(r -> TestCEFUtil.assertFieldsAre(r, expectedFields));
}
private static class TestCEFProcessor extends AbstractProcessor {
private final List<Record> records = new ArrayList<>();
static final Relationship SUCCESS = new Relationship.Builder().name("success").build();
static final PropertyDescriptor READER = new PropertyDescriptor.Builder()
.name("reader")
.identifiesControllerService(CEFReader.class)
.build();
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
final FlowFile flowFile = session.get();
final RecordReaderFactory readerFactory = context.getProperty(READER).asControllerService(RecordReaderFactory.class);
try (
final InputStream in = session.read(flowFile);
final RecordReader reader = readerFactory.createRecordReader(flowFile, in, getLogger())
) {
Record record;
while ((record = reader.nextRecord()) != null) {
records.add(record);
}
} catch (final Exception e) {
throw new RuntimeException(e);
}
session.transfer(flowFile, SUCCESS);
}
@Override
public Set<Relationship> getRelationships() {
return Collections.singleton(SUCCESS);
}
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return Collections.singletonList(READER);
}
public List<Record> getRecords() {
return records;
}
}
}

497
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/cef/TestCEFRecordReader.java Normal file
View File

@ -0,0 +1,497 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import com.fluenda.parcefone.parser.CEFParser;
import org.apache.bval.jsr.ApacheValidationProvider;
import org.apache.nifi.mock.MockComponentLogger;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import javax.validation.Validation;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
public class TestCEFRecordReader {
private final javax.validation.Validator validator = Validation.byProvider(ApacheValidationProvider.class).configure().buildValidatorFactory().getValidator();
private final CEFParser parser = new CEFParser(validator);
private RecordSchema schema;
private InputStream inputStream;
private CEFRecordReader testSubject;
private List<Record> results;
private boolean includeCustomExtensions;
private boolean acceptEmptyExtensions;
private boolean dropUnknownFields;
private String rawField;
private String invalidField;
@BeforeEach
public void setUp() {
schema = null;
inputStream = null;
testSubject = null;
results = null;
includeCustomExtensions = false;
acceptEmptyExtensions = false;
dropUnknownFields = true;
rawField = null;
invalidField = null;
}
@AfterEach
public void tearDown() throws IOException {
if (inputStream != null) {
inputStream.close();
}
}
@Test
public void testReadingSingleLineWithHeaderFieldsUsingSchemaWithHeaderFields() throws Exception {
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingSingleLineWithExtensionFieldsUsingSchemaWithHeaderFields() throws Exception {
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingSingleLineWithCustomExtensionFieldsUsingSchemaWithHeaderFields() throws Exception {
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingSingleLineWithHeaderFieldsUsingSchemaWithExtensionFields() throws Exception {
setSchema(TestCEFUtil.getFieldWithExtensions());
setReader(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, withNulls(TestCEFUtil.EXPECTED_EXTENSION_VALUES));
}
@Test
public void testReadingSingleLineWithExtensionFieldsUsingSchemaWithExtensionFields() throws Exception {
setSchema(TestCEFUtil.getFieldWithExtensions());
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES);
}
@Test
public void testReadingSingleLineWithCustomExtensionFieldsUsingSchemaWithExtensionFields() throws Exception {
setSchema(TestCEFUtil.getFieldWithExtensions());
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES);
}
@Test
public void testReadingSingleLineWithHeaderFieldsUsingSchemaWithCustomExtensionFieldsAsStrings() throws Exception {
setSchema(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD_AS_STRING));
setReader(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, withNulls(TestCEFUtil.EXPECTED_EXTENSION_VALUES), Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, null));
}
@Test
public void testReadingSingleLineWithExtensionFieldsUsingSchemaWithCustomExtensionFieldsAsStrings() throws Exception {
setSchema(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD_AS_STRING));
setIncludeCustomExtensions();
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES, Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, null));
}
@Test
public void testReadingSingleLineWithCustomExtensionFieldsUsingSchemaWithCustomExtensionFieldsAsStrings() throws Exception {
setSchema(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD_AS_STRING));
setIncludeCustomExtensions();
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES, Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, "123"));
}
@Test
public void testReadingSingleLineWithHeaderFieldsUsingSchemaWithCustomExtensionFields() throws Exception {
setSchema(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD));
setIncludeCustomExtensions();
setReader(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, withNulls(TestCEFUtil.EXPECTED_EXTENSION_VALUES), Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, null));
}
@Test
public void testReadingSingleLineWithExtensionFieldsUsingSchemaWithCustomExtensionFields() throws Exception {
setSchema(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD));
setIncludeCustomExtensions();
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES, Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, null));
}
@Test
public void testReadingSingleLineWithCustomExtensionFieldsUsingSchemaWithCustomExtensionFields() throws Exception {
setSchema(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD));
setIncludeCustomExtensions();
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES, Collections.singletonMap(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, 123));
}
@Test
public void testReadingContainingRawEvent() throws Exception {
setSchema(getFieldsWithHeaderAndRaw());
setRawMessageField(TestCEFUtil.RAW_FIELD);
setReader(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
readRecords();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, Collections.singletonMap(TestCEFUtil.RAW_FIELD, TestCEFUtil.RAW_VALUE));
}
@Test
public void testSingleLineWithHeaderFieldsAndNotDroppingUnknownFields() throws Exception {
setKeepUnknownFields();
setSchema(getFieldsWithExtensionsExpect("c6a1", "dmac"));
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
readRecords();
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, without(TestCEFUtil.EXPECTED_EXTENSION_VALUES, "c6a1", "dmac"));
assertFieldIs("c6a1", TestCEFUtil.EXPECTED_EXTENSION_VALUES.get("c6a1"));
assertFieldIs("dmac", TestCEFUtil.EXPECTED_EXTENSION_VALUES.get("dmac"));
}
@Test
public void testReadingEmptyRow() throws Exception {
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_EMPTY_ROW);
readRecords();
assertNumberOfResults(0);
}
@Test
public void testReadingMisformattedRow() throws Exception {
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_MISFORMATTED_ROW);
Assertions.assertThrows(MalformedRecordException.class, this::readRecords);
}
@Test
public void testReadingMisformattedRowWhenInvalidFieldIsSet() throws Exception {
setInvalidField();
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_MISFORMATTED_ROW);
readRecords();
assertNumberOfResults(3);
}
@Test
public void testReadingMultipleIdenticalRows() throws Exception {
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_MULTIPLE_IDENTICAL_ROWS);
readRecords();
assertNumberOfResults(3);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingMultipleIdenticalRowsWithEmptyOnes() throws Exception {
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_MULTIPLE_ROWS_WITH_EMPTY_ROWS);
readRecords(5);
assertNumberOfResults(3);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES);
}
@Test
public void testReadingMultipleRowsWithDecreasingNumberOfExtensionFields() throws Exception {
setSchema(TestCEFUtil.getFieldWithExtensions());
setReader(TestCEFUtil.INPUT_MULTIPLE_ROWS_WITH_DECREASING_NUMBER_OF_EXTENSIONS);
readRecords();
assertNumberOfResults(2);
assertFieldsAre(0, TestCEFUtil.EXPECTED_HEADER_VALUES, TestCEFUtil.EXPECTED_EXTENSION_VALUES);
assertFieldsAre(1, TestCEFUtil.EXPECTED_HEADER_VALUES, withNulls(TestCEFUtil.EXPECTED_EXTENSION_VALUES, "c6a1", "dmac"));
}
@Test
public void testReadingMultipleRowsWithIncreasingNumberOfExtensionFields() throws Exception {
setSchema(getFieldsWithExtensionsExpect("c6a1", "dmac"));
setReader(TestCEFUtil.INPUT_MULTIPLE_ROWS_WITH_INCREASING_NUMBER_OF_EXTENSIONS);
readRecords();
assertNumberOfResults(2);
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, without(TestCEFUtil.EXPECTED_EXTENSION_VALUES, "c6a1", "dmac"));
}
@Test
public void testReadingIncorrectHeaderField() throws Exception {
setSchema(CEFSchemaUtil.getHeaderFields());
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_INCORRECT_HEADER_FIELD);
Assertions.assertThrows(NumberFormatException.class, () -> readRecords());
}
@Test
public void testReadingIncorrectCustomVariableFormat() throws Exception {
setSchema(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD));
setIncludeCustomExtensions();
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_INCORRECT_CUSTOM_EXTENSIONS);
Assertions.assertThrows(NumberFormatException.class, () -> readRecords());
}
@Test
public void testReadingEmptyValuesWhenAcceptingEmptyExtensions() throws Exception {
setSchema(TestCEFUtil.getFieldWithExtensions());
setAcceptEmptyExtensions();
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EMPTY_EXTENSION);
readRecords();
assertNumberOfResults(1);
assertSchemaIsSet();
final Map<String, Object> expectedExtensionValues = new HashMap<>();
expectedExtensionValues.putAll(TestCEFUtil.EXPECTED_EXTENSION_VALUES);
expectedExtensionValues.put("cn1", null);
expectedExtensionValues.put("cn1Label", "");
assertFieldsAre(TestCEFUtil.EXPECTED_HEADER_VALUES, expectedExtensionValues);
}
@Test
public void testReadingEmptyValuesWhenNotAcceptingEmptyExtensions() throws Exception {
setSchema(TestCEFUtil.getFieldWithExtensions());
setReader(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EMPTY_EXTENSION);
// As empty is not accepted, number type fields will not be parsed properly
Assertions.assertThrows(NumberFormatException.class, () -> readRecords());
}
private static List<RecordField> getFieldsWithHeaderAndRaw() {
final List<RecordField> result = new ArrayList<>(CEFSchemaUtil.getHeaderFields());
result.add(new RecordField(TestCEFUtil.RAW_FIELD, RecordFieldType.STRING.getDataType()));
return result;
}
private List<RecordField> getFieldsWithExtensionsExpect(final String... fieldsToExclude) {
final List<RecordField> result = new ArrayList<>();
final Set<String> excludedFields = new HashSet<>();
Arrays.stream(fieldsToExclude).forEach(f -> excludedFields.add(f));
for (final RecordField record : TestCEFUtil.getFieldWithExtensions()) {
if (!excludedFields.contains(record.getFieldName())) {
result.add(record);
}
}
return result;
}
private void setSchema(final List<RecordField> fields) {
this.schema = new SimpleRecordSchema(fields);
}
private void setIncludeCustomExtensions() {
includeCustomExtensions = true;
}
private void setAcceptEmptyExtensions() {
acceptEmptyExtensions = true;
}
private void setRawMessageField(final String rawField) {
this.rawField = rawField;
}
private void setKeepUnknownFields() {
dropUnknownFields = false;
}
private void setInvalidField() {
invalidField = "invalid";
}
private void setReader(final String file) throws FileNotFoundException {
inputStream = new FileInputStream(file);
testSubject = new CEFRecordReader(inputStream, schema, parser, new MockComponentLogger(), Locale.US, rawField, invalidField, includeCustomExtensions, acceptEmptyExtensions);
}
private void readRecords() throws IOException, MalformedRecordException {
final List<Record> results = new ArrayList<>();
Record result;
while ((result = testSubject.nextRecord(true, dropUnknownFields)) != null) {
results.add(result);
}
this.results = results;
}
private void readRecords(final int count) throws IOException, MalformedRecordException {
final List<Record> results = new ArrayList<>();
for (int i=1; i<=count; i++) {
final Record record = testSubject.nextRecord(true, dropUnknownFields);
if (record != null) {
results.add(record);
}
}
this.results = results;
}
@SafeVarargs
private void assertFieldsAre(final Map<String, Object>... fieldGroups) {
final Map<String, Object> expectedFields = new HashMap<>();
Arrays.stream(fieldGroups).forEach(fieldGroup -> expectedFields.putAll(fieldGroup));
results.forEach(r -> TestCEFUtil.assertFieldsAre(r, expectedFields));
}
private void assertFieldsAre(final int number, final Map<String, Object>... fieldGroups) {
final Map<String, Object> expectedFields = new HashMap<>();
Arrays.stream(fieldGroups).forEach(fieldGroup -> expectedFields.putAll(fieldGroup));
TestCEFUtil.assertFieldsAre(results.get(number), expectedFields);
}
private void assertFieldIs(final String field, final Object expectedValue) {
results.forEach(r -> Assertions.assertEquals(expectedValue, r.getValue(field)));
}
private void assertNumberOfResults(final int numberOfResults) {
Assertions.assertEquals(numberOfResults, results.size());
}
private void assertSchemaIsSet() {
results.forEach(r -> Assertions.assertEquals(schema, r.getSchema()));
}
private static Map<String, Object> withNulls(final Map<String, Object> expectedFields) {
final Map<String, Object> result = new HashMap<>();
expectedFields.keySet().forEach(f -> result.put(f, null));
return result;
}
private static Map<String, Object> withNulls(final Map<String, Object> expectedFields, final String... fieldsWithNullValue) {
final Map<String, Object> result = new HashMap<>(expectedFields);
Arrays.stream(fieldsWithNullValue).forEach(field -> result.put(field, null));
return result;
}
private static Map<String, Object> without(final Map<String, Object> expectedFields, final String... fieldsToRemove) {
final Map<String, Object> result = new HashMap<>();
final Set<String> fieldsToNotInclude = new HashSet<>(Arrays.asList(fieldsToRemove));
for (final Map.Entry<String, Object> field : expectedFields.entrySet()) {
if (!fieldsToNotInclude.contains(field.getKey())) {
result.put(field.getKey(), field.getValue());
}
}
return result;
}
}

286
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/cef/TestCEFSchemaInference.java Normal file
View File

@ -0,0 +1,286 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import com.fluenda.parcefone.event.CommonEvent;
import com.fluenda.parcefone.parser.CEFParser;
import org.apache.bval.jsr.ApacheValidationProvider;
import org.apache.nifi.schema.inference.RecordSource;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.apache.nifi.serialization.record.RecordSchema;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import javax.validation.Validation;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.Optional;
import java.util.stream.Collectors;
public class TestCEFSchemaInference {
private final javax.validation.Validator validator = Validation.byProvider(ApacheValidationProvider.class).configure().buildValidatorFactory().getValidator();
private final CEFParser parser = new CEFParser(validator);
private boolean includeExtensions;
private boolean includeCustomExtensions;
private String rawMessageField;
private CEFSchemaInference testSubject;
private RecordSource<CommonEvent> recordSource;
private CEFCustomExtensionTypeResolver typeResolver;
private RecordSchema result;
@BeforeEach
public void setUp() {
includeExtensions = false;
includeCustomExtensions = false;
rawMessageField = null;
testSubject = null;
recordSource = null;
result = null;
}
@Test
public void testInferBasedOnHeaderFields() throws Exception {
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(CEFSchemaUtil.getHeaderFields());
}
@Test
public void testInferBasedOnHeaderFieldsWhenIncludingExtensions() throws Exception {
setIncludeExtensions();
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(CEFSchemaUtil.getHeaderFields());
}
@Test
public void testInferBasedOnHeaderFieldsWithRaw() throws Exception {
setRawMessageField("raw");
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(CEFSchemaUtil.getHeaderFields(), Collections.singletonList(new RecordField("raw", RecordFieldType.STRING.getDataType())));
}
@Test
public void testInferBasedOnHeaderAndExtensionFieldsWhenNotIncludingExtensions() throws Exception {
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(CEFSchemaUtil.getHeaderFields());
}
@Test
public void testInferBasedOnHeaderAndExtensionFieldsWhenIncludingExtensions() throws Exception {
setIncludeExtensions();
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EXTENSIONS);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(TestCEFUtil.getFieldWithExtensions());
}
@Test
public void testInferBasedOnRowWithCustomExtensionsButSkipping() throws Exception {
setIncludeExtensions();
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(TestCEFUtil.getFieldWithExtensions());
}
@Test
public void testInferBasedOnRowWithCustomExtensionsAsString() throws Exception {
setIncludeExtensions();
setIncludeCustomExtensions(CEFCustomExtensionTypeResolver.STRING_RESOLVER);
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD_AS_STRING));
}
@Test
public void testInferBasedOnRowWithCustomExtensions() throws Exception {
setIncludeExtensions();
setIncludeCustomExtensions(CEFCustomExtensionTypeResolver.SIMPLE_RESOLVER);
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD));
}
@Test
public void testInferBasedOnRowWithEmptyExtensions() throws Exception {
setIncludeExtensions();
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EMPTY_EXTENSION);
setUpTestSubject();
inferSchema();
// As extension types are defined by the extension dictionary, even with empty value the data type can be inferred
assertSchemaConsistsOf(TestCEFUtil.getFieldWithExtensions());
}
@Test
public void testInferBasedOnRowWithEmptyCustomExtensions() throws Exception {
setIncludeExtensions();
setIncludeCustomExtensions(CEFCustomExtensionTypeResolver.SIMPLE_RESOLVER);
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_WITH_EMPTY_CUSTOM_EXTENSIONS);
setUpTestSubject();
inferSchema();
// As there is no value provided to infer based on, the empty custom field will be considered as string
assertSchemaConsistsOf(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD_AS_STRING));
}
@Test
public void testInferWhenStartingWithEmptyRow() throws Exception {
setRecordSource(TestCEFUtil.INPUT_MULTIPLE_ROWS_STARTING_WITH_EMPTY_ROW);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(CEFSchemaUtil.getHeaderFields());
}
@Test
public void testInferWithDifferentCustomFieldTypes() throws Exception {
setIncludeExtensions();
setIncludeCustomExtensions(CEFCustomExtensionTypeResolver.SIMPLE_RESOLVER);
setRecordSource(TestCEFUtil.INPUT_MULTIPLE_ROWS_WITH_DIFFERENT_CUSTOM_TYPES);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(TestCEFUtil.getFieldsWithCustomExtensions(TestCEFUtil.CUSTOM_EXTENSION_FIELD_AS_CHOICE));
}
@Test
public void testInferBasedOnEmptyRow() throws Exception{
setRecordSource(TestCEFUtil.INPUT_EMPTY_ROW);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(CEFSchemaUtil.getHeaderFields()); // For CEF it is always expected to have a header
}
@Test
public void testInferBasedOnMisformattedRow() throws Exception{
setRecordSource(TestCEFUtil.INPUT_MISFORMATTED_ROW);
setUpTestSubject();
Assertions.assertThrows(IOException.class, this::inferSchema);
}
@Test
public void testInferBasedOnMisformattedRowWhenNonFailFast() throws Exception{
setRecordSourceWhenNonFailFast(TestCEFUtil.INPUT_MISFORMATTED_ROW);
setUpTestSubject();
inferSchema();
assertSchemaConsistsOf(CEFSchemaUtil.getHeaderFields());
}
@Test
public void testInferBasedOnHeaderFieldsWithInvalid() throws Exception {
setRecordSource(TestCEFUtil.INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY);
setUpTestSubjectWithInvalidField();
inferSchema();
assertSchemaConsistsOf(CEFSchemaUtil.getHeaderFields(), Collections.singletonList(new RecordField("invalid", RecordFieldType.STRING.getDataType())));
}
private void setIncludeExtensions() {
this.includeExtensions = true;
}
private void setIncludeCustomExtensions(final CEFCustomExtensionTypeResolver typeResolver) {
this.includeCustomExtensions = true;
this.typeResolver = typeResolver;
}
private void setUpTestSubject() {
this.testSubject = new CEFSchemaInference(includeExtensions, includeCustomExtensions, typeResolver, rawMessageField, null);
}
private void setUpTestSubjectWithInvalidField() {
this.testSubject = new CEFSchemaInference(includeExtensions, includeCustomExtensions, typeResolver, rawMessageField, "invalid");
}
private void setRawMessageField(final String rawMessageField) {
this.rawMessageField = rawMessageField;
}
private void setRecordSource(final String inputFile) throws FileNotFoundException {
setRecordSource(inputFile, true);
}
private void setRecordSource(final String inputFile, final boolean failFast) throws FileNotFoundException {
final FileInputStream inputStream = new FileInputStream(inputFile);
this.recordSource = new CEFRecordSource(inputStream, parser, Locale.US, true, failFast);
}
private void setRecordSourceWhenNonFailFast(final String inputFile) throws FileNotFoundException {
setRecordSource(inputFile, false);
}
private void inferSchema() throws IOException {
result = testSubject.inferSchema(recordSource);
}
@SafeVarargs
private void assertSchemaConsistsOf(final List<RecordField>... expectedFieldGroups) {
final List<RecordField> expectedFields = Arrays.stream(expectedFieldGroups).flatMap(group -> group.stream()).collect(Collectors.toList());
Assertions.assertEquals(expectedFields.size(), result.getFieldCount());
for (final RecordField expectedField : expectedFields) {
final Optional<RecordField> field = result.getField(expectedField.getFieldName());
Assertions.assertTrue(field.isPresent());
Assertions.assertEquals(expectedField, field.get());
}
}
}

128
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/cef/TestCEFUtil.java Normal file
View File

@ -0,0 +1,128 @@
/*
* 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.
*/
package org.apache.nifi.cef;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.junit.Assert;
import java.sql.Timestamp;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
public class TestCEFUtil {
static final String RAW_FIELD = "raw";
static final String RAW_VALUE = "Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|";
static final String INPUT_SINGLE_ROW_HEADER_FIELDS_ONLY = "src/test/resources/cef/single-row-header-fields-only.txt";
static final String INPUT_SINGLE_ROW_WITH_EXTENSIONS = "src/test/resources/cef/single-row-with-extensions.txt";
static final String INPUT_SINGLE_ROW_WITH_EMPTY_EXTENSION = "src/test/resources/cef/single-row-with-empty-extension.txt";
static final String INPUT_SINGLE_ROW_WITH_CUSTOM_EXTENSIONS = "src/test/resources/cef/single-row-with-custom-extensions.txt";
static final String INPUT_SINGLE_ROW_WITH_EMPTY_CUSTOM_EXTENSIONS = "src/test/resources/cef/single-row-with-empty-custom-extensions.txt";
static final String INPUT_SINGLE_ROW_WITH_INCORRECT_HEADER_FIELD = "src/test/resources/cef/single-row-with-incorrect-header-field.txt";
static final String INPUT_SINGLE_ROW_WITH_INCORRECT_CUSTOM_EXTENSIONS = "src/test/resources/cef/single-row-with-incorrect-custom-extensions.txt";
static final String INPUT_EMPTY_ROW = "src/test/resources/cef/empty-row.txt";
static final String INPUT_MISFORMATTED_ROW = "src/test/resources/cef/misformatted-row.txt";
static final String INPUT_MULTIPLE_IDENTICAL_ROWS = "src/test/resources/cef/multiple-rows.txt";
static final String INPUT_MULTIPLE_ROWS_WITH_DIFFERENT_CUSTOM_TYPES = "src/test/resources/cef/multiple-rows-with-different-custom-types.txt";
static final String INPUT_MULTIPLE_ROWS_STARTING_WITH_EMPTY_ROW = "src/test/resources/cef/multiple-rows-starting-with-empty-row.txt";
static final String INPUT_MULTIPLE_ROWS_WITH_EMPTY_ROWS = "src/test/resources/cef/multiple-rows-with-empty-rows.txt";
static final String INPUT_MULTIPLE_ROWS_WITH_DECREASING_NUMBER_OF_EXTENSIONS = "src/test/resources/cef/multiple-rows-decreasing-number-of-extensions.txt";
static final String INPUT_MULTIPLE_ROWS_WITH_INCREASING_NUMBER_OF_EXTENSIONS = "src/test/resources/cef/multiple-rows-increasing-number-of-extensions.txt";
static final Map<String, Object> EXPECTED_HEADER_VALUES = new HashMap<>();
static final Map<String, Object> EXPECTED_EXTENSION_VALUES = new HashMap<>();
static final String CUSTOM_EXTENSION_FIELD_NAME = "loginsequence";
static final RecordField CUSTOM_EXTENSION_FIELD = new RecordField(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, RecordFieldType.INT.getDataType());
static final RecordField CUSTOM_EXTENSION_FIELD_AS_STRING = new RecordField(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, RecordFieldType.STRING.getDataType());
static final RecordField CUSTOM_EXTENSION_FIELD_AS_CHOICE = new RecordField(TestCEFUtil.CUSTOM_EXTENSION_FIELD_NAME, RecordFieldType.CHOICE.getChoiceDataType(
RecordFieldType.FLOAT.getDataType(), RecordFieldType.STRING.getDataType()
));
static {
EXPECTED_HEADER_VALUES.put("version", Integer.valueOf(0));
EXPECTED_HEADER_VALUES.put("deviceVendor", "Company");
EXPECTED_HEADER_VALUES.put("deviceProduct", "Product");
EXPECTED_HEADER_VALUES.put("deviceVersion", "1.2.3");
EXPECTED_HEADER_VALUES.put("deviceEventClassId", "audit-login");
EXPECTED_HEADER_VALUES.put("name", "Successful login");
EXPECTED_HEADER_VALUES.put("severity", "3");
EXPECTED_EXTENSION_VALUES.put("cn1Label", "userid");
EXPECTED_EXTENSION_VALUES.put("spt", Integer.valueOf(46117));
EXPECTED_EXTENSION_VALUES.put("cn1", Long.valueOf(99999));
EXPECTED_EXTENSION_VALUES.put("cfp1", Float.valueOf(1.23F));
EXPECTED_EXTENSION_VALUES.put("dst", "127.0.0.1");
EXPECTED_EXTENSION_VALUES.put("c6a1", "2345:425:2ca1:0:0:567:5673:23b5"); // Lower case representation is expected and values with leading zeroes shortened
EXPECTED_EXTENSION_VALUES.put("dmac", "00:0d:60:af:1b:61"); // Lower case representation is expected
EXPECTED_EXTENSION_VALUES.put("start", new Timestamp(1479152665000L));
try {
final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MMM-dd HH:mm:ss.SSS", Locale.US);
EXPECTED_EXTENSION_VALUES.put("end", simpleDateFormat.parse("2017-Jan-12 12:23:45.000"));
} catch (ParseException e) {
throw new RuntimeException(e);
}
EXPECTED_EXTENSION_VALUES.put("dlat", Double.valueOf(456.789D));
}
/**
* The list of custom fields in the schema are expected to be generated based on the first event in the flow file. Because of this, it will not contain all the
* possible extension fields. Also because of this, the following rows are handled based on this schema as well.
*/
static List<RecordField> getFieldWithExtensions() {
final List<RecordField> result = new ArrayList<>(CEFSchemaUtil.getHeaderFields());
result.add(new RecordField("cn1Label", RecordFieldType.STRING.getDataType()));
result.add(new RecordField("spt", RecordFieldType.INT.getDataType()));
result.add(new RecordField("cn1", RecordFieldType.LONG.getDataType()));
result.add(new RecordField("cfp1", RecordFieldType.FLOAT.getDataType()));
result.add(new RecordField("dst", RecordFieldType.STRING.getDataType()));
result.add(new RecordField("c6a1", RecordFieldType.STRING.getDataType()));
result.add(new RecordField("dmac", RecordFieldType.STRING.getDataType()));
result.add(new RecordField("start", RecordFieldType.TIMESTAMP.getDataType()));
result.add(new RecordField("end", RecordFieldType.TIMESTAMP.getDataType()));
result.add(new RecordField("dlat", RecordFieldType.DOUBLE.getDataType()));
return result;
}
static List<RecordField> getFieldsWithCustomExtensions(RecordField... customExtensions) {
final List<RecordField> result = TestCEFUtil.getFieldWithExtensions();
for (final RecordField customExtension : customExtensions) {
result.add(customExtension);
}
return result;
}
/*
* Record internally keeps the unknown fields even if they are marked for dropping. Thus direct comparison will not work as expected
*/
static void assertFieldsAre(final Record record, final Map<String, Object> expectedValues) {
Assert.assertEquals(expectedValues.size(), record.getValues().length);
expectedValues.entrySet().forEach(field -> Assert.assertEquals("Field " + field.getKey() + " is incorrect" , field.getValue(), record.getValue(field.getKey())));
}
}

93
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/util/TestSchemaInferenceUtil.java Normal file
View File

@ -0,0 +1,93 @@
/*
* 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.
*/
package org.apache.nifi.util;
import org.apache.nifi.schema.inference.TimeValueInference;
import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.RecordFieldType;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import java.math.BigDecimal;
import java.util.stream.Stream;
class TestSchemaInferenceUtil {
private static final String DATE_FORMAT = "yyyy-MM-dd";
private static final String TIME_FORMAT = "HH:mm:ss";
private static final String TIMESTAMP_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
private final TimeValueInference timeInference = new TimeValueInference(DATE_FORMAT, TIME_FORMAT, TIMESTAMP_FORMAT);
@ParameterizedTest
@MethodSource("data")
public void testWithoutTimeInference(final String input, final DataType expectedResult) {
Assertions.assertEquals(expectedResult, SchemaInferenceUtil.getDataType(input));
}
@ParameterizedTest
@MethodSource("dataForTimeInference")
public void testWithTimeInference(final String input, final DataType expectedResult) {
Assertions.assertEquals(expectedResult, SchemaInferenceUtil.getDataType(input, timeInference));
}
private static Stream<Arguments> data() {
return Stream.of(
Arguments.of(null, null),
Arguments.of("", null),
Arguments.of("true", RecordFieldType.BOOLEAN.getDataType()),
Arguments.of("TRUE", RecordFieldType.BOOLEAN.getDataType()),
Arguments.of("FALSE", RecordFieldType.BOOLEAN.getDataType()),
Arguments.of("tRUE", RecordFieldType.BOOLEAN.getDataType()),
Arguments.of("fALSE", RecordFieldType.BOOLEAN.getDataType()),
Arguments.of(new BigDecimal(Double.MAX_VALUE - 1).toPlainString() + ".01", RecordFieldType.DOUBLE.getDataType()),
Arguments.of(String.valueOf(1.1D), RecordFieldType.FLOAT.getDataType()),
Arguments.of(String.valueOf(1), RecordFieldType.INT.getDataType()),
Arguments.of(String.valueOf(Long.MAX_VALUE), RecordFieldType.LONG.getDataType()),
Arguments.of("c", RecordFieldType.STRING.getDataType()),
Arguments.of("string", RecordFieldType.STRING.getDataType())
);
}
private static Stream<Arguments> dataForTimeInference() {
return Stream.of(
Arguments.of("2017-03-19", RecordFieldType.DATE.getDataType(DATE_FORMAT)),
Arguments.of("2017-3-19", RecordFieldType.STRING.getDataType()),
Arguments.of("2017-44-55", RecordFieldType.STRING.getDataType()),
Arguments.of("2017.03.19", RecordFieldType.STRING.getDataType()),
Arguments.of("12:13:14", RecordFieldType.TIME.getDataType(TIME_FORMAT)),
Arguments.of("12:13:0", RecordFieldType.TIME.getDataType(TIME_FORMAT)),
Arguments.of("12:3:0", RecordFieldType.TIME.getDataType(TIME_FORMAT)),
Arguments.of("25:13:14", RecordFieldType.STRING.getDataType()),
Arguments.of("25::14", RecordFieldType.STRING.getDataType()),
Arguments.of("2019-01-07T14:59:27.817Z", RecordFieldType.TIMESTAMP.getDataType(TIMESTAMP_FORMAT)),
Arguments.of("2019-01-07T14:59:27.Z", RecordFieldType.STRING.getDataType()),
Arguments.of("2019-01-07T14:59:27Z", RecordFieldType.STRING.getDataType()),
Arguments.of("2019-01-07T14:59:27.817", RecordFieldType.STRING.getDataType()),
Arguments.of("2019-01-07 14:59:27.817Z", RecordFieldType.STRING.getDataType())
);
}
}

0
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/empty-row.txt Normal file
View File

3
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/misformatted-row.txt Normal file
View File

@ -0,0 +1,3 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|nxlog.org|nxlog|2.7.1243|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.4|audit-login|Successful login|1|

2
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/multiple-rows-decreasing-number-of-extensions.txt Normal file
View File

@ -0,0 +1,2 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789

2
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/multiple-rows-increasing-number-of-extensions.txt Normal file
View File

@ -0,0 +1,2 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789

8
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/multiple-rows-starting-with-empty-row.txt Normal file
View File

@ -0,0 +1,8 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|

2
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/multiple-rows-with-different-custom-types.txt Normal file
View File

@ -0,0 +1,2 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=1484220225000 dlat=456.789 loginsequence=123.12
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=1484220225000 dlat=456.789 loginsequence=undetermined

4
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/multiple-rows-with-empty-rows.txt Normal file
View File

@ -0,0 +1,4 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|

3
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/multiple-rows.txt Normal file
View File

@ -0,0 +1,3 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|

1
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/single-row-header-fields-only.txt Normal file
View File

@ -0,0 +1 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|

1
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/single-row-with-custom-extensions.txt Normal file
View File

@ -0,0 +1 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789 loginsequence=123

1
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/single-row-with-empty-custom-extensions.txt Normal file
View File

@ -0,0 +1 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789 loginsequence=

1
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/single-row-with-empty-extension.txt Normal file
View File

@ -0,0 +1 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label= spt=46117 cn1= cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789

1
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/single-row-with-extensions.txt Normal file
View File

@ -0,0 +1 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=Jan 12 2017 12:23:45 dlat=456.789

1
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/single-row-with-incorrect-custom-extensions.txt Normal file
View File

@ -0,0 +1 @@
Oct 12 04:16:11 localhost CEF:0|Company|Product|1.2.3|audit-login|Successful login|3|cn1Label=userid spt=46117 cn1=99999 cfp1=1.23 dst=127.0.0.1 c6a1=2345:0425:2CA1:0000:0000:0567:5673:23b5 dmac=00:0D:60:AF:1B:61 start=1479152665000 end=1484220225000 dlat=456.789 loginsequence=123.12

1
nifi-nar-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/resources/cef/single-row-with-incorrect-header-field.txt Normal file
View File

@ -0,0 +1 @@
Oct 12 04:16:11 localhost CEF:Unknown|Company|Product|1.2.3|audit-login|Successful login|3|