Add support for replacing an existing kv prefix with a new set

kv.go

@@ -381,6 +381,34 @@ func (kv *KvSource) ListValuedPairWithPrefix(key string) (map[string][]byte, err
 	return pairs, nil
 }
 
+// ReplaceConfig will replace the existing config in the KV store with
+// the new config. Unused (stale) keys will be removed from the kv
+// store, while existing keys remain and get updated.
+func (kv *KvSource) ReplaceConfig(config interface{}) error {
+	existingConfigPairs, err := kv.ListValuedPairWithPrefix(kv.Prefix)
+
+	if err != nil {
+		return err
+	}
+
+	newConfigPairs := map[string]string{}
+	if err = collateKvRecursive(reflect.ValueOf(config), newConfigPairs, kv.Prefix); err != nil {
+		return err
+	}
+
+	// Loop over the keys in the store and remove the keys that are not present in the new config
+	for existingKey := range existingConfigPairs {
+		if _, ok := newConfigPairs[existingKey]; !ok {
+			err = kv.Delete(existingKey)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return kv.StoreConfig(config)
+}
+
 func convertPairs(pairs map[string][]byte) []*store.KVPair {
 	slicePairs := make([]*store.KVPair, len(pairs))
 	i := 0

kv_mock_test.go

@@ -40,7 +40,16 @@ func (s *Mock) Get(key string, options *store.ReadOptions) (*store.KVPair, error
 }
 
 func (s *Mock) Delete(key string) error {
-	return errors.New("delete not supported")
+	pairs := make([]*store.KVPair, 0)
+	for _, pair := range s.KVPairs {
+		if pair.Key != key {
+			pairs = append(pairs, pair)
+		}
+	}
+
+	s.KVPairs = pairs
+
+	return nil
 }
 
 // Exists mock

kv_test.go

@@ -1238,3 +1238,74 @@ func TestDecodeHookCustomMarshaller(t *testing.T) {
 
 	assert.Exactly(t, data, output)
 }
+
+func TestReplaceConfig(t *testing.T) {
+	kv := &KvSource{
+		&Mock{
+			KVPairs: []*store.KVPair{
+				{Key: "prefix/foo", Value: []byte("foo")},
+				{Key: "prefix/bar/baz", Value: []byte("foo")},
+				{Key: "prefix/bar/boo", Value: []byte("foo")},
+			},
+			WatchTreeMethod: nil,
+		},
+		"prefix",
+	}
+
+	config := &struct {
+		Foo string
+	}{
+		Foo: "bar",
+	}
+
+	err := kv.ReplaceConfig(config)
+	require.NoError(t, err)
+
+	results, err := kv.ListValuedPairWithPrefix("prefix")
+	require.NoError(t, err)
+
+	require.EqualValues(t, map[string][]byte{
+		"prefix/foo": []byte("bar"),
+	}, results)
+}
+
+func TestReplaceNestedConfig(t *testing.T) {
+	kv := &KvSource{
+		&Mock{
+			KVPairs: []*store.KVPair{
+				{Key: "prefix/foo", Value: []byte("foo")},
+				{Key: "prefix/bar/baz", Value: []byte("foo")},
+				{Key: "prefix/bar/boo", Value: []byte("foo")},
+				{Key: "prefix/bar/far/baz", Value: []byte("foo")},
+			},
+			WatchTreeMethod: nil,
+		},
+		"prefix",
+	}
+
+	config := &struct {
+		Foo string
+		Bar map[string]map[string]string
+		New string
+	}{
+		Foo: "bar",
+		Bar: map[string]map[string]string{
+			"far": {
+				"baz": "faz",
+			},
+		},
+		New: "foo",
+	}
+
+	err := kv.ReplaceConfig(config)
+	require.NoError(t, err)
+
+	results, err := kv.ListValuedPairWithPrefix("prefix")
+	require.NoError(t, err)
+
+	require.EqualValues(t, map[string][]byte{
+		"prefix/foo":         []byte("bar"),
+		"prefix/bar/far/baz": []byte("faz"),
+		"prefix/new":         []byte("foo"),
+	}, results)
+}