func (c *Controller) PushEvent(event api.Event) error {
	callerGID := getGID()
	if callerGID == c.evLoopGID {
		if event.IsBlocking() {
			panic("deadlock detected - blocking event sent from within the event loop")
		}
		select {
		case <-c.ctx.Done():
			return ErrClosedController
		case c.followUpEventQueue <- &QueuedEvent{
			event:           event,
			isFollowUp:      true,
			followUpToEvent: c.evSeqNum - 1}:
			return nil
		default:
			return ErrEventQueueFull
		}
	}
	select {
	case <-c.ctx.Done():
		return ErrClosedController
	case c.eventQueue <- &QueuedEvent{event: event}:
		return nil
	default:
		return ErrEventQueueFull
	}
}
PushEvent adds the given event into the queue for processing.
package controller
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"runtime"
	"sync"
	"time"
	"github.com/golang/protobuf/proto"
	"go.ligato.io/cn-infra/v2/db/keyval"
	"go.ligato.io/cn-infra/v2/health/statuscheck"
	"go.ligato.io/cn-infra/v2/infra"
	"go.ligato.io/cn-infra/v2/rpc/rest"
	"go.ligato.io/cn-infra/v2/servicelabel"
	scheduler "go.ligato.io/vpp-agent/v3/plugins/kvscheduler/api"
	"github.com/contiv/vpp/plugins/controller/api"
)
const (
	eventQueueSize = 1000
	eventHistoryTrimmingPeriod = 1 * time.Minute
	defaultDelayLocalResync = 5 * time.Second
	defaultStartupResyncDeadline = 30 * time.Second
	defaultRemoteDBProbingInterval = 3 * time.Second
	defaultEnableRetry = true
	defaultDelayRetry = time.Second
	defaultMaxRetryAttempts = 3
	defaultEnableExpBackoffRetry = true
	defaultEnablePeriodicHealing = false
	defaultPeriodicHealingInterval = time.Minute
	defaultDelayAfterErrorHealing = 5 * time.Second
	defaultRecordEventHistory = true
	defaultEventHistoryAgeLimit = 24 * 60  
	defaultPermanentlyRecordedInitPeriod = 60  
	defaultEnableVerification = false
)
type WithInternalData interface {
	DescribeInternalData() string
}
type Controller struct {
	Deps
	config *Config
	dbWatcher      *dbWatcher
	kubeStateData  api.KubeStateData
	externalConfig map[string]api.KeyValuePairs  
	internalConfig api.KeyValuePairs
	evLoopGID            string  
	revEventHandlers     []api.EventHandler
	delayedEvents        []*QueuedEvent  
	eventQueue           chan *QueuedEvent
	followUpEventQueue   chan *QueuedEvent  
	startupResyncCheck   chan struct{}
	eventHistoryTrimming chan struct{}
	healingScheduled bool
	resyncCount      int
	aborting         bool
	evSeqNum         uint64
	historyLock  sync.Mutex
	eventHistory []*EventRecord
	startTime    time.Time
	wg     sync.WaitGroup
	ctx    context.Context
	cancel context.CancelFunc
	txn *kvSchedulerTxn  
}
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
}
type Config struct {
	EnableRetry           bool          `json:"enableRetry"`
	DelayRetry            time.Duration `json:"delayRetry"`
	MaxRetryAttempts      int           `json:"maxRetryAttempts"`
	EnableExpBackoffRetry bool          `json:"enableExpBackoffRetry"`
	DelayLocalResync      time.Duration `json:"delayLocalResync"`
	StartupResyncDeadline time.Duration `json:"startupResyncDeadline"`
	EnablePeriodicHealing   bool          `json:"enablePeriodicHealing"`
	PeriodicHealingInterval time.Duration `json:"periodicHealingInterval"`
	DelayAfterErrorHealing  time.Duration `json:"delayAfterErrorHealing"`
	RemoteDBProbingInterval time.Duration `json:"remoteDBProbingInterval"`
	RecordEventHistory            bool   `json:"recordEventHistory"`
	EventHistoryAgeLimit          uint32 `json:"eventHistoryAgeLimit"`
	PermanentlyRecordedInitPeriod uint32 `json:"permanentlyRecordedInitPeriod"`
	EnableVerification bool `json:"enableVerification"`
}
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
}
type EventHandlingRecord struct {
	Handler  string
	Revert   bool
	Change   string  
	Error    error   
	ErrorStr string  
}
type QueuedEvent struct {
	event           api.Event
	isFollowUp      bool
	followUpToEvent uint64  
}
type ExternalConfigSource interface {
	String() string
	GetConfigSnapshot() (api.KeyValuePairs, error)
}
var (
	ErrClosedController = errors.New("controller was closed")
	ErrEventQueueFull = errors.New("queue with events is full")
	ErrEventLoopIsAborting = errors.New("event loop is aborting after a fatal error")
)
func (c *Controller) Init() error {
	c.startTime = time.Now()
	c.ctx, c.cancel = context.WithCancel(context.Background())
	c.eventQueue = make(chan *QueuedEvent, eventQueueSize)
	c.followUpEventQueue = make(chan *QueuedEvent, eventQueueSize)
	c.startupResyncCheck = make(chan struct{}, 1)
	c.eventHistoryTrimming = make(chan struct{}, 1)
	c.internalConfig = make(api.KeyValuePairs)
	c.externalConfig = make(map[string]api.KeyValuePairs)
	for i := len(c.EventHandlers) - 1; i >= 0; i-- {
		c.revEventHandlers = append(c.revEventHandlers, c.EventHandlers[i])
	}
	c.config = &Config{
		DelayLocalResync:              defaultDelayLocalResync,
		StartupResyncDeadline:         defaultStartupResyncDeadline,
		RemoteDBProbingInterval:       defaultRemoteDBProbingInterval,
		EnableRetry:                   defaultEnableRetry,
		DelayRetry:                    defaultDelayRetry,
		MaxRetryAttempts:              defaultMaxRetryAttempts,
		EnableExpBackoffRetry:         defaultEnableExpBackoffRetry,
		EnablePeriodicHealing:         defaultEnablePeriodicHealing,
		PeriodicHealingInterval:       defaultPeriodicHealingInterval,
		DelayAfterErrorHealing:        defaultDelayAfterErrorHealing,
		RecordEventHistory:            defaultRecordEventHistory,
		EventHistoryAgeLimit:          defaultEventHistoryAgeLimit,
		PermanentlyRecordedInitPeriod: defaultPermanentlyRecordedInitPeriod,
		EnableVerification:            defaultEnableVerification,
	}
	err := c.loadConfig(c.config)
	if err != nil {
		c.Log.Error(err)
		return err
	}
	c.Log.Infof("Controller configuration: %+v", *c.config)
	if c.StatusCheck != nil {
		c.StatusCheck.Register(c.PluginName, nil)
	}
	c.wg.Add(1)
	go c.eventLoop()
	c.wg.Add(1)
	go c.signalStartupResyncCheck()
	if c.config.RecordEventHistory {
		c.wg.Add(1)
		go c.signalEventHistoryTrimming()
	}
	c.registerHandlers()
	return nil
}
func (c *Controller) AfterInit() error {
	c.dbWatcher = newDBWatcher(&dbWatcherArgs{
		log:                     c.Log.NewLogger("dbwatcher"),
		agentPrefix:             c.ServiceLabel.GetAgentPrefix(),
		eventLoop:               c,
		localDB:                 c.LocalDB,
		remoteDB:                c.RemoteDB,
		delayLocalResync:        c.config.DelayLocalResync,
		remoteDBProbingInterval: c.config.RemoteDBProbingInterval,
	})
	return nil
}