kvdbsync

Documentation

Overview

Package kvdbsync defines a key-value data store client API for unified access among key-value datastore servers. The datasync API contains the Data Broker & KeyValProtoWatcher APIs, which are just facades in front of different key-value or SQL stores.

A key-value store is used as a transport channel between a remote client and an agent (server). It stores data/configuration for multiple agents (servers). Therefore, a client only needs to know the address of the key-value store but does not need to know the addresses of individual agents. The client can write data/configuration independently of the agent's (server's) lifecycle.

The Data KeyValProtoWatcher is used during regular operation to efficiently propagate data/configuration changes from the key-value store to the agents (servers). Upon receiving a data change event, the watcher makes an incremental update to its data. When data resynchronization (RESYNC) is triggered, then the Data Broker is used to read all particular keys & values from the key-value store. Reading all particular keys & values is more reliable but less efficient data synchronization method.

Index

• type ChangeWatchResp
  • func NewChangeWatchResp(delegate keyval.ProtoWatchResp, prevVal datasync.LazyValue) *ChangeWatchResp
  • func (ev *ChangeWatchResp) GetChangeType() datasync.PutDel
  • func (ev *ChangeWatchResp) GetKey() string
  • func (ev *ChangeWatchResp) GetPrevValue(prevVal proto.Message) (exists bool, err error)
  • func (ev *ChangeWatchResp) GetRevision() (rev int64)
  • func (ev *ChangeWatchResp) GetValue(val proto.Message) (err error)
• type Deps
• type Iterator
  • func NewIterator(delegate keyval.ProtoKeyValIterator) *Iterator
  • func (it *Iterator) GetNext() (kv datasync.KeyVal, stop bool)
• type Plugin
  • func (plugin *Plugin) AfterInit() error
  • func (plugin *Plugin) Close() error
  • func (plugin *Plugin) Delete(key string, opts ...datasync.DelOption) (existed bool, err error)
  • func (plugin *Plugin) Init() error
  • func (plugin Plugin) OfDifferentAgent(microserviceLabel string, infraDeps infraDeps) *Plugin
  • func (plugin *Plugin) Put(key string, data proto.Message, opts ...datasync.PutOption) error
  • func (plugin *Plugin) String() string
  • func (plugin *Plugin) Watch(resyncName string, changeChan chan datasync.ChangeEvent, ...) (datasync.WatchRegistration, error)

Types

type ChangeWatchResp

type ChangeWatchResp struct {
	*syncbase.DoneChannel
	// contains filtered or unexported fields
}

ChangeWatchResp is a structure that adapts the BytesWatchResp to the datasync api.

func NewChangeWatchResp

func NewChangeWatchResp(delegate keyval.ProtoWatchResp, prevVal datasync.LazyValue) *ChangeWatchResp

NewChangeWatchResp creates a new instance of ChangeWatchResp.

func (*ChangeWatchResp) GetChangeType

func (ev *ChangeWatchResp) GetChangeType() datasync.PutDel

GetChangeType returns the type of the associated change.

func (*ChangeWatchResp) GetKey

func (ev *ChangeWatchResp) GetKey() string

GetKey returns the key associated with the change.

func (*ChangeWatchResp) GetPrevValue

func (ev *ChangeWatchResp) GetPrevValue(prevVal proto.Message) (exists bool, err error)

GetPrevValue delegates to WatchResp. For description of parameter and output values see the comment in implemented interface datasync.ChangeEvent

func (*ChangeWatchResp) GetRevision

func (ev *ChangeWatchResp) GetRevision() (rev int64)

GetRevision returns revision associated with the change.

func (*ChangeWatchResp) GetValue

func (ev *ChangeWatchResp) GetValue(val proto.Message) (err error)

GetValue delegates to WatchResp. For description of parameter and output values see the comment in implemented interface datasync.ChangeEvent

type Deps

type Deps struct {
	local.PluginInfraDeps                      // inject
	ResyncOrch            resync.Subscriber    // inject
	KvPlugin              keyval.KvProtoPlugin // inject
}

Deps groups dependencies injected into the plugin so that they are logically separated from other plugin fields.

type Iterator

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

Iterator adapts the db_proto.KeyValIterator to the datasync.KeyValIterator

func NewIterator

func NewIterator(delegate keyval.ProtoKeyValIterator) *Iterator

NewIterator creates a new instance of Iterator.

func (*Iterator) GetNext

func (it *Iterator) GetNext() (kv datasync.KeyVal, stop bool)

GetNext just delegate GetNext

type Plugin

type Plugin struct {
	Deps // inject
	// contains filtered or unexported fields
}

Plugin dbsync implements synchronization between local memory and db. Other plugins can be notified when DB changes occur or resync is needed. This plugin reads/pulls the data from db when resync is needed.

func (*Plugin) AfterInit

func (plugin *Plugin) AfterInit() error

AfterInit uses provided connection to build new transport watcher.

Plugin.registry subscriptions (registered by Watch method) are used for resync. Resync is called only if ResyncOrch was injected (is not nil). The order of plugins in flavor is not important to resync since Watch() is called in Plugin.Init() but Resync.Register() is called in Plugin.AfterInit().

func (*Plugin) Close

func (plugin *Plugin) Close() error

Close resources

func (*Plugin) Delete

func (plugin *Plugin) Delete(key string, opts ...datasync.DelOption) (existed bool, err error)

Delete propagates this call to a particular kvdb.Plugin unless the kvdb.Plugin is Disabled().

This method is supposed to be called in Plugin.AfterInit() or later (even from different go routine).

func (*Plugin) Init

func (plugin *Plugin) Init() error

Init just initializes plugin.registry.

func (Plugin) OfDifferentAgent

func (plugin Plugin) OfDifferentAgent(
	microserviceLabel string, infraDeps infraDeps) *Plugin

OfDifferentAgent allows access DB of different agent (with a particular microservice label). This method is a shortcut to simplify creating new instance of plugin that is supposed to watch different agent DB. Method intentionally copies instance of plugin (assuming it has set all dependencies) and sets microservice label.

func (*Plugin) Put

func (plugin *Plugin) Put(key string, data proto.Message, opts ...datasync.PutOption) error

Put propagates this call to a particular kvdb.Plugin unless the kvdb.Plugin is Disabled().

This method is supposed to be called in Plugin.AfterInit() or later (even from different go routine).

func (*Plugin) String

func (plugin *Plugin) String() string

String returns if set Deps.PluginName or "kvdbsync" otherwise

func (*Plugin) Watch

func (plugin *Plugin) Watch(resyncName string, changeChan chan datasync.ChangeEvent,
	resyncChan chan datasync.ResyncEvent, keyPrefixes ...string) (datasync.WatchRegistration, error)

Watch adds entry to the plugin.registry. By doing this other plugins will receive notifications about data changes and data resynchronization.

This method is supposed to be called in Plugin.Init(). Calling this method later than kvdbsync.Plugin.AfterInit() will have no effect (no notifications will be received).