Adding functionality to death letter index instead of dropping events.

libbeat/outputs/elasticsearch/client.go

@@ -45,24 +45,26 @@ type Client struct {
 	index    outputs.IndexSelector
 	pipeline *outil.Selector
 
-	observer outputs.Observer
+	observer           outputs.Observer
+	NonIndexableAction string
 
 	log *logp.Logger
 }
 
 // ClientSettings contains the settings for a client.
 type ClientSettings struct {
 	eslegclient.ConnectionSettings
-	Index    outputs.IndexSelector
-	Pipeline *outil.Selector
-	Observer outputs.Observer
+	Index              outputs.IndexSelector
+	Pipeline           *outil.Selector
+	Observer           outputs.Observer
+	NonIndexableAction string
 }
 
 type bulkResultStats struct {
 	acked        int // number of events ACKed by Elasticsearch
 	duplicates   int // number of events failed with `create` due to ID already being indexed
 	fails        int // number of failed events (can be retried)
-	nonIndexable int // number of failed events (not indexable -> must be dropped)
+	nonIndexable int // number of failed events (not indexable)
 	tooMany      int // number of events receiving HTTP 429 Too Many Requests
 }
 
@@ -123,11 +125,11 @@ func NewClient(
 	}
 
 	client := &Client{
-		conn:     *conn,
-		index:    s.Index,
-		pipeline: pipeline,
-
-		observer: s.Observer,
+		conn:               *conn,
+		index:              s.Index,
+		pipeline:           pipeline,
+		observer:           s.Observer,
+		NonIndexableAction: s.NonIndexableAction,
 
 		log: logp.NewLogger("elasticsearch"),
 	}
@@ -235,7 +237,7 @@ func (client *Client) publishEvents(ctx context.Context, data []publisher.Event)
 		failedEvents = data
 		stats.fails = len(failedEvents)
 	} else {
-		failedEvents, stats = bulkCollectPublishFails(client.log, result, data)
+		failedEvents, stats = bulkCollectPublishFails(client.log, result, data, client.NonIndexableAction)
 	}
 
 	failed := len(failedEvents)
