Introduce statestore internal testing support (elastic#19177)

This change introduces an internal package to the statestore that
provides testing helpers.
The addition of the statestore package is split up into multiple
changeset to ease review. The final version of the package can be found
[here](https://github.com/urso/beats/tree/fb-input-v2-combined/libbeat/statestore).

Once finalized, the libbeat/statestore package contains:
- The statestore frontend and interface for use within Beats
- Interfaces for the store backend
- A common set of tests store backends need to support
- a storetest package for testing new features that require a store. The
  testing helpers use map[string]interface{} that can be initialized or
  queried after the test run for validation purposes.
- The default memlog backend + tests

This change include common testing support, that will be used by the
memlog backend, and to verify the storetest package.

libbeat/statestore/internal/storecompliance/reg.go

@@ -0,0 +1,113 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package storecompliance
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/v7/libbeat/statestore/backend"
+)
+
+// Registry helper for writing tests.
+// The registry uses a testing.T and provides some MustX methods that fail if
+// an error occured.
+type Registry struct {
+	T testing.TB
+	backend.Registry
+}
+
+// Store helper for writing tests.
+// The store needs a reference to the Registry with the current test context.
+// The Store provides additional helpers for reopening the store, MustX methods
+// that will fail the test if an error has occured.
+type Store struct {
+	backend.Store
+
+	Registry *Registry
+	name     string
+}
+
+// Access uses the backend Registry to create a new Store.
+func (r *Registry) Access(name string) (*Store, error) {
+	s, err := r.Registry.Access(name)
+	if err != nil {
+		return nil, err
+	}
+	return &Store{Store: s, Registry: r, name: name}, nil
+}
+
+// MustAccess opens a Store. It fails the test if an error has occured.
+func (r *Registry) MustAccess(name string) *Store {
+	store, err := r.Access(name)
+	must(r.T, err, "open store")
+	return store
+}
+
+// Close closes the testing store.
+func (s *Store) Close() {
+	err := s.Store.Close()
+	must(s.Registry.T, err, "closing store %q failed", s.name)
+}
+
+// ReopenIf reopens the store if b is true.
+func (s *Store) ReopenIf(b bool) {
+	if b {
+		s.Reopen()
+	}
+}
+
+// Reopen reopens the store by closing the backend store and using the registry
+// backend to access the same store again.
+func (s *Store) Reopen() {
+	t := s.Registry.T
+
+	s.Close()
+	if t.Failed() {
+		t.Fatal("Test already failed")
+	}
+
+	store, err := s.Registry.Registry.Access(s.name)
+	must(s.Registry.T, err, "reopen failed")
+
+	s.Store = store
+}
+
+// MustHave fails the test if an error occured in a call to Has.
+func (s *Store) MustHave(key string) bool {
+	b, err := s.Has(key)
+	must(s.Registry.T, err, "unexpected error on store/has call")
+	return b
+}
+
+// MustGet fails the test if an error occured in a call to Get.
+func (s *Store) MustGet(key string, into interface{}) {
+	err := s.Get(key, into)
+	must(s.Registry.T, err, "unexpected error on store/get call")
+}
+
+// MustSet fails the test if an error occured in a call to Set.
+func (s *Store) MustSet(key string, from interface{}) {
+	err := s.Set(key, from)
+	must(s.Registry.T, err, "unexpected error on store/set call")
+}
+
+// MustRemove fails the test if an error occured in a call to Remove.
+func (s *Store) MustRemove(key string) {
+	err := s.Store.Remove(key)
+	must(s.Registry.T, err, "unexpected error remove key")
+}

libbeat/statestore/internal/storecompliance/storecompliance.go

@@ -0,0 +1,188 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package storecompliance provides a common test suite that a store
+// implementation must succeed in order to be compliant to the beats
+// statestore. The Internal tests are used by statestore/storetest and
+// statestore/backend/memlog.
+//
+// The package adds the `-keep` and `-dir <path>` CLI flags:
+//   - `-dir <path>`: configure path where to create test folders in (defaults
+//     to OS specific temporary directory)
+//   - `-keep`: The test directories will not be deleted after a test has
+//     finished. The test directory is added to the test logs.
+//
+package storecompliance
+
+import (
+	"errors"
+	"flag"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/v7/libbeat/statestore/backend"
+)
+
+// BackendFactory is used by TestBackendCompliance to create
+// store instances for testing. Each store will be configured
+// with an unique temporary directory.
+type BackendFactory func(testPath string) (backend.Registry, error)
+
+var defaultTempDir string
+var keepTmpDir bool
+
+func init() {
+	flag.StringVar(&defaultTempDir, "dir", "", "Temporary directory for use by the test")
+	flag.BoolVar(&keepTmpDir, "keep", false, "Keep temporary test directories")
+}
+
+// TestBackendCompliance runs a set of tests the verifies that the store
+// implementation can be used by beats.
+// Most tests are executed twice if they modify data. Once with keeping the
+// store open between operations, and once with reopening the store between
+// updates.
+// For a store backend that supports different 'modes' that can impact storage,
+// the compliance tests should be run with the different modes enabled.
+//
+// Note: The tests only check for interoperability. Implementations should add
+// additional tests as well.
+func TestBackendCompliance(t *testing.T, factory BackendFactory) {
+	t.Run("init and close registry", WithPath(factory, func(t *testing.T, reg *Registry) {
+	}))
+
+	t.Run("open stores", WithPath(factory, func(t *testing.T, reg *Registry) {
+		store := reg.MustAccess("test1")
+		defer store.Close()
+
+		store2 := reg.MustAccess("test2")
+		defer store2.Close()
+	}))
+
+	t.Run("set-get", withBackend(factory, testSetGet))
+	t.Run("remove", withBackend(factory, testRemove))
+	t.Run("iteration", withBackend(factory, testIteration))
+}
+
+func testSetGet(t *testing.T, factory BackendFactory) {
+	t.Run("unknown key", WithStore(factory, func(t *testing.T, store *Store) {
+		has := store.MustHave("key")
+		assert.False(t, has)
+	}))
+
+	runWithBools(t, "reopen", func(t *testing.T, reopen bool) {
+		t.Run("has key after set", WithStore(factory, func(t *testing.T, store *Store) {
+			type entry struct{ A int }
+			store.MustSet("key", entry{A: 1})
+
+			store.ReopenIf(reopen)
+
+			has := store.MustHave("key")
+			assert.True(t, has)
+		}))
+
+		t.Run("set and get one entry only", WithStore(factory, func(t *testing.T, store *Store) {
+			type entry struct{ A int }
+			key := "key"
+			value := entry{A: 1}
+
+			store.MustSet(key, value)
+			store.ReopenIf(reopen)
+
+			var actual entry
+			store.MustGet(key, &actual)
+			assert.Equal(t, value, actual)
+		}))
+	})
+}
+
+func testRemove(t *testing.T, factory BackendFactory) {
+	t.Run("no error when removing unknown key", WithStore(factory, func(t *testing.T, store *Store) {
+		store.MustRemove("key")
+	}))
+
+	runWithBools(t, "reopen", func(t *testing.T, reopen bool) {
+		t.Run("remove key", WithStore(factory, func(t *testing.T, store *Store) {
+			type entry struct{ A int }
+			key := "key"
+			store.MustSet(key, entry{A: 1})
+			store.ReopenIf(reopen)
+			store.MustRemove(key)
+			store.ReopenIf(reopen)
+			has := store.MustHave(key)
+			assert.False(t, has)
+		}))
+	})
+}
+
+func testIteration(t *testing.T, factory BackendFactory) {
+	data := map[string]interface{}{
+		"a": map[string]interface{}{"field": "hello"},
+		"b": map[string]interface{}{"field": "world"},
+	}
+
+	addTestData := func(store *Store, reopen bool, data map[string]interface{}) {
+		for k, v := range data {
+			store.MustSet(k, v)
+		}
+		store.ReopenIf(reopen)
+	}
+
+	runWithBools(t, "reopen", func(t *testing.T, reopen bool) {
+		t.Run("all keys", WithStore(factory, func(t *testing.T, store *Store) {
+			addTestData(store, reopen, data)
+
+			got := map[string]interface{}{}
+			err := store.Each(func(key string, dec backend.ValueDecoder) (bool, error) {
+				var tmp interface{}
+				if err := dec.Decode(&tmp); err != nil {
+					return false, err
+				}
+
+				got[key] = tmp
+				return true, nil
+			})
+
+			assert.NoError(t, err)
+			assert.Equal(t, data, got)
+		}))
+
+		t.Run("stop on error", WithStore(factory, func(t *testing.T, store *Store) {
+			addTestData(store, reopen, data)
+
+			count := 0
+			err := store.Each(func(_ string, _ backend.ValueDecoder) (bool, error) {
+				count++
+				return true, errors.New("oops")
+			})
+			assert.Equal(t, 1, count)
+			assert.Error(t, err)
+		}))
+
+		t.Run("stop on bool", WithStore(factory, func(t *testing.T, store *Store) {
+			addTestData(store, reopen, data)
+
+			count := 0
+			err := store.Each(func(_ string, _ backend.ValueDecoder) (bool, error) {
+				count++
+				return false, nil
+			})
+			assert.Equal(t, 1, count)
+			assert.NoError(t, err)
+		}))
+	})
+}