Merge branch 'dev'

Author: pctreddy

.gitignore

@@ -14,4 +14,5 @@ ui/.settings
 .settings
 dump.rdb
 .idea
-out/
+*.iml
+out/

build.gradle

@@ -10,7 +10,7 @@ buildscript {
     }
 }
 plugins {
-    id 'nebula.netflixoss' version '3.6.0'
+    id 'nebula.netflixoss' version '5.0.0'
 }
 
 // Establish version and status

client/go/httpclient/httpclient.go

@@ -49,7 +49,6 @@ func (c *HttpClient) logResponse(statusCode string, response string) {
 }
 
 func genParamString(paramMap map[string]string) string {
-
     if paramMap == nil || len(paramMap) == 0 {
         return ""
     }
@@ -78,7 +77,6 @@ func (c *HttpClient) httpRequest(url string, requestType string, headers map[str
     if err != nil {
         return "", err
     }
-
     // Default Headers
     for key, value := range c.Headers {
         req.Header.Set(key, value)

common/src/main/java/com/netflix/conductor/common/run/Workflow.java

@@ -16,9 +16,11 @@
 package com.netflix.conductor.common.run;
 
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import com.netflix.conductor.common.metadata.Auditable;
 import com.netflix.conductor.common.metadata.tasks.Task;
@@ -50,13 +52,13 @@ public boolean isSuccessful(){
 	private WorkflowStatus status = WorkflowStatus.RUNNING;
 
 	private long endTime;
-	
+
 	private String workflowId;
 
 	private String parentWorkflowId;
-	
+
 	private String parentWorkflowTaskId;
-	
+
 	private List<Task> tasks = new LinkedList<>();
 
 	private Map<String, Object> input = new HashMap<>();
@@ -79,6 +81,8 @@ public boolean isSuccessful(){
 
 	private Map<String, String> taskToDomain = new HashMap<>();
 
+	private Set<String> failedReferenceTaskNames = new HashSet<>();
+
 	public Workflow(){
 
 	}
@@ -301,7 +305,15 @@ public String getEvent() {
 	public void setEvent(String event) {
 		this.event = event;
 	}
-	
+
+	public Set<String> getFailedReferenceTaskNames() {
+		return failedReferenceTaskNames;
+	}
+
+	public void setFailedReferenceTaskNames(Set<String> failedReferenceTaskNames) {
+		this.failedReferenceTaskNames = failedReferenceTaskNames;
+	}
+
 	@Override
 	public String toString() {
 		return workflowType + "." + version + "/" + workflowId + "." + status; 

common/src/main/java/com/netflix/conductor/common/run/WorkflowSummary.java

@@ -21,6 +21,7 @@
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.TimeZone;
+import java.util.stream.Collectors;
 
 import com.netflix.conductor.common.run.Workflow.WorkflowStatus;
 
@@ -60,6 +61,8 @@ public class WorkflowSummary {
 	private long executionTime;
 
 	private String event;
+
+	private String failedReferenceTaskNames = "";
 
 	public WorkflowSummary() {
 
@@ -90,6 +93,7 @@ public WorkflowSummary(Workflow workflow) {
 			this.executionTime = workflow.getEndTime() - workflow.getStartTime();
 		}
 		this.event = workflow.getEvent();
+		this.failedReferenceTaskNames = workflow.getFailedReferenceTaskNames().stream().collect(Collectors.joining(","));
 	}
 
 	/**
@@ -193,4 +197,12 @@ public String getEvent() {
 	public void setEvent(String event) {
 		this.event = event;
 	}
+
+	public String getFailedReferenceTaskNames() {
+		return failedReferenceTaskNames;
+	}
+
+	public void setFailedReferenceTaskNames(String failedReferenceTaskNames) {
+		this.failedReferenceTaskNames = failedReferenceTaskNames;
+	}
 }

core/src/main/java/com/netflix/conductor/core/execution/DeciderService.java

@@ -130,6 +130,10 @@ private DeciderOutcome decide(final WorkflowDef def, final Workflow workflow, Li
 			TaskDef taskDef = metadata.getTaskDef(task.getTaskDefName());
 			if(taskDef != null) {
 				checkForTimeout(taskDef, task);
+				// If the task has not been updated for "responseTimeout" then rescheduled it.
+				if(checkForResponseTimeout(taskDef, task)){
+					outcome.tasksToBeRequeued.add(task);
+				}
 			}
 
 			if (!task.getStatus().isSuccessful()) {
@@ -374,6 +378,30 @@ void checkForTimeout(TaskDef taskType, Task task) {
 		return;
 	}
 
+	@VisibleForTesting
+	boolean checkForResponseTimeout(TaskDef taskType, Task task) {
+		
+		if(taskType == null){
+			logger.warn("missing task type " + task.getTaskDefName() + ", workflowId=" + task.getWorkflowInstanceId());
+			return false;
+		}
+		if (task.getStatus().isTerminal() || taskType.getTimeoutSeconds() <= 0 || 
+				!task.getStatus().equals(Status.IN_PROGRESS) || taskType.getResponseTimeoutSeconds() == 0) {
+			return false;
+		}
+
+		long responseTimeout = 1000 * taskType.getResponseTimeoutSeconds();
+		long now = System.currentTimeMillis();
+		long noResponseTime = now - task.getUpdateTime();
+		
+		if (noResponseTime < responseTimeout) {
+			return false;
+		}
+		Monitors.recordTaskResponseTimeout(task.getTaskDefName());
+		
+		return true;
+	}
+	
 	private List<Task> getTasksToBeScheduled(WorkflowDef def, Workflow workflow, WorkflowTask taskToSchedule, int retryCount)  {
 		return getTasksToBeScheduled(def, workflow, taskToSchedule, retryCount, null);
 	}
@@ -672,6 +700,8 @@ public static class DeciderOutcome {
 		List<Task> tasksToBeScheduled = new LinkedList<>();
 
 		List<Task> tasksToBeUpdated = new LinkedList<>();
+
+		List<Task> tasksToBeRequeued = new LinkedList<>();
 
 		boolean isComplete;
 

core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java

@@ -434,8 +434,16 @@ public void updateTask(TaskResult result) throws Exception {
 		if (task.getStatus().isTerminal()) {
 			task.setEndTime(System.currentTimeMillis());
 		}
+
 		edao.updateTask(task);
 
+		//If the task has failed update the failed task reference name in the workflow.
+		//This gives the ability to look at workflow and see what tasks have failed at a high level.
+		if(Status.FAILED.equals(task.getStatus())) {
+			wf.getFailedReferenceTaskNames().add(task.getReferenceTaskName());
+			edao.updateWorkflow(wf);
+		}
+
 		result.getLogs().forEach(tl -> tl.setTaskId(task.getTaskId()));
 		edao.addTaskExecLog(result.getLogs());
 
@@ -511,8 +519,12 @@ public boolean decide(String workflowId) throws Exception {
 			List<Task> tasksToBeScheduled = outcome.tasksToBeScheduled;
 			setTaskDomains(tasksToBeScheduled, workflow);
 			List<Task> tasksToBeUpdated = outcome.tasksToBeUpdated;
+			List<Task> tasksToBeRequeued = outcome.tasksToBeRequeued;
 			boolean stateChanged = false;
 
+			if(!tasksToBeRequeued.isEmpty()){
+				addTaskToQueue(tasksToBeRequeued);
+			}
 			workflow.getTasks().addAll(tasksToBeScheduled);
 			for(Task task : tasksToBeScheduled) {
 				if (SystemTaskType.is(task.getTaskType()) && !task.getStatus().isTerminal()) {

core/src/main/java/com/netflix/conductor/metrics/Monitors.java

@@ -184,6 +184,10 @@ public static void recordTaskTimeout(String taskType) {
 		counter(classQualifier, "task_timeout", "taskType", taskType);
 	}
 
+	public static void recordTaskResponseTimeout(String taskType) {
+		counter(classQualifier, "task_response_timeout", "taskType", taskType);
+	}
+	
 	public static void recordWorkflowTermination(String workflowType, WorkflowStatus status, String ownerApp) {
 		counter(classQualifier, "workflow_failure", "workflowName", workflowType, "status", status.name(), "ownerApp", ""+ownerApp);
 	}

es5-persistence/README.md

@@ -1,6 +1,6 @@
 ## Usage
 
-1. In `server/build.gradle` file,  add `compile project(':conductor-es5-persistence')`  dependencie
+1. In `server/build.gradle` file,  add `compile project(':conductor-es5-persistence')` in dependencies
 1. In `server/ConductorServer.java` file , replace  `import com.netflix.conductor.dao.es.EmbeddedElasticSearch` with `import com.netflix.conductor.dao.es5.es.EmbeddedElasticSearch`
 1. In `server/ServerModule.java` file,  replace  `import com.netflix.conductor.dao.index.ElasticSearchDAO; import com.netflix.conductor.dao.index.ElasticsearchModule` with `import com.netflix.conductor.dao.es5.index.ElasticSearchDAO;  import com.netflix.conductor.dao.es5.index.ElasticsearchModule;`
 1. Config property 'workflow.elasticsearch.cluster.name' , value with your elasticsearch cluster name

gradle/wrapper/gradle-wrapper.properties

@@ -3,4 +3,4 @@ distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-3.1-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.4-bin.zip