kubeedge/cloud/pkg/dynamiccontroller/application/eventhandler.go

/*
Copyright 2021 The KubeEdge Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package application

import (
	"fmt"
	"sync"

	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/dynamic/dynamicinformer"
	"k8s.io/client-go/tools/cache"
	"k8s.io/klog/v2"

	genericinformers "github.com/kubeedge/kubeedge/cloud/pkg/common/informers"
	"github.com/kubeedge/kubeedge/cloud/pkg/common/messagelayer"
	"github.com/kubeedge/kubeedge/pkg/metaserver/util"
)

// HandlerCenter is used to prepare corresponding CommonResourceEventHandler for listener
// CommonResourceEventHandler then will send to the listener the objs it is interested in,
// including subsequent changes (watch event)
type HandlerCenter interface {
	AddListener(s *SelectorListener) error
	DeleteListener(s *SelectorListener)
	ForResource(gvr schema.GroupVersionResource) *CommonResourceEventHandler
	GetListenersForNode(nodeName string) map[string]*SelectorListener
}

type handlerCenter struct {
	// handlerLock used to protect the handlers map
	handlerLock                  sync.Mutex
	listenerManager              *listenerManager
	handlers                     map[schema.GroupVersionResource]*CommonResourceEventHandler
	dynamicSharedInformerFactory dynamicinformer.DynamicSharedInformerFactory
	messageLayer                 messagelayer.MessageLayer
}

func NewHandlerCenter(informerFactory dynamicinformer.DynamicSharedInformerFactory) HandlerCenter {
	handlers := make(map[schema.GroupVersionResource]*CommonResourceEventHandler)

	c := handlerCenter{
		listenerManager:              newListenerManager(),
		handlers:                     handlers,
		dynamicSharedInformerFactory: informerFactory,
		messageLayer:                 messagelayer.DynamicControllerMessageLayer(),
	}
	return &c
}

func (c *handlerCenter) ForResource(gvr schema.GroupVersionResource) *CommonResourceEventHandler {
	c.handlerLock.Lock()
	defer c.handlerLock.Unlock()

	if handler, ok := c.handlers[gvr]; ok {
		return handler
	}

	klog.Infof("[metaserver/HandlerCenter] prepare a new resourceEventHandler(%v)", gvr)

	handler := NewCommonResourceEventHandler(gvr, c.listenerManager, c.messageLayer)
	c.handlers[gvr] = handler

	return handler
}

// AddListener dispatch listeners to corresponding CommonResourceEventHandler according it's gvr
func (c *handlerCenter) AddListener(s *SelectorListener) error {
	return c.ForResource(s.gvr).AddListener(s)
}

func (c *handlerCenter) DeleteListener(s *SelectorListener) {
	c.handlerLock.Lock()
	c.handlers[s.gvr].DeleteListener(s)
	c.handlerLock.Unlock()
}

func (c *handlerCenter) GetListenersForNode(nodeName string) map[string]*SelectorListener {
	return c.listenerManager.GetListenersForNode(nodeName)
}

// CommonResourceEventHandler can be used by configmapManager and podManager
type CommonResourceEventHandler struct {
	events chan watch.Event

	//TODO: num of listeners is proportional to the number of request, need reduce.
	listenerManager *listenerManager
	messageLayer    messagelayer.MessageLayer
	gvr             schema.GroupVersionResource
	informer        *genericinformers.InformerPair
}

func NewCommonResourceEventHandler(
	gvr schema.GroupVersionResource,
	listenerManager *listenerManager,
	layer messagelayer.MessageLayer) *CommonResourceEventHandler {
	handler := &CommonResourceEventHandler{
		listenerManager: listenerManager,
		events:          make(chan watch.Event, 100),
		messageLayer:    layer,
		gvr:             gvr,
	}

	klog.Infof("[metaserver/resourceEventHandler] handler(%v) init, prepare informer...", gvr)
	informerPair, err := genericinformers.GetInformersManager().GetInformerPair(gvr)
	if err != nil {
		klog.Exitf("get informer for %s err: %v", gvr.String(), err)
	}

	_, err = informerPair.Informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			handler.objToEvent(watch.Added, obj)
		},
		UpdateFunc: func(oldObj, obj interface{}) {
			handler.objToEvent(watch.Modified, obj)
		},
		DeleteFunc: func(obj interface{}) {
			handler.objToEvent(watch.Deleted, obj)
		},
	})
	if err != nil {
		klog.Exitf("add evenet handler err: %v", err)
	}

	handler.informer = informerPair
	klog.Infof("[metaserver/resourceEventHandler] handler(%v) init successfully, start to dispatch events to it's listeners", gvr)
	go handler.dispatchEvents()
	return handler
}

func (c *CommonResourceEventHandler) objToEvent(t watch.EventType, obj interface{}) {
	eventObj, ok := obj.(runtime.Object)
	if !ok {
		klog.Warningf("Unknown type: %T, ignore", obj)
		return
	}
	// All obj from client has been removed the information of apiversion/kind called MetaType,
	// which is fatal to decode the obj as unstructured.Unstructure or unstructured.UnstructureList at edge.
	err := util.SetMetaType(eventObj)
	if err != nil {
		klog.Warningf("Failed to set metatype :%v", err)
	}
	c.events <- watch.Event{Type: t, Object: eventObj}
}

func (c *CommonResourceEventHandler) AddListener(s *SelectorListener) error {
	// filter s.selector.field when sendAllObjects
	ret, err := c.informer.Lister.List(s.selector.Label)
	if err != nil {
		return fmt.Errorf("Failed to list: %v", err)
	}
	s.sendAllObjects(ret, c)

	c.listenerManager.AddListener(s)

	return nil
}

func (c *CommonResourceEventHandler) DeleteListener(s *SelectorListener) {
	c.listenerManager.DeleteListener(s)
}

func (c *CommonResourceEventHandler) dispatchEvents() {
	for event := range c.events {
		klog.V(4).Infof("[metaserver/resourceEventHandler] handler(%v), send obj event{%v/%v} to listeners", c.gvr, event.Type, event.Object.GetObjectKind().GroupVersionKind().String())
		for _, listener := range c.listenerManager.GetListenersForGVR(c.gvr) {
			listener.sendObj(event, c.messageLayer)
		}
	}
	klog.Warningf("[metaserver/resourceEventHandler] handler(%v) stopped!", c.gvr.String())
}