ProcessJob extends AbstractProcessJob { @Override public void run() throws Exception { try { resolveProps(); } catch (final Exception e) { handleError("Bad property definition! " + e.getMessage(), e); } if (this.getSysProps().getBoolean(MEMCHECK_ENABLED, true) && this.getJobProps().getBoolean(AZKABAN_MEMORY_CHECK, true)) { final Pair<Long, Long> memPair = getProcMemoryRequirement(); final long xms = memPair.getFirst(); final long xmx = memPair.getSecond(); final String oomMsg = String .format("Cannot request memory (Xms %d kb, Xmx %d kb) from system for job %s", xms, xmx, getId()); int attempt; boolean isMemGranted = true; final SystemMemoryInfo memInfo = SERVICE_PROVIDER.getInstance(SystemMemoryInfo.class); for (attempt = 1; attempt <= Constants.MEMORY_CHECK_RETRY_LIMIT; attempt++) { isMemGranted = memInfo.canSystemGrantMemory(xmx); if (isMemGranted) { info(String.format("Memory granted for job %s", getId())); if (attempt > 1) { this.commonMetrics.decrementOOMJobWaitCount(); } break; } if (attempt < Constants.MEMORY_CHECK_RETRY_LIMIT) { info(String.format(oomMsg + ", sleep for %s secs and retry, attempt %s of %s", TimeUnit.MILLISECONDS.toSeconds( Constants.MEMORY_CHECK_INTERVAL_MS), attempt, Constants.MEMORY_CHECK_RETRY_LIMIT)); if (attempt == 1) { this.commonMetrics.incrementOOMJobWaitCount(); } synchronized (this) { try { this.wait(Constants.MEMORY_CHECK_INTERVAL_MS); } catch (final InterruptedException e) { info(String .format("Job %s interrupted while waiting for memory check retry", getId())); } } if (this.killed) { this.commonMetrics.decrementOOMJobWaitCount(); info(String.format("Job %s was killed while waiting for memory check retry", getId())); return; } } } if (!isMemGranted) { this.commonMetrics.decrementOOMJobWaitCount(); handleError(oomMsg, null); } } List<String> commands = null; try { commands = getCommandList(); } catch (final Exception e) { handleError("Job set up failed: " + e.getMessage(), e); } final long startMs = System.currentTimeMillis(); if (commands == null) { handleError("There are no commands to execute", null); } info(commands.size() + " commands to execute."); final File[] propFiles = initPropsFiles(); final Map<String, String> envVars = getEnvironmentVariables(); envVars.put(KRB5CCNAME, getKrb5ccname(this.getJobProps())); String executeAsUserBinaryPath = null; String effectiveUser = null; final boolean isExecuteAsUser = this.getSysProps().getBoolean(EXECUTE_AS_USER, true); final Set<String> blackListedUsers = new HashSet<>( Arrays.asList( this.getSysProps() .getString(Constants.ConfigurationKeys.BLACK_LISTED_USERS, "root,azkaban") .split(",") ) ); if (isExecuteAsUser) { final String nativeLibFolder = this.getSysProps().getString(AZKABAN_SERVER_NATIVE_LIB_FOLDER); executeAsUserBinaryPath = String.format("%s/%s", nativeLibFolder, "execute-as-user"); effectiveUser = getEffectiveUser(this.getJobProps()); if (blackListedUsers.contains(effectiveUser)) { throw new RuntimeException( String.format("Not permitted to proxy as '%s' through Azkaban", effectiveUser) ); } if (!canWriteInCurrentWorkingDirectory(effectiveUser)) { info("Changing current working directory ownership"); assignUserFileOwnership(effectiveUser, getWorkingDirectory()); } for (int i = 0; i < 2; i++) { info("Changing properties files ownership"); assignUserFileOwnership(effectiveUser, propFiles[i].getAbsolutePath()); } } for (String command : commands) { AzkabanProcessBuilder builder = null; if (isExecuteAsUser) { command = String.format("%s %s %s", executeAsUserBinaryPath, effectiveUser, command); info("Command: " + command); builder = new AzkabanProcessBuilder(partitionCommandLine(command)) .setEnv(envVars).setWorkingDir(getCwd()).setLogger(getLog()) .enableExecuteAsUser().setExecuteAsUserBinaryPath(executeAsUserBinaryPath) .setEffectiveUser(effectiveUser); } else { info("Command: " + command); builder = new AzkabanProcessBuilder(partitionCommandLine(command)) .setEnv(envVars).setWorkingDir(getCwd()).setLogger(getLog()); } if (builder.getEnv().size() > 0) { info("Environment variables: " + builder.getEnv()); } info("Working directory: " + builder.getWorkingDir()); this.logJobProperties(); synchronized (this) { if (this.killed) { info("The job is killed. Abort. No job process created."); return; } this.process = builder.build(); } try { this.process.run(); this.success = true; } catch (final Throwable e) { for (final File file : propFiles) { if (file != null && file.exists()) { file.delete(); } } throw new RuntimeException(e); } finally { info("Process with id " + this.process.getProcessId() + " completed " + (this.success ? "successfully" : "unsuccessfully") + " in " + ((System.currentTimeMillis() - startMs) / 1000) + " seconds."); } } generateProperties(propFiles[1]); } ProcessJob(final String jobId, final Props sysProps,
      final Props jobProps, final Logger log);  ProcessJob(final String jobId, final Props sysProps,
      final Props jobProps, final Props privateProps, final Logger log); static String[] partitionCommandLine(final String command); @Override void run(); @Override void cancel(); @Override double getProgress(); int getProcessId(); String getPath(); static final String COMMAND; static final String AZKABAN_MEMORY_CHECK; @Deprecated
  static final String NATIVE_LIB_FOLDER; static final String EXECUTE_AS_USER; static final String KRB5CCNAME; }
