/*
    * 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.datacollection.writer.localfs;
  
  import java.io.File;
  import java.io.IOException;
  import java.net.InetAddress;
  import java.net.UnknownHostException;
  import java.nio.ByteBuffer;
  import java.util.Calendar;
  import java.util.List;
  import java.util.Timer;
  import java.util.TimerTask;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.LinkedBlockingQueue;
  
  import org.apache.avro.Schema;
  import org.apache.avro.generic.GenericData;
  import org.apache.avro.generic.GenericRecord;
  import org.apache.hadoop.chukwa.ChukwaArchiveKey;
  import org.apache.hadoop.chukwa.Chunk;
  import org.apache.hadoop.chukwa.datacollection.writer.ChukwaWriter;
  import org.apache.hadoop.chukwa.datacollection.writer.WriterException;
  import org.apache.hadoop.chukwa.datacollection.writer.parquet.ChukwaAvroSchema;
  import org.apache.hadoop.chukwa.util.ExceptionUtil;
  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.log4j.Logger;
  import org.apache.parquet.avro.AvroParquetWriter;
  import org.apache.parquet.hadoop.metadata.CompressionCodecName;
  
  /**
   * <p>This class <b>is</b> thread-safe -- rotate() and save() both synchronize on
   * lock object.
   * </p>
   * <p>
   * Write data to a local fileSystem then move it to the remote HDFS
   * <br>
   * Warning:
   * <br>
   * There's no lock/waiting time for the remote client.
   * The connection is released as soon as the last append is done,
   * so therefore there is no guarantee that this class will not loose 
   * any data.
   * <br>
   * This class has been designed this way for performance reason.
   * </p>
   * <p>
   * In order to use this class, you need to define some parameters,
   * in chukwa-collector-conf.xml
   * <p>
   * <br>
   *  &lt;property&gt;<br>
   *   &lt;name&gt;chukwaCollector.localOutputDir&lt;/name&gt;<br>
   *   &lt;value&gt;/grid/0/gs/chukwa/chukwa-0.1.2/dataSink/&lt;/value&gt;<br>
   *   &lt;description&gt;Chukwa data sink directory&lt;/description&gt;<br>
   *  &lt;/property&gt;<br>
   *<br>
   *  &lt;property&gt;<br>
   *    &lt;name&gt;chukwaCollector.writerClass&lt;/name&gt;<br>
   *    &lt;value&gt;org.apache.hadoop.chukwa.datacollection.writer.localfs.LocalWriter&lt;/value&gt;<br>
   *    &lt;description&gt;Local chukwa writer&lt;/description&gt;<br>
   *  &lt;/property&gt;<br>
   * <br>
   */
  public class LocalWriter implements ChukwaWriter {
  
    static Logger log = Logger.getLogger(LocalWriter.class);
    static final int STAT_INTERVAL_SECONDS = 30;
    static String localHostAddr = null;
    private int blockSize = 128 * 1024 * 1024;
    private int pageSize = 1 * 1024 * 1024;
  
    private final Object lock = new Object();
    private BlockingQueue<String> fileQueue = null;
    @SuppressWarnings("unused")
    private LocalToRemoteHdfsMover localToRemoteHdfsMover = null;
    private FileSystem fs = null;
    private Configuration conf = null;
  
    private String localOutputDir = null;
   private Calendar calendar = Calendar.getInstance();
 
   private Path currentPath = null;
   private String currentFileName = null;
   private AvroParquetWriter<GenericRecord> parquetWriter = null;
   private int rotateInterval = 1000 * 60;
 
  
   private volatile long dataSize = 0;
   private volatile boolean isRunning = false;
   
   private Timer rotateTimer = null;
   private Timer statTimer = null;
   
   private Schema avroSchema = null;
   private int initWriteChunkRetries = 10;
   private int writeChunkRetries = initWriteChunkRetries;
   private boolean chunksWrittenThisRotate = false;
 
   private long timePeriod = -1;
   private long nextTimePeriodComputation = -1;
   private int minPercentFreeDisk = 20;
   
   static {
     try {
       localHostAddr = "_" + InetAddress.getLocalHost().getHostName() + "_";
     } catch (UnknownHostException e) {
       localHostAddr = "-NA-";
     }
   }
 
   public LocalWriter(Configuration conf) throws WriterException {
     setup(conf);
   }
 
   public void init(Configuration conf) throws WriterException {
   }
 
   public void setup(Configuration conf) throws WriterException {
     this.conf = conf;
 
     // load Chukwa Avro schema
     avroSchema = ChukwaAvroSchema.getSchema();
 
     try {
       fs = FileSystem.getLocal(conf);
       localOutputDir = conf.get("chukwaCollector.localOutputDir",
           "/chukwa/datasink/");
       if (!localOutputDir.endsWith("/")) {
         localOutputDir += "/";
       }
       Path pLocalOutputDir = new Path(localOutputDir);
       if (!fs.exists(pLocalOutputDir)) {
         boolean exist = fs.mkdirs(pLocalOutputDir);
         if (!exist) {
           throw new WriterException("Cannot create local dataSink dir: "
               + localOutputDir);
         }
       } else {
         FileStatus fsLocalOutputDir = fs.getFileStatus(pLocalOutputDir);
         if (!fsLocalOutputDir.isDir()) {
           throw new WriterException("local dataSink dir is not a directory: "
               + localOutputDir);
         }
       }
     } catch (Throwable e) {
       log.fatal("Cannot initialize LocalWriter", e);
       throw new WriterException(e);
     }
 
     
     minPercentFreeDisk = conf.getInt("chukwaCollector.minPercentFreeDisk",20);
     
     rotateInterval = conf.getInt("chukwaCollector.rotateInterval",
         1000 * 60 * 5);// defaults to 5 minutes
    
     initWriteChunkRetries = conf
         .getInt("chukwaCollector.writeChunkRetries", 10);
     writeChunkRetries = initWriteChunkRetries;
 
     log.info("rotateInterval is " + rotateInterval);
     log.info("outputDir is " + localOutputDir);
     log.info("localFileSystem is " + fs.getUri().toString());
     log.info("minPercentFreeDisk is " + minPercentFreeDisk);
 
     if(rotateTimer==null) {
       rotateTimer = new Timer();
       rotateTimer.schedule(new RotateTask(), 0,
         rotateInterval);
     }
     if(statTimer==null) {
       statTimer = new Timer();
       statTimer.schedule(new StatReportingTask(), 0,
         STAT_INTERVAL_SECONDS * 1000);
     }
     fileQueue = new LinkedBlockingQueue<String>();
     localToRemoteHdfsMover = new LocalToRemoteHdfsMover(fileQueue, conf);
     
   }
 
   private class RotateTask extends TimerTask {
         public void run() {
           try {
             rotate();
           } catch(WriterException e) {
             log.error(ExceptionUtil.getStackTrace(e));
           }
       };
   }
   
   private class StatReportingTask extends TimerTask {
     private long lastTs = System.currentTimeMillis();
 
     public void run() {
 
       long time = System.currentTimeMillis();
       long currentDs = dataSize;
       dataSize = 0;
 
       long interval = time - lastTs;
       lastTs = time;
 
       if(interval <= 0) {
         interval = 1;
       }
       long dataRate = 1000 * currentDs / interval; // kb/sec
       log.info("stat:datacollection.writer.local.LocalWriter dataSize="
           + currentDs + " dataRate=" + dataRate);
     }
   };
 
   protected void computeTimePeriod() {
     synchronized (calendar) {
       calendar.setTimeInMillis(System.currentTimeMillis());
       calendar.set(Calendar.MINUTE, 0);
       calendar.set(Calendar.SECOND, 0);
       calendar.set(Calendar.MILLISECOND, 0);
       timePeriod = calendar.getTimeInMillis();
       calendar.add(Calendar.HOUR, 1);
       nextTimePeriodComputation = calendar.getTimeInMillis();
     }
   }
 
 
   /**
    *  Best effort, there's no guarantee that chunks 
    *  have really been written to disk
    */
   public CommitStatus add(List<Chunk> chunks) throws WriterException {
     if (!isRunning) {
       throw new WriterException("Writer not yet ready");
     }
     long now = System.currentTimeMillis();
     if (chunks != null) {
       try {
         chunksWrittenThisRotate = true;
         ChukwaArchiveKey archiveKey = new ChukwaArchiveKey();
 
         synchronized (lock) {
           if (System.currentTimeMillis() >= nextTimePeriodComputation) {
             computeTimePeriod();
           }
 
           for (Chunk chunk : chunks) {
             archiveKey.setTimePartition(timePeriod);
             archiveKey.setDataType(chunk.getDataType());
             archiveKey.setStreamName(chunk.getTags() + "/" + chunk.getSource()
                 + "/" + chunk.getStreamName());
             archiveKey.setSeqId(chunk.getSeqID());
             GenericRecord record = new GenericData.Record(avroSchema);
             record.put("dataType", chunk.getDataType());
             record.put("data", ByteBuffer.wrap(chunk.getData()));
             record.put("tags", chunk.getTags());
             record.put("seqId", chunk.getSeqID());
             record.put("source", chunk.getSource());
             record.put("stream", chunk.getStreamName());
             parquetWriter.write(record);
             // compute size for stats
             dataSize += chunk.getData().length;
           }
         }// End synchro
         long end = System.currentTimeMillis();
         if (log.isDebugEnabled()) {
           log.debug("duration=" + (end-now) + " size=" + chunks.size());
         }
         
       } catch (IOException e) {
         writeChunkRetries--;
         log.error("Could not save the chunk. ", e);
 
         if (writeChunkRetries < 0) {
           log
               .fatal("Too many IOException when trying to write a chunk, Collector is going to exit!");
         }
         throw new WriterException(e);
       }
     }
     return COMMIT_OK;
   }
 
   protected String getNewFileName() {
     calendar.setTimeInMillis(System.currentTimeMillis());
     String newName = new java.text.SimpleDateFormat("yyyyddHHmmssSSS")
     .format(calendar.getTime());
     newName += localHostAddr + new java.rmi.server.UID().toString();
     newName = newName.replace("-", "");
     newName = newName.replace(":", "");
     newName = newName.replace(".", "");
     newName = localOutputDir + "/" + newName.trim();
     return newName;
   }
 
   protected void rotate() throws WriterException {
     isRunning = true;
     log.info("start Date [" + calendar.getTime() + "]");
     log.info("Rotate from " + Thread.currentThread().getName());
 
     String newName = getNewFileName();
 
     synchronized (lock) {
       try {
         if (currentPath != null) {
           Path previousPath = currentPath;
           if (chunksWrittenThisRotate) {
             String previousFileName = previousPath.getName().replace(".chukwa", ".done");
             if(fs.exists(previousPath)) {
               fs.rename(previousPath, new Path(previousFileName + ".done"));
             }
             fileQueue.add(previousFileName + ".done");
           } else {
             log.info("no chunks written to " + previousPath + ", deleting");
             fs.delete(previousPath, false);
           }
         }
         
         Path newOutputPath = new Path(newName + ".chukwa");
         while(fs.exists(newOutputPath)) {
           newName = getNewFileName();
           newOutputPath = new Path(newName + ".chukwa");
         }
 
         currentPath = newOutputPath;
         currentFileName = newName;
         chunksWrittenThisRotate = false;
         parquetWriter = new AvroParquetWriter<GenericRecord>(newOutputPath, avroSchema, CompressionCodecName.SNAPPY, blockSize, pageSize);
 
       } catch (IOException e) {
         log.fatal("IO Exception in rotate: ", e);
       }
     }
  
     // Check for disk space
     File directory4Space = new File(localOutputDir);
     long totalSpace = directory4Space.getTotalSpace();
     long freeSpace = directory4Space.getFreeSpace();
     long minFreeAvailable = (totalSpace * minPercentFreeDisk) /100;
     
     if (log.isDebugEnabled()) {
       log.debug("Directory: " + localOutputDir + ", totalSpace: " + totalSpace 
           + ", freeSpace: " + freeSpace + ", minFreeAvailable: " + minFreeAvailable
           + ", percentFreeDisk: " + minPercentFreeDisk);
     }
   
     if (freeSpace < minFreeAvailable) {
       log.fatal("No space left on device.");
       throw new WriterException("No space left on device.");
     } 
     
     log.debug("finished rotate()");
   }
 
   public void close() {
     synchronized (lock) {
   
       if (rotateTimer != null) {
         rotateTimer.cancel();
       }
 
       if (statTimer != null) {
         statTimer.cancel();
       }
 
       try {
         if (parquetWriter != null) {
           parquetWriter.close();
         }
         if (localToRemoteHdfsMover != null) {
           localToRemoteHdfsMover.shutdown();
         }
         
         fs.rename(currentPath, new Path(currentFileName + ".done"));
       } catch (IOException e) {
         log.error("failed to close and rename stream", e);
       }
     }
   }
 }