Staging debug

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

@@ -291,6 +291,10 @@ public String updateWorkflow(WorkflowModel workflowModel) {
             workflowModel.setEndTime(System.currentTimeMillis());
         }
         externalizeWorkflowData(workflowModel);
+        LOGGER.debug(
+                "Updating workflow: {} in the ExecutionFacadeDao with status {} ",
+                workflowModel.getWorkflowId(),
+                workflowModel.getStatus());
         executionDAO.updateWorkflow(workflowModel);
         if (properties.isAsyncIndexingEnabled()) {
             if (workflowModel.getStatus().isTerminal()
@@ -322,6 +326,10 @@ public String updateWorkflow(WorkflowModel workflowModel) {
         } else {
             indexDAO.indexWorkflow(new WorkflowSummary(workflowModel.toWorkflow()));
         }
+        LOGGER.debug(
+                "Updated workflow: {} in the ExecutionFacadeDao with status {} ",
+                workflowModel.getWorkflowId(),
+                workflowModel.getStatus());
         return workflowModel.getWorkflowId();
     }
 

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

@@ -847,6 +847,7 @@ public List<TaskModel> getTasksToBeScheduled(
                                                 || runningTask.getStatus().isTerminal())
                         .map(TaskModel::getReferenceTaskName)
                         .collect(Collectors.toList());
+        LOGGER.info("DeciderService getTasksToBeScheduled tasksInWorkflow {}", tasksInWorkflow);
 
         String taskId = idGenerator.generate();
         TaskMapperContext taskMapperContext =
@@ -866,6 +867,12 @@ public List<TaskModel> getTasksToBeScheduled(
         // fork.
         // A new task must only be scheduled if a task, with the same reference name is not already
         // in this workflow instance
+        LOGGER.info("DeciderService getTasksToBeScheduled filtered tasks by referenceName {}", taskMappers
+                .getOrDefault(type, taskMappers.get(USER_DEFINED.name()))
+                .getMappedTasks(taskMapperContext)
+                .stream()
+                .filter(task -> !tasksInWorkflow.contains(task.getReferenceTaskName()))
+                .collect(Collectors.toList()));
         return taskMappers
                 .getOrDefault(type, taskMappers.get(USER_DEFINED.name()))
                 .getMappedTasks(taskMapperContext)

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

@@ -556,6 +556,7 @@ WorkflowModel completeWorkflow(WorkflowModel workflow) {
 
         executionLockService.releaseLock(workflow.getWorkflowId());
         executionLockService.deleteLock(workflow.getWorkflowId());
+        LOGGER.debug("Completed workflow execution && released all the locks for {}", workflow.getWorkflowId());
         return workflow;
     }
 

core/src/main/java/com/netflix/conductor/core/reconciliation/WorkflowReconciler.java

@@ -94,6 +94,9 @@ public void pollAndSweep() {
 
     private void recordQueueDepth() {
         int currentQueueSize = queueDAO.getSize(DECIDER_QUEUE);
+        LOGGER.debug(
+                "recordQueueDepth currentQueueSize processed from reconciler ",
+                DECIDER_QUEUE + "__" + currentQueueSize);
         Monitors.recordGauge(DECIDER_QUEUE, currentQueueSize);
     }
 }

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

@@ -84,13 +84,15 @@ public void reportMetrics() {
                 refreshCounter = metadataRefreshInterval;
             }
 
-            getPendingWorkflowToOwnerAppMap(workflowDefs)
-                    .forEach(
-                            (workflowName, ownerApp) -> {
-                                long count =
-                                        executionDAOFacade.getPendingWorkflowCount(workflowName);
-                                Monitors.recordRunningWorkflows(count, workflowName, ownerApp);
-                            });
+            // Commented out as we don't have use case for Pending Workflows as of now and hence we don't have pending workflow implementation from
+            // scylla persistence.
+//            getPendingWorkflowToOwnerAppMap(workflowDefs)
+//                    .forEach(
+//                            (workflowName, ownerApp) -> {
+//                                long count =
+//                                        executionDAOFacade.getPendingWorkflowCount(workflowName);
+//                                Monitors.recordRunningWorkflows(count, workflowName, ownerApp);
+//                            });
 
             taskDefs.forEach(
                     taskDef -> {

docker/server/config/config-conductor-server-bg.properties

@@ -0,0 +1,8 @@
+# Servers background workers configuration
+conductor.system-task-workers.enabled=false
+conductor.app.isolated-system-task-enabled=false
+conductor.workflow-reconciler.enabled=true
+conductor.workflow-monitor.enabled=true
+conductor.workflow-repair-service.enabled=true
+conductor.workflow-monitor.metadata-refresh-interval=200
+conductor.sweep-frequency.millis=1000

docker/server/config/config-conductor-server-main.properties

@@ -0,0 +1,6 @@
+# Servers background workers configuration
+conductor.system-task-workers.enabled=true
+conductor.app.isolated-system-task-enabled=true
+conductor.workflow-reconciler.enabled=false
+conductor.workflow-monitor.enabled=false
+conductor.workflow-repair-service.enabled=false

docker/server/config/config-conductor-server.properties

@@ -5,16 +5,12 @@ conductor.queue.type=redis_standalone
 conductor.redis.hosts=${REDIS_HOST}:${REDIS_PORT}:${REDIS_RACK_HOSTED_ZONE}:${REDIS_PASSWORD}
 conductor.redis.workflowNamespacePrefix=${REDIS_WORKFLOW_NS_PREFIX}
 conductor.redis.queueNamespacePrefix=${REDIS_QUEUE_NS_PREFIX}
-#conductor.redis.queuesNonQuorumPort=6379
 
 conductor.app.stack=staging
 
 # Database persistence type.
 conductor.db.type=scylla
 
-#spring.data.cassandra.keyspace-name=conductor
-#spring.data.cassandra.contact-points=cassandra
-#spring.data.cassandra.port=9042
 
 conductor.scylla.hostAddress=${SCYLLA_HOST}
 conductor.scylla.password=${SCYLLA_PASSWORD}
@@ -26,23 +22,14 @@ conductor.scylla.replicationFactorValue=${SCYLLA_REPLICATIONFACTOR}
 
 conductor.workflow-status-listener.type=archive
 
-conductor.system-task-workers.enabled=true
+conductor.workflow-status-listener.enabled=false
+conductor.default-event-processor.enabled=false
+conductor.event-queues.default.enabled=false
+conductor.default-event-queue-processor.enabled=true
 
-# Hikari pool sizes are -1 by default and prevent startup
-#spring.datasource.hikari.maximum-pool-size=10
-#spring.datasource.hikari.minimum-idle=2
+# Conditional configuration based on process-name(main/bg) property from k8s
+spring.config.import=optional:classpath:/config-conductor-server-${PROCESS_NAME}.properties
 
-# Elastic search instance indexing is enabled.
-conductor.indexing.enabled=false
-
-# Transport address to elasticsearch
-#conductor.elasticsearch.url=http://es:9200
-
-# Name of the elasticsearch cluster
-#conductor.elasticsearch.indexName=conductor
-
-# Load sample kitchen sink workflow
-loadSample=false
 
 # Pushing Prom Metrics
 conductor.metrics-prometheus.enabled = true

docker/server/config/config-scylla.properties

@@ -25,9 +25,7 @@ conductor.scylla.username=scylla
 conductor.scylla.password=scylla
 
 conductor.workflow-status-listener.type=archive
-
 conductor.system-task-workers.enabled=true
-
 # Hikari pool sizes are -1 by default and prevent startup
 #spring.datasource.hikari.maximum-pool-size=10
 #spring.datasource.hikari.minimum-idle=2

scylla-persistence/src/main/java/com/netflix/conductor/scylla/dao/ScyllaExecutionDAO.java

@@ -542,12 +542,29 @@ public String updateWorkflow(WorkflowModel workflow) {
             recordCassandraDaoRequests("updateWorkflow", "n/a", workflow.getWorkflowName());
             recordCassandraDaoPayloadSize(
                     "updateWorkflow", payload.length(), "n/a", workflow.getWorkflowName());
-            session.execute(
-                    updateWorkflowStatement.bind(
-                            payload, UUID.fromString(workflow.getWorkflowId()),correlationId));
+            WorkflowModel prevWorkflow = getWorkflow(workflow.getWorkflowId(), false);
+            LOGGER.debug(
+                    "prevWorkflow status updated for workflow_id {} is {}",
+                    prevWorkflow.getWorkflowId(),
+                    prevWorkflow.getStatus());
+            // Added this change to update the workflow only if it is not completed, Rerun cannot be done on completed workflows
+            if (!prevWorkflow.getStatus().equals(WorkflowModel.Status.COMPLETED)) {
+                session.execute(
+                        updateWorkflowStatement.bind(
+                                payload, UUID.fromString(workflow.getWorkflowId()), correlationId));
+            } else {
+                LOGGER.info(
+                        "Workflow {} is already completed. Not updating the workflow current status {}",
+                        workflow.getWorkflowId(), workflow.getStatus());
+                return workflow.getWorkflowId();
+            }
+            LOGGER.debug(
+                    "Workflow status updated for workflow_id {} is {}",
+                    workflow.getWorkflowId(),
+                    workflow.getStatus());
             workflow.setTasks(tasks);
             return workflow.getWorkflowId();
-        } catch (DriverException e) {
+        } catch (Exception e) {
             Monitors.error(CLASS_NAME, "updateWorkflow");
             String errorMsg =
                     String.format("Failed to update workflow: %s", workflow.getWorkflowId());

server/src/main/resources/application.properties

@@ -126,3 +126,5 @@ conductor.workflow-execution-lock.type=noop_lock
 # Additional modules for metrics collection exposed to Datadog (optional)
 management.metrics.export.datadog.enabled=${conductor.metrics-datadog.enabled:false}
 management.metrics.export.datadog.api-key=${conductor.metrics-datadog.api-key:}
+
+#logging.level.root=DEBUG