content: /*
   * 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 org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
  import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
  import org.apache.hadoop.conf.Configured;
  import org.apache.hadoop.fs.FileStatus;
  import org.apache.hadoop.fs.FileSystem;
  import org.apache.hadoop.fs.FileUtil;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.fs.PathFilter;
  import org.apache.hadoop.mapred.FileInputFormat;
  import org.apache.hadoop.mapred.FileOutputFormat;
  import org.apache.hadoop.mapred.JobClient;
  import org.apache.hadoop.mapred.JobConf;
  import org.apache.hadoop.mapred.SequenceFileInputFormat;
  import org.apache.hadoop.mapred.SequenceFileOutputFormat;
  import org.apache.hadoop.mapred.lib.IdentityMapper;
  import org.apache.hadoop.mapred.lib.IdentityReducer;
  import org.apache.hadoop.util.Tool;
  import org.apache.hadoop.util.ToolRunner;
  
  public class MoveOrMergeRecordFile extends Configured implements Tool {
    static ChukwaConfiguration conf = null;
    static FileSystem fs = null;
    static final String HadoopLogDir = "_logs";
    static final String hadoopTempDir = "_temporary";
  
    public int run(String[] args) throws Exception {
      JobConf conf = new JobConf(getConf(), MoveOrMergeRecordFile.class);
  
      conf.setJobName("Chukwa-MoveOrMergeLogFile");
      conf.setInputFormat(SequenceFileInputFormat.class);
  
      conf.setMapperClass(IdentityMapper.class);
      conf.setReducerClass(IdentityReducer.class);
  
      // conf.setPartitionerClass(ChukwaPartitioner.class);
      // conf.setOutputFormat(ChukwaOutputFormat.class);
  
      conf.setOutputKeyClass(ChukwaRecordKey.class);
      conf.setOutputValueClass(ChukwaRecord.class);
      conf.setOutputFormat(SequenceFileOutputFormat.class);
  
      FileInputFormat.setInputPaths(conf, args[0]);
      FileOutputFormat.setOutputPath(conf, new Path(args[1]));
  
      JobClient.runJob(conf);
      return 0;
    }
  
    static void moveOrMergeOneCluster(Path srcDir, String destDir)
        throws Exception {
      System.out.println("moveOrMergeOneCluster (" + srcDir.getName() + ","
          + destDir + ")");
      FileStatus fstat = fs.getFileStatus(srcDir);
  
      if (!fstat.isDir()) {
        throw new IOException(srcDir + " is not a directory!");
      } else {
        FileStatus[] datasourceDirectories = fs.listStatus(srcDir);
        for (FileStatus datasourceDirectory : datasourceDirectories) {
          System.out.println(datasourceDirectory.getPath() + " isDir?"
              + datasourceDirectory.isDir());
          if (!datasourceDirectory.isDir()) {
            throw new IOException("Top level should just contains directories :"
                + datasourceDirectory.getPath());
          }
  
          String dirName = datasourceDirectory.getPath().getName();
  
          Path destPath = new Path(destDir + "/" + dirName);
          System.out.println("dest directory path: " + destPath);
  
          if (!fs.exists(destPath)) {
            System.out.println("create datasource directory [" + destDir + "/"
                + dirName + "]");
            fs.mkdirs(destPath);
          }
  
          FileStatus[] evts = fs.listStatus(datasourceDirectory.getPath(),
              new EventFileFilter());
          for (FileStatus eventFile : evts) {
  
            Path eventFilePath = eventFile.getPath();
            String filename = eventFilePath.getName();
            System.out.println("src dir File: [" + filename + "]");
            Path destFilePath = new Path(destDir + "/" + dirName + "/" + filename);
            if (!fs.exists(destFilePath)) {
              System.out.println("Moving File: [" + destFilePath + "]");
              // Copy to final Location
              FileUtil.copy(fs, eventFilePath, fs, destFilePath, false, false,
                  conf);
            } else {
              System.out.println("Need to merge! : [" + destFilePath + "]");
              String strMrPath = datasourceDirectory.getPath().toString() + "/"
                  + "MR_" + System.currentTimeMillis();
              Path mrPath = new Path(strMrPath);
              System.out.println("\t New MR directory : [" + mrPath + "]");
              // Create MR input Dir
              fs.mkdirs(mrPath);
              // Move Input files
              FileUtil.copy(fs, eventFilePath, fs,
                  new Path(strMrPath + "/1.evt"), false, false, conf);
              fs.rename(destFilePath, new Path(strMrPath + "/2.evt"));
  
              // Merge
              String[] mergeArgs = new String[2];
              mergeArgs[0] = strMrPath;
              mergeArgs[1] = strMrPath + "/mrOutput";
              DoMerge merge = new DoMerge(conf, fs, eventFilePath, destFilePath,
                  mergeArgs);
              merge.start();
            }
          }
        }
      }
  
    }
  
    /**
     * @param args is command line parameters
     * @throws Exception if unable to process data
     */
    public static void main(String[] args) throws Exception {
      conf = new ChukwaConfiguration();
      String fsName = conf.get("writer.hdfs.filesystem");
      fs = FileSystem.get(new URI(fsName), conf);
  
      Path srcDir = new Path(args[0]);
      String destDir = args[1];
  
      FileStatus fstat = fs.getFileStatus(srcDir);
  
      if (!fstat.isDir()) {
        throw new IOException(srcDir + " is not a directory!");
      } else {
        FileStatus[] clusters = fs.listStatus(srcDir);
        // Run a moveOrMerge on all clusters
        String name = null;
        for (FileStatus cluster : clusters) {
          name = cluster.getPath().getName();
          // Skip hadoop outDir
          if ((name.intern() == HadoopLogDir.intern())
              || (name.intern() == hadoopTempDir.intern())) {
            continue;
          }
          moveOrMergeOneCluster(cluster.getPath(), destDir + "/"
              + cluster.getPath().getName());
        }
      }
      System.out.println("Done with moveOrMerge main()");
    }
  }
  
  
  class DoMerge extends Thread {
    ChukwaConfiguration conf = null;
    FileSystem fs = null;
    String[] mergeArgs = new String[2];
    Path destFilePath = null;
    Path eventFilePath = null;
  
    public DoMerge(ChukwaConfiguration conf, FileSystem fs, Path eventFilePath,
                   Path destFilePath, String[] mergeArgs) {
      this.conf = conf;
      this.fs = fs;
      this.eventFilePath = eventFilePath;
      this.destFilePath = destFilePath;
      this.mergeArgs = mergeArgs;
    }
  
    @Override
    public void run() {
      System.out.println("\t Running Merge! : output [" + mergeArgs[1] + "]");
      int res;
      try {
        res = ToolRunner.run(new ChukwaConfiguration(),
            new MoveOrMergeRecordFile(), mergeArgs);
        System.out.println("MR exit status: " + res);
        if (res == 0) {
          System.out.println("\t Moving output file : to [" + destFilePath + "]");
          FileUtil.copy(fs, new Path(mergeArgs[1] + "/part-00000"), fs,
              destFilePath, false, false, conf);
          fs.rename(new Path(mergeArgs[1] + "/part-00000"), eventFilePath);
        } else {
          throw new RuntimeException("Error in M/R merge operation!");
        }
  
      } catch (Exception e) {
        e.printStackTrace();
        throw new RuntimeException("Error in M/R merge operation!", e);
      }
    }
  
  }
  
  
  class EventFileFilter implements PathFilter {
    public boolean accept(Path path) {
      return (path.toString().endsWith(".evt"));
    }
  }
  