Package org.apache.nifi.processors.standard

Class TailFile

java.lang.Object
org.apache.nifi.components.AbstractConfigurableComponent
org.apache.nifi.processor.AbstractSessionFactoryProcessor
org.apache.nifi.processor.AbstractProcessor
org.apache.nifi.processors.standard.TailFile
All Implemented Interfaces:
org.apache.nifi.components.ConfigurableComponent, org.apache.nifi.processor.Processor
@TriggerSerially @InputRequirement(INPUT_FORBIDDEN) @Tags({"tail","file","log","text","source"}) @CapabilityDescription("\"Tails\" a file, or a list of files, ingesting data from the file as it is written to the file. The file is expected to be textual. Data is ingested only when a new line is encountered (carriage return or new-line character or combination). If the file to tail is periodically \"rolled over\", as is generally the case with log files, an optional Rolling Filename Pattern can be used to retrieve data from files that have rolled over, even if the rollover occurred while NiFi was not running (provided that the data still exists upon restart of NiFi). It is generally advisable to set the Run Schedule to a few seconds, rather than running with the default value of 0 secs, as this Processor will consume a lot of resources if scheduled very aggressively. At this time, this Processor does not support ingesting files that have been compressed when \'rolled over\'.") @Stateful(scopes={LOCAL,CLUSTER}, description="Stores state about where in the Tailed File it left off so that on restart it does not have to duplicate data. State is stored either local or clustered depend on the <File Location> property.") @WritesAttribute(attribute="tailfile.original.path", description="Path of the original file the flow file comes from.") @Restricted(restrictions=@Restriction(requiredPermission=READ_FILESYSTEM,explanation="Provides operator the ability to read from any file that NiFi has access to.")) @DefaultSchedule(strategy=TIMER_DRIVEN, period="30 sec") public class TailFile extends org.apache.nifi.processor.AbstractProcessor

  • Field Details

    • MAP_PREFIX

      static final String MAP_PREFIX
      See Also:

    • NEW_LINE_BYTES

      private static final byte[] NEW_LINE_BYTES

    • LOCATION_LOCAL

      static final org.apache.nifi.components.AllowableValue LOCATION_LOCAL

    • LOCATION_REMOTE

      static final org.apache.nifi.components.AllowableValue LOCATION_REMOTE

    • MODE_SINGLEFILE

      static final org.apache.nifi.components.AllowableValue MODE_SINGLEFILE

    • MODE_MULTIFILE

      static final org.apache.nifi.components.AllowableValue MODE_MULTIFILE

    • START_BEGINNING_OF_TIME

      static final org.apache.nifi.components.AllowableValue START_BEGINNING_OF_TIME

    • START_CURRENT_FILE

      static final org.apache.nifi.components.AllowableValue START_CURRENT_FILE

    • START_CURRENT_TIME

      static final org.apache.nifi.components.AllowableValue START_CURRENT_TIME

    • BASE_DIRECTORY

      static final org.apache.nifi.components.PropertyDescriptor BASE_DIRECTORY

    • MODE

      static final org.apache.nifi.components.PropertyDescriptor MODE

    • FILENAME

      static final org.apache.nifi.components.PropertyDescriptor FILENAME

    • ROLLING_FILENAME_PATTERN

      static final org.apache.nifi.components.PropertyDescriptor ROLLING_FILENAME_PATTERN

    • POST_ROLLOVER_TAIL_PERIOD

      static final org.apache.nifi.components.PropertyDescriptor POST_ROLLOVER_TAIL_PERIOD

    • STATE_LOCATION

      static final org.apache.nifi.components.PropertyDescriptor STATE_LOCATION

    • START_POSITION

      static final org.apache.nifi.components.PropertyDescriptor START_POSITION

    • RECURSIVE

      static final org.apache.nifi.components.PropertyDescriptor RECURSIVE

    • LOOKUP_FREQUENCY

      static final org.apache.nifi.components.PropertyDescriptor LOOKUP_FREQUENCY

    • MAXIMUM_AGE

      static final org.apache.nifi.components.PropertyDescriptor MAXIMUM_AGE

    • REREAD_ON_NUL

      static final org.apache.nifi.components.PropertyDescriptor REREAD_ON_NUL

    • LINE_START_PATTERN

      static final org.apache.nifi.components.PropertyDescriptor LINE_START_PATTERN

    • MAX_BUFFER_LENGTH

      static final org.apache.nifi.components.PropertyDescriptor MAX_BUFFER_LENGTH

    • PRE_ALLOCATED_BUFFER_SIZE

      static final org.apache.nifi.components.PropertyDescriptor PRE_ALLOCATED_BUFFER_SIZE

    • PROPERTY_DESCRIPTORS

      private static final List<org.apache.nifi.components.PropertyDescriptor> PROPERTY_DESCRIPTORS

    • REL_SUCCESS

      static final org.apache.nifi.processor.Relationship REL_SUCCESS

    • RELATIONSHIPS

      private static final Set<org.apache.nifi.processor.Relationship> RELATIONSHIPS

    • states

      private volatile Map<String,TailFile.TailFileObject> states

    • lastLookup

      private final AtomicLong lastLookup

    • isMultiChanging

      private final AtomicBoolean isMultiChanging

    • requireStateLookup

      private volatile boolean requireStateLookup

    • linesBuffer

      private final ByteArrayOutputStream linesBuffer

    • lineStartPattern

      private volatile Pattern lineStartPattern

    • maxBufferBytes

      private volatile long maxBufferBytes

    • preAllocatedBufferSize

      private volatile int preAllocatedBufferSize

  • Constructor Details

    • TailFile

      public TailFile()

  • Method Details

    • getSupportedPropertyDescriptors

      protected List<org.apache.nifi.components.PropertyDescriptor> getSupportedPropertyDescriptors()
      Overrides:
      getSupportedPropertyDescriptors in class org.apache.nifi.components.AbstractConfigurableComponent

    • getRelationships

      public Set<org.apache.nifi.processor.Relationship> getRelationships()
      Specified by:
      getRelationships in interface org.apache.nifi.processor.Processor
      Overrides:
      getRelationships in class org.apache.nifi.processor.AbstractSessionFactoryProcessor

    • onPropertyModified

      public void onPropertyModified(org.apache.nifi.components.PropertyDescriptor descriptor, String oldValue, String newValue)
      Specified by:
      onPropertyModified in interface org.apache.nifi.components.ConfigurableComponent
      Overrides:
      onPropertyModified in class org.apache.nifi.components.AbstractConfigurableComponent

    • customValidate

      protected Collection<org.apache.nifi.components.ValidationResult> customValidate(org.apache.nifi.components.ValidationContext context)
      Overrides:
      customValidate in class org.apache.nifi.components.AbstractConfigurableComponent

    • onPrimaryNodeChange

      @OnPrimaryNodeStateChange public void onPrimaryNodeChange()

    • lookup

      private List<String> lookup(org.apache.nifi.processor.ProcessContext context)

    • compileRegex

      @OnScheduled public void compileRegex(org.apache.nifi.processor.ProcessContext context)

    • recoverState

      @OnScheduled public void recoverState(org.apache.nifi.processor.ProcessContext context) throws IOException
      Throws:
      IOException

    • initStates

      private void initStates(List<String> filesToTail, Map<String,String> statesMap, boolean isCleared)

    • recoverState

      private void recoverState(List<String> filesToTail, Map<String,String> map) throws IOException
      Throws:
      IOException

    • getFilesToTail

      private List<String> getFilesToTail(String baseDir, String fileRegex, boolean isRecursive, long maxAge)
      Method to list the files to tail according to the given base directory and using the user-provided regular expression
      Parameters:
      baseDir - base directory to recursively look into
      fileRegex - expression regular used to match files to tail
      isRecursive - true if looking for file recursively, false otherwise
      Returns:
      List of files to tail

    • recoverState

      private void recoverState(Map<String,String> stateValues, String filePath) throws IOException
      Updates member variables to reflect the "expected recovery checksum" and seek to the appropriate location in the tailed file, updating our checksum, so that we are ready to proceed with the onTrigger(ProcessContext, ProcessSession) call.
      Parameters:
      stateValues - the values that were recovered from state that was previously stored. This Map should be populated with the keys defined in TailFile.TailFileState.StateKeys.
      filePath - the file of the file for which state must be recovered
      Throws:
      IOException - if unable to seek to the appropriate location in the tailed file.

    • resetState

      private void resetState(String filePath)

    • cleanup

      @OnStopped public void cleanup(org.apache.nifi.processor.ProcessContext context)

    • cleanReader

      private void cleanReader(TailFile.TailFileObject tfo)

    • onTrigger

      public void onTrigger(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session) throws org.apache.nifi.processor.exception.ProcessException
      Specified by:
      onTrigger in class org.apache.nifi.processor.AbstractProcessor
      Throws:
      org.apache.nifi.processor.exception.ProcessException

    • collectKeysToBeRemoved

      private List<String> collectKeysToBeRemoved(Map<String,String> sessionStates)

    • processTailFile

      private void processTailFile(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, String tailFile)

    • readLines

      private long readLines(FileChannel reader, ByteBuffer buffer, OutputStream out, Checksum checksum, Boolean reReadOnNul) throws IOException
      Throws:
      IOException

    • readLines

      private long readLines(FileChannel reader, ByteBuffer buffer, OutputStream out, Checksum checksum, Boolean reReadOnNul, boolean readFully) throws IOException
      Read new lines from the given FileChannel, copying it to the given Output Stream. The Checksum is used in order to later determine whether or not data has been consumed.
      Parameters:
      reader - The FileChannel to read data from
      buffer - the buffer to use for copying data
      out - the OutputStream to copy the data to
      checksum - the Checksum object to use in order to calculate checksum for recovery purposes
      reReadOnNul - If set to 'true', ASCII NUL characters will be treated as temporary values and a NulCharacterEncounteredException is thrown. This allows the caller to re-attempt a read from the same position. If set to 'false' these characters will be treated as regular content.
      readFully - If set to 'true' the last chunk of bytes after the last whole line will be also written to the OutputStream
      Returns:
      The new position after the lines have been read
      Throws:
      IOException - if an I/O error occurs.

    • flushByteArrayOutputStream

      private void flushByteArrayOutputStream(ByteArrayOutputStream baos, OutputStream out, Checksum checksum, boolean ignoreRegex) throws IOException
      Throws:
      IOException

    • flushLinesBuffer

      private void flushLinesBuffer(OutputStream out, Checksum checksum) throws IOException
      Throws:
      IOException

    • getRolledOffFiles

      private List<File> getRolledOffFiles(org.apache.nifi.processor.ProcessContext context, long minTimestamp, String tailFilePath) throws IOException
      Returns a list of all Files that match the following criteria:
      • Filename matches the Rolling Filename Pattern
      • Filename does not match the actual file being tailed
      • The Last Modified Time on the file is equal to or later than the given minimum timestamp

      The List that is returned will be ordered by file timestamp, providing the oldest file first.

      Parameters:
      context - the ProcessContext to use in order to determine Processor configuration
      minTimestamp - any file with a Last Modified Time before this timestamp will not be returned
      Returns:
      a list of all Files that have rolled over
      Throws:
      IOException - if unable to perform the listing of files

    • getStateScope

      private org.apache.nifi.components.state.Scope getStateScope(org.apache.nifi.processor.ProcessContext context)

    • persistState

      private void persistState(TailFile.TailFileObject tfo, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.processor.ProcessContext context)

    • persistState

      private void persistState(Map<String,String> state, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.processor.ProcessContext context)

    • createReader

      private FileChannel createReader(File file, long position)

    • getState

      Map<String,TailFile.TailFileObject> getState()

    • recoverRolledFiles

      private boolean recoverRolledFiles(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, String tailFile, Long expectedChecksum, long timestamp, long position)
      Finds any files that have rolled over and have not yet been ingested by this Processor. Each of these files that is found will be ingested as its own FlowFile. If a file is found that has been partially ingested, the rest of the file will be ingested as a single FlowFile but the data that already has been ingested will not be ingested again.
      Parameters:
      context - the ProcessContext to use in order to obtain Processor configuration.
      session - the ProcessSession to use in order to interact with FlowFile creation and content.
      expectedChecksum - the checksum value that is expected for the oldest file from offset 0 through <position>.
      timestamp - the latest Last Modified Timestamp that has been consumed. Any data that was written before this data will not be ingested.
      position - the byte offset in the file being tailed, where tailing last left off.
      Returns:
      true if the file being tailed has rolled over, false otherwise

    • recoverRolledFiles

      private boolean recoverRolledFiles(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, String tailFile, List<File> rolledOffFiles, Long expectedChecksum, long position)
      Finds any files that have rolled over and have not yet been ingested by this Processor. Each of these files that is found will be ingested as its own FlowFile. If a file is found that has been partially ingested, the rest of the file will be ingested as a single FlowFile but the data that already has been ingested will not be ingested again.
      Parameters:
      context - the ProcessContext to use in order to obtain Processor configuration.
      session - the ProcessSession to use in order to interact with FlowFile creation and content.
      expectedChecksum - the checksum value that is expected for the oldest file from offset 0 through <position>.
      position - the byte offset in the file being tailed, where tailing last left off.
      Returns:
      true if the file being tailed has rolled over, false otherwise

    • tailRolledFile

      private boolean tailRolledFile(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, String tailFile, Long expectedChecksum, long position, TailFile.TailFileObject tfo, File fileToTail, boolean readFully, boolean tailingPostRollover) throws IOException
      Throws:
      IOException

    • consumeFileFully

      private TailFile.TailFileState consumeFileFully(File file, org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, TailFile.TailFileObject tfo) throws IOException
      Creates a new FlowFile that contains the entire contents of the given file and transfers that FlowFile to success. This method will commit the given session and emit an appropriate Provenance Event.
      Parameters:
      file - the file to ingest
      context - the ProcessContext
      session - the ProcessSession
      tfo - the current state
      Returns:
      the new, updated state that reflects that the given file has been ingested.
      Throws:
      IOException

    • getCurrentTimeMs

      public long getCurrentTimeMs()