[Backport 2.2] Task consumer Integration

distribution/docker/src/docker/config/log4j2.properties

@@ -53,3 +53,13 @@ logger.index_indexing_slowlog.name = index.indexing.slowlog.index
 logger.index_indexing_slowlog.level = trace
 logger.index_indexing_slowlog.appenderRef.index_indexing_slowlog_rolling.ref = index_indexing_slowlog_rolling
 logger.index_indexing_slowlog.additivity = false
+
+appender.task_detailslog_rolling.type = Console
+appender.task_detailslog_rolling.name = task_detailslog_rolling
+appender.task_detailslog_rolling.layout.type = OpenSearchJsonLayout
+appender.task_detailslog_rolling.layout.type_name = task_detailslog
+
+logger.task_detailslog_rolling.name = task.detailslog
+logger.task_detailslog_rolling.level = trace
+logger.task_detailslog_rolling.appenderRef.task_detailslog_rolling.ref = task_detailslog_rolling
+logger.task_detailslog_rolling.additivity = false

distribution/src/config/log4j2.properties

@@ -195,3 +195,40 @@ logger.index_indexing_slowlog.level = trace
 logger.index_indexing_slowlog.appenderRef.index_indexing_slowlog_rolling.ref = index_indexing_slowlog_rolling
 logger.index_indexing_slowlog.appenderRef.index_indexing_slowlog_rolling_old.ref = index_indexing_slowlog_rolling_old
 logger.index_indexing_slowlog.additivity = false
+
+######## Task details log JSON ####################
+appender.task_detailslog_rolling.type = RollingFile
+appender.task_detailslog_rolling.name = task_detailslog_rolling
+appender.task_detailslog_rolling.fileName = ${sys:opensearch.logs.base_path}${sys:file.separator}${sys:opensearch.logs.cluster_name}_task_detailslog.json
+appender.task_detailslog_rolling.filePermissions = rw-r-----
+appender.task_detailslog_rolling.layout.type = OpenSearchJsonLayout
+appender.task_detailslog_rolling.layout.type_name = task_detailslog
+appender.task_detailslog_rolling.layout.opensearchmessagefields=taskId,type,action,description,start_time_millis,resource_stats,metadata
+
+appender.task_detailslog_rolling.filePattern = ${sys:opensearch.logs.base_path}${sys:file.separator}${sys:opensearch.logs.cluster_name}_task_detailslog-%i.json.gz
+appender.task_detailslog_rolling.policies.type = Policies
+appender.task_detailslog_rolling.policies.size.type = SizeBasedTriggeringPolicy
+appender.task_detailslog_rolling.policies.size.size = 1GB
+appender.task_detailslog_rolling.strategy.type = DefaultRolloverStrategy
+appender.task_detailslog_rolling.strategy.max = 4
+#################################################
+######## Task details log -  old style pattern ####
+appender.task_detailslog_rolling_old.type = RollingFile
+appender.task_detailslog_rolling_old.name = task_detailslog_rolling_old
+appender.task_detailslog_rolling_old.fileName = ${sys:opensearch.logs.base_path}${sys:file.separator}${sys:opensearch.logs.cluster_name}_task_detailslog.log
+appender.task_detailslog_rolling_old.filePermissions = rw-r-----
+appender.task_detailslog_rolling_old.layout.type = PatternLayout
+appender.task_detailslog_rolling_old.layout.pattern = [%d{ISO8601}][%-5p][%-25c{1.}] [%node_name]%marker %m%n
+
+appender.task_detailslog_rolling_old.filePattern = ${sys:opensearch.logs.base_path}${sys:file.separator}${sys:opensearch.logs.cluster_name}_task_detailslog-%i.log.gz
+appender.task_detailslog_rolling_old.policies.type = Policies
+appender.task_detailslog_rolling_old.policies.size.type = SizeBasedTriggeringPolicy
+appender.task_detailslog_rolling_old.policies.size.size = 1GB
+appender.task_detailslog_rolling_old.strategy.type = DefaultRolloverStrategy
+appender.task_detailslog_rolling_old.strategy.max = 4
+#################################################
+logger.task_detailslog_rolling.name = task.detailslog
+logger.task_detailslog_rolling.level = trace
+logger.task_detailslog_rolling.appenderRef.task_detailslog_rolling.ref = task_detailslog_rolling
+logger.task_detailslog_rolling.appenderRef.task_detailslog_rolling_old.ref = task_detailslog_rolling_old
+logger.task_detailslog_rolling.additivity = false