@@ -368,6 +370,7 @@ func bulkCollectPublishFails(
 	log *logp.Logger,
 	result eslegclient.BulkResult,
 	data []publisher.Event,
+	action string,
 ) ([]publisher.Event, bulkResultStats) {
 	reader := newJSONReader(result)
 	if err := bulkReadToItems(reader); err != nil {
@@ -401,10 +404,31 @@ func bulkCollectPublishFails(
 			if status == http.StatusTooManyRequests {
 				stats.tooMany++
 			} else {
-				// hard failure, don't collect
-				log.Warnf("Cannot index event %#v (status=%v): %s", data[i], status, msg)
-				stats.nonIndexable++
-				continue
+				// hard failure, apply policy action
+				result, _ := data[i].Content.Meta.HasKey("deathlettered")
+				if result {
+					stats.nonIndexable++
+					log.Errorf("Can't deliver to death letter index event %#v (status=%v): %s", data[i], status, msg)
+					// poison pill - this will clog the pipeline if the underlying failure is non transient.
+				} else if action == "death_letter_index" {
+					log.Warnf("Cannot index event %#v (status=%v): %s, trying death letter index", data[i], status, msg)
+					if data[i].Content.Meta == nil {
+						data[i].Content.Meta = common.MapStr{
+							"deathlettered": true,
+						}
+					} else {
+						_, _ = data[i].Content.Meta.Put("deathlettered", true)
+					}
+					data[i].Content.Fields = common.MapStr{
+						"message":       data[i].Content.Fields.String(),
+						"error.type":    status,
+						"error.message": string(msg),
+					}
+				} else { // drop
+					stats.nonIndexable++
+					log.Warnf("Cannot index event %#v (status=%v): %s, dropping event!", data[i], status, msg)
+					continue
+				}
 			}
 		}
 

libbeat/outputs/elasticsearch/client_integration_test.go

@@ -200,6 +200,69 @@ func TestClientPublishEventWithPipeline(t *testing.T) {
 	assert.Equal(t, 1, getCount("testfield:0")) // no pipeline
 }
 
+func TestClientBulkPublishEventsWithDeathletterIndex(t *testing.T) {
+	type obj map[string]interface{}
+
+	logp.TestingSetup(logp.WithSelectors("elasticsearch"))
+
+	index := "beat-int-test-dli-index"
+	deathletterIndex := "beat-int-test-dli-deathletter-index"
+
+	output, client := connectTestEsWithoutStats(t, obj{
+		"index": index,
+		"non_indexable_policy": NonIndexablePolicy{
+			Action: "death_letter_index",
+			Index:  deathletterIndex,
+		},
+	})
+	client.conn.Delete(index, "", "", nil)
+	client.conn.Delete(deathletterIndex, "", "", nil)
+
+	err := output.Publish(context.Background(), outest.NewBatch(beat.Event{
+		Timestamp: time.Now(),
+		Fields: common.MapStr{
+			"type":      "libbeat",
+			"message":   "Test message 1",
+			"testfield": 0,
+		},
+	}))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	batch := outest.NewBatch(beat.Event{
+		Timestamp: time.Now(),
+		Fields: common.MapStr{
+			"type":      "libbeat",
+			"message":   "Test message 2",
+			"testfield": "foo0",
+		},
+	})
+	err = output.Publish(context.Background(), batch)
+	if err == nil {
+		t.Fatal("Expecting mapping conflict")
+	}
+	_, _, err = client.conn.Refresh(deathletterIndex)
+	if err == nil {
+		t.Fatal("expecting index to not exist yet")
+	}
+	err = output.Publish(context.Background(), batch)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, _, err = client.conn.Refresh(index)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, _, err = client.conn.Refresh(deathletterIndex)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+}
+
 func TestClientBulkPublishEventsWithPipeline(t *testing.T) {
 	type obj map[string]interface{}
 

libbeat/outputs/elasticsearch/client_test.go

@@ -54,7 +54,7 @@ func TestCollectPublishFailsNone(t *testing.T) {
 		events[i] = publisher.Event{Content: beat.Event{Fields: event}}
 	}
 
-	res, _ := bulkCollectPublishFails(logp.L(), response, events)
+	res, _ := bulkCollectPublishFails(logp.L(), response, events, "drop")
 	assert.Equal(t, 0, len(res))
 }
 
@@ -71,12 +71,97 @@ func TestCollectPublishFailMiddle(t *testing.T) {
 	eventFail := publisher.Event{Content: beat.Event{Fields: common.MapStr{"field": 2}}}
 	events := []publisher.Event{event, eventFail, event}
 
-	res, stats := bulkCollectPublishFails(logp.L(), response, events)
+	res, stats := bulkCollectPublishFails(logp.L(), response, events, "drop")
 	assert.Equal(t, 1, len(res))
 	if len(res) == 1 {
 		assert.Equal(t, eventFail, res[0])
 	}
-	assert.Equal(t, stats, bulkResultStats{acked: 2, fails: 1, tooMany: 1})
+	assert.Equal(t, bulkResultStats{acked: 2, fails: 1, tooMany: 1}, stats)
+}
+
+func TestCollectPublishFailDeathLetterQueue(t *testing.T) {
+	response := []byte(`
+    { "items": [
+      {"create": {"status": 200}},
+      {"create": {
+		  "error" : {
+			"root_cause" : [
+			  {
+				"type" : "mapper_parsing_exception",
+				"reason" : "failed to parse field [bar] of type [long] in document with id '1'. Preview of field's value: 'bar1'"
+			  }
+			],
+			"type" : "mapper_parsing_exception",
+			"reason" : "failed to parse field [bar] of type [long] in document with id '1'. Preview of field's value: 'bar1'",
+			"caused_by" : {
+			  "type" : "illegal_argument_exception",
+			  "reason" : "For input string: \"bar1\""
+			}
+		  },
+		  "status" : 400
+		}
+      },
+      {"create": {"status": 200}}
+    ]}
+  `)
+
+	event := publisher.Event{Content: beat.Event{Fields: common.MapStr{"bar": 1}}}
+	eventFail := publisher.Event{Content: beat.Event{Fields: common.MapStr{"bar": "bar1"}}}
+	events := []publisher.Event{event, eventFail, event}
+
+	res, stats := bulkCollectPublishFails(logp.L(), response, events, "death_letter_index")
+	assert.Equal(t, 1, len(res))
+	if len(res) == 1 {
+		expected := publisher.Event{
+			Content: beat.Event{
+				Fields: common.MapStr{
+					"message":       "{\"bar\":\"bar1\"}",
+					"error.type":    400,
+					"error.message": "{\n\t\t\t\"root_cause\" : [\n\t\t\t  {\n\t\t\t\t\"type\" : \"mapper_parsing_exception\",\n\t\t\t\t\"reason\" : \"failed to parse field [bar] of type [long] in document with id '1'. Preview of field's value: 'bar1'\"\n\t\t\t  }\n\t\t\t],\n\t\t\t\"type\" : \"mapper_parsing_exception\",\n\t\t\t\"reason\" : \"failed to parse field [bar] of type [long] in document with id '1'. Preview of field's value: 'bar1'\",\n\t\t\t\"caused_by\" : {\n\t\t\t  \"type\" : \"illegal_argument_exception\",\n\t\t\t  \"reason\" : \"For input string: \\\"bar1\\\"\"\n\t\t\t}\n\t\t  }",
+				},
+				Meta: common.MapStr{
+					"deathlettered": true,
+				},
+			},
+		}
+		assert.Equal(t, expected, res[0])
+	}
+	assert.Equal(t, bulkResultStats{acked: 2, fails: 1, nonIndexable: 0}, stats)
+}
+
+func TestCollectPublishFailDrop(t *testing.T) {
+	response := []byte(`
+    { "items": [
+      {"create": {"status": 200}},
+      {"create": {
+		  "error" : {
+			"root_cause" : [
+			  {
+				"type" : "mapper_parsing_exception",
+				"reason" : "failed to parse field [bar] of type [long] in document with id '1'. Preview of field's value: 'bar1'"
+			  }
+			],
+			"type" : "mapper_parsing_exception",
+			"reason" : "failed to parse field [bar] of type [long] in document with id '1'. Preview of field's value: 'bar1'",
+			"caused_by" : {
+			  "type" : "illegal_argument_exception",
+			  "reason" : "For input string: \"bar1\""
+			}
+		  },
+		  "status" : 400
+		}
+      },
+      {"create": {"status": 200}}
+    ]}
+  `)
+
+	event := publisher.Event{Content: beat.Event{Fields: common.MapStr{"bar": 1}}}
+	eventFail := publisher.Event{Content: beat.Event{Fields: common.MapStr{"bar": "bar1"}}}
+	events := []publisher.Event{event, eventFail, event}
+
+	res, stats := bulkCollectPublishFails(logp.L(), response, events, "drop")
+	assert.Equal(t, 0, len(res))
+	assert.Equal(t, bulkResultStats{acked: 2, fails: 0, nonIndexable: 1}, stats)
 }
 
 func TestCollectPublishFailAll(t *testing.T) {
@@ -91,7 +176,7 @@ func TestCollectPublishFailAll(t *testing.T) {
 	event := publisher.Event{Content: beat.Event{Fields: common.MapStr{"field": 2}}}
 	events := []publisher.Event{event, event, event}
 
-	res, stats := bulkCollectPublishFails(logp.L(), response, events)
+	res, stats := bulkCollectPublishFails(logp.L(), response, events, "drop")
 	assert.Equal(t, 3, len(res))
 	assert.Equal(t, events, res)
 	assert.Equal(t, stats, bulkResultStats{fails: 3, tooMany: 3})
@@ -132,7 +217,7 @@ func TestCollectPipelinePublishFail(t *testing.T) {
 	event := publisher.Event{Content: beat.Event{Fields: common.MapStr{"field": 2}}}
 	events := []publisher.Event{event}
 
-	res, _ := bulkCollectPublishFails(logp.L(), response, events)
+	res, _ := bulkCollectPublishFails(logp.L(), response, events, "drop")
 	assert.Equal(t, 1, len(res))
 	assert.Equal(t, events, res)
 }
@@ -150,7 +235,7 @@ func BenchmarkCollectPublishFailsNone(b *testing.B) {
 	events := []publisher.Event{event, event, event}
 
 	for i := 0; i < b.N; i++ {
-		res, _ := bulkCollectPublishFails(logp.L(), response, events)
+		res, _ := bulkCollectPublishFails(logp.L(), response, events, "")
 		if len(res) != 0 {
 			b.Fail()
 		}
@@ -171,7 +256,7 @@ func BenchmarkCollectPublishFailMiddle(b *testing.B) {
 	events := []publisher.Event{event, eventFail, event}
 
 	for i := 0; i < b.N; i++ {
-		res, _ := bulkCollectPublishFails(logp.L(), response, events)
+		res, _ := bulkCollectPublishFails(logp.L(), response, events, "")
 		if len(res) != 1 {
 			b.Fail()
 		}
@@ -191,7 +276,7 @@ func BenchmarkCollectPublishFailAll(b *testing.B) {
 	events := []publisher.Event{event, event, event}
 
 	for i := 0; i < b.N; i++ {
-		res, _ := bulkCollectPublishFails(logp.L(), response, events)
+		res, _ := bulkCollectPublishFails(logp.L(), response, events, "drop")
 		if len(res) != 3 {
 			b.Fail()
 		}