| /** |
| * 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 org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.mapreduce.Job; |
| import org.apache.hadoop.mapreduce.MRConfig; |
| import org.apache.hadoop.mapreduce.SleepJob; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.permission.FsPermission; |
| import org.apache.hadoop.mapreduce.JobID; |
| import org.apache.hadoop.mapreduce.server.tasktracker.TTConfig; |
| import org.apache.hadoop.mapreduce.test.system.FinishTaskControlAction; |
| import org.apache.hadoop.mapreduce.test.system.JTProtocol; |
| import org.apache.hadoop.mapreduce.test.system.JobInfo; |
| import org.apache.hadoop.mapreduce.test.system.MRCluster; |
| import org.apache.hadoop.mapreduce.test.system.TTClient; |
| import org.apache.hadoop.mapreduce.test.system.TTInfo; |
| import org.apache.hadoop.mapreduce.test.system.TTTaskInfo; |
| import org.apache.hadoop.mapreduce.test.system.TaskInfo; |
| import org.junit.AfterClass; |
| import org.junit.Assert; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| |
| public class TestFileOwner { |
| public static MRCluster cluster; |
| |
| private StringBuffer jobIdDir = new StringBuffer(); |
| private JTProtocol wovenClient = null; |
| private static final Log LOG = LogFactory.getLog(TestFileOwner.class); |
| private String taskController = null; |
| private final FsPermission PERM_777 = new FsPermission("777"); |
| private final FsPermission PERM_755 = new FsPermission("755"); |
| private final FsPermission PERM_644 = new FsPermission("644"); |
| |
| @BeforeClass |
| public static void setUp() throws java.lang.Exception { |
| cluster = MRCluster.createCluster(new Configuration()); |
| cluster.setUp(); |
| } |
| |
| /* |
| * The test is used to check the file permission of local files in |
| * mapred.local.dir. The job control is used which will make the tasks wait |
| * for completion until it is signaled |
| * |
| * @throws Exception in case of test errors |
| */ |
| @Test |
| public void testFilePermission() throws Exception { |
| wovenClient = cluster.getJTClient().getProxy(); |
| Configuration conf = new Configuration(cluster.getConf()); |
| FinishTaskControlAction.configureControlActionForJob(conf); |
| SleepJob job = new SleepJob(); |
| job.setConf(conf); |
| Job slpJob = job.createJob(1, 0, 100, 100, 100, 100); |
| JobConf jconf = new JobConf(conf); |
| slpJob.submit(); |
| RunningJob rJob = |
| cluster.getJTClient().getClient().getJob( |
| org.apache.hadoop.mapred.JobID.downgrade(slpJob.getJobID())); |
| taskController = conf.get(TTConfig.TT_TASK_CONTROLLER); |
| // get the job info so we can get the env variables from the daemon. |
| // Now wait for the task to be in the running state, only then the |
| // directories will be created |
| JobInfo info = wovenClient.getJobInfo(rJob.getID()); |
| Assert.assertNotNull("JobInfo is null", info); |
| JobID id = rJob.getID(); |
| while (info.runningMaps() != 1) { |
| Thread.sleep(1000); |
| info = wovenClient.getJobInfo(id); |
| } |
| TaskInfo[] myTaskInfos = wovenClient.getTaskInfo(id); |
| for (TaskInfo tInfo : myTaskInfos) { |
| if (!tInfo.isSetupOrCleanup()) { |
| String[] taskTrackers = tInfo.getTaskTrackers(); |
| for (String taskTracker : taskTrackers) { |
| TTInfo ttInfo = wovenClient.getTTInfo(taskTracker); |
| TTClient ttCli = cluster.getTTClient(ttInfo.getStatus().getHost()); |
| Assert.assertNotNull("TTClient instance is null", ttCli); |
| TTTaskInfo ttTaskInfo = ttCli.getProxy().getTask(tInfo.getTaskID()); |
| Assert.assertNotNull("TTTaskInfo is null", ttTaskInfo); |
| while (ttTaskInfo.getTaskStatus().getRunState() != TaskStatus.State.RUNNING) { |
| Thread.sleep(100); |
| ttTaskInfo = ttCli.getProxy().getTask(tInfo.getTaskID()); |
| } |
| testPermissionWithTaskController(ttCli, conf, info); |
| FinishTaskControlAction action = |
| new FinishTaskControlAction(TaskID.downgrade(tInfo.getTaskID())); |
| for (TTClient cli : cluster.getTTClients()) { |
| cli.getProxy().sendAction(action); |
| } |
| } |
| } |
| } |
| JobInfo jInfo = wovenClient.getJobInfo(id); |
| jInfo = cluster.getJTClient().getProxy().getJobInfo(id); |
| while (!jInfo.getStatus().isJobComplete()) { |
| Thread.sleep(100); |
| jInfo = cluster.getJTClient().getProxy().getJobInfo(id); |
| } |
| } |
| |
| private void testPermissionWithTaskController( |
| TTClient tClient, Configuration conf, JobInfo info) { |
| Assert.assertNotNull("TTclient is null", tClient); |
| FsPermission fsPerm = null; |
| String[] pathInfo = conf.getStrings(MRConfig.LOCAL_DIR); |
| for (int i = 0; i < pathInfo.length; i++) { |
| // First verify the jobid directory exists |
| jobIdDir = new StringBuffer(); |
| String userName = null; |
| try { |
| JobStatus[] jobStatus = cluster.getJTClient().getClient().getAllJobs(); |
| userName = jobStatus[0].getUsername(); |
| } catch (Exception ex) { |
| LOG.error("Failed to get user name"); |
| boolean status = false; |
| Assert.assertTrue("Failed to get the userName", status); |
| } |
| jobIdDir.append(pathInfo[i]).append(Path.SEPARATOR); |
| jobIdDir.append(TaskTracker.getLocalJobDir(userName, info |
| .getID().toString())); |
| FileStatus[] fs = null; |
| try { |
| fs = tClient.listStatus(jobIdDir.toString(), true); |
| } catch (Exception ex) { |
| LOG.error("Failed to get the jobIdDir files " + ex); |
| } |
| Assert.assertEquals("Filestatus length is zero", fs.length != 0, true); |
| for (FileStatus file : fs) { |
| try { |
| String filename = file.getPath().getName(); |
| if (filename.equals(TaskTracker.JOBFILE)) { |
| if (taskController == DefaultTaskController.class.getName()) { |
| fsPerm = file.getPermission(); |
| Assert.assertTrue("FilePermission failed for " + filename, fsPerm |
| .equals(PERM_777)); |
| } |
| } |
| if (filename.startsWith("attempt")) { |
| StringBuffer attemptDir = new StringBuffer(jobIdDir); |
| attemptDir.append(Path.SEPARATOR).append(filename); |
| if (tClient.getFileStatus(attemptDir.toString(), true) != null) { |
| FileStatus[] attemptFs = |
| tClient.listStatus(attemptDir.toString(), true, true); |
| for (FileStatus attemptfz : attemptFs) { |
| Assert.assertNotNull("FileStatus is null", attemptfz); |
| fsPerm = attemptfz.getPermission(); |
| Assert.assertNotNull("FsPermission is null", fsPerm); |
| if (taskController == DefaultTaskController.class.getName()) { |
| if (!attemptfz.isDir()) { |
| Assert.assertTrue( |
| "FilePermission failed for " + filename, fsPerm |
| .equals(PERM_777)); |
| } else { |
| Assert.assertTrue( |
| "FilePermission failed for " + filename, fsPerm |
| .equals(PERM_755)); |
| } |
| } |
| } |
| } |
| } |
| if (filename.equals(TaskTracker.TASKJARDIR)) { |
| StringBuffer jarsDir = new StringBuffer(jobIdDir); |
| jarsDir.append(Path.SEPARATOR).append(filename); |
| FileStatus[] jarsFs = |
| tClient.listStatus(jarsDir.toString(), true, true); |
| for (FileStatus jarsfz : jarsFs) { |
| Assert.assertNotNull("FileStatus is null", jarsfz); |
| fsPerm = jarsfz.getPermission(); |
| Assert.assertNotNull("File permission is null", fsPerm); |
| if (taskController == DefaultTaskController.class.getName()) { |
| if (!jarsfz.isDir()) { |
| if (jarsfz.getPath().getName().equals("job.jar")) { |
| Assert.assertTrue( |
| "FilePermission failed for " + filename, fsPerm |
| .equals(PERM_777)); |
| } else { |
| Assert.assertTrue( |
| "FilePermission failed for " + filename, fsPerm |
| .equals(PERM_644)); |
| } |
| } else { |
| Assert.assertTrue( |
| "FilePermission failed for " + filename, fsPerm |
| .equals(PERM_755)); |
| } |
| } |
| } |
| } |
| } catch (Exception ex) { |
| LOG.error("The exception occurred while searching for nonexsistent" |
| + "file, ignoring and continuing. " + ex); |
| } |
| }// for loop ends |
| }// for loop ends |
| } |
| |
| @AfterClass |
| public static void tearDown() throws java.lang.Exception { |
| cluster.tearDown(); |
| } |
| } |
