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

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URL;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
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.hadoop.io.IOUtils;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
import org.apache.hbase.thirdparty.com.google.common.collect.MapMaker;

/**
 * ClassLoader used to load classes for Coprocessor instances.
 * <p>
 * This ClassLoader always tries to load classes from the specified coprocessor
 * jar first actually using URLClassLoader logic before delegating to the parent
 * ClassLoader, thus avoiding dependency conflicts between HBase's classpath and
 * classes in the coprocessor jar.
 * <p>
 * Certain classes are exempt from being loaded by this ClassLoader because it
 * would prevent them from being cast to the equivalent classes in the region
 * server.  For example, the Coprocessor interface needs to be loaded by the
 * region server's ClassLoader to prevent a ClassCastException when casting the
 * coprocessor implementation.
 * <p>
 * A HDFS path can be used to specify the coprocessor jar. In this case, the jar
 * will be copied to local at first under some folder under ${hbase.local.dir}/jars/tmp/.
 * The local copy will be removed automatically when the HBase server instance is
 * stopped.
 * <p>
 * This ClassLoader also handles resource loading.  In most cases this
 * ClassLoader will attempt to load resources from the coprocessor jar first
 * before delegating to the parent.  However, like in class loading,
 * some resources need to be handled differently.  For all of the Hadoop
 * default configurations (e.g. hbase-default.xml) we will check the parent
 * ClassLoader first to prevent issues such as failing the HBase default
 * configuration version check.
 */
@InterfaceAudience.Private
public class CoprocessorClassLoader extends ClassLoaderBase {
  private static final Logger LOG = LoggerFactory.getLogger(CoprocessorClassLoader.class);

  // A temporary place ${hbase.local.dir}/jars/tmp/ to store the local
  // copy of the jar file and the libraries contained in the jar.
  private static final String TMP_JARS_DIR = File.separator
     + "jars" + File.separator + "tmp" + File.separator;

  /**
   * External class loaders cache keyed by external jar path.
   * ClassLoader instance is stored as a weak-reference
   * to allow GC'ing when it is not used
   * (@see HBASE-7205)
   */
  private static final ConcurrentMap<Path, CoprocessorClassLoader> classLoadersCache =
    new MapMaker().concurrencyLevel(3).weakValues().makeMap();

  /**
   * If the class being loaded starts with any of these strings, we will skip
   * trying to load it from the coprocessor jar and instead delegate
   * directly to the parent ClassLoader.
   */
  private static final String[] CLASS_PREFIX_EXEMPTIONS = new String[] {
    // Java standard library:
    "com.sun.",
    "java.",
    "javax.",
    "org.ietf",
    "org.omg",
    "org.w3c",
    "org.xml",
    "sunw.",
    // logging
    "org.slf4j",
    "org.apache.log4j",
    "com.hadoop",
    // HBase:
    "org.apache.hadoop.hbase",
  };

  /**
   * If the resource being loaded matches any of these patterns, we will first
   * attempt to load the resource with the parent ClassLoader.  Only if the
   * resource is not found by the parent do we attempt to load it from the coprocessor jar.
   */
  private static final Pattern[] RESOURCE_LOAD_PARENT_FIRST_PATTERNS =
      new Pattern[] {
    Pattern.compile("^[^-]+-default\\.xml$")
  };

  private static final Pattern libJarPattern = Pattern.compile("[/]?lib/([^/]+\\.jar)");

  /**
   * A locker used to synchronize class loader initialization per coprocessor jar file
   */
  private static final KeyLocker<String> locker = new KeyLocker<>();

  /**
   * A set used to synchronized parent path clean up.  Generally, there
   * should be only one parent path, but using a set so that we can support more.
   */
  static final HashSet<String> parentDirLockSet = new HashSet<>();

  /**
   * Creates a JarClassLoader that loads classes from the given paths.
   */
  private CoprocessorClassLoader(ClassLoader parent) {
    super(parent);
  }

