if (wfJob.getStatus() == WorkflowJob.Status.PREP) {
try {
completed = workflowInstance.start();
}
catch (WorkflowException e) {
throw new CommandException(e);
}
wfJob.setStatus(WorkflowJob.Status.RUNNING);
wfJob.setStartTime(new Date());
wfJob.setWorkflowInstance(workflowInstance);
// 1. Add SLA status event for WF-JOB with status STARTED
// 2. Add SLA registration events for all WF_ACTIONS
SLADbXOperations.writeStausEvent(wfJob.getSlaXml(), jobId, Status.STARTED, SlaAppType.WORKFLOW_JOB);
writeSLARegistrationForAllActions(workflowInstance.getApp().getDefinition(), wfJob.getUser(), wfJob
.getGroup(), wfJob.getConf());
queue(new NotificationXCommand(wfJob));
}
else {
throw new CommandException(ErrorCode.E0801, wfJob.getId());
}
}
else {
String skipVar = workflowInstance.getVar(wfAction.getName() + WorkflowInstance.NODE_VAR_SEPARATOR
+ ReRunXCommand.TO_SKIP);
if (skipVar != null) {
skipAction = skipVar.equals("true");
}
try {
completed = workflowInstance.signal(wfAction.getExecutionPath(), wfAction.getSignalValue());
}
catch (WorkflowException e) {
throw new CommandException(e);
}
wfJob.setWorkflowInstance(workflowInstance);
wfAction.resetPending();
if (!skipAction) {
wfAction.setTransition(workflowInstance.getTransition(wfAction.getName()));
queue(new NotificationXCommand(wfJob, wfAction));
}
try {
jpaService.execute(new WorkflowActionUpdateJPAExecutor(wfAction));
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
}
if (completed) {
try {
for (String actionToKillId : WorkflowStoreService.getActionsToKill(workflowInstance)) {
WorkflowActionBean actionToKill;
actionToKill = jpaService.execute(new WorkflowActionGetJPAExecutor(actionToKillId));
actionToKill.setPending();
actionToKill.setStatus(WorkflowActionBean.Status.KILLED);
jpaService.execute(new WorkflowActionUpdateJPAExecutor(actionToKill));
queue(new ActionKillXCommand(actionToKill.getId(), actionToKill.getType()));
}
for (String actionToFailId : WorkflowStoreService.getActionsToFail(workflowInstance)) {
WorkflowActionBean actionToFail = jpaService.execute(new WorkflowActionGetJPAExecutor(
actionToFailId));
actionToFail.resetPending();
actionToFail.setStatus(WorkflowActionBean.Status.FAILED);
queue(new NotificationXCommand(wfJob, actionToFail));
SLADbXOperations.writeStausEvent(wfAction.getSlaXml(), wfAction.getId(), Status.FAILED,
SlaAppType.WORKFLOW_ACTION);
jpaService.execute(new WorkflowActionUpdateJPAExecutor(actionToFail));
}
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
wfJob.setStatus(WorkflowJob.Status.valueOf(workflowInstance.getStatus().toString()));
wfJob.setEndTime(new Date());
wfJob.setWorkflowInstance(workflowInstance);
Status slaStatus = Status.SUCCEEDED;
switch (wfJob.getStatus()) {
case SUCCEEDED:
slaStatus = Status.SUCCEEDED;
break;
case KILLED:
slaStatus = Status.KILLED;
break;
case FAILED:
slaStatus = Status.FAILED;
break;
default: // TODO SUSPENDED
break;
}
SLADbXOperations.writeStausEvent(wfJob.getSlaXml(), jobId, slaStatus, SlaAppType.WORKFLOW_JOB);
queue(new NotificationXCommand(wfJob));
if (wfJob.getStatus() == WorkflowJob.Status.SUCCEEDED) {
InstrumentUtils.incrJobCounter(INSTR_SUCCEEDED_JOBS_COUNTER_NAME, 1, getInstrumentation());
}
// output message for Kill node
if (wfAction != null) { // wfAction could be a no-op job
NodeDef nodeDef = workflowInstance.getNodeDef(wfAction.getExecutionPath());
if (nodeDef != null && nodeDef instanceof KillNodeDef) {
boolean isRetry = false;
boolean isUserRetry = false;
ActionExecutorContext context = new ActionXCommand.ActionExecutorContext(wfJob, wfAction, isRetry,
isUserRetry);
try {
String tmpNodeConf = nodeDef.getConf();
String actionConf = context.getELEvaluator().evaluate(tmpNodeConf, String.class);
LOG.debug("Try to resolve KillNode message for jobid [{0}], actionId [{1}], before resolve [{2}], after resolve [{3}]",
jobId, actionId, tmpNodeConf, actionConf);
if (wfAction.getErrorCode() != null) {
wfAction.setErrorInfo(wfAction.getErrorCode(), actionConf);
}
else {
wfAction.setErrorInfo(ErrorCode.E0729.toString(), actionConf);
}
jpaService.execute(new WorkflowActionUpdateJPAExecutor(wfAction));
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
catch (Exception ex) {
LOG.warn("Exception in SignalXCommand ", ex.getMessage(), ex);
throw new CommandException(ErrorCode.E0729, wfAction.getName(), ex);
}
}
}
}
else {
for (WorkflowActionBean newAction : WorkflowStoreService.getStartedActions(workflowInstance)) {
String skipVar = workflowInstance.getVar(newAction.getName() + WorkflowInstance.NODE_VAR_SEPARATOR
+ ReRunXCommand.TO_SKIP);
boolean skipNewAction = false;
if (skipVar != null) {
skipNewAction = skipVar.equals("true");
}
try {
if (skipNewAction) {
WorkflowActionBean oldAction;
oldAction = jpaService.execute(new WorkflowActionGetJPAExecutor(newAction.getId()));
oldAction.setPending();
jpaService.execute(new WorkflowActionUpdateJPAExecutor(oldAction));
queue(new SignalXCommand(jobId, oldAction.getId()));
}
else {
newAction.setPending();
String actionSlaXml = getActionSLAXml(newAction.getName(), workflowInstance.getApp()
.getDefinition(), wfJob.getConf());
newAction.setSlaXml(actionSlaXml);
jpaService.execute(new WorkflowActionInsertJPAExecutor(newAction));
LOG.debug("SignalXCommand: Name: "+ newAction.getName() + ", Id: " +newAction.getId() + ", Authcode:" + newAction.getCred());
queue(new ActionStartXCommand(newAction.getId(), newAction.getType()));
}
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
}
}
try {
jpaService.execute(new WorkflowJobUpdateJPAExecutor(wfJob));
}
catch (JPAExecutorException je) {
throw new CommandException(je);
}
LOG.debug(
"Updated the workflow status to " + wfJob.getId() + " status =" + wfJob.getStatusStr());
if (wfJob.getStatus() != WorkflowJob.Status.RUNNING && wfJob.getStatus() != WorkflowJob.Status.SUSPENDED) {
// update coordinator action