Add total timeout backend

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

@@ -628,6 +628,7 @@ Optional<TaskModel> retry(
         rescheduled.setScheduledTime(0);
         rescheduled.setStartTime(0);
         rescheduled.setEndTime(0);
+        rescheduled.setFirstStartTime(task.getFirstStartTime());
         rescheduled.setWorkerId(null);
 
         if (StringUtils.isNotBlank(task.getExternalInputPayloadStoragePath())) {
@@ -710,20 +711,26 @@ void checkTaskTimeout(TaskDef taskDef, TaskModel task) {
         }
 
         long timeout = 1000L * taskDef.getTimeoutSeconds();
+        long totalTaskTimeout = 1000L * taskDef.getTotalTimeoutSeconds();
         long now = System.currentTimeMillis();
         long elapsedTime =
                 now - (task.getStartTime() + ((long) task.getStartDelayInSeconds() * 1000L));
 
-        if (elapsedTime < timeout) {
+        long elapsedTimeFromFirstTaskExecution =
+                now - (task.getFirstStartTime() + ((long) task.getStartDelayInSeconds() * 1000L));
+
+        if (elapsedTimeFromFirstTaskExecution < totalTaskTimeout && elapsedTime < timeout) {
             return;
         }
 
         String reason =
                 String.format(
                         "Task timed out after %d seconds. Timeout configured as %d seconds. "
+                                + "Total Task Timeout configured as %d seconds. "
                                 + "Timeout policy configured to %s",
                         elapsedTime / 1000L,
                         taskDef.getTimeoutSeconds(),
+                        taskDef.getTotalTimeoutSeconds(),
                         taskDef.getTimeoutPolicy().name());
         timeoutTaskWithTimeoutPolicy(reason, taskDef, task);
     }
@@ -773,6 +780,10 @@ void timeoutTaskWithTimeoutPolicy(String reason, TaskDef taskDef, TaskModel task
             case RETRY:
                 task.setStatus(TIMED_OUT);
                 task.setReasonForIncompletion(reason);
+                if (task.isTotalTimeOutExpired()) {
+                    throw new TerminateWorkflowException(
+                            reason, WorkflowModel.Status.TIMED_OUT, task);
+                }
                 return;
             case TIME_OUT_WF:
                 task.setStatus(TIMED_OUT);

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

@@ -442,6 +442,7 @@ private TaskModel taskToBeRescheduled(WorkflowModel workflow, TaskModel task) {
         taskToBeRetried.setScheduledTime(0);
         taskToBeRetried.setStartTime(0);
         taskToBeRetried.setEndTime(0);
+        taskToBeRetried.setFirstStartTime(task.getFirstStartTime());
         taskToBeRetried.setWorkerId(null);
         taskToBeRetried.setReasonForIncompletion(null);
         taskToBeRetried.setSeq(0);
@@ -1512,7 +1513,9 @@ boolean scheduleTask(WorkflowModel workflow, List<TaskModel> tasks) {
                 if (task.getStatus() != null
                         && !task.getStatus().isTerminal()
                         && task.getStartTime() == 0) {
-                    task.setStartTime(System.currentTimeMillis());
+                    long currTime = System.currentTimeMillis();
+                    task.setStartTime(currTime);
+                    task.setFirstStartTime(currTime);
                 }
                 if (!workflowSystemTask.isAsync()) {
                     try {

core/src/main/java/com/netflix/conductor/model/TaskModel.java

@@ -94,9 +94,6 @@ public boolean isRetriable() {
     /** Time when the task was last updated */
     private long updateTime;
 
-    /** Time when first task started */
-    private long firstStartTime;
-
     private int startDelayInSeconds;
 
     private String retriedTaskId;
@@ -150,6 +147,11 @@ public boolean isRetriable() {
     // Timeout after which the wait task should be marked as completed
     private long waitTimeout;
 
+    // Total task Timeout after which the task should be marked as terminated
+    private long firstStartTime;
+
+    private boolean isTotalTimeOutExpired;
+
     /**
      * Used to note that a sub workflow associated with SUB_WORKFLOW task has an action performed on
      * it directly.
@@ -639,6 +641,22 @@ public void internalizeOutput(Map<String, Object> data) {
         this.outputPayload = data;
     }
 
+    public long getFirstStartTime() {
+        return firstStartTime;
+    }
+
+    public void setFirstStartTime(long firstStartTime) {
+        this.firstStartTime = firstStartTime;
+    }
+
+    public boolean isTotalTimeOutExpired() {
+        return isTotalTimeOutExpired;
+    }
+
+    public void setTotalTimeOutExpired(boolean totalTimeOutExpired) {
+        isTotalTimeOutExpired = totalTimeOutExpired;
+    }
+
     @Override
     public String toString() {
         return "TaskModel{"
@@ -741,6 +759,9 @@ public String toString() {
                 + '\''
                 + ", subworkflowChanged="
                 + subworkflowChanged
+                + '\''
+                + ", firstStartTime="
+                + firstStartTime
                 + '}';
     }
 
@@ -794,7 +815,9 @@ && getStatus() == taskModel.getStatus()
                         taskModel.getExternalOutputPayloadStoragePath())
                 && Objects.equals(getExecutionNameSpace(), taskModel.getExecutionNameSpace())
                 && Objects.equals(getIsolationGroupId(), taskModel.getIsolationGroupId())
-                && Objects.equals(getSubWorkflowId(), taskModel.getSubWorkflowId());
+                && Objects.equals(getSubWorkflowId(), taskModel.getSubWorkflowId())
+                && Objects.equals(getFirstStartTime(), taskModel.getFirstStartTime())
+                && isTotalTimeOutExpired == taskModel.isTotalTimeOutExpired();
     }
 
     @Override
@@ -841,7 +864,9 @@ public int hashCode() {
                 getIsolationGroupId(),
                 getIteration(),
                 getSubWorkflowId(),
-                isSubworkflowChanged());
+                isSubworkflowChanged(),
+                getFirstStartTime(),
+                isTotalTimeOutExpired);
     }
 
     public Task toTask() {

core/src/main/java/com/netflix/conductor/service/ExecutionService.java

@@ -168,6 +168,14 @@ public List<Task> poll(
                 taskModel.setStatus(TaskModel.Status.IN_PROGRESS);
                 if (taskModel.getStartTime() == 0) {
                     taskModel.setStartTime(System.currentTimeMillis());
+
+                    // Use previous task's start time for retries, otherwise use current time
+                    if (taskModel.getRetryCount() != 0) {
+                        taskModel.setFirstStartTime(taskModel.getFirstStartTime());
+                    } else {
+                        taskModel.setFirstStartTime(System.currentTimeMillis());
+                    }
+
                     Monitors.recordQueueWaitTime(
                             taskModel.getTaskDefName(), taskModel.getQueueWaitTime());
                 }