Repository: VictoriaMetrics/VictoriaMetrics
Pull request #4146 (2023-04-19 05:26:30)

vmalert: retry datasource requests with EOF or unexpected EOF errors
Retry failed read request on the closed connection one more time. This may improve rules execution reliability when connection between vmalert and datasource closes unexpectedly.

codecov[bot]: ## [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/4146?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
Patch coverage: **`100.00`**% and project coverage change: **`-0.03`** :warning:
> Comparison is base [(`472fe3f`)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/472fe3fd03ca9a708ebe940ef6d448e67c1dc7bf?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) 61.69% compared to head [(`c5af50e`)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/4146?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) 61.67%.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #4146      +/-   ##
==========================================
- Coverage   61.69%   61.67%   -0.03%     
==========================================
  Files         323      323              
  Lines       58335    58340       +5     
==========================================
- Hits        35991    35982       -9     
- Misses      20380    20392      +12     
- Partials     1964     1966       +2     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/4146?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Coverage Δ | |
|---|---|---|
| [app/vmalert/datasource/vm.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/4146?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvZGF0YXNvdXJjZS92bS5nbw==) | `81.35% <100.00%> (+0.82%)` | :arrow_up: |

... and [6 files with indirect coverage changes](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/4146/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)

</details>

[:umbrella: View full report in Codecov by Sentry](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/4146?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).


Patch:
diff --git a/app/vmalert/datasource/vm.go b/app/vmalert/datasource/vm.go
index e9d579c69..fc538266f 100644
--- a/app/vmalert/datasource/vm.go
+++ b/app/vmalert/datasource/vm.go
@@ -2,6 +2,7 @@ package datasource
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -162,6 +163,11 @@ func (s *VMStorage) do(ctx context.Context, req *http.Request) (*http.Response,
 		logger.Infof("DEBUG datasource request: executing %s request with params %q", req.Method, req.URL.RawQuery)
 	}
 	resp, err := s.c.Do(req.WithContext(ctx))
+	if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+		// something in the middle between client and datasource might be closing
+		// the connection. So we do a one more attempt in hope request will succeed.
+		resp, err = s.c.Do(req.WithContext(ctx))
+	}
 	if err != nil {
 		return nil, fmt.Errorf("error getting response from %s: %w", req.URL.Redacted(), err)
 	}
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 60772c115..bb425540a 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -26,6 +26,7 @@ The following tip changes can be tested by building VictoriaMetrics components f
 * BUGFIX: reduce the probability of sudden increase in the number of small parts on systems with small number of CPU cores.
 * BUGFIX: [vmctl](https://docs.victoriametrics.com/vmctl.html): fix performance issue when migrating data from VictoriaMetrics according to [these docs](https://docs.victoriametrics.com/vmctl.html#migrating-data-from-victoriametrics). Add the ability to speed up the data migration via `--vm-native-disable-retries` command-line flag. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4092).
 * BUGFIX: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): fix a panic when the duration in the query contains uppercase `M` suffix. Such a suffix isn't allowed to use in durations, since it clashes with `a million` suffix, e.g. it isn't clear whether `rate(metric[5M])` means rate over 5 minutes, 5 months or 5 million seconds. See [this](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3589) and [this](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4120) issues.
+* BUGFIX: [vmalert](https://docs.victoriametrics.com/vmalert.html): retry failed read request on the closed connection one more time. This improves rules execution reliability when connection between vmalert and datasource closes unexpectedly.
 
 ## [v1.90.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.90.0)
 


Test patch:
diff --git a/app/vmalert/datasource/vm_test.go b/app/vmalert/datasource/vm_test.go
index e9d728763..d6413ab6e 100644
--- a/app/vmalert/datasource/vm_test.go
+++ b/app/vmalert/datasource/vm_test.go
@@ -38,7 +38,7 @@ func TestVMInstantQuery(t *testing.T) {
 	mux.HandleFunc("/render", func(w http.ResponseWriter, request *http.Request) {
 		c++
 		switch c {
-		case 8:
+		case 7:
 			w.Write([]byte(`[{"target":"constantLine(10)","tags":{"name":"constantLine(10)"},"datapoints":[[10,1611758343],[10,1611758373],[10,1611758403]]}]`))
 		}
 	})
@@ -62,21 +62,18 @@ func TestVMInstantQuery(t *testing.T) {
 		}
 		switch c {
 		case 0:
-			conn, _, _ := w.(http.Hijacker).Hijack()
-			_ = conn.Close()
-		case 1:
 			w.WriteHeader(500)
-		case 2:
+		case 1:
 			w.Write([]byte("[]"))
-		case 3:
+		case 2:
 			w.Write([]byte(`{"status":"error", "errorType":"type:", "error":"some error msg"}`))
-		case 4:
+		case 3:
 			w.Write([]byte(`{"status":"unknown"}`))
-		case 5:
+		case 4:
 			w.Write([]byte(`{"status":"success","data":{"resultType":"matrix"}}`))
-		case 6:
+		case 5:
 			w.Write([]byte(`{"status":"success","data":{"resultType":"vector","result":[{"metric":{"__name__":"vm_rows","foo":"bar"},"value":[1583786142,"13763"]},{"metric":{"__name__":"vm_requests","foo":"baz"},"value":[1583786140,"2000"]}]}}`))
-		case 7:
+		case 6:
 			w.Write([]byte(`{"status":"success","data":{"resultType":"scalar","result":[1583786142, "1"]}}`))
 		}
 	})