  private void init(Path pathPattern, String pathPrefix,
      Configuration conf) throws IOException {
    // Copy the jar to the local filesystem
    String parentDirStr =
      conf.get(LOCAL_DIR_KEY, DEFAULT_LOCAL_DIR) + TMP_JARS_DIR;
    synchronized (parentDirLockSet) {
      if (!parentDirLockSet.contains(parentDirStr)) {
        Path parentDir = new Path(parentDirStr);
        FileSystem fs = FileSystem.getLocal(conf);
        fs.delete(parentDir, true); // it's ok if the dir doesn't exist now
        parentDirLockSet.add(parentDirStr);
        if (!fs.mkdirs(parentDir) && !fs.getFileStatus(parentDir).isDirectory()) {
          throw new RuntimeException("Failed to create local dir " + parentDirStr
            + ", CoprocessorClassLoader failed to init");
        }
      }
    }

    FileSystem fs = pathPattern.getFileSystem(conf);
    Path pathPattern1 = fs.isDirectory(pathPattern) ?
      new Path(pathPattern, "*.jar") : pathPattern;  // append "*.jar" if a directory is specified
    FileStatus[] fileStatuses = fs.globStatus(pathPattern1);  // return all files that match the pattern
    if (fileStatuses == null || fileStatuses.length == 0) {  // if no one matches
      throw new FileNotFoundException(pathPattern1.toString());
    } else {
      boolean validFileEncountered = false;
      for (Path path : FileUtil.stat2Paths(fileStatuses)) {  // for each file that match the pattern
        if (fs.isFile(path)) {  // only process files, skip for directories
          File dst = new File(parentDirStr, "." + pathPrefix + "."
            + path.getName() + "." + System.currentTimeMillis() + ".jar");
          fs.copyToLocalFile(path, new Path(dst.toString()));
          dst.deleteOnExit();

          addURL(dst.getCanonicalFile().toURI().toURL());

          JarFile jarFile = new JarFile(dst.toString());
          try {
            Enumeration<JarEntry> entries = jarFile.entries();  // get entries inside a jar file
            while (entries.hasMoreElements()) {
              JarEntry entry = entries.nextElement();
              Matcher m = libJarPattern.matcher(entry.getName());
              if (m.matches()) {
                File file = new File(parentDirStr, "." + pathPrefix + "."
                  + path.getName() + "." + System.currentTimeMillis() + "." + m.group(1));
                try (FileOutputStream outStream = new FileOutputStream(file)) {
                  IOUtils.copyBytes(jarFile.getInputStream(entry),
                    outStream, conf, true);
                }
                file.deleteOnExit();
                addURL(file.toURI().toURL());
              }
            }
          } finally {
            jarFile.close();
          }

          validFileEncountered = true;  // Set to true when encountering a file
        }
      }
      if (validFileEncountered == false) {  // all items returned by globStatus() are directories
        throw new FileNotFoundException("No file found matching " + pathPattern1.toString());
      }
    }
  }

  // This method is used in unit test
  public static CoprocessorClassLoader getIfCached(final Path path) {
    Preconditions.checkNotNull(path, "The jar path is null!");
    return classLoadersCache.get(path);
  }

  // This method is used in unit test
  public static Collection<? extends ClassLoader> getAllCached() {
    return classLoadersCache.values();
  }

  // This method is used in unit test
  public static void clearCache() {
    classLoadersCache.clear();
  }

  /**
   * Get a CoprocessorClassLoader for a coprocessor jar path from cache.
   * If not in cache, create one.
   *
   * @param path the path to the coprocessor jar file to load classes from
   * @param parent the parent class loader for exempted classes
   * @param pathPrefix a prefix used in temp path name to store the jar file locally
   * @param conf the configuration used to create the class loader, if needed
   * @return a CoprocessorClassLoader for the coprocessor jar path
   * @throws IOException
   */
  public static CoprocessorClassLoader getClassLoader(final Path path,
      final ClassLoader parent, final String pathPrefix,
      final Configuration conf) throws IOException {
    CoprocessorClassLoader cl = getIfCached(path);
    String pathStr = path.toString();
    if (cl != null) {
      LOG.debug("Found classloader "+ cl + " for "+ pathStr);
      return cl;
    }

    if (path.getFileSystem(conf).isFile(path) && !pathStr.endsWith(".jar")) {
      throw new IOException(pathStr + ": not a jar file?");
    }

    Lock lock = locker.acquireLock(pathStr);
    try {
      cl = getIfCached(path);
      if (cl != null) {
        LOG.debug("Found classloader "+ cl + " for "+ pathStr);
        return cl;
      }

      cl = AccessController.doPrivileged(
          new PrivilegedAction<CoprocessorClassLoader>() {
        @Override
        public CoprocessorClassLoader run() {
          return new CoprocessorClassLoader(parent);
        }
      });

      cl.init(path, pathPrefix, conf);

      // Cache class loader as a weak value, will be GC'ed when no reference left
      CoprocessorClassLoader prev = classLoadersCache.putIfAbsent(path, cl);
      if (prev != null) {
        // Lost update race, use already added class loader
        LOG.warn("THIS SHOULD NOT HAPPEN, a class loader"
          +" is already cached for " + pathStr);
        cl = prev;
      }
      return cl;
    } finally {
      lock.unlock();
    }
  }

