feat(kv): support for migrations

CHANGELOG.md

@@ -18,6 +18,7 @@
 1. [17138](https://github.com/influxdata/influxdb/pull/17138): Extend pkger export all capabilities to support filtering by lable name and resource type
 1. [17049](https://github.com/influxdata/influxdb/pull/17049): Added new login and sign-up screen that for cloud users that allows direct login from their region
 1. [17170](https://github.com/influxdata/influxdb/pull/17170): Added new cli multiple profiles management tool
+1. [17145](https://github.com/influxdata/influxdb/pull/17145): Update kv.Store to define schema changes via new kv.Migrator types
 
 ### Bug Fixes
 

bolt/kv.go

@@ -18,6 +18,9 @@ import (
 // check that *KVStore implement kv.Store interface.
 var _ kv.Store = (*KVStore)(nil)
 
+// ensure *KVStore implements kv.AutoMigrationStore.
+var _ kv.AutoMigrationStore = (*KVStore)(nil)
+
 // KVStore is a kv.Store backed by boltdb.
 type KVStore struct {
 	path string
@@ -34,6 +37,11 @@ func NewKVStore(log *zap.Logger, path string) *KVStore {
 	}
 }
 
+// AutoMigrate returns itself as it is safe to automatically apply migrations on initialization.
+func (s *KVStore) AutoMigrate() kv.Store {
+	return s
+}
+
 // Open creates boltDB file it doesn't exists and opens it otherwise.
 func (s *KVStore) Open(ctx context.Context) error {
 	span, _ := tracing.StartSpanFromContext(ctx)

inmem/kv.go

@@ -14,6 +14,9 @@ import (
 // ensure *KVStore implement kv.Store interface
 var _ kv.Store = (*KVStore)(nil)
 
+// ensure *KVStore implements kv.AutoMigrationStore
+var _ kv.AutoMigrationStore = (*KVStore)(nil)
+
 // cursorBatchSize is the size of a batch sent by a forward cursors
 // tree iterator
 const cursorBatchSize = 1000
@@ -61,6 +64,11 @@ func (s *KVStore) Backup(ctx context.Context, w io.Writer) error {
 	panic("not implemented")
 }
 
+// AutoMigrate returns itlsef as *KVStore is safe to migrate automically on initialize.
+func (s *KVStore) AutoMigrate() kv.Store {
+	return s
+}
+
 // Flush removes all data from the buckets.  Used for testing.
 func (s *KVStore) Flush(ctx context.Context) {
 	s.mu.Lock()

kv/index.go

@@ -141,8 +141,7 @@ func NewIndex(mapping IndexMapping, opts ...IndexOption) *Index {
 	return index
 }
 
-// Initialize creates the bucket if it does not already exist.
-func (i *Index) Initialize(ctx context.Context, store Store) error {
+func (i *Index) initialize(ctx context.Context, store Store) error {
 	return store.Update(ctx, func(tx Tx) error {
 		// create bucket if not exist
 		_, err := tx.Bucket(i.IndexBucket())
@@ -180,6 +179,53 @@ func indexKeyParts(indexKey []byte) (fk, pk []byte, err error) {
 	return
 }
 
+// IndexMigration is a migration for adding and removing an index.
+// These are constructed via the Index.Migration function.
+type IndexMigration struct {
+	*Index
+	opts []PopulateOption
+}
+
+// Name returns a readable name for the index migration.
+func (i *IndexMigration) Name() string {
+	return fmt.Sprintf("add index %q", string(i.IndexBucket()))
+}
+
+// Up initializes the index bucket and populates the index.
+func (i *IndexMigration) Up(ctx context.Context, store Store) (err error) {
+	wrapErr := func(err error) error {
+		if err == nil {
+			return nil
+		}
+
+		return fmt.Errorf("migration (up) %s: %w", i.Name(), err)
+	}
+
+	if err = i.initialize(ctx, store); err != nil {
+		return wrapErr(err)
+	}
+
+	_, err = i.Populate(ctx, store, i.opts...)
+	return wrapErr(err)
+}
+
+// Down deletes all entries from the index.
+func (i *IndexMigration) Down(ctx context.Context, store Store) error {
+	if err := i.DeleteAll(ctx, store); err != nil {
+		return fmt.Errorf("migration (down) %s: %w", i.Name(), err)
+	}
+
+	return nil
+}
+
+// Migration creates an IndexMigration for the underlying Index.
+func (i *Index) Migration(opts ...PopulateOption) *IndexMigration {
+	return &IndexMigration{
+		Index: i,
+		opts:  opts,
+	}
+}
+
 // Insert creates a single index entry for the provided primary key on the foreign key.
 func (i *Index) Insert(tx Tx, foreignKey, primaryKey []byte) error {
 	bkt, err := i.indexBucket(tx)