all

Documentation

Overview

package all

This package is the canonical location for all migrations being made against the single shared kv.Store implementation used by InfluxDB (while it remains a single store).

The array all.Migrations contains the list of migration specifications which drives the serial set of migration operations required to correctly configure the backing metadata store for InfluxDB.

This package is arranged like so:

doc.go - this piece of documentation.
all.go - definition of Migration array referencing each of the name migrations in number migration files (below).
migration.go - an implementation of migration.Spec for convenience.
000X_migration_name.go (example) - N files contains the specific implementations of each migration enumerated in `all.go`.
...

Managing this list of files and all.go can be fiddly. There is a buildable cli utility called `kvmigrate` in the `internal/cmd/kvmigrate` package. This has a command `create` which automatically creates a new migration in the expected location and appends it appropriately into the all.go Migration array.

Index

• Variables
• func Up(ctx context.Context, logger *zap.Logger, store kv.SchemaStore) error
• type Migration
  • func UpOnlyMigration(name string, up MigrationFunc) *Migration
  • func (m *Migration) Down(ctx context.Context, store kv.SchemaStore) error
  • func (m *Migration) MigrationName() string
  • func (m *Migration) Up(ctx context.Context, store kv.SchemaStore) error
• type MigrationFunc

Variables

var Migration0001_InitialMigration = kv.InitialMigration{}

Migration0001_InitialMigration contains all the buckets created before the time of migrations in kv

var Migration0002_AddURMByUserIndex = kv.NewIndexMigration(kv.URMByUserIndexMapping, kv.WithIndexMigrationCleanup)

Migration0002_AddURMByUserIndex creates the URM by user index and populates missing entries based on the source.

var Migration0003_TaskOwnerIDUpMigration = UpOnlyMigration(
	"migrate task owner id",
	func(ctx context.Context, store kv.SchemaStore) error {
		var ownerlessTasks []*influxdb.Task

		err := store.View(ctx, func(tx kv.Tx) error {
			taskBucket, err := tx.Bucket(taskBucket)
			if err != nil {
				return influxdb.ErrUnexpectedTaskBucketErr(err)
			}

			c, err := taskBucket.ForwardCursor([]byte{})
			if err != nil {
				return influxdb.ErrUnexpectedTaskBucketErr(err)
			}

			for k, v := c.Next(); k != nil; k, v = c.Next() {
				kvTask := &kvTask{}
				if err := json.Unmarshal(v, kvTask); err != nil {
					return influxdb.ErrInternalTaskServiceError(err)
				}

				t := kvToInfluxTask(kvTask)

				if !t.OwnerID.Valid() {
					ownerlessTasks = append(ownerlessTasks, t)
				}
			}
			if err := c.Err(); err != nil {
				return err
			}

			return c.Close()
		})
		if err != nil {
			return err
		}

		for _, t := range ownerlessTasks {

			err := store.Update(ctx, func(tx kv.Tx) error {
				taskKey, err := taskKey(t.ID)
				if err != nil {
					return err
				}
				b, err := tx.Bucket(taskBucket)
				if err != nil {
					return influxdb.ErrUnexpectedTaskBucketErr(err)
				}

				if !t.OwnerID.Valid() {
					v, err := b.Get(taskKey)
					if kv.IsNotFound(err) {
						return influxdb.ErrTaskNotFound
					}
					authType := struct {
						AuthorizationID influxdb.ID `json:"authorizationID"`
					}{}
					if err := json.Unmarshal(v, &authType); err != nil {
						return influxdb.ErrInternalTaskServiceError(err)
					}

					encodedID, err := authType.AuthorizationID.Encode()
					if err == nil {
						authBucket, err := tx.Bucket([]byte("authorizationsv1"))
						if err != nil {
							return err
						}

						a, err := authBucket.Get(encodedID)
						if err == nil {
							auth := &influxdb.Authorization{}
							if err := json.Unmarshal(a, auth); err != nil {
								return err
							}

							t.OwnerID = auth.GetUserID()
						}
					}

				}

				if !t.OwnerID.Valid() {
					b, err := tx.Bucket([]byte("userresourcemappingsv1"))
					if err != nil {
						return err
					}

					id, err := t.OrganizationID.Encode()
					if err != nil {
						return err
					}

					cur, err := b.ForwardCursor(id, kv.WithCursorPrefix(id))
					if err != nil {
						return err
					}

					for k, v := cur.Next(); k != nil; k, v = cur.Next() {
						m := &influxdb.UserResourceMapping{}
						if err := json.Unmarshal(v, m); err != nil {
							return err
						}
						if m.ResourceID == t.OrganizationID && m.ResourceType == influxdb.OrgsResourceType && m.UserType == influxdb.Owner {
							t.OwnerID = m.UserID
							break
						}
					}

					if err := cur.Close(); err != nil {
						return err
					}
				}

				if !t.OwnerID.Valid() {
					return &influxdb.Error{
						Code: influxdb.EInternal,
						Msg:  "could not populate owner ID for task",
					}
				}

				taskBytes, err := json.Marshal(t)
				if err != nil {
					return influxdb.ErrInternalTaskServiceError(err)
				}

				err = b.Put(taskKey, taskBytes)
				if err != nil {
					return influxdb.ErrUnexpectedTaskBucketErr(err)
				}
				return nil
			})
			if err != nil {
				return err
			}
		}
		return nil
	},
)

