controller

Documentation

Index

• Variables
• func LoadKubeStateForResync(broker keyval.ProtoBroker, log logging.Logger) (event *api.DBResync, values map[string]datasync.KeyVal, err error)
• func ResyncDatabase(broker keyval.ProtoBroker, kubeStateData api.KubeStateData) error
• type Config
• type Controller
  • func NewPlugin(opts ...Option) *Controller
  • func (c *Controller) AfterInit() error
  • func (c *Controller) Close() error
  • func (c *Controller) GetConfig(key string) proto.Message
  • func (c *Controller) Init() error
  • func (c *Controller) PushEvent(event api.Event) error
• type Deps
• type EventHandlingRecord
• type EventRecord
• type ExternalConfigSource
• type Option
  • func UseDeps(f func(*Deps)) Option
• type QueuedEvent
• type WithInternalData

Variables

var (
	// ErrClosedWatcher is returned when dbWatcher is used when it is already closed.
	ErrClosedWatcher = errors.New("dbWatcher was closed")
	// ErrResyncReqQueueFull is returned when queue for resync request is full.
	ErrResyncReqQueueFull = errors.New("queue with resync requests is full")
)

var (
	// ErrClosedController is returned when Controller is used when it is already closed.
	ErrClosedController = errors.New("controller was closed")
	// ErrEventQueueFull is returned when queue for events is full.
	ErrEventQueueFull = errors.New("queue with events is full")
	// ErrEventLoopIsAborting returned to an event producer via method Event.Done()
	// when event loop is aborting after a fatal error has occurred.
	ErrEventLoopIsAborting = errors.New("event loop is aborting after a fatal error")
)

var DefaultPlugin = *NewPlugin()

DefaultPlugin is a default instance of Controller.

Functions

func LoadKubeStateForResync

func LoadKubeStateForResync(broker keyval.ProtoBroker, log logging.Logger) (event *api.DBResync, values map[string]datasync.KeyVal, err error)

LoadKubeStateForResync loads Kubernetes state from given DB for resync. Loaded key-value pairs are returned both as a map and a resync event. Broker should not be prefixed.

func ResyncDatabase

func ResyncDatabase(broker keyval.ProtoBroker, kubeStateData api.KubeStateData) error

ResyncDatabase updates database content to reflect the given Kubernetes state data. External configuration is not supported yet. Broker should not be prefixed.

Types

type Config

type Config struct {
	// retry
	EnableRetry           bool          `json:"enableRetry"`
	DelayRetry            time.Duration `json:"delayRetry"`
	MaxRetryAttempts      int           `json:"maxRetryAttempts"`
	EnableExpBackoffRetry bool          `json:"enableExpBackoffRetry"`

	// startup resync
	DelayLocalResync      time.Duration `json:"delayLocalResync"`
	StartupResyncDeadline time.Duration `json:"startupResyncDeadline"`

	// healing
	EnablePeriodicHealing   bool          `json:"enablePeriodicHealing"`
	PeriodicHealingInterval time.Duration `json:"periodicHealingInterval"`
	DelayAfterErrorHealing  time.Duration `json:"delayAfterErrorHealing"`

	// remote DB status
	RemoteDBProbingInterval time.Duration `json:"remoteDBProbingInterval"`

	// event history
	RecordEventHistory            bool   `json:"recordEventHistory"`
	EventHistoryAgeLimit          uint32 `json:"eventHistoryAgeLimit"`
	PermanentlyRecordedInitPeriod uint32 `json:"permanentlyRecordedInitPeriod"`

	// verification mode
	EnableVerification bool `json:"enableVerification"`
}

Config holds the Controller configuration.

type Controller

type Controller struct {
	Deps
	// contains filtered or unexported fields
}

Controller implements single event loop for Contiv.

Events are represented by instances of the api.Event interface. A new event can be pushed into the loop for processing via the PushEvent method from the api.EventLoop interface, implemented by the Controller plugin.

For a plugin to become a handler for one or more events, it has to implement the api.EventHandler interface. The set of event handlers is passed to Controller via EventHandlers attribute from Deps. The order of event handlers in the array matters - if handler B depends on A, i.e. A has to handle *any* event before B does, then A should precede B in the array. Cyclic dependencies are not allowed. Events then flow through the event handlers either in the forward or reverse order, based on the event direction (api.UpdateEvent.Direction(), "Forward" for Resync) and the event processing stage:

