Lt config load

cassandra-persistence/src/main/java/com/netflix/conductor/cassandra/dao/CassandraExecutionDAO.java

@@ -361,6 +361,14 @@ public void updateTask(TaskModel task) {
         }
     }
 
+    @Override
+    public void updateTasksInBatch(List<TaskModel> tasks) {
+        TaskModel task = tasks.get(0);
+        LOGGER.debug(
+                "Workflow Task removed from TASKS_IN_PROGRESS_STATUS with tasksInProgressKey: {}, workflowId: {}, taskId: {}, taskType: {}, taskStatus: {} during updateTask",
+                task.getTaskDefName(), task.getWorkflowInstanceId(), task.getTaskId(), task.getTaskType(), task.getStatus().name());
+    }
+
     /**
      *  @method to verify the task status and update the task_in_progress table
      *  also removes if its a terminal task

client/src/main/java/com/netflix/conductor/client/automator/TaskPollExecutor.java

@@ -164,7 +164,7 @@ void pollAndExecute(Worker worker) {
             for (Task task : tasks) {
                 if (Objects.nonNull(task) && StringUtils.isNotBlank(task.getTaskId())) {
                     MetricsContainer.incrementTaskPollCount(taskType, 1);
-                    LOGGER.debug(
+                    LOGGER.info(
                             "Polled task: {} of type: {} in domain: '{}', from worker: {}",
                             task.getTaskId(),
                             taskType,
@@ -236,6 +236,7 @@ void shutdownAndAwaitTermination(ExecutorService executorService, int timeout) {
             };
 
     private Task processTask(Task task, Worker worker, PollingSemaphore pollingSemaphore) {
+        long start = System.currentTimeMillis();
         LOGGER.debug(
                 "Executing task: {} of type: {} in worker: {} at {}",
                 task.getTaskId(),
@@ -251,10 +252,13 @@ private Task processTask(Task task, Worker worker, PollingSemaphore pollingSemap
         } finally {
             pollingSemaphore.complete(1);
         }
+        LOGGER.info("[Conductor] time taken by task {} for WF {} in processTask is {}", task.getTaskId(), task.getWorkflowInstanceId(),
+                (System.currentTimeMillis() - start));
         return task;
     }
 
     private void executeTask(Worker worker, Task task) {
+        long start = System.currentTimeMillis();
         StopWatch stopwatch = new StopWatch();
         stopwatch.start();
         TaskResult result = null;
@@ -292,6 +296,8 @@ private void executeTask(Worker worker, Task task) {
                 worker.getIdentity(),
                 result.getStatus());
         updateTaskResult(updateRetryCount, task, result, worker);
+        LOGGER.info("[Conductor] time taken by task {} for WF {} in executeTask is {}", task.getTaskId(), task.getWorkflowInstanceId(),
+                (System.currentTimeMillis() - start));
     }
 
     private void finalizeTask(Task task, Throwable throwable) {
@@ -334,7 +340,11 @@ private void updateTaskResult(int count, Task task, TaskResult result, Worker wo
 
             retryOperation(
                     (TaskResult taskResult) -> {
+                        long start = System.currentTimeMillis();
                         taskClient.updateTask(taskResult);
+                        LOGGER.info("[Conductor] time taken by task {} for WF {} in updateTaskResult is {}", task.getTaskId(),
+                                task.getWorkflowInstanceId(),
+                                (System.currentTimeMillis() - start));
                         return null;
                     },
                     count,

client/src/main/java/com/netflix/conductor/client/http/TaskClient.java

@@ -236,8 +236,12 @@ private void populateTaskPayloads(Task task) {
      * @param taskResult the {@link TaskResult} of the executed task to be updated.
      */
     public void updateTask(TaskResult taskResult) {
+        long start = System.currentTimeMillis();
         Validate.notNull(taskResult, "Task result cannot be null");
         postForEntityWithRequestOnly("tasks", taskResult);
+        LOGGER.info("[Conductor] time taken by task {} for WF {} in updateTask is {}", taskResult.getTaskId(),
+                taskResult.getWorkflowInstanceId(),
+                (System.currentTimeMillis() - start));
     }
 
     public Optional<String> evaluateAndUploadLargePayload(

core/src/main/java/com/netflix/conductor/core/dal/ExecutionDAOFacade.java

@@ -492,6 +492,7 @@ public long getInProgressTaskCount(String taskDefName) {
      *     payload fails.
      */
     public void updateTask(TaskModel taskModel) {
+        long start = System.currentTimeMillis();
         if (taskModel.getStatus() != null) {
             if (!taskModel.getStatus().isTerminal()
                     || (taskModel.getStatus().isTerminal() && taskModel.getUpdateTime() == 0)) {
@@ -524,10 +525,13 @@ public void updateTask(TaskModel taskModel) {
             LOGGER.error(errorMsg, e);
             throw new TransientException(errorMsg, e);
         }
+        LOGGER.info("[Conductor] [ExecutionDAOFacade] updateTask Time taken for task: {},for workflowInstanceId {} and status {} and time is :{}",
+                taskModel.getTaskId(), taskModel.getWorkflowInstanceId(), taskModel.getStatus(), (System.currentTimeMillis() - start));
     }
 
     public void updateTasks(List<TaskModel> tasks) {
-        tasks.forEach(this::updateTask);
+        executionDAO.updateTasksInBatch(tasks);
+        //tasks.forEach(this::updateTask);
     }
 
     public void removeTask(String taskId) {

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

@@ -13,6 +13,7 @@
 package com.netflix.conductor.core.execution;
 
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
@@ -705,6 +706,7 @@ public WorkflowModel terminateWorkflow(
      * @throws NotFoundException if the Task is not found.
      */
     public void updateTask(TaskResult taskResult) {
+        long start = System.currentTimeMillis();
         if (taskResult == null) {
             throw new IllegalArgumentException("Task object is null");
         } else if (taskResult.isExtendLease()) {
@@ -714,14 +716,42 @@ public void updateTask(TaskResult taskResult) {
 
         String workflowId = taskResult.getWorkflowInstanceId();
         WorkflowModel workflowInstance = executionDAOFacade.getWorkflowModel(workflowId, false);
+        LOGGER.info(
+                "[Conductor] [WorkflowExecutor] workflowInstance Time taken for task: {},for workflowInstanceId {} and status {} "
+                        + "and workerId {} and time is :{}",
+                taskResult.getTaskId(), taskResult.getWorkflowInstanceId(), taskResult.getStatus(), taskResult.getWorkerId(),
+                (System.currentTimeMillis() - start));
+
+        long start1 = System.currentTimeMillis();
+        List<TaskModel> tasks = workflowInstance.getTasks();
+        TaskModel task = null;
+        for (TaskModel t : tasks) {
+            if (t.getTaskId().equalsIgnoreCase(taskResult.getTaskId())) {
+                task = t;
+            }
+            LOGGER.info(
+                    "[Conductor] [WorkflowExecutor] getTaskModel Time taken for task: {},for workflowInstanceId {} and status {} "
+                            + "and workerId {} and time is :{}",
+                    taskResult.getTaskId(), taskResult.getWorkflowInstanceId(), taskResult.getStatus(), taskResult.getWorkerId(),
+                    (System.currentTimeMillis() - start1));
+        }
+        if (task == null) {
+            throw new NotFoundException("No such task found by id: %s", taskResult.getTaskId());
+        }
 
+        /*long start1 = System.currentTimeMillis();
         TaskModel task =
                 Optional.ofNullable(executionDAOFacade.getTaskModel(taskResult.getTaskId()))
                         .orElseThrow(
                                 () ->
                                         new NotFoundException(
                                                 "No such task found by id: %s",
                                                 taskResult.getTaskId()));
+        LOGGER.info(
+                "[Conductor] [WorkflowExecutor] getTaskModel Time taken for task: {},for workflowInstanceId {} and status {} "
+                        + "and workerId {} and time is :{}",
+                taskResult.getTaskId(), taskResult.getWorkflowInstanceId(), taskResult.getStatus(), taskResult.getWorkerId(),
+                (System.currentTimeMillis() - start1));*/
 
         LOGGER.debug("WE updateTask: taskId {} with taskStatus {} belonging to workflowId {} being updated with workflowStatus {}",
                 task.getTaskId(), task.getStatus(), workflowInstance, workflowInstance.getStatus());
@@ -838,9 +868,15 @@ public void updateTask(TaskResult taskResult) {
 
         // Throw a TransientException if below operations fail to avoid workflow inconsistencies.
         try {
+            long start2 = System.currentTimeMillis();
             LOGGER.debug("WE calling SED.updateTask: taskId {} with taskStatus {} belonging to workflowId {} and workflowStatus {}",
                     task.getTaskId(), task.getStatus(), workflowInstance, workflowInstance.getStatus());
             executionDAOFacade.updateTask(task);
+            LOGGER.info(
+                        "[Conductor] [WorkflowExecutor] internal updateTask Time taken for task: {},for workflowInstanceId {} and status {} "
+                            + "and workerId {} and time is :{}",
+                    taskResult.getTaskId(), taskResult.getWorkflowInstanceId(), taskResult.getStatus(), taskResult.getWorkerId(),
+                    (System.currentTimeMillis() - start2));
         } catch (Exception e) {
             String errorMsg =
                     String.format(
@@ -861,7 +897,10 @@ public void updateTask(TaskResult taskResult) {
             LOGGER.error(errorMsg, e);
         }
 
-        taskResult.getLogs().forEach(taskExecLog -> taskExecLog.setTaskId(task.getTaskId()));
+        for(TaskExecLog taskExecLog : taskResult.getLogs()) {
+            taskExecLog.setTaskId(task.getTaskId());
+        }
+        //taskResult.getLogs().forEach(taskExecLog -> taskExecLog.setTaskId(task.getTaskId()));
         executionDAOFacade.addTaskExecLog(taskResult.getLogs());
 
         if (task.getStatus().isTerminal()) {
@@ -873,9 +912,19 @@ public void updateTask(TaskResult taskResult) {
                     task.getTaskDefName(), lastDuration, false, task.getStatus());
         }
 
+        long start3 = System.currentTimeMillis();
         if (!isLazyEvaluateWorkflow(workflowInstance.getWorkflowDefinition(), task)) {
-            decide(workflowId);
+            //decide(workflowId);
+            decide(workflowInstance);
         }
+        LOGGER.info(
+                "[Conductor] [WorkflowExecutor] decide Time taken for task: {},for workflowInstanceId {} and status {} "
+                        + "and workerId {} and time is :{}",
+                taskResult.getTaskId(), taskResult.getWorkflowInstanceId(), taskResult.getStatus(), taskResult.getWorkerId(),
+                (System.currentTimeMillis() - start3));
+        LOGGER.info(
+                "[Conductor] [WorkflowExecutor] complete updateTask Time taken for task: {},for workflowInstanceId {} and status {} and time is :{}",
+                taskResult.getTaskId(), taskResult.getWorkflowInstanceId(), taskResult.getStatus(), (System.currentTimeMillis() - start));
     }
 
     private void notifyTaskStatusListener(TaskModel task) {
@@ -1037,6 +1086,9 @@ public WorkflowModel decide(String workflowId) {
      *     No locking is required or lock is acquired externally
      */
     public WorkflowModel decide(WorkflowModel workflow) {
+        long start1 = System.currentTimeMillis();
+
+        long start2 = System.currentTimeMillis();
         if (workflow.getStatus().isTerminal()) {
             if (!workflow.getStatus().isSuccessful()) {
                 cancelNonTerminalTasks(workflow);
@@ -1047,7 +1099,6 @@ public WorkflowModel decide(WorkflowModel workflow) {
         // we find any sub workflow tasks that have changed
         // and change the workflow/task state accordingly
         adjustStateIfSubWorkflowChanged(workflow);
-
         try {
             DeciderService.DeciderOutcome outcome = deciderService.decide(workflow);
             if (outcome.isComplete) {
@@ -1061,8 +1112,19 @@ public WorkflowModel decide(WorkflowModel workflow) {
 
             tasksToBeScheduled = dedupAndAddTasks(workflow, tasksToBeScheduled);
 
+            LOGGER.info(
+                    "[Conductor] [WorkflowExecutor] [decide] 1st adjustStateIfSubWorkflowChanged Time taken for workflowInstanceId {} "
+                            + "and outcome.isComplete {} and time is :{}",
+                    workflow.getWorkflowId(), outcome.isComplete, (System.currentTimeMillis() - start2));
+
+            start2 = System.currentTimeMillis();
             boolean stateChanged = scheduleTask(workflow, tasksToBeScheduled); // start
+            LOGGER.info(
+                    "[Conductor] [WorkflowExecutor] [decide] 2nd find scheduleTask stateChanged Time taken for workflowInstanceId {} "
+                            + "and stateChanged {} and time is :{}",
+                    workflow.getWorkflowId(), stateChanged, (System.currentTimeMillis() - start2));
 
+            long start3 = System.currentTimeMillis();
             for (TaskModel task : outcome.tasksToBeScheduled) {
                 executionDAOFacade.populateTaskData(task);
                 if (systemTaskRegistry.isSystemTask(task.getTaskType())
@@ -1076,21 +1138,35 @@ public WorkflowModel decide(WorkflowModel workflow) {
                     }
                 }
             }
+            LOGGER.info(
+                    "[Conductor] [WorkflowExecutor] [decide] 3rd loop outcome.tasksToBeScheduled Time taken for workflowInstanceId {} "
+                            + "and stateChanged {} and time is :{}",
+                    workflow.getWorkflowId(), stateChanged, (System.currentTimeMillis() - start3));
 
             if (!outcome.tasksToBeUpdated.isEmpty() || !tasksToBeScheduled.isEmpty()) {
                 executionDAOFacade.updateTasks(tasksToBeUpdated);
             }
+            LOGGER.info(
+                    "[Conductor] [WorkflowExecutor] [decide] 4th scheduleTask and stateChanged Time taken for workflowInstanceId {} "
+                            + "and stateChanged {} and time is :{}",
+                    workflow.getWorkflowId(), stateChanged, (System.currentTimeMillis() - start2));
 
             if (stateChanged) {
                 return decide(workflow);
             }
-
-            if (!outcome.tasksToBeUpdated.isEmpty() || !tasksToBeScheduled.isEmpty()) {
+            start2 = System.currentTimeMillis();
+            boolean flag = !outcome.tasksToBeUpdated.isEmpty() || !tasksToBeScheduled.isEmpty();
+            if (flag) {
                 executionDAOFacade.updateWorkflow(workflow);
             }
-
+            LOGGER.info(
+                    "[Conductor] [WorkflowExecutor] [decide] 5th updateWorkflow and flag Time taken for workflowInstanceId {} "
+                            + "and flag {} and time is :{}",
+                    workflow.getWorkflowId(), flag, (System.currentTimeMillis() - start2));
+            LOGGER.info(
+                    "[Conductor] [WorkflowExecutor] [decide] final Whole decide complete Time taken for workflowInstanceId {} and time is :{}",
+                    workflow.getWorkflowId(), (System.currentTimeMillis() - start1));
             return workflow;
-
         } catch (TerminateWorkflowException twe) {
             LOGGER.info("Execution terminated of workflow: {}", workflow, twe);
             terminate(workflow, twe);
@@ -1443,6 +1519,7 @@ private long getTaskDuration(long s, TaskModel task) {
 
     @VisibleForTesting
     boolean scheduleTask(WorkflowModel workflow, List<TaskModel> tasks) {
+        long start1 = System.currentTimeMillis();
         List<TaskModel> tasksToBeQueued;
         boolean startedSystemTasks = false;
 
@@ -1466,8 +1543,12 @@ boolean scheduleTask(WorkflowModel workflow, List<TaskModel> tasks) {
                     workflow.getWorkflowName(),
                     String.valueOf(workflow.getWorkflowVersion()));
 
+            long start2 = System.currentTimeMillis();
             // Save the tasks in the DAO
             executionDAOFacade.createTasks(tasks);
+            LOGGER.info("[Conductor] [WorkflowExecutor] [decide] [scheduleTask] 1st createTasks Time taken for workflowInstanceId {} " +
+                            "and tasks.size {} and time is :{}", workflow.getWorkflowId(), tasks.size(),
+                    (System.currentTimeMillis() - start2));
 
             List<TaskModel> systemTasks =
                     tasks.stream()
@@ -1506,7 +1587,12 @@ boolean scheduleTask(WorkflowModel workflow, List<TaskModel> tasks) {
                         throw new NonTransientException(errorMsg, e);
                     }
                     startedSystemTasks = true;
+                    start2 = System.currentTimeMillis();
+                    // Save the tasks in the DAO
                     executionDAOFacade.updateTask(task);
+                    LOGGER.info("[Conductor] [WorkflowExecutor] [decide] [scheduleTask] 2nd updateTask Time taken for workflowInstanceId {} " +
+                                    "and tasks.size {} and time is :{}", workflow.getWorkflowId(), tasks.size(),
+                            (System.currentTimeMillis() - start2));
                 } else {
                     tasksToBeQueued.add(task);
                 }
@@ -1538,6 +1624,10 @@ boolean scheduleTask(WorkflowModel workflow, List<TaskModel> tasks) {
             LOGGER.warn(errorMsg, e);
             Monitors.error(CLASS_NAME, "scheduleTask");
         }
+
+        LOGGER.info("[Conductor] [WorkflowExecutor] [decide] [scheduleTask] final scheduleTask Time taken for workflowInstanceId {} " +
+                        "and tasksToBeQueued.size {} and time is :{}", workflow.getWorkflowId(), tasksToBeQueued.size(),
+                (System.currentTimeMillis() - start1));
         return startedSystemTasks;
     }
 

core/src/main/java/com/netflix/conductor/dao/ExecutionDAO.java

@@ -52,6 +52,11 @@ public interface ExecutionDAO {
      */
     void updateTask(TaskModel task);
 
+    /**
+     * @param tasks Task to be updated
+     */
+    void updateTasksInBatch(List<TaskModel> tasks);
+
     /**
      * Checks if the number of tasks in progress for the given taskDef will exceed the limit if the
      * task is scheduled to be in progress (given to the worker or for system tasks start() method

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

@@ -16,6 +16,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 

redis-persistence/src/main/java/com/netflix/conductor/redis/dao/RedisExecutionDAO.java

@@ -260,6 +260,18 @@ public void updateTask(TaskModel task) {
         }
     }
 
+    @Override
+    public void updateTasksInBatch(List<TaskModel> tasks) {
+        TaskModel task = tasks.get(0);
+        LOGGER.debug(
+                "Workflow Task removed from TASKS_IN_PROGRESS_STATUS with tasksInProgressKey: {}, workflowId: {}, taskId: {}, taskType: {}, taskStatus: {} during updateTask",
+                nsKey(IN_PROGRESS_TASKS, task.getTaskDefName()),
+                task.getWorkflowInstanceId(),
+                task.getTaskId(),
+                task.getTaskType(),
+                task.getStatus().name());
+    }
+
     @Override
     public boolean exceedsLimit(TaskModel task) {
         Optional<TaskDef> taskDefinition = task.getTaskDefinition();