func LoadKubeStateForResync(broker keyval.ProtoBroker, log logging.Logger) (event *api.DBResync, values map[string]datasync.KeyVal, err error) {
	event = api.NewDBResync()
	values = make(map[string]datasync.KeyVal)
	for _, resource := range dbresources.GetDBResources() {
		iterator, err := broker.ListValues(ksrPrefix + resource.KeyPrefix)
		if err != nil {
			return event, values, err
		}
		for {
			kv, stop := iterator.GetNext()
			if stop {
				break
			}
			valueType := proto.MessageType(resource.ProtoMessageName)
			if valueType == nil {
				log.Warnf("Failed to instantiate proto message for resource: %s", resource.Keyword)
				continue
			}
			value := reflect.New(valueType.Elem()).Interface().(proto.Message)
			err := kv.GetValue(value)
			if err != nil {
				log.Warnf("Failed to de-serialize value for key: %s", kv.GetKey())
				continue
			}
			if values != nil {
				values[kv.GetKey()] = kv
			}
			if event != nil {
				key := strings.TrimPrefix(kv.GetKey(), ksrPrefix)
				event.KubeState[resource.Keyword][key] = value
			}
		}
		iterator.Close()
	}
	return event, values, nil
}
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.
package controller
import (
	"context"
	"errors"
	"reflect"
	"strings"
	"sync"
	"time"
	"github.com/golang/protobuf/proto"
	"go.ligato.io/cn-infra/v2/datasync"
	"go.ligato.io/cn-infra/v2/db/keyval"
	"go.ligato.io/cn-infra/v2/logging"
	"go.ligato.io/cn-infra/v2/servicelabel"
	"go.ligato.io/vpp-agent/v3/plugins/orchestrator"
	"github.com/contiv/vpp/dbresources"
	"github.com/contiv/vpp/plugins/controller/api"
	"github.com/contiv/vpp/plugins/ksr"
)
const (
	dbExtCfgSrc = "db"
	healthCheckProbeKey = "/probe-etcd-connection"
)
var (
	ksrPrefix = servicelabel.GetDifferentAgentPrefix(ksr.MicroserviceLabel)
)
type dbWatcher struct {
	sync.Mutex
	*dbWatcherArgs
	remoteIsConnected bool
	resyncCount       int
	resyncReqs        chan bool  
	ignoreChangesUntilResync bool
	remoteBroker  keyval.ProtoBroker
	remoteWatcher keyval.ProtoWatcher
	localBroker   keyval.ProtoBroker
	remoteChangeCh     chan datasync.ProtoWatchResp
	remoteWatchCloseCh chan string
	processedVals map[string]datasync.KeyVal  
	wg     sync.WaitGroup
	ctx    context.Context
	cancel context.CancelFunc
}
type dbWatcherArgs struct {
	log         logging.Logger
	agentPrefix string
	eventLoop api.EventLoop
	localDB  keyval.KvProtoPlugin
	remoteDB keyval.KvProtoPlugin
	delayLocalResync        time.Duration
	remoteDBProbingInterval time.Duration
}
var (
	ErrClosedWatcher = errors.New("dbWatcher was closed")
	ErrResyncReqQueueFull = errors.New("queue with resync requests is full")
)
func newDBWatcher(args *dbWatcherArgs) *dbWatcher {
	watcher := &dbWatcher{
		dbWatcherArgs:  args,
		resyncReqs:     make(chan bool, 10),
		localBroker:    args.localDB.NewBroker(""),
		remoteChangeCh: make(chan datasync.ProtoWatchResp, 1000),
		processedVals:  make(map[string]datasync.KeyVal),
	}
	watcher.ignoreChangesUntilResync = true
	watcher.ctx, watcher.cancel = context.WithCancel(context.Background())
	watcher.wg.Add(1)
	go watcher.watchDB()
	args.remoteDB.OnConnect(watcher.onFirstConnect)
	watcher.wg.Add(1)
	go watcher.scheduleLocalResync(args.delayLocalResync)
	return watcher
}
func (w *dbWatcher) scheduleLocalResync(delay time.Duration) {
	defer w.wg.Done()
	select {
	case <-w.ctx.Done():
		return
	case <-time.After(delay):
		err := w.requestResync(true)
		if err != nil {
			w.log.Errorf("Failed to request resync against local DB: %v", err)
		}
	}
}
func (w *dbWatcher) onFirstConnect() error {
	w.Lock()
	defer w.Unlock()
	w.remoteBroker = w.remoteDB.NewBroker("")
	w.remoteWatcher = w.remoteDB.NewWatcher("")
	w.wg.Add(1)
	go w.periodicRemoteDBProbing()
	return nil
}
func (w *dbWatcher) periodicRemoteDBProbing() {
	defer w.wg.Done()
	w.probeRemoteDB()
	for {
		select {
		case <-time.After(w.remoteDBProbingInterval):
			w.probeRemoteDB()
		case <-w.ctx.Done():
			return
		}
	}
}
func (w *dbWatcher) probeRemoteDB() {
	w.Lock()
	defer w.Unlock()
	if _, _, err := w.remoteBroker.GetValue(healthCheckProbeKey, nil); err != nil {
		if w.remoteIsConnected == true {
			w.remoteIsConnected = false
			w.stopWatching()
			w.log.Warn("Lost connection to Remote DB")
		}
		return
	}
	if !w.remoteIsConnected {
		w.remoteIsConnected = true
		w.log.Info("Connection to Remote DB was (re-)established")
		w.ignoreChangesUntilResync = true
		w.restartWatching()
		err := w.requestResync(false)
		if err != nil {
			w.log.Errorf("Failed to request resync against remote DB: %v", err)
		}
	}
}
func (w *dbWatcher) requestResync(local bool) error {
	select {
	case <-w.ctx.Done():
		return ErrClosedWatcher
	case w.resyncReqs <- local:
		return nil
	default:
		return ErrResyncReqQueueFull
	}
}
func (w *dbWatcher) watchDB() {
	defer w.wg.Done()
	for {
		select {
		case <-w.ctx.Done():
			return
		case local := <-w.resyncReqs:
			w.runResync(local)
			continue
		case change := <-w.remoteChangeCh:
			w.processChange(change)
			continue
		}
	}
}
func (w *dbWatcher) restartWatching() {
	w.remoteWatchCloseCh = make(chan string)
	w.remoteWatcher.Watch(w.onRemoteDBChange, w.remoteWatchCloseCh,
		servicelabel.GetDifferentAgentPrefix(ksr.MicroserviceLabel),  
		w.agentPrefix)  
}
func (w *dbWatcher) stopWatching() {
	if w.remoteWatchCloseCh != nil {
		close(w.remoteWatchCloseCh)
		w.remoteWatchCloseCh = nil
	}
}
func (w *dbWatcher) onRemoteDBChange(change datasync.ProtoWatchResp) {
	select {
	case w.remoteChangeCh <- change:
		return
	default:
		w.log.Error("Failed to enqueue remote DB data change, requesting resync")
		if err := w.requestResync(false); err != nil {
			w.log.Error("Even queue for resync requests is full")
		}
	}
}
func (w *dbWatcher) runResync(local bool) {
	w.Lock()
	defer w.Unlock()
	if local && (w.resyncCount > 0 || w.remoteIsConnected) {
		w.log.Info("Skipping resync against local DB")
		return
	}
	if !local && !w.remoteIsConnected {
		w.log.Info("Unable to resync against remote DB - connection is not available")
		return
	}
	w.resyncCount++
	if local {
		w.runResyncFromLocalDB()
	} else {
		w.runResyncFromRemoteDB()
	}
}
func (w *dbWatcher) runResyncFromLocalDB() {
	event, values, err := LoadKubeStateForResync(w.localBroker, w.log)
	if err != nil {
		w.log.Errorf("Resync from local DB has failed: %v", err)
		return
	}
	if len(values) == 0 {
		w.log.Error("Local DB is empty - aborting resync")
		return
	}
	event.Local = true
	err = w.eventLoop.PushEvent(event)
	if err != nil {
		w.log.Errorf("Failed to push (local) resync event: %v", err)
	}
}
func (w *dbWatcher) runResyncFromRemoteDB() {
	var err error
	defer func() {
		if err != nil {
			w.log.Errorf("Resync from remote DB has failed: %v, requesting another resync", err)
			if err := w.requestResync(false); err != nil {
				w.log.Errorf("Even queue for resync requests is broken: %v", err)
			}
		} else {
			w.ignoreChangesUntilResync = false
		}
	}()
	var (
		event  *api.DBResync
		values map[string]datasync.KeyVal
	)
	event, values, err = LoadKubeStateForResync(w.remoteBroker, w.log)
	if err != nil {
		return
	}
	var iterator keyval.ProtoKeyValIterator
	iterator, err = w.remoteBroker.ListValues(w.agentPrefix)
	if err != nil {
		return
	}
	for {
		kv, stop := iterator.GetNext()
		if stop {
			break
		}
		values[kv.GetKey()] = kv
		key := strings.TrimPrefix(kv.GetKey(), w.agentPrefix)
		protoVal, err := orchestrator.UnmarshalLazyValue(key, kv)
		if err != nil {
			w.log.Warnf("Attempt to unmarshall value for key=%s returned error: %v",
				kv.GetKey(), err)
			continue
		}
		event.ExternalConfig[key] = protoVal
	}
	iterator.Close()
	err = ResyncDatabase(w.localBroker, event.KubeState)
	if err != nil {
		return
	}
	err = w.eventLoop.PushEvent(event)
	if err != nil {
		return
	}
	w.processedVals = values
}
func (w *dbWatcher) processChange(change datasync.ProtoWatchResp) {
	w.Lock()
	defer w.Unlock()
	key := change.GetKey()
	if w.ignoreChangesUntilResync {
		w.log.Debugf("Ignoring change for key: %v (waiting for resync)", key)
		return
	}
	prevRev, hasPrevRev := w.processedVals[key]
	if hasPrevRev {
		if prevRev.GetRevision() >= change.GetRevision() {
			w.log.Debugf("Ignoring already processed revision for key=%s", key)
			return
		}
	}
	if change.GetChangeType() == datasync.Delete {
		delete(w.processedVals, key)
	} else {
		w.processedVals[key] = change
	}
	resourceMeta, externalCfg := w.getResourceByKey(key)
	if resourceMeta == nil && !externalCfg {
		return
	}
	var (
		err             error
		newVal, prevVal proto.Message
	)
	if resourceMeta != nil {
		valueType := proto.MessageType(resourceMeta.ProtoMessageName)
		if valueType == nil {
			w.log.Warnf("Failed to instantiate proto message for resource: %s", resourceMeta.Keyword)
		} else {
			if change.GetChangeType() != datasync.Delete {
				newVal = reflect.New(valueType.Elem()).Interface().(proto.Message)
			}
			prevVal = reflect.New(valueType.Elem()).Interface().(proto.Message)
			if change.GetChangeType() != datasync.Delete {
				err := change.GetValue(newVal)
				if err != nil {
					w.log.Warnf("Failed to de-serialize new value for key=%s: %v", key, err)
					newVal = nil
				}
			}
			if hasPrevRev {
				err = prevRev.GetValue(prevVal)
			}
			if err != nil {
				w.log.Warnf("Failed to de-serialize previous value for key=%s: %v", key, err)
			}
			if !hasPrevRev || err != nil {
				prevVal = nil
			}
			if newVal == nil && prevVal == nil {
				w.log.Debugf("Ignoring already processed Delete for key=%s", key)
				return
			}
		}
	} else {
		keySuffix := strings.TrimPrefix(key, w.agentPrefix)
		if change.GetChangeType() != datasync.Delete {
			newVal, err = orchestrator.UnmarshalLazyValue(keySuffix, change)
			if err != nil {
				w.log.Warnf("Failed to de-serialize new value for key: %s", key)
			}
		}
	}
	if resourceMeta != nil {
		if change.GetChangeType() == datasync.Delete {
			w.localBroker.Delete(key)
		} else if newVal != nil {
			w.localBroker.Put(key, newVal)
		}
	}
	var event api.Event
	if resourceMeta != nil {
		key = strings.TrimPrefix(key, servicelabel.GetDifferentAgentPrefix(ksr.MicroserviceLabel))
		event = &api.KubeStateChange{
			Key:       key,
			Resource:  resourceMeta.Keyword,
			PrevValue: prevVal,
			NewValue:  newVal,
		}
	} else {
		key = strings.TrimPrefix(key, w.agentPrefix)
		extChangeEv := api.NewExternalConfigChange(dbExtCfgSrc, false)
		extChangeEv.UpdatedKVs[key] = newVal
		event = extChangeEv
	}
	err = w.eventLoop.PushEvent(event)
	if err != nil {
		w.log.Errorf("Failed to push data change event: %v, requesting resync", err)
		if err := w.requestResync(false); err != nil {
			w.log.Errorf("Even queue for resync requests is broken: %v", err)
		}
	}
}
func (w *dbWatcher) getResourceByKey(key string) (resource *dbresources.DBResource, externalConfig bool) {
	if strings.HasPrefix(key, w.agentPrefix) {
		return nil, true
	}
	key = strings.TrimPrefix(key, servicelabel.GetDifferentAgentPrefix(ksr.MicroserviceLabel))
	for _, resource := range dbresources.GetDBResources() {
		if strings.HasPrefix(key, resource.KeyPrefix) {
			return resource, false
		}
	}
	return nil, false  
}
func (w *dbWatcher) close() {
	w.cancel()
	w.wg.Wait()
	w.stopWatching()
}