daemondata

Documentation

Overview

Package daemondata implements daemon journaled data

import "opensvc.com/opensvc/daemon/daemondata" cmdC, cancel := daemondata.Start(context.Background()) defer cancel() dataBus := daemondata.New(cmdC)

status := dataBus.ClusterData() // retrieve daemon data bus.ApplyFull("remote-1", remoteNodeStatus) bus.ApplyPatch("remote-1", patchMsg) bus.CommitPending(context.Background()) status = bus.ClusterData() localNodeStatus := bus.GetLocalNodeStatus()

Index

• Variables
• func BusFromContext(ctx context.Context) chan<- Caller
• func ContextWithBus(parent context.Context, cmd chan<- Caller) context.Context
• func PropagationInterval() time.Duration
• func Start(parent context.Context, drainDuration time.Duration) (chan<- Caller, chan<- *hbtype.Msg, context.CancelFunc)
• type Caller
• type HbMessageType
• type T
  • func FromContext(ctx context.Context) *T
  • func New(cmd chan<- Caller) *T
  • func (t T) ClusterData() *cluster.Data
  • func (t T) ClusterNodeData(n string) *node.Node
  • func (t T) DaemonRefresh()
  • func (t T) DropPeerNode(peerNode string) error
  • func (t T) GetHbMessageType() HbMessageType
  • func (t T) SetHBSendQ(hbSendQ chan<- hbtype.Msg) error
  • func (t T) Stats() callcount.Stats

Variables

var (

	// SubscriptionQueueSize is size of "daemondata" subscription
	SubscriptionQueueSize = 40000

	ErrDrained = errors.New("drained command")
)

Functions

func BusFromContext

func BusFromContext(ctx context.Context) chan<- Caller

BusFromContext function returns the command chan stored in the context

func ContextWithBus

func ContextWithBus(parent context.Context, cmd chan<- Caller) context.Context

ContextWithBus function returns copy of parent, including the daemon data command chan.

func PropagationInterval

func PropagationInterval() time.Duration

func Start

func Start(parent context.Context, drainDuration time.Duration) (chan<- Caller, chan<- *hbtype.Msg, context.CancelFunc)

Start runs the daemon journaled data manager

It returns a cmdC chan to submit actions on cluster data

Types

type Caller

type Caller interface {
	SetError(error)
	// contains filtered or unexported methods
}

Caller defines interface to implement for daemondata loop cmd processing the function will hold the daemondata loop while running

err := Caller.call(ctx, d)
Caller.SetError(err)

type HbMessageType

type HbMessageType struct {
	Type        string
	Nodes       []string
	JoinedNodes []string
	Gens        map[string]uint64
}

type T

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

T struct holds a daemondata manager cmdC to submit orders

func FromContext

func FromContext(ctx context.Context) *T

func New

func New(cmd chan<- Caller) *T

New returns a new *T from an existing daemondata manager

func (T) ClusterData

func (t T) ClusterData() *cluster.Data

ClusterData returns deep copy of status

func (T) ClusterNodeData

func (t T) ClusterNodeData(n string) *node.Node

ClusterNodeData returns deep copy of cluster node data for node n. It returns nil when node n is not found in cluster data.

func (T) DaemonRefresh

func (t T) DaemonRefresh()

DaemonRefresh updates the private dataset of a daemon subsystem (scheduler, dns, ...)

func (T) DropPeerNode

func (t T) DropPeerNode(peerNode string) error

DropPeerNode is a public method to drop peer node from t. It uses private call to func (d *data) dropPeer. It is used by a node stale detector to call drop peer on stale <peer> detection.

func (T) GetHbMessageType

func (t T) GetHbMessageType() HbMessageType

GetHbMessageType returns either "undef", "ping", "full" or "patch" Used by nmon start up to determine if rejoin can be skipped.

func (T) SetHBSendQ

func (t T) SetHBSendQ(hbSendQ chan<- hbtype.Msg) error

SetHBSendQ defines daemondata hbSendQ. The hbSendQ is used during queueNewHbMsg to push heartbeat message to this queue, see usage example for hb msgToTx multiplexer Example:

msgC := make(chan hbtype.Msg)
SetHBSendQ(msgC) // inform daemondata we are listening on this queue
defer SetHBSendQ(nil) // inform daemondata, we are not anymore reading on this queue
for {
   select {
   case msg := <- msgC:
      ...
   case <-ctx.Done():
      return
   }
}

func (T) Stats

func (t T) Stats() callcount.Stats