| /** |
| * 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 java.io.File; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Map; |
| import java.util.HashMap; |
| import java.util.SortedMap; |
| import java.util.TreeMap; |
| import java.util.Set; |
| import java.util.Collections; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.permission.FsPermission; |
| import org.apache.hadoop.fs.permission.FsAction; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.mapreduce.Counters; |
| import org.apache.hadoop.mapreduce.Counter; |
| import org.apache.hadoop.mapreduce.CounterGroup; |
| import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser; |
| import org.apache.hadoop.mapred.RunningJob; |
| import org.apache.hadoop.mapreduce.JobID; |
| import org.apache.hadoop.mapreduce.TaskType; |
| import org.apache.hadoop.mapreduce.test.system.JTClient; |
| import org.apache.hadoop.mapred.JobConf; |
| import org.apache.hadoop.tools.rumen.LoggedJob; |
| import org.apache.hadoop.tools.rumen.ZombieJob; |
| import org.apache.hadoop.tools.rumen.TaskInfo; |
| import org.junit.Assert; |
| import java.text.ParseException; |
| import org.apache.hadoop.security.UserGroupInformation; |
| import org.apache.hadoop.mapred.gridmix.GridmixSystemTestCase; |
| /** |
| * Verifying each Gridmix job with corresponding job story in a trace file. |
| */ |
| public class GridmixJobVerification { |
| |
| private static Log LOG = LogFactory.getLog(GridmixJobVerification.class); |
| private Path path; |
| private Configuration conf; |
| private JTClient jtClient; |
| private Map<String, List<JobConf>> simuAndOrigJobsInfo = |
| new HashMap<String, List<JobConf>>(); |
| /** |
| * Gridmix job verification constructor |
| * @param path - path of the gridmix output directory. |
| * @param conf - cluster configuration. |
| * @param jtClient - jobtracker client. |
| */ |
| public GridmixJobVerification(Path path, Configuration conf, |
| JTClient jtClient) { |
| this.path = path; |
| this.conf = conf; |
| this.jtClient = jtClient; |
| } |
| |
| /** |
| * It verifies the Gridmix jobs with corresponding job story in a trace file. |
| * @param jobids - gridmix job ids. |
| * @throws IOException - if an I/O error occurs. |
| * @throws ParseException - if an parse error occurs. |
| */ |
| public void verifyGridmixJobsWithJobStories(List<JobID> jobids) |
| throws IOException, ParseException { |
| |
| SortedMap <Long, String> origSubmissionTime = new TreeMap <Long, String>(); |
| SortedMap <Long, String> simuSubmissionTime = new TreeMap<Long, String>(); |
| GridmixJobStory gjs = new GridmixJobStory(path, conf); |
| final Iterator<JobID> ite = jobids.iterator(); |
| File destFolder = new File(System.getProperty("java.io.tmpdir") + |
| "/gridmix-st/"); |
| |
| destFolder.mkdir(); |
| while (ite.hasNext()) { |
| long expMapInputBytes = 0; |
| long expMapOutputBytes = 0; |
| long expMapInputRecs = 0; |
| long expMapOutputRecs = 0; |
| long expReduceInputBytes = 0; |
| long expReduceOutputBytes = 0; |
| long expReduceInputRecs = 0; |
| long expReduceOutputRecs = 0; |
| |
| JobID simuJobId = ite.next(); |
| JobHistoryParser.JobInfo jhInfo = getSimulatedJobHistory(simuJobId); |
| Assert.assertNotNull("Job history not found.", jhInfo); |
| Counters counters = jhInfo.getTotalCounters(); |
| JobConf simuJobConf = getSimulatedJobConf(simuJobId,destFolder); |
| String origJobId = simuJobConf.get("gridmix.job.original-job-id"); |
| LOG.info("OriginalJobID<->CurrentJobID:" + |
| origJobId + "<->" + simuJobId); |
| |
| ZombieJob zombieJob = gjs.getZombieJob(JobID.forName(origJobId)); |
| Map<String, Long> mapJobCounters = getJobMapCounters(zombieJob); |
| Map<String, Long> reduceJobCounters = getJobReduceCounters(zombieJob); |
| |
| if (simuJobConf.get("gridmix.job-submission.policy").contains("REPLAY")) { |
| origSubmissionTime.put(zombieJob.getSubmissionTime(), |
| origJobId.toString() + "^" + simuJobId); |
| simuSubmissionTime.put(jhInfo.getSubmitTime() , |
| origJobId.toString() + "^" + simuJobId); ; |
| } |
| |
| LOG.info("Verifying the job <" + simuJobId + "> and wait for a while..."); |
| verifySimulatedJobSummary(zombieJob, jhInfo, simuJobConf); |
| verifyJobMapCounters(counters,mapJobCounters,simuJobConf); |
| verifyJobReduceCounters(counters,reduceJobCounters,simuJobConf); |
| verifyDistributeCache(zombieJob,simuJobConf); |
| setJobDistributedCacheInfo(simuJobId.toString(), simuJobConf, |
| zombieJob.getJobConf()); |
| LOG.info("Done."); |
| } |
| verifyDistributedCacheBetweenJobs(simuAndOrigJobsInfo); |
| } |
| |
| /** |
| * Verify the distributed cache files between the jobs in a gridmix run. |
| * @param jobsInfo - jobConfs of simulated and original jobs as a map. |
| */ |
| public void verifyDistributedCacheBetweenJobs(Map<String, |
| List<JobConf>> jobsInfo) { |
| if (jobsInfo.size() > 1) { |
| Map<String, Integer> simJobfilesOccurBtnJobs = |
| getDistcacheFilesOccurenceBetweenJobs(jobsInfo, 0); |
| Map<String, Integer> origJobfilesOccurBtnJobs = |
| getDistcacheFilesOccurenceBetweenJobs(jobsInfo, 1); |
| List<Integer> simuOccurList = |
| getMapValuesAsList(simJobfilesOccurBtnJobs); |
| Collections.sort(simuOccurList); |
| List<Integer> origOccurList = |
| getMapValuesAsList(origJobfilesOccurBtnJobs); |
| Collections.sort(origOccurList); |
| Assert.assertTrue("The unique count of distibuted cache files in " + |
| "simulated jobs have not matched with the unique count of " + |
| "original jobs distributed files ", |
| simuOccurList.size() == origOccurList.size()); |
| int index = 0; |
| for(Integer origDistFileCount : origOccurList) { |
| Assert.assertTrue("Distributed cache file reused in simulated " + |
| "jobs has not matched with reused of distributed cache file " + |
| "in original jobs.",origDistFileCount == simuOccurList.get(index)); |
| index ++; |
| } |
| } |
| } |
| |
| private List<Integer> getMapValuesAsList(Map<String,Integer> jobOccurs) { |
| List<Integer> occursList = new ArrayList<Integer>(); |
| Set<String> files = jobOccurs.keySet(); |
| Iterator<String > ite = files.iterator(); |
| while(ite.hasNext()) { |
| String file = ite.next(); |
| occursList.add(jobOccurs.get(file)); |
| } |
| return occursList; |
| } |
| |
| |
| /** |
| * Get the unique distributed cache files and occurrence between the jobs. |
| * @param jobsInfo - job's configurations as a map. |
| * @param jobConfIndex - 0 for simulated job configuration and |
| * 1 for original jobs configuration. |
| * @return - unique distributed cache files and occurrences as map. |
| */ |
| private Map<String, Integer> getDistcacheFilesOccurenceBetweenJobs( |
| Map<String, List<JobConf>> jobsInfo, int jobConfIndex) { |
| Map<String,Integer> filesOccurBtnJobs = new HashMap <String,Integer>(); |
| Set<String> jobIds = jobsInfo.keySet(); |
| Iterator<String > ite = jobIds.iterator(); |
| while(ite.hasNext()){ |
| String jobId = ite.next(); |
| List<JobConf> jobconfs = jobsInfo.get(jobId); |
| String [] distCacheFiles = jobconfs.get(jobConfIndex). |
| get(GridMixConfig.GRIDMIX_DISTCACHE_FILES).split(","); |
| String [] distCacheFileTimeStamps = jobconfs.get(jobConfIndex). |
| get(GridMixConfig.GRIDMIX_DISTCACHE_TIMESTAMP).split(","); |
| String [] distCacheFileVisib = jobconfs.get(jobConfIndex). |
| get(GridMixConfig.GRIDMIX_DISTCACHE_VISIBILITIES).split(","); |
| int indx = 0; |
| for (String distCacheFile : distCacheFiles) { |
| String fileAndSize = distCacheFile + "^" + |
| distCacheFileTimeStamps[indx] + "^" + |
| jobconfs.get(jobConfIndex).getUser(); |
| if (filesOccurBtnJobs.get(fileAndSize)!= null) { |
| int count = filesOccurBtnJobs.get(fileAndSize); |
| count ++; |
| filesOccurBtnJobs.put(fileAndSize,count); |
| } else { |
| filesOccurBtnJobs.put(fileAndSize,1); |
| } |
| } |
| } |
| return filesOccurBtnJobs; |
| } |
| |
| private void setJobDistributedCacheInfo(String jobId, JobConf simuJobConf, |
| JobConf origJobConf) { |
| if (simuJobConf.get(GridMixConfig.GRIDMIX_DISTCACHE_FILES)!= null) { |
| List<JobConf> jobConfs = new ArrayList<JobConf>(); |
| jobConfs.add(simuJobConf); |
| jobConfs.add(origJobConf); |
| simuAndOrigJobsInfo.put(jobId,jobConfs); |
| } |
| } |
| |
| /** |
| * Verify the job subimssion order between the jobs in replay mode. |
| * @param origSubmissionTime - sorted map of original jobs submission times. |
| * @param simuSubmissionTime - sorted map of simulated jobs submission times. |
| */ |
| public void verifyJobSumissionTime(SortedMap<Long, String> origSubmissionTime, |
| SortedMap<Long, String> simuSubmissionTime){ |
| Assert.assertTrue("Simulated job's submission time count has " + |
| "not match with Original job's submission time count.", |
| origSubmissionTime.size() == simuSubmissionTime.size()); |
| for ( int index = 0; index < origSubmissionTime.size(); index ++) { |
| String origAndSimuJobID = origSubmissionTime.get(index); |
| String simuAndorigJobID = simuSubmissionTime.get(index); |
| Assert.assertEquals("Simulated jobs have not submitted in same " + |
| "order as original jobs submitted inREPLAY mode.", |
| origAndSimuJobID, simuAndorigJobID); |
| } |
| } |
| |
| /** |
| * It verifies the distributed cache emulation of a job. |
| * @param zombieJob - Original job story. |
| * @param simuJobConf - Simulated job configuration. |
| */ |
| public void verifyDistributeCache(ZombieJob zombieJob, |
| JobConf simuJobConf) throws IOException { |
| |
| if (simuJobConf.getBoolean(GridMixConfig.GRIDMIX_DISTCACHE_ENABLE, false)) { |
| JobConf origJobConf = zombieJob.getJobConf(); |
| checkFileVisibility(simuJobConf); |
| checkDistcacheFiles(simuJobConf,origJobConf); |
| checkFileSizes(simuJobConf,origJobConf); |
| checkFileStamps(simuJobConf,origJobConf); |
| } else { |
| Assert.assertNull("Configuration has distributed cache visibilites" + |
| "without enabled distributed cache emulation.", simuJobConf. |
| get(GridMixConfig.GRIDMIX_DISTCACHE_VISIBILITIES)); |
| Assert.assertNull("Configuration has distributed cache files time " + |
| "stamps without enabled distributed cache emulation.",simuJobConf. |
| get(GridMixConfig.GRIDMIX_DISTCACHE_TIMESTAMP)); |
| Assert.assertNull("Configuration has distributed cache files paths" + |
| "without enabled distributed cache emulation.",simuJobConf. |
| get(GridMixConfig.GRIDMIX_DISTCACHE_FILES)); |
| Assert.assertNull("Configuration has distributed cache files sizes" + |
| "without enabled distributed cache emulation.",simuJobConf. |
| get(GridMixConfig.GRIDMIX_DISTCACHE_FILESSIZE)); |
| } |
| } |
| |
| private void checkFileStamps(JobConf simuJobConf, JobConf origJobConf) { |
| //Verify simulated jobs against distributed cache files time stamps. |
| String [] origDCFTS = origJobConf.get(GridMixConfig. |
| GRIDMIX_DISTCACHE_TIMESTAMP).split(","); |
| String [] simuDCFTS = simuJobConf.get(GridMixConfig. |
| GRIDMIX_DISTCACHE_TIMESTAMP).split(","); |
| for (int index = 0; index < origDCFTS.length; index++) { |
| Assert.assertTrue("Invalid time stamps between original " + |
| "and simulated job",Long.parseLong(origDCFTS[index]) < |
| Long.parseLong(simuDCFTS[index])); |
| } |
| } |
| |
| private void checkFileVisibility(JobConf simuJobConf ) { |
| // Verify simulated jobs against distributed cache files visibilities. |
| String [] distFiles = simuJobConf.get(GridMixConfig. |
| GRIDMIX_DISTCACHE_FILES).split(","); |
| String [] simuDistVisibilities = simuJobConf.get(GridMixConfig. |
| GRIDMIX_DISTCACHE_VISIBILITIES).split(","); |
| List<Boolean> expFileVisibility = new ArrayList<Boolean >(); |
| int index = 0; |
| for (String distFile : distFiles) { |
| if (!GridmixSystemTestCase.isLocalDistCache(distFile, |
| simuJobConf.getUser(),Boolean.valueOf(simuDistVisibilities[index]))) { |
| expFileVisibility.add(true); |
| } else { |
| expFileVisibility.add(false); |
| } |
| index ++; |
| } |
| index = 0; |
| for (String actFileVisibility : simuDistVisibilities) { |
| Assert.assertEquals("Simulated job distributed cache file " + |
| "visibilities has not matched.", |
| expFileVisibility.get(index),Boolean.valueOf(actFileVisibility)); |
| index ++; |
| } |
| } |
| |
| private void checkDistcacheFiles(JobConf simuJobConf, JobConf origJobConf) |
| throws IOException { |
| //Verify simulated jobs against distributed cache files. |
| String [] origDistFiles = origJobConf.get(GridMixConfig. |
| GRIDMIX_DISTCACHE_FILES).split(","); |
| String [] simuDistFiles = simuJobConf.get(GridMixConfig. |
| GRIDMIX_DISTCACHE_FILES).split(","); |
| String [] simuDistVisibilities = simuJobConf.get(GridMixConfig. |
| GRIDMIX_DISTCACHE_VISIBILITIES).split(","); |
| Assert.assertEquals("No. of simulatued job's distcache files " + |
| "haven't matched with no.of original job's distcache files", |
| origDistFiles.length, simuDistFiles.length); |
| |
| int index = 0; |
| for (String simDistFile : simuDistFiles) { |
| Path distPath = new Path(simDistFile); |
| if (!GridmixSystemTestCase.isLocalDistCache(simDistFile, |
| simuJobConf.getUser(),Boolean.valueOf(simuDistVisibilities[index]))) { |
| FileSystem fs = distPath.getFileSystem(conf); |
| FileStatus fstat = fs.getFileStatus(distPath); |
| FsPermission permission = fstat.getPermission(); |
| Assert.assertTrue("HDFS distributed cache file has wrong " + |
| "permissions for users.", FsAction.READ_WRITE.SYMBOL == |
| permission.getUserAction().SYMBOL); |
| Assert.assertTrue("HDFS distributed cache file has wrong " + |
| "permissions for groups.",FsAction.READ.SYMBOL == |
| permission.getGroupAction().SYMBOL); |
| Assert.assertTrue("HDSFS distributed cache file has wrong " + |
| "permissions for others.",FsAction.READ.SYMBOL == |
| permission.getOtherAction().SYMBOL); |
| } |
| } |
| index ++; |
| } |
| |
| private void checkFileSizes(JobConf simuJobConf, JobConf origJobConf) { |
| // Verify simulated jobs against distributed cache files size. |
| List<String> origDistFilesSize = Arrays.asList(origJobConf. |
| get(GridMixConfig.GRIDMIX_DISTCACHE_FILESSIZE).split(",")); |
| Collections.sort(origDistFilesSize); |
| List<String> simuDistFilesSize = Arrays.asList(simuJobConf. |
| get(GridMixConfig.GRIDMIX_DISTCACHE_FILESSIZE).split(",")); |
| Collections.sort(simuDistFilesSize); |
| Assert.assertEquals("Simulated job's file size list has not " + |
| "matched with the Original job's file size list.", |
| origDistFilesSize.size(), |
| simuDistFilesSize.size()); |
| for ( int index = 0; index < origDistFilesSize.size(); index ++ ) { |
| Assert.assertEquals("Simulated job distcache file size has not " + |
| "matched with original job distcache file size.", |
| origDistFilesSize.get(index), simuDistFilesSize.get(index)); |
| } |
| } |
| |
| /** |
| * It verifies the simulated job map counters. |
| * @param counters - Original job map counters. |
| * @param mapJobCounters - Simulated job map counters. |
| * @param jobConf - Simulated job configuration. |
| * @throws ParseException - If an parser error occurs. |
| */ |
| public void verifyJobMapCounters(Counters counters, |
| Map<String,Long> mapCounters, JobConf jobConf) throws ParseException { |
| if (!jobConf.get("gridmix.job.type", "LOADJOB").equals("SLEEPJOB")) { |
| //The below statements have commented due to a bug(MAPREDUCE-2135). |
| /*Assert.assertTrue("Map input bytes have not matched.<exp:[" + |
| convertBytes(mapCounters.get("MAP_INPUT_BYTES").longValue()) +"]><act:[" + |
| convertBytes(getCounterValue(counters,"HDFS_BYTES_READ")) + "]>", |
| convertBytes(mapCounters.get("MAP_INPUT_BYTES").longValue()).equals( |
| convertBytes(getCounterValue(counters,"HDFS_BYTES_READ")))); |
| |
| Assert.assertTrue("Map output bytes has not matched.<exp:[" + |
| convertBytes(mapCounters.get("MAP_OUTPUT_BYTES").longValue()) + "]><act:[" + |
| convertBytes(getCounterValue(counters, "MAP_OUTPUT_BYTES")) + "]>", |
| convertBytes(mapCounters.get("MAP_OUTPUT_BYTES").longValue()).equals( |
| convertBytes(getCounterValue(counters, "MAP_OUTPUT_BYTES"))));*/ |
| |
| Assert.assertEquals("Map input records have not matched.<exp:[" + |
| mapCounters.get("MAP_INPUT_RECS").longValue() + "]><act:[" + |
| getCounterValue(counters, "MAP_INPUT_RECORDS") + "]>", |
| mapCounters.get("MAP_INPUT_RECS").longValue(), |
| getCounterValue(counters, "MAP_INPUT_RECORDS")); |
| |
| // The below statements have commented due to a bug(MAPREDUCE-2154). |
| /*Assert.assertEquals("Map output records have not matched.<exp:[" + |
| mapCounters.get("MAP_OUTPUT_RECS").longValue() + "]><act:[" + |
| getCounterValue(counters, "MAP_OUTPUT_RECORDS") + "]>", |
| mapCounters.get("MAP_OUTPUT_RECS").longValue(), |
| getCounterValue(counters, "MAP_OUTPUT_RECORDS"));*/ |
| } else { |
| Assert.assertTrue("Map Input Bytes are zero", |
| getCounterValue(counters,"HDFS_BYTES_READ") != 0); |
| Assert.assertNotNull("Map Input Records are zero", |
| getCounterValue(counters, "MAP_INPUT_RECORDS")!=0); |
| } |
| } |
| |
| /** |
| * It verifies the simulated job reduce counters. |
| * @param counters - Original job reduce counters. |
| * @param reduceCounters - Simulated job reduce counters. |
| * @param jobConf - simulated job configuration. |
| * @throws ParseException - if an parser error occurs. |
| */ |
| public void verifyJobReduceCounters(Counters counters, |
| Map<String,Long> reduceCounters, JobConf jobConf) throws ParseException { |
| if (!jobConf.get("gridmix.job.type", "LOADJOB").equals("SLEEPJOB")) { |
| /*Assert.assertTrue("Reduce input bytes have not matched.<exp:[" + |
| convertBytes(reduceCounters.get("REDUCE_INPUT_BYTES").longValue()) + "]><act:[" + |
| convertBytes(getCounterValue(counters,"REDUCE_SHUFFLE_BYTES")) + "]>", |
| convertBytes(reduceCounters.get("REDUCE_INPUT_BYTES").longValue()).equals( |
| convertBytes(getCounterValue(counters,"REDUCE_SHUFFLE_BYTES"))));*/ |
| |
| /*Assert.assertEquals("Reduce output bytes have not matched.<exp:[" + |
| convertBytes(reduceCounters.get("REDUCE_OUTPUT_BYTES").longValue()) + "]><act:[" + |
| convertBytes(getCounterValue(counters,"HDFS_BYTES_WRITTEN")) + "]>", |
| convertBytes(reduceCounters.get("REDUCE_OUTPUT_BYTES").longValue()).equals( |
| convertBytes(getCounterValue(counters,"HDFS_BYTES_WRITTEN"))));*/ |
| |
| /*Assert.assertEquals("Reduce output records have not matched.<exp:[" + |
| reduceCounters.get("REDUCE_OUTPUT_RECS").longValue() + "]><act:[" + getCounterValue(counters, |
| "REDUCE_OUTPUT_RECORDS") + "]>", |
| reduceCounters.get("REDUCE_OUTPUT_RECS").longValue(), getCounterValue(counters, |
| "REDUCE_OUTPUT_RECORDS"));*/ |
| |
| /*Assert.assertEquals("Reduce input records have not matched.<exp:[" + |
| reduceCounters.get("REDUCE_INPUT_RECS").longValue() + "]><act:[" + getCounterValue(counters, |
| "REDUCE_INPUT_RECORDS") + "]>", |
| reduceCounters.get("REDUCE_INPUT_RECS").longValue(), |
| getCounterValue(counters,"REDUCE_INPUT_RECORDS"));*/ |
| } else { |
| Assert.assertTrue("Reduce output records are not zero for sleep job.", |
| getCounterValue(counters, "REDUCE_OUTPUT_RECORDS") == 0); |
| Assert.assertTrue("Reduce output bytes are not zero for sleep job.", |
| getCounterValue(counters,"HDFS_BYTES_WRITTEN") == 0); |
| } |
| } |
| |
| /** |
| * It verifies the gridmix simulated job summary. |
| * @param zombieJob - Original job summary. |
| * @param jhInfo - Simulated job history info. |
| * @param jobConf - simulated job configuration. |
| * @throws IOException - if an I/O error occurs. |
| */ |
| public void verifySimulatedJobSummary(ZombieJob zombieJob, |
| JobHistoryParser.JobInfo jhInfo, JobConf jobConf) throws IOException { |
| Assert.assertEquals("Job id has not matched", |
| zombieJob.getJobID(), JobID.forName( |
| jobConf.get("gridmix.job.original-job-id"))); |
| |
| Assert.assertEquals("Job maps have not matched", |
| zombieJob.getNumberMaps(), jhInfo.getTotalMaps()); |
| |
| if (!jobConf.getBoolean("gridmix.sleep.maptask-only",false)) { |
| Assert.assertEquals("Job reducers have not matched", |
| zombieJob.getNumberReduces(), jhInfo.getTotalReduces()); |
| } else { |
| Assert.assertEquals("Job reducers have not matched", |
| 0, jhInfo.getTotalReduces()); |
| } |
| |
| Assert.assertEquals("Job status has not matched.", |
| zombieJob.getOutcome().name(), |
| convertJobStatus(jhInfo.getJobStatus())); |
| |
| LoggedJob loggedJob = zombieJob.getLoggedJob(); |
| Assert.assertEquals("Job priority has not matched.", |
| loggedJob.getPriority().toString(), jhInfo.getPriority()); |
| |
| if (jobConf.get("gridmix.user.resolve.class").contains("RoundRobin")) { |
| Assert.assertTrue(jhInfo.getJobId().toString() + |
| " has not impersonate with other user.", !jhInfo.getUsername() |
| .equals(UserGroupInformation.getLoginUser().getShortUserName())); |
| } |
| } |
| |
| /** |
| * Get the original job map counters from a trace. |
| * @param zombieJob - Original job story. |
| * @return - map counters as a map. |
| */ |
| public Map<String, Long> getJobMapCounters(ZombieJob zombieJob) { |
| long expMapInputBytes = 0; |
| long expMapOutputBytes = 0; |
| long expMapInputRecs = 0; |
| long expMapOutputRecs = 0; |
| Map<String,Long> mapCounters = new HashMap<String,Long>(); |
| for (int index = 0; index < zombieJob.getNumberMaps(); index ++) { |
| TaskInfo mapTask = zombieJob.getTaskInfo(TaskType.MAP, index); |
| expMapInputBytes += mapTask.getInputBytes(); |
| expMapOutputBytes += mapTask.getOutputBytes(); |
| expMapInputRecs += mapTask.getInputRecords(); |
| expMapOutputRecs += mapTask.getOutputRecords(); |
| } |
| mapCounters.put("MAP_INPUT_BYTES", expMapInputBytes); |
| mapCounters.put("MAP_OUTPUT_BYTES", expMapOutputBytes); |
| mapCounters.put("MAP_INPUT_RECS", expMapInputRecs); |
| mapCounters.put("MAP_OUTPUT_RECS", expMapOutputRecs); |
| return mapCounters; |
| } |
| |
| /** |
| * Get the original job reduce counters from a trace. |
| * @param zombieJob - Original job story. |
| * @return - reduce counters as a map. |
| */ |
| public Map<String,Long> getJobReduceCounters(ZombieJob zombieJob) { |
| long expReduceInputBytes = 0; |
| long expReduceOutputBytes = 0; |
| long expReduceInputRecs = 0; |
| long expReduceOutputRecs = 0; |
| Map<String,Long> reduceCounters = new HashMap<String,Long>(); |
| for (int index = 0; index < zombieJob.getNumberReduces(); index ++) { |
| TaskInfo reduceTask = zombieJob.getTaskInfo(TaskType.REDUCE, index); |
| expReduceInputBytes += reduceTask.getInputBytes(); |
| expReduceOutputBytes += reduceTask.getOutputBytes(); |
| expReduceInputRecs += reduceTask.getInputRecords(); |
| expReduceOutputRecs += reduceTask.getOutputRecords(); |
| } |
| reduceCounters.put("REDUCE_INPUT_BYTES", expReduceInputBytes); |
| reduceCounters.put("REDUCE_OUTPUT_BYTES", expReduceOutputBytes); |
| reduceCounters.put("REDUCE_INPUT_RECS", expReduceInputRecs); |
| reduceCounters.put("REDUCE_OUTPUT_RECS", expReduceOutputRecs); |
| return reduceCounters; |
| } |
| |
| /** |
| * Get the simulated job configuration of a job. |
| * @param simulatedJobID - Simulated job id. |
| * @param tmpJHFolder - temporary job history folder location. |
| * @return - simulated job configuration. |
| * @throws IOException - If an I/O error occurs. |
| */ |
| public JobConf getSimulatedJobConf(JobID simulatedJobID, File tmpJHFolder) |
| throws IOException{ |
| FileSystem fs = null; |
| try { |
| String historyFilePath = jtClient.getProxy(). |
| getJobHistoryLocationForRetiredJob(simulatedJobID); |
| Path jhpath = new Path(historyFilePath); |
| fs = jhpath.getFileSystem(conf); |
| fs.copyToLocalFile(jhpath,new Path(tmpJHFolder.toString())); |
| fs.copyToLocalFile(new Path(historyFilePath + "_conf.xml"), |
| new Path(tmpJHFolder.toString())); |
| JobConf jobConf = new JobConf(); |
| jobConf.addResource(new Path(tmpJHFolder.toString() + "/" + |
| simulatedJobID + "_conf.xml")); |
| jobConf.reloadConfiguration(); |
| return jobConf; |
| }finally { |
| fs.close(); |
| } |
| } |
| |
| /** |
| * Get the simulated job history of a job. |
| * @param simulatedJobID - simulated job id. |
| * @return - simulated job information. |
| * @throws IOException - if an I/O error occurs. |
| */ |
| public JobHistoryParser.JobInfo getSimulatedJobHistory(JobID simulatedJobID) |
| throws IOException { |
| FileSystem fs = null; |
| try { |
| String historyFilePath = jtClient.getProxy(). |
| getJobHistoryLocationForRetiredJob(simulatedJobID); |
| Path jhpath = new Path(historyFilePath); |
| fs = jhpath.getFileSystem(conf); |
| JobHistoryParser jhparser = new JobHistoryParser(fs, jhpath); |
| JobHistoryParser.JobInfo jhInfo = jhparser.parse(); |
| return jhInfo; |
| } finally { |
| fs.close(); |
| } |
| } |
| |
| private String convertJobStatus(String jobStatus) { |
| if (jobStatus.equals("SUCCEEDED")) { |
| return "SUCCESS"; |
| } else { |
| return jobStatus; |
| } |
| } |
| |
| private String convertBytes(long bytesValue) { |
| int units = 1024; |
| if( bytesValue < units ) { |
| return String.valueOf(bytesValue)+ "B"; |
| } else { |
| int exp = (int)(Math.log(bytesValue) / Math.log(units)); |
| return String.format("%1d%sB",(long)(bytesValue / Math.pow(units, exp)), |
| "KMGTPE".charAt(exp -1)); |
| } |
| } |
| |
| |
| private long getCounterValue(Counters counters,String key) |
| throws ParseException { |
| for (String groupName : counters.getGroupNames()) { |
| CounterGroup totalGroup = counters.getGroup(groupName); |
| Iterator<Counter> itrCounter = totalGroup.iterator(); |
| while (itrCounter.hasNext()) { |
| Counter counter = itrCounter.next(); |
| if (counter.getName().equals(key)) { |
| return counter.getValue(); |
| } |
| } |
| } |
| return 0; |
| } |
| } |
