001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *      http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.apache.oozie.service;
020
021import org.apache.hadoop.io.Text;
022import org.apache.hadoop.mapred.JobClient;
023import org.apache.hadoop.mapred.JobConf;
024import org.apache.hadoop.fs.FileSystem;
025import org.apache.hadoop.fs.Path;
026import org.apache.hadoop.conf.Configuration;
027import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
028import org.apache.hadoop.net.NetUtils;
029import org.apache.hadoop.security.SecurityUtil;
030import org.apache.hadoop.security.UserGroupInformation;
031import org.apache.hadoop.filecache.DistributedCache;
032import org.apache.hadoop.security.token.Token;
033import org.apache.oozie.ErrorCode;
034import org.apache.oozie.action.hadoop.JavaActionExecutor;
035import org.apache.oozie.util.ParamChecker;
036import org.apache.oozie.util.XConfiguration;
037import org.apache.oozie.util.XLog;
038import org.apache.oozie.util.JobUtils;
039import org.apache.oozie.workflow.lite.LiteWorkflowAppParser;
040
041import java.io.File;
042import java.io.FileInputStream;
043import java.io.FilenameFilter;
044import java.io.IOException;
045import java.io.InputStream;
046import java.lang.reflect.InvocationTargetException;
047import java.lang.reflect.Method;
048import java.net.InetAddress;
049import java.net.URI;
050import java.net.URISyntaxException;
051import java.security.PrivilegedExceptionAction;
052import java.util.Arrays;
053import java.util.Comparator;
054import java.util.HashMap;
055import java.util.Map;
056import java.util.Set;
057import java.util.HashSet;
058import java.util.concurrent.ConcurrentHashMap;
059
060/**
061 * The HadoopAccessorService returns HadoopAccessor instances configured to work on behalf of a user-group. <p/> The
062 * default accessor used is the base accessor which just injects the UGI into the configuration instance used to
063 * create/obtain JobClient and FileSystem instances.
064 */
065public class HadoopAccessorService implements Service {
066
067    private static XLog LOG = XLog.getLog(HadoopAccessorService.class);
068
069    public static final String CONF_PREFIX = Service.CONF_PREFIX + "HadoopAccessorService.";
070    public static final String JOB_TRACKER_WHITELIST = CONF_PREFIX + "jobTracker.whitelist";
071    public static final String NAME_NODE_WHITELIST = CONF_PREFIX + "nameNode.whitelist";
072    public static final String HADOOP_CONFS = CONF_PREFIX + "hadoop.configurations";
073    public static final String ACTION_CONFS = CONF_PREFIX + "action.configurations";
074    public static final String KERBEROS_AUTH_ENABLED = CONF_PREFIX + "kerberos.enabled";
075    public static final String KERBEROS_KEYTAB = CONF_PREFIX + "keytab.file";
076    public static final String KERBEROS_PRINCIPAL = CONF_PREFIX + "kerberos.principal";
077    public static final Text MR_TOKEN_ALIAS = new Text("oozie mr token");
078
079    protected static final String OOZIE_HADOOP_ACCESSOR_SERVICE_CREATED = "oozie.HadoopAccessorService.created";
080    /** The Kerberos principal for the job tracker.*/
081    protected static final String JT_PRINCIPAL = "mapreduce.jobtracker.kerberos.principal";
082    /** The Kerberos principal for the resource manager.*/
083    protected static final String RM_PRINCIPAL = "yarn.resourcemanager.principal";
084    protected static final String HADOOP_JOB_TRACKER = "mapred.job.tracker";
085    protected static final String HADOOP_JOB_TRACKER_2 = "mapreduce.jobtracker.address";
086    protected static final String HADOOP_YARN_RM = "yarn.resourcemanager.address";
087    private static final Map<String, Text> mrTokenRenewers = new HashMap<String, Text>();
088
089    private static final String DEFAULT_ACTIONNAME = "default";
090
091    private Set<String> jobTrackerWhitelist = new HashSet<String>();
092    private Set<String> nameNodeWhitelist = new HashSet<String>();
093    private Map<String, Configuration> hadoopConfigs = new HashMap<String, Configuration>();
094    private Map<String, File> actionConfigDirs = new HashMap<String, File>();
095    private Map<String, Map<String, XConfiguration>> actionConfigs = new HashMap<String, Map<String, XConfiguration>>();
096
097    private UserGroupInformationService ugiService;
098
099    /**
100     * Supported filesystem schemes for namespace federation
101     */
102    public static final String SUPPORTED_FILESYSTEMS = CONF_PREFIX + "supported.filesystems";
103    private Set<String> supportedSchemes;
104    private boolean allSchemesSupported;
105
106    public void init(Services services) throws ServiceException {
107        this.ugiService = services.get(UserGroupInformationService.class);
108        init(services.getConf());
109    }
110
111    //for testing purposes, see XFsTestCase
112    public void init(Configuration conf) throws ServiceException {
113        for (String name : ConfigurationService.getStrings(conf, JOB_TRACKER_WHITELIST)) {
114            String tmp = name.toLowerCase().trim();
115            if (tmp.length() == 0) {
116                continue;
117            }
118            jobTrackerWhitelist.add(tmp);
119        }
120        LOG.info(
121                "JOB_TRACKER_WHITELIST :" + jobTrackerWhitelist.toString()
122                        + ", Total entries :" + jobTrackerWhitelist.size());
123        for (String name : ConfigurationService.getStrings(conf, NAME_NODE_WHITELIST)) {
124            String tmp = name.toLowerCase().trim();
125            if (tmp.length() == 0) {
126                continue;
127            }
128            nameNodeWhitelist.add(tmp);
129        }
130        LOG.info(
131                "NAME_NODE_WHITELIST :" + nameNodeWhitelist.toString()
132                        + ", Total entries :" + nameNodeWhitelist.size());
133
134        boolean kerberosAuthOn = ConfigurationService.getBoolean(conf, KERBEROS_AUTH_ENABLED);
135        LOG.info("Oozie Kerberos Authentication [{0}]", (kerberosAuthOn) ? "enabled" : "disabled");
136        if (kerberosAuthOn) {
137            kerberosInit(conf);
138        }
139        else {
140            Configuration ugiConf = new Configuration();
141            ugiConf.set("hadoop.security.authentication", "simple");
142            UserGroupInformation.setConfiguration(ugiConf);
143        }
144
145        if (ugiService == null) { //for testing purposes, see XFsTestCase
146            this.ugiService = new UserGroupInformationService();
147        }
148
149        loadHadoopConfigs(conf);
150        preLoadActionConfigs(conf);
151
152        supportedSchemes = new HashSet<String>();
153        String[] schemesFromConf = ConfigurationService.getStrings(conf, SUPPORTED_FILESYSTEMS);
154        if(schemesFromConf != null) {
155            for (String scheme: schemesFromConf) {
156                scheme = scheme.trim();
157                // If user gives "*", supportedSchemes will be empty, so that checking is not done i.e. all schemes allowed
158                if(scheme.equals("*")) {
159                    if(schemesFromConf.length > 1) {
160                        throw new ServiceException(ErrorCode.E0100, getClass().getName(),
161                            SUPPORTED_FILESYSTEMS + " should contain either only wildcard or explicit list, not both");
162                    }
163                    allSchemesSupported = true;
164                }
165                supportedSchemes.add(scheme);
166            }
167        }
168
169        setConfigForHadoopSecurityUtil(conf);
170    }
171
172    private void setConfigForHadoopSecurityUtil(Configuration conf) {
173        // Prior to HADOOP-12954 (2.9.0+), Hadoop sets hadoop.security.token.service.use_ip on startup in a static block with no
174        // way for Oozie to change it because Oozie doesn't load *-site.xml files on the classpath.  HADOOP-12954 added a way to
175        // set this property via a setConfiguration method.  Ideally, this would be part of JobClient so Oozie wouldn't have to
176        // worry about it and we could have different values for different clusters, but we can't; so we have to use the same value
177        // for every cluster Oozie is configured for.  To that end, we'll use the default NN's configs.  If that's not defined,
178        // we'll use the wildcard's configs.  And if that's not defined, we'll use an arbitrary cluster's configs.  In any case,
179        // if the version of Hadoop we're using doesn't include HADOOP-12954, we'll do nothing (there's no workaround), and
180        // hadoop.security.token.service.use_ip will have the default value.
181        String nameNode = conf.get(LiteWorkflowAppParser.DEFAULT_NAME_NODE);
182        if (nameNode != null) {
183            nameNode = nameNode.trim();
184            if (nameNode.isEmpty()) {
185                nameNode = null;
186            }
187        }
188        if (nameNode == null && hadoopConfigs.containsKey("*")) {
189            nameNode = "*";
190        }
191        if (nameNode == null) {
192            for (String nn : hadoopConfigs.keySet()) {
193                nn = nn.trim();
194                if (!nn.isEmpty()) {
195                    nameNode = nn;
196                    break;
197                }
198            }
199        }
200        if (nameNode != null) {
201            Configuration hConf = getConfiguration(nameNode);
202            try {
203                Method setConfigurationMethod = SecurityUtil.class.getMethod("setConfiguration", Configuration.class);
204                setConfigurationMethod.invoke(null, hConf);
205                LOG.debug("Setting Hadoop SecurityUtil Configuration to that of {0}", nameNode);
206            } catch (NoSuchMethodException e) {
207                LOG.debug("Not setting Hadoop SecurityUtil Configuration because this version of Hadoop doesn't support it");
208            } catch (Exception e) {
209                LOG.error("An Exception occurred while trying to call setConfiguration on {0} via Reflection.  It won't be called.",
210                        SecurityUtil.class.getName(), e);
211            }
212        }
213    }
214
215    private void kerberosInit(Configuration serviceConf) throws ServiceException {
216            try {
217                String keytabFile = ConfigurationService.get(serviceConf, KERBEROS_KEYTAB).trim();
218                if (keytabFile.length() == 0) {
219                    throw new ServiceException(ErrorCode.E0026, KERBEROS_KEYTAB);
220                }
221                String principal = SecurityUtil.getServerPrincipal(
222                        serviceConf.get(KERBEROS_PRINCIPAL, "oozie/localhost@LOCALHOST"),
223                        InetAddress.getLocalHost().getCanonicalHostName());
224                if (principal.length() == 0) {
225                    throw new ServiceException(ErrorCode.E0026, KERBEROS_PRINCIPAL);
226                }
227                Configuration conf = new Configuration();
228                conf.set("hadoop.security.authentication", "kerberos");
229                UserGroupInformation.setConfiguration(conf);
230                UserGroupInformation.loginUserFromKeytab(principal, keytabFile);
231                LOG.info("Got Kerberos ticket, keytab [{0}], Oozie principal principal [{1}]",
232                        keytabFile, principal);
233            }
234            catch (ServiceException ex) {
235                throw ex;
236            }
237            catch (Exception ex) {
238                throw new ServiceException(ErrorCode.E0100, getClass().getName(), ex.getMessage(), ex);
239            }
240    }
241
242    private static final String[] HADOOP_CONF_FILES =
243        {"core-site.xml", "hdfs-site.xml", "mapred-site.xml", "yarn-site.xml", "hadoop-site.xml", "ssl-client.xml"};
244
245
246    private Configuration loadHadoopConf(File dir) throws IOException {
247        Configuration hadoopConf = new XConfiguration();
248        for (String file : HADOOP_CONF_FILES) {
249            File f = new File(dir, file);
250            if (f.exists()) {
251                InputStream is = new FileInputStream(f);
252                Configuration conf = new XConfiguration(is);
253                is.close();
254                XConfiguration.copy(conf, hadoopConf);
255            }
256        }
257        return hadoopConf;
258    }
259
260    private Map<String, File> parseConfigDirs(String[] confDefs, String type) throws ServiceException, IOException {
261        Map<String, File> map = new HashMap<String, File>();
262        File configDir = new File(ConfigurationService.getConfigurationDirectory());
263        for (String confDef : confDefs) {
264            if (confDef.trim().length() > 0) {
265                String[] parts = confDef.split("=");
266                if (parts.length == 2) {
267                    String hostPort = parts[0];
268                    String confDir = parts[1];
269                    File dir = new File(confDir);
270                    if (!dir.isAbsolute()) {
271                        dir = new File(configDir, confDir);
272                    }
273                    if (dir.exists()) {
274                        map.put(hostPort.toLowerCase(), dir);
275                    }
276                    else {
277                        throw new ServiceException(ErrorCode.E0100, getClass().getName(),
278                                                   "could not find " + type + " configuration directory: " +
279                                                   dir.getAbsolutePath());
280                    }
281                }
282                else {
283                    throw new ServiceException(ErrorCode.E0100, getClass().getName(),
284                                               "Incorrect " + type + " configuration definition: " + confDef);
285                }
286            }
287        }
288        return map;
289    }
290
291    private void loadHadoopConfigs(Configuration serviceConf) throws ServiceException {
292        try {
293            Map<String, File> map = parseConfigDirs(ConfigurationService.getStrings(serviceConf, HADOOP_CONFS),
294                    "hadoop");
295            for (Map.Entry<String, File> entry : map.entrySet()) {
296                hadoopConfigs.put(entry.getKey(), loadHadoopConf(entry.getValue()));
297            }
298        }
299        catch (ServiceException ex) {
300            throw ex;
301        }
302        catch (Exception ex) {
303            throw new ServiceException(ErrorCode.E0100, getClass().getName(), ex.getMessage(), ex);
304        }
305    }
306
307    private void preLoadActionConfigs(Configuration serviceConf) throws ServiceException {
308        try {
309            actionConfigDirs = parseConfigDirs(ConfigurationService.getStrings(serviceConf, ACTION_CONFS), "action");
310            for (String hostport : actionConfigDirs.keySet()) {
311                actionConfigs.put(hostport, new ConcurrentHashMap<String, XConfiguration>());
312            }
313        }
314        catch (ServiceException ex) {
315            throw ex;
316        }
317        catch (Exception ex) {
318            throw new ServiceException(ErrorCode.E0100, getClass().getName(), ex.getMessage(), ex);
319        }
320    }
321
322    public void destroy() {
323    }
324
325    public Class<? extends Service> getInterface() {
326        return HadoopAccessorService.class;
327    }
328
329    private UserGroupInformation getUGI(String user) throws IOException {
330        return ugiService.getProxyUser(user);
331    }
332
333    /**
334     * Creates a JobConf using the site configuration for the specified hostname:port.
335     * <p/>
336     * If the specified hostname:port is not defined it falls back to the '*' site
337     * configuration if available. If the '*' site configuration is not available,
338     * the JobConf has all Hadoop defaults.
339     *
340     * @param hostPort hostname:port to lookup Hadoop site configuration.
341     * @return a JobConf with the corresponding site configuration for hostPort.
342     */
343    public JobConf createJobConf(String hostPort) {
344        JobConf jobConf = new JobConf();
345        XConfiguration.copy(getConfiguration(hostPort), jobConf);
346        jobConf.setBoolean(OOZIE_HADOOP_ACCESSOR_SERVICE_CREATED, true);
347        return jobConf;
348    }
349
350    private XConfiguration loadActionConf(String hostPort, String action) {
351        File dir = actionConfigDirs.get(hostPort);
352        XConfiguration actionConf = new XConfiguration();
353        if (dir != null) {
354            // See if a dir with the action name exists.   If so, load all the xml files in the dir
355            File actionConfDir = new File(dir, action);
356
357            if (actionConfDir.exists() && actionConfDir.isDirectory()) {
358                LOG.info("Processing configuration files under [{0}]"
359                                + " for action [{1}] and hostPort [{2}]",
360                        actionConfDir.getAbsolutePath(), action, hostPort);
361                File[] xmlFiles = actionConfDir.listFiles(
362                        new FilenameFilter() {
363                            @Override
364                            public boolean accept(File dir, String name) {
365                                return name.endsWith(".xml");
366                            }});
367                Arrays.sort(xmlFiles, new Comparator<File>() {
368                    @Override
369                    public int compare(File o1, File o2) {
370                        return o1.getName().compareTo(o2.getName());
371                    }
372                });
373                for (File f : xmlFiles) {
374                    if (f.isFile() && f.canRead()) {
375                        LOG.info("Processing configuration file [{0}]", f.getName());
376                        FileInputStream fis = null;
377                        try {
378                            fis = new FileInputStream(f);
379                            XConfiguration conf = new XConfiguration(fis);
380                            XConfiguration.copy(conf, actionConf);
381                        }
382                        catch (IOException ex) {
383                            LOG
384                                .warn("Could not read file [{0}] for action [{1}] configuration and hostPort [{2}]",
385                                        f.getAbsolutePath(), action, hostPort);
386                        }
387                        finally {
388                            if (fis != null) {
389                                try { fis.close(); } catch(IOException ioe) { }
390                            }
391                        }
392                    }
393                }
394            }
395        }
396
397        // Now check for <action.xml>   This way <action.xml> has priority over <action-dir>/*.xml
398
399        File actionConfFile = new File(dir, action + ".xml");
400        if (actionConfFile.exists()) {
401            try {
402                XConfiguration conf = new XConfiguration(new FileInputStream(actionConfFile));
403                XConfiguration.copy(conf, actionConf);
404            }
405            catch (IOException ex) {
406                LOG.warn("Could not read file [{0}] for action [{1}] configuration for hostPort [{2}]",
407                        actionConfFile.getAbsolutePath(), action, hostPort);
408            }
409        }
410
411        return actionConf;
412    }
413
414    /**
415     * Returns a Configuration containing any defaults for an action for a particular cluster.
416     * <p/>
417     * This configuration is used as default for the action configuration and enables cluster
418     * level default values per action.
419     *
420     * @param hostPort hostname"port to lookup the action default confiugration.
421     * @param action action name.
422     * @return the default configuration for the action for the specified cluster.
423     */
424    public XConfiguration createActionDefaultConf(String hostPort, String action) {
425        hostPort = (hostPort != null) ? hostPort.toLowerCase() : null;
426        Map<String, XConfiguration> hostPortActionConfigs = actionConfigs.get(hostPort);
427        if (hostPortActionConfigs == null) {
428            hostPortActionConfigs = actionConfigs.get("*");
429            hostPort = "*";
430        }
431        XConfiguration actionConf = hostPortActionConfigs.get(action);
432        if (actionConf == null) {
433            // doing lazy loading as we don't know upfront all actions, no need to synchronize
434            // as it is a read operation an in case of a race condition loading and inserting
435            // into the Map is idempotent and the action-config Map is a ConcurrentHashMap
436
437            // We first load a action of type default
438            // This allows for global configuration for all actions - for example
439            // all launchers in one queue and actions in another queue
440            // Are some configuration that applies to multiple actions - like
441            // config libraries path etc
442            actionConf = loadActionConf(hostPort, DEFAULT_ACTIONNAME);
443
444            // Action specific default configuration will override the default action config
445
446            XConfiguration.copy(loadActionConf(hostPort, action), actionConf);
447            hostPortActionConfigs.put(action, actionConf);
448        }
449        return new XConfiguration(actionConf.toProperties());
450    }
451
452    private Configuration getConfiguration(String hostPort) {
453        hostPort = (hostPort != null) ? hostPort.toLowerCase() : null;
454        Configuration conf = hadoopConfigs.get(hostPort);
455        if (conf == null) {
456            conf = hadoopConfigs.get("*");
457            if (conf == null) {
458                conf = new XConfiguration();
459            }
460        }
461        return conf;
462    }
463
464    /**
465     * Return a JobClient created with the provided user/group.
466     *
467     *
468     * @param conf JobConf with all necessary information to create the
469     *        JobClient.
470     * @return JobClient created with the provided user/group.
471     * @throws HadoopAccessorException if the client could not be created.
472     */
473    public JobClient createJobClient(String user, final JobConf conf) throws HadoopAccessorException {
474        ParamChecker.notEmpty(user, "user");
475        if (!conf.getBoolean(OOZIE_HADOOP_ACCESSOR_SERVICE_CREATED, false)) {
476            throw new HadoopAccessorException(ErrorCode.E0903);
477        }
478        String jobTracker = conf.get(JavaActionExecutor.HADOOP_JOB_TRACKER);
479        validateJobTracker(jobTracker);
480        try {
481            UserGroupInformation ugi = getUGI(user);
482            JobClient jobClient = ugi.doAs(new PrivilegedExceptionAction<JobClient>() {
483                public JobClient run() throws Exception {
484                    return new JobClient(conf);
485                }
486            });
487            Token<DelegationTokenIdentifier> mrdt = jobClient.getDelegationToken(getMRDelegationTokenRenewer(conf));
488            conf.getCredentials().addToken(MR_TOKEN_ALIAS, mrdt);
489            return jobClient;
490        }
491        catch (InterruptedException ex) {
492            throw new HadoopAccessorException(ErrorCode.E0902, ex.getMessage(), ex);
493        }
494        catch (IOException ex) {
495            throw new HadoopAccessorException(ErrorCode.E0902, ex.getMessage(), ex);
496        }
497    }
498
499    /**
500     * Return a FileSystem created with the provided user for the specified URI.
501     *
502     *
503     * @param uri file system URI.
504     * @param conf Configuration with all necessary information to create the FileSystem.
505     * @return FileSystem created with the provided user/group.
506     * @throws HadoopAccessorException if the filesystem could not be created.
507     */
508    public FileSystem createFileSystem(String user, final URI uri, final Configuration conf)
509            throws HadoopAccessorException {
510        ParamChecker.notEmpty(user, "user");
511        if (!conf.getBoolean(OOZIE_HADOOP_ACCESSOR_SERVICE_CREATED, false)) {
512            throw new HadoopAccessorException(ErrorCode.E0903);
513        }
514
515        checkSupportedFilesystem(uri);
516
517        String nameNode = uri.getAuthority();
518        if (nameNode == null) {
519            nameNode = conf.get("fs.default.name");
520            if (nameNode != null) {
521                try {
522                    nameNode = new URI(nameNode).getAuthority();
523                }
524                catch (URISyntaxException ex) {
525                    throw new HadoopAccessorException(ErrorCode.E0902, ex.getMessage(), ex);
526                }
527            }
528        }
529        validateNameNode(nameNode);
530
531        try {
532            UserGroupInformation ugi = getUGI(user);
533            return ugi.doAs(new PrivilegedExceptionAction<FileSystem>() {
534                public FileSystem run() throws Exception {
535                    return FileSystem.get(uri, conf);
536                }
537            });
538        }
539        catch (InterruptedException ex) {
540            throw new HadoopAccessorException(ErrorCode.E0902, ex.getMessage(), ex);
541        }
542        catch (IOException ex) {
543            throw new HadoopAccessorException(ErrorCode.E0902, ex.getMessage(), ex);
544        }
545    }
546
547    /**
548     * Validate Job tracker
549     * @param jobTrackerUri
550     * @throws HadoopAccessorException
551     */
552    protected void validateJobTracker(String jobTrackerUri) throws HadoopAccessorException {
553        validate(jobTrackerUri, jobTrackerWhitelist, ErrorCode.E0900);
554    }
555
556    /**
557     * Validate Namenode list
558     * @param nameNodeUri
559     * @throws HadoopAccessorException
560     */
561    protected void validateNameNode(String nameNodeUri) throws HadoopAccessorException {
562        validate(nameNodeUri, nameNodeWhitelist, ErrorCode.E0901);
563    }
564
565    private void validate(String uri, Set<String> whitelist, ErrorCode error) throws HadoopAccessorException {
566        if (uri != null) {
567            uri = uri.toLowerCase().trim();
568            if (whitelist.size() > 0 && !whitelist.contains(uri)) {
569                throw new HadoopAccessorException(error, uri, whitelist);
570            }
571        }
572    }
573
574    public Text getMRDelegationTokenRenewer(JobConf jobConf) throws IOException {
575        if (UserGroupInformation.isSecurityEnabled()) { // secure cluster
576            return getMRTokenRenewerInternal(jobConf);
577        }
578        else {
579            return MR_TOKEN_ALIAS; //Doesn't matter what we pass as renewer
580        }
581    }
582
583    // Package private for unit test purposes
584    Text getMRTokenRenewerInternal(JobConf jobConf) throws IOException {
585        // Getting renewer correctly for JT principal also though JT in hadoop 1.x does not have
586        // support for renewing/cancelling tokens
587        String servicePrincipal = jobConf.get(RM_PRINCIPAL, jobConf.get(JT_PRINCIPAL));
588        Text renewer;
589        if (servicePrincipal != null) { // secure cluster
590            renewer = mrTokenRenewers.get(servicePrincipal);
591            if (renewer == null) {
592                // Mimic org.apache.hadoop.mapred.Master.getMasterPrincipal()
593                String target = jobConf.get(HADOOP_YARN_RM, jobConf.get(HADOOP_JOB_TRACKER_2));
594                if (target == null) {
595                    target = jobConf.get(HADOOP_JOB_TRACKER);
596                }
597                try {
598                    String addr = NetUtils.createSocketAddr(target).getHostName();
599                    renewer = new Text(SecurityUtil.getServerPrincipal(servicePrincipal, addr));
600                    LOG.info("Delegation Token Renewer details: Principal=" + servicePrincipal + ",Target=" + target
601                            + ",Renewer=" + renewer);
602                }
603                catch (IllegalArgumentException iae) {
604                    renewer = new Text(servicePrincipal.split("[/@]")[0]);
605                    LOG.info("Delegation Token Renewer for " + servicePrincipal + " is " + renewer);
606                }
607                mrTokenRenewers.put(servicePrincipal, renewer);
608            }
609        }
610        else {
611            renewer = MR_TOKEN_ALIAS; //Doesn't matter what we pass as renewer
612        }
613        return renewer;
614    }
615
616    public void addFileToClassPath(String user, final Path file, final Configuration conf)
617            throws IOException {
618        ParamChecker.notEmpty(user, "user");
619        try {
620            UserGroupInformation ugi = getUGI(user);
621            ugi.doAs(new PrivilegedExceptionAction<Void>() {
622                @Override
623                public Void run() throws Exception {
624                    JobUtils.addFileToClassPath(file, conf, null);
625                    return null;
626                }
627            });
628
629        }
630        catch (InterruptedException ex) {
631            throw new IOException(ex);
632        }
633
634    }
635
636    /**
637     * checks configuration parameter if filesystem scheme is among the list of supported ones
638     * this makes system robust to filesystems other than HDFS also
639     */
640
641    public void checkSupportedFilesystem(URI uri) throws HadoopAccessorException {
642        if (allSchemesSupported)
643            return;
644        String uriScheme = uri.getScheme();
645        if (uriScheme != null) {    // skip the check if no scheme is given
646            if(!supportedSchemes.isEmpty()) {
647                LOG.debug("Checking if filesystem " + uriScheme + " is supported");
648                if (!supportedSchemes.contains(uriScheme)) {
649                    throw new HadoopAccessorException(ErrorCode.E0904, uriScheme, uri.toString());
650                }
651             }
652         }
653    }
654
655    public Set<String> getSupportedSchemes() {
656        return supportedSchemes;
657    }
658
659}