Optional task execution log indexing (#1465)

* Make tasklog indexing optional

* Create task log index only if it is enabled

* Optional task execution logs

core/src/main/java/com/netflix/conductor/core/config/Configuration.java

@@ -73,6 +73,9 @@ public interface Configuration {
     String IGNORE_LOCKING_EXCEPTIONS_PROPERTY_NAME = "workflow.decider.locking.exceptions.ignore";
     boolean IGNORE_LOCKING_EXCEPTIONS_DEFAULT_VALUE = false;
 
+    String TASKEXECLOG_INDEXING_ENABLED_PROPERTY_NAME = "workflow.taskExecLog.indexing.enabled";
+    boolean TASKEXECLOG_INDEXING_ENABLED_DEFAULT_VALUE = true;
+
     //TODO add constants for input/output external payload related properties.
 
     default DB getDB() {
@@ -102,6 +105,13 @@ default boolean enableWorkflowExecutionLock() {
         return getBooleanProperty(EXECUTION_LOCK_ENABLED_PROPERTY_NAME, EXECUTION_LOCK_ENABLED_DEFAULT_VALUE);
     }
 
+    /**
+     * @return if true(default), enables task execution log indexing
+     */
+    default boolean isTaskExecLogIndexingEnabled() {
+        return getBooleanProperty(TASKEXECLOG_INDEXING_ENABLED_PROPERTY_NAME, TASKEXECLOG_INDEXING_ENABLED_DEFAULT_VALUE);
+    }
+
     /**
      * @return time frequency in seconds, at which the workflow sweeper should run to evaluate running workflows.
      */

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

@@ -29,6 +29,7 @@
 import com.netflix.conductor.dao.RateLimitingDao;
 import com.netflix.conductor.metrics.Monitors;
 import java.io.IOException;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -400,10 +401,13 @@ public boolean exceedsRateLimitPerFrequency(Task task, TaskDef taskDef) {
     }
 
     public void addTaskExecLog(List<TaskExecLog> logs) {
-        if (config.enableAsyncIndexing()) {
-            indexDAO.asyncAddTaskExecutionLogs(logs);
-        } else {
-            indexDAO.addTaskExecutionLogs(logs);
+        if (config.isTaskExecLogIndexingEnabled()) {
+            if (config.enableAsyncIndexing()) {
+                indexDAO.asyncAddTaskExecutionLogs(logs);
+            }
+            else {
+                indexDAO.addTaskExecutionLogs(logs);
+            }
         }
     }
 
@@ -420,7 +424,7 @@ public SearchResult<String> searchTasks(String query, String freeText, int start
     }
 
     public List<TaskExecLog> getTaskExecutionLogs(String taskId) {
-        return indexDAO.getTaskExecutionLogs(taskId);
+        return config.isTaskExecLogIndexingEnabled() ? indexDAO.getTaskExecutionLogs(taskId) : Collections.emptyList();
     }
 
     class DelayWorkflowUpdate implements Runnable {