nifi-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bbe...@apache.org
Subject nifi git commit: NIFI-5868: Added instrumentation around ListFile such that all disk accesses are timed and any unusually long listing times or disk access operations can be logged. Additionally, information is logged at a debug level including significa
Date Wed, 05 Dec 2018 18:01:37 GMT
Repository: nifi
Updated Branches:
  refs/heads/master 8ebb4d197 -> 72ea93a65


NIFI-5868: Added instrumentation around ListFile such that all disk accesses are timed and any unusually long listing times or disk access operations can be logged. Additionally, information is logged at a debug level including significant amounts of troubleshooting information when configured to do so

This closes #3202.

Signed-off-by: Bryan Bende <bbende@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/72ea93a6
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/72ea93a6
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/72ea93a6

Branch: refs/heads/master
Commit: 72ea93a657cd673bfc40ef74e172f5efb950e95d
Parents: 8ebb4d1
Author: Mark Payne <markap14@hotmail.com>
Authored: Wed Dec 5 09:37:20 2018 -0500
Committer: Bryan Bende <bbende@apache.org>
Committed: Wed Dec 5 13:01:14 2018 -0500

----------------------------------------------------------------------
 .../nifi/processors/standard/ListFile.java      | 974 +++++++++++++++++--
 .../nifi/processors/standard/TestListFile.java  | 154 ++-
 2 files changed, 1032 insertions(+), 96 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/nifi/blob/72ea93a6/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListFile.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListFile.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListFile.java
index 8b73b97..42e2d3f 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListFile.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ListFile.java
@@ -27,12 +27,15 @@ import org.apache.nifi.annotation.documentation.CapabilityDescription;
 import org.apache.nifi.annotation.documentation.SeeAlso;
 import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
 import org.apache.nifi.components.AllowableValue;
 import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyDescriptor.Builder;
 import org.apache.nifi.components.state.Scope;
 import org.apache.nifi.context.PropertyContext;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.processor.DataUnit;
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.ProcessorInitializationContext;
@@ -41,6 +44,7 @@ import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.processor.util.list.AbstractListProcessor;
 import org.apache.nifi.processor.util.list.ListedEntityTracker;
 import org.apache.nifi.processors.standard.util.FileInfo;
+import org.apache.nifi.util.Tuple;
 
 import java.io.File;
 import java.io.IOException;
@@ -57,21 +61,32 @@ import java.nio.file.attribute.PosixFilePermissions;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiPredicate;
+import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.apache.nifi.expression.ExpressionLanguageScope.VARIABLE_REGISTRY;
+import static org.apache.nifi.processor.util.StandardValidators.POSITIVE_INTEGER_VALIDATOR;
+import static org.apache.nifi.processor.util.StandardValidators.TIME_PERIOD_VALIDATOR;
+
 @TriggerSerially
 @InputRequirement(Requirement.INPUT_FORBIDDEN)
 @Tags({"file", "get", "list", "ingest", "source", "filesystem"})
@@ -115,15 +130,15 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
     static final AllowableValue LOCATION_REMOTE = new AllowableValue("Remote", "Remote", "Input Directory is located on a remote system. State will be stored across the cluster so that "
         + "the listing can be performed on Primary Node Only and another node can pick up where the last node left off, if the Primary Node changes");
 
-    public static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor DIRECTORY = new Builder()
             .name("Input Directory")
             .description("The input directory from which files to pull files")
             .required(true)
             .addValidator(StandardValidators.createDirectoryExistsValidator(true, false))
-            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .expressionLanguageSupported(VARIABLE_REGISTRY)
             .build();
 
-    public static final PropertyDescriptor RECURSE = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor RECURSE = new Builder()
             .name("Recurse Subdirectories")
             .description("Indicates whether to list files from subdirectories of the directory")
             .required(true)
@@ -131,7 +146,7 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
             .defaultValue("true")
             .build();
 
-    public static final PropertyDescriptor DIRECTORY_LOCATION = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor DIRECTORY_LOCATION = new Builder()
             .name("Input Directory Location")
             .description("Specifies where the Input Directory is located. This is used to determine whether state should be stored locally or across the cluster.")
             .allowableValues(LOCATION_LOCAL, LOCATION_REMOTE)
@@ -139,7 +154,7 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
             .required(true)
             .build();
 
-    public static final PropertyDescriptor FILE_FILTER = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor FILE_FILTER = new Builder()
             .name("File Filter")
             .description("Only files whose names match the given regular expression will be picked up")
             .required(true)
@@ -147,14 +162,14 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
             .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
             .build();
 
-    public static final PropertyDescriptor PATH_FILTER = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor PATH_FILTER = new Builder()
             .name("Path Filter")
             .description("When " + RECURSE.getName() + " is true, then only subdirectories whose path matches the given regular expression will be scanned")
             .required(false)
             .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
             .build();
 
-    public static final PropertyDescriptor INCLUDE_FILE_ATTRIBUTES = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor INCLUDE_FILE_ATTRIBUTES = new Builder()
         .name("Include File Attributes")
         .description("Whether or not to include information such as the file's Last Modified Time and Owner as FlowFile Attributes. "
             + "Depending on the File System being used, gathering this information can be expensive and as a result should be disabled. This is especially true of remote file shares.")
@@ -163,22 +178,22 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
         .required(true)
         .build();
 
-    public static final PropertyDescriptor MIN_AGE = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor MIN_AGE = new Builder()
             .name("Minimum File Age")
             .description("The minimum age that a file must be in order to be pulled; any file younger than this amount of time (according to last modification date) will be ignored")
             .required(true)
-            .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+            .addValidator(TIME_PERIOD_VALIDATOR)
             .defaultValue("0 sec")
             .build();
 
-    public static final PropertyDescriptor MAX_AGE = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor MAX_AGE = new Builder()
             .name("Maximum File Age")
             .description("The maximum age that a file must be in order to be pulled; any file older than this amount of time (according to last modification date) will be ignored")
             .required(false)
             .addValidator(StandardValidators.createTimePeriodValidator(100, TimeUnit.MILLISECONDS, Long.MAX_VALUE, TimeUnit.NANOSECONDS))
             .build();
 
-    public static final PropertyDescriptor MIN_SIZE = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor MIN_SIZE = new Builder()
             .name("Minimum File Size")
             .description("The minimum size that a file must be in order to be pulled")
             .required(true)
@@ -186,14 +201,14 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
             .defaultValue("0 B")
             .build();
 
-    public static final PropertyDescriptor MAX_SIZE = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor MAX_SIZE = new Builder()
             .name("Maximum File Size")
             .description("The maximum size that a file can be in order to be pulled")
             .required(false)
             .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
             .build();
 