• "Forward" event, Update/Resync stage: forward iteration
• "Reverse" event, Update stage: backward iteration
• "Forward" event, Revert stage: backward iteration
• "Reverse" event, Revert stage: forward iteration

For every event, the controller approaches a given handler first by checking if the handler is actually interested in the event using the method: api.EventHandler.HandlesEvent(). Then, based on the event Method (api.Event.Method), it calls either Resync or Update method of the handler. The handler may update its internal state but for Update/RevertOnFailure (api.UpdateEvent.TransactionType) events it also has to be prepared to revert the changes (but only for that iteration of the event loop).

The handler may return error from Update/Resync wrapped in either:

• api.FatalError to signal that the agent should be terminated (and restarted by k8s), or
• api.AbortEventError to signal that the processing of the event should not continue and a resync is needed.

Non-fatal, non-abort error signals the controller that something is wrong and a resync is needed, but if the transaction is of type BestEffort, then the current event processing will continue.

The handler is also provided with Update/Resync transaction for re-synchronizing or applying changes to VPP/Linux network configuration. Transactional errors are treated as non-fatal. If Update/RevertOnFailure event handling fails with non-fatal error, handlers that already reacted to the event will be asked in the reverse order to Revert any internal changes via method api.EventHandler.Revert().

Processing of a given event is finalized by calling the api.Event.Done(error) method. The method can be used for example to deliver the return value back to the sender of the event.

func NewPlugin

func NewPlugin(opts ...Option) *Controller

NewPlugin creates a new Plugin with the provided Options.

func (*Controller) AfterInit

func (c *Controller) AfterInit() error

AfterInit starts DB watcher and registers plugin with the status check.

func (*Controller) Close

func (c *Controller) Close() error

Close stops event loop and database watching.

func (*Controller) GetConfig

func (c *Controller) GetConfig(key string) proto.Message

GetConfig returns value for the given key in the controller's transaction. If data for the key is not part of the transaction stored value from internal config is returned.

func (*Controller) Init

func (c *Controller) Init() error

Init loads config file and starts the event loop.

func (*Controller) PushEvent

func (c *Controller) PushEvent(event api.Event) error

PushEvent adds the given event into the queue for processing.

type Deps

type Deps struct {
	infra.PluginDeps

	Scheduler    scheduler.KVScheduler
	StatusCheck  statuscheck.PluginStatusWriter
	ServiceLabel servicelabel.ReaderAPI
	HTTPHandlers rest.HTTPHandlers

	EventHandlers []api.EventHandler

	LocalDB  keyval.KvProtoPlugin
	RemoteDB keyval.KvProtoPlugin

	ExtSources []ExternalConfigSource
}

Deps lists dependencies of the Controller.

type EventHandlingRecord

type EventHandlingRecord struct {
	Handler  string
	Revert   bool
	Change   string // change description for update events
	Error    error  // nil if none
	ErrorStr string // string representation of the error (if any)
}

EventHandlingRecord is a record of an event being handled by a given handler.

type EventRecord

type EventRecord struct {
	SeqNum          uint64
	ProcessingStart time.Time
	ProcessingEnd   time.Time
	IsFollowUp      bool
	FollowUpTo      uint64
	Name            string
	Description     string
	Method          api.EventMethodType
	Handlers        []*EventHandlingRecord
	TxnError        error
	Txn             *scheduler.RecordedTxn
}

EventRecord is a record of a processed event, added into the history of events, available via REST interface.

type ExternalConfigSource

type ExternalConfigSource interface {
	// String identifies the external config source for the Controller.
	// Note: Plugins already implement Stringer.
	String() string

	// GetConfigSnapshot should return full configuration snapshot that is
	// required by the external source to be applied at the given moment.
	GetConfigSnapshot() (api.KeyValuePairs, error)
}

ExternalConfigSource defines API that a source of external configuration must implement.

type Option

type Option func(*Controller)

Option is a function that can be used in NewPlugin to customize Plugin.

func UseDeps

func UseDeps(f func(*Deps)) Option

UseDeps returns Option that can inject custom dependencies.

type QueuedEvent

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

QueuedEvent wraps event for the event queue.

type WithInternalData

type WithInternalData interface {
	// DescribeInternalData should describe the internal state of the event handler.
	DescribeInternalData() string
}

WithInternalData *can* be implemented by event handlers that have internal data. The controller will use the method during verification to check that the internal data remains in-sync.