  @Override
  public Class<?> loadClass(String name)
      throws ClassNotFoundException {
    return loadClass(name, null);
  }

  public Class<?> loadClass(String name, String[] includedClassPrefixes)
      throws ClassNotFoundException {
    // Delegate to the parent immediately if this class is exempt
    if (isClassExempt(name, includedClassPrefixes)) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Skipping exempt class " + name +
            " - delegating directly to parent");
      }
      return parent.loadClass(name);
    }

    synchronized (getClassLoadingLock(name)) {
      // Check whether the class has already been loaded:
      Class<?> clasz = findLoadedClass(name);
      if (clasz != null) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Class " + name + " already loaded");
        }
      }
      else {
        try {
          // Try to find this class using the URLs passed to this ClassLoader
          if (LOG.isDebugEnabled()) {
            LOG.debug("Finding class: " + name);
          }
          clasz = findClass(name);
        } catch (ClassNotFoundException e) {
          // Class not found using this ClassLoader, so delegate to parent
          if (LOG.isDebugEnabled()) {
            LOG.debug("Class " + name + " not found - delegating to parent");
          }
          try {
            clasz = parent.loadClass(name);
          } catch (ClassNotFoundException e2) {
            // Class not found in this ClassLoader or in the parent ClassLoader
            // Log some debug output before re-throwing ClassNotFoundException
            if (LOG.isDebugEnabled()) {
              LOG.debug("Class " + name + " not found in parent loader");
            }
            throw e2;
          }
        }
      }
      return clasz;
    }
  }

  @Override
  public URL getResource(String name) {
    URL resource = null;
    boolean parentLoaded = false;

    // Delegate to the parent first if necessary
    if (loadResourceUsingParentFirst(name)) {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Checking parent first for resource " + name);
      }
      resource = super.getResource(name);
      parentLoaded = true;
    }

    if (resource == null) {
      synchronized (getClassLoadingLock(name)) {
        // Try to find the resource in this jar
        resource = findResource(name);
        if ((resource == null) && !parentLoaded) {
          // Not found in this jar and we haven't attempted to load
          // the resource in the parent yet; fall back to the parent
          resource = super.getResource(name);
        }
      }
    }
    return resource;
  }

  /**
   * Determines whether the given class should be exempt from being loaded
   * by this ClassLoader.
   * @param name the name of the class to test.
   * @return true if the class should *not* be loaded by this ClassLoader;
   * false otherwise.
   */
  protected boolean isClassExempt(String name, String[] includedClassPrefixes) {
    if (includedClassPrefixes != null) {
      for (String clsName : includedClassPrefixes) {
        if (name.startsWith(clsName)) {
          return false;
        }
      }
    }
    for (String exemptPrefix : CLASS_PREFIX_EXEMPTIONS) {
      if (name.startsWith(exemptPrefix)) {
        return true;
      }
    }
    return false;
  }

  /**
   * Determines whether we should attempt to load the given resource using the
   * parent first before attempting to load the resource using this ClassLoader.
   * @param name the name of the resource to test.
   * @return true if we should attempt to load the resource using the parent
   * first; false if we should attempt to load the resource using this
   * ClassLoader first.
   */
  protected boolean loadResourceUsingParentFirst(String name) {
    for (Pattern resourcePattern : RESOURCE_LOAD_PARENT_FIRST_PATTERNS) {
      if (resourcePattern.matcher(name).matches()) {
        return true;
      }
    }
    return false;
  }
}