core/src/main/java/org/apache/hadoop/chukwa/datacollection/adaptor/filetailer/LWFTAdaptor.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.IOException;
import java.io.RandomAccessFile;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.hadoop.chukwa.ChunkImpl;
import org.apache.hadoop.chukwa.datacollection.ChunkReceiver;
import org.apache.hadoop.chukwa.datacollection.adaptor.AbstractAdaptor;
import org.apache.hadoop.chukwa.datacollection.adaptor.AdaptorException;
import org.apache.hadoop.chukwa.datacollection.adaptor.AdaptorShutdownPolicy;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;

/**
 * A base class for file tailing adaptors.  
 * Intended to mandate as little policy as possible, and to use as 
 * few system resources as possible.
 * 
 * 
 * If the file does not exist, this class will continue to retry quietly
 * forever and will start tailing if it's eventually created.
 */
public class LWFTAdaptor extends AbstractAdaptor {
  
  /**
   * This is the maximum amount we'll read from any one file before moving on to
   * the next. This way, we get quick response time for other files if one file
   * is growing rapidly.
   * 
   */
  public static final int DEFAULT_MAX_READ_SIZE = 128 * 1024;
  public static final String MAX_READ_SIZE_OPT = 
      "chukwaAgent.fileTailingAdaptor.maxReadSize";

  int MAX_READ_SIZE = DEFAULT_MAX_READ_SIZE;
  
  static Logger log;
  static FileTailer tailer;
  
  static {
    tailer = null;
    log = Logger.getLogger(FileTailingAdaptor.class);
  }
  
  
  /**
   * next PHYSICAL offset to read
   */
  protected long fileReadOffset;

  /**
   * The logical offset of the first byte of the file
   */
  protected long offsetOfFirstByte = 0;
  protected Configuration conf = null;
  /**
   * The timestamp of last slurping.
   */ 
  protected long lastSlurpTime = 0l;

  File toWatch;

  @Override
  public void start(long offset) {
    synchronized(LWFTAdaptor.class) {
      if(tailer == null)
        tailer = new FileTailer(control.getConfiguration());
    }
    this.fileReadOffset = offset - offsetOfFirstByte;    
    tailer.startWatchingFile(this);
  }
  
  /**
   * @see org.apache.hadoop.chukwa.datacollection.adaptor.Adaptor#getCurrentStatus()
   */
  public String getCurrentStatus() {
    return type.trim() + " " + offsetOfFirstByte + " " + toWatch.getPath();
  }

  public String toString() {
    return "Lightweight Tailer on " + toWatch;
  }

  public String getStreamName() {
    return toWatch.getPath();
  }
  
  @Override
  public String parseArgs(String params) { 
    conf = control.getConfiguration();
    MAX_READ_SIZE = conf.getInt(MAX_READ_SIZE_OPT, DEFAULT_MAX_READ_SIZE);

    Pattern cmd = Pattern.compile("(\\d+)\\s+(.+)\\s?");
    Matcher m = cmd.matcher(params);
    if (m.matches()) { //check for first-byte offset. If absent, assume we just got a path.
      offsetOfFirstByte = Long.parseLong(m.group(1));
      toWatch = new File(m.group(2));
    } else {
      toWatch = new File(params.trim());
    }
    return toWatch.getAbsolutePath();
  }

  @Override
  public long shutdown(AdaptorShutdownPolicy shutdownPolicy)
      throws AdaptorException {
    tailer.stopWatchingFile(this);
    return fileReadOffset + offsetOfFirstByte;
  }
  

  /**
   * Extract records from a byte sequence
   * 
   * @param eq the queue to stick the new chunk[s] in
   * @param buffOffsetInFile the byte offset in the stream at which buf[] begins
   * @param buf the byte buffer to extract records from
   * @return the number of bytes processed
   * @throws InterruptedException
   */
  protected int extractRecords(ChunkReceiver eq, long buffOffsetInFile,
      byte[] buf) throws InterruptedException {
    if(buf.length == 0)
      return 0;
    
    ChunkImpl chunk = new ChunkImpl(type, toWatch.getAbsolutePath(),
        buffOffsetInFile + buf.length, buf, this);

    eq.add(chunk);
    return buf.length;
  }
  
  protected boolean slurp(long len, RandomAccessFile reader) throws IOException,
  InterruptedException{
    boolean hasMoreData = false;

    log.debug("Adaptor|" + adaptorID + "|seeking|" + fileReadOffset);
    reader.seek(fileReadOffset);

    long bufSize = len - fileReadOffset;

   if (bufSize > MAX_READ_SIZE) {
      bufSize = MAX_READ_SIZE;
      hasMoreData = true;
    }
    byte[] buf = new byte[(int) bufSize];

    long curOffset = fileReadOffset;

    lastSlurpTime = System.currentTimeMillis();
    int bufferRead = reader.read(buf);
    assert reader.getFilePointer() == fileReadOffset + bufSize : " event size arithmetic is broken: "
        + " pointer is "
        + reader.getFilePointer()
        + " but offset is "
        + fileReadOffset + bufSize;

    int bytesUsed = extractRecords(dest,
        fileReadOffset + offsetOfFirstByte, buf);

    // === WARNING ===
    // If we couldn't found a complete record AND
    // we cannot read more, i.e bufferRead == MAX_READ_SIZE
    // it's because the record is too BIG
    // So log.warn, and drop current buffer so we can keep moving
    // instead of being stopped at that point for ever
    if (bytesUsed == 0 && bufferRead == MAX_READ_SIZE) {
      log.warn("bufferRead == MAX_READ_SIZE AND bytesUsed == 0, dropping current buffer: startOffset="
              + curOffset
              + ", MAX_READ_SIZE="
              + MAX_READ_SIZE
              + ", for "
              + toWatch.getPath());
      bytesUsed = buf.length;
    }

    fileReadOffset = fileReadOffset + bytesUsed;

    log.debug("Adaptor|" + adaptorID + "|start|" + curOffset + "|end|"
        + fileReadOffset);
    return hasMoreData;
  }
  
  public boolean tailFile()
  throws InterruptedException {
    boolean hasMoreData = false;
    try {
      
       //if file doesn't exist, length =0 and we just keep waiting for it.
      //if(!toWatch.exists())
      //  deregisterAndStop(false);
      
      long len = toWatch.length();
      if(len < fileReadOffset) {
        //file shrank; probably some data went missing.
        handleShrunkenFile(len);
      } else if(len > fileReadOffset) {
        RandomAccessFile reader = new RandomAccessFile(toWatch, "r");
        hasMoreData = slurp(len, reader);
        reader.close();
      }
    } catch(IOException e) {
      log.warn("IOException in tailer", e);
      deregisterAndStop();
    }
    
    return hasMoreData;
  }

  private void handleShrunkenFile(long measuredLen) {
    log.info("file "+ toWatch +"shrank from " + fileReadOffset + " to " + measuredLen);
    offsetOfFirstByte = measuredLen;
    fileReadOffset = 0;
  }

}