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

appender.go

@@ -409,13 +409,13 @@ 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
-	)
+
+	failureStore := resp.FailureStore
 	docsIndexed = resp.Indexed
 	var failedCount map[BulkIndexerResponseItem]int
 	if len(resp.FailedDocs) > 0 {
@@ -483,11 +483,32 @@ func (a *Appender) flush(ctx context.Context, bulkIndexer *BulkIndexer) error {
 			metric.WithAttributes(attribute.String("status", "FailedServer")),
 		)
 	}
+	if failureStore.Used > 0 {
+		a.addCount(failureStore.Used, nil,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(attribute.String("failure_store", "used")),
+		)
+	}
+	if failureStore.Failed > 0 {
+		a.addCount(failureStore.Failed, nil,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(attribute.String("failure_store", "failed")),
+		)
+	}
+	if failureStore.Unknown > 0 {
+		a.addCount(failureStore.Unknown, nil,
+			a.metrics.docsIndexed,
+			metric.WithAttributes(attribute.String("failure_store", "unknown")),
+		)
+	}
 	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", failureStore.Used),
+		zap.Int64("docs_failure_store_failed", failureStore.Failed),
+		zap.Int64("docs_failure_store_unknown", failureStore.Unknown),
 	)
 	if a.otelTracingEnabled() && span.IsRecording() {
 		span.SetStatus(codes.Ok, "")

appender_test.go

@@ -753,7 +753,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 +805,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 +1371,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,12 @@ type BulkIndexerResponseStat struct {
 	Indexed int64
 	// RetriedDocs contains the total number of retried documents.
 	RetriedDocs int64
+	// FailureStore contains the stats for the documents indexed to failure store.
+	FailureStore struct {
+		Used    int64
+		Failed  int64
+		Unknown int64
+	}
 	// GreatestRetry contains the greatest observed retry count in the entire
 	// bulk request.
 	GreatestRetry int
@@ -140,6 +146,21 @@ func init() {
 								item.Index = i.ReadString()
 							case "status":
 								item.Status = i.ReadInt()
+							case "failure_store":
+								// For the stats the only actionable failure store statuses:
+								// "used" which represents that this document was stored in the failure store successfully.
+								// "failed" which represents that this document was rejected from the failure store.
+								// "not_applicable_or_unknown"represents that there is no information about this response or that the failure store is not applicable.
+								// Ignored non actionable statuses:
+								// "not_enabled" which represents that this document was rejected, but it could have ended up in the failure store if it was enabled.
+								switch fs := i.ReadString(); fs {
+								case "used":
+									(*((*BulkIndexerResponseStat)(ptr))).FailureStore.Used++
+								case "failed":
+									(*((*BulkIndexerResponseStat)(ptr))).FailureStore.Failed++
+								case "not_applicable_or_unknown":
+									(*((*BulkIndexerResponseStat)(ptr))).FailureStore.Unknown++
+								}
 							case "error":
 								i.ReadObjectCB(func(i *jsoniter.Iterator, s string) bool {
 									switch s {