juju: github.com/juju/juju/state/presence Index | Files

package presence

import "github.com/juju/juju/state/presence"

Copyright 2017 Canonical Ltd. Licensed under the AGPLv3, see LICENCE file for details.

The presence package implements an interface for observing liveness of arbitrary keys (agents, processes, etc) on top of MongoDB. The design works by periodically updating the database so that watchers can tell an arbitrary key is alive.

Index

Package Files

pingbatcher.go presence.go pruner.go

func RemovePresenceForModel Uses

func RemovePresenceForModel(base *mgo.Collection, modelTag names.ModelTag) error

RemovePresenceForModel removes all of the records of entities for a given model across all of the collections.

type Agent Uses

type Agent interface {
    AgentPresence() (bool, error)
    SetAgentPresence() (*Pinger, error)
    WaitAgentPresence(time.Duration) error
}

Agent shouldn't really live here -- it's not used in this package, and is implemented by a couple of state types for the convenience of the apiserver -- but one of the methods returns a concrete *Pinger, and that ties it down here quite effectively (until we want to take on the task of cleaning it up and promoting it to core, which might well never happen).

type Change Uses

type Change struct {
    Key   string
    Alive bool
}

Change holds a liveness change notification.

type PingBatcher Uses

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

PingBatcher aggregates several pingers to update the database on a fixed schedule.

func NewDeadPingBatcher Uses

func NewDeadPingBatcher(err error) *PingBatcher

NewDeadPingBatcher returns a PingBatcher that is already stopped with an error.

func NewPingBatcher Uses

func NewPingBatcher(base *mgo.Collection, flushInterval time.Duration) *PingBatcher

NewPingBatcher creates a worker that will batch ping requests and prepare them for insertion into the Pings collection. Pass in the base "presence" collection. flushInterval is how often we will write the contents to the database. It should be shorter than the 30s slot window for us to not cause active pingers to show up as missing. The current default is 1s as it provides a good balance of significant-batching-for-performance while still having responsiveness to agents coming alive. Note that we don't strictly sync on flushInterval times, but use a range of times around that interval to avoid having all ping batchers get synchronized and still be issuing all requests concurrently.

func (*PingBatcher) Kill Uses

func (pb *PingBatcher) Kill()

Kill is part of the worker.Worker interface.

func (*PingBatcher) Ping Uses

func (pb *PingBatcher) Ping(modelUUID string, slot int64, fieldKey string, fieldBit uint64) error

Ping should be called by a Pinger when it is ready to update its time slot. It passes in all of the pre-resolved information (what exact field bit is being set), rather than the higher level "I'm pinging for this Agent". Internally, we synchronize with the main worker loop. Which means that this function will return once the main loop recognizes that we have a ping request but it will not have updated its internal structures, and certainly not the database.

func (*PingBatcher) Stop Uses

func (pb *PingBatcher) Stop() error

Stop this PingBatcher, part of the extended Worker interface.

func (*PingBatcher) Sync Uses

func (pb *PingBatcher) Sync() error

Sync schedules a flush of the current state to the database. This is not immediate, but actually within a short timeout so that many calls to sync in a short time frame will only trigger one write to the database.

func (*PingBatcher) Wait Uses

func (pb *PingBatcher) Wait() error

Wait returns when the Pinger has stopped, and returns the first error it encountered.

type PingRecorder Uses

type PingRecorder interface {
    Ping(modelUUID string, slot int64, fieldKey string, fieldBit uint64) error
}

type Pinger Uses

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

Pinger periodically reports that a specific key is alive, so that watchers interested on that fact can react appropriately.

func NewPinger Uses

func NewPinger(base *mgo.Collection, modelTag names.ModelTag, key string, recorderFunc func() PingRecorder) *Pinger

NewPinger returns a new Pinger to report that key is alive. It starts reporting after Start is called.

func (*Pinger) Kill Uses

func (p *Pinger) Kill()

Kill is part of the worker.Worker interface.

func (*Pinger) KillForTesting Uses

func (p *Pinger) KillForTesting() error

KillForTesting stops p's periodical ping and immediately reports that it is dead. TODO(ericsnow) We should be able to drop this and the two kill* methods.

func (*Pinger) Start Uses

func (p *Pinger) Start() error

Start starts periodically reporting that p's key is alive.

func (*Pinger) Stop Uses

func (p *Pinger) Stop() error

Stop stops p's periodical ping. Watchers will not notice p has stopped pinging until the previous ping times out.

func (*Pinger) Wait Uses

func (p *Pinger) Wait() error

Wait returns when the Pinger has stopped, and returns the first error it encountered.

type Pruner Uses

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

Pruner tracks the state of removing unworthy beings from the presence.beings and presence.pings collections. Being sequences are unworthy once their sequence has been superseded, and pings older than 2 slots are no longer referenced.

func NewPruner Uses

func NewPruner(modelUUID string, beings *mgo.Collection, pings *mgo.Collection, delta time.Duration) *Pruner

NewPruner returns an object that is ready to prune the Beings collection of old beings sequence entries that we no longer need.

func (*Pruner) Prune Uses

func (p *Pruner) Prune(memCache map[int64]string) error

Prune removes beings from the beings collection that have been superseded by another entry with a higher sequence. It also removes pings that are outside of the 'active' range (the last few slots)

type Watcher Uses

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

A Watcher can watch any number of pinger keys for liveness changes.

func NewDeadWatcher Uses

func NewDeadWatcher(err error) *Watcher

NewDeadWatcher returns a new watcher that is already dead and always returns the given error from its Err method.

func NewWatcher Uses

func NewWatcher(base *mgo.Collection, modelTag names.ModelTag) *Watcher

NewWatcher returns a new Watcher.

func (*Watcher) Alive Uses

func (w *Watcher) Alive(key string) (bool, error)

Alive returns whether the key is currently considered alive by w, or an error in case the watcher is dying.

func (*Watcher) Dead Uses

func (w *Watcher) Dead() <-chan struct{}

Dead returns a channel that is closed when the watcher has stopped.

func (*Watcher) Err Uses

func (w *Watcher) Err() error

Err returns the error with which the watcher stopped. It returns nil if the watcher stopped cleanly, tomb.ErrStillAlive if the watcher is still running properly, or the respective error if the watcher is terminating or has terminated with an error.

func (*Watcher) Kill Uses

func (w *Watcher) Kill()

Kill is part of the worker.Worker interface.

func (*Watcher) StartSync Uses

func (w *Watcher) StartSync()

StartSync forces the watcher to load new events from the database.

func (*Watcher) Stop Uses

func (w *Watcher) Stop() error

Stop stops all the watcher activities.

func (*Watcher) String Uses

func (w *Watcher) String() string

func (*Watcher) Sync Uses

func (w *Watcher) Sync()

Sync forces the watcher to load new events from the database and blocks until all events have been dispatched.

func (*Watcher) Unwatch Uses

func (w *Watcher) Unwatch(key string, ch chan<- Change)

Unwatch stops watching the liveness of key via ch.

func (*Watcher) Wait Uses

func (w *Watcher) Wait() error

Wait is part of the worker.Worker interface.

func (*Watcher) Watch Uses

func (w *Watcher) Watch(key string, ch chan<- Change)

Watch starts watching the liveness of key. An event will be sent onto ch to report the initial status for the key, and from then on a new event will be sent whenever a change is detected. Change values sent to the channel must be consumed, or the whole watcher will blocked.

Package presence imports 14 packages (graph) and is imported by 95 packages. Updated 2019-09-14. Refresh now. Tools for package owners.