/*
    * 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.
   */
  
  /* Portions copyright (C) The Apache Software Foundation. All rights reserved.
   *
   * This software is published under the terms of the Apache Software
   * License version 1.1, a copy of which has been included with this
   * distribution in the LICENSE.txt file.  */
  
  package org.apache.hadoop.chukwa.inputtools.log4j;
  
  
  import java.io.File;
  import java.io.FilenameFilter;
  import java.io.IOException;
  import java.text.SimpleDateFormat;
  import java.util.ArrayList;
  import java.util.Calendar;
  import java.util.Collections;
  import java.util.Date;
  import java.util.GregorianCalendar;
  import java.util.List;
  import java.util.Locale;
  import java.util.TimeZone;
  import java.util.regex.Pattern;
  
  import org.apache.hadoop.chukwa.datacollection.controller.ChukwaAgentController;
  import org.apache.hadoop.chukwa.datacollection.controller.ClientFinalizer;
  import org.apache.hadoop.chukwa.util.AdaptorNamingUtils;
  import org.apache.hadoop.chukwa.util.RecordConstants;
  import org.apache.log4j.FileAppender;
  import org.apache.log4j.Layout;
  import org.apache.log4j.Logger;
  import org.apache.log4j.helpers.LogLog;
  import org.apache.log4j.spi.LoggingEvent;
  
  /**
      ChukwaDailyRollingFileAppender is a slightly modified version of
      DailyRollingFileAppender, with modified versions of its
      <code>subAppend()</code> and <code>rollOver()</code> functions. 
      We would have preferred to sub-class DailyRollingFileAppender but
      its authors clearly did not intend that to be a viable option since
      they made too much of the class private or package-private
  
      DailyRollingFileAppender extends {@link FileAppender} so that the
      underlying file is rolled over at a user chosen frequency.
  
      <p>The rolling schedule is specified by the <b>DatePattern</b>
      option. This pattern should follow the {@link SimpleDateFormat}
      conventions. In particular, you <em>must</em> escape literal text
      within a pair of single quotes. A formatted version of the date
      pattern is used as the suffix for the rolled file name.
  
      <p>For example, if the <b>File</b> option is set to
      <code>/foo/bar.log</code> and the <b>DatePattern</b> set to
      <code>'.'yyyy-MM-dd</code>, on 2001-02-16 at midnight, the logging
      file <code>/foo/bar.log</code> will be copied to
      <code>/foo/bar.log.2001-02-16</code> and logging for 2001-02-17
      will continue in <code>/foo/bar.log</code> until it rolls over
      the next day.
  
      <p>Is is possible to specify monthly, weekly, half-daily, daily,
      hourly, or minutely rollover schedules.
  
      <p><table border="1" cellpadding="2">
      <tr>
      <th>DatePattern</th>
      <th>Rollover schedule</th>
      <th>Example</th>
  
      <tr>
      <td><code>'.'yyyy-MM</code>
      <td>Rollover at the beginning of each month</td>
  
      <td>At midnight of May 31st, 2002 <code>/foo/bar.log</code> will be
      copied to <code>/foo/bar.log.2002-05</code>. Logging for the month
      of June will be output to <code>/foo/bar.log</code> until it is
      also rolled over the next month.
  
      <tr>
      <td><code>'.'yyyy-ww</code>
  
      <td>Rollover at the first day of each week. The first day of the
      week depends on the locale.</td>
 
     <td>Assuming the first day of the week is Sunday, on Saturday
     midnight, June 9th 2002, the file <i>/foo/bar.log</i> will be
     copied to <i>/foo/bar.log.2002-23</i>.  Logging for the 24th week
     of 2002 will be output to <code>/foo/bar.log</code> until it is
     rolled over the next week.
 
     <tr>
     <td><code>'.'yyyy-MM-dd</code>
 
     <td>Rollover at midnight each day.</td>
 
     <td>At midnight, on March 8th, 2002, <code>/foo/bar.log</code> will
     be copied to <code>/foo/bar.log.2002-03-08</code>. Logging for the
     9th day of March will be output to <code>/foo/bar.log</code> until
     it is rolled over the next day.
 
     <tr>
     <td><code>'.'yyyy-MM-dd-a</code>
 
     <td>Rollover at midnight and midday of each day.</td>
 
     <td>At noon, on March 9th, 2002, <code>/foo/bar.log</code> will be
     copied to <code>/foo/bar.log.2002-03-09-AM</code>. Logging for the
     afternoon of the 9th will be output to <code>/foo/bar.log</code>
     until it is rolled over at midnight.
 
     <tr>
     <td><code>'.'yyyy-MM-dd-HH</code>
 
     <td>Rollover at the top of every hour.</td>
 
     <td>At approximately 11:00.000 o'clock on March 9th, 2002,
     <code>/foo/bar.log</code> will be copied to
     <code>/foo/bar.log.2002-03-09-10</code>. Logging for the 11th hour
     of the 9th of March will be output to <code>/foo/bar.log</code>
     until it is rolled over at the beginning of the next hour.
 
 
     <tr>
     <td><code>'.'yyyy-MM-dd-HH-mm</code>
 
     <td>Rollover at the beginning of every minute.</td>
 
     <td>At approximately 11:23,000, on March 9th, 2001,
     <code>/foo/bar.log</code> will be copied to
     <code>/foo/bar.log.2001-03-09-10-22</code>. Logging for the minute
     of 11:23 (9th of March) will be output to
     <code>/foo/bar.log</code> until it is rolled over the next minute.
 
     </table>
 
     <p>Do not use the colon ":" character in anywhere in the
     <b>DatePattern</b> option. The text before the colon is interpeted
     as the protocol specificaion of a URL which is probably not what
     you want. */
 
 public class ChukwaDailyRollingFileAppender extends FileAppender {
 
   static Logger log = Logger.getLogger(ChukwaDailyRollingFileAppender.class);
   // The code assumes that the following constants are in a increasing
   // sequence.
   static final int TOP_OF_TROUBLE = -1;
   static final int TOP_OF_MINUTE = 0;
   static final int TOP_OF_HOUR = 1;
   static final int HALF_DAY = 2;
   static final int TOP_OF_DAY = 3;
   static final int TOP_OF_WEEK = 4;
   static final int TOP_OF_MONTH = 5;
 
   static final String adaptorType = ChukwaAgentController.CharFileTailUTF8NewLineEscaped;
 
   static final Object lock = new Object();
   static String lastRotation = "";
 
   /**
    * The date pattern. By default, the pattern is set to "'.'yyyy-MM-dd" meaning
    * daily rollover.
    */
   private String datePattern = "'.'yyyy-MM-dd";
 
   /**
    * The log file will be renamed to the value of the scheduledFilename variable
    * when the next interval is entered. For example, if the rollover period is
    * one hour, the log file will be renamed to the value of "scheduledFilename"
    * at the beginning of the next hour.
    * 
    * The precise time when a rollover occurs depends on logging activity.
    */
   private String scheduledFilename;
 
   /**
    * The next time we estimate a rollover should occur.
    */
   private long nextCheck = System.currentTimeMillis() - 1;
 
   /**
    * Regex to select log files to be deleted
    */
   private String cleanUpRegex = null;
 
   /**
    * Set the maximum number of backup files to keep around.
    */
   private int maxBackupIndex = 10;
 
   private ClientFinalizer clientFinalizer = null;
   
   boolean hasBeenActivated = false;
   Date now = new Date();
 
   SimpleDateFormat sdf;
 
   RollingCalendar rc = new RollingCalendar();
 
   int checkPeriod = TOP_OF_TROUBLE;
 
   ChukwaAgentController chukwaClient;
   boolean chukwaClientIsNull = true;
   static final Object chukwaLock = new Object();
 
   String chukwaClientHostname;
   int chukwaClientPortNum;
   long chukwaClientConnectNumRetry;
   long chukwaClientConnectRetryInterval;
 
   String recordType;
 
   // The gmtTimeZone is used only in computeCheckPeriod() method.
   static final TimeZone gmtTimeZone = TimeZone.getTimeZone("GMT");
 
   /**
    * The default constructor does nothing.
    */
   public ChukwaDailyRollingFileAppender() throws IOException {
     super();
   }
 
 /**
      Instantiate a <code>DailyRollingFileAppender</code> and open the
      file designated by <code>filename</code>. The opened filename will
      become the output destination for this appender.
 
    */
   public ChukwaDailyRollingFileAppender(Layout layout, String filename,
                                         String datePattern) throws IOException {
     super(layout, filename, true);
     System.out
         .println("Daily Rolling File Appender successfully registered file with agent: "
             + filename);
     this.datePattern = datePattern;
   }
 
   /**
    * The <b>DatePattern</b> takes a string in the same format as expected by
    * {@link SimpleDateFormat}. This options determines the rollover schedule.
    */
   public void setDatePattern(String pattern) {
     datePattern = pattern;
   }
 
   /** Returns the value of the <b>DatePattern</b> option. */
   public String getDatePattern() {
     return datePattern;
   }
 
   public String getRecordType() {
     if (recordType != null)
       return recordType;
     else
       return "unknown";
   }
 
   public void setRecordType(String recordType) {
     this.recordType = recordType;
   }
 
   public void activateOptions() {
     
     // Prevent early initialisation
     if (!hasBeenActivated)
     { return;}
     
     super.activateOptions();
     if (datePattern != null && fileName != null) {
       now.setTime(System.currentTimeMillis());
       sdf = new SimpleDateFormat(datePattern);
       int type = computeCheckPeriod();
       printPeriodicity(type);
       rc.setType(type);
       File file = new File(fileName);
       scheduledFilename = fileName + sdf.format(new Date(file.lastModified()));
 
     } else {
       LogLog
           .error("Either File or DatePattern options are not set for appender ["
               + name + "].");
     }
   }
 
   void printPeriodicity(int type) {
     switch (type) {
     case TOP_OF_MINUTE:
       LogLog.debug("Appender [" + name + "] to be rolled every minute.");
       break;
     case TOP_OF_HOUR:
       LogLog
           .debug("Appender [" + name + "] to be rolled on top of every hour.");
       break;
     case HALF_DAY:
       LogLog.debug("Appender [" + name
           + "] to be rolled at midday and midnight.");
       break;
     case TOP_OF_DAY:
       LogLog.debug("Appender [" + name + "] to be rolled at midnight.");
       break;
     case TOP_OF_WEEK:
       LogLog.debug("Appender [" + name + "] to be rolled at start of week.");
       break;
     case TOP_OF_MONTH:
       LogLog.debug("Appender [" + name
           + "] to be rolled at start of every month.");
       break;
     default:
       LogLog.warn("Unknown periodicity for appender [" + name + "].");
     }
   }
 
   // This method computes the roll over period by looping over the
   // periods, starting with the shortest, and stopping when the r0 is
   // different from from r1, where r0 is the epoch formatted according
   // the datePattern (supplied by the user) and r1 is the
   // epoch+nextMillis(i) formatted according to datePattern. All date
   // formatting is done in GMT and not local format because the test
   // logic is based on comparisons relative to 1970-01-01 00:00:00
   // GMT (the epoch).
 
   int computeCheckPeriod() {
     RollingCalendar rollingCalendar = new RollingCalendar(gmtTimeZone,
         Locale.ENGLISH);
     // set sate to 1970-01-01 00:00:00 GMT
     Date epoch = new Date(0);
     if (datePattern != null) {
       for (int i = TOP_OF_MINUTE; i <= TOP_OF_MONTH; i++) {
         SimpleDateFormat simpleDateFormat = new SimpleDateFormat(datePattern);
         simpleDateFormat.setTimeZone(gmtTimeZone); // do all date formatting in
                                                    // GMT
         String r0 = simpleDateFormat.format(epoch);
         rollingCalendar.setType(i);
         Date next = new Date(rollingCalendar.getNextCheckMillis(epoch));
         String r1 = simpleDateFormat.format(next);
         // System.out.println("Type = "+i+", r0 = "+r0+", r1 = "+r1);
         if (r0 != null && r1 != null && !r0.equals(r1)) {
           return i;
         }
       }
     }
     return TOP_OF_TROUBLE; // Deliberately head for trouble...
   }
 
   /**
    * Rollover the current file to a new file.
    */
   void rollOver() throws IOException {
 
     /* Compute filename, but only if datePattern is specified */
     if (datePattern == null) {
       errorHandler.error("Missing DatePattern option in rollOver().");
       return;
     }
 
     String datedFilename = fileName + sdf.format(now);
     // It is too early to roll over because we are still within the
     // bounds of the current interval. Rollover will occur once the
     // next interval is reached.
     if (scheduledFilename.equals(datedFilename)) {
       return;
     }
 
     // close current file, and rename it to datedFilename
     this.closeFile();
 
     File target = new File(scheduledFilename);
     if (target.exists()) {
       target.delete();
     }
 
     File file = new File(fileName);
 
     boolean result = file.renameTo(target);
     if (result) {
       LogLog.debug(fileName + " -> " + scheduledFilename);
     } else {
       LogLog.error("Failed to rename [" + fileName + "] to ["
           + scheduledFilename + "].");
     }
 
     try {
       // This will also close the file. This is OK since multiple
       // close operations are safe.
       this.setFile(fileName, false, this.bufferedIO, this.bufferSize);
     } catch (IOException e) {
       errorHandler.error("setFile(" + fileName + ", false) call failed.");
     }
     scheduledFilename = datedFilename;
     cleanUp();
   }
 
   public String getCleanUpRegex() {
     return cleanUpRegex;
   }
 
   public void setCleanUpRegex(String cleanUpRegex) {
     this.cleanUpRegex = cleanUpRegex;
   }
 
   public int getMaxBackupIndex() {
     return maxBackupIndex;
   }
 
   public void setMaxBackupIndex(int maxBackupIndex) {
     this.maxBackupIndex = maxBackupIndex;
   }
 
   protected synchronized void cleanUp() {
     String regex = "";
     try {
       File actualFile = new File(fileName);
 
       String directoryName = actualFile.getParent();
       String actualFileName = actualFile.getName();
       File dirList = new File(directoryName);
 
       if (cleanUpRegex == null || !cleanUpRegex.contains("$fileName")) {
         LogLog
             .error("cleanUpRegex == null || !cleanUpRegex.contains(\"$fileName\")");
         return;
       }
       regex = cleanUpRegex.replace("$fileName", actualFileName);
       String[] dirFiles = dirList.list(new LogFilter(actualFileName, regex));
 
       List<String> files = new ArrayList<String>();
       for (String file : dirFiles) {
         files.add(file);
       }
       Collections.sort(files);
 
       while (files.size() > maxBackupIndex) {
         String file = files.remove(0);
         File f = new File(directoryName + "/" + file);
         f.delete();
         LogLog.debug("Removing: " + file);
       }
     } catch (Exception e) {
       errorHandler
           .error("cleanUp(" + fileName + "," + regex + ") call failed.");
     }
   }
 
   private class LogFilter implements FilenameFilter {
     private Pattern p = null;
     private String logFile = null;
 
     public LogFilter(String logFile, String regex) {
       this.logFile = logFile;
       p = Pattern.compile(regex);
     }
 
     @Override
     public boolean accept(File dir, String name) {
       // ignore current log file
       if (name.intern() == this.logFile.intern()) {
         return false;
       }
       // ignore file without the same prefix
       if (!name.startsWith(logFile)) {
         return false;
       }
       return p.matcher(name).find();
     }
   }
 
   /**
    * Fix for late-initialisation
    */
   @Override
   protected boolean checkEntryConditions() {
     if (!hasBeenActivated) {
       synchronized(chukwaLock) {
         if (!hasBeenActivated) {
           hasBeenActivated = true;
           activateOptions();
         }
       }
     }
     return super.checkEntryConditions();
   }
 
   /**
    * This method differentiates DailyRollingFileAppender from its super class.
    * 
    * <p>Before actually logging, this method will check whether it is time to do
    * a rollover. If it is, it will schedule the next rollover time and then
    * rollover.
    * */
   protected void subAppend(LoggingEvent event) {
     try {
       // we set up the chukwa adaptor here because this is the first
       // point which is called after all setters have been called with
       // their values from the log4j.properties file, in particular we
       // needed to give setCukwaClientPortNum() and -Hostname() a shot
 
       // Make sure only one thread can do this
       // and use the boolean to avoid the first level locking
       if (chukwaClientIsNull) {
         synchronized (chukwaLock) {
 
           String log4jFileName = getFile(); 
           String recordType = getRecordType();
 
           long currentLength = 0L;
           try {
             File fooLog = new File(log4jFileName);
             log4jFileName = fooLog.getAbsolutePath();
             currentLength = fooLog.length();
           } catch (Throwable e) {
             log.warn("Exception while trying to get current file size for " + log4jFileName);
             currentLength = 0L;
           }
 
           if (chukwaClient == null) {
             if (getChukwaClientHostname() != null
                 && getChukwaClientPortNum() != 0) {
               chukwaClient = new ChukwaAgentController(
                   getChukwaClientHostname(), getChukwaClientPortNum());
               log.debug("setup adaptor with hostname "
                   + getChukwaClientHostname() + " and portnum "
                   + getChukwaClientPortNum());
             } else {
               chukwaClient = new ChukwaAgentController();
               log
               .debug("setup adaptor with no args, which means it used its defaults");
             }
 
             chukwaClientIsNull = false;
 
             // Watchdog is watching for ChukwaAgent only once every 5 minutes,
             // so there's no point in retrying more than once every 5 mins.
             // In practice, if the watchdog is not able to automatically restart
             // the agent, it will take more than 20 minutes to get Ops to
             // restart it.
             // Also its a good to limit the number of communications between
             // Hadoop and Chukwa, that's why 30 minutes.
             long retryInterval = chukwaClientConnectRetryInterval;
             if (retryInterval == 0) {
               retryInterval = 1000 * 60 * 30;
             }
             long numRetries = chukwaClientConnectNumRetry;
             if (numRetries == 0) {
               numRetries = 48;
             }
 
             String name = AdaptorNamingUtils.synthesizeAdaptorID
               (ChukwaAgentController.CharFileTailUTF8NewLineEscaped, recordType, log4jFileName);
             
             String adaptorID = chukwaClient.addByName(name, ChukwaAgentController.CharFileTailUTF8NewLineEscaped,
                 recordType,currentLength + " " + log4jFileName, currentLength,
                 numRetries, retryInterval);
 
             // Setup a shutdownHook for the controller
             clientFinalizer = new ClientFinalizer(chukwaClient);
             Runtime.getRuntime().addShutdownHook(clientFinalizer);
 
             if (adaptorID != null) {
               log.debug("Added file tailing adaptor to chukwa agent for file "
                   + log4jFileName + ", adaptorId:" + adaptorID 
                   + " using this recordType :" + recordType 
                   + ", starting at offset:" + currentLength);
             } else {
               log.debug("Chukwa adaptor not added, addFile(" + log4jFileName
                   + ") returned " + adaptorID 
                   + ", current offset:" + currentLength);
             }
 
           }
         }
       }
 
       long n = System.currentTimeMillis();
       if (n >= nextCheck) {
         now.setTime(n);
         nextCheck = rc.getNextCheckMillis(now);
         try {
           rollOver();
         } catch (IOException ioe) {
           LogLog.error("rollOver() failed.", ioe);
         }
       }
 
       boolean written = false;
       if(layout.ignoresThrowable()) {
         String[] s = event.getThrowableStrRep();
         if (s != null) {
           int len = s.length;
           StringBuilder sb = new StringBuilder();
           sb.append(this.layout.format(event));
           for(int i = 0; i < len; i++) {
             sb.append(s[i]).append("\n");
           }
           //escape the newlines from record bodies, exception and then write this record to the log file
           written = true;
           this.qw.write(RecordConstants.escapeAllButLastRecordSeparator("\n",sb.toString()));
         } 
       }
        
       if (!written) {
         //escape the newlines from record bodies and then write this record to the log file
         this.qw.write(RecordConstants.escapeAllButLastRecordSeparator("\n",this.layout.format(event)));
       }
       
       if (this.immediateFlush) {
         this.qw.flush();
       }
     } catch (Throwable e) {
       System.err.println("Exception in ChukwaRollingAppender: "
           + e.getMessage());
       e.printStackTrace();
     }
 
   }
 
   public String getChukwaClientHostname() {
     return chukwaClientHostname;
   }
 
   public void setChukwaClientHostname(String chukwaClientHostname) {
     this.chukwaClientHostname = chukwaClientHostname;
   }
 
   public int getChukwaClientPortNum() {
     return chukwaClientPortNum;
   }
 
   public void setChukwaClientPortNum(int chukwaClientPortNum) {
     this.chukwaClientPortNum = chukwaClientPortNum;
   }
 
   public void setChukwaClientConnectNumRetry(int i) {
     this.chukwaClientConnectNumRetry = i;
   }
 
   public void setChukwaClientConnectRetryInterval(long i) {
     this.chukwaClientConnectRetryInterval = i;
   }
 
 }
 
 
 /**
  * RollingCalendar is a helper class to DailyRollingFileAppender. Given a
  * periodicity type and the current time, it computes the start of the next
  * interval.
  * */
 class RollingCalendar extends GregorianCalendar {
 
   /**
 	 * 
 	 */
   private static final long serialVersionUID = 2153481574198792767L;
   int type = ChukwaDailyRollingFileAppender.TOP_OF_TROUBLE;
 
   RollingCalendar() {
     super();
   }
 
   RollingCalendar(TimeZone tz, Locale locale) {
     super(tz, locale);
   }
 
   void setType(int type) {
     this.type = type;
   }
 
   public long getNextCheckMillis(Date now) {
     return getNextCheckDate(now).getTime();
   }
 
   public Date getNextCheckDate(Date now) {
     this.setTime(now);
 
     switch (type) {
     case ChukwaDailyRollingFileAppender.TOP_OF_MINUTE:
       this.set(Calendar.SECOND, 0);
       this.set(Calendar.MILLISECOND, 0);
       this.add(Calendar.MINUTE, 1);
       break;
     case ChukwaDailyRollingFileAppender.TOP_OF_HOUR:
       this.set(Calendar.MINUTE, 0);
       this.set(Calendar.SECOND, 0);
       this.set(Calendar.MILLISECOND, 0);
       this.add(Calendar.HOUR_OF_DAY, 1);
       break;
     case ChukwaDailyRollingFileAppender.HALF_DAY:
       this.set(Calendar.MINUTE, 0);
       this.set(Calendar.SECOND, 0);
       this.set(Calendar.MILLISECOND, 0);
       int hour = get(Calendar.HOUR_OF_DAY);
       if (hour < 12) {
         this.set(Calendar.HOUR_OF_DAY, 12);
       } else {
         this.set(Calendar.HOUR_OF_DAY, 0);
         this.add(Calendar.DAY_OF_MONTH, 1);
       }
       break;
     case ChukwaDailyRollingFileAppender.TOP_OF_DAY:
       this.set(Calendar.HOUR_OF_DAY, 0);
       this.set(Calendar.MINUTE, 0);
       this.set(Calendar.SECOND, 0);
       this.set(Calendar.MILLISECOND, 0);
       this.add(Calendar.DATE, 1);
       break;
     case ChukwaDailyRollingFileAppender.TOP_OF_WEEK:
       this.set(Calendar.DAY_OF_WEEK, getFirstDayOfWeek());
       this.set(Calendar.HOUR_OF_DAY, 0);
       this.set(Calendar.SECOND, 0);
       this.set(Calendar.MILLISECOND, 0);
       this.add(Calendar.WEEK_OF_YEAR, 1);
       break;
     case ChukwaDailyRollingFileAppender.TOP_OF_MONTH:
       this.set(Calendar.DATE, 1);
       this.set(Calendar.HOUR_OF_DAY, 0);
       this.set(Calendar.SECOND, 0);
       this.set(Calendar.MILLISECOND, 0);
       this.add(Calendar.MONTH, 1);
       break;
     default:
       throw new IllegalStateException("Unknown periodicity type.");
     }
     return getTime();
   }
 }