@@ -95,17 +92,20 @@ func TestVMInstantQuery(t *testing.T) {
 	ts := time.Now()
 
 	expErr := func(err string) {
-		if _, _, err := pq.Query(ctx, query, ts); err == nil {
+		_, _, gotErr := pq.Query(ctx, query, ts)
+		if gotErr == nil {
 			t.Fatalf("expected %q got nil", err)
 		}
+		if !strings.Contains(gotErr.Error(), err) {
+			t.Fatalf("expected err %q; got %q", err, gotErr)
+		}
 	}
 
-	expErr("connection error")              // 0
-	expErr("invalid response status error") // 1
-	expErr("response body error")           // 2
-	expErr("error status")                  // 3
-	expErr("unknown status")                // 4
-	expErr("non-vector resultType error")   // 5
+	expErr("500")                              // 0
+	expErr("error parsing prometheus metrics") // 1
+	expErr("response error")                   // 2
+	expErr("unknown status")                   // 3
+	expErr("unexpected end of JSON input")     // 4
 
 	m, _, err := pq.Query(ctx, query, ts) // 6 - vector
 	if err != nil {
@@ -165,6 +165,75 @@ func TestVMInstantQuery(t *testing.T) {
 		},
 	}
 	metricsEqual(t, m, exp)
+
+}
+
+func TestVMInstantQueryWithRetry(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", func(_ http.ResponseWriter, _ *http.Request) {
+		t.Errorf("should not be called")
+	})
+	c := -1
+	mux.HandleFunc("/api/v1/query", func(w http.ResponseWriter, r *http.Request) {
+		c++
+		if r.URL.Query().Get("query") != query {
+			t.Errorf("expected %s in query param, got %s", query, r.URL.Query().Get("query"))
+		}
+		switch c {
+		case 0:
+			w.Write([]byte(`{"status":"success","data":{"resultType":"scalar","result":[1583786142, "1"]}}`))
+		case 1:
+			conn, _, _ := w.(http.Hijacker).Hijack()
+			_ = conn.Close()
+		case 2:
+			w.Write([]byte(`{"status":"success","data":{"resultType":"scalar","result":[1583786142, "2"]}}`))
+		case 3:
+			conn, _, _ := w.(http.Hijacker).Hijack()
+			_ = conn.Close()
+		case 4:
+			conn, _, _ := w.(http.Hijacker).Hijack()
+			_ = conn.Close()
+		}
+	})
+
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	s := NewVMStorage(srv.URL, nil, time.Minute, 0, false, srv.Client())
+	pq := s.BuildWithParams(QuerierParams{DataSourceType: string(datasourcePrometheus)})
+
+	expErr := func(err string) {
+		_, _, gotErr := pq.Query(ctx, query, time.Now())
+		if gotErr == nil {
+			t.Fatalf("expected %q got nil", err)
+		}
+		if !strings.Contains(gotErr.Error(), err) {
+			t.Fatalf("expected err %q; got %q", err, gotErr)
+		}
+	}
+
+	expValue := func(v float64) {
+		m, _, err := pq.Query(ctx, query, time.Now())
+		if err != nil {
+			t.Fatalf("unexpected %s", err)
+		}
+		if len(m) != 1 {
+			t.Fatalf("expected 1 metrics got %d in %+v", len(m), m)
+		}
+		expected := []Metric{
+			{
+				Timestamps: []int64{1583786142},
+				Values:     []float64{v},
+			},
+		}
+		if !reflect.DeepEqual(m, expected) {
+			t.Fatalf("unexpected metric %+v want %+v", m, expected)
+		}
+	}
+
+	expValue(1)   // 0
+	expValue(2)   // 1 - fail, 2 - retry
+	expErr("EOF") // 3, 4 - retries
 }
 
 func metricsEqual(t *testing.T, gotM, expectedM []Metric) {