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

NIFI-3008 Enabled expression language on BASE_DIRECTORY and FILENAME properties in TailFile 

This closes #1190
This commit is contained in:
Jeff Storck 2016-11-08 22:02:03 -05:00 committed by Oleg Zhurakousky
parent 3e892c55ec
commit cc2fbcaac4
2 changed files with 32 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java
View File

@ -126,7 +126,7 @@ public class TailFile extends AbstractProcessor {
.name("tail-base-directory")
.displayName("Base directory")
.description("Base directory used to look for files to tail. This property is required when using Multifile mode.")
.expressionLanguageSupported(false)
.expressionLanguageSupported(true)
.addValidator(StandardValidators.FILE_EXISTS_VALIDATOR)
.required(false)
.build();
@ -148,8 +148,8 @@ public class TailFile extends AbstractProcessor {
.description("Path of the file to tail in case of single file mode. If using multifile mode, regular expression to find files "
+ "to tail in the base directory. In case recursivity is set to true, the regular expression will be used to match the "
+ "path starting from the base directory (see additional details for examples).")
.expressionLanguageSupported(false)
.addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
.expressionLanguageSupported(true)
.addValidator(StandardValidators.createRegexValidator(0, Integer.MAX_VALUE, true))
.required(true)
.build();
@ -267,7 +267,7 @@ public class TailFile extends AbstractProcessor {
final List<ValidationResult> results = new ArrayList<>(super.customValidate(context));
if(context.getProperty(MODE).getValue().equals(MODE_MULTIFILE.getValue())) {
String path = context.getProperty(BASE_DIRECTORY).getValue();
String path = context.getProperty(BASE_DIRECTORY).evaluateAttributeExpressions().getValue();
if(path == null) {
results.add(new ValidationResult.Builder().subject(BASE_DIRECTORY.getName()).valid(false)
.explanation("Base directory property cannot be empty in Multifile mode.").build());
@ -291,8 +291,8 @@ public class TailFile extends AbstractProcessor {
}
} else {
long max = context.getProperty(MAXIMUM_AGE).getValue() == null ? Long.MAX_VALUE : context.getProperty(MAXIMUM_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
List<String> filesToTail = getFilesToTail(context.getProperty(BASE_DIRECTORY).getValue(),
context.getProperty(FILENAME).getValue(),
List<String> filesToTail = getFilesToTail(context.getProperty(BASE_DIRECTORY).evaluateAttributeExpressions().getValue(),
context.getProperty(FILENAME).evaluateAttributeExpressions().getValue(),
context.getProperty(RECURSIVE).asBoolean(),
max);
@ -322,12 +322,12 @@ public class TailFile extends AbstractProcessor {
List<String> filesToTail = new ArrayList<String>();
if(context.getProperty(MODE).getValue().equals(MODE_MULTIFILE.getValue())) {
filesToTail.addAll(getFilesToTail(context.getProperty(BASE_DIRECTORY).getValue(),
context.getProperty(FILENAME).getValue(),
filesToTail.addAll(getFilesToTail(context.getProperty(BASE_DIRECTORY).evaluateAttributeExpressions().getValue(),
context.getProperty(FILENAME).evaluateAttributeExpressions().getValue(),
context.getProperty(RECURSIVE).asBoolean(),
maxAge));
} else {
filesToTail.add(context.getProperty(FILENAME).getValue());
filesToTail.add(context.getProperty(FILENAME).evaluateAttributeExpressions().getValue());
}
@ -1130,7 +1130,8 @@ public class TailFile extends AbstractProcessor {
// use a timestamp of lastModified() + 1 so that we do not ingest this file again.
cleanup();
tfo.setState(new TailFileState(context.getProperty(FILENAME).getValue(), null, null, 0L, file.lastModified() + 1L, file.length(), null, tfo.getState().getBuffer()));
tfo.setState(new TailFileState(context.getProperty(FILENAME).evaluateAttributeExpressions().getValue(), null, null, 0L, file.lastModified() + 1L, file.length(), null,
tfo.getState().getBuffer()));
// must ensure that we do session.commit() before persisting state in order to avoid data loss.
session.commit();

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

@ -620,6 +620,27 @@ public class TestTailFile {
runner.getFlowFilesForRelationship(TailFile.REL_SUCCESS).get(4).assertContentEquals("1\n");
}
@Test
public void testMultipleFilesWithBasedirAndFilenameEL() throws IOException, InterruptedException {
runner.setVariable("vrBaseDirectory", "target");
runner.setProperty(TailFile.BASE_DIRECTORY, "${vrBaseDirectory}");
runner.setProperty(TailFile.MODE, TailFile.MODE_MULTIFILE);
runner.setVariable("vrFilename", "(testDir/)?log(ging)?.txt");
runner.setProperty(TailFile.FILENAME, "${vrFilename}");
runner.setProperty(TailFile.ROLLING_FILENAME_PATTERN, "${filename}.?");
runner.setProperty(TailFile.START_POSITION, TailFile.START_CURRENT_FILE);
runner.setProperty(TailFile.RECURSIVE, "true");
runner.run(1);
runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 0);
otherRaf.write("hi\n".getBytes());
raf.write("hello\n".getBytes());
runner.run(1);
runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 2);
}
/**
* This test is used to check the case where we have multiple files in the same directory
* and where it is not possible to specify a single rolling pattern for all files.