distribution/tools/upgrade-cli/src/test/java/org/opensearch/upgrade/ImportLog4jPropertiesTaskTests.java

@@ -67,7 +67,7 @@ public void testImportLog4jPropertiesTask() throws IOException {
         Properties properties = new Properties();
         properties.load(Files.newInputStream(taskInput.getOpenSearchConfig().resolve(ImportLog4jPropertiesTask.LOG4J_PROPERTIES)));
         assertThat(properties, is(notNullValue()));
-        assertThat(properties.entrySet(), hasSize(137));
+        assertThat(properties.entrySet(), hasSize(165));
         assertThat(properties.get("appender.rolling.layout.type"), equalTo("OpenSearchJsonLayout"));
         assertThat(
             properties.get("appender.deprecation_rolling.fileName"),

distribution/tools/upgrade-cli/src/test/resources/config/log4j2.properties

@@ -176,3 +176,38 @@ logger.index_indexing_slowlog.level = trace
 logger.index_indexing_slowlog.appenderRef.index_indexing_slowlog_rolling.ref = index_indexing_slowlog_rolling
 logger.index_indexing_slowlog.appenderRef.index_indexing_slowlog_rolling_old.ref = index_indexing_slowlog_rolling_old
 logger.index_indexing_slowlog.additivity = false
+
+######## Task details log JSON ####################
+appender.task_detailslog_rolling.type = RollingFile
+appender.task_detailslog_rolling.name = task_detailslog_rolling
+appender.task_detailslog_rolling.fileName = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}_task_detailslog.json
+appender.task_detailslog_rolling.layout.type = ESJsonLayout
+appender.task_detailslog_rolling.layout.type_name = task_detailslog
+appender.task_detailslog_rolling.layout.esmessagefields=taskId,type,action,description,start_time_millis,resource_stats,metadata
+
+appender.task_detailslog_rolling.filePattern = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}_task_detailslog-%i.json.gz
+appender.task_detailslog_rolling.policies.type = Policies
+appender.task_detailslog_rolling.policies.size.type = SizeBasedTriggeringPolicy
+appender.task_detailslog_rolling.policies.size.size = 1GB
+appender.task_detailslog_rolling.strategy.type = DefaultRolloverStrategy
+appender.task_detailslog_rolling.strategy.max = 4
+#################################################
+######## Task details log -  old style pattern ####
+appender.task_detailslog_rolling_old.type = RollingFile
+appender.task_detailslog_rolling_old.name = task_detailslog_rolling_old
+appender.task_detailslog_rolling_old.fileName = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}_task_detailslog.log
+appender.task_detailslog_rolling_old.layout.type = PatternLayout
+appender.task_detailslog_rolling_old.layout.pattern = [%d{ISO8601}][%-5p][%-25c{1.}] [%node_name]%marker %m%n
+
+appender.task_detailslog_rolling_old.filePattern = ${sys:es.logs.base_path}${sys:file.separator}${sys:es.logs.cluster_name}_task_detailslog-%i.log.gz
+appender.task_detailslog_rolling_old.policies.type = Policies
+appender.task_detailslog_rolling_old.policies.size.type = SizeBasedTriggeringPolicy
+appender.task_detailslog_rolling_old.policies.size.size = 1GB
+appender.task_detailslog_rolling_old.strategy.type = DefaultRolloverStrategy
+appender.task_detailslog_rolling_old.strategy.max = 4
+#################################################
+logger.task_detailslog_rolling.name = task.detailslog
+logger.task_detailslog_rolling.level = trace
+logger.task_detailslog_rolling.appenderRef.task_detailslog_rolling.ref = task_detailslog_rolling
+logger.task_detailslog_rolling.appenderRef.task_detailslog_rolling_old.ref = task_detailslog_rolling_old
+logger.task_detailslog_rolling.additivity = false

