automation

Documentation

Overview

Package automation contains code to execute high-level cluster operations (e.g. resharding) as a series of low-level operations (e.g. vtctl, shell commands, ...).

Index

• func AddMissingTaskID(tc []*pb.TaskContainer, taskIDGenerator *IDGenerator)
• func AddTask(t *pb.TaskContainer, taskName string, parameters map[string]string)
• func CreateLoggerEventToBufferFunction(output *bytes.Buffer) func(*logutil.LoggerEvent)
• func ExecuteVtctl(ctx context.Context, server string, args []string) (string, error)
• func ExecuteVtworker(ctx context.Context, server string, args []string) (string, error)
• func MarkTaskFailed(t *pb.Task, output string, err error)
• func MarkTaskSucceeded(t *pb.Task, output string)
• func NewTask(taskName string, parameters map[string]string) *pb.Task
• func NewTaskContainer() *pb.TaskContainer
• func NewTaskContainerWithSingleTask(taskName string, parameters map[string]string) *pb.TaskContainer
• type ClusterOperationInstance
  • func NewClusterOperationInstance(clusterOpID string, initialTask *pb.TaskContainer, ...) ClusterOperationInstance
  • func (c ClusterOperationInstance) Clone() ClusterOperationInstance
  • func (c *ClusterOperationInstance) InsertTaskContainers(newTaskContainers []*pb.TaskContainer, pos int)
• type CopySchemaShardTask
  • func (t *CopySchemaShardTask) OptionalParameters() []string
  • func (t *CopySchemaShardTask) RequiredParameters() []string
  • func (t *CopySchemaShardTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)
• type HorizontalReshardingTask
  • func (t *HorizontalReshardingTask) OptionalParameters() []string
  • func (t *HorizontalReshardingTask) RequiredParameters() []string
  • func (t *HorizontalReshardingTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)
• type IDGenerator
  • func (ig *IDGenerator) GetNextID() string
• type MigrateServedTypesTask
  • func (t *MigrateServedTypesTask) OptionalParameters() []string
  • func (t *MigrateServedTypesTask) RequiredParameters() []string
  • func (t *MigrateServedTypesTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)
• type Scheduler
  • func NewScheduler() (*Scheduler, error)
  • func (s *Scheduler) Checkpoint(clusterOp ClusterOperationInstance)
  • func (s *Scheduler) EnqueueClusterOperation(ctx context.Context, req *pb.EnqueueClusterOperationRequest) (*pb.EnqueueClusterOperationResponse, error)
  • func (s *Scheduler) GetClusterOperationDetails(ctx context.Context, req *pb.GetClusterOperationDetailsRequest) (*pb.GetClusterOperationDetailsResponse, error)
  • func (s *Scheduler) Run()
  • func (s *Scheduler) ShutdownAndWait()
• type SplitCloneTask
  • func (t *SplitCloneTask) OptionalParameters() []string
  • func (t *SplitCloneTask) RequiredParameters() []string
  • func (t *SplitCloneTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)
• type SplitDiffTask
  • func (t *SplitDiffTask) OptionalParameters() []string
  • func (t *SplitDiffTask) RequiredParameters() []string
  • func (t *SplitDiffTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)
• type Task
• type WaitForFilteredReplicationTask
  • func (t *WaitForFilteredReplicationTask) OptionalParameters() []string
  • func (t *WaitForFilteredReplicationTask) RequiredParameters() []string
  • func (t *WaitForFilteredReplicationTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)

Functions

func AddMissingTaskID

func AddMissingTaskID(tc []*pb.TaskContainer, taskIDGenerator *IDGenerator)

AddMissingTaskID assigns a task id to each task in "tc".

func AddTask

func AddTask(t *pb.TaskContainer, taskName string, parameters map[string]string)

AddTask adds a new task to an existing task container.

func CreateLoggerEventToBufferFunction

func CreateLoggerEventToBufferFunction(output *bytes.Buffer) func(*logutil.LoggerEvent)

CreateLoggerEventToBufferFunction returns a function to add LoggerEvent structs to a given buffer, one line per event. The buffer can be used to return a multi-line string with all events.

func ExecuteVtctl

func ExecuteVtctl(ctx context.Context, server string, args []string) (string, error)

