datasync

README

Concept

Package datasync defines the interfaces for the abstraction datasync_api.go of a data synchronization between app plugins and different backend data sources (such as data stores, message buses, or RPC-connected clients).

In this context, data synchronization is about multiple data sets that need to be synchronized whenever a particular event is published. The event can be published by:

• database (when particular data was changed);
• by message bus (such as consuming messages from Kafka topics);
• or by RPC clients (when GRPC or REST service call ).

The data synchronization APIs are centered around watching and publishing data change events. These events are processed asynchronously.

The data handled by one plugin can have references to the data of another plugin. Therefore, proper time/order of data resynchronization between plugins needs to be maintained. The datasync plugin initiates full data resync in the same order as the other plugins have been registered in Init().

Watch data API

Watch data API is used by app plugin (see following diagram and the example) to:

1. Subscribe channels for particular data changes Watch() while being abstracted from a particular message source (data store, message bus or RPC)
2. Process Full Data RESYNC (startup, for certain fault recovery) event reprocess whole data set. Feedback is given to the user of this API (e.g. successful configuration or an error) by callback.
3. Process Incremental Data CHANGE. It is Optimized mode that works only with a minimal set of changes (deltas). Again, feedback to the API's user (e.g. successful configuration or an error) is given by callback.

This APIs defines two types of events that a plugin must be able to process:

