/**
    *
    * 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.hbase.regionserver.wal;
  
  import java.io.IOException;
  import java.util.NavigableSet;
  import java.util.TreeSet;
  import java.util.regex.Matcher;
  import java.util.regex.Pattern;
  
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  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.hadoop.fs.PathFilter;
  import org.apache.hadoop.hbase.HConstants;
  import org.apache.hadoop.hbase.HRegionInfo;
  import org.apache.hadoop.hbase.HTableDescriptor;
  import org.apache.hadoop.hbase.ServerName;
  import org.apache.hadoop.hbase.protobuf.generated.WALProtos.CompactionDescriptor;
  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
  import org.apache.hadoop.hbase.util.FSUtils;
  import org.apache.hadoop.hbase.util.Bytes;
  
  public class HLogUtil {
    static final Log LOG = LogFactory.getLog(HLogUtil.class);
  
    /**
     * Pattern used to validate a HLog file name
     */
    private static final Pattern pattern =
        Pattern.compile(".*\\.\\d*("+HLog.META_HLOG_FILE_EXTN+")*");
  
    /**
     * @param filename
     *          name of the file to validate
     * @return <tt>true</tt> if the filename matches an HLog, <tt>false</tt>
     *         otherwise
     */
    public static boolean validateHLogFilename(String filename) {
      return pattern.matcher(filename).matches();
    }
  
    /**
     * Construct the HLog directory name
     *
     * @param serverName
     *          Server name formatted as described in {@link ServerName}
     * @return the relative HLog directory name, e.g.
     *         <code>.logs/1.example.org,60030,12345</code> if
     *         <code>serverName</code> passed is
     *         <code>1.example.org,60030,12345</code>
     */
    public static String getHLogDirectoryName(final String serverName) {
      StringBuilder dirName = new StringBuilder(HConstants.HREGION_LOGDIR_NAME);
      dirName.append("/");
      dirName.append(serverName);
      return dirName.toString();
    }
  
    /**
     * @param regiondir
     *          This regions directory in the filesystem.
     * @return The directory that holds recovered edits files for the region
     *         <code>regiondir</code>
     */
    public static Path getRegionDirRecoveredEditsDir(final Path regiondir) {
      return new Path(regiondir, HConstants.RECOVERED_EDITS_DIR);
    }
  
    /**
     * Move aside a bad edits file.
     *
     * @param fs
     * @param edits
     *          Edits file to move aside.
     * @return The name of the moved aside file.
     * @throws IOException
     */
    public static Path moveAsideBadEditsFile(final FileSystem fs, final Path edits)
       throws IOException {
     Path moveAsideName = new Path(edits.getParent(), edits.getName() + "."
         + System.currentTimeMillis());
     if (!fs.rename(edits, moveAsideName)) {
       LOG.warn("Rename failed from " + edits + " to " + moveAsideName);
     }
     return moveAsideName;
   }
 
   /**
    * @param path
    *          - the path to analyze. Expected format, if it's in hlog directory:
    *          / [base directory for hbase] / hbase / .logs / ServerName /
    *          logfile
    * @return null if it's not a log file. Returns the ServerName of the region
    *         server that created this log file otherwise.
    */
   public static ServerName getServerNameFromHLogDirectoryName(
       Configuration conf, String path) throws IOException {
     if (path == null
         || path.length() <= HConstants.HREGION_LOGDIR_NAME.length()) {
       return null;
     }
 
     if (conf == null) {
       throw new IllegalArgumentException("parameter conf must be set");
     }
 
     final String rootDir = conf.get(HConstants.HBASE_DIR);
     if (rootDir == null || rootDir.isEmpty()) {
       throw new IllegalArgumentException(HConstants.HBASE_DIR
           + " key not found in conf.");
     }
 
     final StringBuilder startPathSB = new StringBuilder(rootDir);
     if (!rootDir.endsWith("/"))
       startPathSB.append('/');
     startPathSB.append(HConstants.HREGION_LOGDIR_NAME);
     if (!HConstants.HREGION_LOGDIR_NAME.endsWith("/"))
       startPathSB.append('/');
     final String startPath = startPathSB.toString();
 
     String fullPath;
     try {
       fullPath = FileSystem.get(conf).makeQualified(new Path(path)).toString();
     } catch (IllegalArgumentException e) {
       LOG.info("Call to makeQualified failed on " + path + " " + e.getMessage());
       return null;
     }
 
     if (!fullPath.startsWith(startPath)) {
       return null;
     }
 
     final String serverNameAndFile = fullPath.substring(startPath.length());
 
     if (serverNameAndFile.indexOf('/') < "a,0,0".length()) {
       // Either it's a file (not a directory) or it's not a ServerName format
       return null;
     }
 
     final String serverName = serverNameAndFile.substring(0,
         serverNameAndFile.indexOf('/') - 1);
 
     if (!ServerName.isFullServerName(serverName)) {
       return null;
     }
 
     return ServerName.parseServerName(serverName);
   }
 
   /**
    * This function returns region server name from a log file name which is in either format:
    * hdfs://<name node>/hbase/.logs/<server name>-splitting/... or hdfs://<name
    * node>/hbase/.logs/<server name>/...
    * @param logFile
    * @return null if the passed in logFile isn't a valid HLog file path
    */
   public static ServerName getServerNameFromHLogDirectoryName(Path logFile) {
     Path logDir = logFile.getParent();
     String logDirName = logDir.getName();
     if (logDirName.equals(HConstants.HREGION_LOGDIR_NAME)) {
       logDir = logFile;
       logDirName = logDir.getName();
     }
     ServerName serverName = null;
     if (logDirName.endsWith(HLog.SPLITTING_EXT)) {
       logDirName = logDirName.substring(0, logDirName.length() - HLog.SPLITTING_EXT.length());
     }
     try {
       serverName = ServerName.parseServerName(logDirName);
     } catch (IllegalArgumentException ex) {
       serverName = null;
       LOG.warn("Invalid log file path=" + logFile, ex);
     }
     if (serverName != null && serverName.getStartcode() < 0) {
       LOG.warn("Invalid log file path=" + logFile);
       return null;
     }
     return serverName;
   }
 
   /**
    * Returns sorted set of edit files made by wal-log splitter, excluding files
    * with '.temp' suffix.
    *
    * @param fs
    * @param regiondir
    * @return Files in passed <code>regiondir</code> as a sorted set.
    * @throws IOException
    */
   public static NavigableSet<Path> getSplitEditFilesSorted(final FileSystem fs,
       final Path regiondir) throws IOException {
     NavigableSet<Path> filesSorted = new TreeSet<Path>();
     Path editsdir = HLogUtil.getRegionDirRecoveredEditsDir(regiondir);
     if (!fs.exists(editsdir))
       return filesSorted;
     FileStatus[] files = FSUtils.listStatus(fs, editsdir, new PathFilter() {
       @Override
       public boolean accept(Path p) {
         boolean result = false;
         try {
           // Return files and only files that match the editfile names pattern.
           // There can be other files in this directory other than edit files.
           // In particular, on error, we'll move aside the bad edit file giving
           // it a timestamp suffix. See moveAsideBadEditsFile.
           Matcher m = HLog.EDITFILES_NAME_PATTERN.matcher(p.getName());
           result = fs.isFile(p) && m.matches();
           // Skip the file whose name ends with RECOVERED_LOG_TMPFILE_SUFFIX,
           // because it means splithlog thread is writting this file.
           if (p.getName().endsWith(HLog.RECOVERED_LOG_TMPFILE_SUFFIX)) {
             result = false;
           }
         } catch (IOException e) {
           LOG.warn("Failed isFile check on " + p);
         }
         return result;
       }
     });
     if (files == null)
       return filesSorted;
     for (FileStatus status : files) {
       filesSorted.add(status.getPath());
     }
     return filesSorted;
   }
 
   public static boolean isMetaFile(Path p) {
     if (p.getName().endsWith(HLog.META_HLOG_FILE_EXTN)) {
       return true;
     }
     return false;
   }
 
   /**
    * Write the marker that a compaction has succeeded and is about to be committed.
    * This provides info to the HMaster to allow it to recover the compaction if
    * this regionserver dies in the middle (This part is not yet implemented). It also prevents
    * the compaction from finishing if this regionserver has already lost its lease on the log.
    */
   public static void writeCompactionMarker(HLog log, HTableDescriptor htd, HRegionInfo info,
       final CompactionDescriptor c) throws IOException {
     WALEdit e = WALEdit.createCompaction(c);
     log.append(info, c.getTableName().toByteArray(), e,
         EnvironmentEdgeManager.currentTimeMillis(), htd);
     LOG.info("Appended compaction marker " + c);
   }
 }