drift

README

dynamo-drift

dynamo-drift is a minimalistic library for performing DynamoDB schema migrations.

Individual migrations are code that is executed in the context of a callback performed per existing table item. You can execute actions within callbacks (update/insert/delete) which are deferred until after all callbacks complete, or use the DynamoDB client directly for immediate table operations. Migrations and actions can modify any table, not just the target table of the migration (this allows a migration from one table into another, for example).

dynamo-drift:

• keeps track of migrations already executed via a separate configurable metadata table
• executes the supplied migrations, each with the requested concurrency
• queues and executes actions (insert/update/delete) after callbacks finish so that they do not interfere with the main table scan

It is the responsibility of the application to determine:

• what a migration consists of and when it may be executed
• when to rollback or undo a migration, and what that means
• how many active migrations exist at any given time

Example

import (
  "github.com/aws/aws-sdk-go/aws/session"
  "github.com/aws/aws-sdk-go/service/dynamodb"
  "github.com/dollarshaveclub/dynamo-drift"
)

func main() {
  // get authenticated DynamoDB client
  client := dynamodb.New(session.Must(session.NewSession()))

  // create migrator
  dd := drift.DynamoDrifter{}
  dd.MetaTableName = "MyMigrationsTable"
  dd.DynamoDB = client

  // initialize
  dd.Init(10, 10)

  // check applied
  migrations, _ := dd.Applied()
  fmt.Printf("migrations already applied: %v\n", len(migrations)) // "migrations already applied: 0"

  // create migration
  migration := &drift.DynamoDrifterMigration{
    Number: 0,
    TableName: "MyApplicationTable",
    Description: "readme test",
    Callback: migrateUp,
  }

  // run migration
  errs := dd.Run(context.TODO(), migration, 1, true)
  for _, err := range errs {
    fmt.Printf("error during up migration: %v", err)
  }

  migrations, _ = dd.Applied()
  fmt.Printf("migrations already applied: %v\n", len(migrations)) // "migrations already applied: 1"

  migration.Callback = migrateDown

  // run undo migration
  errs := dd.Undo(context.TODO(), migration, 1, true)
  for _, err := range errs {
    fmt.Printf("error during down migration: %v", err)
  }

  migrations, _ = dd.Applied()
  fmt.Printf("migrations already applied: %v\n", len(migrations)) // "migrations already applied: 0"
}

// Callbacks are executed once for each item in the target table
func migrateUp(item drift.RawDynamoItem, action *drift.DrifterAction) {
  // modify this table item somehow
}

func migrateDown(item drift.RawDynamoItem, action *drift.DrifterAction) {
  // do something to undo migration
}

Documentation

Index

• func GetBoolAttribute(item RawDynamoItem, attr string) (bool, error)
• func GetByteSliceAttribute(item RawDynamoItem, attr string) ([]byte, error)
• func GetNumberAttribute(item RawDynamoItem, attr string) (string, error)
• func GetStringAttribute(item RawDynamoItem, attr string) (string, error)
• type DrifterAction
  • func (da *DrifterAction) Delete(keys interface{}, tableName string) error
  • func (da *DrifterAction) DynamoDB() *dynamodb.DynamoDB
  • func (da *DrifterAction) Insert(item interface{}, tableName string) error
  • func (da *DrifterAction) Update(keys interface{}, values interface{}, updateExpression string, ...) error
• type DynamoDrifter
  • func (dd *DynamoDrifter) Applied() ([]DynamoDrifterMigration, error)
  • func (dd *DynamoDrifter) Init(pwrite, pread uint) error
  • func (dd *DynamoDrifter) Run(ctx context.Context, migration *DynamoDrifterMigration, concurrency uint, ...) []error
  • func (dd *DynamoDrifter) Undo(ctx context.Context, undoMigration *DynamoDrifterMigration, concurrency uint, ...) []error
• type DynamoDrifterMigration
• type DynamoMigrationFunction
• type MigrationProgress
• type RawDynamoItem

Functions

func GetBoolAttribute

func GetBoolAttribute(item RawDynamoItem, attr string) (bool, error)

GetBoolAttribute returns a string attribute from the raw item, or error if not found or wrong type

func GetByteSliceAttribute

func GetByteSliceAttribute(item RawDynamoItem, attr string) ([]byte, error)

GetByteSliceAttribute returns a string attribute from the raw item, or error if not found or wrong type

func GetNumberAttribute

func GetNumberAttribute(item RawDynamoItem, attr string) (string, error)

GetNumberAttribute returns a number attribute (as string) from the raw item, or error if not found or wrong type

func GetStringAttribute

func GetStringAttribute(item RawDynamoItem, attr string) (string, error)

GetStringAttribute returns a string attribute from the raw item, or error if not found or wrong type

