/*
 * 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;


import java.net.InetAddress;
import java.net.URI;
import java.net.UnknownHostException;
import java.util.Calendar;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.io.IOException;

import org.apache.hadoop.chukwa.ChukwaArchiveKey;
import org.apache.hadoop.chukwa.Chunk;
import org.apache.hadoop.chukwa.ChunkImpl;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.log4j.Logger;

/**
 * This class <b>is</b> thread-safe -- rotate() and save() both synchronize on
 * this object.
 * 
 */
public class SeqFileWriter extends PipelineableWriter implements ChukwaWriter {
  static Logger log = Logger.getLogger(SeqFileWriter.class);
  private static boolean ENABLE_ROTATION_ON_CLOSE = true;

  protected int STAT_INTERVAL_SECONDS = 30;
  private int rotateInterval = 1000 * 60 * 5;
  private int offsetInterval = 1000 * 30;
  private boolean if_fixed_interval = false;
  static final int ACQ_WAIT_ON_TERM = 500; //ms to wait for lock on a SIGTERM before aborting
  
  public static final String STAT_PERIOD_OPT = "chukwaCollector.stats.period";
  public static final String ROTATE_INTERVAL_OPT = "chukwaCollector.rotateInterval";
  public static final String IF_FIXED_INTERVAL_OPT = "chukwaCollector.isFixedTimeRotatorScheme";
  public static final String FIXED_INTERVAL_OFFSET_OPT = "chukwaCollector.fixedTimeIntervalOffset";
  public static final String OUTPUT_DIR_OPT= "chukwaCollector.outputDir";
  public String localHostAddr = null;
  
  protected final Semaphore lock = new Semaphore(1, true);
  
  protected FileSystem fs = null;
  protected Configuration conf = null;

  protected String outputDir = null;
  private Calendar calendar = Calendar.getInstance();

  protected Path currentPath = null;
  protected String currentFileName = null;
  protected FSDataOutputStream currentOutputStr = null;
  protected SequenceFile.Writer seqFileWriter = null;

  protected long timePeriod = -1;
  protected long nextTimePeriodComputation = -1;
  
  protected Timer rotateTimer = null;  
  protected Timer statTimer = null;
  
  protected volatile long dataSize = 0;
  protected volatile long bytesThisRotate = 0;
  protected volatile boolean isRunning = false;

  public SeqFileWriter() {
    try {
      localHostAddr = "_" + InetAddress.getLocalHost().getHostName() + "_";
    } catch (UnknownHostException e) {
      localHostAddr = "-NA-";
    }
  }
  
  public long getBytesWritten() {
    return dataSize;
  }
  
  public void init(Configuration conf) throws WriterException {
    outputDir = conf.get(OUTPUT_DIR_OPT, "/chukwa");

    this.conf = conf;

    rotateInterval = conf.getInt(ROTATE_INTERVAL_OPT,rotateInterval);
    if_fixed_interval = conf.getBoolean(IF_FIXED_INTERVAL_OPT,if_fixed_interval);
    offsetInterval = conf.getInt(FIXED_INTERVAL_OFFSET_OPT,offsetInterval);
    
    STAT_INTERVAL_SECONDS = conf.getInt(STAT_PERIOD_OPT, STAT_INTERVAL_SECONDS);

    // check if they've told us the file system to use
    String fsname = conf.get("writer.hdfs.filesystem");
    if (fsname == null || fsname.equals("")) {
      // otherwise try to get the filesystem from hadoop
      fsname = conf.get("fs.defaultFS");
    }

    log.info("rotateInterval is " + rotateInterval);
    if(if_fixed_interval)
      log.info("using fixed time interval scheme, " +
              "offsetInterval is " + offsetInterval);
    else
      log.info("not using fixed time interval scheme");
    log.info("outputDir is " + outputDir);
    log.info("fsname is " + fsname);
    log.info("filesystem type from core-default.xml is "
        + conf.get("fs.hdfs.impl"));

    if (fsname == null) {
      log.error("no filesystem name");
      throw new WriterException("no filesystem");
    }
    try {
      fs = FileSystem.get(new URI(fsname), conf);
      if (fs == null) {
        log.error("can't connect to HDFS.");
      }
    } catch (Throwable e) {
      log.error(
          "can't connect to HDFS, trying default file system instead (likely to be local)",
          e);
    }

    // Setup everything by rotating

    isRunning = true;
    rotate();

    statTimer = new Timer();
    statTimer.schedule(new StatReportingTask(), 1000,
        STAT_INTERVAL_SECONDS * 1000);

  }