Migration0003_TaskOwnerIDUpMigration adds missing owner IDs to some legacy tasks

var Migration0004_AddDbrpBuckets = migration.CreateBuckets(
	"create DBRP buckets",
	dbrpBucket,
	dbrpIndexBucket,
	dbrpDefaultBucket,
)

Migration0004_AddDbrpBuckets creates the buckets necessary for the DBRP Service to operate.

var Migration0005_AddPkgerBuckets = migration.CreateBuckets(
	"create pkger stacks buckets",
	pkgerStacksBucket,
	pkgerStackIndexBucket,
)

Migration0005_AddPkgerBuckets creates the buckets necessary for the pkger service to operate.

var Migration0006_DeleteBucketSessionsv1 = migration.DeleteBuckets("delete sessionsv1 bucket", []byte("sessionsv1"))

Migration0006_DeleteBucketSessionsv1 removes the sessionsv1 bucket from the backing kv store.

var Migration0007_CreateMetaDataBucket = migration.CreateBuckets(
	"Create TSM metadata buckets",
	meta.BucketName)

var Migration0008_LegacyAuthBuckets = migration.CreateBuckets(
	"Create Legacy authorization buckets",
	[]byte("legacy/authorizationsv1"), []byte("legacy/authorizationindexv1"))

var Migration0009_LegacyAuthPasswordBuckets = migration.CreateBuckets(
	"Create legacy auth password bucket",
	[]byte("legacy/authorizationPasswordv1"))

var Migration0010_AddIndexTelegrafByOrg = kv.NewIndexMigration(telegraf.ByOrganizationIndexMapping, kv.WithIndexMigrationCleanup)

Migration0010_AddIndexTelegrafByOrg adds the index telegraf configs by organization ID

