core/src/main/java/org/apache/hadoop/chukwa/extraction/demux/DemuxManager.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.extraction.demux;

 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.text.SimpleDateFormat;
 import java.util.Date;

 import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
 import org.apache.hadoop.chukwa.extraction.CHUKWA_CONSTANT;
 import org.apache.hadoop.chukwa.util.NagiosHelper;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.util.ToolRunner;
 import org.apache.log4j.Logger;

 public class DemuxManager implements CHUKWA_CONSTANT {
   static Logger log = Logger.getLogger(DemuxManager.class);

   int globalErrorcounter = 0;
   Date firstErrorTime = null;

   protected int ERROR_SLEEP_TIME = 60;
   protected int NO_DATASINK_SLEEP_TIME = 20;

   protected int DEFAULT_MAX_ERROR_COUNT = 6;
   protected int DEFAULT_MAX_FILES_PER_DEMUX = 500;
   protected int DEFAULT_REDUCER_COUNT = 8;

   protected int maxPermittedErrorCount = DEFAULT_MAX_ERROR_COUNT;
   protected int demuxReducerCount = 0;
   protected ChukwaConfiguration conf = null;
   protected FileSystem fs = null;
   protected int reprocess = 0;
   protected boolean sendAlert = true;

   protected SimpleDateFormat dayTextFormat = new java.text.SimpleDateFormat("yyyyMMdd");
   protected volatile boolean isRunning = true;

   final private static PathFilter DATA_SINK_FILTER = new PathFilter() {
     public boolean accept(Path file) {
       return file.getName().endsWith(".done");
     }
   };


   public static void main(String[] args) throws Exception {

     DemuxManager manager = new DemuxManager();
     manager.start();

   }

   public DemuxManager() throws Exception {
     this.conf = new ChukwaConfiguration();
     init();
   }

   public DemuxManager(ChukwaConfiguration conf) throws Exception {
     this.conf = conf;
     init();
   }

   protected void init() throws IOException, URISyntaxException {
     String fsName = conf.get(HDFS_DEFAULT_NAME_FIELD);
     fs = FileSystem.get(new URI(fsName), conf);
   }

   public void shutdown() {
     this.isRunning = false;
   }


   public int getReprocess() {
     return reprocess;
   }

   /**
    * Start the Demux Manager daemon
    * @throws Exception if error in processing data
    */
   public void start() throws Exception {

      String chukwaRootDir = conf.get(CHUKWA_ROOT_DIR_FIELD, DEFAULT_CHUKWA_ROOT_DIR_NAME);
      if ( ! chukwaRootDir.endsWith("/") ) {
        chukwaRootDir += "/";
      }
      log.info("chukwaRootDir:" + chukwaRootDir);

      String demuxRootDir = chukwaRootDir + DEFAULT_DEMUX_PROCESSING_DIR_NAME;
      String demuxErrorDir = demuxRootDir + DEFAULT_DEMUX_IN_ERROR_DIR_NAME;
      String demuxInputDir = demuxRootDir + DEFAULT_DEMUX_MR_INPUT_DIR_NAME;
      String demuxOutputDir = demuxRootDir + DEFAULT_DEMUX_MR_OUTPUT_DIR_NAME;

      String dataSinkDir = conf.get(CHUKWA_DATA_SINK_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_LOGS_DIR_NAME);
      if ( ! dataSinkDir.endsWith("/") ) {
        dataSinkDir += "/";
      }
      log.info("dataSinkDir:" + dataSinkDir);

      String postProcessDir = conf.get(CHUKWA_POST_PROCESS_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_POSTPROCESS_DIR_NAME);
      if ( ! postProcessDir.endsWith("/") ) {
        postProcessDir += "/";
      }
      log.info("postProcessDir:" + postProcessDir);

      String archiveRootDir = conf.get(CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_DATASINK_DIR_NAME);
      if ( ! archiveRootDir.endsWith("/") ) {
        archiveRootDir += "/";
      }
      log.info("archiveRootDir:" + archiveRootDir);

      maxPermittedErrorCount = conf.getInt(CHUKWA_DEMUX_MAX_ERROR_COUNT_FIELD,
                                           DEFAULT_MAX_ERROR_COUNT);
      demuxReducerCount = conf.getInt(CHUKWA_DEMUX_REDUCER_COUNT_FIELD, DEFAULT_REDUCER_COUNT);
      log.info("demuxReducerCount:" + demuxReducerCount);

      String nagiosHost = conf.get(CHUKWA_NAGIOS_HOST_FIELD);
      int nagiosPort = conf.getInt(CHUKWA_NAGIOS_PORT_FIELD,0);
      String reportingHost = conf.get(CHUKWA_REPORTING_HOST_FIELD);

      log.info("Nagios information: nagiosHost:" + nagiosHost + ", nagiosPort:"
          + nagiosPort + ", reportingHost:" + reportingHost);


      if (nagiosHost == null || nagiosHost.length() == 0 || nagiosPort == 0 || reportingHost == null || reportingHost.length() == 0) {
        sendAlert = false;
        log.warn("Alerting is OFF");
      }

      boolean demuxReady = false;


      while (isRunning) {
        try {
          demuxReady = false;

          if (maxPermittedErrorCount != -1 && globalErrorcounter >= maxPermittedErrorCount) {
            log.warn("==================\nToo many errors (" + globalErrorcounter +
                     "), Bail out!\n==================");
            break;
          }

          // Check for anomalies
          if (checkDemuxOutputDir(demuxOutputDir) == true) {
            // delete current demux output dir
            if ( deleteDemuxOutputDir(demuxOutputDir) == false ) {
              log.warn("Cannot delete an existing demux output directory!");
              throw new IOException("Cannot move demuxOutput to postProcess!");
            }
            continue;
          } else if (checkDemuxInputDir(demuxInputDir) == true) { // dataSink already there
            reprocess++;

            // Data has been processed more than 3 times ... move to InError directory
            if (reprocess > 3) {
              if (moveDataSinkFilesToDemuxErrorDirectory(demuxInputDir,demuxErrorDir) == false) {
                log.warn("Cannot move dataSink files to DemuxErrorDir!");
                throw new IOException("Cannot move dataSink files to DemuxErrorDir!");
              }
              reprocess = 0;
              continue;
            }

            log.error("Demux inputDir aready contains some dataSink files,"
                + " going to reprocess, reprocessCount=" + reprocess);
            demuxReady = true;
          } else { // standard code path
            reprocess = 0;
            // Move new dataSink Files
            if (moveDataSinkFilesToDemuxInputDirectory(dataSinkDir, demuxInputDir) == true) {
              demuxReady = true; // if any are available
            } else {
              demuxReady = false; // if none
            }
          }

          // start a new demux ?
          if (demuxReady == true) {
           boolean demuxStatus = processData(dataSinkDir, demuxInputDir, demuxOutputDir,
                postProcessDir, archiveRootDir);
           sendDemuxStatusToNagios(nagiosHost,nagiosPort,reportingHost,demuxErrorDir,demuxStatus,null);

           // if demux suceeds, then we reset these.
           if (demuxStatus) {
            globalErrorcounter = 0;
            firstErrorTime = null;
           }
          } else {
            log.info("Demux not ready so going to sleep ...");
            Thread.sleep(NO_DATASINK_SLEEP_TIME * 1000);
          }
        }catch(Throwable e) {
          globalErrorcounter ++;
          if (firstErrorTime == null) firstErrorTime = new Date();

          log.warn("Consecutive error number " + globalErrorcounter +
                   " encountered since " + firstErrorTime, e);
          sendDemuxStatusToNagios(nagiosHost,nagiosPort,reportingHost,demuxErrorDir,false, e.getMessage());
          try { Thread.sleep(ERROR_SLEEP_TIME * 1000); }
          catch (InterruptedException e1) {/*do nothing*/ }
          init();
        }
      }
    }


    /**
     * Send NSCA status to Nagios
     * @param nagiosHost
     * @param nagiosPort
     * @param reportingHost
     * @param demuxInErrorDir
     * @param demuxStatus
     * @param exception
     */
   protected void sendDemuxStatusToNagios(String nagiosHost,int nagiosPort,String reportingHost,
         String demuxInErrorDir,boolean demuxStatus,String demuxException) {

      if (sendAlert == false) {
        return;
      }

      boolean demuxInErrorStatus = true;
      String demuxInErrorMsg = "";
      try {
        Path pDemuxInErrorDir = new Path(demuxInErrorDir);
        if ( fs.exists(pDemuxInErrorDir)) {
          FileStatus[] demuxInErrorDirs = fs.listStatus(pDemuxInErrorDir);
          if (demuxInErrorDirs.length == 0) {
            demuxInErrorStatus = false;
          }
        }
      } catch (Throwable e) {
        demuxInErrorMsg = e.getMessage();
        log.warn(e);
      }

      // send Demux status
      if (demuxStatus == true) {
        NagiosHelper.sendNsca("Demux OK",NagiosHelper.NAGIOS_OK);
      } else {
        NagiosHelper.sendNsca("Demux failed. " + demuxException,NagiosHelper.NAGIOS_CRITICAL);
      }

      // send DemuxInErrorStatus
      if (demuxInErrorStatus == false) {
        NagiosHelper.sendNsca("DemuxInError OK",NagiosHelper.NAGIOS_OK);
      } else {
        NagiosHelper.sendNsca("DemuxInError not empty -" + demuxInErrorMsg,NagiosHelper.NAGIOS_CRITICAL);
      }

    }

    /**
     * Process Data, i.e.
     * - run demux
     * - move demux output to postProcessDir
     * - move dataSink file to archiveDir
     *
     * @param dataSinkDir
     * @param demuxInputDir
     * @param demuxOutputDir
     * @param postProcessDir
     * @param archiveDir
     * @return True iff succeed
     * @throws IOException
     */
     protected boolean processData(String dataSinkDir, String demuxInputDir,
        String demuxOutputDir, String postProcessDir, String archiveDir) throws IOException {

      boolean demuxStatus = false;

      long startTime = System.currentTimeMillis();
      demuxStatus = runDemux(demuxInputDir, demuxOutputDir);
      log.info("Demux Duration: " + (System.currentTimeMillis() - startTime));

      if (demuxStatus == false) {
        log.warn("Demux failed!");
      } else {

        // Move demux output to postProcessDir
        if (checkDemuxOutputDir(demuxOutputDir)) {
          if (moveDemuxOutputDirToPostProcessDirectory(demuxOutputDir, postProcessDir) == false) {
            log.warn("Cannot move demuxOutput to postProcess! bail out!");
            throw new IOException("Cannot move demuxOutput to postProcess! bail out!");
          }
        } else {
          log.warn("Demux processing OK but no output");
        }

        // Move DataSink Files to archiveDir
        if (moveDataSinkFilesToArchiveDirectory(demuxInputDir, archiveDir) == false) {
          log.warn("Cannot move datasinkFile to archive! bail out!");
          throw new IOException("Cannot move datasinkFile to archive! bail out!");
        }
      }

      return demuxStatus;
    }


    /**
     * Submit and Run demux Job
     * @param demuxInputDir
     * @param demuxOutputDir
     * @return true id Demux succeed
     */
    protected boolean runDemux(String demuxInputDir, String demuxOutputDir) {
      // to reload the configuration, and demux's reduce number
      Configuration tempConf = new Configuration(conf);
      tempConf.reloadConfiguration();
      demuxReducerCount = tempConf.getInt(CHUKWA_DEMUX_REDUCER_COUNT_FIELD, DEFAULT_REDUCER_COUNT);
      String[] demuxParams;
      int i=0;
      Demux.addParsers(tempConf);
      demuxParams = new String[4];
      demuxParams[i++] = "-r";
      demuxParams[i++] = "" + demuxReducerCount;
      demuxParams[i++] = demuxInputDir;
      demuxParams[i++] = demuxOutputDir;
      try {
        return ( 0 == ToolRunner.run(tempConf,new Demux(), demuxParams) );
      } catch (Throwable e) {
        e.printStackTrace();
        globalErrorcounter ++;
        if (firstErrorTime == null) firstErrorTime = new Date();
        log.error("Failed to run demux. Consecutive error number " +
                globalErrorcounter + " encountered since " + firstErrorTime, e);
      }
      return false;
    }


    /**
     * Move dataSink files to Demux input directory
     * @param dataSinkDir
     * @param demuxInputDir
     * @return true if there's any dataSink files ready to be processed
     * @throws IOException
     */
    protected boolean moveDataSinkFilesToDemuxInputDirectory(
        String dataSinkDir, String demuxInputDir) throws IOException {
      Path pDataSinkDir = new Path(dataSinkDir);
      Path pDemuxInputDir = new Path(demuxInputDir);
      log.info("dataSinkDir: " + dataSinkDir);
      log.info("demuxInputDir: " + demuxInputDir);


      boolean containsFile = false;

      FileStatus[] dataSinkFiles = fs.listStatus(pDataSinkDir,DATA_SINK_FILTER);
      if (dataSinkFiles.length > 0) {
        setup(pDemuxInputDir);
      }

      int maxFilesPerDemux = 0;
      for (FileStatus fstatus : dataSinkFiles) {
        boolean rename = fs.rename(fstatus.getPath(),pDemuxInputDir);
        log.info("Moving " + fstatus.getPath() + " to " + pDemuxInputDir +", status is:" + rename);
        maxFilesPerDemux ++;
        containsFile = true;
        if (maxFilesPerDemux >= DEFAULT_MAX_FILES_PER_DEMUX) {
          log.info("Max File per Demux reached:" + maxFilesPerDemux);
          break;
        }
      }
      return containsFile;
    }


    /**
     * Move sourceFolder inside destFolder
     * @param dataSinkDir : ex chukwa/demux/inputDir
     * @param demuxErrorDir : ex /chukwa/demux/inError
     * @return true if able to move chukwa/demux/inputDir to /chukwa/demux/inError/<YYYYMMDD>/demuxInputDirXXX
     * @throws IOException
     */
    protected boolean moveDataSinkFilesToDemuxErrorDirectory(
        String dataSinkDir, String demuxErrorDir) throws IOException {
      demuxErrorDir += "/" + dayTextFormat.format(System.currentTimeMillis());
      return moveFolder(dataSinkDir,demuxErrorDir,"demuxInputDir");
    }

    /**
     * Move sourceFolder inside destFolder
     * @param demuxInputDir: ex chukwa/demux/inputDir
     * @param archiveDirectory: ex /chukwa/archives
     * @return true if able to move chukwa/demux/inputDir to /chukwa/archives/raw/<YYYYMMDD>/dataSinkDirXXX
     * @throws IOException
     */
    protected boolean moveDataSinkFilesToArchiveDirectory(
        String demuxInputDir, String archiveDirectory) throws IOException {
      archiveDirectory += "/" + dayTextFormat.format(System.currentTimeMillis());
      return moveFolder(demuxInputDir,archiveDirectory,"dataSinkDir");
    }

    /**
     * Move sourceFolder inside destFolder
     * @param demuxOutputDir: ex chukwa/demux/outputDir
     * @param postProcessDirectory: ex /chukwa/postProcess
     * @return true if able to move chukwa/demux/outputDir to /chukwa/postProcess/demuxOutputDirXXX
     * @throws IOException
     */
    protected  boolean moveDemuxOutputDirToPostProcessDirectory(
        String demuxOutputDir, String postProcessDirectory) throws IOException {
      return moveFolder(demuxOutputDir,postProcessDirectory,"demuxOutputDir");
    }


    /**
     * Test if demuxInputDir exists
     * @param demuxInputDir
     * @return true if demuxInputDir exists
     * @throws IOException
     */
    protected boolean checkDemuxInputDir(String demuxInputDir)
    throws IOException {
      return dirExists(demuxInputDir);
    }

    /**
     * Test if demuxOutputDir exists
     * @param demuxOutputDir
     * @return true if demuxOutputDir exists
     * @throws IOException
     */
    protected boolean checkDemuxOutputDir(String demuxOutputDir)
    throws IOException {
      return dirExists(demuxOutputDir);
    }


    /**
     * Delete DemuxOutput directory
     * @param demuxOutputDir
     * @return true if succeed
     * @throws IOException
     */
    protected boolean deleteDemuxOutputDir(String demuxOutputDir) throws IOException
    {
      return fs.delete(new Path(demuxOutputDir), true);
    }

    /**
     * Create directory if !exists
     * @param directory
     * @throws IOException
     */
    protected void setup(Path directory) throws IOException {
       if ( ! fs.exists(directory)) {
         fs.mkdirs(directory);
       }
     }

     /**
      * Check if source exists and if source is a directory
      * @param f source file
      */
    protected boolean dirExists(String directory) throws IOException {
       Path pDirectory = new Path(directory);
       return (fs.exists(pDirectory) && fs.getFileStatus(pDirectory).isDir());
     }

     /**
      * Move sourceFolder inside destFolder
      * @param srcDir
      * @param destDir
      * @return
      * @throws IOException
      */
    protected boolean moveFolder(String srcDir,String destDir, String prefix) throws IOException {
       if (!destDir.endsWith("/")) {
         destDir +="/";
       }
       Path pSrcDir = new Path(srcDir);
       Path pDestDir = new Path(destDir );
       setup(pDestDir);
       destDir += prefix +"_" +System.currentTimeMillis();
       Path pFinalDestDir = new Path(destDir );

       return fs.rename(pSrcDir, pFinalDestDir);
     }
 }
	/*
	* 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.extraction.demux;

	import java.io.IOException;
	import java.net.URI;
	import java.net.URISyntaxException;
	import java.text.SimpleDateFormat;
	import java.util.Date;

	import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
	import org.apache.hadoop.chukwa.extraction.CHUKWA_CONSTANT;
	import org.apache.hadoop.chukwa.util.NagiosHelper;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.fs.PathFilter;
	import org.apache.hadoop.util.ToolRunner;
	import org.apache.log4j.Logger;

	public class DemuxManager implements CHUKWA_CONSTANT {
	static Logger log = Logger.getLogger(DemuxManager.class);

	int globalErrorcounter = 0;
	Date firstErrorTime = null;

	protected int ERROR_SLEEP_TIME = 60;
	protected int NO_DATASINK_SLEEP_TIME = 20;

	protected int DEFAULT_MAX_ERROR_COUNT = 6;
	protected int DEFAULT_MAX_FILES_PER_DEMUX = 500;
	protected int DEFAULT_REDUCER_COUNT = 8;

	protected int maxPermittedErrorCount = DEFAULT_MAX_ERROR_COUNT;
	protected int demuxReducerCount = 0;
	protected ChukwaConfiguration conf = null;
	protected FileSystem fs = null;
	protected int reprocess = 0;
	protected boolean sendAlert = true;

	protected SimpleDateFormat dayTextFormat = new java.text.SimpleDateFormat("yyyyMMdd");
	protected volatile boolean isRunning = true;

	final private static PathFilter DATA_SINK_FILTER = new PathFilter() {
	public boolean accept(Path file) {
	return file.getName().endsWith(".done");
	}
	};


	public static void main(String[] args) throws Exception {

	DemuxManager manager = new DemuxManager();
	manager.start();

	}

	public DemuxManager() throws Exception {
	this.conf = new ChukwaConfiguration();
	init();
	}

	public DemuxManager(ChukwaConfiguration conf) throws Exception {
	this.conf = conf;
	init();
	}

	protected void init() throws IOException, URISyntaxException {
	String fsName = conf.get(HDFS_DEFAULT_NAME_FIELD);
	fs = FileSystem.get(new URI(fsName), conf);
	}

	public void shutdown() {
	this.isRunning = false;
	}


	public int getReprocess() {
	return reprocess;
	}

	/**
	* Start the Demux Manager daemon
	* @throws Exception if error in processing data
	*/
	public void start() throws Exception {

	String chukwaRootDir = conf.get(CHUKWA_ROOT_DIR_FIELD, DEFAULT_CHUKWA_ROOT_DIR_NAME);
	if ( ! chukwaRootDir.endsWith("/") ) {
	chukwaRootDir += "/";
	}
	log.info("chukwaRootDir:" + chukwaRootDir);

	String demuxRootDir = chukwaRootDir + DEFAULT_DEMUX_PROCESSING_DIR_NAME;
	String demuxErrorDir = demuxRootDir + DEFAULT_DEMUX_IN_ERROR_DIR_NAME;
	String demuxInputDir = demuxRootDir + DEFAULT_DEMUX_MR_INPUT_DIR_NAME;
	String demuxOutputDir = demuxRootDir + DEFAULT_DEMUX_MR_OUTPUT_DIR_NAME;

	String dataSinkDir = conf.get(CHUKWA_DATA_SINK_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_LOGS_DIR_NAME);
	if ( ! dataSinkDir.endsWith("/") ) {
	dataSinkDir += "/";
	}
	log.info("dataSinkDir:" + dataSinkDir);

	String postProcessDir = conf.get(CHUKWA_POST_PROCESS_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_POSTPROCESS_DIR_NAME);
	if ( ! postProcessDir.endsWith("/") ) {
	postProcessDir += "/";
	}
	log.info("postProcessDir:" + postProcessDir);

	String archiveRootDir = conf.get(CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_DATASINK_DIR_NAME);
	if ( ! archiveRootDir.endsWith("/") ) {
	archiveRootDir += "/";
	}
	log.info("archiveRootDir:" + archiveRootDir);

	maxPermittedErrorCount = conf.getInt(CHUKWA_DEMUX_MAX_ERROR_COUNT_FIELD,
	DEFAULT_MAX_ERROR_COUNT);
	demuxReducerCount = conf.getInt(CHUKWA_DEMUX_REDUCER_COUNT_FIELD, DEFAULT_REDUCER_COUNT);
	log.info("demuxReducerCount:" + demuxReducerCount);

	String nagiosHost = conf.get(CHUKWA_NAGIOS_HOST_FIELD);
	int nagiosPort = conf.getInt(CHUKWA_NAGIOS_PORT_FIELD,0);
	String reportingHost = conf.get(CHUKWA_REPORTING_HOST_FIELD);

	log.info("Nagios information: nagiosHost:" + nagiosHost + ", nagiosPort:"
	+ nagiosPort + ", reportingHost:" + reportingHost);


	if (nagiosHost == null \|\| nagiosHost.length() == 0 \|\| nagiosPort == 0 \|\| reportingHost == null \|\| reportingHost.length() == 0) {
	sendAlert = false;
	log.warn("Alerting is OFF");
	}

	boolean demuxReady = false;


	while (isRunning) {
	try {
	demuxReady = false;

	if (maxPermittedErrorCount != -1 && globalErrorcounter >= maxPermittedErrorCount) {
	log.warn("==================\nToo many errors (" + globalErrorcounter +
	"), Bail out!\n==================");
	break;
	}

	// Check for anomalies
	if (checkDemuxOutputDir(demuxOutputDir) == true) {
	// delete current demux output dir
	if ( deleteDemuxOutputDir(demuxOutputDir) == false ) {
	log.warn("Cannot delete an existing demux output directory!");
	throw new IOException("Cannot move demuxOutput to postProcess!");
	}
	continue;
	} else if (checkDemuxInputDir(demuxInputDir) == true) { // dataSink already there
	reprocess++;

	// Data has been processed more than 3 times ... move to InError directory
	if (reprocess > 3) {
	if (moveDataSinkFilesToDemuxErrorDirectory(demuxInputDir,demuxErrorDir) == false) {
	log.warn("Cannot move dataSink files to DemuxErrorDir!");
	throw new IOException("Cannot move dataSink files to DemuxErrorDir!");
	}
	reprocess = 0;
	continue;
	}

	log.error("Demux inputDir aready contains some dataSink files,"
	+ " going to reprocess, reprocessCount=" + reprocess);
	demuxReady = true;
	} else { // standard code path
	reprocess = 0;
	// Move new dataSink Files
	if (moveDataSinkFilesToDemuxInputDirectory(dataSinkDir, demuxInputDir) == true) {
	demuxReady = true; // if any are available
	} else {
	demuxReady = false; // if none
	}
	}

	// start a new demux ?
	if (demuxReady == true) {
	boolean demuxStatus = processData(dataSinkDir, demuxInputDir, demuxOutputDir,
	postProcessDir, archiveRootDir);
	sendDemuxStatusToNagios(nagiosHost,nagiosPort,reportingHost,demuxErrorDir,demuxStatus,null);

	// if demux suceeds, then we reset these.
	if (demuxStatus) {
	globalErrorcounter = 0;
	firstErrorTime = null;
	}
	} else {
	log.info("Demux not ready so going to sleep ...");
	Thread.sleep(NO_DATASINK_SLEEP_TIME * 1000);
	}
	}catch(Throwable e) {
	globalErrorcounter ++;
	if (firstErrorTime == null) firstErrorTime = new Date();

	log.warn("Consecutive error number " + globalErrorcounter +
	" encountered since " + firstErrorTime, e);
	sendDemuxStatusToNagios(nagiosHost,nagiosPort,reportingHost,demuxErrorDir,false, e.getMessage());
	try { Thread.sleep(ERROR_SLEEP_TIME * 1000); }
	catch (InterruptedException e1) {/do nothing/ }
	init();
	}
	}
	}


	/**
	* Send NSCA status to Nagios
	* @param nagiosHost
	* @param nagiosPort
	* @param reportingHost
	* @param demuxInErrorDir
	* @param demuxStatus
	* @param exception
	*/
	protected void sendDemuxStatusToNagios(String nagiosHost,int nagiosPort,String reportingHost,
	String demuxInErrorDir,boolean demuxStatus,String demuxException) {

	if (sendAlert == false) {
	return;
	}

	boolean demuxInErrorStatus = true;
	String demuxInErrorMsg = "";
	try {
	Path pDemuxInErrorDir = new Path(demuxInErrorDir);
	if ( fs.exists(pDemuxInErrorDir)) {
	FileStatus[] demuxInErrorDirs = fs.listStatus(pDemuxInErrorDir);
	if (demuxInErrorDirs.length == 0) {
	demuxInErrorStatus = false;
	}
	}
	} catch (Throwable e) {
	demuxInErrorMsg = e.getMessage();
	log.warn(e);
	}

	// send Demux status
	if (demuxStatus == true) {
	NagiosHelper.sendNsca("Demux OK",NagiosHelper.NAGIOS_OK);
	} else {
	NagiosHelper.sendNsca("Demux failed. " + demuxException,NagiosHelper.NAGIOS_CRITICAL);
	}

	// send DemuxInErrorStatus
	if (demuxInErrorStatus == false) {
	NagiosHelper.sendNsca("DemuxInError OK",NagiosHelper.NAGIOS_OK);
	} else {
	NagiosHelper.sendNsca("DemuxInError not empty -" + demuxInErrorMsg,NagiosHelper.NAGIOS_CRITICAL);
	}

	}

	/**
	* Process Data, i.e.
	* - run demux
	* - move demux output to postProcessDir
	* - move dataSink file to archiveDir
	*
	* @param dataSinkDir
	* @param demuxInputDir
	* @param demuxOutputDir
	* @param postProcessDir
	* @param archiveDir
	* @return True iff succeed
	* @throws IOException
	*/
	protected boolean processData(String dataSinkDir, String demuxInputDir,
	String demuxOutputDir, String postProcessDir, String archiveDir) throws IOException {

	boolean demuxStatus = false;

	long startTime = System.currentTimeMillis();
	demuxStatus = runDemux(demuxInputDir, demuxOutputDir);
	log.info("Demux Duration: " + (System.currentTimeMillis() - startTime));

	if (demuxStatus == false) {
	log.warn("Demux failed!");
	} else {

	// Move demux output to postProcessDir
	if (checkDemuxOutputDir(demuxOutputDir)) {
	if (moveDemuxOutputDirToPostProcessDirectory(demuxOutputDir, postProcessDir) == false) {
	log.warn("Cannot move demuxOutput to postProcess! bail out!");
	throw new IOException("Cannot move demuxOutput to postProcess! bail out!");
	}
	} else {
	log.warn("Demux processing OK but no output");
	}

	// Move DataSink Files to archiveDir
	if (moveDataSinkFilesToArchiveDirectory(demuxInputDir, archiveDir) == false) {
	log.warn("Cannot move datasinkFile to archive! bail out!");
	throw new IOException("Cannot move datasinkFile to archive! bail out!");
	}
	}

	return demuxStatus;
	}


	/**
	* Submit and Run demux Job
	* @param demuxInputDir
	* @param demuxOutputDir
	* @return true id Demux succeed
	*/
	protected boolean runDemux(String demuxInputDir, String demuxOutputDir) {
	// to reload the configuration, and demux's reduce number
	Configuration tempConf = new Configuration(conf);
	tempConf.reloadConfiguration();
	demuxReducerCount = tempConf.getInt(CHUKWA_DEMUX_REDUCER_COUNT_FIELD, DEFAULT_REDUCER_COUNT);
	String[] demuxParams;
	int i=0;
	Demux.addParsers(tempConf);
	demuxParams = new String[4];
	demuxParams[i++] = "-r";
	demuxParams[i++] = "" + demuxReducerCount;
	demuxParams[i++] = demuxInputDir;
	demuxParams[i++] = demuxOutputDir;
	try {
	return ( 0 == ToolRunner.run(tempConf,new Demux(), demuxParams) );
	} catch (Throwable e) {
	e.printStackTrace();
	globalErrorcounter ++;
	if (firstErrorTime == null) firstErrorTime = new Date();
	log.error("Failed to run demux. Consecutive error number " +
	globalErrorcounter + " encountered since " + firstErrorTime, e);
	}
	return false;
	}



	/**
	* Move dataSink files to Demux input directory
	* @param dataSinkDir
	* @param demuxInputDir
	* @return true if there's any dataSink files ready to be processed
	* @throws IOException
	*/
	protected boolean moveDataSinkFilesToDemuxInputDirectory(
	String dataSinkDir, String demuxInputDir) throws IOException {
	Path pDataSinkDir = new Path(dataSinkDir);
	Path pDemuxInputDir = new Path(demuxInputDir);
	log.info("dataSinkDir: " + dataSinkDir);
	log.info("demuxInputDir: " + demuxInputDir);


	boolean containsFile = false;

	FileStatus[] dataSinkFiles = fs.listStatus(pDataSinkDir,DATA_SINK_FILTER);
	if (dataSinkFiles.length > 0) {
	setup(pDemuxInputDir);
	}

	int maxFilesPerDemux = 0;
	for (FileStatus fstatus : dataSinkFiles) {
	boolean rename = fs.rename(fstatus.getPath(),pDemuxInputDir);
	log.info("Moving " + fstatus.getPath() + " to " + pDemuxInputDir +", status is:" + rename);
	maxFilesPerDemux ++;
	containsFile = true;
	if (maxFilesPerDemux >= DEFAULT_MAX_FILES_PER_DEMUX) {
	log.info("Max File per Demux reached:" + maxFilesPerDemux);
	break;
	}
	}
	return containsFile;
	}




	/**
	* Move sourceFolder inside destFolder
	* @param dataSinkDir : ex chukwa/demux/inputDir
	* @param demuxErrorDir : ex /chukwa/demux/inError
	* @return true if able to move chukwa/demux/inputDir to /chukwa/demux/inError/<YYYYMMDD>/demuxInputDirXXX
	* @throws IOException
	*/
	protected boolean moveDataSinkFilesToDemuxErrorDirectory(
	String dataSinkDir, String demuxErrorDir) throws IOException {
	demuxErrorDir += "/" + dayTextFormat.format(System.currentTimeMillis());
	return moveFolder(dataSinkDir,demuxErrorDir,"demuxInputDir");
	}

	/**
	* Move sourceFolder inside destFolder
	* @param demuxInputDir: ex chukwa/demux/inputDir
	* @param archiveDirectory: ex /chukwa/archives
	* @return true if able to move chukwa/demux/inputDir to /chukwa/archives/raw/<YYYYMMDD>/dataSinkDirXXX
	* @throws IOException
	*/
	protected boolean moveDataSinkFilesToArchiveDirectory(
	String demuxInputDir, String archiveDirectory) throws IOException {
	archiveDirectory += "/" + dayTextFormat.format(System.currentTimeMillis());
	return moveFolder(demuxInputDir,archiveDirectory,"dataSinkDir");
	}

	/**
	* Move sourceFolder inside destFolder
	* @param demuxOutputDir: ex chukwa/demux/outputDir
	* @param postProcessDirectory: ex /chukwa/postProcess
	* @return true if able to move chukwa/demux/outputDir to /chukwa/postProcess/demuxOutputDirXXX
	* @throws IOException
	*/
	protected boolean moveDemuxOutputDirToPostProcessDirectory(
	String demuxOutputDir, String postProcessDirectory) throws IOException {
	return moveFolder(demuxOutputDir,postProcessDirectory,"demuxOutputDir");
	}


	/**
	* Test if demuxInputDir exists
	* @param demuxInputDir
	* @return true if demuxInputDir exists
	* @throws IOException
	*/
	protected boolean checkDemuxInputDir(String demuxInputDir)
	throws IOException {
	return dirExists(demuxInputDir);
	}

	/**
	* Test if demuxOutputDir exists
	* @param demuxOutputDir
	* @return true if demuxOutputDir exists
	* @throws IOException
	*/
	protected boolean checkDemuxOutputDir(String demuxOutputDir)
	throws IOException {
	return dirExists(demuxOutputDir);
	}


	/**
	* Delete DemuxOutput directory
	* @param demuxOutputDir
	* @return true if succeed
	* @throws IOException
	*/
	protected boolean deleteDemuxOutputDir(String demuxOutputDir) throws IOException
	{
	return fs.delete(new Path(demuxOutputDir), true);
	}

	/**
	* Create directory if !exists
	* @param directory
	* @throws IOException
	*/
	protected void setup(Path directory) throws IOException {
	if ( ! fs.exists(directory)) {
	fs.mkdirs(directory);
	}
	}

	/**
	* Check if source exists and if source is a directory
	* @param f source file
	*/
	protected boolean dirExists(String directory) throws IOException {
	Path pDirectory = new Path(directory);
	return (fs.exists(pDirectory) && fs.getFileStatus(pDirectory).isDir());
	}

	/**
	* Move sourceFolder inside destFolder
	* @param srcDir
	* @param destDir
	* @return
	* @throws IOException
	*/
	protected boolean moveFolder(String srcDir,String destDir, String prefix) throws IOException {
	if (!destDir.endsWith("/")) {
	destDir +="/";
	}
	Path pSrcDir = new Path(srcDir);
	Path pDestDir = new Path(destDir );
	setup(pDestDir);
	destDir += prefix +"_" +System.currentTimeMillis();
	Path pFinalDestDir = new Path(destDir );

	return fs.rename(pSrcDir, pFinalDestDir);
	}
	}