Redirect failed ingest node operations to a failure store when available

docs/changelog/103481.yaml

@@ -0,0 +1,5 @@
+pr: 103481
+summary: Redirect failed ingest node operations to a failure store when available
+area: Data streams
+type: feature
+issues: []

modules/data-streams/src/yamlRestTest/resources/rest-api-spec/test/data_stream/190_failure_store_redirection.yml

@@ -0,0 +1,110 @@
+---
+teardown:
+  - do:
+      indices.delete_data_stream:
+        name: logs-foobar
+        ignore: 404
+
+  - do:
+      indices.delete:
+        index: .fs-logs-foobar-*
+        ignore: 404
+
+  - do:
+      indices.delete_index_template:
+        name: generic_logs_template
+        ignore: 404
+
+  - do:
+      ingest.delete_pipeline:
+        id: "failing_pipeline"
+        ignore: 404
+
+---
+"Redirect ingest failure in data stream to failure store":
+  - skip:
+      version: " - 8.12.99"
+      reason: "data stream failure stores only redirect ingest failures in 8.13+"
+      features: [allowed_warnings, contains]
+
+  - do:
+      ingest.put_pipeline:
+        id: "failing_pipeline"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "fail" : {
+                  "message" : "error_message"
+                }
+              }
+            ]
+          }
+  - 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:
+            failure_store: true
+          template:
+            settings:
+              number_of_shards:   1
+              number_of_replicas: 1
+              index:
+                default_pipeline: "failing_pipeline"
+
+  - do:
+      index:
+        index:  logs-foobar
+        refresh: true
+        body:
+          '@timestamp': '2020-12-12'
+          foo: bar
+
+  - do:
+      indices.get_data_stream:
+        name: logs-foobar
+  - match: { data_streams.0.name: logs-foobar }
+  - match: { data_streams.0.timestamp_field.name: '@timestamp' }
+  - length: { data_streams.0.indices: 1 }
+  - match: { data_streams.0.indices.0.index_name: '/\.ds-logs-foobar-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+  - match: { data_streams.0.failure_store: true }
+  - length: { data_streams.0.failure_indices: 1 }
+  - match: { data_streams.0.failure_indices.0.index_name: '/\.fs-logs-foobar-(\d{4}\.\d{2}\.\d{2}-)?000001/' }
+
+  - do:
+      search:
+        index: logs-foobar
+        body: { query: { match_all: {} } }
+  - length:   { hits.hits: 0  }
+
+  - do:
+      search:
+        index: .fs-logs-foobar-*
+  - length:   { hits.hits: 1  }
+  - match: { hits.hits.0._index: "/\\.fs-logs-foobar-(\\d{4}\\.\\d{2}\\.\\d{2}-)?000001/" }
+  - exists: hits.hits.0._source.@timestamp
+  - not_exists: hits.hits.0._source.foo
+  - not_exists: hits.hits.0._source.document.id
+  - match: { hits.hits.0._source.document.index: 'logs-foobar' }
+  - match: { hits.hits.0._source.document.source.@timestamp: '2020-12-12' }
+  - match: { hits.hits.0._source.document.source.foo: 'bar' }
+  - match: { hits.hits.0._source.error.type: 'fail_processor_exception' }
+  - match: { hits.hits.0._source.error.message: 'error_message' }
+  - contains: { hits.hits.0._source.error.stack_trace: 'org.elasticsearch.ingest.common.FailProcessorException: error_message' }
+
+  - do:
+      indices.delete_data_stream:
+        name: logs-foobar
+  - is_true: acknowledged
+
+  - do:
+      indices.delete:
+        index: .fs-logs-foobar-*
+  - is_true: acknowledged

server/src/main/java/org/elasticsearch/action/bulk/FailureStoreDocument.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.action.bulk;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.json.JsonXContent;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * Transforms an indexing request using error information into a new index request to be stored in a data stream's failure store.
+ */
+public final class FailureStoreDocument {
+
+    private FailureStoreDocument() {}
+
+    /**
+     * Combines an {@link IndexRequest} that has failed during the bulk process with the error thrown for that request. The result is a
+     * new {@link IndexRequest} that can be stored in a data stream's failure store.
+     * @param source The original request that has failed to be ingested
+     * @param exception The exception that was thrown that caused the request to fail to be ingested
+     * @param targetIndexName The index that the request was targeting at time of failure
+     * @return A new {@link IndexRequest} with a failure store compliant structure
+     * @throws IOException If there is a problem when the document's new source is serialized
+     */
+    public static IndexRequest transformFailedRequest(IndexRequest source, Exception exception, String targetIndexName) throws IOException {
+        return transformFailedRequest(source, exception, targetIndexName, System::currentTimeMillis);
+    }
+
+    /**
+     * Combines an {@link IndexRequest} that has failed during the bulk process with the error thrown for that request. The result is a
+     * new {@link IndexRequest} that can be stored in a data stream's failure store.
+     * @param source The original request that has failed to be ingested
+     * @param exception The exception that was thrown that caused the request to fail to be ingested
+     * @param targetIndexName The index that the request was targeting at time of failure
+     * @param timeSupplier Supplies the value for the document's timestamp
+     * @return A new {@link IndexRequest} with a failure store compliant structure
+     * @throws IOException If there is a problem when the document's new source is serialized
+     */
+    public static IndexRequest transformFailedRequest(
+        IndexRequest source,
+        Exception exception,
+        String targetIndexName,
+        Supplier<Long> timeSupplier
+    ) throws IOException {
+        return new IndexRequest().index(targetIndexName)
+            .source(createSource(source, exception, targetIndexName, timeSupplier))
+            .opType(DocWriteRequest.OpType.CREATE)
+            .setWriteToFailureStore(true);
+    }
+
+    private static XContentBuilder createSource(
+        IndexRequest source,
+        Exception exception,
+        String targetIndexName,
+        Supplier<Long> timeSupplier
+    ) throws IOException {
+        Objects.requireNonNull(source, "source must not be null");
+        Objects.requireNonNull(exception, "exception must not be null");
+        Objects.requireNonNull(targetIndexName, "targetIndexName must not be null");
+        Objects.requireNonNull(timeSupplier, "timeSupplier must not be null");
+        Throwable unwrapped = ExceptionsHelper.unwrapCause(exception);
+        XContentBuilder builder = JsonXContent.contentBuilder();
+        builder.startObject();
+        {
+            builder.timeField("@timestamp", timeSupplier.get());
+            builder.startObject("document");
+            {
+                if (source.id() != null) {
+                    builder.field("id", source.id());
+                }
+                if (source.routing() != null) {
+                    builder.field("routing", source.routing());
+                }
+                builder.field("index", source.index());
+                // Unmapped source field
+                builder.startObject("source");
+                {
+                    builder.mapContents(source.sourceAsMap());
+                }
+                builder.endObject();
+            }
+            builder.endObject();
+            builder.startObject("error");
+            {
+                builder.field("type", ElasticsearchException.getExceptionName(unwrapped));
+                builder.field("message", unwrapped.getMessage());
+                builder.field("stack_trace", ExceptionsHelper.stackTrace(unwrapped));
+                // Further fields not yet tracked (Need to expose via specific exceptions)
+                // - pipeline
+                // - pipeline_trace
+                // - processor
+            }
+            builder.endObject();
+        }
+        builder.endObject();
+        return builder;
+    }
+}