server/src/main/java/org/opensearch/action/search/SearchShardTask.java

@@ -32,12 +32,14 @@
 
 package org.opensearch.action.search;
 
+import org.opensearch.common.MemoizedSupplier;
 import org.opensearch.search.fetch.ShardFetchSearchRequest;
 import org.opensearch.search.internal.ShardSearchRequest;
 import org.opensearch.tasks.CancellableTask;
 import org.opensearch.tasks.TaskId;
 
 import java.util.Map;
+import java.util.function.Supplier;
 
 /**
  * Task storing information about a currently running search shard request.
@@ -46,9 +48,28 @@
  * @opensearch.internal
  */
 public class SearchShardTask extends CancellableTask {
+    // generating metadata in a lazy way since source can be quite big
+    private final MemoizedSupplier<String> metadataSupplier;
 
     public SearchShardTask(long id, String type, String action, String description, TaskId parentTaskId, Map<String, String> headers) {
+        this(id, type, action, description, parentTaskId, headers, () -> "");
+    }
+
+    public SearchShardTask(
+        long id,
+        String type,
+        String action,
+        String description,
+        TaskId parentTaskId,
+        Map<String, String> headers,
+        Supplier<String> metadataSupplier
+    ) {
         super(id, type, action, description, parentTaskId, headers);
+        this.metadataSupplier = new MemoizedSupplier<>(metadataSupplier);
+    }
+
+    public String getTaskMetadata() {
+        return metadataSupplier.get();
     }
 
     @Override

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -41,6 +41,7 @@
 import org.opensearch.index.ShardIndexingPressureMemoryManager;
 import org.opensearch.index.ShardIndexingPressureSettings;
 import org.opensearch.index.ShardIndexingPressureStore;
+import org.opensearch.tasks.TaskManager;
 import org.opensearch.tasks.TaskResourceTrackingService;
 import org.opensearch.watcher.ResourceWatcherService;
 import org.opensearch.action.admin.cluster.configuration.TransportAddVotingConfigExclusionsAction;
@@ -575,7 +576,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 ShardIndexingPressureMemoryManager.SUCCESSFUL_REQUEST_ELAPSED_TIMEOUT,
                 ShardIndexingPressureMemoryManager.MAX_OUTSTANDING_REQUESTS,
                 IndexingPressure.MAX_INDEXING_BYTES,
-                TaskResourceTrackingService.TASK_RESOURCE_TRACKING_ENABLED
+                TaskResourceTrackingService.TASK_RESOURCE_TRACKING_ENABLED,
+                TaskManager.TASK_RESOURCE_CONSUMERS_ENABLED
             )
         )
     );

server/src/main/java/org/opensearch/search/internal/ShardSearchRequest.java

@@ -51,6 +51,7 @@
 import org.opensearch.common.io.stream.StreamInput;
 import org.opensearch.common.io.stream.StreamOutput;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.xcontent.ToXContent;
 import org.opensearch.index.Index;
 import org.opensearch.index.query.BoolQueryBuilder;
 import org.opensearch.index.query.MatchNoneQueryBuilder;
@@ -71,6 +72,7 @@
 import org.opensearch.transport.TransportRequest;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.Arrays;
 import java.util.Map;
 import java.util.function.Function;
