Clean up / document metrics monitor fields (#39413)

Document the semantics of many metrics monitoring variables, and rename some metrics APIs to more clearly indicate their function. As a side effect, fix several metrics reporting / publishing bugs in the Elasticsearch output, including #39146.

Add the `output.events.dead_letter` metric to distinguish events that were ingested to the dead letter index after a fatal error (previously these events were just reported as "acked").

This could have been a shorter fix, but it was hard to properly test since the metrics were changed from two separate functions with a lot of special cases. I ended up reorganizing the Elasticsearch `Publish` helpers to make the logic more clear. The new layout makes it much easier to test the error handling and metrics reporting.

The bugs fixed by this refactor are:
- When a batch was split, the events in it were not reported to the observer via `RetryableErrors`. This caused `activeEvents` to increase permanently even after the events were handled.
- When a previously-failed event was ingested to the dead letter index as a raw string, it was reported to the observer as `Acked` (success). The new logic creates a new `dead_letter` metric specifically for this case.
- When a previously-failed event encountered a fatal error ingesting to the dead letter index:
  * It was reported to the observer as both a permanent error and a retryable error, giving incorrect event counts. It's now reported as just a permanent error.
  * It was added to the retry list anyway, which would block all ingestion if the error really was fatal since the queue could never advance past that event. It's now dropped, the same as with a permanent error in the main index.
- If the Elasticsearch bulk index response was invalid, the associated events were dropped and reported as acknowledged

Author: faec

filebeat/input/filestream/internal/task/group_test.go

@@ -195,8 +195,8 @@ func TestGroup_Go(t *testing.T) {
 	})
 
 	t.Run("without limit, all goroutines run", func(t *testing.T) {
-		// 100 <= limit <= 100000
-		limit := rand.Int63n(100000-100) + 100
+		// 100 <= limit <= 10000
+		limit := rand.Int63n(10000-100) + 100
 		t.Logf("running %d goroutines", limit)
 		g := NewGroup(uint64(limit), time.Second, noopLogger{}, "")
 

libbeat/esleg/eslegclient/bulkapi.go

@@ -23,7 +23,6 @@ import (
 	"encoding/json"
 	"errors"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"strings"
 
@@ -60,16 +59,16 @@ type bulkRequest struct {
 	requ *http.Request
 }
 
-// BulkResult contains the result of a bulk API request.
-type BulkResult json.RawMessage
+// BulkResponse contains the result of a bulk API request.
+type BulkResponse json.RawMessage
 
 // Bulk performs many index/delete operations in a single API call.
 // Implements: http://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
 func (conn *Connection) Bulk(
 	ctx context.Context,
 	index, docType string,
 	params map[string]string, body []interface{},
-) (int, BulkResult, error) {
+) (int, BulkResponse, error) {
 	if len(body) == 0 {
 		return 0, nil, nil
 	}
@@ -142,7 +141,7 @@ func (r *bulkRequest) reset(body BodyEncoder) {
 
 	rc, ok := bdy.(io.ReadCloser)
 	if !ok && body != nil {
-		rc = ioutil.NopCloser(bdy)
+		rc = io.NopCloser(bdy)
 	}
 
 	switch v := bdy.(type) {
@@ -160,9 +159,9 @@ func (r *bulkRequest) reset(body BodyEncoder) {
 	body.AddHeader(&r.requ.Header)
 }
 
-func (conn *Connection) sendBulkRequest(requ *bulkRequest) (int, BulkResult, error) {
+func (conn *Connection) sendBulkRequest(requ *bulkRequest) (int, BulkResponse, error) {
 	status, resp, err := conn.execHTTPRequest(requ.requ)
-	return status, BulkResult(resp), err
+	return status, BulkResponse(resp), err
 }
 
 func bulkEncode(log *logp.Logger, out BulkWriter, body []interface{}) error {

libbeat/monitoring/report/elasticsearch/client.go

@@ -224,7 +224,7 @@ func getMonitoringIndexName() string {
 	return fmt.Sprintf(".monitoring-beats-%v-%s", version, date)
 }
 
-func logBulkFailures(log *logp.Logger, result eslegclient.BulkResult, events []report.Event) {
+func logBulkFailures(log *logp.Logger, result eslegclient.BulkResponse, events []report.Event) {
 	var response struct {
 		Items []map[string]map[string]interface{} `json:"items"`
 	}

libbeat/outputs/console/console.go

@@ -111,8 +111,8 @@ func (c *console) Publish(_ context.Context, batch publisher.Batch) error {
 	c.writer.Flush()
 	batch.ACK()
 
-	st.Dropped(dropped)
-	st.Acked(len(events) - dropped)
+	st.PermanentErrors(dropped)
+	st.AckedEvents(len(events) - dropped)
 
 	return nil
 }

libbeat/outputs/discard/discard.go

@@ -70,7 +70,7 @@ func (out *discardOutput) Publish(_ context.Context, batch publisher.Batch) erro
 	st := out.observer
 	events := batch.Events()
 	st.NewBatch(len(events))
-	st.Acked(len(events))
+	st.AckedEvents(len(events))
 	return nil
 }