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
• type Datastore
• type MemoryDatastore
  • func NewMemoryDatastore(cfg config.Config, immediate time.Time) *MemoryDatastore
  • func (md *MemoryDatastore) GetReplJobs(storage string, since time.Time, count int) ([]ReplJob, error)
  • func (md *MemoryDatastore) GetSecondaries(primary Repository) ([]string, error)
  • func (md *MemoryDatastore) PutReplJob(target Repository, scheduled time.Time) error
  • func (md *MemoryDatastore) SetSecondaries(primary Repository, secondaries []string) 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) *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.

type Datastore

type Datastore interface {
	ReplJobsDatastore
	ReplicasDatastore
}

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

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, immediate time.Time) *MemoryDatastore

NewMemoryDatastore returns an initialized in-memory datastore

func (*MemoryDatastore) GetReplJobs

func (md *MemoryDatastore) GetReplJobs(storage string, since time.Time, count int) ([]ReplJob, error)

GetReplJobs will return any replications jobs for the specified storage since the specified scheduled time up to the specified result limit.

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) PutReplJob

func (md *MemoryDatastore) PutReplJob(target Repository, scheduled time.Time) error

PutReplJob will create or update an existing replication job by scheduling it at the specified time

func (*MemoryDatastore) SetSecondaries

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

SetSecondaries will replace the set of replicas for a repository

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 {
	Target    string     // which storage location to replicate to?
	Source    Repository // source for replication
	Scheduled time.Time
}

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 {
	// GetReplJobs fetches a list of chronologically ordered replication
	// jobs for the given storage replica
	GetReplJobs(storage string, since time.Time, count int) ([]ReplJob, error)

	// PutReplJob will update or create a replication job for the specified repo
	// on a specific storage node
	PutReplJob(repo Repository, when time.Time) 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.