migration

Documentation

Index

• Variables
• type Migration
• type Migrator
  • func New(ds datastore.TxnDatastore, migrations map[int]Migration) *Migrator
  • func (m *Migrator) Ensure() error

Variables

var V1MultitenancyMigration = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		cidOwners, err := v0CidOwners(ds)
		if err != nil {
			return fmt.Errorf("getting cid owners: %s", err)
		}
		log.Infof("Starting job logger migration...")
		if err := migrateJobLogger(ds, cidOwners); err != nil {
			return fmt.Errorf("migrating job logger: %s", err)
		}
		log.Infof("Job logger migration finished")

		log.Infof("Starting storage info migration...")
		if err := migrateStorageInfo(ds, cidOwners); err != nil {
			return fmt.Errorf("migrating storage info: %s", err)
		}
		log.Infof("Storage info migration finished")

		log.Infof("Starting trackstore migration...")
		if err := migrateTrackstore(ds, cidOwners); err != nil {
			return fmt.Errorf("migrating trackstore: %s", err)
		}
		log.Infof("Trackstore migration finished")

		log.Infof("Starting started deals migration...")
		if err := migrateStartedDeals(ds); err != nil {
			return fmt.Errorf("migrating trackstore: %s", err)
		}
		log.Infof("Started deals migration finished")

		log.Infof("Starting pinstore filling migration...")
		if err := pinstoreFilling(ds, cidOwners); err != nil {
			return fmt.Errorf("filling pinstore: %s", err)
		}
		log.Infof("Pinstore filling migration finished")

		return nil
	},
}

V1MultitenancyMigration contains the logic to upgrade a datastore from version 0 to version 1. Transactionality is disabled since is a big migration.

var V2StorageInfoDealIDs = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		propDealIDMap, err := v2GenProposalCidToDealID(ds)
		if err != nil {
			return fmt.Errorf("getting cid owners: %s", err)
		}
		log.Infof("Starting DealID filling...")
		if err := v2MigrationDealIDFilling(ds, propDealIDMap); err != nil {
			return fmt.Errorf("migrating job logger: %s", err)
		}
		log.Infof("DealID filling migration finished")

		return nil
	},
}

V2StorageInfoDealIDs contains the logic to upgrade a datastore from version 1 to version 2. Transactionality is disabled since is a big migration.

var V3StorageJobsIndexMigration = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		q := query.Query{Prefix: "/ffs/scheduler/sjstore/job"}
		res, err := ds.Query(q)
		if err != nil {
			return fmt.Errorf("querying sjstore jobs: %s", err)
		}
		defer func() { _ = res.Close() }()

		var count int
		var lock sync.Mutex
		var errors []string
		lim := make(chan struct{}, 1000)
		for r := range res.Next() {
			if r.Error != nil {
				return fmt.Errorf("iterating results: %s", r.Error)
			}

			lim <- struct{}{}

			r := r
			go func() {
				defer func() { <-lim }()

				var job storageJob
				if err := json.Unmarshal(r.Value, &job); err != nil {
					lock.Lock()
					errors = append(errors, fmt.Sprintf("unmarshaling job: %s", err))
					lock.Unlock()
					return
				}

				apiidKey := datastore.NewKey("/ffs/scheduler/sjstore/apiid").ChildString(job.APIID).ChildString(job.Cid.String()).ChildString(fmt.Sprintf("%d", job.CreatedAt))
				cidKey := datastore.NewKey("/ffs/scheduler/sjstore/cid").ChildString(job.Cid.String()).ChildString(job.APIID).ChildString(fmt.Sprintf("%d", job.CreatedAt))

				if err := ds.Put(apiidKey, []byte(job.ID)); err != nil {
					lock.Lock()
					errors = append(errors, fmt.Sprintf("putting apiid index record in datastore: %s", err))
					lock.Unlock()
					return

				}
				if err := ds.Put(cidKey, []byte(job.ID)); err != nil {
					lock.Lock()
					errors = append(errors, fmt.Sprintf("putting cid index record in datastore: %s", err))
					lock.Unlock()
					return

				}

			}()
			count++
		}

		for i := 0; i < cap(lim); i++ {
			lim <- struct{}{}
		}

		if len(errors) > 0 {
			for _, m := range errors {
				log.Error(m)
			}
			return fmt.Errorf("migration had %d errors", len(errors))
		}

		log.Infof("migration indexed %d storage-jobs", count)

		return nil
	},
}

V3StorageJobsIndexMigration contains the logic to upgrade a datastore from version 2 to version 3.

var V4RecordsMigration = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		if err := v4IncludeIDInRetrievalRecords(ds); err != nil {
			return fmt.Errorf("including ids in retrieval records: %s", err)
		}

		if err := v4PopulateUpdatedAtIndex(ds); err != nil {
			return fmt.Errorf("populating updated at index: %s", err)
		}

		return nil
	},
}

V4RecordsMigration contains the logic to upgrade a datastore from version 3 to version 4.

var V5DeleteOldMinerIndex = Migration{
	UseTxn: false,
	Run: func(ds datastoreReaderWriter) error {
		q := query.Query{Prefix: "/index/miner/chainstore"}
		res, err := ds.Query(q)
		if err != nil {
			return fmt.Errorf("querying records: %s", err)
		}
		defer func() { _ = res.Close() }()

		var count int
		for v := range res.Next() {
			if err := ds.Delete(datastore.NewKey(v.Key)); err != nil {
				return fmt.Errorf("deleting miner chainstore key: %s", err)
			}
			count++
		}
		log.Infof("deleted %d chainstore keys", count)

		return nil
	},
}

V5DeleteOldMinerIndex contains the logic to upgrade a datastore from version 4 to version 5.

Types

type Migration

type Migration struct {
	Run    func(datastoreReaderWriter) error
	UseTxn bool
}

Migration runs a vA->v(A+1) migration. UseTxn indicates if this migration should be run in a transaction.

type Migrator

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

Migrator ensures a datastore goes through all the needed Migrations to upgrade its current version to the latest version.

func New

func New(ds datastore.TxnDatastore, migrations map[int]Migration) *Migrator

New returns a new Migrator.

func (*Migrator) Ensure

func (m *Migrator) Ensure() error

Ensure detects the current datastore version, and runs all the known migrations to upgrade to the latest known version.