var Migration0011_PopulateDashboardsOwnerId = UpOnlyMigration("populate dashboards owner id", func(ctx context.Context, store kv.SchemaStore) error {
	var urmBucket = []byte("userresourcemappingsv1")
	type userResourceMapping struct {
		UserID       influxdb.ID           `json:"userID"`
		UserType     influxdb.UserType     `json:"userType"`
		MappingType  influxdb.MappingType  `json:"mappingType"`
		ResourceType influxdb.ResourceType `json:"resourceType"`
		ResourceID   influxdb.ID           `json:"resourceID"`
	}

	var mappings []*userResourceMapping
	if err := store.View(ctx, func(tx kv.Tx) error {
		bkt, err := tx.Bucket(urmBucket)
		if err != nil {
			return err
		}

		cursor, err := bkt.ForwardCursor(nil)
		if err != nil {
			return err
		}

		return kv.WalkCursor(ctx, cursor, func(_, v []byte) (bool, error) {
			var mapping userResourceMapping
			if err := json.Unmarshal(v, &mapping); err != nil {
				return false, err
			}

			if mapping.ResourceType == influxdb.DashboardsResourceType &&
				mapping.UserType == influxdb.Owner {
				mappings = append(mappings, &mapping)
			}

			return true, nil
		})
	}); err != nil {
		return err
	}

	var dashboardsBucket = []byte("dashboardsv2")
	// dashboard represents all visual and query data for a dashboard.
	type dashboard struct {
		ID             influxdb.ID            `json:"id,omitempty"`
		OrganizationID influxdb.ID            `json:"orgID,omitempty"`
		Name           string                 `json:"name"`
		Description    string                 `json:"description"`
		Cells          []*influxdb.Cell       `json:"cells"`
		Meta           influxdb.DashboardMeta `json:"meta"`
		OwnerID        *influxdb.ID           `json:"owner,omitempty"`
	}

	var (
		batchSize = 100
		flush     = func(batch []*userResourceMapping) (err error) {
			ids := make([][]byte, len(batch))
			for i, urm := range batch {
				ids[i], err = urm.ResourceID.Encode()
				if err != nil {
					return
				}
			}

			return store.Update(ctx, func(tx kv.Tx) error {
				bkt, err := tx.Bucket(dashboardsBucket)
				if err != nil {
					return err
				}

				values, err := bkt.GetBatch(ids...)
				if err != nil {
					return err
				}

				for i, value := range values {
					var dashboard dashboard
					if err := json.Unmarshal(value, &dashboard); err != nil {
						return err
					}

					if dashboard.OwnerID != nil {
						fmt.Printf("dashboard %q already has owner %q", dashboard.ID, dashboard.OwnerID)
						continue
					}

					dashboard.OwnerID = &batch[i].UserID

					updated, err := json.Marshal(dashboard)
					if err != nil {
						return err
					}

					return bkt.Put(ids[i], updated)
				}

				return nil
			})
		}
	)

	for i := 0; i < len(mappings); i += batchSize {
		end := i + batchSize
		if end > len(mappings) {
			end = len(mappings)
		}

		flush(mappings[i:end])
	}

	return nil
})

Migration0011_PopulateDashboardsOwnerId backfills owner IDs on dashboards based on the presence of user resource mappings

var Migrations = [...]migration.Spec{

	Migration0001_InitialMigration,

	Migration0002_AddURMByUserIndex,

	Migration0003_TaskOwnerIDUpMigration,

	Migration0004_AddDbrpBuckets,

	Migration0005_AddPkgerBuckets,

	Migration0006_DeleteBucketSessionsv1,

	Migration0007_CreateMetaDataBucket,

	Migration0008_LegacyAuthBuckets,

	Migration0009_LegacyAuthPasswordBuckets,

	Migration0010_AddIndexTelegrafByOrg,

	Migration0011_PopulateDashboardsOwnerId,
}

Migrations contains all the migrations required for the entire of the kv store backing influxdb's metadata.

Functions

func Up

func Up(ctx context.Context, logger *zap.Logger, store kv.SchemaStore) error

Up is a convenience methods which creates a migrator for all migrations and calls Up on it.

Types

type Migration

type Migration struct {
	// contains filtered or unexported fields
}

Migration is a type which implements the migration packages Spec interface It can be used to conveniently create migration specs for the all package

func UpOnlyMigration

func UpOnlyMigration(name string, up MigrationFunc) *Migration

UpOnlyMigration is a migration with an up function and a noop down function

func (*Migration) Down

func (m *Migration) Down(ctx context.Context, store kv.SchemaStore) error

Down delegates to the underlying anonymous down migration function

func (*Migration) MigrationName

func (m *Migration) MigrationName() string

MigrationName returns the underlying name of the migation

func (*Migration) Up

func (m *Migration) Up(ctx context.Context, store kv.SchemaStore) error

Up delegates to the underlying anonymous up migration function

type MigrationFunc

type MigrationFunc func(context.Context, kv.SchemaStore) error

MigrationFunc is a function which can be used as either an up or down operation.