Failure Store: Update go-docappender to respect failure store status

appender.go

@@ -70,6 +70,8 @@ type Appender struct {
 	docsFailedClient       int64
 	docsFailedServer       int64
 	docsIndexed            int64
+	docsFailureStoreUsed   int64
+	docsFailureStoreFailed int64
 	tooManyRequests        int64
 	bytesTotal             int64
 	bytesUncompressedTotal int64
@@ -255,6 +257,8 @@ func (a *Appender) Stats() Stats {
 		IndexersActive:         a.scalingInformation().activeIndexers,
 		IndexersCreated:        atomic.LoadInt64(&a.activeCreated),
 		IndexersDestroyed:      atomic.LoadInt64(&a.activeDestroyed),
+		FailureStoreUsed:       atomic.LoadInt64(&a.docsFailureStoreUsed),
+		FailureStoreFailed:     atomic.LoadInt64(&a.docsFailureStoreFailed),
 	}
 }
 
@@ -409,13 +413,12 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *BulkIndexer) error {
 		}
 		return err
 	}
-	var (
-		docsFailed, docsIndexed,
+	var docsFailed, docsIndexed,
 		// breakdown of failed docs:
 		tooManyRequests, // failed after document retries (if it applies) and final status is 429
 		clientFailed, // failed after document retries (if it applies) and final status is 400s excluding 429
 		serverFailed int64 // failed after document retries (if it applies) and final status is 500s
-	)
+
 	docsIndexed = resp.Indexed
 	var failedCount map[BulkIndexerResponseItem]int
 	if len(resp.FailedDocs) > 0 {
@@ -483,11 +486,31 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *BulkIndexer) error {
 			metric.WithAttributes(attribute.String("status", "FailedServer")),
 		)
 	}
+	if resp.FailureStoreUsed > 0 {
+		a.addCount(resp.FailureStoreUsed, &a.docsFailureStoreUsed,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(
+				attribute.String("status", "FailureStore"),
+				attribute.String("failure_store", string(FailureStoreStatusUsed)),
+			),
+		)
+	}
+	if resp.FailureStoreFailed > 0 {
+		a.addCount(resp.FailureStoreFailed, &a.docsFailureStoreFailed,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(
+				attribute.String("status", "FailureStore"),
+				attribute.String("failure_store", string(FailureStoreStatusFailed)),
+			),
+		)
+	}
 	logger.Debug(
 		"bulk request completed",
 		zap.Int64("docs_indexed", docsIndexed),
 		zap.Int64("docs_failed", docsFailed),
 		zap.Int64("docs_rate_limited", tooManyRequests),
+		zap.Int64("docs_failure_store_used", resp.FailureStoreUsed),
+		zap.Int64("docs_failure_store_failed", resp.FailureStoreFailed),
 	)
 	if a.otelTracingEnabled() && span.IsRecording() {
 		span.SetStatus(codes.Ok, "")
@@ -826,6 +849,14 @@ type Stats struct {
 
 	// IndexersDestroyed represents the number of times an active indexer was destroyed.
 	IndexersDestroyed int64
+
+	// FailureStoreUsed represents the number of indexing operations that have resolved
+	// in indexing to failure store.
+	FailureStoreUsed int64
+
+	// FailureStoreFailed represents the number of indexing operations that have failed
+	// while indexing to failure store.
+	FailureStoreFailed int64
 }
 
 func timeFunc(f func()) time.Duration {

appender_test.go

@@ -67,18 +67,22 @@ func TestAppender(t *testing.T) {
 		// "too many requests". These will be recorded as failures in indexing
 		// stats.
 		for i := range result.Items {
-			if i > 2 {
+			if i > 4 {
 				break
 			}
-			status := http.StatusInternalServerError
-			switch i {
-			case 1:
-				status = http.StatusTooManyRequests
-			case 2:
-				status = http.StatusUnauthorized
-			}
 			for action, item := range result.Items[i] {
-				item.Status = status
+				switch i {
+				case 0:
+					item.Status = http.StatusInternalServerError
+				case 1:
+					item.Status = http.StatusTooManyRequests
+				case 2:
+					item.Status = http.StatusUnauthorized
+				case 3:
+					item.FailureStore = string(docappender.FailureStoreStatusUsed)
+				case 4:
+					item.FailureStore = string(docappender.FailureStoreStatusFailed)
+				}
 				result.Items[i][action] = item
 			}
 		}
@@ -145,6 +149,8 @@ loop:
 		AvailableBulkRequests:  10,
 		BytesTotal:             bytesTotal,
 		BytesUncompressedTotal: bytesUncompressed,
+		FailureStoreUsed:       1,
+		FailureStoreFailed:     1,
 	}, stats)
 
 	var rm metricdata.ResourceMetrics
@@ -173,6 +179,9 @@ loop:
 			case "TooMany":
 				processedAsserted++
 				assert.Equal(t, stats.TooManyRequests, dp.Value)
+			case "FailureStore":
+				processedAsserted++
+				assert.Equal(t, stats.TooManyRequests, dp.Value)
 			default:
 				assert.FailNow(t, "Unexpected metric with status: "+status.AsString())
 			}
@@ -206,7 +215,7 @@ loop:
 
 	assert.Empty(t, unexpectedMetrics)
 	assert.Equal(t, int64(7), asserted.Load())
-	assert.Equal(t, 4, processedAsserted)
+	assert.Equal(t, 6, processedAsserted)
 }
 
 func TestAppenderRetry(t *testing.T) {
@@ -753,7 +762,6 @@ func TestAppenderFlushRequestError(t *testing.T) {
 			}
 		})
 		assert.Equal(t, int64(1), asserted.Load())