@@ -85,6 +87,8 @@
  * @opensearch.internal
  */
 public class ShardSearchRequest extends TransportRequest implements IndicesRequest {
+    public static final ToXContent.Params FORMAT_PARAMS = new ToXContent.MapParams(Collections.singletonMap("pretty", "false"));
+
     private final String clusterAlias;
     private final ShardId shardId;
     private final int numberOfShards;
@@ -501,7 +505,7 @@ public String getClusterAlias() {
 
     @Override
     public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
-        return new SearchShardTask(id, type, action, getDescription(), parentTaskId, headers);
+        return new SearchShardTask(id, type, action, getDescription(), parentTaskId, headers, this::getMetadataSupplier);
     }
 
     @Override
@@ -510,6 +514,16 @@ public String getDescription() {
         return "shardId[" + shardId() + "]";
     }
 
+    public String getMetadataSupplier() {
+        StringBuilder sb = new StringBuilder();
+        if (source != null) {
+            sb.append("source[").append(source.toString(FORMAT_PARAMS)).append("]");
+        } else {
+            sb.append("source[]");
+        }
+        return sb.toString();
+    }
+
     public Rewriteable<Rewriteable> getRewriteable() {
         return new RequestRewritable(this);
     }

server/src/main/java/org/opensearch/search/query/QuerySearchRequest.java

@@ -123,7 +123,7 @@ public IndicesOptions indicesOptions() {
 
     @Override
     public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
-        return new SearchShardTask(id, type, action, getDescription(), parentTaskId, headers);
+        return new SearchShardTask(id, type, action, getDescription(), parentTaskId, headers, this::getMetadataSupplier);
     }
 
     public String getDescription() {
@@ -137,4 +137,7 @@ public String getDescription() {
         return sb.toString();
     }
 
+    public String getMetadataSupplier() {
+        return shardSearchRequest().getMetadataSupplier();
+    }
 }

server/src/main/java/org/opensearch/tasks/TaskManager.java

@@ -51,13 +51,16 @@
 import org.opensearch.cluster.node.DiscoveryNodes;
 import org.opensearch.common.lease.Releasable;
 import org.opensearch.common.lease.Releasables;
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.ByteSizeValue;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.concurrent.AbstractRunnable;
 import org.opensearch.common.util.concurrent.ConcurrentCollections;
 import org.opensearch.common.util.concurrent.ConcurrentMapLong;
 import org.opensearch.common.util.concurrent.ThreadContext;
+import org.opensearch.tasks.consumer.TopNSearchTasksLogger;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TcpChannel;
 
@@ -75,6 +78,7 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
@@ -92,6 +96,15 @@ public class TaskManager implements ClusterStateApplier {
 
     private static final TimeValue WAIT_FOR_COMPLETION_POLL = timeValueMillis(100);
 
+    public static final String TASK_RESOURCE_CONSUMERS_ATTRIBUTES = "task_resource_consumers.enabled";
+
+    public static final Setting<Boolean> TASK_RESOURCE_CONSUMERS_ENABLED = Setting.boolSetting(
+        TASK_RESOURCE_CONSUMERS_ATTRIBUTES,
+        false,
+        Setting.Property.Dynamic,
+        Setting.Property.NodeScope
+    );
+
     /**
      * Rest headers that are copied to the task
      */
@@ -116,10 +129,26 @@ public class TaskManager implements ClusterStateApplier {
     private final Map<TcpChannel, ChannelPendingTaskTracker> channelPendingTaskTrackers = ConcurrentCollections.newConcurrentMap();
     private final SetOnce<TaskCancellationService> cancellationService = new SetOnce<>();
 
+    private volatile boolean taskResourceConsumersEnabled;
+    private final Set<Consumer<Task>> taskResourceConsumer;
+
+    public static TaskManager createTaskManagerWithClusterSettings(
+        Settings settings,
+        ClusterSettings clusterSettings,
+        ThreadPool threadPool,
+        Set<String> taskHeaders
+    ) {
+        final TaskManager taskManager = new TaskManager(settings, threadPool, taskHeaders);
+        clusterSettings.addSettingsUpdateConsumer(TASK_RESOURCE_CONSUMERS_ENABLED, taskManager::setTaskResourceConsumersEnabled);
+        return taskManager;
+    }
+
     public TaskManager(Settings settings, ThreadPool threadPool, Set<String> taskHeaders) {
         this.threadPool = threadPool;
         this.taskHeaders = new ArrayList<>(taskHeaders);
         this.maxHeaderSize = SETTING_HTTP_MAX_HEADER_SIZE.get(settings);
+        this.taskResourceConsumersEnabled = TASK_RESOURCE_CONSUMERS_ENABLED.get(settings);
+        this.taskResourceConsumer = Set.of(new TopNSearchTasksLogger(settings));
     }
 
     public void setTaskResultsService(TaskResultsService taskResultsService) {
@@ -135,6 +164,10 @@ public void setTaskResourceTrackingService(TaskResourceTrackingService taskResou
         this.taskResourceTrackingService.set(taskResourceTrackingService);
     }
 
+    public void setTaskResourceConsumersEnabled(boolean taskResourceConsumersEnabled) {
+        this.taskResourceConsumersEnabled = taskResourceConsumersEnabled;
+    }
+
     /**
      * Registers a task without parent task
      */
@@ -240,6 +273,16 @@ public Task unregister(Task task) {
         // Decrement the task's self-thread as part of unregistration.
         task.decrementResourceTrackingThreads();
 
+        if (taskResourceConsumersEnabled) {
+            for (Consumer<Task> taskConsumer : taskResourceConsumer) {
+                try {
+                    taskConsumer.accept(task);
+                } catch (Exception e) {
+                    logger.error("error encountered when updating the consumer", e);
+                }
+            }
+        }
+
         if (task instanceof CancellableTask) {
             CancellableTaskHolder holder = cancellableTasks.remove(task.getId());
             if (holder != null) {

server/src/main/java/org/opensearch/tasks/consumer/SearchShardTaskDetailsLogMessage.java

@@ -0,0 +1,67 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.tasks.consumer;
+
+import org.opensearch.action.search.SearchShardTask;
+import org.opensearch.common.logging.OpenSearchLogMessage;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Search shard task information that will be extracted from Task and converted into
+ * format that will be logged
+ *
+ * @opensearch.internal
+ */
+public final class SearchShardTaskDetailsLogMessage extends OpenSearchLogMessage {
+    SearchShardTaskDetailsLogMessage(SearchShardTask task) {
+        super(prepareMap(task), message(task));
+    }
+
+    private static Map<String, Object> prepareMap(SearchShardTask task) {
+        Map<String, Object> messageFields = new HashMap<>();
+        messageFields.put("taskId", task.getId());
+        messageFields.put("type", task.getType());
+        messageFields.put("action", task.getAction());
+        messageFields.put("description", task.getDescription());
+        messageFields.put("start_time_millis", task.getStartTime());
+        messageFields.put("parentTaskId", task.getParentTaskId());
+        messageFields.put("resource_stats", task.getResourceStats());
+        messageFields.put("metadata", task.getTaskMetadata());
+        return messageFields;
+    }
+
+    // Message will be used in plaintext logs
+    private static String message(SearchShardTask task) {
+        StringBuilder sb = new StringBuilder();
+        sb.append("taskId:[")
+            .append(task.getId())
+            .append("], ")
+            .append("type:[")
+            .append(task.getType())
+            .append("], ")
+            .append("action:[")
+            .append(task.getAction())
+            .append("], ")
+            .append("description:[")
+            .append(task.getDescription())
+            .append("], ")
+            .append("start_time_millis:[")
+            .append(task.getStartTime())
+            .append("], ")
+            .append("resource_stats:[")
+            .append(task.getResourceStats())
+            .append("], ")
+            .append("metadata:[")
+            .append(task.getTaskMetadata())
+            .append("]");
+        return sb.toString();
+    }
+}