ExecuteVtctl runs vtctl using vtctlclient. The stream of LoggerEvent messages is concatenated into one output string.

func ExecuteVtworker

func ExecuteVtworker(ctx context.Context, server string, args []string) (string, error)

ExecuteVtworker runs vtworker using vtworkerclient. The stream of LoggerEvent messages is concatenated into one output string.

func MarkTaskFailed

func MarkTaskFailed(t *pb.Task, output string, err error)

MarkTaskFailed marks the task as failed.

func MarkTaskSucceeded

func MarkTaskSucceeded(t *pb.Task, output string)

MarkTaskSucceeded marks the task as done.

func NewTask

func NewTask(taskName string, parameters map[string]string) *pb.Task

NewTask creates a new task protobuf message for "taskName" with "parameters".

func NewTaskContainer

func NewTaskContainer() *pb.TaskContainer

NewTaskContainer creates an empty task container. Use AddTask() to add tasks to it.

func NewTaskContainerWithSingleTask

func NewTaskContainerWithSingleTask(taskName string, parameters map[string]string) *pb.TaskContainer

NewTaskContainerWithSingleTask creates a new task container with exactly one task.

Types

type ClusterOperationInstance

type ClusterOperationInstance struct {
	pb.ClusterOperation
	// contains filtered or unexported fields
}

ClusterOperationInstance is a runtime type which enhances the protobuf message "ClusterOperation" with runtime specific data. Unlike the protobuf message, the additional runtime data will not be part of a checkpoint. Methods of this struct are not thread-safe.

func NewClusterOperationInstance

func NewClusterOperationInstance(clusterOpID string, initialTask *pb.TaskContainer, taskIDGenerator *IDGenerator) ClusterOperationInstance

NewClusterOperationInstance creates a new cluster operation instance with one initial task.

func (ClusterOperationInstance) Clone

func (c ClusterOperationInstance) Clone() ClusterOperationInstance

Clone creates a deep copy of the inner protobuf. Other elements e.g. taskIDGenerator are not deep-copied.

func (*ClusterOperationInstance) InsertTaskContainers

func (c *ClusterOperationInstance) InsertTaskContainers(newTaskContainers []*pb.TaskContainer, pos int)

InsertTaskContainers inserts "newTaskContainers" at pos in the current list of task containers. Existing task containers will be moved after the new task containers.

type CopySchemaShardTask

type CopySchemaShardTask struct {
}

CopySchemaShardTask runs vtctl CopySchemaShard to copy the schema from one shard to another.

func (*CopySchemaShardTask) OptionalParameters

func (t *CopySchemaShardTask) OptionalParameters() []string

OptionalParameters is part of the Task interface.

func (*CopySchemaShardTask) RequiredParameters

func (t *CopySchemaShardTask) RequiredParameters() []string

RequiredParameters is part of the Task interface.

func (*CopySchemaShardTask) Run

func (t *CopySchemaShardTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)

Run is part of the Task interface.

type HorizontalReshardingTask

type HorizontalReshardingTask struct {
}

HorizontalReshardingTask is a cluster operation which allows to increase the number of shards.

func (*HorizontalReshardingTask) OptionalParameters

func (t *HorizontalReshardingTask) OptionalParameters() []string

OptionalParameters is part of the Task interface.

func (*HorizontalReshardingTask) RequiredParameters

func (t *HorizontalReshardingTask) RequiredParameters() []string

RequiredParameters is part of the Task interface.

func (*HorizontalReshardingTask) Run

func (t *HorizontalReshardingTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)

Run is part of the Task interface.

type IDGenerator

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

IDGenerator generates unique task and cluster operation IDs.

func (*IDGenerator) GetNextID

func (ig *IDGenerator) GetNextID() string

GetNextID returns an ID which wasn't returned before.

type MigrateServedTypesTask

type MigrateServedTypesTask struct {
}

MigrateServedTypesTask runs vtctl MigrateServedTypes to migrate a serving type from the source shard to the shards that it replicates to.

func (*MigrateServedTypesTask) OptionalParameters

func (t *MigrateServedTypesTask) OptionalParameters() []string

OptionalParameters is part of the Task interface.

func (*MigrateServedTypesTask) RequiredParameters