-    public static final PropertyDescriptor IGNORE_HIDDEN_FILES = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor IGNORE_HIDDEN_FILES = new Builder()
             .name("Ignore Hidden Files")
             .description("Indicates whether or not hidden files should be ignored")
             .allowableValues("true", "false")
@@ -201,11 +216,63 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
             .required(true)
             .build();
 
+    public static final PropertyDescriptor TRACK_PERFORMANCE = new Builder()
+        .name("track-performance")
+        .displayName("Track Performance")
+        .description("Whether or not the Processor should track the performance of disk access operations. If true, all accesses to disk will be recorded, including the file being accessed, the " +
+            "information being obtained, and how long it takes. This is then logged periodically at a DEBUG level. While the amount of data will be capped, " +
+            "this option may still consume a significant amount of heap (controlled by the 'Maximum Number of Files to Track' property), " +
+            "but it can be very useful for troubleshooting purposes if performance is poor is degraded.")
+        .required(true)
+        .allowableValues("true", "false")
+        .defaultValue("false")
+        .build();
+
+    public static final PropertyDescriptor MAX_TRACKED_FILES = new Builder()
+        .name("max-performance-metrics")
+        .displayName("Maximum Number of Files to Track")
+        .description("If the 'Track Performance' property is set to 'true', this property indicates the maximum number of files whose performance metrics should be held onto. A smaller value for " +
+            "this property will result in less heap utilization, while a larger value may provide more accurate insights into how the disk access operations are performing")
+        .required(true)
+        .addValidator(POSITIVE_INTEGER_VALIDATOR)
+        .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+        .defaultValue("100000")
+        .build();
+
+    public static final PropertyDescriptor MAX_DISK_OPERATION_TIME = new Builder()
+        .name("max-operation-time")
+        .displayName("Max Disk Operation Time")
+        .description("The maximum amount of time that any single disk operation is expected to take. If any disk operation takes longer than this amount of time, a warning bulletin will be " +
+            "generated for each operation that exceeds this amount of time.")
+        .required(false)
+        .addValidator(TIME_PERIOD_VALIDATOR)
+        .expressionLanguageSupported(VARIABLE_REGISTRY)
+        .defaultValue("10 secs")
+        .build();
+
+    public static final PropertyDescriptor MAX_LISTING_TIME = new Builder()
+        .name("max-listing-time")
+        .displayName("Max Directory Listing Time")
+        .description("The maximum amount of time that listing any single directory is expected to take. If the listing for the directory specified by the 'Input Directory' property, " +
+            "or the listing of any subdirectory (if 'Recurse' is set to true) takes longer than this amount of time, a warning bulletin will be generated for each directory listing " +
+            "that exceeds this amount of time.")
+        .required(false)
+        .addValidator(TIME_PERIOD_VALIDATOR)
+        .expressionLanguageSupported(VARIABLE_REGISTRY)
+        .defaultValue("3 mins")
+        .build();
+
+
     private List<PropertyDescriptor> properties;
     private Set<Relationship> relationships;
 
+    private volatile ScheduledExecutorService monitoringThreadPool;
+    private volatile Future<?> monitoringFuture;
+
     private volatile boolean includeFileAttributes;
-    private final AtomicReference<BiPredicate<Path, BasicFileAttributes>> fileFilterRef = new AtomicReference<BiPredicate<Path, BasicFileAttributes>>();
+    private volatile PerformanceTracker performanceTracker;
+    private volatile long performanceLoggingTimestamp = System.currentTimeMillis();
+    private final AtomicReference<BiPredicate<Path, BasicFileAttributes>> fileFilterRef = new AtomicReference<>();
 
     public static final String FILE_CREATION_TIME_ATTRIBUTE = "file.creationTime";
     public static final String FILE_LAST_MODIFY_TIME_ATTRIBUTE = "file.lastModifiedTime";
@@ -219,8 +286,8 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
     @Override
     protected void init(final ProcessorInitializationContext context) {
         final List<PropertyDescriptor> properties = new ArrayList<>();
-        properties.add(LISTING_STRATEGY);
         properties.add(DIRECTORY);
+        properties.add(LISTING_STRATEGY);
         properties.add(RECURSE);
         properties.add(DIRECTORY_LOCATION);
         properties.add(FILE_FILTER);
@@ -236,11 +303,23 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
         properties.add(ListedEntityTracker.TRACKING_TIME_WINDOW);
         properties.add(ListedEntityTracker.INITIAL_LISTING_TARGET);
         properties.add(ListedEntityTracker.NODE_IDENTIFIER);
+        properties.add(TRACK_PERFORMANCE);
+        properties.add(MAX_TRACKED_FILES);
+        properties.add(MAX_DISK_OPERATION_TIME);
+        properties.add(MAX_LISTING_TIME);
         this.properties = Collections.unmodifiableList(properties);
 
         final Set<Relationship> relationships = new HashSet<>();
         relationships.add(REL_SUCCESS);
         this.relationships = Collections.unmodifiableSet(relationships);
+
+        monitoringThreadPool = Executors.newScheduledThreadPool(1, r -> {
+            final Thread t = Executors.defaultThreadFactory().newThread(r);
+            t.setName("Monitor ListFile Performance [UUID=" + context.getIdentifier() + "]");
+            t.setDaemon(true);
+
+            return t;
+        });
     }
 
     @Override
@@ -257,8 +336,76 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
     public void onScheduled(final ProcessContext context) {
         fileFilterRef.set(createFileFilter(context));
         includeFileAttributes = context.getProperty(INCLUDE_FILE_ATTRIBUTES).asBoolean();
+
+        final long maxDiskOperationMillis = context.getProperty(MAX_DISK_OPERATION_TIME).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS);
+        final long maxListingMillis = context.getProperty(MAX_LISTING_TIME).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS);
+
+        final boolean trackPerformance = context.getProperty(TRACK_PERFORMANCE).asBoolean();
+        if (trackPerformance) {
+            final int maxEntries = context.getProperty(MAX_TRACKED_FILES).evaluateAttributeExpressions().asInteger();
+            performanceTracker = new RollingMetricPerformanceTracker(getLogger(), maxDiskOperationMillis, maxEntries);
+        } else {
+            performanceTracker = new UntrackedPerformanceTracker(getLogger(), maxDiskOperationMillis);
+        }
+
+        final long millisToKeepStats = TimeUnit.MINUTES.toMillis(15);
+        final MonitorActiveTasks monitorTask = new MonitorActiveTasks(performanceTracker, getLogger(), maxDiskOperationMillis, maxListingMillis, millisToKeepStats);
+        monitoringFuture = monitoringThreadPool.scheduleAtFixedRate(monitorTask, 15, 15, TimeUnit.SECONDS);
+    }
+
+    @OnStopped
+    public void onStopped(final ProcessContext context) {
+        if (monitoringFuture != null) {
+            monitoringFuture.cancel(true);
+        }
+
+        final boolean trackPerformance = context.getProperty(TRACK_PERFORMANCE).asBoolean();
+        if (trackPerformance) {
+            logPerformance();
+        }
+    }
+
+    protected PerformanceTracker getPerformanceTracker() {
+        return performanceTracker;
     }
 
