| /** |
| * 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; |
| |
| import java.io.File; |
| import java.io.IOException; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.FileUtil; |
| import org.apache.hadoop.mapred.UtilsForTests.FakeClock; |
| import org.apache.hadoop.mapreduce.MRConfig; |
| import org.apache.hadoop.mapreduce.MRJobConfig; |
| import org.apache.hadoop.mapreduce.server.tasktracker.Localizer; |
| import org.apache.hadoop.mapreduce.util.MRAsyncDiskService; |
| |
| import static org.junit.Assert.*; |
| |
| import org.junit.After; |
| import org.junit.Test; |
| |
| public class TestUserLogCleanup { |
| private static String jtid = "test"; |
| private static long ONE_HOUR = 1000 * 60 * 60; |
| private Localizer localizer; |
| private UserLogCleaner taskLogCleanupThread; |
| private TaskTracker tt; |
| private FakeClock myClock = new FakeClock(); |
| private JobID jobid1 = new JobID(jtid, 1); |
| private JobID jobid2 = new JobID(jtid, 2); |
| private JobID jobid3 = new JobID(jtid, 3); |
| private JobID jobid4 = new JobID(jtid, 4); |
| private File foo = new File(TaskLog.getUserLogDir(), "foo"); |
| private File bar = new File(TaskLog.getUserLogDir(), "bar"); |
| |
| public TestUserLogCleanup() throws IOException { |
| Configuration conf = new Configuration(); |
| localizer = new Localizer(FileSystem.get(conf), conf |
| .getStrings(MRConfig.LOCAL_DIR), new DefaultTaskController()); |
| taskLogCleanupThread = new UserLogCleaner(conf); |
| taskLogCleanupThread.setClock(myClock); |
| tt = new TaskTracker(); |
| tt.setConf(new JobConf(conf)); |
| tt.setLocalizer(localizer); |
| tt.setTaskLogCleanupThread(taskLogCleanupThread); |
| } |
| |
| @After |
| public void tearDown() throws IOException { |
| FileUtil.fullyDelete(TaskLog.getUserLogDir()); |
| } |
| |
| private File localizeJob(JobID jobid) throws IOException { |
| File jobUserlog = TaskLog.getJobDir(jobid); |
| |
| JobConf conf = new JobConf(); |
| // localize job log directory |
| tt.initializeJobLogDir(jobid, conf); |
| assertTrue(jobUserlog + " directory is not created.", jobUserlog.exists()); |
| return jobUserlog; |
| } |
| |
| private void jobFinished(JobID jobid, int logRetainHours) { |
| Configuration jobconf = new Configuration(); |
| jobconf.setInt(MRJobConfig.USER_LOG_RETAIN_HOURS, logRetainHours); |
| taskLogCleanupThread.markJobLogsForDeletion(myClock.getTime(), jobconf, |
| jobid); |
| } |
| |
| /** |
| * Tests job user-log directory deletion. |
| * |
| * Adds two jobs for log deletion. One with one hour retain hours, other with |
| * two retain hours. After an hour, |
| * TaskLogCleanupThread.processCompletedJobs() call, |
| * makes sure job with 1hr retain hours is removed and other is retained. |
| * After one more hour, job with 2hr retain hours is also removed. |
| * |
| * @throws IOException |
| */ |
| @Test |
| public void testJobLogCleanup() throws IOException { |
| File jobUserlog1 = localizeJob(jobid1); |
| File jobUserlog2 = localizeJob(jobid2); |
| |
| // add job user log directory for deletion, with 2 hours for deletion |
| jobFinished(jobid1, 2); |
| |
| // add the job for deletion with one hour as retain hours |
| jobFinished(jobid2, 1); |
| |
| // remove old logs and see jobid1 is not removed and jobid2 is removed |
| myClock.advance(ONE_HOUR); |
| taskLogCleanupThread.processCompletedJobs(); |
| assertTrue(jobUserlog1 + " got deleted", jobUserlog1.exists()); |
| assertFalse(jobUserlog2 + " still exists.", jobUserlog2.exists()); |
| |
| myClock.advance(ONE_HOUR); |
| // remove old logs and see jobid1 is removed now |
| taskLogCleanupThread.processCompletedJobs(); |
| assertFalse(jobUserlog1 + " still exists.", jobUserlog1.exists()); |
| } |
| |
| /** |
| * Tests user-log directory cleanup on a TT re-init with 3 hours as log |
| * retain hours for tracker. |
| * |
| * Adds job1 deletion before the re-init with 2 hour retain hours. |
| * Adds job2 for which there are no tasks/killJobAction after the re-init. |
| * Adds job3 for which there is localizeJob followed by killJobAction |
| * with 3 hours as retain hours. |
| * Adds job4 for which there are some tasks after the re-init. |
| * |
| * @throws IOException |
| */ |
| @Test |
| public void testUserLogCleanup() throws IOException { |
| File jobUserlog1 = localizeJob(jobid1); |
| File jobUserlog2 = localizeJob(jobid2); |
| File jobUserlog3 = localizeJob(jobid3); |
| File jobUserlog4 = localizeJob(jobid4); |
| // create a some files/dirs in userlog |
| foo.mkdirs(); |
| bar.createNewFile(); |
| |
| // add the jobid1 for deletion with retainhours = 2 |
| jobFinished(jobid1, 2); |
| |
| // time is now 1. |
| myClock.advance(ONE_HOUR); |
| |
| // mimic TaskTracker reinit |
| // clear userlog directory |
| // job directories will be added with 3 hours as retain hours. They will be |
| // deleted at time 4. |
| Configuration conf = new Configuration(); |
| conf.setInt(MRJobConfig.USER_LOG_RETAIN_HOURS, 3); |
| taskLogCleanupThread.clearOldUserLogs(conf); |
| assertFalse(foo.exists()); |
| assertFalse(bar.exists()); |
| assertTrue(jobUserlog1.exists()); |
| assertTrue(jobUserlog2.exists()); |
| assertTrue(jobUserlog3.exists()); |
| assertTrue(jobUserlog4.exists()); |
| assertTrue(new File(TaskLog.getUserLogDir(), MRAsyncDiskService.TOBEDELETED) |
| .exists()); |
| |
| myClock.advance(ONE_HOUR); |
| // time is now 2. |
| taskLogCleanupThread.processCompletedJobs(); |
| assertFalse(jobUserlog1.exists()); |
| assertTrue(jobUserlog2.exists()); |
| assertTrue(jobUserlog3.exists()); |
| assertTrue(jobUserlog4.exists()); |
| |
| // mimic localizeJob followed KillJobAction for jobid3 |
| // add the job for deletion with retainhours = 3. |
| // jobid3 should be deleted at time 5. |
| jobUserlog3 = localizeJob(jobid3); |
| jobFinished(jobid3, 3); |
| |
| // mimic localizeJob for jobid4 |
| jobUserlog4 = localizeJob(jobid4); |
| |
| // do cleanup |
| myClock.advance(2 * ONE_HOUR); |
| // time is now 4. |
| taskLogCleanupThread.processCompletedJobs(); |
| |
| // jobid2 will be deleted |
| assertFalse(jobUserlog1.exists()); |
| assertFalse(jobUserlog2.exists()); |
| assertTrue(jobUserlog3.exists()); |
| assertTrue(jobUserlog4.exists()); |
| |
| myClock.advance(ONE_HOUR); |
| // time is now 5. |
| // do cleanup again |
| taskLogCleanupThread.processCompletedJobs(); |
| |
| // jobid3 will be deleted |
| assertFalse(jobUserlog1.exists()); |
| assertFalse(jobUserlog2.exists()); |
| assertFalse(jobUserlog3.exists()); |
| assertTrue(jobUserlog4.exists()); |
| } |
| |
| /** |
| * Tests user-log directory cleanup on a TT restart. |
| * |
| * Adds job1 deletion before the restart with 2 hour retain hours. |
| * Adds job2 for which there are no tasks/killJobAction after the restart. |
| * Adds job3 for which there is localizeJob followed by killJobAction after |
| * the restart with 3 hours retain hours. |
| * Adds job4 for which there are some tasks after the restart. |
| * |
| * @throws IOException |
| */ |
| @Test |
| public void testUserLogCleanupAfterRestart() throws IOException { |
| File jobUserlog1 = localizeJob(jobid1); |
| File jobUserlog2 = localizeJob(jobid2); |
| File jobUserlog3 = localizeJob(jobid3); |
| File jobUserlog4 = localizeJob(jobid4); |
| // create a some files/dirs in userlog |
| foo.mkdirs(); |
| bar.createNewFile(); |
| |
| // add the jobid1 for deletion with retainhours = 2 |
| jobFinished(jobid1, 2); |
| |
| // time is now 1. |
| myClock.advance(ONE_HOUR); |
| |
| // mimic TaskTracker restart |
| // clear userlog directory |
| // job directories will be added with 3 hours as retain hours. |
| Configuration conf = new Configuration(); |
| conf.setInt(MRJobConfig.USER_LOG_RETAIN_HOURS, 3); |
| taskLogCleanupThread = new UserLogCleaner(conf); |
| myClock = new FakeClock(); // clock is reset. |
| taskLogCleanupThread.setClock(myClock); |
| taskLogCleanupThread.clearOldUserLogs(conf); |
| tt.setTaskLogCleanupThread(taskLogCleanupThread); |
| assertFalse(foo.exists()); |
| assertFalse(bar.exists()); |
| assertTrue(jobUserlog1.exists()); |
| assertTrue(jobUserlog2.exists()); |
| assertTrue(jobUserlog3.exists()); |
| assertTrue(jobUserlog4.exists()); |
| assertTrue(new File(TaskLog.getUserLogDir(), MRAsyncDiskService.TOBEDELETED) |
| .exists()); |
| |
| myClock.advance(ONE_HOUR); |
| // time is now 1. |
| taskLogCleanupThread.processCompletedJobs(); |
| assertTrue(jobUserlog1.exists()); |
| assertTrue(jobUserlog2.exists()); |
| assertTrue(jobUserlog3.exists()); |
| assertTrue(jobUserlog4.exists()); |
| |
| // mimic localizeJob followed KillJobAction for jobid3 |
| // add the job for deletion with retainhours = 3. |
| // jobid3 should be deleted at time 4. |
| jobUserlog3 = localizeJob(jobid3); |
| jobFinished(jobid3, 3); |
| |
| // mimic localizeJob for jobid4 |
| jobUserlog4 = localizeJob(jobid4); |
| |
| // do cleanup |
| myClock.advance(2 * ONE_HOUR); |
| // time is now 3. |
| taskLogCleanupThread.processCompletedJobs(); |
| |
| // jobid1 and jobid2 will be deleted |
| assertFalse(jobUserlog1.exists()); |
| assertFalse(jobUserlog2.exists()); |
| assertTrue(jobUserlog3.exists()); |
| assertTrue(jobUserlog4.exists()); |
| |
| myClock.advance(ONE_HOUR); |
| // time is now 4. |
| // do cleanup again |
| taskLogCleanupThread.processCompletedJobs(); |
| |
| // jobid3 will be deleted |
| assertFalse(jobUserlog1.exists()); |
| assertFalse(jobUserlog2.exists()); |
| assertFalse(jobUserlog3.exists()); |
| assertTrue(jobUserlog4.exists()); |
| } |
| } |