praefect

Documentation

Overview

Package praefect provides data models and datastore persistence abstractions for tracking the state of repository replicas.

See original design discussion: https://gitlab.com/gitlab-org/gitaly/issues/1495

Package praefect is a Gitaly reverse proxy for transparently routing gRPC calls to a set of Gitaly services.

Index

• Variables
• type Coordinator
  • func NewCoordinator(l *logrus.Logger, storageLoc string, ...) *Coordinator
  • func (c *Coordinator) GetStorageNode(storage string) (Node, error)
  • func (c *Coordinator) RegisterNode(storageLoc, listenAddr string) error
  • func (c *Coordinator) RegisterProtos(protos ...*descriptor.FileDescriptorProto) error
• type Datastore
• type JobState
• type MemoryDatastore
  • func NewMemoryDatastore(cfg config.Config) *MemoryDatastore
  • func (md *MemoryDatastore) CreateSecondaryReplJobs(source Repository) ([]uint64, error)
  • func (md *MemoryDatastore) GetJobs(state JobState, storage string, count int) ([]ReplJob, error)
  • func (md *MemoryDatastore) GetSecondaries(primary Repository) ([]string, error)
  • func (md *MemoryDatastore) SetSecondaries(primary Repository, secondaries []string) error
  • func (md *MemoryDatastore) UpdateReplJob(jobID uint64, newState JobState) error
• type Node
• type ReplJob
• type ReplJobsDatastore
• type ReplMgr
  • func NewReplMgr(storage string, log *logrus.Logger, ds ReplJobsDatastore, c *Coordinator, ...) ReplMgr
  • func (r ReplMgr) ProcessBacklog(ctx context.Context) error
  • func (r ReplMgr) ScheduleReplication(ctx context.Context, repo Repository) error
• type ReplMgrOpt
  • func WithReplicator(r Replicator) ReplMgrOpt
  • func WithWhitelist(whitelistedRepos []string) ReplMgrOpt
• type ReplicasDatastore
• type Replicator
• type Repository
• type Server
  • func NewServer(c *Coordinator, repl ReplMgr, grpcOpts []grpc.ServerOption, l *logrus.Logger) *Server
  • func (srv *Server) Shutdown(ctx context.Context) error
  • func (srv *Server) Start(lis net.Listener) error

Variables

var ErrInvalidReplTarget = errors.New("target repository fails preconditions for replication")

ErrInvalidReplTarget indicates a target repository cannot be chosen because it fails preconditions for being replicatable

var ErrSecondariesMissing = errors.New("repository missing secondary replicas")

ErrSecondariesMissing indicates the repository does not have any backup replicas

Types

type Coordinator

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

Coordinator takes care of directing client requests to the appropriate downstream server. The coordinator is thread safe; concurrent calls to register nodes are safe.

func NewCoordinator

func NewCoordinator(l *logrus.Logger, storageLoc string, fileDescriptors ...*descriptor.FileDescriptorProto) *Coordinator

NewCoordinator returns a new Coordinator that utilizes the provided logger

func (*Coordinator) GetStorageNode

func (c *Coordinator) GetStorageNode(storage string) (Node, error)

GetStorageNode returns the registered node for the given storage location

func (*Coordinator) RegisterNode

func (c *Coordinator) RegisterNode(storageLoc, listenAddr string) error

RegisterNode will direct traffic to the supplied downstream connection when the storage location is encountered.

func (*Coordinator) RegisterProtos

func (c *Coordinator) RegisterProtos(protos ...*descriptor.FileDescriptorProto) error

RegisterProtos allows coordinator to register new protos on the fly

type Datastore

type Datastore interface {
	ReplJobsDatastore
	ReplicasDatastore
}

Datastore is a data persistence abstraction for all of Praefect's persistence needs

type JobState

type JobState uint8

JobState is an enum that indicates the state of a job

const (
	// JobStatePending is the initial job state when it is not yet ready to run
	// and may indicate recovery from a failure prior to the ready-state
	JobStatePending JobState = 1 << iota
	// JobStateReady indicates the job is now ready to proceed
	JobStateReady
	// JobStateInProgress indicates the job is being processed by a worker
	JobStateInProgress
	// JobStateComplete indicates the job is now complete
	JobStateComplete
	// JobStateCancelled indicates the job was cancelled. This can occur if the
	// job is no longer relevant (e.g. a node is moved out of a shard)
	JobStateCancelled
)

type MemoryDatastore

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

MemoryDatastore is a simple datastore that isn't persisted to disk. It is only intended for early beta requirements and as a reference implementation for the eventual SQL implementation

func NewMemoryDatastore

func NewMemoryDatastore(cfg config.Config) *MemoryDatastore

NewMemoryDatastore returns an initialized in-memory datastore

func (*MemoryDatastore) CreateSecondaryReplJobs

func (md *MemoryDatastore) CreateSecondaryReplJobs(source Repository) ([]uint64, error)

