monitoring metrics libbeat.output.active < 0

[carsonip]

APM Server version (apm-server version): confirmed on 8.15.0 and 8.15 branch

Description of the problem including expected versus actual behavior:

libbeat.output.events.active can be negative

"libbeat": {
        "output": {
            "events": {
                "acked": 2,
                "active": -1,
                "batches": 3,
                "failed": 2,
                "toomany": 0,
                "total": 3
            },
            "type": "elasticsearch",
            "write": {
                "bytes": 1501
            }
        },
        "pipeline": {
            "events": {
                "total": 3
            }
        }
    },

Steps to reproduce:

Please include a minimal but complete recreation of the problem,
including server configuration, agent(s) used, etc. The easier you make it
for us to reproduce it, the more likely that somebody will take the time to
look at it.

1. run mock ES server (see code below)
2. send a simple intake v2 request to apm-server running with monitoring
3. wait for event and its aggregated metrics to be processed by go-docappender
4. ES server should return 200, 400 and 500.
5. Check localhost:5066/stats

mock ES code

package main

import (
	"bufio"
	"compress/gzip"
	"fmt"
	"log"
	"net/http"
	"sync/atomic"
	"time"

	"go.elastic.co/fastjson"
)

func main() {
	var num atomic.Int64
	mux := http.NewServeMux()
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("X-Elastic-Product", "Elasticsearch")
		fmt.Fprintln(w, `{
    "name": "3a69e54dc7a0",
    "cluster_name": "docker-cluster",
    "cluster_uuid": "fJBzdwAYQ2ecCDYKkoq_Vw",
    "version": {
        "number": "8.15.1-SNAPSHOT",
        "build_flavor": "default",
        "build_type": "docker",
        "build_hash": "dcba83954afd584d495a1919af52f644a2564f53",
        "build_date": "2024-08-29T15:49:24.262263441Z",
        "build_snapshot": true,
        "lucene_version": "9.11.1",
        "minimum_wire_compatibility_version": "7.17.0",
        "minimum_index_compatibility_version": "7.0.0"
    },
    "tagline": "You Know, for Search"
}`)
	})

	mux.HandleFunc("/_bulk", func(w http.ResponseWriter, r *http.Request) {
		id := num.Add(1)
		gzr, _ := gzip.NewReader(r.Body)
		defer gzr.Close()

		var jsonw fastjson.Writer
		jsonw.RawString(`{"items":[`)
		first := true

		scanner := bufio.NewScanner(gzr)
		var n int64

		for scanner.Scan() && len(scanner.Bytes()) != 0 { // index
			if first {
				first = false
			} else {
				jsonw.RawByte(',')
			}
			jsonw.RawString(`{"create":{"status":201}}`)
			n++
		}
		jsonw.RawString(`]}`)

		w.Header().Set("X-Elastic-Product", "Elasticsearch")

		switch (id - 1) % 3 {
		case 0:
			w.WriteHeader(200)
			w.Write(jsonw.Bytes())
		case 1:
			w.WriteHeader(400)
		case 2:
			w.WriteHeader(500)
		}

	})
	s := &http.Server{
		Addr:           ":3000",
		Handler:        mux,
		ReadTimeout:    10 * time.Second,
		WriteTimeout:   10 * time.Second,
		MaxHeaderBytes: 1 << 20,
	}
	log.Fatal(s.ListenAndServe())
}

Provide logs (if relevant):

[simitt]

@carsonip are you aware whether this is a newly introduced bug in 8.15.0 or whether this has been the case already in previous versions?

[carsonip]

I just double checked, there is nothing wrong with apm-server. The mock ES was wrong. Sorry for the noise.

It returns more success document responses in the bulk response than in the request, causing active to be <0. I've attached the updated code. It should only count number of index lines, and skip the document line with scanner.Scan() in bulk request.

Updated mock ES

package main

import (
	"bufio"
	"compress/gzip"
	"fmt"
	"log"
	"net/http"
	"sync/atomic"
	"time"

	"go.elastic.co/fastjson"
)

func main() {
	var num atomic.Int64
	mux := http.NewServeMux()
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("X-Elastic-Product", "Elasticsearch")
		fmt.Fprintln(w, `{
    "name": "3a69e54dc7a0",
    "cluster_name": "docker-cluster",
    "cluster_uuid": "fJBzdwAYQ2ecCDYKkoq_Vw",
    "version": {
        "number": "8.15.1-SNAPSHOT",
        "build_flavor": "default",
        "build_type": "docker",
        "build_hash": "dcba83954afd584d495a1919af52f644a2564f53",
        "build_date": "2024-08-29T15:49:24.262263441Z",
        "build_snapshot": true,
        "lucene_version": "9.11.1",
        "minimum_wire_compatibility_version": "7.17.0",
        "minimum_index_compatibility_version": "7.0.0"
    },
    "tagline": "You Know, for Search"
}`)
	})

	mux.HandleFunc("/_bulk", func(w http.ResponseWriter, r *http.Request) {
		id := num.Add(1)
		gzr, _ := gzip.NewReader(r.Body)
		defer gzr.Close()

		var jsonw fastjson.Writer
		jsonw.RawString(`{"items":[`)
		first := true

		scanner := bufio.NewScanner(gzr)
		var n int64

		for scanner.Scan() && len(scanner.Bytes()) != 0 { // index
			if first {
				first = false
			} else {
				jsonw.RawByte(',')
			}
			jsonw.RawString(`{"create":{"status":201}}`)
			n++
			scanner.Scan()
		}
		jsonw.RawString(`]}`)

		w.Header().Set("X-Elastic-Product", "Elasticsearch")

		switch (id - 1) % 3 {
		case 0:
			w.WriteHeader(200)
			w.Write(jsonw.Bytes())
		case 1:
			w.WriteHeader(400)
		case 2:
			w.WriteHeader(500)
		}

	})
	s := &http.Server{
		Addr:           ":3000",
		Handler:        mux,
		ReadTimeout:    10 * time.Second,
		WriteTimeout:   10 * time.Second,
		MaxHeaderBytes: 1 << 20,
	}
	log.Fatal(s.ListenAndServe())
}