@Test public void testOneUnixCommand() throws Exception { this.props.put(ProcessJob.COMMAND, "ls -al"); this.job.run(); }
@Test public void testOneUnixCommandWithProxyUserInsteadOfSubmitUser() throws Exception { this.props.removeLocal(CommonJobProperties.SUBMIT_USER); this.props.put(JobProperties.USER_TO_PROXY, "test_user"); this.props.put(ProcessJob.COMMAND, "ls -al"); this.job.run(); }
@Test(expected = RuntimeException.class) public void testOneUnixCommandWithNoUser() throws Exception { this.props.removeLocal(CommonJobProperties.SUBMIT_USER); this.props.put(ProcessJob.COMMAND, "ls -al"); this.job.run(); }
@Test(expected = RuntimeException.class) public void testOneUnixCommandWithRootUser() throws Exception { this.props.removeLocal(CommonJobProperties.SUBMIT_USER); this.props.put(JobProperties.USER_TO_PROXY, "root"); this.props.put("execute.as.user", "true"); this.props.put(ProcessJob.COMMAND, "ls -al"); this.job.run(); }
@Test(expected = RuntimeException.class) public void testOneUnixCommandWithAzkabanUser() throws Exception { this.props.removeLocal(CommonJobProperties.SUBMIT_USER); this.props.put(JobProperties.USER_TO_PROXY, "azkaban"); this.props.put("execute.as.user", "true"); this.props.put(ProcessJob.COMMAND, "ls -al"); this.job.run(); }
@Test public void testFailedUnixCommand() throws Exception { this.props.put(ProcessJob.COMMAND, "xls -al"); try { this.job.run(); } catch (final RuntimeException e) { Assert.assertTrue(true); e.printStackTrace(); } }
@Test public void testMultipleUnixCommands() throws Exception { this.props.put(ProcessJob.COMMAND, "pwd"); this.props.put("command.1", "date"); this.props.put("command.2", "whoami"); this.job.run(); }