[BUG] Add missing fields to resolve Strict Dynamic Mapping issue in .…

…tasks index (opensearch-project#16060) - Fixed issue where `.tasks` index failed to update due to StrictDynamicMappingException when a task was cancelled. - Added missing `cancellation_time_millis` and `resource_stats` fields to `task-index-mapping.json`. - Ensured proper task result storage by updating the mappings. - Changed the version in the meta field from 4 to 5 to reflect the updated mappings. Signed-off-by: inpink <inpink@kakao.com>
inpink · Oct 18, 2024 · 1cf934f · 1cf934f
1 parent 0bded88
commit 1cf934f
Show file tree

Hide file tree

Showing 5 changed files with 168 additions and 19 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -87,6 +87,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Workload Management] Make query groups persistent across process restarts [#16370](https://github.com/opensearch-project/OpenSearch/pull/16370)
 
 - Fix inefficient Stream API call chains ending with count() ([#15386](https://github.com/opensearch-project/OpenSearch/pull/15386))
+- Fix missing fields in task index mapping to ensure proper task result storage ([#16201](https://github.com/opensearch-project/OpenSearch/pull/16201))
 
 ### Security
 

diff --git a/.../src/internalClusterTest/java/org/opensearch/action/admin/cluster/node/tasks/TasksIT.java b/.../src/internalClusterTest/java/org/opensearch/action/admin/cluster/node/tasks/TasksIT.java
@@ -32,6 +32,12 @@
 
 package org.opensearch.action.admin.cluster.node.tasks;
 
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import org.opensearch.ExceptionsHelper;
 import org.opensearch.OpenSearchException;
 import org.opensearch.OpenSearchTimeoutException;
@@ -58,9 +64,14 @@
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.common.regex.Regex;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.io.Streams;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.tasks.TaskId;
+import org.opensearch.core.tasks.resourcetracker.TaskResourceStats;
+import org.opensearch.core.tasks.resourcetracker.TaskResourceUsage;
+import org.opensearch.core.tasks.resourcetracker.TaskThreadUsage;
 import org.opensearch.core.xcontent.MediaTypeRegistry;
+import org.opensearch.index.mapper.StrictDynamicMappingException;
 import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.search.builder.SearchSourceBuilder;
 import org.opensearch.tasks.Task;
@@ -86,6 +97,8 @@
 
 import static java.util.Collections.emptyList;
 import static java.util.Collections.singleton;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
 import static org.opensearch.common.unit.TimeValue.timeValueMillis;
 import static org.opensearch.common.unit.TimeValue.timeValueSeconds;
 import static org.opensearch.http.HttpTransportSettings.SETTING_HTTP_MAX_HEADER_SIZE;
@@ -112,6 +125,26 @@
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.SUITE, minNumDataNodes = 2)
 public class TasksIT extends AbstractTasksIT {
 
+    protected final TaskInfo taskInfo = new TaskInfo(
+        new TaskId("fake", 1),
+        "test_type",
+        "test_action",
+        "test_description",
+        null,
+        0L,
+        1L,
+        false,
+        false,
+        TaskId.EMPTY_TASK_ID,
+        Collections.emptyMap(),
+        new TaskResourceStats(new HashMap<>() {
+            {
+                put("dummy-type1", new TaskResourceUsage(10, 20));
+            }
+        }, new TaskThreadUsage(30, 40)),
+        2L
+    );
+
     public void testTaskCounts() {
         // Run only on data nodes
         ListTasksResponse response = client().admin()
@@ -881,20 +914,7 @@ public void testNodeNotFoundButTaskFound() throws Exception {
         TaskResultsService resultsService = internalCluster().getInstance(TaskResultsService.class);
         resultsService.storeResult(
             new TaskResult(
-                new TaskInfo(
-                    new TaskId("fake", 1),
-                    "test",
-                    "test",
-                    "",
-                    null,
-                    0,
-                    0,
-                    false,
-                    false,
-                    TaskId.EMPTY_TASK_ID,
-                    Collections.emptyMap(),
-                    null
-                ),
+                taskInfo,
                 new RuntimeException("test")
             ),
             new ActionListener<Void>() {
@@ -917,8 +937,66 @@ public void onFailure(Exception e) {
 
         // Now we can find it!
         GetTaskResponse response = expectFinishedTask(new TaskId("fake:1"));
-        assertEquals("test", response.getTask().getTask().getAction());
-        assertNotNull(response.getTask().getError());
-        assertNull(response.getTask().getResponse());
+        TaskResult taskResult = response.getTask();
+        TaskInfo task = taskResult.getTask();
+
+        assertEquals("fake", task.getTaskId().getNodeId());
+        assertEquals(1, task.getTaskId().getId());
+        assertEquals("test_type", task.getType());
+        assertEquals("test_action", task.getAction());
+        assertEquals("test_description", task.getDescription());
+        assertEquals(0L, task.getStartTime());
+        assertEquals(1L, task.getRunningTimeNanos());
+        assertFalse(task.isCancellable());
+        assertFalse(task.isCancelled());
+        assertEquals(TaskId.EMPTY_TASK_ID, task.getParentTaskId());
+        assertEquals(1, task.getResourceStats().getResourceUsageInfo().size());
+        assertEquals(30, task.getResourceStats().getThreadUsage().getThreadExecutions());
+        assertEquals(40, task.getResourceStats().getThreadUsage().getActiveThreads());
+        assertEquals(Long.valueOf(2L), task.getCancellationStartTime());
+
+
+        assertNotNull(taskResult.getError());
+        assertNull(taskResult.getResponse());
+        System.out.println("original:"+resultsService.taskResultIndexMapping());
+    }
+
+    public void testStoreTaskResultFailsDueToMissingIndexMappingFields() throws IOException {
+        // given
+        TaskResultsService resultsService = spy(internalCluster().getInstance(TaskResultsService.class));
+
+        InputStream mockInputStream = getClass().getResourceAsStream("/org/opensearch/tasks/missing-fields-task-index-mapping.json");
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        Streams.copy(mockInputStream, out);
+        String mockJsonString = out.toString(StandardCharsets.UTF_8.name());
+
+        // when & then
+        doReturn(mockJsonString).when(resultsService).taskResultIndexMapping();
+
+        CompletionException thrown = assertThrows(CompletionException.class, () -> {
+            CompletableFuture<Void> future = new CompletableFuture<>();
+
+            resultsService.storeResult(
+                new TaskResult(
+                    taskInfo,
+                    new RuntimeException("test")
+                ),
+                new ActionListener<Void>() {
+                    @Override
+                    public void onResponse(Void response) {
+                        future.complete(null);
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        future.completeExceptionally(e);
+                    }
+                }
+            );
+
+            future.join();
+        });
+
+        assertTrue(thrown.getCause() instanceof StrictDynamicMappingException);
     }
 }
diff --git a/server/src/main/java/org/opensearch/tasks/TaskResultsService.java b/server/src/main/java/org/opensearch/tasks/TaskResultsService.java
@@ -85,7 +85,7 @@ public class TaskResultsService {
 
     public static final String TASK_RESULT_MAPPING_VERSION_META_FIELD = "version";
 
-    public static final int TASK_RESULT_MAPPING_VERSION = 4; // must match version in task-index-mapping.json
+    public static final int TASK_RESULT_MAPPING_VERSION = 5; // must match version in task-index-mapping.json
 
     /**
      * The backoff policy to use when saving a task result fails. The total wait

diff --git a/server/src/main/resources/org/opensearch/tasks/task-index-mapping.json b/server/src/main/resources/org/opensearch/tasks/task-index-mapping.json
@@ -1,7 +1,7 @@
 {
   "_doc" : {
     "_meta": {
-      "version": 4
+      "version": 5
     },
     "dynamic" : "strict",
     "properties" : {
@@ -34,6 +34,9 @@
           "start_time_in_millis": {
             "type": "long"
           },
+          "cancellation_time_millis": {
+            "type": "long"
+          },
           "type": {
             "type": "keyword"
           },
@@ -47,6 +50,10 @@
           "headers": {
             "type" : "object",
             "enabled" : false
+          },
+          "resource_stats": {
+            "type" : "object",
+            "enabled" : false
           }
         }
       },

diff --git a/server/src/test/resources/org/opensearch/tasks/missing-fields-task-index-mapping.json b/server/src/test/resources/org/opensearch/tasks/missing-fields-task-index-mapping.json
@@ -0,0 +1,63 @@
+{
+  "_doc" : {
+    "_meta": {
+      "version": 5
+    },
+    "dynamic" : "strict",
+    "properties" : {
+      "completed": {
+        "type": "boolean"
+      },
+      "task" : {
+        "properties": {
+          "action": {
+            "type": "keyword"
+          },
+          "cancellable": {
+            "type": "boolean"
+          },
+          "cancelled": {
+            "type": "boolean"
+          },
+          "id": {
+            "type": "long"
+          },
+          "parent_task_id": {
+            "type": "keyword"
+          },
+          "node": {
+            "type": "keyword"
+          },
+          "running_time_in_nanos": {
+            "type": "long"
+          },
+          "start_time_in_millis": {
+            "type": "long"
+          },
+          "type": {
+            "type": "keyword"
+          },
+          "status": {
+            "type" : "object",
+            "enabled" : false
+          },
+          "description": {
+            "type": "text"
+          },
+          "headers": {
+            "type" : "object",
+            "enabled" : false
+          }
+        }
+      },
+      "response" : {
+        "type" : "object",
+        "enabled" : false
+      },
+      "error" : {
+        "type" : "object",
+        "enabled" : false
+      }
+    }
+  }
+}