/*
    * 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.text.SimpleDateFormat;
  import java.util.Calendar;
  import java.util.Collection;
  import java.util.HashSet;
  import java.util.List;
  
  import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
  import org.apache.hadoop.chukwa.util.HierarchyDataType;
  import org.apache.hadoop.fs.PathFilter;
  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.log4j.Logger;
  
  // TODO
  // First version of the Spill
  // need some polishing
  
  public class MoveToRepository {
    static Logger log = Logger.getLogger(MoveToRepository.class);
  
    static ChukwaConfiguration conf = null;
    static FileSystem fs = null;
    static SimpleDateFormat sdf = new java.text.SimpleDateFormat("yyyyMMdd");
    static Calendar calendar = Calendar.getInstance();
  
    static Collection<Path> processClusterDirectory(Path srcDir, String destDir)
        throws Exception {
      log.info("processClusterDirectory (" + srcDir.getName() + "," + destDir
          + ")");
      FileStatus fstat = fs.getFileStatus(srcDir);
      Collection<Path> destFiles = new HashSet<Path>();
  
      if (!fstat.isDir()) {
        throw new IOException(srcDir + " is not a directory!");
      } else {
        FileStatus[] datasourceDirectories = fs.listStatus(srcDir);
  
        for (FileStatus datasourceDirectory : datasourceDirectories) {
          log.info(datasourceDirectory.getPath() + " isDir?"
              + datasourceDirectory.isDir());
          if (!datasourceDirectory.isDir()) {
            throw new IOException(
                "Top level datasource directory should be a directory :"
                    + datasourceDirectory.getPath());
          }
  
          PathFilter filter = new PathFilter()
          {public boolean accept(Path file) {
            return file.getName().endsWith(".evt");
          }  };
          //CHUKWA-648:  Make Chukwa Reduce Type to support hierarchy format
          // to processDataSourceDirectory according to hierarchy data type format 
          List<FileStatus> eventfiles = HierarchyDataType.globStatus(fs, datasourceDirectory.getPath(),filter,true);
          for (FileStatus eventfile : eventfiles){
            Path datatypeDir = eventfile.getPath().getParent();
            String dirName = HierarchyDataType.getDataType(datatypeDir, srcDir);
          
            Path destPath = new Path(destDir + "/" + dirName);
            log.info("dest directory path: " + destPath);
            log.info("processClusterDirectory processing Datasource: (" + dirName
                + ")");
            StringBuilder dtDir = new StringBuilder(srcDir.toString()).append("/").append(dirName);
            log.debug("srcDir: " + dtDir.toString());
            processDatasourceDirectory(srcDir.toString(), new Path(dtDir.toString()), destDir + "/" + dirName);
          }
        }
      }
      return destFiles;
    }
  
    static Collection<Path> processDatasourceDirectory(String clusterpath, Path srcDir,
        String destDir) throws Exception {
      Path cPath = new Path(clusterpath);
      String cluster = cPath.getName();
     
     Collection<Path> destFiles = new HashSet<Path>();
     String fileName = null;
     int fileDay = 0;
     int fileHour = 0;
     int fileMin = 0;
 
     FileStatus[] recordFiles = fs.listStatus(srcDir);
     for (FileStatus recordFile : recordFiles) {
       // dataSource_20080915_18_15.1.evt
       // <datasource>_<yyyyMMdd_HH_mm>.1.evt
 
       fileName = recordFile.getPath().getName();
       log.info("processDatasourceDirectory processing RecordFile: (" + fileName
           + ")");
       log.info("fileName: " + fileName);
 
       int l = fileName.length();
       String dataSource = HierarchyDataType.getDataType(srcDir, cPath);
       log.info("Datasource: " + dataSource);
 
       if (fileName.endsWith(".D.evt")) {
         // Hadoop_dfs_datanode_20080919.D.evt
 
         fileDay = Integer.parseInt(fileName.substring(l - 14, l - 6));
         Path destFile = writeRecordFile(destDir + "/" + fileDay + "/",
             recordFile.getPath(),
             HierarchyDataType.getHierarchyDataTypeFileName(dataSource) + "_"
                 + fileDay);
         if (destFile != null) {
           destFiles.add(destFile);
         }
       } else if (fileName.endsWith(".H.evt")) {
         // Hadoop_dfs_datanode_20080925_1.H.evt
         // Hadoop_dfs_datanode_20080925_12.H.evt
 
         String day = null;
         String hour = null;
         if (fileName.charAt(l - 8) == '_') {
           day = fileName.substring(l - 16, l - 8);
           log.info("day->" + day);
           hour = "" + fileName.charAt(l - 7);
           log.info("hour->" + hour);
         } else {
           day = fileName.substring(l - 17, l - 9);
           log.info("day->" + day);
           hour = fileName.substring(l - 8, l - 6);
           log.info("hour->" + hour);
         }
         fileDay = Integer.parseInt(day);
         fileHour = Integer.parseInt(hour);
         // rotate there so spill
         Path destFile = writeRecordFile(destDir + "/" + fileDay + "/"
             + fileHour + "/", recordFile.getPath(),
             HierarchyDataType.getHierarchyDataTypeFileName(dataSource) + "_"
                 + fileDay + "_" + fileHour);
         if (destFile != null) {
           destFiles.add(destFile);
         }
         // mark this directory for daily rotate
         addDirectory4Rolling(true, fileDay, fileHour, cluster, dataSource);
       } else if (fileName.endsWith(".R.evt")) {
         if (fileName.charAt(l - 11) == '_') {
           fileDay = Integer.parseInt(fileName.substring(l - 19, l - 11));
           fileHour = Integer.parseInt("" + fileName.charAt(l - 10));
           fileMin = Integer.parseInt(fileName.substring(l - 8, l - 6));
         } else {
           fileDay = Integer.parseInt(fileName.substring(l - 20, l - 12));
           fileHour = Integer.parseInt(fileName.substring(l - 11, l - 9));
           fileMin = Integer.parseInt(fileName.substring(l - 8, l - 6));
         }
 
         log.info("fileDay: " + fileDay);
         log.info("fileHour: " + fileHour);
         log.info("fileMin: " + fileMin);
         Path destFile = writeRecordFile(
             destDir + "/" + fileDay + "/" + fileHour + "/" + fileMin,
             recordFile.getPath(),
             HierarchyDataType.getHierarchyDataTypeFileName(HierarchyDataType.trimSlash(dataSource))
             + "_" + fileDay + "_" + fileHour + "_" + fileMin);
         if (destFile != null) {
           destFiles.add(destFile);
         }
         // mark this directory for hourly rotate
         addDirectory4Rolling(false, fileDay, fileHour, cluster, dataSource);
       } else {
         throw new RuntimeException("Wrong fileName format! [" + fileName + "]");
       }
     }
 
     return destFiles;
   }
 
   static void addDirectory4Rolling(boolean isDailyOnly, int day, int hour,
       String cluster, String dataSource) throws IOException {
     // TODO get root directory from config
     String rollingDirectory = "/chukwa/rolling/";
 
     Path path = new Path(rollingDirectory + "/daily/" + day + "/" + cluster
         + "/" + dataSource);
     if (!fs.exists(path)) {
       fs.mkdirs(path);
     }
 
     if (!isDailyOnly) {
       path = new Path(rollingDirectory + "/hourly/" + day + "/" + hour + "/"
           + cluster + "/" + dataSource);
       if (!fs.exists(path)) {
         fs.mkdirs(path);
       }
     }
   }
 
   static Path writeRecordFile(String destDir, Path recordFile, String fileName)
       throws IOException {
     boolean done = false;
     int count = 1;
     do {
       Path destDirPath = new Path(destDir);
       Path destFilePath = new Path(destDir + "/" + fileName + "." + count
           + ".evt");
 
       if (!fs.exists(destDirPath)) {
         fs.mkdirs(destDirPath);
         log.info(">>>>>>>>>>>> create Dir" + destDirPath);
       }
 
       if (!fs.exists(destFilePath)) {
         log.info(">>>>>>>>>>>> Before Rename" + recordFile + " -- "
             + destFilePath);
         boolean rename = fs.rename(recordFile,destFilePath);
         done = true;
         log.info(">>>>>>>>>>>> after Rename" + destFilePath + " , rename:"+rename);
         return destFilePath;
       } 
       count++;
 
       if (count > 1000) {
         log.warn("too many files in this directory: " + destDir);
       }
     } while (!done);
 
     return null;
   }
 
   static boolean checkRotate(String directoryAsString,
       boolean createDirectoryIfNotExist) throws IOException {
     Path directory = new Path(directoryAsString);
     boolean exist = fs.exists(directory);
 
     if (!exist) {
       if (createDirectoryIfNotExist == true) {
         fs.mkdirs(directory);
       }
       return false;
     } else {
       return fs.exists(new Path(directoryAsString + "/rotateDone"));
     }
   }
 
   public static Path[] doMove(Path srcDir, String destDir) throws Exception {
     conf = new ChukwaConfiguration();
     String fsName = conf.get("writer.hdfs.filesystem");
     fs = FileSystem.get(new URI(fsName), conf);
     log.info("Start MoveToRepository doMove()");
 
     FileStatus fstat = fs.getFileStatus(srcDir);
 
     Collection<Path> destinationFiles = new HashSet<Path>();
     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 M/R outputDir
         if (name.startsWith("_")) {
           continue;
         }
         log.info("main procesing Cluster (" + cluster.getPath().getName() + ")");
         destinationFiles.addAll(processClusterDirectory(cluster.getPath(),
             destDir + "/" + cluster.getPath().getName()));
 
         // Delete the demux's cluster dir
         FileUtil.fullyDelete(fs, cluster.getPath());
       }
     }
 
     log.info("Done with MoveToRepository doMove()");
     return destinationFiles.toArray(new Path[destinationFiles.size()]);
   }
 
   /**
    * @param args is command line parameter
    * @throws Exception if error in processing data
    */
   public static void main(String[] args) throws Exception {
 
     Path srcDir = new Path(args[0]);
     String destDir = args[1];
     doMove(srcDir, destDir);
   }
 
 }