-
 	}
 	t.Run("400", func(t *testing.T) {
 		test(t, http.StatusBadRequest, "flush failed (400): {\"error\":{\"type\":\"x_content_parse_exception\",\"caused_by\":{\"type\":\"json_parse_exception\"}}}")
@@ -806,7 +814,7 @@ func TestAppenderIndexFailedLogging(t *testing.T) {
 
 	core, observed := observer.New(zap.NewAtomicLevelAt(zapcore.DebugLevel))
 	indexer, err := docappender.New(client, docappender.Config{
-		FlushBytes: 500,
+		FlushBytes: 5000,
 		Logger:     zap.New(core),
 	})
 	require.NoError(t, err)
@@ -1372,7 +1380,8 @@ func TestAppenderCloseBusyIndexer(t *testing.T) {
 		BytesTotal:             bytesTotal,
 		BytesUncompressedTotal: bytesUncompressedTotal,
 		AvailableBulkRequests:  10,
-		IndexersActive:         0}, indexer.Stats())
+		IndexersActive:         0,
+	}, indexer.Stats())
 }
 
 func TestAppenderPipeline(t *testing.T) {

bulk_indexer.go

@@ -105,6 +105,10 @@ type BulkIndexerResponseStat struct {
 	Indexed int64
 	// RetriedDocs contains the total number of retried documents.
 	RetriedDocs int64
+	// FailureStoreUsed contains the total number of documents indexed to failure store.
+	FailureStoreUsed int64
+	// FailureStoreFailed contains the total number of documents which failed when indexed to failure store.
+	FailureStoreFailed int64
 	// GreatestRetry contains the greatest observed retry count in the entire
 	// bulk request.
 	GreatestRetry int
@@ -125,6 +129,22 @@ type BulkIndexerResponseItem struct {
 	} `json:"error,omitempty"`
 }
 
+// FailureStoreStatus defines enumeration type for all known failure store statuses.
+type FailureStoreStatus string
+
+const (
+	// FailureStoreStatusUknown implicit status which represents that there is no information about
+	// this response or that the failure store is not applicable.
+	FailureStoreStatusUnknown FailureStoreStatus = "not_applicable_or_unknown"
+	// FailureStoreStatusUsed status which represents that this document was stored in the failure store successfully.
+	FailureStoreStatusUsed FailureStoreStatus = "used"
+	// FailureStoreStatusFailed status which represents that this document was rejected from the failure store.
+	FailureStoreStatusFailed FailureStoreStatus = "failed"
+	// FailureStoreStatusNotEnabled status which represents that this document was rejected, but
+	// it could have ended up in the failure store if it was enabled.
+	FailureStoreStatusNotEnabled FailureStoreStatus = "not_enabled"
+)
+
 func init() {
 	jsoniter.RegisterTypeDecoderFunc("docappender.BulkIndexerResponseStat", func(ptr unsafe.Pointer, iter *jsoniter.Iterator) {
 		iter.ReadObjectCB(func(i *jsoniter.Iterator, s string) bool {
@@ -140,6 +160,15 @@ func init() {
 								item.Index = i.ReadString()
 							case "status":
 								item.Status = i.ReadInt()
+							case "failure_store":
+								// For the stats the only actionable failure store statuses are "used" and "failed".
+								// All other statuses are non actionable and therefore ignored.
+								switch fs := i.ReadString(); FailureStoreStatus(fs) {
+								case FailureStoreStatusUsed:
+									(*((*BulkIndexerResponseStat)(ptr))).FailureStoreUsed++
+								case FailureStoreStatusFailed:
+									(*((*BulkIndexerResponseStat)(ptr))).FailureStoreFailed++
+								}
 							case "error":
 								i.ReadObjectCB(func(i *jsoniter.Iterator, s string) bool {
 									switch s {
@@ -376,7 +405,7 @@ func (b *BulkIndexer) Flush(ctx context.Context) (BulkIndexerResponseStat, error
 		// See: https://github.com/golang/go/issues/51907
 		Body:       bytes.NewReader(b.buf.Bytes()),
 		Header:     make(http.Header),
-		FilterPath: []string{"items.*._index", "items.*.status", "items.*.error.type", "items.*.error.reason"},
+		FilterPath: []string{"items.*._index", "items.*.status", "items.*.failure_store", "items.*.error.type", "items.*.error.reason"},
 		Pipeline:   b.config.Pipeline,
 	}
 	if b.requireDataStream {

bulk_indexer_test.go

@@ -203,3 +203,46 @@ func TestPipeline(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int64(2), stat.Indexed)
 }
+
+func TestBulkIndexer_FailureStore(t *testing.T) {
+	client := docappendertest.NewMockElasticsearchClient(t, func(w http.ResponseWriter, r *http.Request) {
+		_, result := docappendertest.DecodeBulkRequest(r)
+		var i int
+		for _, itemsMap := range result.Items {
+			for k, item := range itemsMap {
+				switch i % 3 {
+				case 0:
+					item.FailureStore = string(docappender.FailureStoreStatusUsed)
+				case 1:
+					item.FailureStore = string(docappender.FailureStoreStatusFailed)
+				case 2:
+					item.FailureStore = string(docappender.FailureStoreStatusUnknown)
+				}
+				itemsMap[k] = item
+				i++
+			}
+		}
+		err := json.NewEncoder(w).Encode(result)
+		require.NoError(t, err)
+	})
+	indexer, err := docappender.NewBulkIndexer(docappender.BulkIndexerConfig{
+		Client: client,
+	})
+	require.NoError(t, err)
+
+	for range 3 {
+		err = indexer.Add(docappender.BulkIndexerItem{
+			Index: "testidx",
+			Body: newJSONReader(map[string]any{
+				"@timestamp": time.Now().Format(docappendertest.TimestampFormat),
+			}),
+		})
+		require.NoError(t, err)
+	}
+
+	stat, err := indexer.Flush(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, int64(3), stat.Indexed)
+	require.Equal(t, int64(1), stat.FailureStoreUsed)
+	require.Equal(t, int64(1), stat.FailureStoreFailed)
+}

go.mod

@@ -42,3 +42,5 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	howett.net/plist v1.0.0 // indirect
 )
+
+replace github.com/elastic/go-elasticsearch/v8 v8.17.0 => github.com/1pkg/go-elasticsearch/v8 v8.0.0-20250131000307-5fee144a41e3

go.sum

@@ -1,12 +1,12 @@
+github.com/1pkg/go-elasticsearch/v8 v8.0.0-20250131000307-5fee144a41e3 h1:o29f7gDStJ4fBNN1PoW+Zwu9j/JSS6JFRfmTHvB0Tr4=
+github.com/1pkg/go-elasticsearch/v8 v8.0.0-20250131000307-5fee144a41e3/go.mod h1:lGMlgKIbYoRvay3xWBeKahAiJOgmFDsjZC39nmO3H64=
 github.com/armon/go-radix v1.0.0 h1:F4z6KzEeeQIMeLFa97iZU6vupzoecKdU5TX24SNppXI=
 github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/elastic/elastic-transport-go/v8 v8.6.0 h1:Y2S/FBjx1LlCv5m6pWAF2kDJAHoSjSRSJCApolgfthA=
 github.com/elastic/elastic-transport-go/v8 v8.6.0/go.mod h1:YLHer5cj0csTzNFXoNQ8qhtGY1GTvSqPnKWKaqQE3Hk=
-github.com/elastic/go-elasticsearch/v8 v8.17.0 h1:e9cWksE/Fr7urDRmGPGp47Nsp4/mvNOrU8As1l2HQQ0=
-github.com/elastic/go-elasticsearch/v8 v8.17.0/go.mod h1:lGMlgKIbYoRvay3xWBeKahAiJOgmFDsjZC39nmO3H64=
 github.com/elastic/go-sysinfo v1.7.1 h1:Wx4DSARcKLllpKT2TnFVdSUJOsybqMYCNQZq1/wO+s0=
 github.com/elastic/go-sysinfo v1.7.1/go.mod h1:i1ZYdU10oLNfRzq4vq62BEwD2fH8KaWh6eh0ikPT9F0=
 github.com/elastic/go-windows v1.0.0/go.mod h1:TsU0Nrp7/y3+VwE82FoZF8gC/XFg/Elz6CcloAxnPgU=