core/src/main/java/org/apache/hadoop/chukwa/datacollection/adaptor/filetailer/RCheckFTAdaptor.java

/*
 * 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.hadoop.chukwa.datacollection.adaptor.filetailer;

import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.Collections;
import java.util.LinkedList;

import org.apache.commons.lang3.builder.HashCodeBuilder;

/**
 * Checkpoint state:
 *   date modified of most-recently tailed file, offset of first byte of that file,
 *   then regular FTA arts 
 *
 */
public class RCheckFTAdaptor extends LWFTAdaptor implements FileFilter {
  
  private static class FPair implements Comparable<FPair> {
    File f;
    long mod;
    FPair(File f) {
      this.f = f;
      mod = f.lastModified();
    }
    /**
     * -1   implies this is LESS THAN o
     */
    @Override
    public int compareTo(FPair o) {
      if(mod < o.mod)
        return -1;
      else if (mod > o.mod)
        return 1;
      //want toWatch to be last after a rotation; otherwise, this is basically 
      //just a heuristic that hasn't been tuned yet
      else return (o.f.getName().compareTo(f.getName()));
    }
    
    @Override
    public boolean equals(Object o) {
      if(o instanceof FPair) {
        return mod == ((FPair) o).mod;
      } else {
        return false;
      }
    }
    
    @Override
    public int hashCode() {
      return new HashCodeBuilder(643, 1321).
          append(this.mod).
          toHashCode();
    }
  }
  
  long prevFileLastModDate = 0;
  LinkedList<FPair> fileQ = new LinkedList<FPair>(); 
  String fBaseName;
  File cur; //this is the actual physical file being watched.  
            // in contrast, toWatch is the path specified by the user
  boolean caughtUp = false;
  /**
   * Check for date-modified and offset; if absent assume we just got a name.
   */
  @Override
  public String parseArgs(String params) { 
    Pattern cmd = Pattern.compile("d:(\\d+)\\s+(\\d+)\\s+(.+)\\s?");
    Matcher m = cmd.matcher(params);
    if (m.matches()) {
      prevFileLastModDate = Long.parseLong(m.group(1));
      offsetOfFirstByte = Long.parseLong(m.group(2));
      toWatch = new File(m.group(3)).getAbsoluteFile();
    } else {
      toWatch = new File(params.trim()).getAbsoluteFile();
    }
    fBaseName = toWatch.getName();
    return toWatch.getAbsolutePath();
  }
  
  public String getCurrentStatus() {
    return type.trim() + " d:" + prevFileLastModDate + " "  + offsetOfFirstByte + " " + toWatch.getPath();
  }

  @Override
  public boolean accept(File pathname) {
    return pathname.getName().startsWith(fBaseName) &&
     ( pathname.getName().equals(fBaseName) ||
       pathname.lastModified() > prevFileLastModDate);
  }
  
  
  protected void mkFileQ() {
    
    File toWatchDir = toWatch.getParentFile();
    File[] candidates = toWatchDir.listFiles(this);
    if(candidates == null) {
      log.error(toWatchDir + " is not a directory in "+adaptorID);
    } else {
      log.debug("saw " + candidates.length + " files matching pattern");
      fileQ = new LinkedList<FPair>();
      for(File f:candidates)
        fileQ.add(new FPair(f));
      Collections.sort(fileQ);
    } 
  }
  
  protected void advanceQ() {
    FPair next = fileQ.poll();
    if(next != null) {
      cur = next.f;
      caughtUp = toWatch.equals(cur);
      if(caughtUp && !fileQ.isEmpty()) 
        log.warn("expected rotation queue to be empty when caught up...");
    }
    else {
      cur = null;
      caughtUp = true;
    }
  }
  
  @Override
  public void start(long offset) {
    mkFileQ(); //figure out what to watch
    advanceQ();
    super.start(offset);
  }
  
  @Override
  public boolean tailFile()
  throws InterruptedException {
    boolean hasMoreData = false;
    try {
      
      if(caughtUp) {
        //we're caught up and watching an unrotated file
        mkFileQ(); //figure out what to watch
        advanceQ();
      }
      if(cur == null) //file we're watching doesn't exist
        return false;
      
      
      long len = cur.length();
      long tsPreTail = cur.exists() ? cur.lastModified() : prevFileLastModDate;

      if(log.isDebugEnabled())
        log.debug(adaptorID + " treating " + cur + " as " + toWatch + " len = " + len);
      
      if(len < fileReadOffset) {
        log.info("file "+ cur +" shrank from " + fileReadOffset + " to " + len);
        //no unseen changes to prev version, since mod date is older than last scan.
        offsetOfFirstByte += fileReadOffset;
        fileReadOffset = 0;
      } else if(len > fileReadOffset) {
        log.debug("slurping from " + cur+ " at offset " + fileReadOffset);
        RandomAccessFile reader = new RandomAccessFile(cur, "r");
        slurp(len, reader);
        reader.close();
      } else {
        //we're either caught up or at EOF
        if (!caughtUp) {
          prevFileLastModDate = cur.lastModified();
          //Hit EOF on an already-rotated file. Move on!
          offsetOfFirstByte += fileReadOffset;
          fileReadOffset = 0;
          advanceQ();
          log.debug("not caught up, and hit EOF.  Moving forward in queue to " + cur);
        } else 
          prevFileLastModDate = tsPreTail;

      }
        
    } catch(IOException e) {
      log.warn("IOException in "+adaptorID, e);
      deregisterAndStop();
    }
    
    return hasMoreData;
  }
  

  public String toString() {
    return "Rotation-aware Tailer on " + toWatch;
  }
}