func (t *MigrateServedTypesTask) RequiredParameters() []string

RequiredParameters is part of the Task interface.

func (*MigrateServedTypesTask) Run

func (t *MigrateServedTypesTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)

Run is part of the Task interface.

type Scheduler

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

Scheduler executes automation tasks and maintains the execution state.

func NewScheduler

func NewScheduler() (*Scheduler, error)

NewScheduler creates a new instance.

func (*Scheduler) Checkpoint

func (s *Scheduler) Checkpoint(clusterOp ClusterOperationInstance)

Checkpoint should be called every time the state of the cluster op changes. It is used to update the copy of the state in activeClusterOperations.

func (*Scheduler) EnqueueClusterOperation

func (s *Scheduler) EnqueueClusterOperation(ctx context.Context, req *pb.EnqueueClusterOperationRequest) (*pb.EnqueueClusterOperationResponse, error)

EnqueueClusterOperation can be used to start a new cluster operation.

func (*Scheduler) GetClusterOperationDetails

func (s *Scheduler) GetClusterOperationDetails(ctx context.Context, req *pb.GetClusterOperationDetailsRequest) (*pb.GetClusterOperationDetailsResponse, error)

GetClusterOperationDetails can be used to query the full details of active or finished operations.

func (*Scheduler) Run

func (s *Scheduler) Run()

Run processes queued cluster operations.

func (*Scheduler) ShutdownAndWait

func (s *Scheduler) ShutdownAndWait()

ShutdownAndWait shuts down the scheduler and waits infinitely until all pending cluster operations have finished.

type SplitCloneTask

type SplitCloneTask struct {
}

SplitCloneTask runs SplitClone on a remote vtworker to split an existing shard.

func (*SplitCloneTask) OptionalParameters

func (t *SplitCloneTask) OptionalParameters() []string

OptionalParameters is part of the Task interface.

func (*SplitCloneTask) RequiredParameters

func (t *SplitCloneTask) RequiredParameters() []string

RequiredParameters is part of the Task interface.

func (*SplitCloneTask) Run

func (t *SplitCloneTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)

Run is part of the Task interface.

type SplitDiffTask

type SplitDiffTask struct {
}

SplitDiffTask runs SplitDiff on a remote vtworker to compare the old shard against its new split shards.

func (*SplitDiffTask) OptionalParameters

func (t *SplitDiffTask) OptionalParameters() []string

OptionalParameters is part of the Task interface.

func (*SplitDiffTask) RequiredParameters

func (t *SplitDiffTask) RequiredParameters() []string

RequiredParameters is part of the Task interface.

func (*SplitDiffTask) Run

func (t *SplitDiffTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)

Run is part of the Task interface.

type Task

type Task interface {
	// Run executes the task using the key/values from parameters.
	// "newTaskContainers" contains new tasks which the task can emit. They'll be inserted in the cluster operation directly after this task. It may be "nil".
	// "output" may be empty. It contains any text which maybe must e.g. to debug the task or show it in the UI.
	Run(parameters map[string]string) (newTaskContainers []*pb.TaskContainer, output string, err error)

	// RequiredParameters() returns a list of parameter keys which must be provided as input for Run().
	RequiredParameters() []string

	// OptionalParameters() returns a list of parameter keys which are optional input for Run().
	OptionalParameters() []string
}

Task implementations can be executed by the scheduler.

type WaitForFilteredReplicationTask

type WaitForFilteredReplicationTask struct {
}

WaitForFilteredReplicationTask runs vtctl WaitForFilteredReplication to block until the destination master (i.e. the receiving side of the filtered replication) has caught up up to max_delay with the source shard.

func (*WaitForFilteredReplicationTask) OptionalParameters

func (t *WaitForFilteredReplicationTask) OptionalParameters() []string

OptionalParameters is part of the Task interface.

func (*WaitForFilteredReplicationTask) RequiredParameters

func (t *WaitForFilteredReplicationTask) RequiredParameters() []string

RequiredParameters is part of the Task interface.

func (*WaitForFilteredReplicationTask) Run

func (t *WaitForFilteredReplicationTask) Run(parameters map[string]string) ([]*pb.TaskContainer, string, error)

Run is part of the Task interface.