/*

  * 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.giraph.utils;

 import static org.apache.giraph.conf.GiraphConstants.COMPUTATION_CLASS;

 import static org.apache.giraph.conf.GiraphConstants.TYPES_HOLDER_CLASS;

 import java.io.IOException;

 import java.util.List;

 import org.apache.commons.cli.BasicParser;

 import org.apache.commons.cli.CommandLine;

 import org.apache.commons.cli.CommandLineParser;

 import org.apache.commons.cli.HelpFormatter;

 import org.apache.commons.cli.Options;

 import org.apache.commons.cli.ParseException;

 import org.apache.giraph.Algorithm;

 import org.apache.giraph.aggregators.AggregatorWriter;

 import org.apache.giraph.combiner.MessageCombiner;

 import org.apache.giraph.conf.GiraphConfiguration;

 import org.apache.giraph.conf.GiraphConfigurationSettable;

 import org.apache.giraph.conf.GiraphConstants;

 import org.apache.giraph.conf.GiraphTypes;

 import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;

 import org.apache.giraph.conf.TypesHolder;

 import org.apache.giraph.edge.OutEdges;

 import org.apache.giraph.factories.VertexValueFactory;

 import org.apache.giraph.graph.Computation;

 import org.apache.giraph.graph.Language;

 import org.apache.giraph.graph.VertexValueCombiner;

 import org.apache.giraph.io.EdgeInputFormat;

 import org.apache.giraph.io.EdgeOutputFormat;

 import org.apache.giraph.io.VertexInputFormat;

 import org.apache.giraph.io.VertexOutputFormat;

 import org.apache.giraph.io.formats.GiraphFileInputFormat;

 import org.apache.giraph.job.GiraphConfigurationValidator;

 import org.apache.giraph.jython.JythonUtils;

 import org.apache.giraph.master.MasterCompute;

 import org.apache.giraph.partition.Partition;

 import org.apache.giraph.scripting.DeployType;

 import org.apache.giraph.scripting.ScriptLoader;

 import org.apache.giraph.worker.WorkerContext;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.hadoop.fs.FileSystem;

 import org.apache.hadoop.fs.Path;

 import org.apache.log4j.Level;

 import org.apache.log4j.Logger;

 import org.apache.zookeeper.ZooKeeper;

 import com.google.common.base.Splitter;

 import com.google.common.collect.Iterables;

 /**

  * Translate command line args into Configuration Key-Value pairs.

  */

 public final class ConfigurationUtils {

   /** Class logger */

   private static final Logger LOG =

     Logger.getLogger(ConfigurationUtils.class);

 /*if[PURE_YARN]

   // The base path for output dirs as saved in GiraphConfiguration

   private static final Path BASE_OUTPUT_PATH;

   static {

     // whether local or remote, if there's no *-site.xml's to find, we're done

     try {

       BASE_OUTPUT_PATH = FileSystem.get(new Configuration()).getHomeDirectory();

     } catch (IOException ioe) {

       throw new IllegalStateException("Error locating default base path!", ioe);

     }

   }

 end[PURE_YARN]*/

   /** Maintains our accepted options in case the caller wants to add some */

   private static Options OPTIONS;

   static {

     OPTIONS = new Options();

     OPTIONS.addOption("h", "help", false, "Help");

     OPTIONS.addOption("la", "listAlgorithms", false, "List supported " +

         "algorithms");

     OPTIONS.addOption("q", "quiet", false, "Quiet output");

     OPTIONS.addOption("yj", "yarnjars", true, "comma-separated list of JAR " +

       "filenames to distribute to Giraph tasks and ApplicationMaster. " +

       "YARN only. Search order: CLASSPATH, HADOOP_HOME, user current dir.");

     OPTIONS.addOption("yh", "yarnheap", true, "Heap size, in MB, for each " +

       "Giraph task (YARN only.) Defaults to " +

       GiraphConstants.GIRAPH_YARN_TASK_HEAP_MB + " MB.");

     OPTIONS.addOption("w", "workers", true, "Number of workers");

     OPTIONS.addOption("vif", "vertexInputFormat", true, "Vertex input format");

     OPTIONS.addOption("eif", "edgeInputFormat", true, "Edge input format");

     OPTIONS.addOption("vof", "vertexOutputFormat", true,

         "Vertex output format");

     OPTIONS.addOption("eof", "edgeOutputFormat", true, "Edge output format");

     OPTIONS.addOption("vip", "vertexInputPath", true, "Vertex input path");

     OPTIONS.addOption("eip", "edgeInputPath", true, "Edge input path");

     OPTIONS.addOption("op",  "outputPath", true, "Output path");

     OPTIONS.addOption("vsd",  "vertexSubDir", true, "subdirectory to be used " +

         "for the vertex output");

     OPTIONS.addOption("esd",  "edgeSubDir", true, "subdirectory to be used " +

         "for the edge output");

     OPTIONS.addOption("c", "combiner", true, "MessageCombiner class");

     OPTIONS.addOption("ve", "outEdges", true, "Vertex edges class");

     OPTIONS.addOption("wc", "workerContext", true, "WorkerContext class");

     OPTIONS.addOption("aw", "aggregatorWriter", true, "AggregatorWriter class");

     OPTIONS.addOption("mc", "masterCompute", true, "MasterCompute class");

     OPTIONS.addOption("cf", "cacheFile", true, "Files for distributed cache");

     OPTIONS.addOption("pc", "partitionClass", true, "Partition class");

     OPTIONS.addOption("vvf", "vertexValueFactoryClass", true,

         "Vertex value factory class");

     OPTIONS.addOption("th", "typesHolder", true,

         "Class that holds types. Needed only if Computation is not set");

     OPTIONS.addOption("jyc", "jythonClass", true,

         "Jython class name, used if computation passed in is a python script");

     OPTIONS.addOption("ca", "customArguments", true, "provide custom" +

         " arguments for the job configuration in the form:" +

         " -ca <param1>=<value1>,<param2>=<value2> -ca <param3>=<value3> etc." +

         " It can appear multiple times, and the last one has effect" +

         " for the same param.");

   }

   /**

    * No constructing this utility class

    */

   private ConfigurationUtils() { }

   /**

    * Configure an object with an

    * {@link org.apache.giraph.conf.ImmutableClassesGiraphConfiguration}

    * if that objects supports it.

    *

    * @param object The object to configure

    * @param configuration The configuration

    */

   public static void configureIfPossible(Object object,

       ImmutableClassesGiraphConfiguration configuration) {

     if (configuration != null) {

       configuration.configureIfPossible(object);

     } else if (object instanceof GiraphConfigurationSettable) {

       throw new IllegalArgumentException(

           "Trying to configure configurable object without value, " +

           object.getClass());

     }

   }

   /**

    * Get a class which is parameterized by the graph types defined by user.

    * The types holder is actually an interface that any class which holds all of

    * Giraph types can implement. It is used with reflection to infer the Giraph

    * types.

    *

    * The current order of type holders we try are:

    * 1) The {@link TypesHolder} class directly.

    * 2) The {@link Computation} class, as that holds all the types.

    *

    * @param conf Configuration

    * @return {@link TypesHolder} or null if could not find one.

    */

   public static Class<? extends TypesHolder> getTypesHolderClass(

       Configuration conf) {

     Class<? extends TypesHolder> klass = TYPES_HOLDER_CLASS.get(conf);

     if (klass != null) {

       return klass;

     }

     klass = COMPUTATION_CLASS.get(conf);

     return klass;

   }

   /**

    * Translate CLI arguments to GiraphRunner or 'bin/hadoop jar' into

    * Configuration Key-Value pairs.

    * @param giraphConf the current job Configuration.

    * @param args the raw CLI args to parse

    * @return a CommandLine object, or null if the job run should exit.

    */

   public static CommandLine parseArgs(final GiraphConfiguration giraphConf,

     final String[] args) throws ClassNotFoundException, ParseException,

     IOException {

     // verify we have args at all (can't run without them!)

     if (args.length == 0) {

       throw new IllegalArgumentException("No arguments were provided (try -h)");

     }

     CommandLineParser parser = new BasicParser();

     CommandLine cmd = parser.parse(OPTIONS, args);

     // simply printing help or info, return normally but kill job run

     if (cmd.hasOption("h")) {

       printHelp();

       return null;

     }

     if (cmd.hasOption("la")) {

       printSupportedAlgorithms();

       return null;

     }

     // Be certain that there are no critical args missing, die if so.

     performSanityCheck(cmd);

     // Args are OK; attempt to populate the GiraphConfiguration with them.

     final String computationClassName = args[0];

     final int workers = Integer.parseInt(cmd.getOptionValue('w'));

     populateGiraphConfiguration(giraphConf, cmd, computationClassName, workers);

     // validate generic parameters chosen are correct or

     // throw IllegalArgumentException, halting execution.

     @SuppressWarnings("rawtypes")

     GiraphConfigurationValidator<?, ?, ?, ?, ?> gtv =

       new GiraphConfigurationValidator(giraphConf);

     gtv.validateConfiguration();

     // successfully populated/validated GiraphConfiguration, ready to run job

     return cmd;

   }

   /**

    * Callers can place additional options to be parsed and stored in our job's

    * GiraphConfiguration via this utility call. These options will ONLY be

    * parsed and placed into the CommandLine returned from <code>parseArgs</code>

    * Calling code must query this CommandLine to take action on these options.

    * @param opt short options name, i.e. -h

    * @param longOpt long option name, i.e. --help

    * @param hasArg should we expect an argument for this option?

    * @param description English description of this option.

    */

   public static void addOption(final String opt, final String longOpt,

     final boolean hasArg, final String description) {

     if (OPTIONS.hasOption(opt)) {

       printHelp();

       throw new IllegalArgumentException("GiraphConfiguration already " +

         "provides a '" + opt + "' option, please choose another identifier.");

     }

     OPTIONS.addOption(opt, longOpt, hasArg, description);

   }

   /**

    * Utility to check mission-critical args are populated. The validity of

    * the values provided in these args is checked elsewhere.

    * @param cmd our parsed CommandLine

    */

   private static void performSanityCheck(final CommandLine cmd) {

     // Verify all the required options have been provided

     if (!cmd.hasOption("w")) {

       throw new IllegalArgumentException("Need to choose the " +

         "number of workers (-w)");

     }

     if (!cmd.hasOption("vif") && !cmd.hasOption("eif")) {

       throw new IllegalArgumentException("Need to set an input " +

         "format (-vif or -eif)");

     }

   }

   /**

    * Populate GiraphConfiguration for this job with all cmd line args found.

    * Any global configuration data that Giraph on any platform might need

    * should be captured here.

    *

    * @param conf config for this job run

    * @param cmd parsed command line options to store in conf

    * @param computationClassName the computation class (application) to run in

    *                             this job.

    * @param workers the number of worker tasks for this job run.

    */

   private static void populateGiraphConfiguration(final GiraphConfiguration

     conf, final CommandLine cmd,

     final String computationClassName, final int workers)

     throws ClassNotFoundException, IOException {

     conf.setWorkerConfiguration(workers, workers, 100.0f);

     if (cmd.hasOption("typesHolder")) {

       Class<? extends TypesHolder> typesHolderClass =

           (Class<? extends TypesHolder>)

               Class.forName(cmd.getOptionValue("typesHolder"));

       TYPES_HOLDER_CLASS.set(conf, typesHolderClass);

     }

     if (cmd.hasOption("c")) {

       conf.setMessageCombinerClass(

           (Class<? extends MessageCombiner>)

               Class.forName(cmd.getOptionValue("c")));

     }

     if (cmd.hasOption("vc")) {

       conf.setVertexValueCombinerClass(

           (Class<? extends VertexValueCombiner>)

               Class.forName(cmd.getOptionValue("vc")));

     }

     if (cmd.hasOption("ve")) {

       conf.setOutEdgesClass(

           (Class<? extends OutEdges>) Class.forName(cmd.getOptionValue("ve")));

     }

     if (cmd.hasOption("ive")) {

       conf.setInputOutEdgesClass(

           (Class<? extends OutEdges>) Class.forName(cmd.getOptionValue("ive")));

     }

     if (cmd.hasOption("wc")) {

       conf.setWorkerContextClass(

           (Class<? extends WorkerContext>) Class

               .forName(cmd.getOptionValue("wc")));

     }

     if (cmd.hasOption("mc")) {

       conf.setMasterComputeClass(

           (Class<? extends MasterCompute>) Class

               .forName(cmd.getOptionValue("mc")));

     }

     if (cmd.hasOption("aw")) {

       conf.setAggregatorWriterClass(

           (Class<? extends AggregatorWriter>) Class

               .forName(cmd.getOptionValue("aw")));

     }

     if (cmd.hasOption("vif")) {

       conf.setVertexInputFormatClass(

           (Class<? extends VertexInputFormat>) Class

               .forName(cmd.getOptionValue("vif")));

     } else {

       if (LOG.isInfoEnabled()) {

         LOG.info("No vertex input format specified. Ensure your " +

           "InputFormat does not require one.");

       }

     }

     if (cmd.hasOption("eif")) {

       conf.setEdgeInputFormatClass(

           (Class<? extends EdgeInputFormat>) Class

               .forName(cmd.getOptionValue("eif")));

     } else {

       if (LOG.isInfoEnabled()) {

         LOG.info("No edge input format specified. Ensure your " +

           "InputFormat does not require one.");

       }

     }

     if (cmd.hasOption("vof")) {

       conf.setVertexOutputFormatClass(

           (Class<? extends VertexOutputFormat>) Class

               .forName(cmd.getOptionValue("vof")));

     } else {

       if (LOG.isInfoEnabled()) {

         LOG.info("No vertex output format specified. Ensure your " +

           "OutputFormat does not require one.");

       }

     }

     if (cmd.hasOption("vof")) {

       if (cmd.hasOption("vsd")) {

         conf.setVertexOutputFormatSubdir(cmd.getOptionValue("vsd"));

       }

     }

     if (cmd.hasOption("eof")) {

       conf.setEdgeOutputFormatClass(

           (Class<? extends EdgeOutputFormat>) Class

               .forName(cmd.getOptionValue("eof")));

     } else {

       if (LOG.isInfoEnabled()) {

         LOG.info("No edge output format specified. Ensure your " +

           "OutputFormat does not require one.");

       }

     }

     if (cmd.hasOption("eof")) {

       if (cmd.hasOption("esd")) {

         conf.setEdgeOutputFormatSubdir(cmd.getOptionValue("esd"));

       }

     }

     /* check for path clashes */

     if (cmd.hasOption("vof") && cmd.hasOption("eof") && cmd.hasOption("op")) {

       if (!cmd.hasOption("vsd") || cmd.hasOption("esd")) {

         if (!conf.hasEdgeOutputFormatSubdir() ||

             !conf.hasVertexOutputFormatSubdir()) {

           throw new IllegalArgumentException("If VertexOutputFormat and " +

               "EdgeOutputFormat are both set, it is mandatory to provide " +

               "both vertex subdirectory as well as edge subdirectory");

         }

       }

     }

     if (cmd.hasOption("pc")) {

       conf.setPartitionClass(

           (Class<? extends Partition>) Class.forName(cmd.getOptionValue("pc")));

     }

     if (cmd.hasOption("vvf")) {

       conf.setVertexValueFactoryClass(

           (Class<? extends VertexValueFactory>) Class

               .forName(cmd.getOptionValue("vvf")));

     }

     if (cmd.hasOption("ca")) {

       for (String caOptionValue : cmd.getOptionValues("ca")) {

         for (String paramValue :

             Splitter.on(',').split(caOptionValue)) {

           String[] parts = Iterables.toArray(Splitter.on('=').split(paramValue),

                                               String.class);

           if (parts.length != 2) {

             throw new IllegalArgumentException("Unable to parse custom " +

                 " argument: " + paramValue);

           }

           if (LOG.isInfoEnabled()) {

             LOG.info("Setting custom argument [" + parts[0] + "] to [" +

                 parts[1] + "] in GiraphConfiguration");

           }

           conf.set(parts[0], parts[1]);

         }

       }

     }

     // Now, we parse options that are specific to Hadoop MR Job

     if (cmd.hasOption("vif")) {

       if (cmd.hasOption("vip")) {

         if (FileSystem.get(new Configuration()).listStatus(

               new Path(cmd.getOptionValue("vip"))) == null) {

           throw new IllegalArgumentException(

               "Invalid vertex input path (-vip): " +

               cmd.getOptionValue("vip"));

         }

         GiraphFileInputFormat.addVertexInputPath(conf,

           new Path(cmd.getOptionValue("vip")));

       } else {

         if (LOG.isInfoEnabled()) {

           LOG.info("No input path for vertex data was specified. Ensure your " +

             "InputFormat does not require one.");

         }

       }

     }

     if (cmd.hasOption("eif")) {

       if (cmd.hasOption("eip")) {

         if (FileSystem.get(new Configuration()).listStatus(

               new Path(cmd.getOptionValue("eip"))) == null) {

           throw new IllegalArgumentException(

               "Invalid edge input path (-eip): " +

               cmd.getOptionValue("eip"));

         }

         GiraphFileInputFormat.addEdgeInputPath(conf,

           new Path(cmd.getOptionValue("eip")));

       } else {

         if (LOG.isInfoEnabled()) {

           LOG.info("No input path for edge data was specified. Ensure your " +

             "InputFormat does not require one.");

         }

       }

     }

     // YARN-ONLY OPTIONS

     if (cmd.hasOption("yj")) {

       conf.setYarnLibJars(cmd.getOptionValue("yj"));

     }

     if (cmd.hasOption("yh")) {

       conf.setYarnTaskHeapMb(

           Integer.parseInt(cmd.getOptionValue("yh")));

     }

 /*if[PURE_YARN]

     if (cmd.hasOption("vof") || cmd.hasOption("eof")) {

       if (cmd.hasOption("op")) {

         // For YARN conf to get the out dir we need w/o a Job obj

         Path outputDir =

             new Path(BASE_OUTPUT_PATH, cmd.getOptionValue("op"));

         outputDir =

           outputDir.getFileSystem(conf).makeQualified(outputDir);

         conf.set(

             org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.OUTDIR,

             outputDir.toString());

       } else {

         if (LOG.isInfoEnabled()) {

           LOG.info("No output path specified. Ensure your OutputFormat " +

             "does not require one.");

         }

       }

     }

 end[PURE_YARN]*/

     // END YARN-ONLY OPTIONS

     handleComputationClass(conf, cmd, computationClassName);

   }

   /**

    * Helper to deal with computation class.

    *

    * @param conf Configuration

    * @param cmd CommandLine

    * @param computationClassName Name of computation

    * @throws ClassNotFoundException error finding class

    */

   private static void handleComputationClass(GiraphConfiguration conf,

     CommandLine cmd, String computationClassName)

     throws ClassNotFoundException {

     if (computationClassName.endsWith("py")) {

       handleJythonComputation(conf, cmd, computationClassName);

     } else {

       conf.setComputationClass(

           (Class<? extends Computation>) Class.forName(computationClassName));

     }

   }

   /**

    * Helper to handle Computations implemented in Python.

    *

    * @param conf Configuration

    * @param cmd CommandLine

    * @param scriptPath Path to python script

    */

   private static void handleJythonComputation(GiraphConfiguration conf,

     CommandLine cmd, String scriptPath) {

     String jythonClass = cmd.getOptionValue("jythonClass");

     if (jythonClass == null) {

       throw new IllegalArgumentException(

           "handleJythonComputation: Need to set Jython Computation class " +

           "name with --jythonClass");

     }

     String typesHolderClass = cmd.getOptionValue("typesHolder");

     if (typesHolderClass == null) {

       throw new IllegalArgumentException(

           "handleJythonComputation: Need to set TypesHolder class name " +

           "with --typesHolder");

     }

     Path path = new Path(scriptPath);

     Path remotePath = DistributedCacheUtils.copyAndAdd(path, conf);

     ScriptLoader.setScriptsToLoad(conf, remotePath.toString(),

         DeployType.DISTRIBUTED_CACHE, Language.JYTHON);

     GiraphTypes.readFrom(conf).writeIfUnset(conf);

     JythonUtils.init(conf, jythonClass);

   }

   /**

    * Utility to print CLI help messsage for registered options.

    */

   private static void printHelp() {

     HelpFormatter formatter = new HelpFormatter();

     formatter.printHelp(ConfigurationUtils.class.getName(), OPTIONS, true);

   }

   /**

    * Prints description of algorithms annotated with

    * {@link org.apache.giraph.Algorithm}

    */

   private static void printSupportedAlgorithms() {

     Logger.getLogger(ZooKeeper.class).setLevel(Level.OFF);

     List<Class<?>> classes = AnnotationUtils.getAnnotatedClasses(

       Algorithm.class, "org.apache.giraph");

     System.out.print("  Supported algorithms:\n");

     for (Class<?> clazz : classes) {

       if (Computation.class.isAssignableFrom(clazz)) {

         Algorithm algorithm = clazz.getAnnotation(Algorithm.class);

         StringBuilder sb = new StringBuilder();

         sb.append(algorithm.name()).append(" - ").append(clazz.getName())

             .append("\n");

         if (!algorithm.description().equals("")) {

           sb.append("    ").append(algorithm.description()).append("\n");

         }

         System.out.print(sb.toString());

       }

     }

   }

 }