github
/
DolphinScheduler
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git
1
0
Fork 0
Code Issues Releases Wiki Activity
Browse Source

[FIX-#3177]Task time parameter parsing error (#4224) 

* [FIX-#3177]Task time parameter parsing error

rerun schedule time error

this closes #3177

* [FIX-#3177]Task time parameter parsing error

rerun schedule time error

this closes #3177
pull/3/MERGE
Kirs 4 years ago committed by GitHub
parent
95d320d7da
commit
8919e3838f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 108 additions and 107 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/ParameterUtils.java
  2. 211
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/shell/ShellTask.java

4
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/ParameterUtils.java
Unescape View File

@ -111,7 +111,9 @@ public class ParameterUtils {
} }
// replace variable ${} form,refers to the replacement of system variables and custom variables // replace variable ${} form,refers to the replacement of system variables and custom variables
parameterString = PlaceholderUtils.replacePlaceholders(parameterString, parameterMap, true); if (!parameterMap.isEmpty()) {
parameterString = PlaceholderUtils.replacePlaceholders(parameterString, parameterMap, true);
}
// replace time $[...] form, eg. $[yyyyMMdd] // replace time $[...] form, eg. $[yyyyMMdd]
if (cronTime != null) { if (cronTime != null) {

211
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/shell/ShellTask.java
Unescape View File

@ -14,8 +14,8 @@
* See the License for the specific language governing permissions and * See the License for the specific language governing permissions and
* limitations under the License. * limitations under the License.
*/ */
package org.apache.dolphinscheduler.server.worker.task.shell;
package org.apache.dolphinscheduler.server.worker.task.shell;
import org.apache.dolphinscheduler.common.Constants; import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.CommandType; import org.apache.dolphinscheduler.common.enums.CommandType;
@ -23,7 +23,7 @@ import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.task.AbstractParameters; import org.apache.dolphinscheduler.common.task.AbstractParameters;
import org.apache.dolphinscheduler.common.task.shell.ShellParameters; import org.apache.dolphinscheduler.common.task.shell.ShellParameters;
import org.apache.dolphinscheduler.common.utils.DateUtils; import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.*; import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils; import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.common.utils.ParameterUtils; import org.apache.dolphinscheduler.common.utils.ParameterUtils;
import org.apache.dolphinscheduler.server.entity.TaskExecutionContext; import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
@ -31,7 +31,6 @@ import org.apache.dolphinscheduler.server.utils.ParamUtils;
import org.apache.dolphinscheduler.server.worker.task.AbstractTask; import org.apache.dolphinscheduler.server.worker.task.AbstractTask;
import org.apache.dolphinscheduler.server.worker.task.CommandExecuteResult; import org.apache.dolphinscheduler.server.worker.task.CommandExecuteResult;
import org.apache.dolphinscheduler.server.worker.task.ShellCommandExecutor; import org.apache.dolphinscheduler.server.worker.task.ShellCommandExecutor;
import org.slf4j.Logger;
import java.io.File; import java.io.File;
import java.nio.file.Files; import java.nio.file.Files;
@ -43,136 +42,136 @@ import java.nio.file.attribute.PosixFilePermissions;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
import org.slf4j.Logger;
/** /**
* shell task * shell task
*/ */
public class ShellTask extends AbstractTask { public class ShellTask extends AbstractTask {
/** /**
* shell parameters * shell parameters
*/ */
private ShellParameters shellParameters; private ShellParameters shellParameters;
/** /**
* shell command executor * shell command executor
*/ */
private ShellCommandExecutor shellCommandExecutor; private ShellCommandExecutor shellCommandExecutor;
/** /**
* taskExecutionContext * taskExecutionContext
*/ */
private TaskExecutionContext taskExecutionContext; private TaskExecutionContext taskExecutionContext;
/** /**
* constructor * constructor
* @param taskExecutionContext taskExecutionContext *
* @param logger logger * @param taskExecutionContext taskExecutionContext
*/ * @param logger logger
public ShellTask(TaskExecutionContext taskExecutionContext, Logger logger) { */
super(taskExecutionContext, logger); public ShellTask(TaskExecutionContext taskExecutionContext, Logger logger) {
super(taskExecutionContext, logger);
this.taskExecutionContext = taskExecutionContext;
this.shellCommandExecutor = new ShellCommandExecutor(this::logHandle, this.taskExecutionContext = taskExecutionContext;
this.shellCommandExecutor = new ShellCommandExecutor(this::logHandle,
taskExecutionContext, taskExecutionContext,
logger); logger);
} }
@Override @Override
public void init() { public void init() {
logger.info("shell task params {}", taskExecutionContext.getTaskParams()); logger.info("shell task params {}", taskExecutionContext.getTaskParams());
shellParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), ShellParameters.class); shellParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), ShellParameters.class);
if (!shellParameters.checkParameters()) { if (!shellParameters.checkParameters()) {
throw new RuntimeException("shell task params is not valid"); throw new RuntimeException("shell task params is not valid");
} }
}
@Override
public void handle() throws Exception {
try {
// construct process
CommandExecuteResult commandExecuteResult = shellCommandExecutor.run(buildCommand());
setExitStatusCode(commandExecuteResult.getExitStatusCode());
setAppIds(commandExecuteResult.getAppIds());
setProcessId(commandExecuteResult.getProcessId());
} catch (Exception e) {
logger.error("shell task error", e);
setExitStatusCode(Constants.EXIT_CODE_FAILURE);
throw e;
} }
}
@Override
public void cancelApplication(boolean cancelApplication) throws Exception {
// cancel process
shellCommandExecutor.cancelApplication();
}
/**
* create command
* @return file name
* @throws Exception exception
*/
private String buildCommand() throws Exception {
// generate scripts
String fileName = String.format("%s/%s_node.%s",
taskExecutionContext.getExecutePath(),
taskExecutionContext.getTaskAppId(), OSUtils.isWindows() ? "bat" : "sh");
Path path = new File(fileName).toPath(); @Override
public void handle() throws Exception {
try {
// construct process
CommandExecuteResult commandExecuteResult = shellCommandExecutor.run(buildCommand());
setExitStatusCode(commandExecuteResult.getExitStatusCode());
setAppIds(commandExecuteResult.getAppIds());
setProcessId(commandExecuteResult.getProcessId());
} catch (Exception e) {
logger.error("shell task error", e);
setExitStatusCode(Constants.EXIT_CODE_FAILURE);
throw e;
}
}
if (Files.exists(path)) { @Override
return fileName; public void cancelApplication(boolean cancelApplication) throws Exception {
// cancel process
shellCommandExecutor.cancelApplication();
} }
String script = shellParameters.getRawScript().replaceAll("\\r\\n", "\n");
/** /**
* combining local and global parameters * create command
*
* @return file name
* @throws Exception exception
*/ */
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()), private String buildCommand() throws Exception {
// generate scripts
String fileName = String.format("%s/%s_node.%s",
taskExecutionContext.getExecutePath(),
taskExecutionContext.getTaskAppId(), OSUtils.isWindows() ? "bat" : "sh");
Path path = new File(fileName).toPath();
if (Files.exists(path)) {
return fileName;
}
String script = shellParameters.getRawScript().replaceAll("\\r\\n", "\n");
// combining local and global parameters
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(), taskExecutionContext.getDefinedParams(),
shellParameters.getLocalParametersMap(), shellParameters.getLocalParametersMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()), CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime()); taskExecutionContext.getScheduleTime());
if (paramsMap != null){
script = ParameterUtils.convertParameterPlaceholders(script, ParamUtils.convert(paramsMap));
}
// new
// replace variable TIME with $[YYYYmmddd...] in shell file when history run job and batch complement job
if (paramsMap != null) {
if (taskExecutionContext.getScheduleTime() != null) {
String dateTime = DateUtils.format(taskExecutionContext.getScheduleTime(), Constants.PARAMETER_FORMAT_TIME);
Property p = new Property();
p.setValue(dateTime);
p.setProp(Constants.PARAMETER_SHECDULE_TIME);
paramsMap.put(Constants.PARAMETER_SHECDULE_TIME, p);
}
script = ParameterUtils.convertParameterPlaceholders2(script, ParamUtils.convert(paramsMap));
}
shellParameters.setRawScript(script); // replace variable TIME with $[YYYYmmddd...] in shell file when history run job and batch complement job
if (paramsMap != null) {
if (taskExecutionContext.getScheduleTime() != null) {
String dateTime = DateUtils.format(taskExecutionContext.getScheduleTime(), Constants.PARAMETER_FORMAT_TIME);
Property p = new Property();
p.setValue(dateTime);
p.setProp(Constants.PARAMETER_SHECDULE_TIME);
paramsMap.put(Constants.PARAMETER_SHECDULE_TIME, p);
}
}
logger.info("raw script : {}", shellParameters.getRawScript()); script = ParameterUtils.convertParameterPlaceholders2(script, ParamUtils.convert(paramsMap));
logger.info("task execute path : {}", taskExecutionContext.getExecutePath());
Set<PosixFilePermission> perms = PosixFilePermissions.fromString(Constants.RWXR_XR_X); shellParameters.setRawScript(script);
FileAttribute<Set<PosixFilePermission>> attr = PosixFilePermissions.asFileAttribute(perms);
if (OSUtils.isWindows()) { logger.info("raw script : {}", shellParameters.getRawScript());
Files.createFile(path); logger.info("task execute path : {}", taskExecutionContext.getExecutePath());
} else {
Files.createFile(path, attr);
}
Files.write(path, shellParameters.getRawScript().getBytes(), StandardOpenOption.APPEND); Set<PosixFilePermission> perms = PosixFilePermissions.fromString(Constants.RWXR_XR_X);
FileAttribute<Set<PosixFilePermission>> attr = PosixFilePermissions.asFileAttribute(perms);
return fileName; if (OSUtils.isWindows()) {
} Files.createFile(path);
} else {
Files.createFile(path, attr);
}
@Override Files.write(path, shellParameters.getRawScript().getBytes(), StandardOpenOption.APPEND);
public AbstractParameters getParameters() {
return shellParameters; return fileName;
} }
@Override
public AbstractParameters getParameters() {
return shellParameters;
}
} }

Write Preview
Loading…
Cancel
Save