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

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

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

 public class ChukwaArchiveManager implements CHUKWA_CONSTANT {
   static Logger log = Logger.getLogger(ChukwaArchiveManager.class);
   SimpleDateFormat day = new java.text.SimpleDateFormat("yyyyMMdd");

   static final  int ONE_HOUR = 60 * 60 * 1000;
   static final int ONE_DAY = 24*ONE_HOUR;
   static final int MAX_FILES = 500;

   private static final int DEFAULT_MAX_ERROR_COUNT = 4;

   protected ChukwaConfiguration conf = null;
   protected FileSystem fs = null;
   protected boolean isRunning = true;

   public ChukwaArchiveManager() throws Exception {
     conf = new ChukwaConfiguration();
     init();
   }

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

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

     ChukwaArchiveManager manager = new ChukwaArchiveManager();
     manager.start();
   }

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

   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 archiveRootDir = conf.get(CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_DATASINK_DIR_NAME);
     if ( ! archiveRootDir.endsWith("/") ) {
       archiveRootDir += "/";
     }
     log.info("archiveDir:" + archiveRootDir);
     Path pArchiveRootDir = new Path(archiveRootDir);
     setup(pArchiveRootDir);

     String archivesRootProcessingDir = chukwaRootDir + ARCHIVES_PROCESSING_DIR_NAME;
     // String archivesErrorDir = archivesRootProcessingDir + DEFAULT_ARCHIVES_IN_ERROR_DIR_NAME;
     String archivesMRInputDir = archivesRootProcessingDir + ARCHIVES_MR_INPUT_DIR_NAME;
     String archivesMROutputDir = archivesRootProcessingDir+ ARCHIVES_MR_OUTPUT_DIR_NAME;
     String finalArchiveOutput = chukwaRootDir + DEFAULT_FINAL_ARCHIVES;

     int maxPermittedErrorCount = conf.getInt(CHUKWA_ARCHIVE_MAX_ERROR_COUNT_FIELD,
                                              DEFAULT_MAX_ERROR_COUNT);

     Path pDailyRawArchivesInput = new Path(archiveRootDir);
     Path pArchivesMRInputDir = new Path(archivesMRInputDir);
     Path pArchivesRootProcessingDir = new Path(archivesRootProcessingDir);
     Path pFinalArchiveOutput =  new Path(finalArchiveOutput);


     if (!archivesMRInputDir.endsWith("/")) {
       archivesMRInputDir +="/";
     }
     setup( pArchivesRootProcessingDir );
     setup( pDailyRawArchivesInput );
     setup( pFinalArchiveOutput );

     int errorCount = 0;

     long lastRun = 0l;

     while (isRunning) {
       try {

         if (maxPermittedErrorCount != -1 && errorCount >= maxPermittedErrorCount) {
           log.warn("==================\nToo many errors (" + errorCount +
                    "), Bail out!\n==================");
           break;
         }
         // /chukwa/archives/<YYYYMMDD>/dataSinkDirXXX
         //  to
         // /chukwa/archives/final/<YYYYMMDD>_<TS>

         if (fs.exists(pArchivesMRInputDir)) {
           FileStatus[] days = fs.listStatus(pArchivesMRInputDir);
           if (days.length > 0) {
             log.info("reprocessing current Archive input" +  days[0].getPath());

             runArchive(archivesMRInputDir + days[0].getPath().getName() + "/",archivesMROutputDir,finalArchiveOutput);
             errorCount = 0;
             continue;
           }
         }


         log.info("Raw Archive dir:" + pDailyRawArchivesInput);
         long now = System.currentTimeMillis();
         int currentDay = Integer.parseInt(day.format(System.currentTimeMillis()));
         FileStatus[] daysInRawArchiveDir = fs.listStatus(pDailyRawArchivesInput);

         if (daysInRawArchiveDir.length == 0 ) {
           log.debug( pDailyRawArchivesInput + " is empty, going to sleep for 1 minute");
           Thread.sleep(1 * 60 * 1000);
           continue;
         }
         // We don't want to process DataSink file more than once every 2 hours
         // for current day
         if (daysInRawArchiveDir.length == 1 ) {
           int workingDay = Integer.parseInt(daysInRawArchiveDir[0].getPath().getName());
           long nextRun = lastRun + (2*ONE_HOUR) - (1*60*1000);// 2h -1min
           if (workingDay == currentDay && now < nextRun) {
             log.info("lastRun < 2 hours so skip archive for now, going to sleep for 30 minutes, currentDate is:" + new java.util.Date());
             Thread.sleep(30 * 60 * 1000);
             continue;
           }
         }

         String dayArchivesMRInputDir = null;
         for (FileStatus fsDay : daysInRawArchiveDir) {
           dayArchivesMRInputDir = archivesMRInputDir + fsDay.getPath().getName() + "/";
           processDay(fsDay, dayArchivesMRInputDir,archivesMROutputDir, finalArchiveOutput);
           lastRun = now;
         }

       }catch (Throwable e) {
         errorCount ++;
         e.printStackTrace();
         log.warn(e);
       }

     }

   }

   public void processDay(FileStatus fsDay, String archivesMRInputDir,
       String archivesMROutputDir,String finalArchiveOutput) throws Exception {
     FileStatus[] dataSinkDirsInRawArchiveDir = fs.listStatus(fsDay.getPath());
     long now = System.currentTimeMillis();

     int currentDay = Integer.parseInt(day.format(System.currentTimeMillis()));
     int workingDay = Integer.parseInt(fsDay.getPath().getName());

     long oneHourAgo = now -  ONE_HOUR;
     if (dataSinkDirsInRawArchiveDir.length == 0 && workingDay < currentDay) {
       fs.delete(fsDay.getPath(),false);
       log.info("deleting raw dataSink dir for day:" + fsDay.getPath().getName());
       return;
     }

     int fileCount = 0;
     for (FileStatus fsDataSinkDir : dataSinkDirsInRawArchiveDir) {
       long modificationDate = fsDataSinkDir.getModificationTime();
       if (modificationDate < oneHourAgo || workingDay < currentDay) {
         log.info("processDay,modificationDate:" + modificationDate +", adding: " + fsDataSinkDir.getPath() );
         fileCount += fs.listStatus(fsDataSinkDir.getPath()).length;
         moveDataSinkFilesToArchiveMrInput(fsDataSinkDir,archivesMRInputDir);
         // process no more than MAX_FILES directories
         if (fileCount >= MAX_FILES) {
           log.info("processDay, reach capacity");
           runArchive(archivesMRInputDir,archivesMROutputDir,finalArchiveOutput);
           fileCount = 0;
         } else {
           log.info("processDay,modificationDate:" + modificationDate +", skipping: " + fsDataSinkDir.getPath() );
         }
       }
     }
   }

   public void runArchive(String archivesMRInputDir,String archivesMROutputDir,
       String finalArchiveOutput) throws Exception {
     String[] args = new String[3];


     args[0] = conf.get("archive.grouper","Stream");
     args[1] = archivesMRInputDir + "*/*.done" ;
     args[2] = archivesMROutputDir;

     Path pArchivesMRInputDir = new Path(archivesMRInputDir);
     Path pArchivesMROutputDir = new Path(archivesMROutputDir);


     if (fs.exists(pArchivesMROutputDir)) {
       log.warn("Deleteing mroutput dir for archive ...");
       fs.delete(pArchivesMROutputDir, true);
     }

     log.info("ChukwaArchiveManager processing :" + args[1] + " going to output to " + args[2] );
     int res = ToolRunner.run(this.conf, new ChukwaArchiveBuilder(),args);
     log.info("Archive result: " + res);
     if (res != 0) {
       throw new Exception("Archive result != 0");
     }

     if (!finalArchiveOutput.endsWith("/")) {
       finalArchiveOutput +="/";
     }
     String day = pArchivesMRInputDir.getName();
     finalArchiveOutput += day;
     Path pDay = new Path(finalArchiveOutput);
     setup(pDay);

     finalArchiveOutput += "/archive_" + System.currentTimeMillis();
     Path pFinalArchiveOutput = new Path(finalArchiveOutput);

     log.info("Final move: moving " + pArchivesMROutputDir + " to " + pFinalArchiveOutput);

     if (fs.rename(pArchivesMROutputDir, pFinalArchiveOutput ) ) {
       log.info("deleting " + pArchivesMRInputDir);
       fs.delete(pArchivesMRInputDir, true);
     } else {
       log.warn("move to final archive folder failed!");
     }


   }

   public void moveDataSinkFilesToArchiveMrInput(FileStatus fsDataSinkDir,
       String archivesMRInputDir) throws IOException {

     if (!archivesMRInputDir.endsWith("/")) {
       archivesMRInputDir +="/";
     }

     Path pArchivesMRInputDir = new Path(archivesMRInputDir);
     setup(pArchivesMRInputDir);
     fs.rename(fsDataSinkDir.getPath(), pArchivesMRInputDir);
     log.info("moving " + fsDataSinkDir.getPath() + " to " + pArchivesMRInputDir);
   }

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

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

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

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

	public class ChukwaArchiveManager implements CHUKWA_CONSTANT {
	static Logger log = Logger.getLogger(ChukwaArchiveManager.class);
	SimpleDateFormat day = new java.text.SimpleDateFormat("yyyyMMdd");

	static final int ONE_HOUR = 60 * 60 * 1000;
	static final int ONE_DAY = 24*ONE_HOUR;
	static final int MAX_FILES = 500;

	private static final int DEFAULT_MAX_ERROR_COUNT = 4;

	protected ChukwaConfiguration conf = null;
	protected FileSystem fs = null;
	protected boolean isRunning = true;

	public ChukwaArchiveManager() throws Exception {
	conf = new ChukwaConfiguration();
	init();
	}

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

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

	ChukwaArchiveManager manager = new ChukwaArchiveManager();
	manager.start();
	}

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

	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 archiveRootDir = conf.get(CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_DATASINK_DIR_NAME);
	if ( ! archiveRootDir.endsWith("/") ) {
	archiveRootDir += "/";
	}
	log.info("archiveDir:" + archiveRootDir);
	Path pArchiveRootDir = new Path(archiveRootDir);
	setup(pArchiveRootDir);

	String archivesRootProcessingDir = chukwaRootDir + ARCHIVES_PROCESSING_DIR_NAME;
	// String archivesErrorDir = archivesRootProcessingDir + DEFAULT_ARCHIVES_IN_ERROR_DIR_NAME;
	String archivesMRInputDir = archivesRootProcessingDir + ARCHIVES_MR_INPUT_DIR_NAME;
	String archivesMROutputDir = archivesRootProcessingDir+ ARCHIVES_MR_OUTPUT_DIR_NAME;
	String finalArchiveOutput = chukwaRootDir + DEFAULT_FINAL_ARCHIVES;

	int maxPermittedErrorCount = conf.getInt(CHUKWA_ARCHIVE_MAX_ERROR_COUNT_FIELD,
	DEFAULT_MAX_ERROR_COUNT);

	Path pDailyRawArchivesInput = new Path(archiveRootDir);
	Path pArchivesMRInputDir = new Path(archivesMRInputDir);
	Path pArchivesRootProcessingDir = new Path(archivesRootProcessingDir);
	Path pFinalArchiveOutput = new Path(finalArchiveOutput);


	if (!archivesMRInputDir.endsWith("/")) {
	archivesMRInputDir +="/";
	}
	setup( pArchivesRootProcessingDir );
	setup( pDailyRawArchivesInput );
	setup( pFinalArchiveOutput );

	int errorCount = 0;

	long lastRun = 0l;

	while (isRunning) {
	try {

	if (maxPermittedErrorCount != -1 && errorCount >= maxPermittedErrorCount) {
	log.warn("==================\nToo many errors (" + errorCount +
	"), Bail out!\n==================");
	break;
	}
	// /chukwa/archives/<YYYYMMDD>/dataSinkDirXXX
	// to
	// /chukwa/archives/final/<YYYYMMDD>_<TS>

	if (fs.exists(pArchivesMRInputDir)) {
	FileStatus[] days = fs.listStatus(pArchivesMRInputDir);
	if (days.length > 0) {
	log.info("reprocessing current Archive input" + days[0].getPath());

	runArchive(archivesMRInputDir + days[0].getPath().getName() + "/",archivesMROutputDir,finalArchiveOutput);
	errorCount = 0;
	continue;
	}
	}


	log.info("Raw Archive dir:" + pDailyRawArchivesInput);
	long now = System.currentTimeMillis();
	int currentDay = Integer.parseInt(day.format(System.currentTimeMillis()));
	FileStatus[] daysInRawArchiveDir = fs.listStatus(pDailyRawArchivesInput);

	if (daysInRawArchiveDir.length == 0 ) {
	log.debug( pDailyRawArchivesInput + " is empty, going to sleep for 1 minute");
	Thread.sleep(1 * 60 * 1000);
	continue;
	}
	// We don't want to process DataSink file more than once every 2 hours
	// for current day
	if (daysInRawArchiveDir.length == 1 ) {
	int workingDay = Integer.parseInt(daysInRawArchiveDir[0].getPath().getName());
	long nextRun = lastRun + (2ONE_HOUR) - (160*1000);// 2h -1min
	if (workingDay == currentDay && now < nextRun) {
	log.info("lastRun < 2 hours so skip archive for now, going to sleep for 30 minutes, currentDate is:" + new java.util.Date());
	Thread.sleep(30 * 60 * 1000);
	continue;
	}
	}

	String dayArchivesMRInputDir = null;
	for (FileStatus fsDay : daysInRawArchiveDir) {
	dayArchivesMRInputDir = archivesMRInputDir + fsDay.getPath().getName() + "/";
	processDay(fsDay, dayArchivesMRInputDir,archivesMROutputDir, finalArchiveOutput);
	lastRun = now;
	}

	}catch (Throwable e) {
	errorCount ++;
	e.printStackTrace();
	log.warn(e);
	}

	}

	}

	public void processDay(FileStatus fsDay, String archivesMRInputDir,
	String archivesMROutputDir,String finalArchiveOutput) throws Exception {
	FileStatus[] dataSinkDirsInRawArchiveDir = fs.listStatus(fsDay.getPath());
	long now = System.currentTimeMillis();

	int currentDay = Integer.parseInt(day.format(System.currentTimeMillis()));
	int workingDay = Integer.parseInt(fsDay.getPath().getName());

	long oneHourAgo = now - ONE_HOUR;
	if (dataSinkDirsInRawArchiveDir.length == 0 && workingDay < currentDay) {
	fs.delete(fsDay.getPath(),false);
	log.info("deleting raw dataSink dir for day:" + fsDay.getPath().getName());
	return;
	}

	int fileCount = 0;
	for (FileStatus fsDataSinkDir : dataSinkDirsInRawArchiveDir) {
	long modificationDate = fsDataSinkDir.getModificationTime();
	if (modificationDate < oneHourAgo \|\| workingDay < currentDay) {
	log.info("processDay,modificationDate:" + modificationDate +", adding: " + fsDataSinkDir.getPath() );
	fileCount += fs.listStatus(fsDataSinkDir.getPath()).length;
	moveDataSinkFilesToArchiveMrInput(fsDataSinkDir,archivesMRInputDir);
	// process no more than MAX_FILES directories
	if (fileCount >= MAX_FILES) {
	log.info("processDay, reach capacity");
	runArchive(archivesMRInputDir,archivesMROutputDir,finalArchiveOutput);
	fileCount = 0;
	} else {
	log.info("processDay,modificationDate:" + modificationDate +", skipping: " + fsDataSinkDir.getPath() );
	}
	}
	}
	}

	public void runArchive(String archivesMRInputDir,String archivesMROutputDir,
	String finalArchiveOutput) throws Exception {
	String[] args = new String[3];


	args[0] = conf.get("archive.grouper","Stream");
	args[1] = archivesMRInputDir + "/.done" ;
	args[2] = archivesMROutputDir;

	Path pArchivesMRInputDir = new Path(archivesMRInputDir);
	Path pArchivesMROutputDir = new Path(archivesMROutputDir);


	if (fs.exists(pArchivesMROutputDir)) {
	log.warn("Deleteing mroutput dir for archive ...");
	fs.delete(pArchivesMROutputDir, true);
	}

	log.info("ChukwaArchiveManager processing :" + args[1] + " going to output to " + args[2] );
	int res = ToolRunner.run(this.conf, new ChukwaArchiveBuilder(),args);
	log.info("Archive result: " + res);
	if (res != 0) {
	throw new Exception("Archive result != 0");
	}

	if (!finalArchiveOutput.endsWith("/")) {
	finalArchiveOutput +="/";
	}
	String day = pArchivesMRInputDir.getName();
	finalArchiveOutput += day;
	Path pDay = new Path(finalArchiveOutput);
	setup(pDay);

	finalArchiveOutput += "/archive_" + System.currentTimeMillis();
	Path pFinalArchiveOutput = new Path(finalArchiveOutput);

	log.info("Final move: moving " + pArchivesMROutputDir + " to " + pFinalArchiveOutput);

	if (fs.rename(pArchivesMROutputDir, pFinalArchiveOutput ) ) {
	log.info("deleting " + pArchivesMRInputDir);
	fs.delete(pArchivesMRInputDir, true);
	} else {
	log.warn("move to final archive folder failed!");
	}



	}

	public void moveDataSinkFilesToArchiveMrInput(FileStatus fsDataSinkDir,
	String archivesMRInputDir) throws IOException {

	if (!archivesMRInputDir.endsWith("/")) {
	archivesMRInputDir +="/";
	}

	Path pArchivesMRInputDir = new Path(archivesMRInputDir);
	setup(pArchivesMRInputDir);
	fs.rename(fsDataSinkDir.getPath(), pArchivesMRInputDir);
	log.info("moving " + fsDataSinkDir.getPath() + " to " + pArchivesMRInputDir);
	}

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

	}