fix in handling states and purging

CHANGELOG.next.asciidoc

@@ -60,6 +60,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 - Fix reporting of `filebeat.events.active` in log events such that the current value is always reported instead of the difference from the last value. {pull}33597[33597]
 - Fix splitting array of strings/arrays in httpjson input {issue}30345[30345] {pull}33609[33609]
 - Fix Google workspace pagination and document ID generation. {pull}33666[33666]
+- Fix handling of error in states in direct aws-s3 listing input {issue}33513[33513] {pull}33722[33722]
 
 *Heartbeat*
 - Fix bug affecting let's encrypt and other users of cross-signed certs, where cert expiration was incorrectly calculated. {issue}33215[33215]

x-pack/filebeat/input/awss3/input_integration_test.go

@@ -31,11 +31,8 @@ import (
 	"golang.org/x/sync/errgroup"
 	"gopkg.in/yaml.v2"
 
-	"github.com/elastic/beats/v7/filebeat/beater"
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	pubtest "github.com/elastic/beats/v7/libbeat/publisher/testing"
-	"github.com/elastic/beats/v7/libbeat/statestore"
-	"github.com/elastic/beats/v7/libbeat/statestore/storetest"
 	awscommon "github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -134,28 +131,6 @@ file_selectors:
 `, queueURL))
 }
 
-type testInputStore struct {
-	registry *statestore.Registry
-}
-
-func openTestStatestore() beater.StateStore {
-	return &testInputStore{
-		registry: statestore.NewRegistry(storetest.NewMemoryStoreBackend()),
-	}
-}
-
-func (s *testInputStore) Close() {
-	s.registry.Close()
-}
-
-func (s *testInputStore) Access() (*statestore.Store, error) {
-	return s.registry.Get("filebeat")
-}
-
-func (s *testInputStore) CleanupInterval() time.Duration {
-	return 24 * time.Hour
-}
-
 func createInput(t *testing.T, cfg *conf.C) *s3Input {
 	inputV2, err := Plugin(openTestStatestore()).Manager.Create(cfg)
 	if err != nil {

x-pack/filebeat/input/awss3/s3.go

@@ -282,8 +282,8 @@ func (p *s3Poller) Purge() {
 
 		for _, state := range p.states.GetStatesByListingID(listingID) {
 			// it is not stored, keep
-			if !state.Stored {
-				p.log.Debugw("state not stored, skip purge", "state", state)
+			if !state.Stored && !state.Error {
+				p.log.Debugw("state not stored or with error, skip purge", "state", state)
 				continue
 			}
 
@@ -294,7 +294,7 @@ func (p *s3Poller) Purge() {
 				var commitWriteState commitWriteState
 				err := p.store.Get(awsS3WriteCommitPrefix+state.Bucket+state.ListPrefix, &commitWriteState)
 				if err == nil {
-					// we have no entry in the map and we have no entry in the store
+					// we have no entry in the map, and we have no entry in the store
 					// set zero time
 					latestStoredTime = time.Time{}
 					p.log.Debugw("last stored time is zero time", "bucket", state.Bucket, "listPrefix", state.ListPrefix)

x-pack/filebeat/input/awss3/states.go

@@ -78,6 +78,12 @@ func (s *states) MustSkip(state state, store *statestore.Store) bool {
 		return true
 	}
 
+	// the previous state is stored or has error: let's skip
+	if !previousState.IsEmpty() && (previousState.Stored || previousState.Error) {
+		s.log.Debugw("previous state is stored or has error", "state", state)
+		return true
+	}
+
 	// we have no previous state or the previous state
 	// is not stored: refresh the state
 	if previousState.IsEmpty() || (!previousState.Stored && !previousState.Error) {

x-pack/filebeat/input/awss3/states_test.go

@@ -9,12 +9,38 @@ import (
 	"testing"
 	"time"
 
+	"github.com/elastic/beats/v7/filebeat/beater"
+	"github.com/elastic/beats/v7/libbeat/statestore"
+	"github.com/elastic/beats/v7/libbeat/statestore/storetest"
+
 	"github.com/stretchr/testify/assert"
 
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
 	"github.com/elastic/elastic-agent-libs/logp"
 )
 
+type testInputStore struct {
+	registry *statestore.Registry
+}
+
+func openTestStatestore() beater.StateStore {
+	return &testInputStore{
+		registry: statestore.NewRegistry(storetest.NewMemoryStoreBackend()),
+	}
+}
+
+func (s *testInputStore) Close() {
+	_ = s.registry.Close()
+}
+
+func (s *testInputStore) Access() (*statestore.Store, error) {
+	return s.registry.Get("filebeat")
+}
+
+func (s *testInputStore) CleanupInterval() time.Duration {
+	return 24 * time.Hour
+}
+
 var inputCtx = v2.Context{
 	Logger:      logp.NewLogger("test"),
 	Cancelation: context.Background(),
@@ -83,6 +109,139 @@ func TestStatesIsNew(t *testing.T) {
 	}
 }
 
+func TestMustSkip(t *testing.T) {
+	type stateTestCase struct {
+		states            func() *states
+		state             state
+		mustBeNew         bool
+		persistentStoreKV map[string]interface{}
+		expected          bool
+	}
+	lastModified := time.Date(2022, time.June, 30, 14, 13, 00, 0, time.UTC)
+	tests := map[string]stateTestCase{
+		"with empty states": {
+			states: func() *states {
+				return newStates(inputCtx)
+			},
+			state:    newState("bucket", "key", "etag", "listPrefix", lastModified),
+			expected: false,
+		},
+		"not existing state": {
+			states: func() *states {
+				states := newStates(inputCtx)
+				states.Update(newState("bucket", "key", "etag", "listPrefix", lastModified), "")
+				return states
+			},
+			state:    newState("bucket1", "key1", "etag1", "listPrefix1", lastModified),
+			expected: false,
+		},
+		"existing state": {
+			states: func() *states {
+				states := newStates(inputCtx)
+				states.Update(newState("bucket", "key", "etag", "listPrefix", lastModified), "")
+				return states
+			},
+			state:    newState("bucket", "key", "etag", "listPrefix", lastModified),
+			expected: true,
+		},
+		"with different etag": {
+			states: func() *states {
+				states := newStates(inputCtx)
+				states.Update(newState("bucket", "key", "etag1", "listPrefix", lastModified), "")
+				return states
+			},
+			state:    newState("bucket", "key", "etag2", "listPrefix", lastModified),
+			expected: false,
+		},
+		"with different lastmodified": {
+			states: func() *states {
+				states := newStates(inputCtx)
+				states.Update(newState("bucket", "key", "etag", "listPrefix", lastModified), "")
+				return states
+			},
+			state:    newState("bucket", "key", "etag", "listPrefix", lastModified.Add(1*time.Second)),
+			expected: false,
+		},
+		"with stored state": {
+			states: func() *states {
+				states := newStates(inputCtx)
+				aState := newState("bucket", "key", "etag", "listPrefix", lastModified)
+				aState.Stored = true
+				states.Update(aState, "")
+				return states
+			},
+			state:     newState("bucket", "key", "etag", "listPrefix", lastModified),
+			mustBeNew: true,
+			expected:  true,
+		},
+		"with error state": {
+			states: func() *states {
+				states := newStates(inputCtx)
+				aState := newState("bucket", "key", "etag", "listPrefix", lastModified)
+				aState.Error = true
+				states.Update(aState, "")
+				return states
+			},
+			state:     newState("bucket", "key", "etag", "listPrefix", lastModified),
+			mustBeNew: true,
+			expected:  true,
+		},
+		"before commit write": {
+			states: func() *states {
+				return newStates(inputCtx)
+			},
+			persistentStoreKV: map[string]interface{}{
+				awsS3WriteCommitPrefix + "bucket" + "listPrefix": &commitWriteState{lastModified},
+			},
+			state:    newState("bucket", "key", "etag", "listPrefix", lastModified.Add(-1*time.Second)),
+			expected: true,
+		},
+		"same commit write": {
+			states: func() *states {
+				return newStates(inputCtx)
+			},
+			persistentStoreKV: map[string]interface{}{
+				awsS3WriteCommitPrefix + "bucket" + "listPrefix": &commitWriteState{lastModified},
+			},
+			state:    newState("bucket", "key", "etag", "listPrefix", lastModified),
+			expected: true,
+		},
+		"after commit write": {
+			states: func() *states {
+				return newStates(inputCtx)
+			},
+			persistentStoreKV: map[string]interface{}{
+				awsS3WriteCommitPrefix + "bucket" + "listPrefix": &commitWriteState{lastModified},
+			},
+			state:    newState("bucket", "key", "etag", "listPrefix", lastModified.Add(time.Second)),
+			expected: false,
+		},
+	}
+
+	for name, test := range tests {
+		test := test
+		t.Run(name, func(t *testing.T) {
+			states := test.states()
+			store := openTestStatestore()
+			persistentStore, err := store.Access()
+			if err != nil {
+				t.Fatalf("unexpected err: %v", err)
+			}
+			for key, value := range test.persistentStoreKV {
+				_ = persistentStore.Set(key, value)
+			}
+
+			if test.mustBeNew {
+				test.state.LastModified = test.state.LastModified.Add(1 * time.Second)
+			}
+
+			isNew := states.MustSkip(test.state, persistentStore)
+			assert.Equal(t, test.expected, isNew)
+			_ = persistentStore.Close()
+		})
+	}
+}
+
 func TestStatesDelete(t *testing.T) {
 	type stateTestCase struct {
 		states   func() *states