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}