src/main/java/org/apache/hadoop/chukwa/datacollection/adaptor/FileAdaptor.java - chukwa - Git at Google

 /*
  * 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;

 import java.io.File;
 import java.io.RandomAccessFile;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;

 import org.apache.hadoop.chukwa.ChunkImpl;
 import org.apache.hadoop.chukwa.datacollection.ChunkReceiver;
 import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.chukwa.util.ExceptionUtil;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;


 class FileAdaptorTailer extends Thread {
   static Logger log = Logger.getLogger(FileAdaptorTailer.class);
   private List<FileAdaptor> adaptors = null;

   /**
    * How often to call each adaptor.
    */
   int DEFAULT_SAMPLE_PERIOD_MS = 1000 * 10;
   int SAMPLE_PERIOD_MS = DEFAULT_SAMPLE_PERIOD_MS;


   public FileAdaptorTailer() {
     ChukwaAgent agent = ChukwaAgent.getAgent();
     if (agent != null) {
       Configuration conf = agent.getConfiguration();
       if (conf != null) {
         SAMPLE_PERIOD_MS = conf.getInt(
             "chukwaAgent.adaptor.context.switch.time",
             DEFAULT_SAMPLE_PERIOD_MS);
       }
     }

     // iterations are much more common than adding a new adaptor
     adaptors = new CopyOnWriteArrayList<FileAdaptor>();

     setDaemon(true);
     start();// start the FileAdaptorTailer thread
   }
   @Override
   public void run() {
     while(true) {
       try {

         long startTime = System.currentTimeMillis();
         for (FileAdaptor adaptor: adaptors) {
           log.info("calling sendFile for " + adaptor.toWatch.getCanonicalPath());
           adaptor.sendFile();
         }

         long timeToReadFiles = System.currentTimeMillis() - startTime;
         if (timeToReadFiles < SAMPLE_PERIOD_MS) {
           Thread.sleep(SAMPLE_PERIOD_MS);
         }

       }catch (Throwable e) {
         log.warn("Exception in FileAdaptorTailer:",e);
       }
     }
   }

   public void addFileAdaptor(FileAdaptor adaptor) {
     adaptors.add(adaptor);
   }

   public void removeFileAdaptor(FileAdaptor adaptor) {
     adaptors.remove(adaptor);
   }
 }

 /**
  * File Adaptor push small size file in one chunk to collector
  */
 public class FileAdaptor extends AbstractAdaptor {

   static Logger log = Logger.getLogger(FileAdaptor.class);
   static FileAdaptorTailer tailer = null;

   static final int DEFAULT_TIMEOUT_PERIOD = 5*60*1000;
   int TIMEOUT_PERIOD = DEFAULT_TIMEOUT_PERIOD;

   static {
     tailer = new FileAdaptorTailer();
   }

   private long startTime = 0;
   private long timeOut = 0;

   protected volatile boolean finished = false;
   File toWatch;
   protected RandomAccessFile reader = null;
   protected long fileReadOffset;
   protected boolean deleteFileOnClose = false;
   protected boolean shutdownCalled = false;

   /**
    * The logical offset of the first byte of the file
    */
   private long offsetOfFirstByte = 0;

   public void start(long bytes) {
     // in this case params = filename
     log.info("adaptor id: " + adaptorID + " started file adaptor on file "
         + toWatch);
     this.startTime = System.currentTimeMillis();
     TIMEOUT_PERIOD = control.getConfiguration().getInt(
         "chukwaAgent.adaptor.fileadaptor.timeoutperiod",
         DEFAULT_TIMEOUT_PERIOD);
     this.timeOut = startTime + TIMEOUT_PERIOD;

     tailer.addFileAdaptor(this);
   }

   void sendFile() {
     long now = System.currentTimeMillis() ;
     long oneMinAgo = now - (60*1000);
     if (toWatch.exists()) {
      if (toWatch.lastModified() > oneMinAgo && now < timeOut) {
        log.info("Last modified time less than one minute, keep waiting");
        return;
      } else {
        try {

          long bufSize = toWatch.length();
          byte[] buf = new byte[(int) bufSize];

          reader = new RandomAccessFile(toWatch, "r");
          reader.read(buf);
          reader.close();
          reader = null;

          long fileTime = toWatch.lastModified();
          int bytesUsed = extractRecords(dest, 0, buf, fileTime);
          this.fileReadOffset = bytesUsed;
          finished = true;
          deregisterAndStop();
          cleanUp();
        } catch(Exception e) {
          log.warn("Exception while trying to read: " + toWatch.getAbsolutePath(),e);
        }  finally {
          if (reader != null) {
            try {
              reader.close();
            } catch (Exception e) {
              log.debug(ExceptionUtil.getStackTrace(e));
           }
            reader = null;
          }
        }
      }
     } else {
       if (now > timeOut) {
         finished = true;
         log.warn("Couldn't read this file: " + toWatch.getAbsolutePath());
         deregisterAndStop();
         cleanUp() ;
       }
     }
   }

   private void cleanUp() {
     tailer.removeFileAdaptor(this);
     if (reader != null) {
       try {
         reader.close();
       } catch (Exception e) {
         log.debug(ExceptionUtil.getStackTrace(e));
      }
       reader = null;
     }
   }


   @Override
   public long shutdown(AdaptorShutdownPolicy shutdownPolicy) {
     log.info("Enter Shutdown:" + shutdownPolicy.name()+ " - ObjectId:" + this);
     switch(shutdownPolicy) {
       case GRACEFULLY : {
         int retry = 0;
         while (!finished && retry < 60) {
           try {
             log.info("GRACEFULLY Retry:" + retry);
             Thread.sleep(1000);
             retry++;
           } catch (InterruptedException ex) {
           }
         }
       }
       break;
       case WAIT_TILL_FINISHED : {
         int retry = 0;
         while (!finished) {
           try {
             if (retry%100 == 0) {
               log.info("WAIT_TILL_FINISHED Retry:" + retry);
             }

             Thread.sleep(1000);
             retry++;
           } catch (InterruptedException ex) {
           }
         }
       }

       break;
       default :
         cleanUp();
         break;
     }

     if (deleteFileOnClose && toWatch != null) {
       if (log.isDebugEnabled()) {
         log.debug("About to delete " + toWatch.getAbsolutePath());
       }
       if (toWatch.delete()) {
         if (log.isInfoEnabled()) {
           log.debug("Successfully deleted " + toWatch.getAbsolutePath());
         }
       } else {
         if (log.isEnabledFor(Level.WARN)) {
           log.warn("Could not delete " + toWatch.getAbsolutePath() + " (for unknown reason)");
         }
       }
     }

     log.info("Exist Shutdown:" + shutdownPolicy.name()+ " - ObjectId:" + this);
     return fileReadOffset + offsetOfFirstByte;
   }

   public String parseArgs(String params) {

     String[] words = params.split(" ");
     if (words.length == 2) {
       if (words[1].equals("delete")) {
         deleteFileOnClose = true;
         toWatch = new File(words[0]);
       } else {
         offsetOfFirstByte = Long.parseLong(words[0]);
         toWatch = new File(words[1]);
       }
     } else if (words.length == 3) {
       offsetOfFirstByte = Long.parseLong(words[0]);
       toWatch = new File(words[1]);
       deleteFileOnClose = words[2].equals("delete");
     } else {
       toWatch = new File(params);
     }
     return toWatch.getAbsolutePath();
   }

   /**
    * 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(final ChunkReceiver eq, long buffOffsetInFile,
       byte[] buf, long fileTime) throws InterruptedException {
     final ChunkImpl chunk = new ChunkImpl(type, toWatch.getAbsolutePath(),
         buffOffsetInFile + buf.length, buf, this);
     chunk.addTag("time=\"" + fileTime + "\"");
     log.info("Adding " + toWatch.getAbsolutePath() + " to the queue");
     eq.add(chunk);
     log.info( toWatch.getAbsolutePath() + " added to the queue");
     return buf.length;
   }

   @Override
   public String getCurrentStatus() {
     return type.trim() + " " + offsetOfFirstByte + " " + toWatch.getPath();
   }

 }
	/*
	* 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;

	import java.io.File;
	import java.io.RandomAccessFile;
	import java.util.List;
	import java.util.concurrent.CopyOnWriteArrayList;

	import org.apache.hadoop.chukwa.ChunkImpl;
	import org.apache.hadoop.chukwa.datacollection.ChunkReceiver;
	import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.chukwa.util.ExceptionUtil;
	import org.apache.log4j.Level;
	import org.apache.log4j.Logger;


	class FileAdaptorTailer extends Thread {
	static Logger log = Logger.getLogger(FileAdaptorTailer.class);
	private List<FileAdaptor> adaptors = null;

	/**
	* How often to call each adaptor.
	*/
	int DEFAULT_SAMPLE_PERIOD_MS = 1000 * 10;
	int SAMPLE_PERIOD_MS = DEFAULT_SAMPLE_PERIOD_MS;


	public FileAdaptorTailer() {
	ChukwaAgent agent = ChukwaAgent.getAgent();
	if (agent != null) {
	Configuration conf = agent.getConfiguration();
	if (conf != null) {
	SAMPLE_PERIOD_MS = conf.getInt(
	"chukwaAgent.adaptor.context.switch.time",
	DEFAULT_SAMPLE_PERIOD_MS);
	}
	}

	// iterations are much more common than adding a new adaptor
	adaptors = new CopyOnWriteArrayList<FileAdaptor>();

	setDaemon(true);
	start();// start the FileAdaptorTailer thread
	}
	@Override
	public void run() {
	while(true) {
	try {

	long startTime = System.currentTimeMillis();
	for (FileAdaptor adaptor: adaptors) {
	log.info("calling sendFile for " + adaptor.toWatch.getCanonicalPath());
	adaptor.sendFile();
	}

	long timeToReadFiles = System.currentTimeMillis() - startTime;
	if (timeToReadFiles < SAMPLE_PERIOD_MS) {
	Thread.sleep(SAMPLE_PERIOD_MS);
	}

	}catch (Throwable e) {
	log.warn("Exception in FileAdaptorTailer:",e);
	}
	}
	}

	public void addFileAdaptor(FileAdaptor adaptor) {
	adaptors.add(adaptor);
	}

	public void removeFileAdaptor(FileAdaptor adaptor) {
	adaptors.remove(adaptor);
	}
	}

	/**
	* File Adaptor push small size file in one chunk to collector
	*/
	public class FileAdaptor extends AbstractAdaptor {

	static Logger log = Logger.getLogger(FileAdaptor.class);
	static FileAdaptorTailer tailer = null;

	static final int DEFAULT_TIMEOUT_PERIOD = 5601000;
	int TIMEOUT_PERIOD = DEFAULT_TIMEOUT_PERIOD;

	static {
	tailer = new FileAdaptorTailer();
	}

	private long startTime = 0;
	private long timeOut = 0;

	protected volatile boolean finished = false;
	File toWatch;
	protected RandomAccessFile reader = null;
	protected long fileReadOffset;
	protected boolean deleteFileOnClose = false;
	protected boolean shutdownCalled = false;

	/**
	* The logical offset of the first byte of the file
	*/
	private long offsetOfFirstByte = 0;

	public void start(long bytes) {
	// in this case params = filename
	log.info("adaptor id: " + adaptorID + " started file adaptor on file "
	+ toWatch);
	this.startTime = System.currentTimeMillis();
	TIMEOUT_PERIOD = control.getConfiguration().getInt(
	"chukwaAgent.adaptor.fileadaptor.timeoutperiod",
	DEFAULT_TIMEOUT_PERIOD);
	this.timeOut = startTime + TIMEOUT_PERIOD;

	tailer.addFileAdaptor(this);
	}

	void sendFile() {
	long now = System.currentTimeMillis() ;
	long oneMinAgo = now - (60*1000);
	if (toWatch.exists()) {
	if (toWatch.lastModified() > oneMinAgo && now < timeOut) {
	log.info("Last modified time less than one minute, keep waiting");
	return;
	} else {
	try {

	long bufSize = toWatch.length();
	byte[] buf = new byte[(int) bufSize];

	reader = new RandomAccessFile(toWatch, "r");
	reader.read(buf);
	reader.close();
	reader = null;

	long fileTime = toWatch.lastModified();
	int bytesUsed = extractRecords(dest, 0, buf, fileTime);
	this.fileReadOffset = bytesUsed;
	finished = true;
	deregisterAndStop();
	cleanUp();
	} catch(Exception e) {
	log.warn("Exception while trying to read: " + toWatch.getAbsolutePath(),e);
	} finally {
	if (reader != null) {
	try {
	reader.close();
	} catch (Exception e) {
	log.debug(ExceptionUtil.getStackTrace(e));
	}
	reader = null;
	}
	}
	}
	} else {
	if (now > timeOut) {
	finished = true;
	log.warn("Couldn't read this file: " + toWatch.getAbsolutePath());
	deregisterAndStop();
	cleanUp() ;
	}
	}
	}

	private void cleanUp() {
	tailer.removeFileAdaptor(this);
	if (reader != null) {
	try {
	reader.close();
	} catch (Exception e) {
	log.debug(ExceptionUtil.getStackTrace(e));
	}
	reader = null;
	}
	}


	@Override
	public long shutdown(AdaptorShutdownPolicy shutdownPolicy) {
	log.info("Enter Shutdown:" + shutdownPolicy.name()+ " - ObjectId:" + this);
	switch(shutdownPolicy) {
	case GRACEFULLY : {
	int retry = 0;
	while (!finished && retry < 60) {
	try {
	log.info("GRACEFULLY Retry:" + retry);
	Thread.sleep(1000);
	retry++;
	} catch (InterruptedException ex) {
	}
	}
	}
	break;
	case WAIT_TILL_FINISHED : {
	int retry = 0;
	while (!finished) {
	try {
	if (retry%100 == 0) {
	log.info("WAIT_TILL_FINISHED Retry:" + retry);
	}

	Thread.sleep(1000);
	retry++;
	} catch (InterruptedException ex) {
	}
	}
	}

	break;
	default :
	cleanUp();
	break;
	}

	if (deleteFileOnClose && toWatch != null) {
	if (log.isDebugEnabled()) {
	log.debug("About to delete " + toWatch.getAbsolutePath());
	}
	if (toWatch.delete()) {
	if (log.isInfoEnabled()) {
	log.debug("Successfully deleted " + toWatch.getAbsolutePath());
	}
	} else {
	if (log.isEnabledFor(Level.WARN)) {
	log.warn("Could not delete " + toWatch.getAbsolutePath() + " (for unknown reason)");
	}
	}
	}

	log.info("Exist Shutdown:" + shutdownPolicy.name()+ " - ObjectId:" + this);
	return fileReadOffset + offsetOfFirstByte;
	}

	public String parseArgs(String params) {

	String[] words = params.split(" ");
	if (words.length == 2) {
	if (words[1].equals("delete")) {
	deleteFileOnClose = true;
	toWatch = new File(words[0]);
	} else {
	offsetOfFirstByte = Long.parseLong(words[0]);
	toWatch = new File(words[1]);
	}
	} else if (words.length == 3) {
	offsetOfFirstByte = Long.parseLong(words[0]);
	toWatch = new File(words[1]);
	deleteFileOnClose = words[2].equals("delete");
	} else {
	toWatch = new File(params);
	}
	return toWatch.getAbsolutePath();
	}

	/**
	* 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(final ChunkReceiver eq, long buffOffsetInFile,
	byte[] buf, long fileTime) throws InterruptedException {
	final ChunkImpl chunk = new ChunkImpl(type, toWatch.getAbsolutePath(),
	buffOffsetInFile + buf.length, buf, this);
	chunk.addTag("time=\"" + fileTime + "\"");
	log.info("Adding " + toWatch.getAbsolutePath() + " to the queue");
	eq.add(chunk);
	log.info( toWatch.getAbsolutePath() + " added to the queue");
	return buf.length;
	}

	@Override
	public String getCurrentStatus() {
	return type.trim() + " " + offsetOfFirstByte + " " + toWatch.getPath();
	}

	}