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

import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hadoop.chukwa.extraction.CHUKWA_CONSTANT;
import org.apache.log4j.Logger;
import java.io.IOException;

/**
 * A lightweight tool for archiving, suitable for small-to-medium-size Chukwa
 * deployments that don't use Demux.
 * Grabs everything in the data sink, runs the Archiver MapReduce job,
 * then promotes output to the archive dir.
 * 
 * Input is determined by conf option chukwaArchiveDir; defaults to
 *   /chukwa/logs
 *   
 *   Uses /chukwa/archivesProcessing/mr[Input/Output] as tmp storage
 *   
 *   Outputs to /chukwa/archives
 * 
 */
public class SinkArchiver implements CHUKWA_CONSTANT {
  
  final public static PathFilter DATA_SINK_FILTER = new PathFilter() {
    public boolean accept(Path file) {
      return file.getName().endsWith(".done");
    }     
  };
  
  static Logger log = Logger.getLogger(SinkArchiver.class);
  
  public static void main(String[] args) {
    try {
      Configuration conf = new ChukwaConfiguration();
      if(conf.get(ChukwaArchiveDataTypeOutputFormat.GROUP_BY_CLUSTER_OPTION_NAME) == null )
        conf.set(ChukwaArchiveDataTypeOutputFormat.GROUP_BY_CLUSTER_OPTION_NAME, "true");
      FileSystem fs = FileSystem.get(conf);
      SinkArchiver archiver = new SinkArchiver();
      archiver.exec(fs, conf);    
    } catch(IOException e) {
      e.printStackTrace();
    }
  }
  

  /*
   * Pull most of the logic into instance methods so that we can
   * more easily unit-test, by altering passed-in configuration.
   */
  public void exec(FileSystem fs, Configuration conf) {
    try {
      
      String chukwaRootDir = conf.get(CHUKWA_ROOT_DIR_FIELD, DEFAULT_CHUKWA_ROOT_DIR_NAME);
      if ( ! chukwaRootDir.endsWith("/") ) {
        chukwaRootDir += "/";
      }
      String archiveSource = conf.get(CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +DEFAULT_CHUKWA_LOGS_DIR_NAME);
      if ( ! archiveSource.endsWith("/") ) {
        archiveSource += "/";
      }
      String archivesRootProcessingDir = chukwaRootDir + ARCHIVES_PROCESSING_DIR_NAME;
      
      //String archivesErrorDir = archivesRootProcessingDir + ARCHIVES_IN_ERROR_DIR_NAME;
      String archivesMRInputDir = archivesRootProcessingDir + ARCHIVES_MR_INPUT_DIR_NAME;
      String archivesMROutputDir = archivesRootProcessingDir+ ARCHIVES_MR_OUTPUT_DIR_NAME;

      Path pSource = new Path(archiveSource);
      
      Path pMRInputDir = new Path(archivesMRInputDir);
      if(!fs.exists(pMRInputDir))
        fs.mkdirs(pMRInputDir);
      
      Path pOutputDir = new Path(archivesMROutputDir);
      if(!fs.exists(pOutputDir))
        fs.mkdirs(pOutputDir);
      
      if(fs.listStatus(pOutputDir).length == 0)
        fs.delete(pOutputDir, true);
      Path archive = new Path(chukwaRootDir + "archive");
      
      selectInputs(fs, pSource, pMRInputDir);

      int result = runMapRedJob(conf, archivesMRInputDir, archivesMROutputDir);
      if(result == 0) { //success, so empty input dir
        fs.delete(pMRInputDir, true);
      }
      
      if(!fs.exists(archive)) {
        fs.mkdirs(archive);
      }
      FileStatus[] files = fs.listStatus(pOutputDir);
      for(FileStatus f: files) {
        if(!f.getPath().getName().endsWith("_logs"))
          promoteAndMerge(fs, f.getPath(), archive);
      }

      fs.delete(pOutputDir, true);
      
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  
  private void selectInputs(FileSystem fs, Path pSource,
      Path pMRInputDir) throws IOException {
    
    FileStatus[] dataSinkFiles = fs.listStatus(pSource, DATA_SINK_FILTER);
    for(FileStatus fstatus: dataSinkFiles) {
      boolean rename = fs.rename(fstatus.getPath(),pMRInputDir);
      log.info("Moving " + fstatus.getPath() + " to " + pMRInputDir 
          +", status is: " + rename);
    }
    
  }

  public int runMapRedJob(Configuration conf, String in, String out)
    throws Exception {
    String grouper = conf.get("archive.grouper","DataType");
    String[] args = new String[] {grouper, in, out};
    int res = ToolRunner.run(conf, new ChukwaArchiveBuilder(),
        args);
    return res;
  }
  /**
   * Merges the contents of src into dest.
   * If src is a file, moves it to dest.
   * 
   * @param fs the filesystem in question
   * @param src a file or directory to merge into dest
   * @param dest a directory to merge into
   * @throws IOException if error in promote and merge
   */
  public void promoteAndMerge(FileSystem fs, Path src, Path dest) 
  throws IOException {
    FileStatus stat = fs.getFileStatus(src);
    String baseName = src.getName();
    Path target = new Path(dest, baseName);
    if(!fs.exists(target)) {
      fs.rename(src, dest);
      System.out.println("moving " + src + " to " + dest);
    } else if(stat.isDir()) {//recurse
      FileStatus[] files = fs.listStatus(src);
      for(FileStatus f: files) {
        promoteAndMerge(fs, f.getPath(), target);
      }
    } else { //append a number to unique-ify filename
      int i=0;
      do {
        //FIXME: can make this more generic
        String newName;
        if(baseName.endsWith(".arc")) {
          newName = baseName.substring(0, baseName.length() - 4) + "-"+i+".arc";
        }
        else
          newName = baseName+"-"+i;
        target = new Path(dest, newName);
      } while(fs.exists(target));
      fs.rename(src, target);
      System.out.println("promoting " + src + " to " + target);
    }

  }
}