  public 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;

      long dataRate = 1000 * currentDs / interval; // kb/sec
      log.info("stat:datacollection.writer.hdfs dataSize=" + currentDs
          + " dataRate=" + dataRate);
    }
    
    public StatReportingTask() {}
  };

  void rotate() {
     if (rotateTimer != null) {
      rotateTimer.cancel();
    } 
     
    if(!isRunning)
      return;
    
    calendar.setTimeInMillis(System.currentTimeMillis());

    String newName = new java.text.SimpleDateFormat("yyyyMMddHHmmssSSS")
        .format(calendar.getTime());
    newName += localHostAddr + new java.rmi.server.UID().toString();
    newName = newName.replace("-", "");
    newName = newName.replace(":", "");
    newName = newName.replace(".", "");
    newName = outputDir + "/" + newName.trim();

    try {
      lock.acquire();

      FSDataOutputStream previousOutputStr = currentOutputStr;
      Path previousPath = currentPath;
      String previousFileName = currentFileName;

      if (previousOutputStr != null) {
        boolean closed = false;
        try {
          log.info("closing sink file" + previousFileName);
          previousOutputStr.close();
          closed = true;
        }catch (Throwable e) {
          log.error("couldn't close file" + previousFileName, e);
          //we probably have an orphaned 0 byte file at this point due to an
          //intermitant HDFS outage. Once HDFS comes up again we'll be able to
          //close it, although it will be empty.
        }

        if (bytesThisRotate > 0) {
          if (closed) {
            log.info("rotating sink file " + previousPath);
            fs.rename(previousPath, new Path(previousFileName + ".done"));
          }
          else {
            log.warn(bytesThisRotate + " bytes potentially lost, since " +
                    previousPath + " could not be closed.");
          }
        } else {
          log.info("no chunks written to " + previousPath + ", deleting");
          fs.delete(previousPath, false);
        }
      }

      Path newOutputPath = new Path(newName + ".chukwa");
      FSDataOutputStream newOutputStr = fs.create(newOutputPath);
      // Uncompressed for now
      seqFileWriter = SequenceFile.createWriter(conf, newOutputStr,
          ChukwaArchiveKey.class, ChunkImpl.class,
          SequenceFile.CompressionType.NONE, null);

      // reset these once we know that seqFileWriter was created
      currentOutputStr = newOutputStr;
      currentPath = newOutputPath;
      currentFileName = newName;
      bytesThisRotate = 0;
    } catch (Throwable e) {
      log.warn("Got an exception trying to rotate. Will try again in " +
              rotateInterval/1000 + " seconds." ,e);
    } finally {
      lock.release();
    }
    
    // Schedule the next timer
    scheduleNextRotation();

  }

  /**
   * Schedules the rotate task using either a fixed time interval scheme or a
   * relative time interval scheme as specified by the
   * chukwaCollector.isFixedTimeRotatorScheme configuration parameter. If the
   * value of this parameter is true then next rotation will be scheduled at a
   * fixed offset from the current rotateInterval. This fixed offset is
   * provided by chukwaCollector.fixedTimeIntervalOffset configuration
   * parameter.
   */
  void scheduleNextRotation(){
    long delay = rotateInterval;
    if (if_fixed_interval) {
      long currentTime = System.currentTimeMillis();
      delay = getDelayForFixedInterval(currentTime, rotateInterval, offsetInterval);
    }
    rotateTimer = new Timer();
    rotateTimer.schedule(new TimerTask() {
      public void run() {
        rotate();
      }
    }, delay);
  }

  /**
   * Calculates delay for scheduling the next rotation in case of
   * FixedTimeRotatorScheme. This delay is the time difference between the
   * currentTimestamp (t1) and the next time the collector should rotate the
   * sequence files (t2). t2 is the time when the current rotateInterval ends
   * plus an offset (as set by chukwaCollector.FixedTimeIntervalOffset).
   * So, delay = t2 - t1
   *
   * @param currentTime - the current timestamp
   * @param rotateInterval - chukwaCollector.rotateInterval
   * @param offsetInterval - chukwaCollector.fixedTimeIntervalOffset
   * @return delay for scheduling next rotation
   */
  long getDelayForFixedInterval(long currentTime, long rotateInterval, long offsetInterval){
    // time since last rounded interval
    long remainder = (currentTime % rotateInterval);
    long prevRoundedInterval = currentTime - remainder;
    long nextRoundedInterval = prevRoundedInterval + rotateInterval;
    long delay = nextRoundedInterval - currentTime + offsetInterval;

    if (log.isInfoEnabled()) {
     log.info("currentTime="+currentTime+" prevRoundedInterval="+
             prevRoundedInterval+" nextRoundedInterval" +
            "="+nextRoundedInterval+" delay="+delay);
    }

    return delay;
  }


  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();
    }
  }
  
  @Override
  public CommitStatus add(List<Chunk> chunks) throws WriterException {
    COMMIT_PENDING result = new COMMIT_PENDING(chunks.size());
    if (!isRunning) {
      log.info("Collector not ready");
      throw new WriterException("Collector not ready");
    }

    ChukwaArchiveKey archiveKey = new ChukwaArchiveKey();
    
    if (System.currentTimeMillis() >= nextTimePeriodComputation) {
      computeTimePeriod();
    }
    try {
      lock.acquire();
      for (Chunk chunk : chunks) {
        archiveKey.setTimePartition(timePeriod);
        archiveKey.setDataType(chunk.getDataType());
        archiveKey.setStreamName(chunk.getTags() + "/" + chunk.getSource()
            + "/" + chunk.getStreamName());
        archiveKey.setSeqId(chunk.getSeqID());

        seqFileWriter.append(archiveKey, chunk);

        // compute size for stats only if append succeeded. Note though that
        // seqFileWriter.append can continue taking data for quite some time
        // after HDFS goes down while the client is trying to reconnect. Hence
        // these stats might not reflect reality during an HDFS outage.
        dataSize += chunk.getData().length;
        bytesThisRotate += chunk.getData().length;

        String futureName = currentPath.getName().replace(".chukwa", ".done");
        result.addPend(futureName, currentOutputStr.getPos());

      }
    }
    catch (IOException e) {
      log.error("IOException when trying to write a chunk, Collector will return error and keep running.", e);
      return COMMIT_FAIL;
    }
    catch (Throwable e) {
      // We don't want to loose anything
      log.fatal("IOException when trying to write a chunk, Collector is going to exit!", e);
      isRunning = false;
    } finally {
      lock.release();
    }
    return result;
  }

  public void close() {
    
    isRunning = false;

    if (statTimer != null) {
      statTimer.cancel();
    }

    if (rotateTimer != null) {
      rotateTimer.cancel();
    }

    // If we are here it's either because of an HDFS exception
    // or Collector has received a kill -TERM
    boolean gotLock = false;
    try {
      gotLock = lock.tryAcquire(ACQ_WAIT_ON_TERM, TimeUnit.MILLISECONDS);
      if(gotLock) {
        
        if (this.currentOutputStr != null) {
          this.currentOutputStr.close();
        }
        if(ENABLE_ROTATION_ON_CLOSE)
          if(bytesThisRotate > 0)
            fs.rename(currentPath, new Path(currentFileName + ".done"));
          else
            fs.delete(currentPath, false);
      }
    } catch (Throwable e) {
     log.warn("cannot rename dataSink file:" + currentPath,e);
    } finally {
      if(gotLock)
        lock.release();
    }
  }
  
  public static void setEnableRotationOnClose(boolean b) {
    ENABLE_ROTATION_ON_CLOSE = b;
  }

}