kubernetes/pkg/kubelet/pleg/evented.go

const globalCacheUpdatePeriod …

var eventedPLEGUsage …

var eventedPLEGUsageMu …

// isEventedPLEGInUse indicates whether Evented PLEG is in use. Even after enabling
// the Evented PLEG feature gate, there could be several reasons it may not be in use.
// e.g. Streaming data issues from the runtime or the runtime does not implement the
// container events stream.
func isEventedPLEGInUse() bool { … }

// setEventedPLEGUsage should only be accessed from
// Start/Stop of Evented PLEG.
func setEventedPLEGUsage(enable bool) { … }

type EventedPLEG …

// NewEventedPLEG instantiates a new EventedPLEG object and return it.
func NewEventedPLEG(logger klog.Logger, runtime kubecontainer.Runtime, runtimeService internalapi.RuntimeService, eventChannel chan *PodLifecycleEvent,
	cache kubecontainer.Cache, genericPleg PodLifecycleEventGenerator, eventedPlegMaxStreamRetries int,
	relistDuration *RelistDuration, clock clock.Clock) (PodLifecycleEventGenerator, error) { … }

// Watch returns a channel from which the subscriber can receive PodLifecycleEvent events.
func (e *EventedPLEG) Watch() chan *PodLifecycleEvent { … }

// Relist relists all containers using GenericPLEG
func (e *EventedPLEG) Relist() { … }

// Start starts the Evented PLEG
func (e *EventedPLEG) Start() { … }

// Stop stops the Evented PLEG
func (e *EventedPLEG) Stop() { … }

// In case the Evented PLEG experiences undetectable issues in the underlying
// GRPC connection there is a remote chance the pod might get stuck in a
// given state while it has progressed in its life cycle. This function will be
// called periodically to update the global timestamp of the cache so that those
// pods stuck at GetNewerThan in pod workers will get unstuck.
func (e *EventedPLEG) updateGlobalCache() { … }

// Update the relisting period and threshold
func (e *EventedPLEG) Update(relistDuration *RelistDuration) { … }

// Healthy check if PLEG work properly.
func (e *EventedPLEG) Healthy() (bool, error) { … }

func (e *EventedPLEG) watchEventsChannel() { … }

func (e *EventedPLEG) processCRIEvents(containerEventsResponseCh chan *runtimeapi.ContainerEventResponse) { … }

func (e *EventedPLEG) processCRIEvent(event *runtimeapi.ContainerEventResponse) { … }

func (e *EventedPLEG) getPodIPs(pid types.UID, status *kubecontainer.PodStatus) []string { … }

func (e *EventedPLEG) sendPodLifecycleEvent(event *PodLifecycleEvent) { … }

func getPodSandboxState(podStatus *kubecontainer.PodStatus) kubecontainer.State { … }

func (e *EventedPLEG) updateRunningPodMetric(podStatus *kubecontainer.PodStatus) { … }

func getContainerStateCount(podStatus *kubecontainer.PodStatus) map[kubecontainer.State]int { … }

func (e *EventedPLEG) updateRunningContainerMetric(podStatus *kubecontainer.PodStatus) { … }

func (e *EventedPLEG) updateLatencyMetric(event *runtimeapi.ContainerEventResponse) { … }

func (e *EventedPLEG) UpdateCache(pod *kubecontainer.Pod, pid types.UID) (error, bool) { … }