Repository: VictoriaMetrics/VictoriaMetrics
Pull request #605 (2020-07-04 08:52:48)

app/vmalert: add retries to remotewrite
Remotewrite pkg now does limited number of retries if write request failed.
This suppose to make vmalert state persisting more reliable.

New metrics were added to remotewrite in order to track rows/bytes sent/dropped.

defaultFlushInterval was increased from 1s to 5s for sanity reasons.

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605?src=pr&el=h1) Report
> Merging [#605](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605?src=pr&el=desc) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/ff6a0955eb99e1fbb16b60e66153d19dbb8dd9f6&el=desc) will **decrease** coverage by `0.09%`.
> The diff coverage is `63.63%`.

[![Impacted file tree graph](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605/graphs/tree.svg?width=650&height=150&src=pr&token=YND8dYVcXn)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #605      +/-   ##
==========================================
- Coverage   68.08%   67.99%   -0.10%     
==========================================
  Files         154      156       +2     
  Lines       21298    21431     +133     
==========================================
+ Hits        14500    14571      +71     
- Misses       5633     5685      +52     
- Partials     1165     1175      +10     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [app/vmalert/group.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvZ3JvdXAuZ28=) | `75.62% <ø> (+0.46%)` | :arrow_up: |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `63.55% <63.63%> (ø)` | |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `87.97% <0.00%> (-5.50%)` | :arrow_down: |
| [app/vmalert/remotewrite/init.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvaW5pdC5nbw==) | `0.00% <0.00%> (ø)` | |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `68.29% <0.00%> (+0.27%)` | :arrow_up: |
| [lib/storage/index\_db.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvaW5kZXhfZGIuZ28=) | `76.22% <0.00%> (+0.43%)` | :arrow_up: |
| [lib/storage/merge.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvbWVyZ2UuZ28=) | `80.99% <0.00%> (+1.65%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605?src=pr&el=footer). Last update [ff6a095...8d588ac](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/605?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/app/vmalert/group.go b/app/vmalert/group.go
index ec65ea508..cccd0f460 100644
--- a/app/vmalert/group.go
+++ b/app/vmalert/group.go
@@ -144,7 +144,6 @@ var (
 	alertsSent       = metrics.NewCounter(`vmalert_alerts_sent_total`)
 	alertsSendErrors = metrics.NewCounter(`vmalert_alerts_send_errors_total`)
 
-	remoteWriteSent   = metrics.NewCounter(`vmalert_remotewrite_sent_total`)
 	remoteWriteErrors = metrics.NewCounter(`vmalert_remotewrite_errors_total`)
 )
 
@@ -255,7 +254,6 @@ func (e *executor) exec(ctx context.Context, rule Rule, returnSeries bool, inter
 	}
 
 	if len(tss) > 0 && e.rw != nil {
-		remoteWriteSent.Add(len(tss))
 		for _, ts := range tss {
 			if err := e.rw.Push(ts); err != nil {
 				remoteWriteErrors.Inc()
diff --git a/app/vmalert/remotewrite/remotewrite.go b/app/vmalert/remotewrite/remotewrite.go
index f508fded7..caff50a5c 100644
--- a/app/vmalert/remotewrite/remotewrite.go
+++ b/app/vmalert/remotewrite/remotewrite.go
@@ -12,6 +12,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+	"github.com/VictoriaMetrics/metrics"
 	"github.com/golang/snappy"
 )
 
@@ -61,7 +62,7 @@ const (
 	defaultConcurrency   = 4
 	defaultMaxBatchSize  = 1e3
 	defaultMaxQueueSize  = 1e5
-	defaultFlushInterval = time.Second
+	defaultFlushInterval = 5 * time.Second
 	defaultWriteTimeout  = 30 * time.Second
 )
 
@@ -85,6 +86,9 @@ func NewClient(ctx context.Context, cfg Config) (*Client, error) {
 	if cfg.WriteTimeout == 0 {
 		cfg.WriteTimeout = defaultWriteTimeout
 	}
+	if cfg.Transport == nil {
+		cfg.Transport = http.DefaultTransport.(*http.Transport).Clone()
+	}
 	c := &Client{
 		c: &http.Client{
 			Timeout:   cfg.WriteTimeout,
@@ -138,14 +142,11 @@ func (c *Client) Close() error {
 
 func (c *Client) run(ctx context.Context) {
 	ticker := time.NewTicker(c.flushInterval)
-	wr := prompbmarshal.WriteRequest{}
+	wr := &prompbmarshal.WriteRequest{}
 	shutdown := func() {
 		for ts := range c.input {
 			wr.Timeseries = append(wr.Timeseries, ts)
 		}
-		if len(wr.Timeseries) < 1 {
-			return
-		}
 		lastCtx, cancel := context.WithTimeout(context.Background(), defaultWriteTimeout)
 		c.flush(lastCtx, wr)
 		cancel()
@@ -164,44 +165,82 @@ func (c *Client) run(ctx context.Context) {
 				return
 			case <-ticker.C:
 				c.flush(ctx, wr)
-				wr = prompbmarshal.WriteRequest{}
-			case ts := <-c.input:
+			case ts, ok := <-c.input:
+				if !ok {
+					continue
+				}
 				wr.Timeseries = append(wr.Timeseries, ts)
 				if len(wr.Timeseries) >= c.maxBatchSize {
 					c.flush(ctx, wr)
-					wr = prompbmarshal.WriteRequest{}
 				}
 			}
 		}
 	}()
 }
 
-func (c *Client) flush(ctx context.Context, wr prompbmarshal.WriteRequest) {
+var (
+	sentRows     = metrics.NewCounter(`vmalert_remotewrite_sent_rows_total`)
+	sentBytes    = metrics.NewCounter(`vmalert_remotewrite_sent_bytes_total`)
+	droppedRows  = metrics.NewCounter(`vmalert_remotewrite_dropped_rows_total`)
+	droppedBytes = metrics.NewCounter(`vmalert_remotewrite_dropped_bytes_total`)
+)
+
+// flush is a blocking function that marshals WriteRequest and sends
+// it to remote write endpoint. Flush performs limited amount of retries
+// if request fails.
+func (c *Client) flush(ctx context.Context, wr *prompbmarshal.WriteRequest) {
 	if len(wr.Timeseries) < 1 {
 		return
 	}
+	defer prompbmarshal.ResetWriteRequest(wr)
+
 	data, err := wr.Marshal()
 	if err != nil {
 		logger.Errorf("failed to marshal WriteRequest: %s", err)
 		return
 	}
-	req, err := http.NewRequest("POST", c.addr, bytes.NewReader(snappy.Encode(nil, data)))
+
+	const attempts = 5
+	b := snappy.Encode(nil, data)
+	for i := 0; i < attempts; i++ {
+		err := c.send(ctx, b)
+		if err == nil {
+			sentRows.Add(len(wr.Timeseries))
+			sentBytes.Add(len(b))
+			return
+		}
+
+		logger.Errorf("attempt %d to send request failed: %s", i+1, err)
+		// sleeping to avoid remote db hammering
+		time.Sleep(time.Second)
+		continue
+	}
+
+	droppedRows.Add(len(wr.Timeseries))
+	droppedBytes.Add(len(b))
+	logger.Errorf("all %d attempts to send request failed - dropping %d timeseries",
+		attempts, len(wr.Timeseries))
+}
+
+func (c *Client) send(ctx context.Context, data []byte) error {
+	r := bytes.NewReader(data)
+	req, err := http.NewRequest("POST", c.addr, r)
 	if err != nil {
-		logger.Errorf("failed to create new HTTP request: %s", err)
-		return
+		return fmt.Errorf("failed to create new HTTP request: %s", err)
 	}
 	if c.baPass != "" {
 		req.SetBasicAuth(c.baUser, c.baPass)
 	}
 	resp, err := c.c.Do(req.WithContext(ctx))
 	if err != nil {
-		logger.Errorf("error getting response from %s:%s", req.URL, err)
-		return
+		return fmt.Errorf("error while sending request to %s: %s; Data len %d(%d)",
+			req.URL, err, len(data), r.Size())
 	}
 	defer func() { _ = resp.Body.Close() }()
 	if resp.StatusCode != http.StatusNoContent {
 		body, _ := ioutil.ReadAll(resp.Body)
-		logger.Errorf("unexpected response code %d for %s. Response body %s", resp.StatusCode, req.URL, body)
-		return
+		return fmt.Errorf("unexpected response code %d for %s. Response body %q",
+			resp.StatusCode, req.URL, body)
 	}
+	return nil
 }
diff --git a/app/vmselect/prometheus/prometheus.go b/app/vmselect/prometheus/prometheus.go
index eb106f9b0..cf95f4099 100644
--- a/app/vmselect/prometheus/prometheus.go
+++ b/app/vmselect/prometheus/prometheus.go
@@ -791,8 +791,8 @@ func queryRangeHandler(w http.ResponseWriter, query string, start, end, step int
 		return fmt.Errorf("cannot execute query: %w", err)
 	}
 	queryOffset := getLatencyOffsetMilliseconds()
-	if ct-end < queryOffset {
-		result = adjustLastPoints(result)
+	if ct-queryOffset < end {
+		result = adjustLastPoints(result, ct-queryOffset, end)
 	}
 
 	// Remove NaN values as Prometheus does.
@@ -837,40 +837,24 @@ func removeNaNValuesInplace(tss []netstorage.Result) {
 
 var queryRangeDuration = metrics.NewSummary(`vm_request_duration_seconds{path="/api/v1/query_range"}`)
 
-// adjustLastPoints substitutes the last point values with the previous
-// point values, since the last points may contain garbage.
-func adjustLastPoints(tss []netstorage.Result) []netstorage.Result {
-	if len(tss) == 0 {
-		return nil
-	}
-
-	// Search for the last non-NaN value across all the timeseries.
-	lastNonNaNIdx := -1
+// adjustLastPoints substitutes the last point values on the time range (start..end]
+// with the previous point values, since these points may contain garbage.
+func adjustLastPoints(tss []netstorage.Result, start, end int64) []netstorage.Result {
 	for i := range tss {
-		values := tss[i].Values
-		j := len(values) - 1
-		for j >= 0 && math.IsNaN(values[j]) {
+		ts := &tss[i]
+		values := ts.Values
+		timestamps := ts.Timestamps
+		j := len(timestamps) - 1
+		for j >= 0 && timestamps[j] > start {
 			j--
 		}
-		if j > lastNonNaNIdx {
-			lastNonNaNIdx = j
+		j++
+		if j <= 0 {
+			continue
 		}
-	}
-	if lastNonNaNIdx == -1 {
-		// All timeseries contain only NaNs.
-		return nil
-	}
-
-	// Substitute the last two values starting from lastNonNaNIdx
-	// with the previous values for each timeseries.
-	for i := range tss {
-		values := tss[i].Values
-		for j := 0; j < 2; j++ {
-			idx := lastNonNaNIdx + j
-			if idx <= 0 || idx >= len(values) || math.IsNaN(values[idx-1]) {
-				continue
-			}
-			values[idx] = values[idx-1]
+		for j < len(timestamps) && timestamps[j] <= end {
+			values[j] = values[j-1]
+			j++
 		}
 	}
 	return tss


Test patch:
diff --git a/app/victoria-metrics/testdata/graphite/max_lookback_set.json b/app/victoria-metrics/testdata/graphite/max_lookback_set.json
index 23ca9c947..b477d6cff 100644
--- a/app/victoria-metrics/testdata/graphite/max_lookback_set.json
+++ b/app/victoria-metrics/testdata/graphite/max_lookback_set.json
@@ -16,6 +16,8 @@
 	      ["{TIME_S-120s}","3"],
 	      ["{TIME_S-60s}","2"],
 	      ["{TIME_S-30s}","1"],
-	      ["{TIME_S-20s}","1"]
+	      ["{TIME_S-20s}","1"],
+	      ["{TIME_S-10s}","1"],
+	      ["{TIME_S-0s}","1"]
       ]}]}}
 }
diff --git a/app/vmalert/remotewrite/remotewrite_test.go b/app/vmalert/remotewrite/remotewrite_test.go
new file mode 100644
index 000000000..7dab24712
--- /dev/null
+++ b/app/vmalert/remotewrite/remotewrite_test.go
@@ -0,0 +1,102 @@
+package remotewrite
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/golang/snappy"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
+)
+
+func TestClient_Push(t *testing.T) {
+	testSrv := newRWServer()
+	cfg := Config{
+		Addr:         testSrv.URL,
+		MaxBatchSize: 100,
+	}
+	client, err := NewClient(context.Background(), cfg)
+	if err != nil {
+		t.Fatalf("failed to create client: %s", err)
+	}
+	const rowsN = 1e4
+	var sent int
+	for i := 0; i < rowsN; i++ {
+		s := prompbmarshal.TimeSeries{
+			Samples: []prompbmarshal.Sample{{
+				Value:     rand.Float64(),
+				Timestamp: time.Now().Unix(),
+			}},
+		}
+		err := client.Push(s)
+		if err == nil {
+			sent++
+		}
+	}
+	if sent == 0 {
+		t.Fatalf("0 series sent")
+	}
+	if err := client.Close(); err != nil {
+		t.Fatalf("failed to close client: %s", err)
+	}
+	got := testSrv.accepted()
+	if got != sent {
+		t.Fatalf("expected to have %d series; got %d", sent, got)
+	}
+}
+
+func newRWServer() *rwServer {
+	rw := &rwServer{}
+	rw.Server = httptest.NewServer(http.HandlerFunc(rw.handler))
+	return rw
+}
+
+type rwServer struct {
+	// WARN: ordering of fields is important for alignment!
+	// see https://golang.org/pkg/sync/atomic/#pkg-note-BUG
+	acceptedRows uint64
+	*httptest.Server
+}
+
+func (rw *rwServer) accepted() int {
+	return int(atomic.LoadUint64(&rw.acceptedRows))
+}
+
+func (rw *rwServer) err(w http.ResponseWriter, err error) {
+	w.WriteHeader(http.StatusBadRequest)
+	w.Write([]byte(err.Error()))
+}
+
+func (rw *rwServer) handler(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		rw.err(w, fmt.Errorf("bad method %q", r.Method))
+		return
+	}
+	data, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		rw.err(w, fmt.Errorf("body read err: %s", err))
+		return
+	}
+	defer func() { _ = r.Body.Close() }()
+
+	b, err := snappy.Decode(nil, data)
+	if err != nil {
+		rw.err(w, fmt.Errorf("decode err: %s", err))
+		return
+	}
+	wr := &prompb.WriteRequest{}
+	if err := wr.Unmarshal(b); err != nil {
+		rw.err(w, fmt.Errorf("unmarhsal err: %s", err))
+		return
+	}
+	atomic.AddUint64(&rw.acceptedRows, uint64(len(wr.Timeseries)))
+	w.WriteHeader(http.StatusNoContent)
+}
diff --git a/app/vmselect/prometheus/prometheus_test.go b/app/vmselect/prometheus/prometheus_test.go
index 69c2c1237..32bfdd585 100644
--- a/app/vmselect/prometheus/prometheus_test.go
+++ b/app/vmselect/prometheus/prometheus_test.go
@@ -113,3 +113,130 @@ func TestGetTimeError(t *testing.T) {
 	f("-292273086-05-16T16:47:07Z")
 	f("292277025-08-18T07:12:54.999999998Z")
 }
+
+func TestAdjustLastPoints(t *testing.T) {
+	f := func(tss []netstorage.Result, start, end int64, tssExpected []netstorage.Result) {
+		t.Helper()
+		tss = adjustLastPoints(tss, start, end)
+		for i, ts := range tss {
+			for j, value := range ts.Values {
+				expectedValue := tssExpected[i].Values[j]
+				if math.IsNaN(expectedValue) {
+					if !math.IsNaN(value) {
+						t.Fatalf("unexpected result; got %v; want nan", value)
+					}
+				} else if expectedValue != value {
+					t.Fatalf("unexpected result; got %v; want %v", value, expectedValue)
+				}
+			}
+			if !reflect.DeepEqual(ts.Timestamps, tssExpected[i].Timestamps) {
+				t.Fatalf("unexpected result; got %v; want %v", tss, tssExpected)
+			}
+		}
+
+	}
+
+	nan := math.NaN()
+
+	f(nil, 300, 500, nil)
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 4, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, nan, nan},
+		},
+	}, 400, 500, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 4, 4},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, nan, nan},
+		},
+	})
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, nan, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, nan, nan, nan},
+		},
+	}, 300, 500, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 3, 3},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, nan, nan, nan},
+		},
+	})
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, nan, nan, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, nan, nan, nan, nan},
+		},
+	}, 500, 300, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, nan, nan, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, nan, nan, nan, nan},
+		},
+	})
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 4, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400},
+			Values:     []float64{1, 2, 3, 4},
+		},
+	}, 400, 500, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 4, 4},
+		},
+		{
+			Timestamps: []int64{100, 200, 300, 400},
+			Values:     []float64{1, 2, 3, 4},
+		},
+	})
+
+	f([]netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, nan, nan},
+		},
+		{
+			Timestamps: []int64{100, 200, 300},
+			Values:     []float64{1, 2, nan},
+		},
+	}, 300, 600, []netstorage.Result{
+		{
+			Timestamps: []int64{100, 200, 300, 400, 500},
+			Values:     []float64{1, 2, 3, 3, 3},
+		},
+		{
+			Timestamps: []int64{100, 200, 300},
+			Values:     []float64{1, 2, nan},
+		},
+	})
+}