8704: Reproducing unexpected EOF on vmagent

8704.md

VictoriaMetrics/VictoriaMetrics#8704

1. Checkout VictoriaMetrics repo
2. Build binaries

make vmagent
make victoria-metrics

3. Run storage:

./bin/victoria-metrics 

4. Run vmagent

./bin/vmagent    -remoteWrite.url=http://0.0.0.0:8428/api/v1/write -maxConcurrentInserts=2 -insert.maxQueueDuration=10s

5. Run load script with commented out client's timeout:

go run main.go

You should obeserve a lot of succesful writes and some rare warnings like:

2025-04-12T13:49:47.551Z	warn	VictoriaMetrics/app/vmagent/main.go:291	remoteAddr: "127.0.0.1:60319"; requestURI: /api/v1/write; cannot read compressed request in 10 seconds: cannot process insert request for 10.000 seconds because 2 concurrent insert requests are executed. Possible solutions: to reduce workload; to increase compute resources at the server; to increase -insert.maxQueueDuration; to increase -maxConcurrentInserts

6. Uncomment timeout and run load script:

You should obeserve a lot of succesful writes and some rare warnings like:

2025-04-12T13:49:47.374Z	warn	VictoriaMetrics/app/vmagent/main.go:291	remoteAddr: "127.0.0.1:60308"; requestURI: /api/v1/write; cannot read compressed request in 10 seconds: unexpected EOF

2025-04-12T13:49:47.551Z	warn	VictoriaMetrics/app/vmagent/main.go:291	remoteAddr: "127.0.0.1:60319"; requestURI: /api/v1/write; cannot read compressed request in 10 seconds: cannot process insert request for 10.000 seconds because 2 concurrent insert requests are executed. Possible solutions: to reduce workload; to increase compute resources at the server; to increase -insert.maxQueueDuration; to increase -maxConcurrentInserts

main.go

package main

import (
	"bytes"
	"fmt"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
	"github.com/golang/snappy"
)

func main() {

	var wg sync.WaitGroup

	concurrencyCh := make(chan struct{}, 300)

	c := &http.Client{
		// Timeout: time.Second,
	}

	for i := 0; i < 300; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			for {
				concurrencyCh <- struct{}{}
				handleRequest(c)
				<-concurrencyCh
			}
		}()
	}

	wg.Wait()
}

func handleRequest(c *http.Client) {
	payload := genPayload(100000)

	req, _ := http.NewRequest("POST", `http://127.0.0.1:8429/api/v1/write`, bytes.NewReader(payload))
	req.Header.Set("Content-Type", "application/x-protobuf")
	req.Header.Set("Connection", "keep-alive")
	req.Header.Set("User-Agent", "aUserAgent")
	req.Header.Set("Content-Encoding", "snappy")
	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
	req.Header.Set("Content-Length", fmt.Sprintf("%d", len(payload)))

	resp, err := c.Do(req)
	if err != nil {
		log.Println("http: do: ", err)
		return
	}
	defer resp.Body.Close()

	if resp.StatusCode != 204 {
		log.Println("http: do: ", resp.Status)
	}
}

func genPayload(size int) []byte {
	r := &prompbmarshal.WriteRequest{}
	for i := 0; i < size; i++ {
		r.Timeseries = append(r.Timeseries, prompbmarshal.TimeSeries{
			Labels: []prompbmarshal.Label{
				{Name: "__name__", Value: fmt.Sprintf(`a_metric_case_%d`, i)},
				{Name: "foo", Value: "fooVal"},
				{Name: "bar", Value: "barVal"},
				{Name: "baz", Value: "bazVal"},
			},
			Samples: []prompbmarshal.Sample{
				{Value: 1, Timestamp: time.Now().UnixMilli()},
			},
		})
	}

	payload := r.MarshalProtobuf(nil)
	return snappy.Encode(nil, payload)
}