kubernetes-添加controller-demo-v1.5.2

上一次分析介绍了如何添加资源Application，现在就要在kube-controller-manager中添加一个controller对Appliaction进行一些操作。

添加application controller

在/pkg/controller目录下新建application目录，然后在application目录下添加application_controller.go：

/*
Copyright 2015 The Kubernetes 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 contains all the logic for handling Kubernetes Application.
package application

import (
	"time"
	"strings"

	"github.com/golang/glog"
	"k8s.io/kubernetes/pkg/api"
	"k8s.io/kubernetes/pkg/client/cache"
	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
	unversionedcore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
	"k8s.io/kubernetes/pkg/client/record"
	"k8s.io/kubernetes/pkg/apis/extensions"
	"k8s.io/kubernetes/pkg/controller"
	"k8s.io/kubernetes/pkg/controller/informers"
	"k8s.io/kubernetes/pkg/util/metrics"
	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
	"k8s.io/kubernetes/pkg/util/wait"
	"k8s.io/kubernetes/pkg/util/workqueue"
)

// ApplicationController is responsible for synchronizing Application objects.
type ApplicationController struct {
	client        clientset.Interface
	eventRecorder record.EventRecorder

	syncHandler func(dKey string) error

	// Applications that need to be synced
	queue workqueue.RateLimitingInterface
}

// NewApplicationController creates a new ApplicationController.
func NewApplicationController(aInformer informers.ApplicationInformer, client clientset.Interface) *ApplicationController {
	eventBroadcaster := record.NewBroadcaster()
	eventBroadcaster.StartLogging(glog.Infof)
	eventBroadcaster.StartRecordingToSink(&unversionedcore.EventSinkImpl{Interface: client.Core().Events("")})

	if client != nil && client.Core().RESTClient().GetRateLimiter() != nil {
		metrics.RegisterMetricAndTrackRateLimiterUsage("application_controller", client.Core().RESTClient().GetRateLimiter())
	}
	ac := &ApplicationController{
		client:        client,
		eventRecorder: eventBroadcaster.NewRecorder(api.EventSource{Component: "application-controller"}),
		queue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "application"),
	}

	aInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    ac.addApplicationNotification,
		UpdateFunc: ac.updateApplicationNotification,
		// This will enter the sync loop and no-op, because the application has been deleted from the store.
		DeleteFunc: ac.deleteApplicationNotification,
	})

	ac.syncHandler = ac.syncApplication
	return ac
}

// Run begins watching and syncing.
func (ac *ApplicationController) Run(workers int, stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	glog.Infof("Starting application controller")

	go wait.Until(ac.worker, time.Second, stopCh)

	<-stopCh
	glog.Infof("Shutting down application controller")
}

func (ac *ApplicationController) worker() {
	work := func() bool {
		key, quit := ac.queue.Get()
		if quit {
			return true
		}
		defer ac.queue.Done(key)

		ac.syncHandler(key.(string))

		return false
	}

	for {
		if quit := work(); quit {
			return
		}
	}
}

func (ac *ApplicationController) enqueueApplication(application *extensions.Application) {
	key, err := controller.KeyFunc(application)
	if err != nil {
		glog.Errorf("Couldn't get key for object %#v: %v", application, err)
		return
	}

	ac.queue.Add(key)
}

func (ac *ApplicationController) addApplicationNotification(obj interface{}) {
	d := obj.(*extensions.Application)
	glog.V(4).Infof("Adding application %s", d.Name)
	ac.enqueueApplication(d)
}

func (ac *ApplicationController) updateApplicationNotification(old, cur interface{}) {
	oldD := old.(*extensions.Application)
	glog.V(4).Infof("Updating application %s", oldD.Name)
	// Resync on application object relist.
	ac.enqueueApplication(cur.(*extensions.Application))
}

func (ac *ApplicationController) deleteApplicationNotification(obj interface{}) {
	d, ok := obj.(*extensions.Application)
	if !ok {
		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
		if !ok {
			glog.Errorf("Couldn't get object from tombstone %#v", obj)
			return
		}
		d, ok = tombstone.Obj.(*extensions.Application)
		if !ok {
			glog.Errorf("Tombstone contained object that is not a Application %#v", obj)
			return
		}
	}
	glog.V(4).Infof("Deleting application %s", d.Name)
	ac.enqueueApplication(d)
}

func (ac *ApplicationController) syncApplication(key string) error {
	segs := strings.Split(key, "/")
	namespace := segs[0]
	name := segs[1]
	application, err := ac.client.Extensions().Applications(namespace).Get(name)
	if err != nil {
		return err
	}

	glog.V(0).Infof("This is from syncApplication: %s/%s", application.Namespace, application.Name)
	return nil
}

该controller实现的功能很简单，不断在日志中记录("This is from syncApplication: %s/%s", application.Namespace, application.Name)。

当然，application_controller.go中用到了informers.ApplicationInformer，但还没定义。

informers.ApplicationInformer

在/pkg/controller/informers/extensions.go中定义ApplicationInformer：

// ApplicationInformer is a type of SharedIndexInformer which watches and lists all deployments.
type ApplicationInformer interface {
        Informer() cache.SharedIndexInformer
        Lister() *cache.StoreToApplicationLister
}

type applicationInformer struct {
        *sharedInformerFactory
}

func (f *applicationInformer) Informer() cache.SharedIndexInformer {
        f.lock.Lock()
        defer f.lock.Unlock()

        informerType := reflect.TypeOf(&extensions.Application{})
        informer, exists := f.informers[informerType]
        if exists {
                return informer
        }
        informer = cache.NewSharedIndexInformer(
                &cache.ListWatch{
                        ListFunc: func(options api.ListOptions) (runtime.Object, error) {
                                return f.client.Extensions().Applications(api.NamespaceAll).List(options)
                        },
                        WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
                                return f.client.Extensions().Applications(api.NamespaceAll).Watch(options)
                        },
                },
                &extensions.Application{},
                // TODO remove this.  It is hardcoded so that "Waiting for the second deployment to clear overlapping annotation" in
                // "overlapping deployment should not fight with each other" will work since it requires a full resync to work properly.
                30*time.Second,
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        f.informers[informerType] = informer

        return informer
}

func (f *applicationInformer) Lister() *cache.StoreToApplicationLister {
        informer := f.Informer()
        return &cache.StoreToApplicationLister{Indexer: informer.GetIndexer()}
}

可以看到，informers.ApplicationInformer中用到了cache.StoreToApplicationLister，但还未定义。

StoreToApplicationLister

修改pkg/client/cache/listers_extensions.go：

type StoreToApplicationLister struct {
        Indexer Indexer
}

func (s *StoreToApplicationLister) List(selector labels.Selector) (ret []*extensions.Application, err error) {
        err = ListAll(s.Indexer, selector, func(m interface{}) {
                ret = append(ret, m.(*extensions.Application))
        })
        return ret, err
}

func (s *StoreToApplicationLister) Applications(namespace string) storeApplicationsNamespacer {
        return storeApplicationsNamespacer{Indexer: s.Indexer, namespace: namespace}
}

type storeApplicationsNamespacer struct {
        Indexer   Indexer
        namespace string
}

func (s storeApplicationsNamespacer) List(selector labels.Selector) (ret []*extensions.Application, err error) {
        err = ListAllByNamespace(s.Indexer, s.namespace, selector, func(m interface{}) {
                ret = append(ret, m.(*extensions.Application))
        })
        return ret, err
}

func (s storeApplicationsNamespacer) Get(name string) (*extensions.Application, error) {
        obj, exists, err := s.Indexer.GetByKey(s.namespace + "/" + name)
        if err != nil {
                return nil, err
        }
        if !exists {
                return nil, errors.NewNotFound(extensions.Resource("application"), name)
        }
        return obj.(*extensions.Application), nil
}

informer factory

现在我们有了ApplicationInformer，可以在informer factory中注册ApplicationInformer了，修改/pkg/controller/informers/factory.go：

type SharedInformerFactory interface {
        // Start starts informers that can start AFTER the API server and controllers have started
        Start(stopCh <-chan struct{})

        ForResource(unversioned.GroupResource) (GenericInformer, error)

        // when you update these, update generic.go/ForResource, same package
        ......
        Applications() ApplicationInformer
		......
)

func (f *sharedInformerFactory) Applications() ApplicationInformer {
        return &applicationInformer{sharedInformerFactory: f}
}

controller cmd

修改/cmd/kube-controller-manager/app/controllermanager.go：

import (
	......
	"k8s.io/kubernetes/pkg/controller/application"
	......
)

在StartControllers()中修改：

if containsVersion(versions, groupVersion) && found {
		glog.Infof("Starting %s apis", groupVersion)

		......

		if containsResource(resources, "applications") {
				glog.Infof("Starting application controller")
				go application.NewApplicationController(sharedInformers.Applications(), client("application-controller")).Run(int(s.ConcurrentDeploymentSyncs), wait.NeverStop)
				time.Sleep(wait.Jitter(s.ControllerStartInterval.Duration, ControllerStartJitter))
		}

		......
}

验证

编译，启动kube-controller-manager，有如下日志重复输出：

I1209 21:28:25.478233   19090 application_controller.go:157] This is from syncApplication: default/ubuntu
I1209 21:28:25.534602   19090 application_controller.go:157] This is from syncApplication: default/ubuntu1