Types

type DrifterAction

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

DrifterAction is an object useful for performing actions within the migration callback. All actions performed by methods on DrifterAction are queued and performed *after* all existing items have been iterated over and callbacks performed. DrifterAction can be used in multiple goroutines by the callback, but must not be retained after the callback returns. If concurrency > 1, order of queued operations cannot be guaranteed.

func (*DrifterAction) Delete

func (da *DrifterAction) Delete(keys interface{}, tableName string) error

Delete deletes the specified item(s). keys is an arbitrary struct with "dynamodbav" annotations. tableName is optional (defaults to migration table).

func (*DrifterAction) DynamoDB

func (da *DrifterAction) DynamoDB() *dynamodb.DynamoDB

DynamoDB returns the DynamoDB client object

func (*DrifterAction) Insert

func (da *DrifterAction) Insert(item interface{}, tableName string) error

Insert inserts item into the specified table. item is an arbitrary struct with "dynamodbav" annotations or a RawDynamoItem tableName is optional (defaults to migration table).

func (*DrifterAction) Update

func (da *DrifterAction) Update(keys interface{}, values interface{}, updateExpression string, expressionAttributeNames map[string]string, tableName string) error

Update mutates the given keys using fields and updateExpression. keys and values are arbitrary structs with "dynamodbav" annotations. IMPORTANT: annotation names must match the names used in updateExpression. updateExpression is the native DynamoDB update expression. Ex: "SET foo = :bar" (in this example keys must have a field annotated "foo" and values must have a field annotated ":bar"). expressionAttributeNames is optional but used if item attribute names are reserved keywords. For example: "SET #n = :name", expressionAttributeNames: map[string]string{"#n":"Name"}.

Required: keys, values, updateExpression

Optional: expressionAttributeNames, tableName (defaults to migration table)

type DynamoDrifter

type DynamoDrifter struct {
	MetaTableName string             // Table to store migration tracking metadata
	DynamoDB      *dynamodb.DynamoDB // Fully initialized and authenticated DynamoDB client
	// contains filtered or unexported fields
}

DynamoDrifter is the object that manages and performs migrations

func (*DynamoDrifter) Applied

func (dd *DynamoDrifter) Applied() ([]DynamoDrifterMigration, error)

Applied returns all applied migrations as tracked in metadata table in ascending order

func (*DynamoDrifter) Init

func (dd *DynamoDrifter) Init(pwrite, pread uint) error

Init creates the metadata table if necessary. It is safe to run Init multiple times (it's a noop if metadata table already exists). pread and pwrite are the provisioned read and write values to use with table creation, if necessary

func (*DynamoDrifter) Run

func (dd *DynamoDrifter) Run(ctx context.Context, migration *DynamoDrifterMigration, concurrency uint, failOnFirstError bool, progressChan chan *MigrationProgress) []error

Run runs an individual migration at the specified concurrency and blocks until finished. concurrency controls the number of table items processed concurrently (value of one will guarantee order of migration actions). failOnFirstError causes Run to abort on first error, otherwise the errors will be queued and reported only after all items have been processed. progressChan is an optional channel on which periodic MigrationProgress messages will be sent

func (*DynamoDrifter) Undo

func (dd *DynamoDrifter) Undo(ctx context.Context, undoMigration *DynamoDrifterMigration, concurrency uint, failOnFirstError bool, progressChan chan *MigrationProgress) []error

Undo "undoes" a migration by running the supplied migration but deletes the corresponding metadata record if successful

type DynamoDrifterMigration

type DynamoDrifterMigration struct {
	Number      uint                    `dynamodbav:"Number" json:"number"`           // Monotonic number of the migration (ascending)
	TableName   string                  `dynamodbav:"TableName" json:"tablename"`     // DynamoDB table the migration applies to
	Description string                  `dynamodbav:"Description" json:"description"` // Free-form description of what the migration does
	Callback    DynamoMigrationFunction `dynamodbav:"-" json:"-"`                     // Callback for each item in the table
}

DynamoDrifterMigration models an individual migration

type DynamoMigrationFunction

type DynamoMigrationFunction func(item RawDynamoItem, action *DrifterAction) error

DynamoMigrationFunction is a callback run for each item in the DynamoDB table item is the raw item action is the DrifterAction object used to mutate/add/remove items

type MigrationProgress

type MigrationProgress struct {
	CallbacksProcessed uint
	ActionsExecuted    uint
	CallbackErrors     []error
	ActionErrors       []error
}

MigrationProgress models periodic progress information communicated back to the caller

type RawDynamoItem

type RawDynamoItem map[string]*dynamodb.AttributeValue

RawDynamoItem models an item from DynamoDB as returned by the API