src/test/system/test/org/apache/hadoop/mapred/TestDistributedCachePrivateFile.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;

 import java.io.DataOutputStream;
 import java.net.URI;
 import java.util.Collection;
 import org.apache.commons.logging.LogFactory;
 import org.apache.commons.logging.Log;
 import org.apache.hadoop.mapreduce.test.system.JTProtocol;
 import org.apache.hadoop.mapreduce.test.system.TTClient;
 import org.apache.hadoop.mapreduce.test.system.JobInfo;
 import org.apache.hadoop.mapreduce.test.system.TaskInfo;
 import org.apache.hadoop.mapreduce.test.system.MRCluster;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.UtilsForTests;

 import org.apache.hadoop.mapreduce.test.system.FinishTaskControlAction;
 import org.apache.hadoop.filecache.DistributedCache;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.SleepJob;

 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.AfterClass;
 import org.junit.Test;

 /**
  * Verify the Distributed Cache functionality.
  * This test scenario is for a distributed cache file behaviour
  * when the file is private. Once a job uses a distributed
  * cache file with private permissions that file is stored in the
  * mapred.local.dir, under the directory which has the same name
  * as job submitter's username. The directory has 700 permission
  * and the file under it, should have 777 permissions.
 */

 public class TestDistributedCachePrivateFile {

   private static MRCluster cluster = null;
   private static FileSystem dfs = null;
   private static JobClient client = null;
   private static FsPermission permission = new FsPermission((short)00770);

   private static String uriPath = "hdfs:///tmp/test.txt";
   private static final Path URIPATH = new Path(uriPath);
   private String distributedFileName = "test.txt";

   static final Log LOG = LogFactory.
                            getLog(TestDistributedCachePrivateFile.class);

   public TestDistributedCachePrivateFile() throws Exception {
   }

   @BeforeClass
   public static void setUp() throws Exception {
     cluster = MRCluster.createCluster(new Configuration());
     cluster.setUp();
     client = cluster.getJTClient().getClient();
     dfs = client.getFs();
     //Deleting the file if it already exists
     dfs.delete(URIPATH, true);

     Collection<TTClient> tts = cluster.getTTClients();
     //Stopping all TTs
     for (TTClient tt : tts) {
       tt.kill();
     }
     //Starting all TTs
     for (TTClient tt : tts) {
       tt.start();
     }

     String input = "This will be the content of\n" + "distributed cache\n";
     //Creating the path with the file
     DataOutputStream file =
         UtilsForTests.createTmpFileDFS(dfs, URIPATH, permission, input);
   }

   @AfterClass
   public static void tearDown() throws Exception {
     cluster.tearDown();
     dfs.delete(URIPATH, true);

     Collection<TTClient> tts = cluster.getTTClients();
     //Stopping all TTs
     for (TTClient tt : tts) {
       tt.kill();
     }
     //Starting all TTs
     for (TTClient tt : tts) {
       tt.start();
     }
   }

   @Test
   /**
    * This tests Distributed Cache for private file
    * @param none
    * @return void
    */
   public void testDistributedCache() throws Exception {
     Configuration conf = new Configuration(cluster.getConf());
     JTProtocol wovenClient = cluster.getJTClient().getProxy();

     //This counter will check for count of a loop,
     //which might become infinite.
     int count = 0;

     SleepJob job = new SleepJob();
     job.setConf(conf);
     Job slpJob = job.createJob(5, 1, 1000, 1000, 100, 100);

     DistributedCache.createSymlink(conf);
     URI uri = URI.create(uriPath);
     DistributedCache.addCacheFile(uri, conf);
     JobConf jconf = new JobConf(conf);

     //Controls the job till all verification is done
     FinishTaskControlAction.configureControlActionForJob(conf);

     //Submitting the job
     slpJob.submit();
     RunningJob rJob =
         cluster.getJTClient().getClient().getJob(org.apache.hadoop.mapred.JobID.downgrade(slpJob.getJobID()));

     JobStatus[] jobStatus = client.getAllJobs();
     String userName = jobStatus[0].getUsername();

     TTClient tClient = null;
     JobInfo jInfo = wovenClient.getJobInfo(rJob.getID());
     LOG.info("jInfo is :" + jInfo);

     //Assert if jobInfo is null
     Assert.assertNotNull("jobInfo is null", jInfo);

     //Wait for the job to start running.
     count = 0;
     while (jInfo.getStatus().getRunState() != JobStatus.RUNNING) {
       UtilsForTests.waitFor(10000);
       count++;
       jInfo = wovenClient.getJobInfo(rJob.getID());
       //If the count goes beyond a point, then Assert; This is to avoid
       //infinite loop under unforeseen circumstances.
       if (count > 10) {
         Assert.fail("job has not reached running state for more than" +
             "100 seconds. Failing at this point");
       }
     }

     LOG.info("job id is :" + rJob.getID().toString());

     TaskInfo[] taskInfos = cluster.getJTClient().getProxy()
            .getTaskInfo(rJob.getID());

     boolean distCacheFileIsFound;

     for (TaskInfo taskInfo : taskInfos) {
       distCacheFileIsFound = false;
       String[] taskTrackers = taskInfo.getTaskTrackers();

       for(String taskTracker : taskTrackers) {
         //Getting the exact FQDN of the tasktracker from
         //the tasktracker string.
         taskTracker = UtilsForTests.getFQDNofTT(taskTracker);
         tClient =  cluster.getTTClient(taskTracker);
         String[] localDirs = tClient.getMapredLocalDirs();
         int distributedFileCount = 0;
         String localDirOnly = null;

         boolean FileNotPresentForThisDirectoryPath = false;

         //Go to every single path
         for (String localDir : localDirs) {
           FileNotPresentForThisDirectoryPath = false;
           localDirOnly = localDir;

           //Public Distributed cache will always be stored under
           //mapred.local.dir/tasktracker/archive
           localDirOnly = localDir + Path.SEPARATOR + TaskTracker.SUBDIR +
               Path.SEPARATOR +  userName;

           //Private Distributed cache will always be stored under
           //mapre.local.dir/taskTracker/<username>/distcache
           //Checking for username directory to check if it has the
           //proper permissions
           localDir = localDir + Path.SEPARATOR +
                   TaskTracker.getPrivateDistributedCacheDir(userName);

           FileStatus fileStatusMapredLocalDirUserName = null;

           try {
             fileStatusMapredLocalDirUserName = tClient.
                             getFileStatus(localDirOnly, true);
           } catch (Exception e) {
             LOG.info("LocalDirOnly :" + localDirOnly + " not found");
             FileNotPresentForThisDirectoryPath = true;
           }

           //File will only be stored under one of the mapred.lcoal.dir
           //If other paths were hit, just continue
           if (FileNotPresentForThisDirectoryPath)
             continue;

           Path pathMapredLocalDirUserName =
               fileStatusMapredLocalDirUserName.getPath();
           FsPermission fsPermMapredLocalDirUserName =
               fileStatusMapredLocalDirUserName.getPermission();
           Assert.assertTrue("Directory Permission is not 700",
             fsPermMapredLocalDirUserName.equals(new FsPermission("700")));

           //Get file status of all the directories
           //and files under that path.
           FileStatus[] fileStatuses = tClient.listStatus(localDir,
               true, true);
           for (FileStatus  fileStatus : fileStatuses) {
             Path path = fileStatus.getPath();
             LOG.info("path is :" + path.toString());
             //Checking if the received path ends with
             //the distributed filename
             distCacheFileIsFound = (path.toString()).
                 endsWith(distributedFileName);
             //If file is found, check for its permission.
             //Since the file is found break out of loop
             if (distCacheFileIsFound){
               LOG.info("PATH found is :" + path.toString());
               distributedFileCount++;
               String filename = path.getName();
               FsPermission fsPerm = fileStatus.getPermission();
               Assert.assertTrue("File Permission is not 777",
                 fsPerm.equals(new FsPermission("777")));
             }
           }
         }

         LOG.info("Distributed File count is :" + distributedFileCount);

         if (distributedFileCount > 1) {
           Assert.fail("The distributed cache file is more than one");
         } else if (distributedFileCount < 1)
           Assert.fail("The distributed cache file is less than one");
         if (!distCacheFileIsFound) {
           Assert.assertEquals("The distributed cache file does not exist",
               distCacheFileIsFound, false);
         }
       }

       //Allow the job to continue through MR control job.
       for (TaskInfo taskInfoRemaining : taskInfos) {
         FinishTaskControlAction action = new FinishTaskControlAction(TaskID
            .downgrade(taskInfoRemaining.getTaskID()));
         Collection<TTClient> tts = cluster.getTTClients();
         for (TTClient cli : tts) {
           cli.getProxy().sendAction(action);
         }
       }

       //Killing the job because all the verification needed
       //for this testcase is completed.
       rJob.killJob();
     }
   }
 }
	/**
	* 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.DataOutputStream;
	import java.net.URI;
	import java.util.Collection;
	import org.apache.commons.logging.LogFactory;
	import org.apache.commons.logging.Log;
	import org.apache.hadoop.mapreduce.test.system.JTProtocol;
	import org.apache.hadoop.mapreduce.test.system.TTClient;
	import org.apache.hadoop.mapreduce.test.system.JobInfo;
	import org.apache.hadoop.mapreduce.test.system.TaskInfo;
	import org.apache.hadoop.mapreduce.test.system.MRCluster;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.mapred.UtilsForTests;

	import org.apache.hadoop.mapreduce.test.system.FinishTaskControlAction;
	import org.apache.hadoop.filecache.DistributedCache;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.permission.FsPermission;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.mapreduce.Job;
	import org.apache.hadoop.mapreduce.SleepJob;

	import org.junit.Assert;
	import org.junit.BeforeClass;
	import org.junit.AfterClass;
	import org.junit.Test;

	/**
	* Verify the Distributed Cache functionality.
	* This test scenario is for a distributed cache file behaviour
	* when the file is private. Once a job uses a distributed
	* cache file with private permissions that file is stored in the
	* mapred.local.dir, under the directory which has the same name
	* as job submitter's username. The directory has 700 permission
	* and the file under it, should have 777 permissions.
	*/

	public class TestDistributedCachePrivateFile {

	private static MRCluster cluster = null;
	private static FileSystem dfs = null;
	private static JobClient client = null;
	private static FsPermission permission = new FsPermission((short)00770);

	private static String uriPath = "hdfs:///tmp/test.txt";
	private static final Path URIPATH = new Path(uriPath);
	private String distributedFileName = "test.txt";

	static final Log LOG = LogFactory.
	getLog(TestDistributedCachePrivateFile.class);

	public TestDistributedCachePrivateFile() throws Exception {
	}

	@BeforeClass
	public static void setUp() throws Exception {
	cluster = MRCluster.createCluster(new Configuration());
	cluster.setUp();
	client = cluster.getJTClient().getClient();
	dfs = client.getFs();
	//Deleting the file if it already exists
	dfs.delete(URIPATH, true);

	Collection<TTClient> tts = cluster.getTTClients();
	//Stopping all TTs
	for (TTClient tt : tts) {
	tt.kill();
	}
	//Starting all TTs
	for (TTClient tt : tts) {
	tt.start();
	}

	String input = "This will be the content of\n" + "distributed cache\n";
	//Creating the path with the file
	DataOutputStream file =
	UtilsForTests.createTmpFileDFS(dfs, URIPATH, permission, input);
	}

	@AfterClass
	public static void tearDown() throws Exception {
	cluster.tearDown();
	dfs.delete(URIPATH, true);

	Collection<TTClient> tts = cluster.getTTClients();
	//Stopping all TTs
	for (TTClient tt : tts) {
	tt.kill();
	}
	//Starting all TTs
	for (TTClient tt : tts) {
	tt.start();
	}
	}

	@Test
	/**
	* This tests Distributed Cache for private file
	* @param none
	* @return void
	*/
	public void testDistributedCache() throws Exception {
	Configuration conf = new Configuration(cluster.getConf());
	JTProtocol wovenClient = cluster.getJTClient().getProxy();

	//This counter will check for count of a loop,
	//which might become infinite.
	int count = 0;

	SleepJob job = new SleepJob();
	job.setConf(conf);
	Job slpJob = job.createJob(5, 1, 1000, 1000, 100, 100);

	DistributedCache.createSymlink(conf);
	URI uri = URI.create(uriPath);
	DistributedCache.addCacheFile(uri, conf);
	JobConf jconf = new JobConf(conf);

	//Controls the job till all verification is done
	FinishTaskControlAction.configureControlActionForJob(conf);

	//Submitting the job
	slpJob.submit();
	RunningJob rJob =
	cluster.getJTClient().getClient().getJob(org.apache.hadoop.mapred.JobID.downgrade(slpJob.getJobID()));

	JobStatus[] jobStatus = client.getAllJobs();
	String userName = jobStatus[0].getUsername();

	TTClient tClient = null;
	JobInfo jInfo = wovenClient.getJobInfo(rJob.getID());
	LOG.info("jInfo is :" + jInfo);

	//Assert if jobInfo is null
	Assert.assertNotNull("jobInfo is null", jInfo);

	//Wait for the job to start running.
	count = 0;
	while (jInfo.getStatus().getRunState() != JobStatus.RUNNING) {
	UtilsForTests.waitFor(10000);
	count++;
	jInfo = wovenClient.getJobInfo(rJob.getID());
	//If the count goes beyond a point, then Assert; This is to avoid
	//infinite loop under unforeseen circumstances.
	if (count > 10) {
	Assert.fail("job has not reached running state for more than" +
	"100 seconds. Failing at this point");
	}
	}

	LOG.info("job id is :" + rJob.getID().toString());

	TaskInfo[] taskInfos = cluster.getJTClient().getProxy()
	.getTaskInfo(rJob.getID());

	boolean distCacheFileIsFound;

	for (TaskInfo taskInfo : taskInfos) {
	distCacheFileIsFound = false;
	String[] taskTrackers = taskInfo.getTaskTrackers();

	for(String taskTracker : taskTrackers) {
	//Getting the exact FQDN of the tasktracker from
	//the tasktracker string.
	taskTracker = UtilsForTests.getFQDNofTT(taskTracker);
	tClient = cluster.getTTClient(taskTracker);
	String[] localDirs = tClient.getMapredLocalDirs();
	int distributedFileCount = 0;
	String localDirOnly = null;

	boolean FileNotPresentForThisDirectoryPath = false;

	//Go to every single path
	for (String localDir : localDirs) {
	FileNotPresentForThisDirectoryPath = false;
	localDirOnly = localDir;

	//Public Distributed cache will always be stored under
	//mapred.local.dir/tasktracker/archive
	localDirOnly = localDir + Path.SEPARATOR + TaskTracker.SUBDIR +
	Path.SEPARATOR + userName;

	//Private Distributed cache will always be stored under
	//mapre.local.dir/taskTracker/<username>/distcache
	//Checking for username directory to check if it has the
	//proper permissions
	localDir = localDir + Path.SEPARATOR +
	TaskTracker.getPrivateDistributedCacheDir(userName);

	FileStatus fileStatusMapredLocalDirUserName = null;

	try {
	fileStatusMapredLocalDirUserName = tClient.
	getFileStatus(localDirOnly, true);
	} catch (Exception e) {
	LOG.info("LocalDirOnly :" + localDirOnly + " not found");
	FileNotPresentForThisDirectoryPath = true;
	}

	//File will only be stored under one of the mapred.lcoal.dir
	//If other paths were hit, just continue
	if (FileNotPresentForThisDirectoryPath)
	continue;

	Path pathMapredLocalDirUserName =
	fileStatusMapredLocalDirUserName.getPath();
	FsPermission fsPermMapredLocalDirUserName =
	fileStatusMapredLocalDirUserName.getPermission();
	Assert.assertTrue("Directory Permission is not 700",
	fsPermMapredLocalDirUserName.equals(new FsPermission("700")));

	//Get file status of all the directories
	//and files under that path.
	FileStatus[] fileStatuses = tClient.listStatus(localDir,
	true, true);
	for (FileStatus fileStatus : fileStatuses) {
	Path path = fileStatus.getPath();
	LOG.info("path is :" + path.toString());
	//Checking if the received path ends with
	//the distributed filename
	distCacheFileIsFound = (path.toString()).
	endsWith(distributedFileName);
	//If file is found, check for its permission.
	//Since the file is found break out of loop
	if (distCacheFileIsFound){
	LOG.info("PATH found is :" + path.toString());
	distributedFileCount++;
	String filename = path.getName();
	FsPermission fsPerm = fileStatus.getPermission();
	Assert.assertTrue("File Permission is not 777",
	fsPerm.equals(new FsPermission("777")));
	}
	}
	}

	LOG.info("Distributed File count is :" + distributedFileCount);

	if (distributedFileCount > 1) {
	Assert.fail("The distributed cache file is more than one");
	} else if (distributedFileCount < 1)
	Assert.fail("The distributed cache file is less than one");
	if (!distCacheFileIsFound) {
	Assert.assertEquals("The distributed cache file does not exist",
	distCacheFileIsFound, false);
	}
	}

	//Allow the job to continue through MR control job.
	for (TaskInfo taskInfoRemaining : taskInfos) {
	FinishTaskControlAction action = new FinishTaskControlAction(TaskID
	.downgrade(taskInfoRemaining.getTaskID()));
	Collection<TTClient> tts = cluster.getTTClients();
	for (TTClient cli : tts) {
	cli.getProxy().sendAction(action);
	}
	}

	//Killing the job because all the verification needed
	//for this testcase is completed.
	rJob.killJob();
	}
	}
	}