CreateSecondaryReplJobs creates a replication job for each secondary that backs the specified repository. Upon success, the job IDs will be returned.

func (*MemoryDatastore) GetJobs

func (md *MemoryDatastore) GetJobs(state JobState, storage string, count int) ([]ReplJob, error)

GetJobs is a more general method to retrieve jobs of a certain state from the datastore

func (*MemoryDatastore) GetSecondaries

func (md *MemoryDatastore) GetSecondaries(primary Repository) ([]string, error)

GetSecondaries will return the set of secondary storage locations for a given repository if they exist

func (*MemoryDatastore) SetSecondaries

func (md *MemoryDatastore) SetSecondaries(primary Repository, secondaries []string) error

SetSecondaries will replace the set of replicas for a repository

func (*MemoryDatastore) UpdateReplJob

func (md *MemoryDatastore) UpdateReplJob(jobID uint64, newState JobState) error

UpdateReplJob updates an existing replication job's state

type Node

type Node struct {
	Storage string
	// contains filtered or unexported fields
}

Node is a wrapper around the grpc client connection for a backend Gitaly node

type ReplJob

type ReplJob struct {
	ID     uint64     // autoincrement ID
	Target string     // which storage location to replicate to?
	Source Repository // source for replication
	State  JobState
}

ReplJob is an instance of a queued replication job. A replication job is meant for updating the repository so that it is synced with the primary copy. Scheduled indicates when a replication job should be performed.

type ReplJobsDatastore

type ReplJobsDatastore interface {
	// GetJobs fetches a list of chronologically ordered replication
	// jobs for the given storage replica. The returned list will be at most
	// count-length.
	GetJobs(flag JobState, storage string, count int) ([]ReplJob, error)

	// CreateSecondaryJobs will create replication jobs for each secondary
	// replica of a repository known to the datastore. A set of replication job
	// ID's for the created jobs will be returned upon success.
	CreateSecondaryReplJobs(source Repository) ([]uint64, error)

	// UpdateReplJob updates the state of an existing replication job
	UpdateReplJob(jobID uint64, newState JobState) error
}

ReplJobsDatastore represents the behavior needed for fetching and updating replication jobs from the datastore

type ReplMgr

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

ReplMgr is a replication manager for handling replication jobs

func NewReplMgr

func NewReplMgr(storage string, log *logrus.Logger, ds ReplJobsDatastore, c *Coordinator, opts ...ReplMgrOpt) ReplMgr

NewReplMgr initializes a replication manager with the provided dependencies and options

func (ReplMgr) ProcessBacklog

func (r ReplMgr) ProcessBacklog(ctx context.Context) error

ProcessBacklog will process queued jobs. It will block while processing jobs.

func (ReplMgr) ScheduleReplication

func (r ReplMgr) ScheduleReplication(ctx context.Context, repo Repository) error

ScheduleReplication will store a replication job in the datastore for later execution. It filters out projects that are not whitelisted. TODO: add a parameter to delay replication

type ReplMgrOpt

type ReplMgrOpt func(*ReplMgr)

ReplMgrOpt allows a replicator to be configured with additional options

func WithReplicator

func WithReplicator(r Replicator) ReplMgrOpt

WithReplicator overrides the default replicator

func WithWhitelist

func WithWhitelist(whitelistedRepos []string) ReplMgrOpt

WithWhitelist will configure a whitelist for repos to allow replication

type ReplicasDatastore

type ReplicasDatastore interface {
	// GetSecondaries will retrieve all secondary replica storage locations for
	// a primary replica
	GetSecondaries(primary Repository) ([]string, error)

	// SetSecondaries will set the secondary storage locations for a repository
	// in a primary replica.
	SetSecondaries(primary Repository, secondaries []string) error
}

ReplicasDatastore manages accessing and setting which secondary replicas backup a repository

type Replicator

type Replicator interface {
	Replicate(ctx context.Context, source Repository, target Node) error
}

Replicator performs the actual replication logic between two nodes

type Repository

type Repository struct {
	RelativePath string // relative path of repository
	Storage      string // storage location, e.g. default
}

Repository provides all necessary information to address a repository hosted in a specific Gitaly replica

type Server

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

Server is a praefect server

func NewServer

func NewServer(c *Coordinator, repl ReplMgr, grpcOpts []grpc.ServerOption, l *logrus.Logger) *Server

NewServer returns an initialized praefect gPRC proxy server configured with the provided gRPC server options

func (*Server) Shutdown

func (srv *Server) Shutdown(ctx context.Context) error

Shutdown will attempt a graceful shutdown of the grpc server. If unable to gracefully shutdown within the context deadline, it will then forcefully shutdown the server and return a context cancellation error.

func (*Server) Start

func (srv *Server) Start(lis net.Listener) error

Start will start the praefect gRPC proxy server listening at the provided listener. Function will block until the server is stopped or an unrecoverable error occurs.