diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc index 46a86a51ecdc..eef3e2304167 100644 --- a/CHANGELOG.next.asciidoc +++ b/CHANGELOG.next.asciidoc @@ -100,6 +100,9 @@ fields added to events containing the Beats version. {pull}37553[37553] *Heartbeat* +- Fix panics when parsing dereferencing invalid parsed url. {pull}34702[34702] +- Fix setuid root when running under cgroups v2. {pull}37794[37794] +- Adjust State loader to only retry when response code status is 5xx {pull}37981[37981] *Metricbeat* diff --git a/heartbeat/monitors/wrappers/monitorstate/esloader.go b/heartbeat/monitors/wrappers/monitorstate/esloader.go index d63be5aada5b..5705ec4b1468 100644 --- a/heartbeat/monitors/wrappers/monitorstate/esloader.go +++ b/heartbeat/monitors/wrappers/monitorstate/esloader.go @@ -32,13 +32,21 @@ import ( var DefaultDataStreams = "synthetics-*,heartbeat-*" +type LoaderError struct { + err error + Retry bool +} + +func (e LoaderError) Error() string { + return e.err.Error() +} + func MakeESLoader(esc *eslegclient.Connection, indexPattern string, beatLocation *config.LocationWithID) StateLoader { if indexPattern == "" { // Should never happen, but if we ever make a coding error... logp.L().Warn("ES state loader initialized with no index pattern, will not load states from ES") return NilStateLoader } - return func(sf stdfields.StdMonitorFields) (*State, error) { var runFromID string if sf.RunFrom != nil { @@ -74,10 +82,11 @@ func MakeESLoader(esc *eslegclient.Connection, indexPattern string, beatLocation }, }, } - status, body, err := esc.Request("POST", strings.Join([]string{"/", indexPattern, "/", "_search", "?size=1"}, ""), "", nil, reqBody) if err != nil || status > 299 { - return nil, fmt.Errorf("error executing state search for %s in loc=%s: %w", sf.ID, runFromID, err) + sErr := fmt.Errorf("error executing state search for %s in loc=%s: %w", sf.ID, runFromID, err) + retry := shouldRetry(status) + return nil, LoaderError{err: sErr, Retry: retry} } type stateHits struct { @@ -94,7 +103,8 @@ func MakeESLoader(esc *eslegclient.Connection, indexPattern string, beatLocation sh := stateHits{} err = json.Unmarshal(body, &sh) if err != nil { - return nil, fmt.Errorf("could not unmarshal state hits for %s: %w", sf.ID, err) + sErr := fmt.Errorf("could not unmarshal state hits for %s: %w", sf.ID, err) + return nil, LoaderError{err: sErr, Retry: false} } if len(sh.Hits.Hits) == 0 { @@ -107,3 +117,7 @@ func MakeESLoader(esc *eslegclient.Connection, indexPattern string, beatLocation return state, nil } } + +func shouldRetry(status int) bool { + return status >= 500 +} diff --git a/heartbeat/monitors/wrappers/monitorstate/esloader_test.go b/heartbeat/monitors/wrappers/monitorstate/esloader_test.go index 42b1c6c31c31..db70c1cc72e7 100644 --- a/heartbeat/monitors/wrappers/monitorstate/esloader_test.go +++ b/heartbeat/monitors/wrappers/monitorstate/esloader_test.go @@ -21,6 +21,9 @@ package monitorstate import ( "fmt" + "io" + "net/http" + "strings" "testing" "time" @@ -33,6 +36,7 @@ import ( "github.com/elastic/beats/v7/heartbeat/config" "github.com/elastic/beats/v7/heartbeat/esutil" "github.com/elastic/beats/v7/heartbeat/monitors/stdfields" + "github.com/elastic/beats/v7/libbeat/esleg/eslegclient" "github.com/elastic/beats/v7/libbeat/processors/util" ) @@ -51,7 +55,7 @@ func TestStatesESLoader(t *testing.T) { monID := etc.createTestMonitorStateInES(t, testStatus) // Since we've continued this state it should register the initial state - ms := etc.tracker.GetCurrentState(monID) + ms := etc.tracker.GetCurrentState(monID, RetryConfig{}) require.True(t, ms.StartedAt.After(testStart.Add(-time.Nanosecond)), "timestamp for new state is off") requireMSStatusCount(t, ms, testStatus, 1) @@ -89,8 +93,61 @@ func TestStatesESLoader(t *testing.T) { } } +func TestMakeESLoaderError(t *testing.T) { + tests := []struct { + name string + statusCode int + expected bool + }{ + { + name: "should return a retryable error", + statusCode: http.StatusInternalServerError, + expected: true, + }, + { + name: "should not return a retryable error", + statusCode: http.StatusNotFound, + expected: false, + }, + { + name: "should not return a retryable error when handling malformed data", + statusCode: http.StatusOK, + expected: false, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + etc := newESTestContext(t) + etc.ec.HTTP = fakeHTTPClient{respStatus: test.statusCode} + loader := MakeESLoader(etc.ec, "fakeIndexPattern", etc.location) + + _, err := loader(stdfields.StdMonitorFields{}) + + var loaderError LoaderError + require.ErrorAs(t, err, &loaderError) + require.Equal(t, loaderError.Retry, test.expected) + }) + } +} + +type fakeHTTPClient struct { + respStatus int +} + +func (fc fakeHTTPClient) Do(req *http.Request) (resp *http.Response, err error) { + return &http.Response{ + StatusCode: fc.respStatus, + Body: io.NopCloser(strings.NewReader("test response")), + }, nil +} + +func (fc fakeHTTPClient) CloseIdleConnections() { + // noop +} + type esTestContext struct { namespace string + ec *eslegclient.Connection esc *elasticsearch.Client loader StateLoader tracker *Tracker @@ -106,10 +163,12 @@ func newESTestContext(t *testing.T) *esTestContext { } namespace, _ := uuid.NewV4() esc := IntegApiClient(t) + ec := IntegES(t) etc := &esTestContext{ namespace: namespace.String(), esc: esc, - loader: IntegESLoader(t, fmt.Sprintf("synthetics-*-%s", namespace.String()), location), + ec: ec, + loader: IntegESLoader(t, ec, fmt.Sprintf("synthetics-*-%s", namespace.String()), location), location: location, } diff --git a/heartbeat/monitors/wrappers/monitorstate/testutil.go b/heartbeat/monitors/wrappers/monitorstate/testutil.go index 540983097587..28a6c2606557 100644 --- a/heartbeat/monitors/wrappers/monitorstate/testutil.go +++ b/heartbeat/monitors/wrappers/monitorstate/testutil.go @@ -33,8 +33,8 @@ import ( // Helpers for tests here and elsewhere -func IntegESLoader(t *testing.T, indexPattern string, location *config.LocationWithID) StateLoader { - return MakeESLoader(IntegES(t), indexPattern, location) +func IntegESLoader(t *testing.T, esc *eslegclient.Connection, indexPattern string, location *config.LocationWithID) StateLoader { + return MakeESLoader(esc, indexPattern, location) } func IntegES(t *testing.T) (esc *eslegclient.Connection) { diff --git a/heartbeat/monitors/wrappers/monitorstate/tracker.go b/heartbeat/monitors/wrappers/monitorstate/tracker.go index e350294e46e8..40a4e8f2ded1 100644 --- a/heartbeat/monitors/wrappers/monitorstate/tracker.go +++ b/heartbeat/monitors/wrappers/monitorstate/tracker.go @@ -62,7 +62,7 @@ func (t *Tracker) RecordStatus(sf stdfields.StdMonitorFields, newStatus StateSta t.mtx.Lock() defer t.mtx.Unlock() - state := t.GetCurrentState(sf) + state := t.GetCurrentState(sf, RetryConfig{}) if state == nil { state = newMonitorState(sf, newStatus, 0, t.flappingEnabled) logp.L().Infof("initializing new state for monitor %s: %s", sf.ID, state.String()) @@ -75,22 +75,33 @@ func (t *Tracker) RecordStatus(sf stdfields.StdMonitorFields, newStatus StateSta } func (t *Tracker) GetCurrentStatus(sf stdfields.StdMonitorFields) StateStatus { - s := t.GetCurrentState(sf) + s := t.GetCurrentState(sf, RetryConfig{}) if s == nil { return StatusEmpty } return s.Status } -func (t *Tracker) GetCurrentState(sf stdfields.StdMonitorFields) (state *State) { +type RetryConfig struct { + attempts int + waitFn func() time.Duration +} + +func (t *Tracker) GetCurrentState(sf stdfields.StdMonitorFields, rc RetryConfig) (state *State) { if state, ok := t.states[sf.ID]; ok { return state } - tries := 3 + // Default number of attempts + attempts := 3 + if rc.attempts != 0 { + attempts = rc.attempts + } + var loadedState *State var err error - for i := 0; i < tries; i++ { + var i int + for i = 0; i < attempts; i++ { loadedState, err = t.stateLoader(sf) if err == nil { if loadedState != nil { @@ -98,13 +109,22 @@ func (t *Tracker) GetCurrentState(sf stdfields.StdMonitorFields) (state *State) } break } + var loaderError LoaderError + if errors.As(err, &loaderError) && !loaderError.Retry { + logp.L().Warnf("could not load last externally recorded state: %v", loaderError) + break + } + // Default sleep time sleepFor := (time.Duration(i*i) * time.Second) + (time.Duration(rand.Intn(500)) * time.Millisecond) - logp.L().Warnf("could not load last externally recorded state, will retry again in %d milliseconds: %w", sleepFor.Milliseconds(), err) + if rc.waitFn != nil { + sleepFor = rc.waitFn() + } + logp.L().Warnf("could not load last externally recorded state, will retry again in %d milliseconds: %v", sleepFor.Milliseconds(), err) time.Sleep(sleepFor) } if err != nil { - logp.L().Warnf("could not load prior state from elasticsearch after %d attempts, will create new state for monitor: %s", tries, sf.ID) + logp.L().Warnf("could not load prior state from elasticsearch after %d attempts, will create new state for monitor: %s", i+1, sf.ID) } if loadedState != nil { diff --git a/heartbeat/monitors/wrappers/monitorstate/tracker_test.go b/heartbeat/monitors/wrappers/monitorstate/tracker_test.go index ec1217b86150..fd34371ce810 100644 --- a/heartbeat/monitors/wrappers/monitorstate/tracker_test.go +++ b/heartbeat/monitors/wrappers/monitorstate/tracker_test.go @@ -18,6 +18,7 @@ package monitorstate import ( + "errors" "testing" "time" @@ -131,3 +132,51 @@ func TestDeferredStateLoader(t *testing.T) { resState, _ = dsl(stdfields.StdMonitorFields{}) require.Equal(t, stateA, resState) } + +func TestStateLoaderRetry(t *testing.T) { + // While testing the sleep time between retries should be negligible + waitFn := func() time.Duration { + return time.Microsecond + } + + tests := []struct { + name string + retryable bool + rc RetryConfig + expectedCalls int + }{ + { + "should retry 3 times when fails with retryable error", + true, + RetryConfig{waitFn: waitFn}, + 3, + }, + { + "should not retry when fails with non-retryable error", + false, + RetryConfig{waitFn: waitFn}, + 1, + }, + { + "should honour the configured number of attempts when fails with retryable error", + true, + RetryConfig{attempts: 5, waitFn: waitFn}, + 5, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + calls := 0 + errorStateLoader := func(_ stdfields.StdMonitorFields) (*State, error) { + calls += 1 + return nil, LoaderError{err: errors.New("test error"), Retry: tt.retryable} + } + + mst := NewTracker(errorStateLoader, true) + mst.GetCurrentState(stdfields.StdMonitorFields{}, tt.rc) + + require.Equal(t, calls, tt.expectedCalls) + }) + } +}