src/contrib/gridmix/src/test/system/org/apache/hadoop/mapred/gridmix/test/system/UtilsForGridmix.java - hadoop-mapreduce - Git at Google

 /**
  * 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.mapred.gridmix.test.system;

 import java.io.IOException;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.ToolRunner;
 import org.apache.hadoop.mapred.gridmix.Gridmix;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobClient;
 import org.apache.hadoop.mapred.JobStatus;
 import org.apache.hadoop.mapreduce.JobID;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Arrays;
 import java.net.URI;
 import java.text.SimpleDateFormat;
 import java.io.OutputStream;
 import java.util.Set;
 import java.util.List;
 import java.util.Iterator;
 import java.util.Map;
 import java.io.File;
 import java.io.FileOutputStream;
 import org.apache.hadoop.test.system.ProxyUserDefinitions;
 import org.apache.hadoop.test.system.ProxyUserDefinitions.GroupsAndHost;

 /**
  * Gridmix utilities.
  */
 public class UtilsForGridmix {
   private static final Log LOG = LogFactory.getLog(UtilsForGridmix.class);
   private static final Path DEFAULT_TRACES_PATH =
     new Path(System.getProperty("user.dir") + "/src/test/system/resources/");

   /**
    * cleanup the folder or file.
    * @param path - folder or file path.
    * @param conf - cluster configuration
    * @throws IOException - If an I/O error occurs.
    */
   public static void cleanup(Path path, Configuration conf)
      throws IOException {
     FileSystem fs = path.getFileSystem(conf);
     fs.delete(path, true);
     fs.close();
   }

   /**
    * Get the login user.
    * @return - login user as string..
    * @throws IOException - if an I/O error occurs.
    */
   public static String getUserName() throws IOException {
     return UserGroupInformation.getLoginUser().getUserName();
   }

   /**
    * Get the argument list for gridmix job.
    * @param gridmixDir - gridmix parent directory.
    * @param gridmixRunMode - gridmix modes either 1,2,3.
    * @param values - gridmix runtime values.
    * @param otherArgs - gridmix other generic args.
    * @return - argument list as string array.
    */
   public static String [] getArgsList(Path gridmixDir, int gridmixRunMode,
                                       String [] values, String [] otherArgs) {
     String [] runtimeArgs = {
         "-D", GridMixConfig.GRIDMIX_LOG_MODE + "=DEBUG",
         "-D", GridMixConfig.GRIDMIX_OUTPUT_DIR + "=gridmix",
         "-D", GridMixConfig.GRIDMIX_JOB_SUBMISSION_QUEUE_IN_TRACE + "=true",
         "-D", GridMixConfig.GRIDMIX_JOB_TYPE + "=" + values[0],
         "-D", GridMixConfig.GRIDMIX_USER_RESOLVER + "=" + values[1],
         "-D", GridMixConfig.GRIDMIX_SUBMISSION_POLICY + "=" + values[2]
     };

     String [] classArgs;
     if ((gridmixRunMode == GridMixRunMode.DATA_GENERATION.getValue()
        || gridmixRunMode
        == GridMixRunMode.DATA_GENERATION_AND_RUN_GRIDMIX.getValue())
        && values[1].indexOf("RoundRobinUserResolver") > 0) {
       classArgs = new String[] {
           "-generate", values[3],
           "-users", values[4],
           gridmixDir.toString(),
           values[5]
       };
     } else if (gridmixRunMode == GridMixRunMode.DATA_GENERATION.getValue()
               || gridmixRunMode
               == GridMixRunMode.DATA_GENERATION_AND_RUN_GRIDMIX.getValue()) {
       classArgs = new String[] {
           "-generate", values[3],
           gridmixDir.toString(),
           values[4]
       };
     } else if (gridmixRunMode == GridMixRunMode.RUN_GRIDMIX.getValue()
               && values[1].indexOf("RoundRobinUserResolver") > 0) {
       classArgs = new String[] {
           "-users", values[3],
           gridmixDir.toString(),
           values[4]
       };
     } else {
       classArgs = new String[] {
          gridmixDir.toString(),values[3]
       };
     }

     String [] args = new String [runtimeArgs.length +
        classArgs.length + ((otherArgs != null)?otherArgs.length:0)];
     System.arraycopy(runtimeArgs, 0, args, 0, runtimeArgs.length);

     if (otherArgs != null) {
       System.arraycopy(otherArgs, 0, args, runtimeArgs.length,
                        otherArgs.length);
       System.arraycopy(classArgs, 0, args, (runtimeArgs.length +
                        otherArgs.length), classArgs.length);
     } else {
       System.arraycopy(classArgs, 0, args, runtimeArgs.length,
                        classArgs.length);
     }
     return args;
   }

   /**
    * Create a file with specified size in mb.
    * @param sizeInMB - file size in mb.
    * @param inputDir - input directory.
    * @param conf - cluster configuration.
    * @throws Exception - if an exception occurs.
    */
   public static void createFile(int sizeInMB, Path inputDir,
       Configuration conf) throws Exception {
     Date d = new Date();
     SimpleDateFormat sdf = new SimpleDateFormat("ddMMyy_HHmmssS");
     String formatDate = sdf.format(d);
     FileSystem fs = inputDir.getFileSystem(conf);
     OutputStream out = fs.create(new Path(inputDir,"datafile_" + formatDate));
     final byte[] b = new byte[1024 * 1024];
     for (int index = 0; index < sizeInMB; index++) {
       out.write(b);
     }
     out.close();
     fs.close();
   }

   /**
    * Create directories for a path.
    * @param path - directories path.
    * @param conf  - cluster configuration.
    * @throws IOException  - if an I/O error occurs.
    */
   public static void createDirs(Path path,Configuration conf)
      throws IOException {
     FileSystem fs = path.getFileSystem(conf);
     if (!fs.exists(path)) {
        fs.mkdirs(path);
     }
   }

   /**
    * Run the Gridmix job with given runtime arguments.
    * @param gridmixDir - Gridmix parent directory.
    * @param conf - cluster configuration.
    * @param gridmixRunMode - gridmix run mode either 1,2,3
    * @param runtimeValues -gridmix runtime values.
    * @return - gridmix status either 0 or 1.
    * @throws Exception
    */
   public static int runGridmixJob(Path gridmixDir, Configuration conf,
      int gridmixRunMode, String [] runtimeValues) throws Exception {
     return runGridmixJob(gridmixDir, conf, gridmixRunMode, runtimeValues, null);
   }
   /**
    * Run the Gridmix job with given runtime arguments.
    * @param gridmixDir - Gridmix parent directory
    * @param conf - cluster configuration.
    * @param gridmixRunMode - gridmix run mode.
    * @param runtimeValues - gridmix runtime values.
    * @param otherArgs - gridmix other generic args.
    * @return - gridmix status either 0 or 1.
    * @throws Exception
    */

   public static int runGridmixJob(Path gridmixDir, Configuration conf,
                                   int gridmixRunMode, String [] runtimeValues,
                                   String [] otherArgs) throws Exception {
     Path  outputDir = new Path(gridmixDir, "gridmix");
     Path inputDir = new Path(gridmixDir, "input");
     LOG.info("Cleanup the data if data already exists.");
     String modeName = new String();
     switch (gridmixRunMode) {
       case 1 :
         cleanup(inputDir, conf);
         cleanup(outputDir, conf);
         modeName = GridMixRunMode.DATA_GENERATION.name();
         break;
       case 2 :
         cleanup(outputDir, conf);
         modeName = GridMixRunMode.RUN_GRIDMIX.name();
         break;
       case 3 :
         cleanup(inputDir, conf);
         cleanup(outputDir, conf);
         modeName = GridMixRunMode.DATA_GENERATION_AND_RUN_GRIDMIX.name();
         break;
     }

     final String [] args =
         UtilsForGridmix.getArgsList(gridmixDir, gridmixRunMode,
                                     runtimeValues, otherArgs);
     Gridmix gridmix = new Gridmix();
     LOG.info("Submit a Gridmix job in " + runtimeValues[1]
             + " mode for " + modeName);
     int exitCode = ToolRunner.run(conf, gridmix, args);
     return exitCode;
   }

   /**
    * Get the proxy users file.
    * @param conf - cluster configuration.
    * @return String - proxy users file.
    * @Exception - if no proxy users found in configuration.
    */
   public static String getProxyUsersFile(Configuration conf)
       throws Exception {
      ProxyUserDefinitions pud = getProxyUsersData(conf);
      String fileName = buildProxyUsersFile(pud.getProxyUsers());
      if (fileName == null) {
         LOG.error("Proxy users file not found.");
         throw new Exception("Proxy users file not found.");
      } else {
         return fileName;
      }
   }

   /**
   * List the current gridmix jobid's.
   * @param client - job client.
   * @param execJobCount - number of executed jobs.
   * @return - list of gridmix jobid's.
   */
  public static List<JobID> listGridmixJobIDs(JobClient client,
      int execJobCount) throws IOException {
    List<JobID> jobids = new ArrayList<JobID>();
    JobStatus [] jobStatus = client.getAllJobs();
    int numJobs = jobStatus.length;
    for (int index = 1; index <= execJobCount; index++) {
      JobStatus js = jobStatus[numJobs - index];
      JobID jobid = js.getJobID();
      String jobName = js.getJobName();
      if (!jobName.equals("GRIDMIX_GENERATE_INPUT_DATA") &&
          !jobName.equals("GRIDMIX_GENERATE_DISTCACHE_DATA")) {
        jobids.add(jobid);
      }
    }
    return (jobids.size() == 0)? null : jobids;
  }

  /**
   * List the proxy users.
   * @param conf
   * @return
   * @throws Exception
   */
  public static List<String> listProxyUsers(Configuration conf,
      String loginUser) throws Exception {
    List<String> proxyUsers = new ArrayList<String>();
    ProxyUserDefinitions pud = getProxyUsersData(conf);
    Map<String, GroupsAndHost> usersData = pud.getProxyUsers();
    Collection users = usersData.keySet();
    Iterator<String> itr = users.iterator();
    while (itr.hasNext()) {
      String user = itr.next();
      if (!user.equals(loginUser)){ proxyUsers.add(user); };
    }
    return proxyUsers;
  }

   private static String buildProxyUsersFile(final Map<String, GroupsAndHost>
       proxyUserData) throws Exception {
      FileOutputStream fos = null;
      File file = null;
      StringBuffer input = new StringBuffer();
      Set users = proxyUserData.keySet();
      Iterator itr = users.iterator();
      while (itr.hasNext()) {
        String user = itr.next().toString();
        if (!user.equals(
            UserGroupInformation.getLoginUser().getShortUserName())) {
          input.append(user);
          final GroupsAndHost gah = proxyUserData.get(user);
          final List <String> groups = gah.getGroups();
          for (String group : groups) {
            input.append(",");
            input.append(group);
          }
          input.append("\n");
        }
      }
      if (input.length() > 0) {
         try {
            file = File.createTempFile("proxyusers", null);
            fos = new FileOutputStream(file);
            fos.write(input.toString().getBytes());
         } catch(IOException ioexp) {
            LOG.warn(ioexp.getMessage());
            return null;
         } finally {
            fos.close();
            file.deleteOnExit();
         }
         LOG.info("file.toString():" + file.toString());
         return file.toString();
      } else {
         return null;
      }
   }

   private static ProxyUserDefinitions getProxyUsersData(Configuration conf)
       throws Exception {
     Iterator itr = conf.iterator();
     List<String> proxyUsersData = new ArrayList<String>();
     while (itr.hasNext()) {
       String property = itr.next().toString();
       if (property.indexOf("hadoop.proxyuser") >= 0
          && property.indexOf("groups=") >= 0) {
         proxyUsersData.add(property.split("\\.")[2]);
       }
     }

     if (proxyUsersData.size() == 0) {
        LOG.error("No proxy users found in the configuration.");
        throw new Exception("No proxy users found in the configuration.");
     }

     ProxyUserDefinitions pud = new ProxyUserDefinitions() {
        public boolean writeToFile(URI filePath) throws IOException {
            throw new UnsupportedOperationException("No such methood exists.");
        };
     };

      for (String userName : proxyUsersData) {
         List<String> groups = Arrays.asList(conf.get("hadoop.proxyuser." +
             userName + ".groups").split("//,"));
         List<String> hosts = Arrays.asList(conf.get("hadoop.proxyuser." +
             userName + ".hosts").split("//,"));
         ProxyUserDefinitions.GroupsAndHost definitions =
             pud.new GroupsAndHost();
         definitions.setGroups(groups);
         definitions.setHosts(hosts);
         pud.addProxyUser(userName, definitions);
      }
      return pud;
   }

   /**
    *  Gives the list of paths for MR traces against different time
    *  intervals.It fetches only the paths which followed the below
    *  file convention.
    *    Syntax : &lt;FileName&gt;_&lt;TimeIntervals&gt;.json.gz
    *  There is a restriction in a  file and user has to
    *  follow  the below convention for time interval.
    *    Syntax: &lt;numeric&gt;[m|h|d]
    *    e.g : for 10 minutes trace should specify 10m,
    *    same way for 1 hour traces should specify 1h,
    *    for 1 day traces should specify 1d.
    *
    * @param conf - cluster configuration.
    * @return - list of MR paths as key/value pair based on time interval.
    * @throws IOException - if an I/O error occurs.
    */
   public static Map<String, String> getMRTraces(Configuration conf)
      throws IOException {
     return getMRTraces(conf, DEFAULT_TRACES_PATH);
   }

   /**
    *  It gives the list of paths for MR traces against different time
    *  intervals. It fetches only the paths which followed the below
    *  file convention.
    *    Syntax : &lt;FileNames&gt;_&lt;TimeInterval&gt;.json.gz
    *  There is a restriction in a file and user has to follow the
    *  below convention for time interval.
    *    Syntax: &lt;numeric&gt;[m|h|d]
    *    e.g : for 10 minutes trace should specify 10m,
    *    same way for 1 hour traces should specify 1h,
    *    for 1 day  traces should specify 1d.
    *
    * @param conf - cluster configuration object.
    * @param tracesPath - MR traces path.
    * @return - list of MR paths as key/value pair based on time interval.
    * @throws IOException - If an I/O error occurs.
    */
   public static Map<String,String> getMRTraces(Configuration conf,
       Path tracesPath) throws IOException {
      Map <String, String> jobTraces = new HashMap <String, String>();
      final FileSystem fs = FileSystem.getLocal(conf);
      final FileStatus fstat[] = fs.listStatus(tracesPath);
      for (FileStatus fst : fstat) {
         final String fileName = fst.getPath().getName();
         if (fileName.endsWith("m.json.gz")
             || fileName.endsWith("h.json.gz")
             || fileName.endsWith("d.json.gz")) {
            jobTraces.put(fileName.substring(fileName.indexOf("_") + 1,
               fileName.indexOf(".json.gz")), fst.getPath().toString());
         }
      }
      if (jobTraces.size() == 0) {
         LOG.error("No traces found in " + tracesPath.toString() + " path.");
         throw new IOException("No traces found in "
                              + tracesPath.toString() + " path.");
      }
      return jobTraces;
   }

   /**
    * It list the all the MR traces path irrespective of time.
    * @param conf - cluster configuration.
    * @param tracesPath - MR traces path
    * @return - MR paths as a list.
    * @throws IOException - if an I/O error occurs.
    */
   public static List<String> listMRTraces(Configuration conf,
       Path tracesPath) throws IOException {
      List<String> jobTraces = new ArrayList<String>();
      final FileSystem fs = FileSystem.getLocal(conf);
      final FileStatus fstat[] = fs.listStatus(tracesPath);
      for (FileStatus fst : fstat) {
         jobTraces.add(fst.getPath().toString());
      }
      if (jobTraces.size() == 0) {
         LOG.error("No traces found in " + tracesPath.toString() + " path.");
         throw new IOException("No traces found in "
                              + tracesPath.toString() + " path.");
      }
      return jobTraces;
   }

   /**
    * It list the all the MR traces path irrespective of time.
    * @param conf - cluster configuration.
    * @param tracesPath - MR traces path
    * @return - MR paths as a list.
    * @throws IOException - if an I/O error occurs.
    */
   public static List<String> listMRTraces(Configuration conf)
       throws IOException {
      return listMRTraces(conf, DEFAULT_TRACES_PATH);
   }

   /**
    * Gives the list of MR traces for given time interval.
    * The time interval should be following convention.
    *   Syntax : &lt;numeric&gt;[m|h|d]
    *   e.g : 10m or 1h or 2d etc.
    * @param conf - cluster configuration
    * @param timeInterval - trace time interval.
    * @param tracesPath - MR traces Path.
    * @return - MR paths as a list for a given time interval.
    * @throws IOException - If an I/O error occurs.
    */
   public static List<String> listMRTracesByTime(Configuration conf,
       String timeInterval, Path tracesPath) throws IOException {
      List<String> jobTraces = new ArrayList<String>();
      final FileSystem fs = FileSystem.getLocal(conf);
      final FileStatus fstat[] = fs.listStatus(tracesPath);
      for (FileStatus fst : fstat) {
         final String fileName = fst.getPath().getName();
         if (fileName.indexOf(timeInterval) >= 0) {
            jobTraces.add(fst.getPath().toString());
         }
      }
      return jobTraces;
   }

   /**
    * Gives the list of MR traces for given time interval.
    * The time interval should be following convention.
    *   Syntax : &lt;numeric&gt;[m|h|d]
    *   e.g : 10m or 1h or 2d etc.
    * @param conf - cluster configuration
    * @param timeInterval - trace time interval.
    * @return - MR paths as a list for a given time interval.
    * @throws IOException - If an I/O error occurs.
    */
   public static List<String> listMRTracesByTime(Configuration conf,
       String timeInterval) throws IOException {
      return listMRTracesByTime(conf, timeInterval, DEFAULT_TRACES_PATH);
   }
 }
	/**
	* 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.mapred.gridmix.test.system;

	import java.io.IOException;
	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.security.UserGroupInformation;
	import org.apache.hadoop.util.ToolRunner;
	import org.apache.hadoop.mapred.gridmix.Gridmix;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.mapred.JobClient;
	import org.apache.hadoop.mapred.JobStatus;
	import org.apache.hadoop.mapreduce.JobID;
	import java.util.Date;
	import java.util.HashMap;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.Arrays;
	import java.net.URI;
	import java.text.SimpleDateFormat;
	import java.io.OutputStream;
	import java.util.Set;
	import java.util.List;
	import java.util.Iterator;
	import java.util.Map;
	import java.io.File;
	import java.io.FileOutputStream;
	import org.apache.hadoop.test.system.ProxyUserDefinitions;
	import org.apache.hadoop.test.system.ProxyUserDefinitions.GroupsAndHost;

	/**
	* Gridmix utilities.
	*/
	public class UtilsForGridmix {
	private static final Log LOG = LogFactory.getLog(UtilsForGridmix.class);
	private static final Path DEFAULT_TRACES_PATH =
	new Path(System.getProperty("user.dir") + "/src/test/system/resources/");

	/**
	* cleanup the folder or file.
	* @param path - folder or file path.
	* @param conf - cluster configuration
	* @throws IOException - If an I/O error occurs.
	*/
	public static void cleanup(Path path, Configuration conf)
	throws IOException {
	FileSystem fs = path.getFileSystem(conf);
	fs.delete(path, true);
	fs.close();
	}

	/**
	* Get the login user.
	* @return - login user as string..
	* @throws IOException - if an I/O error occurs.
	*/
	public static String getUserName() throws IOException {
	return UserGroupInformation.getLoginUser().getUserName();
	}

	/**
	* Get the argument list for gridmix job.
	* @param gridmixDir - gridmix parent directory.
	* @param gridmixRunMode - gridmix modes either 1,2,3.
	* @param values - gridmix runtime values.
	* @param otherArgs - gridmix other generic args.
	* @return - argument list as string array.
	*/
	public static String [] getArgsList(Path gridmixDir, int gridmixRunMode,
	String [] values, String [] otherArgs) {
	String [] runtimeArgs = {
	"-D", GridMixConfig.GRIDMIX_LOG_MODE + "=DEBUG",
	"-D", GridMixConfig.GRIDMIX_OUTPUT_DIR + "=gridmix",
	"-D", GridMixConfig.GRIDMIX_JOB_SUBMISSION_QUEUE_IN_TRACE + "=true",
	"-D", GridMixConfig.GRIDMIX_JOB_TYPE + "=" + values[0],
	"-D", GridMixConfig.GRIDMIX_USER_RESOLVER + "=" + values[1],
	"-D", GridMixConfig.GRIDMIX_SUBMISSION_POLICY + "=" + values[2]
	};

	String [] classArgs;
	if ((gridmixRunMode == GridMixRunMode.DATA_GENERATION.getValue()
	\|\| gridmixRunMode
	== GridMixRunMode.DATA_GENERATION_AND_RUN_GRIDMIX.getValue())
	&& values[1].indexOf("RoundRobinUserResolver") > 0) {
	classArgs = new String[] {
	"-generate", values[3],
	"-users", values[4],
	gridmixDir.toString(),
	values[5]
	};
	} else if (gridmixRunMode == GridMixRunMode.DATA_GENERATION.getValue()
	\|\| gridmixRunMode
	== GridMixRunMode.DATA_GENERATION_AND_RUN_GRIDMIX.getValue()) {
	classArgs = new String[] {
	"-generate", values[3],
	gridmixDir.toString(),
	values[4]
	};
	} else if (gridmixRunMode == GridMixRunMode.RUN_GRIDMIX.getValue()
	&& values[1].indexOf("RoundRobinUserResolver") > 0) {
	classArgs = new String[] {
	"-users", values[3],
	gridmixDir.toString(),
	values[4]
	};
	} else {
	classArgs = new String[] {
	gridmixDir.toString(),values[3]
	};
	}

	String [] args = new String [runtimeArgs.length +
	classArgs.length + ((otherArgs != null)?otherArgs.length:0)];
	System.arraycopy(runtimeArgs, 0, args, 0, runtimeArgs.length);

	if (otherArgs != null) {
	System.arraycopy(otherArgs, 0, args, runtimeArgs.length,
	otherArgs.length);
	System.arraycopy(classArgs, 0, args, (runtimeArgs.length +
	otherArgs.length), classArgs.length);
	} else {
	System.arraycopy(classArgs, 0, args, runtimeArgs.length,
	classArgs.length);
	}
	return args;
	}

	/**
	* Create a file with specified size in mb.
	* @param sizeInMB - file size in mb.
	* @param inputDir - input directory.
	* @param conf - cluster configuration.
	* @throws Exception - if an exception occurs.
	*/
	public static void createFile(int sizeInMB, Path inputDir,
	Configuration conf) throws Exception {
	Date d = new Date();
	SimpleDateFormat sdf = new SimpleDateFormat("ddMMyy_HHmmssS");
	String formatDate = sdf.format(d);
	FileSystem fs = inputDir.getFileSystem(conf);
	OutputStream out = fs.create(new Path(inputDir,"datafile_" + formatDate));
	final byte[] b = new byte[1024 * 1024];
	for (int index = 0; index < sizeInMB; index++) {
	out.write(b);
	}
	out.close();
	fs.close();
	}

	/**
	* Create directories for a path.
	* @param path - directories path.
	* @param conf - cluster configuration.
	* @throws IOException - if an I/O error occurs.
	*/
	public static void createDirs(Path path,Configuration conf)
	throws IOException {
	FileSystem fs = path.getFileSystem(conf);
	if (!fs.exists(path)) {
	fs.mkdirs(path);
	}
	}

	/**
	* Run the Gridmix job with given runtime arguments.
	* @param gridmixDir - Gridmix parent directory.
	* @param conf - cluster configuration.
	* @param gridmixRunMode - gridmix run mode either 1,2,3
	* @param runtimeValues -gridmix runtime values.
	* @return - gridmix status either 0 or 1.
	* @throws Exception
	*/
	public static int runGridmixJob(Path gridmixDir, Configuration conf,
	int gridmixRunMode, String [] runtimeValues) throws Exception {
	return runGridmixJob(gridmixDir, conf, gridmixRunMode, runtimeValues, null);
	}
	/**
	* Run the Gridmix job with given runtime arguments.
	* @param gridmixDir - Gridmix parent directory
	* @param conf - cluster configuration.
	* @param gridmixRunMode - gridmix run mode.
	* @param runtimeValues - gridmix runtime values.
	* @param otherArgs - gridmix other generic args.
	* @return - gridmix status either 0 or 1.
	* @throws Exception
	*/

	public static int runGridmixJob(Path gridmixDir, Configuration conf,
	int gridmixRunMode, String [] runtimeValues,
	String [] otherArgs) throws Exception {
	Path outputDir = new Path(gridmixDir, "gridmix");
	Path inputDir = new Path(gridmixDir, "input");
	LOG.info("Cleanup the data if data already exists.");
	String modeName = new String();
	switch (gridmixRunMode) {
	case 1 :
	cleanup(inputDir, conf);
	cleanup(outputDir, conf);
	modeName = GridMixRunMode.DATA_GENERATION.name();
	break;
	case 2 :
	cleanup(outputDir, conf);
	modeName = GridMixRunMode.RUN_GRIDMIX.name();
	break;
	case 3 :
	cleanup(inputDir, conf);
	cleanup(outputDir, conf);
	modeName = GridMixRunMode.DATA_GENERATION_AND_RUN_GRIDMIX.name();
	break;
	}

	final String [] args =
	UtilsForGridmix.getArgsList(gridmixDir, gridmixRunMode,
	runtimeValues, otherArgs);
	Gridmix gridmix = new Gridmix();
	LOG.info("Submit a Gridmix job in " + runtimeValues[1]
	+ " mode for " + modeName);
	int exitCode = ToolRunner.run(conf, gridmix, args);
	return exitCode;
	}

	/**
	* Get the proxy users file.
	* @param conf - cluster configuration.
	* @return String - proxy users file.
	* @Exception - if no proxy users found in configuration.
	*/
	public static String getProxyUsersFile(Configuration conf)
	throws Exception {
	ProxyUserDefinitions pud = getProxyUsersData(conf);
	String fileName = buildProxyUsersFile(pud.getProxyUsers());
	if (fileName == null) {
	LOG.error("Proxy users file not found.");
	throw new Exception("Proxy users file not found.");
	} else {
	return fileName;
	}
	}

	/**
	* List the current gridmix jobid's.
	* @param client - job client.
	* @param execJobCount - number of executed jobs.
	* @return - list of gridmix jobid's.
	*/
	public static List<JobID> listGridmixJobIDs(JobClient client,
	int execJobCount) throws IOException {
	List<JobID> jobids = new ArrayList<JobID>();
	JobStatus [] jobStatus = client.getAllJobs();
	int numJobs = jobStatus.length;
	for (int index = 1; index <= execJobCount; index++) {
	JobStatus js = jobStatus[numJobs - index];
	JobID jobid = js.getJobID();
	String jobName = js.getJobName();
	if (!jobName.equals("GRIDMIX_GENERATE_INPUT_DATA") &&
	!jobName.equals("GRIDMIX_GENERATE_DISTCACHE_DATA")) {
	jobids.add(jobid);
	}
	}
	return (jobids.size() == 0)? null : jobids;
	}

	/**
	* List the proxy users.
	* @param conf
	* @return
	* @throws Exception
	*/
	public static List<String> listProxyUsers(Configuration conf,
	String loginUser) throws Exception {
	List<String> proxyUsers = new ArrayList<String>();
	ProxyUserDefinitions pud = getProxyUsersData(conf);
	Map<String, GroupsAndHost> usersData = pud.getProxyUsers();
	Collection users = usersData.keySet();
	Iterator<String> itr = users.iterator();
	while (itr.hasNext()) {
	String user = itr.next();
	if (!user.equals(loginUser)){ proxyUsers.add(user); };
	}
	return proxyUsers;
	}

	private static String buildProxyUsersFile(final Map<String, GroupsAndHost>
	proxyUserData) throws Exception {
	FileOutputStream fos = null;
	File file = null;
	StringBuffer input = new StringBuffer();
	Set users = proxyUserData.keySet();
	Iterator itr = users.iterator();
	while (itr.hasNext()) {
	String user = itr.next().toString();
	if (!user.equals(
	UserGroupInformation.getLoginUser().getShortUserName())) {
	input.append(user);
	final GroupsAndHost gah = proxyUserData.get(user);
	final List <String> groups = gah.getGroups();
	for (String group : groups) {
	input.append(",");
	input.append(group);
	}
	input.append("\n");
	}
	}
	if (input.length() > 0) {
	try {
	file = File.createTempFile("proxyusers", null);
	fos = new FileOutputStream(file);
	fos.write(input.toString().getBytes());
	} catch(IOException ioexp) {
	LOG.warn(ioexp.getMessage());
	return null;
	} finally {
	fos.close();
	file.deleteOnExit();
	}
	LOG.info("file.toString():" + file.toString());
	return file.toString();
	} else {
	return null;
	}
	}

	private static ProxyUserDefinitions getProxyUsersData(Configuration conf)
	throws Exception {
	Iterator itr = conf.iterator();
	List<String> proxyUsersData = new ArrayList<String>();
	while (itr.hasNext()) {
	String property = itr.next().toString();
	if (property.indexOf("hadoop.proxyuser") >= 0
	&& property.indexOf("groups=") >= 0) {
	proxyUsersData.add(property.split("\\.")[2]);
	}
	}

	if (proxyUsersData.size() == 0) {
	LOG.error("No proxy users found in the configuration.");
	throw new Exception("No proxy users found in the configuration.");
	}

	ProxyUserDefinitions pud = new ProxyUserDefinitions() {
	public boolean writeToFile(URI filePath) throws IOException {
	throw new UnsupportedOperationException("No such methood exists.");
	};
	};

	for (String userName : proxyUsersData) {
	List<String> groups = Arrays.asList(conf.get("hadoop.proxyuser." +
	userName + ".groups").split("//,"));
	List<String> hosts = Arrays.asList(conf.get("hadoop.proxyuser." +
	userName + ".hosts").split("//,"));
	ProxyUserDefinitions.GroupsAndHost definitions =
	pud.new GroupsAndHost();
	definitions.setGroups(groups);
	definitions.setHosts(hosts);
	pud.addProxyUser(userName, definitions);
	}
	return pud;
	}

	/**
	* Gives the list of paths for MR traces against different time
	* intervals.It fetches only the paths which followed the below
	* file convention.
	* Syntax : <FileName>_<TimeIntervals>.json.gz
	* There is a restriction in a file and user has to
	* follow the below convention for time interval.
	* Syntax: <numeric>[m\|h\|d]
	* e.g : for 10 minutes trace should specify 10m,
	* same way for 1 hour traces should specify 1h,
	* for 1 day traces should specify 1d.
	*
	* @param conf - cluster configuration.
	* @return - list of MR paths as key/value pair based on time interval.
	* @throws IOException - if an I/O error occurs.
	*/
	public static Map<String, String> getMRTraces(Configuration conf)
	throws IOException {
	return getMRTraces(conf, DEFAULT_TRACES_PATH);
	}

	/**
	* It gives the list of paths for MR traces against different time
	* intervals. It fetches only the paths which followed the below
	* file convention.
	* Syntax : <FileNames>_<TimeInterval>.json.gz
	* There is a restriction in a file and user has to follow the
	* below convention for time interval.
	* Syntax: <numeric>[m\|h\|d]
	* e.g : for 10 minutes trace should specify 10m,
	* same way for 1 hour traces should specify 1h,
	* for 1 day traces should specify 1d.
	*
	* @param conf - cluster configuration object.
	* @param tracesPath - MR traces path.
	* @return - list of MR paths as key/value pair based on time interval.
	* @throws IOException - If an I/O error occurs.
	*/
	public static Map<String,String> getMRTraces(Configuration conf,
	Path tracesPath) throws IOException {
	Map <String, String> jobTraces = new HashMap <String, String>();
	final FileSystem fs = FileSystem.getLocal(conf);
	final FileStatus fstat[] = fs.listStatus(tracesPath);
	for (FileStatus fst : fstat) {
	final String fileName = fst.getPath().getName();
	if (fileName.endsWith("m.json.gz")
	\|\| fileName.endsWith("h.json.gz")
	\|\| fileName.endsWith("d.json.gz")) {
	jobTraces.put(fileName.substring(fileName.indexOf("_") + 1,
	fileName.indexOf(".json.gz")), fst.getPath().toString());
	}
	}
	if (jobTraces.size() == 0) {
	LOG.error("No traces found in " + tracesPath.toString() + " path.");
	throw new IOException("No traces found in "
	+ tracesPath.toString() + " path.");
	}
	return jobTraces;
	}

	/**
	* It list the all the MR traces path irrespective of time.
	* @param conf - cluster configuration.
	* @param tracesPath - MR traces path
	* @return - MR paths as a list.
	* @throws IOException - if an I/O error occurs.
	*/
	public static List<String> listMRTraces(Configuration conf,
	Path tracesPath) throws IOException {
	List<String> jobTraces = new ArrayList<String>();
	final FileSystem fs = FileSystem.getLocal(conf);
	final FileStatus fstat[] = fs.listStatus(tracesPath);
	for (FileStatus fst : fstat) {
	jobTraces.add(fst.getPath().toString());
	}
	if (jobTraces.size() == 0) {
	LOG.error("No traces found in " + tracesPath.toString() + " path.");
	throw new IOException("No traces found in "
	+ tracesPath.toString() + " path.");
	}
	return jobTraces;
	}

	/**
	* It list the all the MR traces path irrespective of time.
	* @param conf - cluster configuration.
	* @param tracesPath - MR traces path
	* @return - MR paths as a list.
	* @throws IOException - if an I/O error occurs.
	*/
	public static List<String> listMRTraces(Configuration conf)
	throws IOException {
	return listMRTraces(conf, DEFAULT_TRACES_PATH);
	}

	/**
	* Gives the list of MR traces for given time interval.
	* The time interval should be following convention.
	* Syntax : <numeric>[m\|h\|d]
	* e.g : 10m or 1h or 2d etc.
	* @param conf - cluster configuration
	* @param timeInterval - trace time interval.
	* @param tracesPath - MR traces Path.
	* @return - MR paths as a list for a given time interval.
	* @throws IOException - If an I/O error occurs.
	*/
	public static List<String> listMRTracesByTime(Configuration conf,
	String timeInterval, Path tracesPath) throws IOException {
	List<String> jobTraces = new ArrayList<String>();
	final FileSystem fs = FileSystem.getLocal(conf);
	final FileStatus fstat[] = fs.listStatus(tracesPath);
	for (FileStatus fst : fstat) {
	final String fileName = fst.getPath().getName();
	if (fileName.indexOf(timeInterval) >= 0) {
	jobTraces.add(fst.getPath().toString());
	}
	}
	return jobTraces;
	}

	/**
	* Gives the list of MR traces for given time interval.
	* The time interval should be following convention.
	* Syntax : <numeric>[m\|h\|d]
	* e.g : 10m or 1h or 2d etc.
	* @param conf - cluster configuration
	* @param timeInterval - trace time interval.
	* @return - MR paths as a list for a given time interval.
	* @throws IOException - If an I/O error occurs.
	*/
	public static List<String> listMRTracesByTime(Configuration conf,
	String timeInterval) throws IOException {
	return listMRTracesByTime(conf, timeInterval, DEFAULT_TRACES_PATH);
	}
	}