+    public void logPerformance() {
+        final ComponentLog logger = getLogger();
+        if (!logger.isDebugEnabled()) {
+            return;
+        }
+
+        final long earliestTimestamp = performanceTracker.getEarliestTimestamp();
+        final long millis = System.currentTimeMillis() - earliestTimestamp;
+        final long seconds = TimeUnit.MILLISECONDS.toSeconds(millis);
+
+        for (final DiskOperation operation : DiskOperation.values()) {
+            final OperationStatistics stats = performanceTracker.getOperationStatistics(operation);
+
+            final StringBuilder sb = new StringBuilder();
+            if (stats.getCount() == 0) {
+                sb.append("Over the past ").append(seconds).append(" seconds, for Operation '").append(operation).append("' there were no operations performed");
+            } else {
+                sb.append("Over the past ").append(seconds).append(" seconds, For Operation '").append(operation).append("' there were ")
+                    .append(stats.getCount()).append(" operations performed with an average time of ")
+                    .append(stats.getAverage()).append(" milliseconds; Standard Deviation = ").append(stats.getStandardDeviation()).append(" millis; Min Time = ")
+                    .append(stats.getMin()).append(" millis, Max Time = ").append(stats.getMax()).append(" millis");
+
+                if (logger.isDebugEnabled()) {
+                    final Map<String, Long> outliers = stats.getOutliers();
+
+                    sb.append("; ").append(stats.getOutliers().size()).append(" significant outliers: ");
+                    sb.append(outliers);
+                }
+            }
+
+            logger.debug(sb.toString());
+        }
+
+        performanceLoggingTimestamp = System.currentTimeMillis();
+    }
+
+
     @Override
     protected Map<String, String> createAttributes(final FileInfo fileInfo, final ProcessContext context) {
         final Map<String, String> attributes = new HashMap<>();
@@ -284,34 +431,43 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
         attributes.put(FILE_LAST_MODIFY_TIME_ATTRIBUTE, formatter.format(new Date(fileInfo.getLastModifiedTime())));
 
         if (includeFileAttributes) {
+            final TimingInfo timingInfo = performanceTracker.getTimingInfo(relativePath.toString(), file.getName());
+
             try {
                 FileStore store = Files.getFileStore(filePath);
-                if (store.supportsFileAttributeView("basic")) {
-                    try {
-                        BasicFileAttributeView view = Files.getFileAttributeView(filePath, BasicFileAttributeView.class);
-                        BasicFileAttributes attrs = view.readAttributes();
-                        attributes.put(FILE_CREATION_TIME_ATTRIBUTE, formatter.format(new Date(attrs.creationTime().toMillis())));
-                        attributes.put(FILE_LAST_ACCESS_TIME_ATTRIBUTE, formatter.format(new Date(attrs.lastAccessTime().toMillis())));
-                    } catch (Exception ignore) {
-                    } // allow other attributes if these fail
-                }
 
-                if (store.supportsFileAttributeView("owner")) {
-                    try {
-                        FileOwnerAttributeView view = Files.getFileAttributeView(filePath, FileOwnerAttributeView.class);
-                        attributes.put(FILE_OWNER_ATTRIBUTE, view.getOwner().getName());
-                    } catch (Exception ignore) {
-                    } // allow other attributes if these fail
-                }
+                timingInfo.timeOperation(DiskOperation.RETRIEVE_BASIC_ATTRIBUTES, () -> {
+                    if (store.supportsFileAttributeView("basic")) {
+                        try {
+                            BasicFileAttributeView view = Files.getFileAttributeView(filePath, BasicFileAttributeView.class);
+                            BasicFileAttributes attrs = view.readAttributes();
+                            attributes.put(FILE_CREATION_TIME_ATTRIBUTE, formatter.format(new Date(attrs.creationTime().toMillis())));
+                            attributes.put(FILE_LAST_ACCESS_TIME_ATTRIBUTE, formatter.format(new Date(attrs.lastAccessTime().toMillis())));
+                        } catch (Exception ignore) {
+                        } // allow other attributes if these fail
+                    }
+                });
 
-                if (store.supportsFileAttributeView("posix")) {
-                    try {
-                        PosixFileAttributeView view = Files.getFileAttributeView(filePath, PosixFileAttributeView.class);
-                        attributes.put(FILE_PERMISSIONS_ATTRIBUTE, PosixFilePermissions.toString(view.readAttributes().permissions()));
-                        attributes.put(FILE_GROUP_ATTRIBUTE, view.readAttributes().group().getName());
-                    } catch (Exception ignore) {
-                    } // allow other attributes if these fail
-                }
+                timingInfo.timeOperation(DiskOperation.RETRIEVE_OWNER_ATTRIBUTES, () -> {
+                    if (store.supportsFileAttributeView("owner")) {
+                        try {
+                            FileOwnerAttributeView view = Files.getFileAttributeView(filePath, FileOwnerAttributeView.class);
+                            attributes.put(FILE_OWNER_ATTRIBUTE, view.getOwner().getName());
+                        } catch (Exception ignore) {
+                        } // allow other attributes if these fail
+                    }
+                });
+
+                timingInfo.timeOperation(DiskOperation.RETRIEVE_POSIX_ATTRIBUTES, () -> {
+                    if (store.supportsFileAttributeView("posix")) {
+                        try {
+                            PosixFileAttributeView view = Files.getFileAttributeView(filePath, PosixFileAttributeView.class);
+                            attributes.put(FILE_PERMISSIONS_ATTRIBUTE, PosixFilePermissions.toString(view.readAttributes().permissions()));
+                            attributes.put(FILE_GROUP_ATTRIBUTE, view.readAttributes().group().getName());
+                        } catch (Exception ignore) {
+                        } // allow other attributes if these fail
+                    }
+                });
             } catch (IOException ioe) {
                 // well then this FlowFile gets none of these attributes
                 getLogger().warn("Error collecting attributes for file {}, message is {}", new Object[] {absPathString, ioe.getMessage()});
@@ -338,36 +494,94 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
 
     @Override
     protected List<FileInfo> performListing(final ProcessContext context, final Long minTimestamp) throws IOException {
-        final Path path = new File(getPath(context)).toPath();
+        final Path basePath = new File(getPath(context)).toPath();
         final Boolean recurse = context.getProperty(RECURSE).asBoolean();
         final Map<Path, BasicFileAttributes> lastModifiedMap = new HashMap<>();
 
         final BiPredicate<Path, BasicFileAttributes> fileFilter = fileFilterRef.get();
         int maxDepth = recurse ? Integer.MAX_VALUE : 1;
-        BiPredicate<Path, BasicFileAttributes> matcher = (p, attributes) -> {
-            if (!attributes.isDirectory()
-                    && (minTimestamp == null || attributes.lastModifiedTime().toMillis() >= minTimestamp)
-                    && fileFilter.test(p, attributes)) {
-                // We store the attributes for each Path we are returning in order to avoid to
-                // retrieve them again later when creating the FileInfo
-                lastModifiedMap.put(p, attributes);
-                return true;
-            }
-            return false;
+
+        final BiPredicate<Path, BasicFileAttributes> matcher = new BiPredicate<Path, BasicFileAttributes>() {
+            private long lastTimestamp = System.currentTimeMillis();
+
+            @Override
+            public boolean test(final Path path, final BasicFileAttributes attributes) {
+                if (!isScheduled()) {
+                    throw new ProcessorStoppedException();
+                }
+
+                final long now = System.currentTimeMillis();
+                final long timeToList = now - lastTimestamp;
+                lastTimestamp = now;
+
+                final Path relativeDirectory = basePath.relativize(path).getParent();
+                final String relativePath = relativeDirectory == null ? "" : relativeDirectory.toString();
+                final String filename = path.getFileName().toString();
+                performanceTracker.acceptOperation(DiskOperation.RETRIEVE_NEXT_FILE_FROM_OS, relativePath, filename, timeToList);
+
+                final boolean isDirectory = attributes.isDirectory();
+                if (isDirectory) {
+                    performanceTracker.setActiveDirectory(relativePath);
+                }
+
+                final TimedOperationKey operationKey = performanceTracker.beginOperation(DiskOperation.FILTER, relativePath, filename);
+
+                try {
+                    if (!isDirectory && (minTimestamp == null || attributes.lastModifiedTime().toMillis() >= minTimestamp)
+                        && fileFilter.test(path, attributes)) {
+                        // We store the attributes for each Path we are returning in order to avoid to
+                        // retrieve them again later when creating the FileInfo
+                        lastModifiedMap.put(path, attributes);
+
+                        return true;
+                    }
+
+                    return false;
+                } finally {
+                    performanceTracker.completeOperation(operationKey);
+
+                    if (TimeUnit.MILLISECONDS.toMinutes(System.currentTimeMillis() - performanceLoggingTimestamp) >= 5) {
+                        logPerformance();
+                    }
+                }
+            }
         };
-        Stream<Path> inputStream = Files.find(path, maxDepth, matcher, FileVisitOption.FOLLOW_LINKS);
-        Stream<FileInfo> listing = inputStream.map(p -> {
+
+        final Stream<Path> inputStream = getPathStream(basePath, maxDepth, matcher);
+
+        final Stream<FileInfo> listing = inputStream.map(p -> {
             File file = p.toFile();
             BasicFileAttributes attributes = lastModifiedMap.get(p);
-            return new FileInfo.Builder()
+
+            final FileInfo fileInfo = new FileInfo.Builder()
                 .directory(false)
                 .filename(file.getName())
                 .fullPathFileName(file.getAbsolutePath())
                 .lastModifiedTime(attributes.lastModifiedTime().toMillis())
                 .size(attributes.size())
                 .build();
+
+            return fileInfo;
         });
-        return listing.collect(Collectors.toList());
+
+        // Perform the actual listing
+        try {
+            final long start = System.currentTimeMillis();
+            final List<FileInfo> fileInfos = listing.collect(Collectors.toList());
+            final long millis = System.currentTimeMillis() - start;
+
+            getLogger().debug("Took {} milliseconds to perform listing and gather {} entries", new Object[] {millis, fileInfos.size()});
+            return fileInfos;
+        } catch (final ProcessorStoppedException pse) {
+            getLogger().info("Processor was stopped so will not complete listing of Files");
+            return Collections.emptyList();
+        } finally {
+            performanceTracker.completeActiveDirectory();
+        }
+    }
+
+    protected Stream<Path> getPathStream(final Path basePath, final int maxDepth, final BiPredicate<Path, BasicFileAttributes> matcher) throws IOException {
+        return Files.find(basePath, maxDepth, matcher, FileVisitOption.FOLLOW_LINKS);
     }
 
     @Override
@@ -389,11 +603,15 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
         final long minAge = context.getProperty(MIN_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
         final Long maxAge = context.getProperty(MAX_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
         final boolean ignoreHidden = context.getProperty(IGNORE_HIDDEN_FILES).asBoolean();
-        final Pattern filePattern = Pattern.compile(context.getProperty(FILE_FILTER).getValue());
+        final String fileFilter = context.getProperty(FILE_FILTER).getValue();
+        final Pattern filePattern = Pattern.compile(fileFilter);
         final String indir = context.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue();
         final boolean recurseDirs = context.getProperty(RECURSE).asBoolean();
         final String pathPatternStr = context.getProperty(PATH_FILTER).getValue();
         final Pattern pathPattern = (!recurseDirs || pathPatternStr == null) ? null : Pattern.compile(pathPatternStr);
+
+        final Path basePath = Paths.get(indir);
+
         return (path, attributes) -> {
             if (minSize > attributes.size()) {
                 return false;
@@ -408,22 +626,652 @@ public class ListFile extends AbstractListProcessor<FileInfo> {
             if (maxAge != null && maxAge < fileAge) {
                 return false;
             }
-            if (ignoreHidden && path.toFile().isHidden()) {
-                return false;
-            }
+
+            final Path relativePath = basePath.relativize(path).getParent();
+            final String relativeDir = relativePath == null ? "" : relativePath.toString();
+            final String filename = path.getFileName().toString();
+            final TimingInfo timingInfo = performanceTracker.getTimingInfo(relativeDir, filename);
+
+            final File file = path.toFile();
+
             if (pathPattern != null) {
-                Path reldir = Paths.get(indir).relativize(path).getParent();
-                if (reldir != null && !reldir.toString().isEmpty()) {
-                    if (!pathPattern.matcher(reldir.toString()).matches()) {
+                if (relativePath != null && !relativePath.toString().isEmpty()) {
+                    if (!pathPattern.matcher(relativePath.toString()).matches()) {
                         return false;
                     }
                 }
             }
+
+            final boolean matchesFilter = filePattern.matcher(filename).matches();
+            if (!matchesFilter) {
+                return false;
+            }
+
             // Verify that we have at least read permissions on the file we're considering grabbing
-            if (!Files.isReadable(path)) {
+            if (!timingInfo.timeOperation(DiskOperation.CHECK_READABLE, () -> Files.isReadable(path))) {
+                return false;
+            }
+
+            if (ignoreHidden && timingInfo.timeOperation(DiskOperation.CHECK_HIDDEN, file::isHidden)) {
                 return false;
             }
-            return filePattern.matcher(path.getFileName().toString()).matches();
+
+            return true;
         };
     }
+
+    /**
+     * A PerformanceTracker that is capable of tracking which disk access operation is active and which directory is actively being listed,
+     * as well as timing specific operations, but does not track metrics over any amount of time. This implementation does not provide the ability
+     * to glean information such as which operations or files are taking the longest to operate on but uses very little heap.
+     */
+    public static class UntrackedPerformanceTracker implements PerformanceTracker {
+        private TimedOperationKey activeOperation = null;
+        private String activeDirectory;
+        private long activeDirectoryStartTime = -1L;
+
+        private final ComponentLog logger;
+        private final long maxDiskOperationMillis;
+
+        public UntrackedPerformanceTracker(final ComponentLog logger, final long maxDiskOperationMillis) {
+            this.logger = logger;
+            this.maxDiskOperationMillis = maxDiskOperationMillis;
+        }
+
+        @Override
+        public TimedOperationKey beginOperation(final DiskOperation operation, final String directory, final String filename) {
+            return null;
+        }
+
+        @Override
+        public void completeOperation(final TimedOperationKey operationKey) {
+        }
+
+        @Override
+        public void acceptOperation(final DiskOperation operation, final String directory, final String filename, final long millis) {
+        }
+
+        @Override
+        public TimingInfo getTimingInfo(final String directory, final String filename) {
+            return new TimingInfo(directory, filename, this, logger, maxDiskOperationMillis);
+        }
+
+        @Override
+        public OperationStatistics getOperationStatistics(final DiskOperation operation) {
+            return OperationStatistics.EMPTY;
+        }
+
+        @Override
+        public synchronized void setActiveOperation(final TimedOperationKey operationKey) {
+            this.activeOperation = operationKey;
+        }
+
+        @Override
+        public synchronized void completeActiveOperation() {
+            this.activeOperation = null;
+        }
+
+        @Override
+        public synchronized TimedOperationKey getActiveOperation() {
+            return activeOperation;
+        }
+
+        @Override
+        public void purgeTimingInfo(final long cutoff) {
+        }
+
+        @Override
+        public long getEarliestTimestamp() {
+            return System.currentTimeMillis();
+        }
+
+        @Override
+        public synchronized void setActiveDirectory(final String directory) {
+            activeDirectory = directory;
+            activeDirectoryStartTime = System.currentTimeMillis();
+        }
+
+        @Override
+        public synchronized void completeActiveDirectory() {
+            activeDirectory = null;
+            activeDirectoryStartTime = -1L;
+        }
+
+        @Override
+        public synchronized long getActiveDirectoryStartTime() {
+            return activeDirectoryStartTime;
+        }
+
+        @Override
+        public synchronized String getActiveDirectory() {
+            return activeDirectory;
+        }
+
+        @Override
+        public int getTrackedFileCount() {
+            return 0;
+        }
+    }
+
+    /**
+     * Tracks metrics using a rolling window of time, in which older metrics are 'aged off' by calling {@link #purgeTimingInfo(long)}. Tracking these metrics allows information
+     * to be gleaned, such as which files are expensive to operate on or which operations are most expensive. However, the heap utilization is significant.
+     */
+    public static final class RollingMetricPerformanceTracker implements PerformanceTracker {
+        private final Map<String, String> directoryCanonicalization = new HashMap<>();
+        private final Map<Tuple<String, String>, TimingInfo> directoryToTimingInfo;
+        private TimedOperationKey activeOperation;
+        private long earliestTimestamp = System.currentTimeMillis();
+        private final long maxDiskOperationMillis;
+        private final ComponentLog logger;
+
+        private String activeDirectory;
+        private long activeDirectoryStartTime = -1L;
+
+        public RollingMetricPerformanceTracker(final ComponentLog logger, final long maxDiskOperationMillis, final int maxEntries) {
+            this.logger = logger;
+            this.maxDiskOperationMillis = maxDiskOperationMillis;
+
+            directoryToTimingInfo = new LinkedHashMap<Tuple<String, String>, TimingInfo>() {
+                @Override
+                protected boolean removeEldestEntry(final Map.Entry<Tuple<String, String>, TimingInfo> eldest) {
+                    return size() > maxEntries;
+                }
+            };
+        }
+
+        @Override
+        public synchronized TimedOperationKey beginOperation(final DiskOperation operation, final String directory, final String filename) {
+            return new TimedOperationKey(operation, directory, filename, System.currentTimeMillis());
+        }
+
+        @Override
+        public synchronized void completeOperation(final TimedOperationKey operationKey) {
+            final TimingInfo timingInfo = getTimingInfo(operationKey.getDirectory(), operationKey.getFilename());
+            timingInfo.accept(operationKey.getOperation(), System.currentTimeMillis() - operationKey.getStartTime());
+        }
+
+        @Override
+        public synchronized void acceptOperation(final DiskOperation operation, final String directory, final String filename, final long millis) {
+            final String canonicalDirectory = directoryCanonicalization.computeIfAbsent(directory, key -> directory);
+            final Tuple<String, String> key = new Tuple<>(canonicalDirectory, filename);
+            final TimingInfo timingInfo = directoryToTimingInfo.computeIfAbsent(key, k -> new TimingInfo(directory, filename, this, logger, maxDiskOperationMillis));
+            timingInfo.accept(operation, millis);
+        }
+
+        @Override
+        public synchronized TimingInfo getTimingInfo(final String directory, final String filename) {
+            final String canonicalDirectory = directoryCanonicalization.computeIfAbsent(directory, key -> directory);
+            final Tuple<String, String> key = new Tuple<>(canonicalDirectory, filename);
+            final TimingInfo timingInfo = directoryToTimingInfo.computeIfAbsent(key, k -> new TimingInfo(directory, filename, this, logger, maxDiskOperationMillis));
+
+            return timingInfo;
+        }
+
+        @Override
+        public void setActiveOperation(final TimedOperationKey activeOperation) {
+            this.activeOperation = activeOperation;
+        }
+
+        @Override
+        public void completeActiveOperation() {
+            this.activeOperation = null;
+        }
+
+        @Override
+        public synchronized TimedOperationKey getActiveOperation() {
+            return activeOperation;
+        }
+
+        @Override
+        public synchronized void setActiveDirectory(final String directory) {
+            activeDirectory = directory;
+            activeDirectoryStartTime = System.currentTimeMillis();
+        }
+
+        @Override
+        public synchronized void completeActiveDirectory() {
+            activeDirectory = null;
+            activeDirectoryStartTime = -1L;
+        }
+
+        @Override
+        public synchronized long getActiveDirectoryStartTime() {
+            return activeDirectoryStartTime;
+        }
+
+        @Override
+        public synchronized String getActiveDirectory() {
+            return activeDirectory;
+        }
+
+        @Override
+        public synchronized int getTrackedFileCount() {
+            return directoryToTimingInfo.size();
+        }
+
+        @Override
+        public synchronized void purgeTimingInfo(final long cutoff) {
+            logger.debug("Purging any entries from Performance Tracker that is older than {}", new Object[] {new Date(cutoff)});
+            final Iterator<Map.Entry<Tuple<String, String>, TimingInfo>> itr = directoryToTimingInfo.entrySet().iterator();
+
+            int purgedCount = 0;
+            long earliestTimestamp = System.currentTimeMillis();
+            while (itr.hasNext()) {
+                final Map.Entry<Tuple<String, String>, TimingInfo> entry = itr.next();
+                final TimingInfo timingInfo = entry.getValue();
+                final long creationTime = timingInfo.getCreationTimestamp();
+
+                if (creationTime < cutoff) {
+                    itr.remove();
+                    purgedCount++;
+
+                    directoryCanonicalization.remove(entry.getKey().getKey());
+                } else {
+                    earliestTimestamp = Math.min(earliestTimestamp, creationTime);
+                }
+            }
+
+            this.earliestTimestamp = earliestTimestamp;
+            logger.debug("Purged {} entries from Performance Tracker; now holding {} entries", new Object[] {purgedCount, directoryToTimingInfo.size()});
+        }
+
+        public long getEarliestTimestamp() {
+            return earliestTimestamp;
+        }
+
+        public synchronized OperationStatistics getOperationStatistics(final DiskOperation operation) {
+            long count = 0L;
+            long sum = 0L;
+            long min = 0L;
+            long max = 0L;
+
+            // Calculate min/max/mean
+            for (final TimingInfo timingInfo : directoryToTimingInfo.values()) {
+                final long operationTime = timingInfo.getOperationTime(operation);
+
+                if (operationTime < 0) { // operation not conducted
+                    continue;
+                }
+
+                sum += operationTime;
+
+                if (count++ == 0) {
+                    min = operationTime;
+                    max = operationTime;
+                } else {
+                    min = Math.min(min, operationTime);
+                    max = Math.max(max, operationTime);
+                }
+            }
+
+            if (count == 0) {
+                return OperationStatistics.EMPTY;
+            }
+
+            double average = (double) sum / (double) count;
+
+            // Calculate Standard Deviation
+            final double stdDeviation = calculateStdDev(average, (double) count, operation);
+            final double outlierCutoff = average + 2 * stdDeviation;
+
+            final Map<String, Long> outliers = new HashMap<>();
+            for (final TimingInfo timingInfo : directoryToTimingInfo.values()) {
+                final long operationTime = timingInfo.getOperationTime(operation);
+
+                if (operationTime > 2 && operationTime > outlierCutoff) {
+                    final String directory = timingInfo.getDirectory();
+                    final String filename = timingInfo.getFilename();
+                    final String fullPath = directory.endsWith("/") ? directory + filename : directory + "/" + filename;
+                    outliers.put(fullPath, operationTime);
+                }
+            }
+
+            return new StandardOperationStatistics(min, max, count, average, stdDeviation, outliers);
+        }
+
+        private double calculateStdDev(final double average, final double count, final DiskOperation operation) {
+            double squaredDifferenceSum = 0D;
+            for (final TimingInfo timingInfo : directoryToTimingInfo.values()) {
+                final long operationTime = timingInfo.getOperationTime(operation);
+                if (operationTime < 0) {
+                    continue;
+                }
+
+                final double differenceSquared = Math.pow(((double) operationTime - average), 2);
+                squaredDifferenceSum += differenceSquared;
+            }
+
+            final double squaredDifferenceAverage = squaredDifferenceSum / count;
+            final double stdDeviation = Math.pow(squaredDifferenceAverage, 0.5);
+            return stdDeviation;
+        }
+    }
+
+    /**
+     * Provides a mechanism for timing how long a particular operation takes to complete, logging if it takes longer than the configured threshold.
+     */
+    private static class TimingInfo {
+        private final String directory;
+        private final String filename;
+        private final int[] operationTimes;
+        private final PerformanceTracker tracker;
+        private final long creationTimestamp;
+        private final ComponentLog logger;
+        private final long maxDiskOperationMillis;
+
+        public TimingInfo(final String directory, final String filename, final PerformanceTracker tracker, final ComponentLog logger, final long maxDiskOperationMillis) {
+            this.directory = directory;
+            this.filename = filename;
+            this.tracker = tracker;
+            this.logger = logger;
+            this.maxDiskOperationMillis = maxDiskOperationMillis;
+
+            this.creationTimestamp = System.currentTimeMillis();
+
+            operationTimes = new int[DiskOperation.values().length];
+            Arrays.fill(operationTimes, -1);
+        }
+
+        public String getDirectory() {
+            return directory;
+        }
+
+        public String getFilename() {
+            return filename;
+        }
+
+        public void accept(final DiskOperation operation, final long duration) {
+            operationTimes[operation.ordinal()] = (int) duration;
+
+            if (duration > maxDiskOperationMillis) {
+                final String fullPath = getFullPath();
+                logger.warn("This Processor completed action {} on {} in {} milliseconds, which exceeds the configured threshold of {} milliseconds",
+                    new Object[] {operation, fullPath, duration, maxDiskOperationMillis});
+            }
+
+            if (logger.isTraceEnabled()) {
+                logger.trace("Performing operation {} on {} took {} milliseconds", new Object[] {operation, getFullPath(), duration});
+            }
+        }
+
+        private String getFullPath() {
+            if (directory.isEmpty()) {
+                return filename;
+            } else {
+                return directory.endsWith("/") ? directory + filename : directory + "/" + filename;
+            }
+        }
+
+        public long getOperationTime(final DiskOperation operation) {
+            return operationTimes[operation.ordinal()];
+        }
+
+        private <T> T timeOperation(final DiskOperation operation, final Supplier<T> function) {
+            final long start = System.currentTimeMillis();
+            final TimedOperationKey operationKey = new TimedOperationKey(operation, directory, filename, start);
+            tracker.setActiveOperation(operationKey);
+
+            try {
+                final T value = function.get();
+                final long millis = System.currentTimeMillis() - start;
+                accept(operation, millis);
+                return value;
+            } finally {
+                tracker.completeActiveOperation();
+            }
+        }
+
+        private void timeOperation(final DiskOperation operation, final Runnable task) {
+            final long start = System.currentTimeMillis();
+            final TimedOperationKey operationKey = new TimedOperationKey(operation, directory, filename, start);
+            tracker.setActiveOperation(operationKey);
+
+            try {
+                task.run();
+                final long millis = System.currentTimeMillis() - start;
+                accept(operation, millis);
+            } finally {
+                tracker.completeActiveOperation();
+            }
+        }
+
+        public long getCreationTimestamp() {
+            return creationTimestamp;
+        }
+    }
+
+    /**
+     * PerformanceTracker is responsible for providing a mechanism by which any disk operation can be timed and the timing information
+     * can both be used to issue warnings as well as be aggregated for some amount of time, in order to understand how long certain disk operations
+     * take and which files may be responsible for causing longer-than-usual operations to be performed.
+     */
+    interface PerformanceTracker {
+        TimedOperationKey beginOperation(DiskOperation operation, String directory, String filename);
+
+        void completeOperation(TimedOperationKey operationKey);
+
+        void acceptOperation(DiskOperation operation, String directory, String filename, long millis);
+
+        TimingInfo getTimingInfo(String directory, String filename);
+
+        OperationStatistics getOperationStatistics(DiskOperation operation);
+
+        void setActiveOperation(TimedOperationKey operationKey);
+
+        void completeActiveOperation();
+
+        TimedOperationKey getActiveOperation();
+
+        void purgeTimingInfo(long cutoff);
+
+        long getEarliestTimestamp();
+
+        void setActiveDirectory(String directory);
+
+        void completeActiveDirectory();
+
+        String getActiveDirectory();
+
+        long getActiveDirectoryStartTime();
+
+        int getTrackedFileCount();
+    }
+
+
+    interface OperationStatistics {
+        long getMin();
+        long getMax();
+        long getCount();
+        double getAverage();
+        double getStandardDeviation();
+
+        Map<String, Long> getOutliers();
+
+        OperationStatistics EMPTY = new OperationStatistics() {
+            @Override
+            public long getMin() {
+                return 0;
+            }
+
+            @Override
+            public long getMax() {
+                return 0;
+            }
+
+            @Override
+            public long getCount() {
+                return 0;
+            }
+
+            @Override
+            public double getAverage() {
+                return 0;
+            }
+
+            @Override
+            public double getStandardDeviation() {
+                return 0;
+            }
+
+            @Override
+            public Map<String, Long> getOutliers() {
+                return Collections.emptyMap();
+            }
+        };
+    }
+
+    private static class StandardOperationStatistics implements OperationStatistics {
+        private final long min;
+        private final long max;
+        private final long count;
+        private final double average;
+        private final double stdDev;
+        private final Map<String, Long> outliers;
+
+        public StandardOperationStatistics(final long min, final long max, final long count, final double average, final double stdDev, final Map<String, Long> outliers) {
+            this.min = min;
+            this.max = max;
+            this.count = count;
+            this.average = average;
+            this.stdDev = stdDev;
+            this.outliers = outliers;
+        }
+
+        public long getMin() {
+            return min;
+        }
+
+        public long getMax() {
+            return max;
+        }
+
+        public long getCount() {
+            return count;
+        }
+
+        public double getAverage() {
+            return average;
+        }
+
+        public double getStandardDeviation() {
+            return stdDev;
+        }
+
+        public Map<String, Long> getOutliers() {
+            return outliers;
+        }
+    }
+
+
+    private static class TimedOperationKey {
+        private final DiskOperation operation;
+        private final String directory;
+        private final String filename;
+        private final long startTime;
+
+        public TimedOperationKey(final DiskOperation operation, final String directory, final String filename, final long startTime) {
+            this.operation = operation;
+            this.startTime = startTime;
+            this.directory = directory;
+            this.filename = filename;
+        }
+
+        public DiskOperation getOperation() {
+            return operation;
+        }
+
+        public String getDirectory() {
+            return directory;
+        }
+
+        public String getFilename() {
+            return filename;
+        }
+
+        public long getStartTime() {
+            return startTime;
+        }
+    }
+
+    private enum DiskOperation {
+        RETRIEVE_BASIC_ATTRIBUTES,
+        RETRIEVE_OWNER_ATTRIBUTES,
+        RETRIEVE_POSIX_ATTRIBUTES,
+        CHECK_HIDDEN,
+        CHECK_READABLE,
+        FILTER,
+        RETRIEVE_NEXT_FILE_FROM_OS;
+    }
+
+    private static class ProcessorStoppedException extends RuntimeException {
+    }
+
+    static class MonitorActiveTasks implements Runnable {
+        private final PerformanceTracker performanceTracker;
+        private final ComponentLog logger;
+        private final long maxDiskOperationMillis;
+        private final long maxListingMillis;
+        private final long millisToKeepStats;
+        private long lastPurgeTimestamp = 0L;
+
+        public MonitorActiveTasks(final PerformanceTracker tracker, final ComponentLog logger, final long maxDiskOperationMillis, final long maxListingMillis, final long millisToKeepStats) {
+            this.performanceTracker = tracker;
+            this.logger = logger;
+            this.maxDiskOperationMillis = maxDiskOperationMillis;
+            this.maxListingMillis = maxListingMillis;
+            this.millisToKeepStats = millisToKeepStats;
+        }
+
+        @Override
+        public void run() {
+            monitorActiveOperation();
+            monitorActiveDirectory();
+
+            final long now = System.currentTimeMillis();
+            final long millisSincePurge = now - lastPurgeTimestamp;
+            if (millisSincePurge > TimeUnit.SECONDS.toMillis(60)) {
+                performanceTracker.purgeTimingInfo(now - millisToKeepStats);
+                lastPurgeTimestamp = System.currentTimeMillis();
+            }
+        }
+
+        private void monitorActiveOperation() {
+            final TimedOperationKey activeOperation = performanceTracker.getActiveOperation();
+            if (activeOperation == null) {
+                return;
+            }
+
+            final long activeTime = System.currentTimeMillis() - activeOperation.getStartTime();
+            if (activeTime > maxDiskOperationMillis) {
+                final String directory = activeOperation.getDirectory();
+                final String filename = activeOperation.getFilename();
+
+                final String fullPath;
+                if (directory.isEmpty()) {
+                    fullPath = filename;
+                } else {
+                    fullPath = directory.endsWith("/") ? directory + filename : directory + "/" + filename;
+                }
+
+                logger.warn("This Processor has currently spent {} milliseconds performing the {} action on {}, which exceeds the configured threshold of {} milliseconds",
+                    new Object[] {activeTime, activeOperation.getOperation(), fullPath, maxDiskOperationMillis});
+            }
+        }
+
+        private void monitorActiveDirectory() {
+            final String activeDirectory = performanceTracker.getActiveDirectory();
+            final long startTime = performanceTracker.getActiveDirectoryStartTime();
+            if (startTime <= 0) {
+                return;
+            }
+
+            final long activeMillis = System.currentTimeMillis() - startTime;
+            if (activeMillis > maxListingMillis) {
+                final String fullPath = activeDirectory.isEmpty() ? "the base directory" : activeDirectory;
+                logger.warn("This processor has currently spent {} milliseconds performing the listing of {}, which exceeds the configured threshold of {} milliseconds",
+                    new Object[] {activeMillis, fullPath, maxListingMillis});
+            }
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/nifi/blob/72ea93a6/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListFile.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListFile.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListFile.java
index bf2755b..b9276be 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListFile.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListFile.java
@@ -17,10 +17,22 @@
 
 package org.apache.nifi.processors.standard;
 
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.util.list.AbstractListProcessor;
+import org.apache.nifi.processor.util.list.ListProcessorTestWatcher;
+import org.apache.nifi.processors.standard.util.FileInfo;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.Description;
 
 import java.io.File;
 import java.io.FileOutputStream;
@@ -28,6 +40,8 @@ import java.io.IOException;
 import java.nio.file.FileStore;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.nio.file.attribute.FileTime;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
@@ -38,27 +52,15 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiPredicate;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.state.Scope;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.flowfile.attributes.CoreAttributes;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.Relationship;
-import org.apache.nifi.processor.util.list.AbstractListProcessor;
-import org.apache.nifi.processor.util.list.ListProcessorTestWatcher;
-import org.apache.nifi.processors.standard.util.FileInfo;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.runner.Description;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 public class TestListFile {
 
@@ -160,13 +162,99 @@ public class TestListFile {
         dumpState.dumpState(startedAtMillis);
     }
 
+
     @Test
-    public void testGetRelationships() throws Exception {
-        Set<Relationship> relationships = processor.getRelationships();
-        assertEquals(1, relationships.size());
-        assertEquals(AbstractListProcessor.REL_SUCCESS, relationships.toArray()[0]);
+    @Ignore("Intended only for manual testing, as is very expensive to run as a unit test. Performs listing of 1,000,000 files (doesn't actually create the files, though - injects them in) to " +
+        "ensure performance is not harmed")
+    public void testPerformanceOnLargeListing() {
+        final List<Path> paths = new ArrayList<>(1_000_000);
+        final File base = new File("target");
+
+        for (int firstLevel=0; firstLevel < 1000; firstLevel++) {
+            final File dir = new File(base, String.valueOf(firstLevel));
+
+            for (int secondLevel = 0; secondLevel < 1000; secondLevel++) {
+                final File file = new File(dir, String.valueOf(secondLevel));
+                paths.add(file.toPath());
+            }
+        }
+
+        final BasicFileAttributes basicFileAttributes = new BasicFileAttributes() {
+            @Override
+            public FileTime lastModifiedTime() {
+                return FileTime.fromMillis(System.currentTimeMillis());
+            }
+
+            @Override
+            public FileTime lastAccessTime() {
+                return FileTime.fromMillis(System.currentTimeMillis());
+            }
+
+            @Override
+            public FileTime creationTime() {
+                return FileTime.fromMillis(System.currentTimeMillis());
+            }
+
+            @Override
+            public boolean isRegularFile() {
+                return false;
+            }
+
+            @Override
+            public boolean isDirectory() {
+                return false;
+            }
+
+            @Override
+            public boolean isSymbolicLink() {
+                return false;
+            }
+
+            @Override
+            public boolean isOther() {
+                return false;
+            }
+
+            @Override
+            public long size() {
+                return 0;
+            }
+
+            @Override
+            public Object fileKey() {
+                return null;
+            }
+        };
+
+        processor = new ListFile() {
+            @Override
+            protected Stream<Path> getPathStream(final Path basePath, final int maxDepth, final BiPredicate<Path, BasicFileAttributes> matcher) throws IOException {
+                return paths.stream()
+                    .filter(path -> matcher.test(path, basicFileAttributes));
+            }
+        };
+
+        runner = TestRunners.newTestRunner(processor);
+        runner.setProperty(AbstractListProcessor.TARGET_SYSTEM_TIMESTAMP_PRECISION, AbstractListProcessor.PRECISION_SECONDS.getValue());
+        runner.setProperty(ListFile.TRACK_PERFORMANCE, "true");
+        runner.setProperty(ListFile.MAX_TRACKED_FILES, "100000");
+        runner.setProperty(ListFile.DIRECTORY, "target");
+
+        runner.run();
+
+        final ListFile.PerformanceTracker tracker = processor.getPerformanceTracker();
+        assertEquals(100_000, tracker.getTrackedFileCount());
+
+        final ListFile.MonitorActiveTasks monitorActiveTasks = new ListFile.MonitorActiveTasks(tracker, runner.getLogger(), 1000, 1000, 1);
+
+        while (tracker.getTrackedFileCount() > 0) {
+            monitorActiveTasks.run();
+        }
+
+        assertEquals(0, tracker.getTrackedFileCount());
     }
 
+
     @Test
     public void testGetPath() {
         runner.setProperty(ListFile.DIRECTORY, "/dir/test1");
@@ -636,14 +724,14 @@ public class TestListFile {
 
     @Test
     public void testIsListingResetNecessary() throws Exception {
-        assertEquals(true, processor.isListingResetNecessary(ListFile.DIRECTORY));
-        assertEquals(true, processor.isListingResetNecessary(ListFile.RECURSE));
-        assertEquals(true, processor.isListingResetNecessary(ListFile.FILE_FILTER));
-        assertEquals(true, processor.isListingResetNecessary(ListFile.PATH_FILTER));
-        assertEquals(true, processor.isListingResetNecessary(ListFile.MIN_AGE));
-        assertEquals(true, processor.isListingResetNecessary(ListFile.MAX_AGE));
-        assertEquals(true, processor.isListingResetNecessary(ListFile.MIN_SIZE));
-        assertEquals(true, processor.isListingResetNecessary(ListFile.MAX_SIZE));
+        assertTrue(processor.isListingResetNecessary(ListFile.DIRECTORY));
+        assertTrue(processor.isListingResetNecessary(ListFile.RECURSE));
+        assertTrue(processor.isListingResetNecessary(ListFile.FILE_FILTER));
+        assertTrue(processor.isListingResetNecessary(ListFile.PATH_FILTER));
+        assertTrue(processor.isListingResetNecessary(ListFile.MIN_AGE));
+        assertTrue(processor.isListingResetNecessary(ListFile.MAX_AGE));
+        assertTrue(processor.isListingResetNecessary(ListFile.MIN_SIZE));
+        assertTrue(processor.isListingResetNecessary(ListFile.MAX_SIZE));
         assertEquals(true, processor.isListingResetNecessary(ListFile.IGNORE_HIDDEN_FILES));
         assertEquals(false, processor.isListingResetNecessary(new PropertyDescriptor.Builder().name("x").build()));
     }


Mime
View raw message