You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by om...@apache.org on 2011/03/04 05:14:55 UTC
svn commit: r1077437 [3/5] - in
/hadoop/common/branches/branch-0.20-security-patches: ./
src/test/aop/build/ src/test/org/apache/hadoop/mapred/
src/test/system/aop/org/apache/hadoop/hdfs/
src/test/system/aop/org/apache/hadoop/hdfs/server/ src/test/syst...
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/configure.ac
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/c%2B%2B/runAs/configure.ac?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/configure.ac (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/configure.ac Fri Mar 4 04:14:53 2011
@@ -0,0 +1,65 @@
+# -*- Autoconf -*-
+# Process this file with autoconf to produce a configure script.
+
+#
+# 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.
+#
+
+AC_PREREQ(2.59)
+AC_INIT([runAs],[0.1])
+
+#changing default prefix value to empty string, so that binary does not
+#gets installed within system
+AC_PREFIX_DEFAULT(.)
+
+#add new arguments --with-home
+AC_ARG_WITH(home,[--with-home path to hadoop home dir])
+AC_CONFIG_SRCDIR([main.c])
+AC_CONFIG_HEADER([runAs.h])
+
+# Checks for programs.
+AC_PROG_CC
+
+# Checks for libraries.
+
+# Checks for header files.
+AC_HEADER_STDC
+AC_CHECK_HEADERS([stdlib.h string.h unistd.h fcntl.h])
+
+#check for HADOOP_HOME
+if test "$with_home" != ""
+then
+AC_DEFINE_UNQUOTED(HADOOP_HOME,"$with_home")
+fi
+
+# Checks for typedefs, structures, and compiler characteristics.
+AC_C_CONST
+AC_TYPE_PID_T
+AC_TYPE_MODE_T
+AC_TYPE_SIZE_T
+
+# Checks for library functions.
+AC_FUNC_MALLOC
+AC_FUNC_REALLOC
+AC_FUNC_CHOWN
+AC_CHECK_FUNCS([strerror memset mkdir rmdir strdup])
+
+AC_CONFIG_FILES([Makefile])
+AC_OUTPUT
+
+AC_HEADER_STDBOOL
+AC_PROG_MAKE_SET
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/main.c
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/c%2B%2B/runAs/main.c?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/main.c (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/main.c Fri Mar 4 04:14:53 2011
@@ -0,0 +1,59 @@
+/**
+ * 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.
+ */
+
+#include "runAs.h"
+
+/**
+ * The binary would be accepting the command of following format:
+ * cluster-controller user hostname hadoop-daemon.sh-command
+ */
+int main(int argc, char **argv) {
+ int errorcode;
+ char *user;
+ char *hostname;
+ char *command;
+ struct passwd user_detail;
+ int i = 1;
+ /*
+ * Minimum number of arguments required for the binary to perform.
+ */
+ if (argc < 4) {
+ fprintf(stderr, "Invalid number of arguments passed to the binary\n");
+ return INVALID_ARGUMENT_NUMER;
+ }
+
+ user = argv[1];
+ if (user == NULL) {
+ fprintf(stderr, "Invalid user name\n");
+ return INVALID_USER_NAME;
+ }
+
+ if (getuserdetail(user, &user_detail) != 0) {
+ fprintf(stderr, "Invalid user name\n");
+ return INVALID_USER_NAME;
+ }
+
+ if (user_detail.pw_gid == 0 || user_detail.pw_uid == 0) {
+ fprintf(stderr, "Cannot run tasks as super user\n");
+ return SUPER_USER_NOT_ALLOWED_TO_RUN_COMMANDS;
+ }
+
+ hostname = argv[2];
+ command = argv[3];
+ return process_controller_command(user, hostname, command);
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/runAs.c
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/c%2B%2B/runAs/runAs.c?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/runAs.c (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/runAs.c Fri Mar 4 04:14:53 2011
@@ -0,0 +1,111 @@
+/**
+ * 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.
+ */
+
+#include "runAs.h"
+
+/*
+ * Function to get the user details populated given a user name.
+ */
+int getuserdetail(char *user, struct passwd *user_detail) {
+ struct passwd *tempPwdPtr;
+ int size = sysconf(_SC_GETPW_R_SIZE_MAX);
+ char pwdbuffer[size];
+ if ((getpwnam_r(user, user_detail, pwdbuffer, size, &tempPwdPtr)) != 0) {
+ fprintf(stderr, "Invalid user provided to getpwnam\n");
+ return -1;
+ }
+ return 0;
+}
+
+/**
+ * Function to switch the user identity and set the appropriate
+ * group control as the user specified in the argument.
+ */
+int switchuser(char *user) {
+ //populate the user details
+ struct passwd user_detail;
+ if ((getuserdetail(user, &user_detail)) != 0) {
+ return INVALID_USER_NAME;
+ }
+ //set the right supplementary groups for the user.
+ if (initgroups(user_detail.pw_name, user_detail.pw_gid) != 0) {
+ fprintf(stderr, "Init groups call for the user : %s failed\n",
+ user_detail.pw_name);
+ return INITGROUPS_FAILED;
+ }
+ errno = 0;
+ //switch the group.
+ setgid(user_detail.pw_gid);
+ if (errno != 0) {
+ fprintf(stderr, "Setgid for the user : %s failed\n", user_detail.pw_name);
+ return SETUID_OPER_FAILED;
+ }
+ errno = 0;
+ //swith the user
+ setuid(user_detail.pw_uid);
+ if (errno != 0) {
+ fprintf(stderr, "Setuid for the user : %s failed\n", user_detail.pw_name);
+ return SETUID_OPER_FAILED;
+ }
+ errno = 0;
+ //set the effective user id.
+ seteuid(user_detail.pw_uid);
+ if (errno != 0) {
+ fprintf(stderr, "Seteuid for the user : %s failed\n", user_detail.pw_name);
+ return SETUID_OPER_FAILED;
+ }
+ return 0;
+}
+
+/*
+ * Top level method which processes a cluster management
+ * command.
+ */
+int process_cluster_command(char * user, char * node , char *command) {
+ char *finalcommandstr;
+ int len;
+ int errorcode = 0;
+ if (strncmp(command, "", strlen(command)) == 0) {
+ fprintf(stderr, "Invalid command passed\n");
+ return INVALID_COMMAND_PASSED;
+ }
+ len = STRLEN + strlen(command);
+ finalcommandstr = (char *) malloc((len + 1) * sizeof(char));
+ snprintf(finalcommandstr, len, SCRIPT_DIR_PATTERN, HADOOP_HOME,
+ command);
+ finalcommandstr[len + 1] = '\0';
+ errorcode = switchuser(user);
+ if (errorcode != 0) {
+ fprintf(stderr, "switch user failed\n");
+ return errorcode;
+ }
+ errno = 0;
+ execlp(SSH_COMMAND, SSH_COMMAND, node, finalcommandstr, NULL);
+ if (errno != 0) {
+ fprintf(stderr, "Excelp failed dude to : %s\n", strerror(errno));
+ }
+ return 0;
+}
+
+/*
+ * Process cluster controller command the API exposed to the
+ * main in order to execute the cluster commands.
+ */
+int process_controller_command(char *user, char * node, char *command) {
+ return process_cluster_command(user, node, command);
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/runAs.h.in
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/c%2B%2B/runAs/runAs.h.in?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/runAs.h.in (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/c++/runAs/runAs.h.in Fri Mar 4 04:14:53 2011
@@ -0,0 +1,59 @@
+/**
+ * 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.
+ */
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdarg.h>
+#include <string.h>
+#include <errno.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <pwd.h>
+#include <assert.h>
+#include <getopt.h>
+#include <grp.h>
+
+/*
+* List of possible error codes.
+*/
+enum errorcodes {
+ INVALID_ARGUMENT_NUMER = 1,
+ INVALID_USER_NAME, //2
+ SUPER_USER_NOT_ALLOWED_TO_RUN_COMMANDS, //3
+ INITGROUPS_FAILED, //4
+ SETUID_OPER_FAILED, //5
+ INVALID_COMMAND_PASSED, //6
+};
+
+#undef HADOOP_HOME
+
+#define SSH_COMMAND "ssh"
+
+#define SCRIPT_DIR_PATTERN "%s/bin/hadoop-daemon.sh %s" //%s to be substituded
+
+#define STRLEN strlen(SCRIPT_DIR_PATTERN) + strlen(HADOOP_HOME)
+
+/*
+ * Function to get the user details populated given a user name.
+ */
+int getuserdetails(char *user, struct passwd *user_detail);
+
+ /*
+ * Process cluster controller command the API exposed to the
+ * main in order to execute the cluster commands.
+ */
+int process_controller_command(char *user, char *node, char *command);
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/conf/hadoop-policy-system-test.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/conf/hadoop-policy-system-test.xml?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/conf/hadoop-policy-system-test.xml (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/conf/hadoop-policy-system-test.xml Fri Mar 4 04:14:53 2011
@@ -0,0 +1,52 @@
+<?xml version="1.0"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+
+<configuration>
+<!--
+ This is Herriot specific protocols. This section shouldn't be present in
+ a production cluster configuration. This file needs to be linked up to the
+ main conf/hadoop-policy.xml in the deployment process
+-->
+ <property>
+ <name>security.daemon.protocol.acl</name>
+ <value>*</value>
+ <description>ACL for DaemonProtocol, extended by all other
+ Herriot RPC protocols.
+ The ACL is a comma-separated list of user and group names. The user and
+ group list is separated by a blank. For e.g. "alice,bob users,wheel".
+ A special value of "*" means all users are allowed.</description>
+ </property>
+
+ <property>
+ <name>security.nn.protocol.acl</name>
+ <value>*</value>
+ <description>ACL for NNProtocol, used by the
+ Herriot AbstractDaemonCluster's implementations to connect to a remote
+ NameNode.
+ The ACL is a comma-separated list of user and group names. The user and
+ group list is separated by a blank. For e.g. "alice,bob users,wheel".
+ A special value of "*" means all users are allowed.</description>
+ </property>
+
+ <property>
+ <name>security.dn.protocol.acl</name>
+ <value>*</value>
+ <description>ACL for DNProtocol, used by the
+ Herriot AbstractDaemonCluster's implementations to connect to a remote
+ DataNode.
+ The ACL is a comma-separated list of user and group names. The user and
+ group list is separated by a blank. For e.g. "alice,bob users,wheel".
+ A special value of "*" means all users are allowed.</description>
+ </property>
+
+ <property>
+ <name>security.tt.protocol.acl</name>
+ <value>*</value>
+ <description>ACL for TTProtocol, used by the
+ Herriot AbstractDaemonCluster's implementations to connect to a remote
+ TaskTracker.
+ The ACL is a comma-separated list of user and group names. The user and
+ group list is separated by a blank. For e.g. "alice,bob users,wheel".
+ A special value of "*" means all users are allowed.</description>
+ </property>
+</configuration>
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/conf/system-test.xml
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/conf/system-test.xml?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/conf/system-test.xml (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/conf/system-test.xml Fri Mar 4 04:14:53 2011
@@ -0,0 +1,128 @@
+<?xml version="1.0"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+
+<!-- Put site-specific property overrides in this file. -->
+
+<configuration>
+<!-- Mandatory properties that are to be set and uncommented before running the tests -->
+
+<!-- $(YINST_ROOT) variable needs to be replace with something corporate
+neutral at the forward-port stage -->
+<property>
+ <name>test.system.hdrc.hadoophome</name>
+ <value>$(YINST_ROOT)/share/hadoop-current</value>
+ <description> This is the path to the home directory of the hadoop deployment.
+ </description>
+</property>
+<property>
+ <name>test.system.hdrc.hadoopconfdir</name>
+ <value>$(YINST_ROOT)/conf/hadoop</value>
+ <description> This is the path to the configuration directory of the hadoop
+ cluster that is deployed.
+ </description>
+</property>
+
+<property>
+ <name>test.system.hdrc.tt.hostfile</name>
+ <value>slaves.localcopy.txt</value>
+ <description> File name containing the hostnames where the TaskTrackers are running.
+ </description>
+</property>
+<property>
+ <name>test.system.hdrc.dn.hostfile</name>
+ <value>slaves.localcopy.txt</value>
+ <description> File name containing the hostnames where the DataNodes are running.
+ </description>
+</property>
+<property>
+ <name>test.system.mr.clusterprocess.impl.class</name>
+ <value>org.apache.hadoop.mapreduce.test.system.MRCluster$MRProcessManager</value>
+ <description>
+ Cluster process manager for the Mapreduce subsystem of the cluster. The value
+ org.apache.hadoop.mapreduce.test.system.MRCluster$MultiMRProcessManager can
+ be used to enable multi user support.
+ </description>
+</property>
+<property>
+ <name>test.system.hdfs.clusterprocess.impl.class</name>
+ <value>org.apache.hadoop.hdfs.test.system.HDFSCluster$HDFSProcessManager</value>
+ <description>
+ Cluster process manager for the Hdfs subsystem of the cluster. The value
+ org.apache.hadoop.hdfs.test.system.HDFSCluster$MultiUserHDFSProcessManager can
+ be used to enable multi user support.
+ </description>
+</property>
+<property>
+ <name>test.system.hdrc.deployed.scripts.dir</name>
+ <value>./src/test/system/scripts</value>
+ <description>
+ This directory hosts the scripts in the deployed location where
+ the system test client runs.
+ </description>
+</property>
+<property>
+ <name>test.system.hdrc.hadoopnewconfdir</name>
+ <value>$(TO_DO_GLOBAL_TMP_DIR)</value>
+ <description>
+ The directory where the new config files will be copied to in all
+ the clusters is pointed out this directory.
+ </description>
+</property>
+
+<!-- Mandatory keys to be set for the multi user support to be enabled. -->
+
+ <property>
+ <name>test.system.mr.clusterprocess.impl.class</name>
+ <value>org.apache.hadoop.mapreduce.test.system.MRCluster$MultiMRProcessManager</value>
+ <description>
+ Enabling multi user based cluster process manger.
+ </description>
+</property>
+<property>
+ <name>test.system.hdfs.clusterprocess.impl.class</name>
+ <value>org.apache.hadoop.hdfs.test.system.HDFSCluster$MultiUserHDFSProcessManager</value>
+ <description>
+ Enabling multi user based cluster process manger.
+ </description>
+</property>
+<property>
+ <name>test.system.hdrc.multi-user.binary.path</name>
+ <value>$(YINST_ROOT)/conf/hadoop/runAs</value>
+ <description>
+ Local file system path on gate way to cluster-controller binary including the binary name.
+ To build the binary the following commands need to be executed:
+ % ant run-as -Drun-as.hadoop.home.dir=(HADOOP_HOME of setup cluster)
+ % cp build-fi/system/c++-build/runAs $(test.system.hdrc.multi-user.binary.path)
+ Location of the cluster is important security precaution.
+ The binary should be owned by root and test user group permission should be set such a
+ way that it can be executed by binary. Example usage would be:
+ % sudo chown root binary
+ % sudo chmod 6511 binary
+ Change permission appropriately to make it more secure.
+ </description>
+</property>
+<property>
+ <name>test.system.hdrc.multi-user.managinguser.namenode</name>
+ <value>hdfs</value>
+ <description>
+ User value for managing the particular daemon, please note that these user should be
+ present on gateways also, an example configuration for the above would be
+ key name = test.system.hdrc.multi-user.managinguser.jobtracker
+ key value = guest
+ Please note the daemon names are all lower case, corresponding to hadoop-daemon.sh command.
+ </description>
+</property>
+<property>
+ <name>test.system.hdrc.multi-user.managinguser.datanode</name>
+ <value>hdfs</value>
+</property>
+<property>
+ <name>test.system.hdrc.multi-user.managinguser.jobtracker</name>
+ <value>mapred</value>
+</property>
+<property>
+ <name>test.system.hdrc.multi-user.managinguser.tasktracker</name>
+ <value>mapred</value>
+</property>
+
+</configuration>
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/TestHL040.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/TestHL040.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/TestHL040.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/TestHL040.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,69 @@
+/**
+ * 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.hdfs;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.test.system.DNClient;
+import org.apache.hadoop.hdfs.test.system.HDFSCluster;
+import org.apache.hadoop.hdfs.test.system.NNClient;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestHL040 {
+ private HDFSCluster cluster = null;
+ private static final Log LOG = LogFactory.getLog(TestHL040.class);
+
+ public TestHL040() throws Exception {
+ }
+
+ @Before
+ public void setupUp() throws Exception {
+ cluster = HDFSCluster.createCluster(new Configuration());
+ cluster.setUp();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ cluster.tearDown();
+ }
+
+ @Test
+ public void testConnect() throws IOException {
+ LOG.info("Staring TestHL040: connecting to the HDFSCluster ");
+ LOG.info("================ Getting namenode info ================");
+ NNClient dfsMaster = cluster.getNNClient();
+ LOG.info("Process info of namenode " + dfsMaster.getHostName() + " is: " +
+ dfsMaster.getProcessInfo());
+ LOG.info("================ Getting datanode info ================");
+ Collection<DNClient> clients = cluster.getDNClients();
+ for (DNClient dnC : clients) {
+ LOG.info("Process info of datanode " + dnC.getHostName() + " is: " +
+ dnC.getProcessInfo());
+ Assert.assertNotNull("Datanode process info isn't suppose to be null",
+ dnC.getProcessInfo());
+ }
+ }
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/DNClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/DNClient.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/DNClient.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/DNClient.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,82 @@
+/**
+ * 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.hdfs.test.system;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.test.system.process.RemoteProcess;
+
+/**
+ * Datanode client for system tests. Assumption of the class is that the
+ * configuration key is set for the configuration key : {@code
+ * DFSConfigKeys.DFS_DATANODE_IPC_ADDRESS_KEY} is set, only the port portion of
+ * the address is used.
+ */
+public class DNClient extends HDFSDaemonClient<DNProtocol> {
+
+ DNProtocol proxy;
+
+ public DNClient(Configuration conf, RemoteProcess process) throws IOException {
+ super(conf, process);
+ }
+
+ @Override
+ public void connect() throws IOException {
+ if (isConnected()) {
+ return;
+ }
+ String sockAddrStr = getConf().get(DFSConfigKeys.DFS_DATANODE_IPC_ADDRESS_KEY);
+ if (sockAddrStr == null) {
+ throw new IllegalArgumentException("Datenode IPC address is not set."
+ + "Check if " + DFSConfigKeys.DFS_DATANODE_IPC_ADDRESS_KEY
+ + " is configured.");
+ }
+ String[] splits = sockAddrStr.split(":");
+ if (splits.length != 2) {
+ throw new IllegalArgumentException(
+ "Datanode IPC address is not correctly configured");
+ }
+ String port = splits[1];
+ String sockAddr = getHostName() + ":" + port;
+ InetSocketAddress bindAddr = NetUtils.createSocketAddr(sockAddr);
+ proxy = (DNProtocol) RPC.getProxy(DNProtocol.class, DNProtocol.versionID,
+ bindAddr, getConf());
+ setConnected(true);
+ }
+
+ @Override
+ public void disconnect() throws IOException {
+ RPC.stopProxy(proxy);
+ setConnected(false);
+ }
+
+ @Override
+ protected DNProtocol getProxy() {
+ return proxy;
+ }
+
+ public Configuration getDatanodeConfig() throws IOException {
+ return getProxy().getDaemonConf();
+ }
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/DNProtocol.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/DNProtocol.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/DNProtocol.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/DNProtocol.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,36 @@
+/**
+ * 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.hdfs.test.system;
+
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.security.KerberosInfo;
+import org.apache.hadoop.test.system.DaemonProtocol;
+
+/**
+ * Client side API exposed from Datanode.
+ * Actual implementations are likely to be injected
+ *
+ * The protocol has to be annotated so KerberosInfo can be filled in during
+ * creation of a ipc.Client connection
+ */
+@KerberosInfo(
+ serverPrincipal = DFSConfigKeys.DFS_DATANODE_USER_NAME_KEY)
+public interface DNProtocol extends DaemonProtocol {
+ public static final long versionID = 1L;
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/HDFSCluster.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/HDFSCluster.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/HDFSCluster.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/HDFSCluster.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,149 @@
+/**
+ * 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.hdfs.test.system;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.test.system.AbstractDaemonClient;
+import org.apache.hadoop.test.system.AbstractDaemonCluster;
+import org.apache.hadoop.test.system.process.ClusterProcessManager;
+import org.apache.hadoop.test.system.process.HadoopDaemonRemoteCluster;
+import org.apache.hadoop.test.system.process.MultiUserHadoopDaemonRemoteCluster;
+import org.apache.hadoop.test.system.process.RemoteProcess;
+import org.apache.hadoop.test.system.process.HadoopDaemonRemoteCluster.HadoopDaemonInfo;
+
+public class HDFSCluster extends AbstractDaemonCluster {
+
+ static {
+ Configuration.addDefaultResource("hdfs-site.xml");
+ }
+
+ private static final Log LOG = LogFactory.getLog(HDFSCluster.class);
+ public static final String CLUSTER_PROCESS_MGR_IMPL =
+ "test.system.hdfs.clusterprocess.impl.class";
+
+ private HDFSCluster(Configuration conf, ClusterProcessManager rCluster)
+ throws IOException {
+ super(conf, rCluster);
+ }
+
+ /**
+ * Key is used to to point to the file containing hostnames of tasktrackers
+ */
+ public static final String CONF_HADOOP_DN_HOSTFILE_NAME =
+ "test.system.hdrc.dn.hostfile";
+
+ private static List<HadoopDaemonInfo> hdfsDaemonInfos;
+
+ private static String nnHostName;
+ private static String DN_hostFileName;
+
+ protected enum Role {NN, DN}
+
+ @Override
+ protected AbstractDaemonClient
+ createClient(RemoteProcess process) throws IOException {
+ Enum<?> pRole = process.getRole();
+ if (Role.NN.equals(pRole)) {
+ return createNNClient(process);
+ } else if (Role.DN.equals(pRole)) {
+ return createDNClient(process);
+ } else throw new IOException("Role " + pRole +
+ " is not supported by HDFSCluster");
+ }
+
+ protected DNClient createDNClient(RemoteProcess dnDaemon) throws IOException {
+ return new DNClient(getConf(), dnDaemon);
+ }
+
+ protected NNClient createNNClient(RemoteProcess nnDaemon) throws IOException {
+ return new NNClient(getConf(), nnDaemon);
+ }
+
+ public NNClient getNNClient () {
+ Iterator<AbstractDaemonClient> iter = getDaemons().get(Role.NN).iterator();
+ return (NNClient) iter.next();
+ }
+
+ public List<DNClient> getDNClients () {
+ return (List) getDaemons().get(Role.DN);
+ }
+
+ public DNClient getDNClient (String hostname) {
+ for (DNClient dnC : getDNClients()) {
+ if (dnC.getHostName().equals(hostname))
+ return dnC;
+ }
+ return null;
+ }
+
+ public static class HDFSProcessManager extends HadoopDaemonRemoteCluster {
+ public HDFSProcessManager() {
+ super(hdfsDaemonInfos);
+ }
+ }
+
+ public static class MultiUserHDFSProcessManager
+ extends MultiUserHadoopDaemonRemoteCluster {
+ public MultiUserHDFSProcessManager() {
+ super(hdfsDaemonInfos);
+ }
+ }
+
+
+ public static HDFSCluster createCluster(Configuration conf) throws Exception {
+ conf.addResource("system-test.xml");
+ String sockAddrStr = FileSystem.getDefaultUri(conf).getAuthority();
+ if (sockAddrStr == null) {
+ throw new IllegalArgumentException("Namenode IPC address is not set");
+ }
+ String[] splits = sockAddrStr.split(":");
+ if (splits.length != 2) {
+ throw new IllegalArgumentException(
+ "Namenode report IPC is not correctly configured");
+ }
+ nnHostName = splits[0];
+ DN_hostFileName = conf.get(CONF_HADOOP_DN_HOSTFILE_NAME, "slaves");
+
+ hdfsDaemonInfos = new ArrayList<HadoopDaemonInfo>();
+ hdfsDaemonInfos.add(new HadoopDaemonInfo("namenode",
+ Role.NN, Arrays.asList(new String[]{nnHostName})));
+ hdfsDaemonInfos.add(new HadoopDaemonInfo("datanode",
+ Role.DN, DN_hostFileName));
+
+ String implKlass = conf.get(CLUSTER_PROCESS_MGR_IMPL);
+ if (implKlass == null || implKlass.isEmpty()) {
+ implKlass = HDFSCluster.HDFSProcessManager.class.getName();
+ }
+ Class<ClusterProcessManager> klass =
+ (Class<ClusterProcessManager>) Class.forName(implKlass);
+ ClusterProcessManager clusterProcessMgr = klass.newInstance();
+ LOG.info("Created ClusterProcessManager as " + implKlass);
+ clusterProcessMgr.init(conf);
+ return new HDFSCluster(conf, clusterProcessMgr);
+ }
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/HDFSDaemonClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/HDFSDaemonClient.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/HDFSDaemonClient.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/HDFSDaemonClient.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,43 @@
+/**
+ * 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.hdfs.test.system;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.test.system.AbstractDaemonClient;
+import org.apache.hadoop.test.system.DaemonProtocol;
+import org.apache.hadoop.test.system.process.RemoteProcess;
+
+public abstract class HDFSDaemonClient<PROXY extends DaemonProtocol>
+ extends AbstractDaemonClient<PROXY> {
+
+ public HDFSDaemonClient(Configuration conf, RemoteProcess process)
+ throws IOException {
+ super(conf, process);
+ }
+
+ public String[] getHDFSDataDirs() throws IOException {
+ return getProxy().getDaemonConf().getStrings("dfs.data.dir");
+ }
+
+ public String getHDFSNameDirs() throws IOException {
+ return getProxy().getDaemonConf().getStrings("dfs.name.dir")[0];
+ }
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/NNClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/NNClient.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/NNClient.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/NNClient.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,71 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * 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.hdfs.test.system;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.test.system.process.RemoteProcess;
+
+public class NNClient extends HDFSDaemonClient<NNProtocol> {
+
+ NNProtocol proxy;
+
+ public NNClient(Configuration conf, RemoteProcess process) throws IOException {
+ super(conf, process);
+ }
+
+ @Override
+ public void connect() throws IOException {
+ if (isConnected())
+ return;
+ String sockAddrStr = FileSystem.getDefaultUri(getConf()).getAuthority();
+ if (sockAddrStr == null) {
+ throw new IllegalArgumentException("Namenode IPC address is not set");
+ }
+ String[] splits = sockAddrStr.split(":");
+ if (splits.length != 2) {
+ throw new IllegalArgumentException(
+ "Namenode report IPC is not correctly configured");
+ }
+ String port = splits[1];
+ String sockAddr = getHostName() + ":" + port;
+
+ InetSocketAddress bindAddr = NetUtils.createSocketAddr(sockAddr);
+ proxy = (NNProtocol) RPC.getProxy(NNProtocol.class, NNProtocol.versionID,
+ bindAddr, getConf());
+ setConnected(true);
+ }
+
+ @Override
+ public void disconnect() throws IOException {
+ RPC.stopProxy(proxy);
+ setConnected(false);
+ }
+
+ @Override
+ protected NNProtocol getProxy() {
+ return proxy;
+ }
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/NNProtocol.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/NNProtocol.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/NNProtocol.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/hdfs/test/system/NNProtocol.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,36 @@
+/**
+ * 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.hdfs.test.system;
+
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.security.KerberosInfo;
+import org.apache.hadoop.test.system.DaemonProtocol;
+
+/**
+ * Client side API exposed from Namenode.
+ * Actual implementations are likely to be injected
+ *
+ * The protocol has to be annotated so KerberosInfo can be filled in during
+ * creation of a ipc.Client connection
+ */
+@KerberosInfo(
+ serverPrincipal = DFSConfigKeys.DFS_NAMENODE_USER_NAME_KEY)
+public interface NNProtocol extends DaemonProtocol {
+ public static final long versionID = 1L;
+}
Modified: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TTTaskInfoImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TTTaskInfoImpl.java?rev=1077437&r1=1077436&r2=1077437&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TTTaskInfoImpl.java (original)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TTTaskInfoImpl.java Fri Mar 4 04:14:53 2011
@@ -37,13 +37,14 @@ abstract class TTTaskInfoImpl implements
Configuration conf;
String user;
boolean isTaskCleanupTask;
+ private String pid;
public TTTaskInfoImpl() {
}
public TTTaskInfoImpl(boolean slotTaken, boolean wasKilled,
TaskStatus status, Configuration conf, String user,
- boolean isTaskCleanupTask) {
+ boolean isTaskCleanupTask, String pid) {
super();
this.slotTaken = slotTaken;
this.wasKilled = wasKilled;
@@ -51,6 +52,7 @@ abstract class TTTaskInfoImpl implements
this.conf = conf;
this.user = user;
this.isTaskCleanupTask = isTaskCleanupTask;
+ this.pid = pid;
}
@Override
@@ -82,6 +84,11 @@ abstract class TTTaskInfoImpl implements
}
@Override
+ public String getPid() {
+ return pid;
+ }
+
+ @Override
public void readFields(DataInput in) throws IOException {
slotTaken = in.readBoolean();
wasKilled = in.readBoolean();
@@ -89,6 +96,7 @@ abstract class TTTaskInfoImpl implements
conf.readFields(in);
user = in.readUTF();
isTaskCleanupTask = in.readBoolean();
+ pid = in.readUTF();
}
@Override
@@ -98,6 +106,11 @@ abstract class TTTaskInfoImpl implements
conf.write(out);
out.writeUTF(user);
out.writeBoolean(isTaskCleanupTask);
+ if (pid != null) {
+ out.writeUTF(pid);
+ } else {
+ out.writeUTF("");
+ }
status.write(out);
}
@@ -109,8 +122,8 @@ abstract class TTTaskInfoImpl implements
public MapTTTaskInfo(boolean slotTaken, boolean wasKilled,
MapTaskStatus status, Configuration conf, String user,
- boolean isTaskCleanup) {
- super(slotTaken, wasKilled, status, conf, user, isTaskCleanup);
+ boolean isTaskCleanup,String pid) {
+ super(slotTaken, wasKilled, status, conf, user, isTaskCleanup, pid);
}
@Override
@@ -133,8 +146,8 @@ abstract class TTTaskInfoImpl implements
public ReduceTTTaskInfo(boolean slotTaken, boolean wasKilled,
ReduceTaskStatus status, Configuration conf, String user,
- boolean isTaskCleanup) {
- super(slotTaken, wasKilled, status, conf, user, isTaskCleanup);
+ boolean isTaskCleanup, String pid) {
+ super(slotTaken, wasKilled, status, conf, user, isTaskCleanup, pid);
}
@Override
Modified: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestCluster.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestCluster.java?rev=1077437&r1=1077436&r2=1077437&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestCluster.java (original)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestCluster.java Fri Mar 4 04:14:53 2011
@@ -18,6 +18,7 @@
package org.apache.hadoop.mapred;
+import java.security.PrivilegedExceptionAction;
import java.util.Collection;
import junit.framework.Assert;
@@ -39,6 +40,8 @@ import org.apache.hadoop.mapreduce.test.
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.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.test.system.AbstractDaemonClient;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -86,22 +89,41 @@ public class TestCluster {
cluster.getJTClient().verifyJobHistory(rJob.getID());
}
- @Test
+ //@Test
public void testFileStatus() throws Exception {
- JTClient jt = cluster.getJTClient();
- String dir = ".";
- checkFileStatus(jt.getFileStatus(dir, true));
- checkFileStatus(jt.listStatus(dir, false, true), dir);
- for (TTClient tt : cluster.getTTClients()) {
- String[] localDirs = tt.getMapredLocalDirs();
- for (String localDir : localDirs) {
- checkFileStatus(tt.listStatus(localDir, true, false), localDir);
- checkFileStatus(tt.listStatus(localDir, true, true), localDir);
+ UserGroupInformation ugi =
+ UserGroupInformation.createRemoteUser(cluster
+ .getJTClient().getProxy().getDaemonUser());
+ ugi.doAs(new PrivilegedExceptionAction<Void>() {
+ @Override
+ public Void run() throws Exception {
+ MRCluster myCluster = null;
+ try {
+ myCluster = MRCluster.createCluster(cluster.getConf());
+ myCluster.connect();
+ JTClient jt = myCluster.getJTClient();
+ String dir = ".";
+ checkFileStatus(jt.getFileStatus(dir, true));
+ checkFileStatus(jt.listStatus(dir, false, true), dir);
+ for (TTClient tt : myCluster.getTTClients()) {
+ String[] localDirs = tt.getMapredLocalDirs();
+ for (String localDir : localDirs) {
+ checkFileStatus(tt.listStatus(localDir, true, false), localDir);
+ checkFileStatus(tt.listStatus(localDir, true, true), localDir);
+ }
+ }
+ String systemDir = jt.getClient().getSystemDir().toString();
+ checkFileStatus(jt.listStatus(systemDir, false, true), systemDir);
+ checkFileStatus(jt.listStatus(jt.getLogDir(), true, true), jt
+ .getLogDir());
+ } finally {
+ if (myCluster != null) {
+ myCluster.disconnect();
+ }
+ }
+ return null;
}
- }
- String systemDir = jt.getClient().getSystemDir().toString();
- checkFileStatus(jt.listStatus(systemDir, false, true), systemDir);
- checkFileStatus(jt.listStatus(jt.getLogDir(), true, true), jt.getLogDir());
+ });
}
private void checkFileStatus(FileStatus[] fs, String path) {
@@ -149,6 +171,10 @@ public class TestCluster {
LOG.info("Waiting till job starts running one map");
TaskInfo[] myTaskInfos = wovenClient.getTaskInfo(id);
+ boolean isOneTaskStored = false;
+ String sometaskpid = null;
+ org.apache.hadoop.mapreduce.TaskAttemptID sometaskId = null;
+ TTClient myCli = null;
for(TaskInfo info : myTaskInfos) {
if(!info.isSetupOrCleanup()) {
String[] taskTrackers = info.getTaskTrackers();
@@ -162,6 +188,27 @@ public class TestCluster {
Assert.assertNotNull(ttTaskInfo.getUser());
Assert.assertTrue(ttTaskInfo.getTaskStatus().getProgress() >= 0.0);
Assert.assertTrue(ttTaskInfo.getTaskStatus().getProgress() <= 1.0);
+ //Get the pid of the task attempt. The task need not have
+ //reported the pid of the task by the time we are checking
+ //the pid. So perform null check.
+ String pid = ttTaskInfo.getPid();
+ int i = 1;
+ while(pid.isEmpty()) {
+ Thread.sleep(1000);
+ LOG.info("Waiting for task to report its pid back");
+ ttTaskInfo = ttCli.getProxy().getTask(taskId);
+ pid = ttTaskInfo.getPid();
+ if(i == 40) {
+ Assert.fail("The task pid not reported for 40 seconds.");
+ }
+ i++;
+ }
+ if(!isOneTaskStored) {
+ sometaskpid = pid;
+ sometaskId = ttTaskInfo.getTaskStatus().getTaskID();
+ myCli = ttCli;
+ isOneTaskStored = true;
+ }
LOG.info("verified task progress to be between 0 and 1");
State state = ttTaskInfo.getTaskStatus().getRunState();
if (ttTaskInfo.getTaskStatus().getProgress() < 1.0 &&
@@ -176,5 +223,97 @@ public class TestCluster {
}
}
rJob.killJob();
+ int i = 1;
+ while (!rJob.isComplete()) {
+ Thread.sleep(1000);
+ if (i == 40) {
+ Assert
+ .fail("The job not completed within 40 seconds after killing it.");
+ }
+ i++;
+ }
+ TTTaskInfo myTaskInfo = myCli.getProxy().getTask(sometaskId.getTaskID());
+ i = 0;
+ while (myTaskInfo != null && !myTaskInfo.getPid().isEmpty()) {
+ LOG.info("sleeping till task is retired from TT memory");
+ Thread.sleep(1000);
+ myTaskInfo = myCli.getProxy().getTask(sometaskId.getTaskID());
+ if (i == 40) {
+ Assert
+ .fail("Task not retired from TT memory within 40 seconds of job completeing");
+ }
+ i++;
+ }
+ Assert.assertFalse(myCli.getProxy().isProcessTreeAlive(sometaskpid));
}
+
+ @Test
+ public void testClusterRestart() throws Exception {
+ cluster.stop();
+ // Give the cluster time to stop the whole cluster.
+ AbstractDaemonClient cli = cluster.getJTClient();
+ int i = 1;
+ while (i < 40) {
+ try {
+ cli.ping();
+ Thread.sleep(1000);
+ i++;
+ } catch (Exception e) {
+ break;
+ }
+ }
+ if (i >= 40) {
+ Assert.fail("JT on " + cli.getHostName() + " Should have been down.");
+ }
+ i = 1;
+ for (AbstractDaemonClient tcli : cluster.getTTClients()) {
+ i = 1;
+ while (i < 40) {
+ try {
+ tcli.ping();
+ Thread.sleep(1000);
+ i++;
+ } catch (Exception e) {
+ break;
+ }
+ }
+ if (i >= 40) {
+ Assert.fail("TT on " + tcli.getHostName() + " Should have been down.");
+ }
+ }
+ cluster.start();
+ cli = cluster.getJTClient();
+ i = 1;
+ while (i < 40) {
+ try {
+ cli.ping();
+ break;
+ } catch (Exception e) {
+ i++;
+ Thread.sleep(1000);
+ LOG.info("Waiting for Jobtracker on host : "
+ + cli.getHostName() + " to come up.");
+ }
+ }
+ if (i >= 40) {
+ Assert.fail("JT on " + cli.getHostName() + " Should have been up.");
+ }
+ for (AbstractDaemonClient tcli : cluster.getTTClients()) {
+ i = 1;
+ while (i < 40) {
+ try {
+ tcli.ping();
+ break;
+ } catch (Exception e) {
+ i++;
+ Thread.sleep(1000);
+ LOG.info("Waiting for Tasktracker on host : "
+ + tcli.getHostName() + " to come up.");
+ }
+ }
+ if (i >= 40) {
+ Assert.fail("TT on " + tcli.getHostName() + " Should have been Up.");
+ }
+ }
+ }
}
Modified: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestControlledJob.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestControlledJob.java?rev=1077437&r1=1077436&r2=1077437&view=diff
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestControlledJob.java (original)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestControlledJob.java Fri Mar 4 04:14:53 2011
@@ -98,11 +98,22 @@ public class TestControlledJob {
}
jInfo = wovenClient.getJobInfo(id);
- while(!jInfo.getStatus().isJobComplete()) {
- Thread.sleep(1000);
- jInfo = wovenClient.getJobInfo(id);
+ int i = 1;
+ if (jInfo != null) {
+ while (!jInfo.getStatus().isJobComplete()) {
+ Thread.sleep(1000);
+ jInfo = wovenClient.getJobInfo(id);
+ if (jInfo == null) {
+ break;
+ }
+ if(i > 40) {
+ Assert.fail("Controlled Job with ID : "
+ + jInfo.getID()
+ + " has not completed in 40 seconds after signalling.");
+ }
+ i++;
+ }
}
-
LOG.info("Job sucessfully completed after signalling!!!!");
}
}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestDistributedCacheModifiedFile.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestDistributedCacheModifiedFile.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestDistributedCacheModifiedFile.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestDistributedCacheModifiedFile.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,342 @@
+/**
+ * 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 java.util.ArrayList;
+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.examples.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 it is modified before and after being
+ * accessed by maximum two jobs. Once a job uses a distributed cache file
+ * that file is stored in the mapred.local.dir. If the next job
+ * uses the same file, but with differnt timestamp, then that
+ * file is stored again. So, if two jobs choose
+ * the same tasktracker for their job execution
+ * then, the distributed cache file should be found twice.
+ *
+ * This testcase runs a job with a distributed cache file. All the
+ * tasks' corresponding tasktracker's handle is got and checked for
+ * the presence of distributed cache with proper permissions in the
+ * proper directory. Next when job
+ * runs again and if any of its tasks hits the same tasktracker, which
+ * ran one of the task of the previous job, then that
+ * file should be uploaded again and task should not use the old file.
+ * This is verified.
+*/
+
+public class TestDistributedCacheModifiedFile {
+
+ private static MRCluster cluster = null;
+ private static FileSystem dfs = null;
+ private static FileSystem ttFs = null;
+ private static JobClient client = null;
+ private static FsPermission permission = new FsPermission((short)00777);
+
+ 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(TestDistributedCacheModifiedFile.class);
+
+ public TestDistributedCacheModifiedFile() 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();
+ }
+ //Waiting for 5 seconds to make sure tasktrackers are ready
+ Thread.sleep(5000);
+ }
+
+ @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 modified 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;
+ //This boolean will decide whether to run job again
+ boolean continueLoop = true;
+ //counter for job Loop
+ int countLoop = 0;
+ //This counter increases with all the tasktrackers in which tasks ran
+ int taskTrackerCounter = 0;
+ //This will store all the tasktrackers in which tasks ran
+ ArrayList<String> taskTrackerCollection = new ArrayList<String>();
+ //This boolean tells if two tasks ran onteh same tasktracker or not
+ boolean taskTrackerFound = false;
+
+ do {
+ SleepJob job = new SleepJob();
+ job.setConf(conf);
+ conf = job.setupJobConf(5, 1, 1000, 1000, 100, 100);
+
+ //Before starting, Modify the file
+ 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);
+
+ 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
+ RunningJob rJob = cluster.getJTClient().getClient().submitJob(jconf);
+
+ //counter for job Loop
+ countLoop++;
+
+ 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 break; This is to avoid
+ //infinite loop under unforeseen circumstances. Testcase will anyway
+ //fail later.
+ 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) {
+ //Formatting tasktracker to get just its FQDN
+ taskTracker = UtilsForTests.getFQDNofTT(taskTracker);
+ LOG.info("taskTracker is :" + taskTracker);
+
+ //The tasktrackerFound variable is initialized
+ taskTrackerFound = false;
+
+ //This will be entered from the second job onwards
+ if (countLoop > 1) {
+ if (taskTracker != null) {
+ continueLoop = taskTrackerCollection.contains(taskTracker);
+ }
+ if (continueLoop) {
+ taskTrackerFound = true;
+ }
+ }
+ //Collecting the tasktrackers
+ if (taskTracker != null)
+ taskTrackerCollection.add(taskTracker);
+
+ //we have loopped through two times to look for task
+ //getting submitted on same tasktrackers.The same tasktracker
+ //for subsequent jobs was not hit maybe because of many number
+ //of tasktrackers. So, testcase has to stop here.
+ if (countLoop > 1) {
+ continueLoop = false;
+ }
+
+ tClient = cluster.getTTClient(taskTracker);
+
+ //tClient maybe null because the task is already dead. Ex: setup
+ if (tClient == null) {
+ continue;
+ }
+
+ String[] localDirs = tClient.getMapredLocalDirs();
+ int distributedFileCount = 0;
+ //Go to every single path
+ for (String localDir : localDirs) {
+ //Public Distributed cache will always be stored under
+ //mapre.local.dir/tasktracker/archive
+ localDir = localDir + Path.SEPARATOR +
+ TaskTracker.getPublicDistributedCacheDir();
+ LOG.info("localDir is : " + localDir);
+
+ //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.debug("The distributed FileCount is :" + distributedFileCount);
+ LOG.debug("The taskTrackerFound is :" + taskTrackerFound);
+
+ // If distributed cache is modified in dfs
+ // between two job runs, it can be present more than once
+ // in any of the task tracker, in which job ran.
+ if (distributedFileCount != 2 && taskTrackerFound) {
+ Assert.fail("The distributed cache file has to be two. " +
+ "But found was " + distributedFileCount);
+ } else if (distributedFileCount > 1 && !taskTrackerFound) {
+ Assert.fail("The distributed cache file cannot more than one." +
+ " But found was " + distributedFileCount);
+ } else if (distributedFileCount < 1)
+ Assert.fail("The distributed cache file is less than one. " +
+ "But found was " + distributedFileCount);
+ 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();
+
+ //Waiting for 3 seconds for cleanup to start
+ Thread.sleep(3000);
+
+ //Getting the last cleanup task's tasktracker also, as
+ //distributed cache gets uploaded even during cleanup.
+ TaskInfo[] myTaskInfos = wovenClient.getTaskInfo(rJob.getID());
+ if (myTaskInfos != null) {
+ for(TaskInfo info : myTaskInfos) {
+ if(info.isSetupOrCleanup()) {
+ String[] taskTrackers = info.getTaskTrackers();
+ for(String taskTracker : taskTrackers) {
+ //Formatting tasktracker to get just its FQDN
+ taskTracker = UtilsForTests.getFQDNofTT(taskTracker);
+ LOG.info("taskTracker is :" + taskTracker);
+ //Collecting the tasktrackers
+ if (taskTracker != null)
+ taskTrackerCollection.add(taskTracker);
+ }
+ }
+ }
+ }
+
+ //Making sure that the job is complete.
+ while (jInfo != null && !jInfo.getStatus().isJobComplete()) {
+ Thread.sleep(10000);
+ jInfo = wovenClient.getJobInfo(rJob.getID());
+ }
+
+ } while (continueLoop);
+ }
+}
Added: hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestDistributedCachePrivateFile.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestDistributedCachePrivateFile.java?rev=1077437&view=auto
==============================================================================
--- hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestDistributedCachePrivateFile.java (added)
+++ hadoop/common/branches/branch-0.20-security-patches/src/test/system/java/org/apache/hadoop/mapred/TestDistributedCachePrivateFile.java Fri Mar 4 04:14:53 2011
@@ -0,0 +1,281 @@
+/**
+ * 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.examples.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);
+ conf = job.setupJobConf(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
+ RunningJob rJob = cluster.getJTClient().getClient().submitJob(jconf);
+
+ 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();
+ }
+ }
+}