Return failure store status on ingest failures for data streams without fs enabled.

jbaiera · jbaiera · commit 9bd4fbc1f2e2 · 2025-02-03T11:12:02.000-05:00
diff --git a/modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/190_failure_store_redirection.yml b/modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/190_failure_store_redirection.yml
@@ -786,7 +786,7 @@ teardown:
   - is_false: items.1.create.failure_store
 
 ---
-"Test failure store status with bulk request":
+"Test failure store status with bulk request failing on mappings":
   - do:
       allowed_warnings:
         - "index template [generic_logs_template] has index patterns [logs-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [generic_logs_template] will take precedence during new index creation"
@@ -865,3 +865,90 @@ teardown:
   - match: { items.3.create.status: 400 }
   - match: { items.3.create.error.type: document_parsing_exception }
   - match: { items.3.create.failure_store: not_enabled }
+
+---
+"Test failure store status with bulk request failing in ingest":
+  - do:
+      ingest.put_pipeline:
+        id: "failing_pipeline"
+        body: >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "fail": {
+                  "message" : "error_message",
+                  "tag": "foo-tag"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      allowed_warnings:
+        - "index template [generic_logs_template] has index patterns [logs-*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [generic_logs_template] will take precedence during new index creation"
+      indices.put_index_template:
+        name: generic_logs_template
+        body:
+          index_patterns: logs-*
+          data_stream: {}
+          template:
+            settings:
+              number_of_shards:   1
+              number_of_replicas: 1
+              index:
+                default_pipeline: "failing_pipeline"
+            mappings:
+              properties:
+                '@timestamp':
+                  type: date
+                count:
+                  type: long
+            data_stream_options:
+              failure_store:
+                enabled: true
+  - do:
+      allowed_warnings:
+        - "index template [no-fs] has index patterns [no-fs*] matching patterns from existing older templates [global] with patterns (global => [*]); this template [no-fs] will take precedence during new index creation"
+      indices.put_index_template:
+        name: no-fs
+        body:
+          index_patterns: no-fs*
+          data_stream: {}
+          template:
+            settings:
+              number_of_shards: 1
+              number_of_replicas: 0
+              index:
+                default_pipeline: "failing_pipeline"
+            mappings:
+              properties:
+                '@timestamp':
+                  type: date
+                count:
+                  type: long
+            data_stream_options:
+              failure_store:
+                enabled: false
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - '{ "create": { "_index": "logs-foobar", "_id": "1" } }'
+          - '{ "@timestamp": "2022-01-01", "count": 1 }'
+          - '{ "create": { "_index": "no-fs", "_id": "1" } }'
+          - '{ "@timestamp": "2022-01-01", "count": 1 }'
+  - is_true: errors
+  # Successfully indexed to backing index
+  - match: { items.0.create._index: '/\.fs-logs-foobar-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - match: { items.0.create.status: 201 }
+  - match: { items.0.create.failure_store: used }
+
+  # Rejected, eligible to go to failure store, but failure store not enabled
+  - match: { items.1.create._index: 'no-fs' }
+  - match: { items.1.create.status: 500 }
+  - match: { items.1.create.failure_store: not_enabled }
+  - match: { items.1.create.error.type: fail_processor_exception }
+  - match: { items.1.create.error.reason: error_message }
diff --git a/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestModifier.java b/server/src/main/java/org/elasticsearch/action/bulk/BulkRequestModifier.java
@@ -179,15 +179,15 @@ private void assertResponsesAreCorrect(BulkItemResponse[] bulkResponses, BulkIte
      * @param slot the slot in the bulk request to mark as failed.
      * @param e the failure encountered.
      */
-    synchronized void markItemAsFailed(int slot, Exception e) {
+    synchronized void markItemAsFailed(int slot, Exception e, IndexDocFailureStoreStatus failureStoreStatus) {
         final DocWriteRequest<?> docWriteRequest = bulkRequest.requests().get(slot);
         final String id = Objects.requireNonNullElse(docWriteRequest.id(), DROPPED_OR_FAILED_ITEM_WITH_AUTO_GENERATED_ID);
         // We hit a error during preprocessing a request, so we:
         // 1) Remember the request item slot from the bulk, so that when we're done processing all requests we know what failed
         // 2) Add a bulk item failure for this request
         // 3) Continue with the next request in the bulk.
         failedSlots.set(slot);
-        BulkItemResponse.Failure failure = new BulkItemResponse.Failure(docWriteRequest.index(), id, e);
+        BulkItemResponse.Failure failure = new BulkItemResponse.Failure(docWriteRequest.index(), id, e, failureStoreStatus);
         itemResponses.add(BulkItemResponse.failure(slot, docWriteRequest.opType(), failure));
     }
 
@@ -223,7 +223,7 @@ public void markItemForFailureStore(int slot, String targetIndexName, Exception
             assert false
                 : "Attempting to route a failed write request type to a failure store but the failure store is not enabled! "
                     + "This should be guarded against in TransportBulkAction#shouldStoreFailure()";
-            markItemAsFailed(slot, e);
+            markItemAsFailed(slot, e, IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN);
         } else {
             // We get the index write request to find the source of the failed document
             IndexRequest indexRequest = TransportBulkAction.getIndexWriteRequest(bulkRequest.requests().get(slot));
@@ -238,7 +238,7 @@ public void markItemForFailureStore(int slot, String targetIndexName, Exception
                         + "], index: ["
                         + targetIndexName
                         + "]";
-                markItemAsFailed(slot, e);
+                markItemAsFailed(slot, e, IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN);
                 logger.debug(
                     () -> "Attempted to redirect an invalid write operation after ingest failure - type: ["
                         + bulkRequest.requests().get(slot).getClass().getName()
@@ -267,7 +267,7 @@ public void markItemForFailureStore(int slot, String targetIndexName, Exception
                             + "]",
                         ioException
                     );
-                    markItemAsFailed(slot, e);
+                    markItemAsFailed(slot, e, IndexDocFailureStoreStatus.FAILED);
                 }
             }
         }
diff --git a/server/src/main/java/org/elasticsearch/ingest/IngestService.java b/server/src/main/java/org/elasticsearch/ingest/IngestService.java
@@ -21,6 +21,7 @@
 import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
 import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
 import org.elasticsearch.action.bulk.FailureStoreMetrics;
+import org.elasticsearch.action.bulk.IndexDocFailureStoreStatus;
 import org.elasticsearch.action.bulk.TransportBulkAction;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.ingest.DeletePipelineRequest;
@@ -729,12 +730,32 @@ void validatePipeline(Map<DiscoveryNode, IngestInfo> ingestInfos, String pipelin
         ExceptionsHelper.rethrowAndSuppress(exceptions);
     }
 
-    private record IngestPipelinesExecutionResult(boolean success, boolean shouldKeep, Exception exception, String failedIndex) {
-
-        private static final IngestPipelinesExecutionResult SUCCESSFUL_RESULT = new IngestPipelinesExecutionResult(true, true, null, null);
-        private static final IngestPipelinesExecutionResult DISCARD_RESULT = new IngestPipelinesExecutionResult(true, false, null, null);
+    private record IngestPipelinesExecutionResult(
+        boolean success,
+        boolean shouldKeep,
+        Exception exception,
+        String failedIndex,
+        IndexDocFailureStoreStatus failureStoreStatus
+    ) {
+        private static final IngestPipelinesExecutionResult SUCCESSFUL_RESULT = new IngestPipelinesExecutionResult(
+            true,
+            true,
+            null,
+            null,
+            IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN
+        );
+        private static final IngestPipelinesExecutionResult DISCARD_RESULT = new IngestPipelinesExecutionResult(
+            true,
+            false,
+            null,
+            null,
+            IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN
+        );
         private static IngestPipelinesExecutionResult failAndStoreFor(String index, Exception e) {
-            return new IngestPipelinesExecutionResult(false, true, e, index);
+            return new IngestPipelinesExecutionResult(false, true, e, index, IndexDocFailureStoreStatus.USED);
+        }
+        private static IngestPipelinesExecutionResult failWithoutStoringIn(String index, Exception e) {
+            return new IngestPipelinesExecutionResult(false, true, e, index, IndexDocFailureStoreStatus.NOT_ENABLED);
         }
     }
 
@@ -764,7 +785,7 @@ public void executeBulkRequest(
         final IntConsumer onDropped,
         final Function<String, Boolean> resolveFailureStore,
         final TriConsumer<Integer, String, Exception> onStoreFailure,
-        final BiConsumer<Integer, Exception> onFailure,
+        final TriConsumer<Integer, Exception, IndexDocFailureStoreStatus> onFailure,
         final BiConsumer<Thread, Exception> onCompletion,
         final Executor executor
     ) {
@@ -821,18 +842,26 @@ public void onResponse(IngestPipelinesExecutionResult result) {
                                             firstPipeline.getMetrics().postIngestBytes(indexRequest.ramBytesUsed());
                                         }
                                     } else {
-                                        // We were given a failure result in the onResponse method, so we must store the failure
-                                        // Recover the original document state, track a failed ingest, and pass it along
-                                        updateIndexRequestMetadata(indexRequest, originalDocumentMetadata);
                                         totalMetrics.ingestFailed();
-                                        onStoreFailure.apply(slot, result.failedIndex, result.exception);
+                                        if (IndexDocFailureStoreStatus.NOT_ENABLED.equals(result.failureStoreStatus)) {
+                                            // A failure result, but despite the target being a data stream, it does not have failure
+                                            // storage enabled currently. Capture the status in the onFailure call and skip any further
+                                            // processing
+                                            onFailure.apply(slot, result.exception, result.failureStoreStatus);
+                                        } else {
+                                            // We were given a failure result in the onResponse method, so we must store the failure
+                                            // Recover the original document state, track a failed ingest, and pass it along
+                                            updateIndexRequestMetadata(indexRequest, originalDocumentMetadata);
+                                            onStoreFailure.apply(slot, result.failedIndex, result.exception);
+                                        }
                                     }
                                 }
 
                                 @Override
                                 public void onFailure(Exception e) {
+                                    // The target of the request does not allow failure storage, or failed for unforeseen reason
                                     totalMetrics.ingestFailed();
-                                    onFailure.accept(slot, e);
+                                    onFailure.apply(slot, e, IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN);
                                 }
                             },
                             () -> {
@@ -954,15 +983,15 @@ private void executePipelines(
             if (failureStoreResolution != null && failureStoreResolution) {
                 failureStoreMetrics.incrementFailureStore(originalIndex, errorType, FailureStoreMetrics.ErrorLocation.PIPELINE);
                 listener.onResponse(IngestPipelinesExecutionResult.failAndStoreFor(originalIndex, e));
+            } else if (failureStoreResolution != null) {
+                // If this document targeted a data stream that didn't have the failure store enabled, we increment
+                // the rejected counter.
+                // We also increment the total counter because this request will not reach the code that increments
+                // the total counter for non-rejected documents.
+                failureStoreMetrics.incrementTotal(originalIndex);
+                failureStoreMetrics.incrementRejected(originalIndex, errorType, FailureStoreMetrics.ErrorLocation.PIPELINE, false);
+                listener.onResponse(IngestPipelinesExecutionResult.failWithoutStoringIn(originalIndex, e));
             } else {
-                if (failureStoreResolution != null) {
-                    // If this document targeted a data stream that didn't have the failure store enabled, we increment
-                    // the rejected counter.
-                    // We also increment the total counter because this request will not reach the code that increments
-                    // the total counter for non-rejected documents.
-                    failureStoreMetrics.incrementTotal(originalIndex);
-                    failureStoreMetrics.incrementRejected(originalIndex, errorType, FailureStoreMetrics.ErrorLocation.PIPELINE, false);
-                }
                 listener.onFailure(e);
             }
         };
diff --git a/server/src/test/java/org/elasticsearch/action/bulk/BulkRequestModifierTests.java b/server/src/test/java/org/elasticsearch/action/bulk/BulkRequestModifierTests.java
@@ -44,7 +44,7 @@ public void testBulkRequestModifier() {
         for (int i = 0; modifier.hasNext(); i++) {
             modifier.next();
             if (randomBoolean()) {
-                modifier.markItemAsFailed(i, new RuntimeException());
+                modifier.markItemAsFailed(i, new RuntimeException(), randomFrom(IndexDocFailureStoreStatus.values()));
                 failedSlots.add(i);
             }
         }
@@ -110,7 +110,7 @@ public void testPipelineFailures() {
 
         // actually mark the failures
         for (int i : failures) {
-            modifier.markItemAsFailed(i, new RuntimeException());
+            modifier.markItemAsFailed(i, new RuntimeException(), randomFrom(IndexDocFailureStoreStatus.values()));
         }
 
         // So half of the requests have "failed", so only the successful requests are left:
diff --git a/server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java b/server/src/test/java/org/elasticsearch/action/bulk/TransportBulkActionIngestTests.java
@@ -117,7 +117,7 @@ public class TransportBulkActionIngestTests extends ESTestCase {
     @Captor
     ArgumentCaptor<TriConsumer<Integer, String, Exception>> redirectHandler;
     @Captor
-    ArgumentCaptor<BiConsumer<Integer, Exception>> failureHandler;
+    ArgumentCaptor<TriConsumer<Integer, Exception, IndexDocFailureStoreStatus>> failureHandler;
     @Captor
     ArgumentCaptor<BiConsumer<Thread, Exception>> completionHandler;
     @Captor
@@ -412,7 +412,8 @@ public void testIngestLocal() throws Exception {
 
         // now check success
         Iterator<DocWriteRequest<?>> req = bulkDocsItr.getValue().iterator();
-        failureHandler.getValue().accept(0, exception); // have an exception for our one index request
+        // have an exception for our one index request
+        failureHandler.getValue().apply(0, exception, IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN);
         indexRequest2.setPipeline(IngestService.NOOP_PIPELINE_NAME); // this is done by the real pipeline execution service when processing
         // ensure redirects on failure store data stream
         assertTrue(redirectPredicate.getValue().apply(WITH_FAILURE_STORE_ENABLED + "-1"));
@@ -509,7 +510,8 @@ public void testIngestSystemLocal() throws Exception {
 
         // now check success
         Iterator<DocWriteRequest<?>> req = bulkDocsItr.getValue().iterator();
-        failureHandler.getValue().accept(0, exception); // have an exception for our one index request
+        // have an exception for our one index request
+        failureHandler.getValue().apply(0, exception, IndexDocFailureStoreStatus.NOT_APPLICABLE_OR_UNKNOWN);
         indexRequest2.setPipeline(IngestService.NOOP_PIPELINE_NAME); // this is done by the real pipeline execution service when processing
         completionHandler.getValue().accept(DUMMY_WRITE_THREAD, null);
         assertTrue(action.isExecuted);
diff --git a/server/src/test/java/org/elasticsearch/ingest/IngestServiceTests.java b/server/src/test/java/org/elasticsearch/ingest/IngestServiceTests.java

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ public void testBulkRequestModifier() {`
`44`	`44`	`for (int i = 0; modifier.hasNext(); i++) {`
`45`	`45`	`modifier.next();`
`46`	`46`	`if (randomBoolean()) {`
`47`		`- modifier.markItemAsFailed(i, new RuntimeException());`
	`47`	`+ modifier.markItemAsFailed(i, new RuntimeException(), randomFrom(IndexDocFailureStoreStatus.values()));`
`48`	`48`	`failedSlots.add(i);`
`49`	`49`	`}`
`50`	`50`	`}`
`@@ -110,7 +110,7 @@ public void testPipelineFailures() {`
`110`	`110`
`111`	`111`	`// actually mark the failures`
`112`	`112`	`for (int i : failures) {`
`113`		`- modifier.markItemAsFailed(i, new RuntimeException());`
	`113`	`+ modifier.markItemAsFailed(i, new RuntimeException(), randomFrom(IndexDocFailureStoreStatus.values()));`
`114`	`114`	`}`
`115`	`115`
`116`	`116`	`// So half of the requests have "failed", so only the successful requests are left:`