1. Full Data RESYNC (resynchronization) event is defined to trigger resynchronization of the whole configuration. This event is used after agent start/restart, or for fault recovery (when agent's connectivity to an external data source is lost and restored).
2. Incremental Data CHANGE event is defined to trigger incremental processing of configuration changes. Data change events are sent after the data resync has completed. Each data change event contains both the previous and the new/current values for the data. The Data synchronization is switched to optimized mode after successful Full Data RESYNC.

Publish data API

Publish data API is used by app plugins to asynchronously publish events with particular data change values and still abstract from data store, message bus, local/RPC client.

Documentation

Overview

Package datasync defines the datasync API, which abstracts the data transport between app plugins and backend data sources. Data sources can be data stores, clients connected to a message bus, or remote clients connected to a CN-Infra app. Transport can be, for example, HTTP or gRPC.

These events are processed asynchronously. The app plugin that watches data changes gives callback for each event (e.g. successful configuration or an error).

See the examples under the dedicated examples package.

Index

• Constants
• type AggregatedRegistration
  • func (wa *AggregatedRegistration) Close() error
• type CallbackResult
• type ChangeEvent
• type ChangeValue
• type CompositeKVProtoWatcher
  • func (ta *CompositeKVProtoWatcher) Watch(resyncName string, changeChan chan ChangeEvent, resyncChan chan ResyncEvent, ...) (WatchRegistration, error)
• type CompositeKVProtoWriter
  • func (ta *CompositeKVProtoWriter) Put(key string, data proto.Message) error
• type DelOption
• type DelOptionMarker
  • func (marker *DelOptionMarker) DelOptionMark()
• type KeyProtoValWriter
• type KeyVal
• type KeyValIterator
• type KeyValProtoWatcher
• type LazyValue
• type LazyValueWithRev
• type ProtoWatchResp
• type PutDel
• type PutOption
• type PutOptionMarker
  • func (marker *PutOptionMarker) PutOptionMark()
• type ResyncEvent
• type WatchRegistration
• type WithChangeType
• type WithKey
• type WithPrefixOpt
  • func WithPrefix() *WithPrefixOpt
• type WithPrevValue
• type WithRevision
• type WithTTLOpt
  • func WithTTL(TTL time.Duration) *WithTTLOpt

Constants

const DefaultNotifTimeout = 2 * time.Second

DefaultNotifTimeout for delivery of notification

Types

type AggregatedRegistration

type AggregatedRegistration struct {
	Registrations []WatchRegistration
}

AggregatedRegistration is cumulative adapter which contains all available transport types

func (*AggregatedRegistration) Close

func (wa *AggregatedRegistration) Close() error

Close every registration under watch aggregator

type CallbackResult

type CallbackResult interface {
	// Done allows plugins that are processing data change/resync to send feedback
	// If there was no error the Done(nil) needs to be called. Use the noError=nil
	// definition for better readability, for example:
	//     Done(noError).
	Done(error)
}

CallbackResult can be used by an event receiver to indicate to the event producer whether an operation was successful (error is nil) or unsuccessful (error is not nil)

DoneMethod is reused later. There are at least two implementations DoneChannel, DoneCallback

type ChangeEvent

type ChangeEvent interface {
	CallbackResult

	ProtoWatchResp
}

ChangeEvent is used as the data type for the change channel (see the VPP Standard Plugins API). A data change event contains a key identifying where the change happened and two values for data stored under that key: the value *before* the change (previous value) and the value *after* the change (current value).

type ChangeValue

type ChangeValue interface {
	LazyValueWithRev
	WithChangeType
}

ChangeValue represents single propagated change.

type CompositeKVProtoWatcher

type CompositeKVProtoWatcher struct {
	Adapters []KeyValProtoWatcher
}

CompositeKVProtoWatcher is a slice of watchers

func (*CompositeKVProtoWatcher) Watch

func (ta *CompositeKVProtoWatcher) Watch(resyncName string, changeChan chan ChangeEvent, resyncChan chan ResyncEvent,
	keyPrefixes ...string) (WatchRegistration, error)

Watch subscribes to every transport available within transport aggregator

type CompositeKVProtoWriter

type CompositeKVProtoWriter struct {
	Adapters []KeyProtoValWriter
}

CompositeKVProtoWriter is cumulative adapter which contains all available transport types

func (*CompositeKVProtoWriter) Put

func (ta *CompositeKVProtoWriter) Put(key string, data proto.Message) error

Put to every available transport

type DelOption

type DelOption interface {
	//DelOptionMark is just for marking implementation that it implements this interface
	DelOptionMark()
}

DelOption defines options for Del operation. The particular options can be found below.

type DelOptionMarker

type DelOptionMarker struct{}

DelOptionMarker is meant for anonymous composition in With*Opt structs

func (*DelOptionMarker) DelOptionMark

func (marker *DelOptionMarker) DelOptionMark()

DelOptionMark is just for marking implementation that it implements this interface

type KeyProtoValWriter

type KeyProtoValWriter interface {
	// Put to ETCD or any other data transport (from other Agent Plugins)
	Put(key string, data proto.Message, opts ...PutOption) error
}

KeyProtoValWriter allows plugins to push their data changes to a data store.

type KeyVal

type KeyVal interface {
	WithKey
	LazyValueWithRev
}

KeyVal represents a single key-value pair

type KeyValIterator

type KeyValIterator interface {
	// GetNext retrieves the next value from the iterator context.  The retrieved
	// value is unmarshaled into the provided argument. The allReceived flag is
	// set to true on the last KeyVal pair in the context.
	GetNext() (kv KeyVal, allReceived bool)
}

KeyValIterator is an iterator for KeyVals

type KeyValProtoWatcher

type KeyValProtoWatcher interface {
	// Watch using ETCD or any other data transport
	Watch(resyncName string, changeChan chan ChangeEvent, resyncChan chan ResyncEvent,
		keyPrefixes ...string) (WatchRegistration, error)
}

KeyValProtoWatcher is used by plugin to subscribe to both data change events and data resync events. Multiple keys can be specified, the caller will be subscribed to events on each key. See README.md for description of the Events.

type LazyValue

type LazyValue interface {
	// GetValue gets the current in the data change event.
	// The caller must provide an address of a proto message buffer
	// for each value.
	// returns:
	// - revision associated with the latest change in the key-value pair
	// - error if value argument can not be properly filled
	GetValue(value proto.Message) error
}

LazyValue defines value that is unmarshalled into proto message on demand. The reason for defining interface with only one method is primary to unify interfaces in this package

type LazyValueWithRev

type LazyValueWithRev interface {
	LazyValue
	WithRevision
}

LazyValueWithRev defines value that is unmarshalled into proto message on demand with a revision. The reason for defining interface with only one method is primary to unify interfaces in this package

type ProtoWatchResp

type ProtoWatchResp interface {
	ChangeValue
	WithKey
	WithPrevValue
}

ProtoWatchResp contains changed value

type PutDel

type PutDel string

PutDel enumerates 'Put' (meaning 'Create' or 'Update') and 'Delete' operations

const (
	// Put represents a Create or Update operation
	Put PutDel = "Put"
	// Delete operation
	Delete PutDel = "Delete"
)

type PutOption

type PutOption interface {
	//PutOptionMark is just for marking implementation that it implements this interface
	PutOptionMark()
}

PutOption defines options for Put operation. The particular options can be found below.

type PutOptionMarker

type PutOptionMarker struct{}

PutOptionMarker is meant for anonymous composition in With*Opt structs

func (*PutOptionMarker) PutOptionMark

func (marker *PutOptionMarker) PutOptionMark()

PutOptionMark is just for marking implementation that it implements this interface

type ResyncEvent

type ResyncEvent interface {
	CallbackResult

	GetValues() map[string]KeyValIterator
}

ResyncEvent is used as the data type for the resync channel (see the ifplugin API)

type WatchRegistration

type WatchRegistration interface {
	io.Closer
}

WatchRegistration is a facade that avoids importing the io.Closer package into Agent plugin implementations.

type WithChangeType

type WithChangeType interface {
	GetChangeType() PutDel
}

WithChangeType is a helper interface which intent is to ensure that same method declaration is used in different interfaces (composition of interfaces)

type WithKey

type WithKey interface {
	// GetKey returns the key of the pair
	GetKey() string
}

WithKey is a helper interface which intent is to ensure that same method declaration is used in different interfaces (composition of interfaces)

type WithPrefixOpt

type WithPrefixOpt struct {
	DelOptionMarker
}

WithPrefixOpt applies an operation to all items with the specified prefix.

func WithPrefix

func WithPrefix() *WithPrefixOpt

WithPrefix creates new instance of WithPrefixOpt.

type WithPrevValue

type WithPrevValue interface {
	// GetPrevValue gets previous value in the data change event.
	// The caller must provide an address of a proto message buffer
	// for each value.
	// returns:
	// - prevValueExist flag is set to 'true' if prevValue was filled
	// - error if value argument can not be properly filled
	GetPrevValue(prevValue proto.Message) (prevValueExist bool, err error)
}

WithPrevValue is a helper interface which intent is to ensure that same method declaration is used in different interfaces (composition of interfaces)

type WithRevision

type WithRevision interface {
	// GetRevision gets revision of current value
	GetRevision() (rev int64)
}

WithRevision is a helper interface which intent is to ensure that same method declaration is used in different interfaces (composition of interfaces)

type WithTTLOpt

type WithTTLOpt struct {
	PutOptionMarker
	TTL time.Duration
}

WithTTLOpt defines a TTL for data being put. Once TTL elapses the data is removed from data store.

func WithTTL

func WithTTL(TTL time.Duration) *WithTTLOpt

WithTTL creates new instance of TTL option. Once TTL elapses data is removed. Beware: some implementation might be using TTL with lower precision.