Set size limit for task exec log and add metrics (#3096)

* Set size limit for task exec log and add metrics * refactor
Netflix · Jul 8, 2022 · 036c0b2 · 036c0b2
1 parent 7882c61
commit 036c0b2
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 1 deletion.
diff --git a/core/src/main/java/com/netflix/conductor/core/config/ConductorProperties.java b/core/src/main/java/com/netflix/conductor/core/config/ConductorProperties.java
@@ -210,6 +210,9 @@ public class ConductorProperties {
     @DataSizeUnit(DataUnit.KILOBYTES)
     private DataSize maxWorkflowVariablesPayloadSizeThreshold = DataSize.ofKilobytes(256L);
 
+    /** Used to limit the size of task execution logs. */
+    private int taskExecLogSizeLimit = 10;
+
     public String getStack() {
         return stack;
     }
@@ -510,6 +513,14 @@ public void setMaxWorkflowVariablesPayloadSizeThreshold(
         this.maxWorkflowVariablesPayloadSizeThreshold = maxWorkflowVariablesPayloadSizeThreshold;
     }
 
+    public int getTaskExecLogSizeLimit() {
+        return taskExecLogSizeLimit;
+    }
+
+    public void setTaskExecLogSizeLimit(int taskExecLogSizeLimit) {
+        this.taskExecLogSizeLimit = taskExecLogSizeLimit;
+    }
+
     /**
      * @return Returns all the configurations in a map.
      */

diff --git a/core/src/main/java/com/netflix/conductor/core/dal/ExecutionDAOFacade.java b/core/src/main/java/com/netflix/conductor/core/dal/ExecutionDAOFacade.java
@@ -572,7 +572,17 @@ public boolean exceedsRateLimitPerFrequency(TaskModel task, TaskDef taskDef) {
     }
 
     public void addTaskExecLog(List<TaskExecLog> logs) {
-        if (properties.isTaskExecLogIndexingEnabled()) {
+        if (properties.isTaskExecLogIndexingEnabled() && !logs.isEmpty()) {
+            Monitors.recordTaskExecLogSize(logs.size());
+            int taskExecLogSizeLimit = properties.getTaskExecLogSizeLimit();
+            if (logs.size() > taskExecLogSizeLimit) {
+                LOGGER.warn(
+                        "Task Execution log size: {} for taskId: {} exceeds the limit: {}",
+                        logs.size(),
+                        logs.get(0).getTaskId(),
+                        taskExecLogSizeLimit);
+                logs = logs.stream().limit(taskExecLogSizeLimit).collect(Collectors.toList());
+            }
             if (properties.isAsyncIndexingEnabled()) {
                 indexDAO.asyncAddTaskExecutionLogs(logs);
             } else {

diff --git a/core/src/main/java/com/netflix/conductor/metrics/Monitors.java b/core/src/main/java/com/netflix/conductor/metrics/Monitors.java
@@ -569,4 +569,8 @@ public static void recordEventQueuePollSize(String queueType, int val) {
     public static void recordQueueMessageRepushFromRepairService(String queueName) {
         counter(classQualifier, "queue_message_repushed", "queueName", queueName);
     }
+
+    public static void recordTaskExecLogSize(int val) {
+        gauge(classQualifier, "task_exec_log_size", val);
+    }
 }