replication

README

Replication: facilities for scheduling a pod across multiple nodes

The intent store is a relatively open book - for maintenance and introspection purposes, it should be modifable by users. This facility allows declarative management of a pod across multiple nodes, with potentially different leader/follower or other configurations.

Replication is managed by clients exclusively. The only state tracked by these utilities are the manifests found in the intent store.

General usage

package mycorp

import (
    "fmt"

    "github.com/square/p2/pkg/allocation"
    "github.com/square/p2/pkg/health"
    "github.com/square/p2/pkg/pods"
    "github.com/square/p2/pkg/store/consul"
    "github.com/square/p2/pkg/replication"
    "github.com/mycorp/myp2allocator"
)

func LaunchMysqlPod(mysqlManfiest pods.PodManifest, allocator allocation.Allocator, store consul.Store, healthChecker *health.Checker) (replication.Result, error){
    request := allocation.Request {
        Manifest: mysqlManifest,
        Replicas: 3,
    }
    allocated, err := allocator.Allocate(request)
    master := allocated.MasterNode()
    if !master.Valid() {
        return nil, fmt.Errorf("No ste")
    }
    mysqlManifest.Config["master_node"] = master.Name
    if err != nil {
        return nil, fmt.Errorf("Could not allocate for pod %s: %s", mysqlManifest.Id, err)
    }

    replicator := replication.NewReplicator(mysqlManifest, allocated)
    replicator.MinimumNodes = 2
    return replicator.Enact(store, healthChecker)
}

This example code allocates and deploys 3 instances of the given MySQL pod to different nodes and updates the pod manifest's configuration.

Documentation

Index

• Constants
• func AggregateHealth(id types.PodID, checker checker.HealthChecker, watchDelay time.Duration) *podHealth
• func IsFatalError(err error) bool
• type Labeler
• type Replication
• type Replicator
  • func NewReplicator(manifest manifest.Manifest, logger logging.Logger, nodes []types.NodeName, ...) (Replicator, error)
• type Store

Constants

const (
	DefaultConcurrentReality = 3

	// Normal replications will have no timeout, but daemon sets will
	// because it is unlikely that all hosts are healthy at all times
	NoTimeout = time.Duration(-1)
)

Functions

func AggregateHealth

func AggregateHealth(id types.PodID, checker checker.HealthChecker, watchDelay time.Duration) *podHealth

func IsFatalError

func IsFatalError(err error) bool

Types

type Labeler

type Labeler interface {
	GetLabels(labelType labels.Type, id string) (labels.Labeled, error)
	SetLabelsTxn(ctx context.Context, labelType labels.Type, id string, labels map[string]string) error
}

type Replication

type Replication interface {
	// Proceed with the prescribed replication
	Enact()

	// Cancel the prescribed replication
	Cancel()

	// Will block until the r.quitCh is closed
	// this is used to synchronize updates which quickly cancel and re-enact the replicaton
	WaitForReplication()

	CompletedCount() int32

	InProgress() bool

	// SetManifest() can be used to change the manifest while a replication is in progress
	SetManifest(manifest.Manifest)

	// SetTimeout() is used to change the timeout used for the replication while it is in progress
	SetTimeout(timeout time.Duration)
}

type Replicator

type Replicator interface {
	InitializeReplication(
		overrideLock bool,
		ignoreControllers bool,
		concurrentRealityNodes int,
		rateLimitInterval time.Duration,
		podLabels map[string]string,
	) (Replication, chan error, error)

	// InitializeDaemonSetReplication creates a Replication with parameters suitable for a daemon set.
	// Specifically:
	// * hosts are not locked
	// * replication controllers are ignored
	// * and preparers are not checked.
	InitializeDaemonSetReplication(
		nodeQueue chan types.NodeName,
		concurrentRealityRequests int,
		rateLimitInterval time.Duration,
		podLabels map[string]string,
	) (Replication, chan error, error)
}

func NewReplicator

func NewReplicator(
	manifest manifest.Manifest,
	logger logging.Logger,
	nodes []types.NodeName,
	active int,
	store Store,
	txner transaction.Txner,
	labeler Labeler,
	health checker.HealthChecker,
	threshold health.HealthState,
	lockMessage string,
	timeout time.Duration,
	healthWatchDelay time.Duration,
) (Replicator, error)

type Store

type Store interface {
	SetPodTxn(
		ctx context.Context,
		podPrefix consul.PodPrefix,
		nodename types.NodeName,
		manifest manifest.Manifest,
	) error
	Pod(podPrefix consul.PodPrefix, nodename types.NodeName, podId types.PodID) (manifest.Manifest, time.Duration, error)
	NewSession(name string, renewalCh <-chan time.Time) (consul.Session, chan error, error)
	LockHolder(key string) (string, string, error)
	DestroyLockHolder(id string) error
}