123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648 |
- package deployment
- import (
- "fmt"
- "reflect"
- "time"
- "k8s.io/klog"
- apps "k8s.io/api/apps/v1"
- "k8s.io/api/core/v1"
- "k8s.io/apimachinery/pkg/api/errors"
- metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
- "k8s.io/apimachinery/pkg/labels"
- "k8s.io/apimachinery/pkg/types"
- utilruntime "k8s.io/apimachinery/pkg/util/runtime"
- "k8s.io/apimachinery/pkg/util/wait"
- appsinformers "k8s.io/client-go/informers/apps/v1"
- coreinformers "k8s.io/client-go/informers/core/v1"
- clientset "k8s.io/client-go/kubernetes"
- "k8s.io/client-go/kubernetes/scheme"
- v1core "k8s.io/client-go/kubernetes/typed/core/v1"
- appslisters "k8s.io/client-go/listers/apps/v1"
- corelisters "k8s.io/client-go/listers/core/v1"
- "k8s.io/client-go/tools/cache"
- "k8s.io/client-go/tools/record"
- "k8s.io/client-go/util/workqueue"
- "k8s.io/kubernetes/pkg/controller"
- "k8s.io/kubernetes/pkg/controller/deployment/util"
- "k8s.io/kubernetes/pkg/util/metrics"
- )
- const (
-
-
-
-
-
- maxRetries = 15
- )
- var controllerKind = apps.SchemeGroupVersion.WithKind("Deployment")
- type DeploymentController struct {
-
- rsControl controller.RSControlInterface
- client clientset.Interface
- eventRecorder record.EventRecorder
-
- syncHandler func(dKey string) error
- // used for unit testing
- enqueueDeployment func(deployment *apps.Deployment)
- // dLister can list/get deployments from the shared informer's store
- dLister appslisters.DeploymentLister
- // rsLister can list/get replica sets from the shared informer's store
- rsLister appslisters.ReplicaSetLister
- // podLister can list/get pods from the shared informer's store
- podLister corelisters.PodLister
- // dListerSynced returns true if the Deployment store has been synced at least once.
- // Added as a member to the struct to allow injection for testing.
- dListerSynced cache.InformerSynced
- // rsListerSynced returns true if the ReplicaSet store has been synced at least once.
- // Added as a member to the struct to allow injection for testing.
- rsListerSynced cache.InformerSynced
- // podListerSynced returns true if the pod store has been synced at least once.
- // Added as a member to the struct to allow injection for testing.
- podListerSynced cache.InformerSynced
- // Deployments that need to be synced
- queue workqueue.RateLimitingInterface
- }
- // NewDeploymentController creates a new DeploymentController.
- func NewDeploymentController(dInformer appsinformers.DeploymentInformer, rsInformer appsinformers.ReplicaSetInformer, podInformer coreinformers.PodInformer, client clientset.Interface) (*DeploymentController, error) {
- eventBroadcaster := record.NewBroadcaster()
- eventBroadcaster.StartLogging(klog.Infof)
- eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: client.CoreV1().Events("")})
- if client != nil && client.CoreV1().RESTClient().GetRateLimiter() != nil {
- if err := metrics.RegisterMetricAndTrackRateLimiterUsage("deployment_controller", client.CoreV1().RESTClient().GetRateLimiter()); err != nil {
- return nil, err
- }
- }
- dc := &DeploymentController{
- client: client,
- eventRecorder: eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "deployment-controller"}),
- queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "deployment"),
- }
- dc.rsControl = controller.RealRSControl{
- KubeClient: client,
- Recorder: dc.eventRecorder,
- }
- dInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
- AddFunc: dc.addDeployment,
- UpdateFunc: dc.updateDeployment,
-
- DeleteFunc: dc.deleteDeployment,
- })
- rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
- AddFunc: dc.addReplicaSet,
- UpdateFunc: dc.updateReplicaSet,
- DeleteFunc: dc.deleteReplicaSet,
- })
- podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
- DeleteFunc: dc.deletePod,
- })
- dc.syncHandler = dc.syncDeployment
- dc.enqueueDeployment = dc.enqueue
- dc.dLister = dInformer.Lister()
- dc.rsLister = rsInformer.Lister()
- dc.podLister = podInformer.Lister()
- dc.dListerSynced = dInformer.Informer().HasSynced
- dc.rsListerSynced = rsInformer.Informer().HasSynced
- dc.podListerSynced = podInformer.Informer().HasSynced
- return dc, nil
- }
- func (dc *DeploymentController) Run(workers int, stopCh <-chan struct{}) {
- defer utilruntime.HandleCrash()
- defer dc.queue.ShutDown()
- klog.Infof("Starting deployment controller")
- defer klog.Infof("Shutting down deployment controller")
- if !controller.WaitForCacheSync("deployment", stopCh, dc.dListerSynced, dc.rsListerSynced, dc.podListerSynced) {
- return
- }
- for i := 0; i < workers; i++ {
- go wait.Until(dc.worker, time.Second, stopCh)
- }
- <-stopCh
- }
- func (dc *DeploymentController) addDeployment(obj interface{}) {
- d := obj.(*apps.Deployment)
- klog.V(4).Infof("Adding deployment %s", d.Name)
- dc.enqueueDeployment(d)
- }
- func (dc *DeploymentController) updateDeployment(old, cur interface{}) {
- oldD := old.(*apps.Deployment)
- curD := cur.(*apps.Deployment)
- klog.V(4).Infof("Updating deployment %s", oldD.Name)
- dc.enqueueDeployment(curD)
- }
- func (dc *DeploymentController) deleteDeployment(obj interface{}) {
- d, ok := obj.(*apps.Deployment)
- if !ok {
- tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
- if !ok {
- utilruntime.HandleError(fmt.Errorf("Couldn't get object from tombstone %#v", obj))
- return
- }
- d, ok = tombstone.Obj.(*apps.Deployment)
- if !ok {
- utilruntime.HandleError(fmt.Errorf("Tombstone contained object that is not a Deployment %#v", obj))
- return
- }
- }
- klog.V(4).Infof("Deleting deployment %s", d.Name)
- dc.enqueueDeployment(d)
- }
- func (dc *DeploymentController) addReplicaSet(obj interface{}) {
- rs := obj.(*apps.ReplicaSet)
- if rs.DeletionTimestamp != nil {
-
-
- dc.deleteReplicaSet(rs)
- return
- }
-
- if controllerRef := metav1.GetControllerOf(rs); controllerRef != nil {
- d := dc.resolveControllerRef(rs.Namespace, controllerRef)
- if d == nil {
- return
- }
- klog.V(4).Infof("ReplicaSet %s added.", rs.Name)
- dc.enqueueDeployment(d)
- return
- }
-
-
- ds := dc.getDeploymentsForReplicaSet(rs)
- if len(ds) == 0 {
- return
- }
- klog.V(4).Infof("Orphan ReplicaSet %s added.", rs.Name)
- for _, d := range ds {
- dc.enqueueDeployment(d)
- }
- }
- func (dc *DeploymentController) getDeploymentsForReplicaSet(rs *apps.ReplicaSet) []*apps.Deployment {
- deployments, err := dc.dLister.GetDeploymentsForReplicaSet(rs)
- if err != nil || len(deployments) == 0 {
- return nil
- }
-
-
-
-
- if len(deployments) > 1 {
-
-
- klog.V(4).Infof("user error! more than one deployment is selecting replica set %s/%s with labels: %#v, returning %s/%s",
- rs.Namespace, rs.Name, rs.Labels, deployments[0].Namespace, deployments[0].Name)
- }
- return deployments
- }
- func (dc *DeploymentController) updateReplicaSet(old, cur interface{}) {
- curRS := cur.(*apps.ReplicaSet)
- oldRS := old.(*apps.ReplicaSet)
- if curRS.ResourceVersion == oldRS.ResourceVersion {
-
-
- return
- }
- curControllerRef := metav1.GetControllerOf(curRS)
- oldControllerRef := metav1.GetControllerOf(oldRS)
- controllerRefChanged := !reflect.DeepEqual(curControllerRef, oldControllerRef)
- if controllerRefChanged && oldControllerRef != nil {
-
- if d := dc.resolveControllerRef(oldRS.Namespace, oldControllerRef); d != nil {
- dc.enqueueDeployment(d)
- }
- }
-
- if curControllerRef != nil {
- d := dc.resolveControllerRef(curRS.Namespace, curControllerRef)
- if d == nil {
- return
- }
- klog.V(4).Infof("ReplicaSet %s updated.", curRS.Name)
- dc.enqueueDeployment(d)
- return
- }
-
-
- labelChanged := !reflect.DeepEqual(curRS.Labels, oldRS.Labels)
- if labelChanged || controllerRefChanged {
- ds := dc.getDeploymentsForReplicaSet(curRS)
- if len(ds) == 0 {
- return
- }
- klog.V(4).Infof("Orphan ReplicaSet %s updated.", curRS.Name)
- for _, d := range ds {
- dc.enqueueDeployment(d)
- }
- }
- }
- func (dc *DeploymentController) deleteReplicaSet(obj interface{}) {
- rs, ok := obj.(*apps.ReplicaSet)
-
-
-
-
- if !ok {
- tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
- if !ok {
- utilruntime.HandleError(fmt.Errorf("Couldn't get object from tombstone %#v", obj))
- return
- }
- rs, ok = tombstone.Obj.(*apps.ReplicaSet)
- if !ok {
- utilruntime.HandleError(fmt.Errorf("Tombstone contained object that is not a ReplicaSet %#v", obj))
- return
- }
- }
- controllerRef := metav1.GetControllerOf(rs)
- if controllerRef == nil {
-
- return
- }
- d := dc.resolveControllerRef(rs.Namespace, controllerRef)
- if d == nil {
- return
- }
- klog.V(4).Infof("ReplicaSet %s deleted.", rs.Name)
- dc.enqueueDeployment(d)
- }
- func (dc *DeploymentController) deletePod(obj interface{}) {
- pod, ok := obj.(*v1.Pod)
-
-
-
-
- if !ok {
- tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
- if !ok {
- utilruntime.HandleError(fmt.Errorf("Couldn't get object from tombstone %#v", obj))
- return
- }
- pod, ok = tombstone.Obj.(*v1.Pod)
- if !ok {
- utilruntime.HandleError(fmt.Errorf("Tombstone contained object that is not a pod %#v", obj))
- return
- }
- }
- klog.V(4).Infof("Pod %s deleted.", pod.Name)
- if d := dc.getDeploymentForPod(pod); d != nil && d.Spec.Strategy.Type == apps.RecreateDeploymentStrategyType {
-
- rsList, err := util.ListReplicaSets(d, util.RsListFromClient(dc.client.AppsV1()))
- if err != nil {
- return
- }
- podMap, err := dc.getPodMapForDeployment(d, rsList)
- if err != nil {
- return
- }
- numPods := 0
- for _, podList := range podMap {
- numPods += len(podList.Items)
- }
- if numPods == 0 {
- dc.enqueueDeployment(d)
- }
- }
- }
- func (dc *DeploymentController) enqueue(deployment *apps.Deployment) {
- key, err := controller.KeyFunc(deployment)
- if err != nil {
- utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %#v: %v", deployment, err))
- return
- }
- dc.queue.Add(key)
- }
- func (dc *DeploymentController) enqueueRateLimited(deployment *apps.Deployment) {
- key, err := controller.KeyFunc(deployment)
- if err != nil {
- utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %#v: %v", deployment, err))
- return
- }
- dc.queue.AddRateLimited(key)
- }
- func (dc *DeploymentController) enqueueAfter(deployment *apps.Deployment, after time.Duration) {
- key, err := controller.KeyFunc(deployment)
- if err != nil {
- utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %#v: %v", deployment, err))
- return
- }
- dc.queue.AddAfter(key, after)
- }
- func (dc *DeploymentController) getDeploymentForPod(pod *v1.Pod) *apps.Deployment {
-
- var rs *apps.ReplicaSet
- var err error
- controllerRef := metav1.GetControllerOf(pod)
- if controllerRef == nil {
-
- return nil
- }
- if controllerRef.Kind != apps.SchemeGroupVersion.WithKind("ReplicaSet").Kind {
-
- return nil
- }
- rs, err = dc.rsLister.ReplicaSets(pod.Namespace).Get(controllerRef.Name)
- if err != nil || rs.UID != controllerRef.UID {
- klog.V(4).Infof("Cannot get replicaset %q for pod %q: %v", controllerRef.Name, pod.Name, err)
- return nil
- }
-
- controllerRef = metav1.GetControllerOf(rs)
- if controllerRef == nil {
- return nil
- }
- return dc.resolveControllerRef(rs.Namespace, controllerRef)
- }
- func (dc *DeploymentController) resolveControllerRef(namespace string, controllerRef *metav1.OwnerReference) *apps.Deployment {
-
-
- if controllerRef.Kind != controllerKind.Kind {
- return nil
- }
- d, err := dc.dLister.Deployments(namespace).Get(controllerRef.Name)
- if err != nil {
- return nil
- }
- if d.UID != controllerRef.UID {
-
-
- return nil
- }
- return d
- }
- func (dc *DeploymentController) worker() {
- for dc.processNextWorkItem() {
- }
- }
- func (dc *DeploymentController) processNextWorkItem() bool {
- key, quit := dc.queue.Get()
- if quit {
- return false
- }
- defer dc.queue.Done(key)
- err := dc.syncHandler(key.(string))
- dc.handleErr(err, key)
- return true
- }
- func (dc *DeploymentController) handleErr(err error, key interface{}) {
- if err == nil {
- dc.queue.Forget(key)
- return
- }
- if dc.queue.NumRequeues(key) < maxRetries {
- klog.V(2).Infof("Error syncing deployment %v: %v", key, err)
- dc.queue.AddRateLimited(key)
- return
- }
- utilruntime.HandleError(err)
- klog.V(2).Infof("Dropping deployment %q out of the queue: %v", key, err)
- dc.queue.Forget(key)
- }
- func (dc *DeploymentController) getReplicaSetsForDeployment(d *apps.Deployment) ([]*apps.ReplicaSet, error) {
-
-
- rsList, err := dc.rsLister.ReplicaSets(d.Namespace).List(labels.Everything())
- if err != nil {
- return nil, err
- }
- deploymentSelector, err := metav1.LabelSelectorAsSelector(d.Spec.Selector)
- if err != nil {
- return nil, fmt.Errorf("deployment %s/%s has invalid label selector: %v", d.Namespace, d.Name, err)
- }
-
-
- canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
- fresh, err := dc.client.AppsV1().Deployments(d.Namespace).Get(d.Name, metav1.GetOptions{})
- if err != nil {
- return nil, err
- }
- if fresh.UID != d.UID {
- return nil, fmt.Errorf("original Deployment %v/%v is gone: got uid %v, wanted %v", d.Namespace, d.Name, fresh.UID, d.UID)
- }
- return fresh, nil
- })
- cm := controller.NewReplicaSetControllerRefManager(dc.rsControl, d, deploymentSelector, controllerKind, canAdoptFunc)
- return cm.ClaimReplicaSets(rsList)
- }
- func (dc *DeploymentController) getPodMapForDeployment(d *apps.Deployment, rsList []*apps.ReplicaSet) (map[types.UID]*v1.PodList, error) {
-
- selector, err := metav1.LabelSelectorAsSelector(d.Spec.Selector)
- if err != nil {
- return nil, err
- }
- pods, err := dc.podLister.Pods(d.Namespace).List(selector)
- if err != nil {
- return nil, err
- }
-
- podMap := make(map[types.UID]*v1.PodList, len(rsList))
- for _, rs := range rsList {
- podMap[rs.UID] = &v1.PodList{}
- }
- for _, pod := range pods {
-
-
- controllerRef := metav1.GetControllerOf(pod)
- if controllerRef == nil {
- continue
- }
-
- if podList, ok := podMap[controllerRef.UID]; ok {
- podList.Items = append(podList.Items, *pod)
- }
- }
- return podMap, nil
- }
- func (dc *DeploymentController) syncDeployment(key string) error {
- startTime := time.Now()
- klog.V(4).Infof("Started syncing deployment %q (%v)", key, startTime)
- defer func() {
- klog.V(4).Infof("Finished syncing deployment %q (%v)", key, time.Since(startTime))
- }()
- namespace, name, err := cache.SplitMetaNamespaceKey(key)
- if err != nil {
- return err
- }
- deployment, err := dc.dLister.Deployments(namespace).Get(name)
- if errors.IsNotFound(err) {
- klog.V(2).Infof("Deployment %v has been deleted", key)
- return nil
- }
- if err != nil {
- return err
- }
-
-
- d := deployment.DeepCopy()
- everything := metav1.LabelSelector{}
- if reflect.DeepEqual(d.Spec.Selector, &everything) {
- dc.eventRecorder.Eventf(d, v1.EventTypeWarning, "SelectingAll", "This deployment is selecting all pods. A non-empty selector is required.")
- if d.Status.ObservedGeneration < d.Generation {
- d.Status.ObservedGeneration = d.Generation
- dc.client.AppsV1().Deployments(d.Namespace).UpdateStatus(d)
- }
- return nil
- }
-
-
- rsList, err := dc.getReplicaSetsForDeployment(d)
- if err != nil {
- return err
- }
-
-
-
-
-
- podMap, err := dc.getPodMapForDeployment(d, rsList)
- if err != nil {
- return err
- }
- if d.DeletionTimestamp != nil {
- return dc.syncStatusOnly(d, rsList)
- }
-
-
-
- if err = dc.checkPausedConditions(d); err != nil {
- return err
- }
- if d.Spec.Paused {
- return dc.sync(d, rsList)
- }
-
-
-
- if getRollbackTo(d) != nil {
- return dc.rollback(d, rsList)
- }
- scalingEvent, err := dc.isScalingEvent(d, rsList)
- if err != nil {
- return err
- }
- if scalingEvent {
- return dc.sync(d, rsList)
- }
- switch d.Spec.Strategy.Type {
- case apps.RecreateDeploymentStrategyType:
- return dc.rolloutRecreate(d, rsList, podMap)
- case apps.RollingUpdateDeploymentStrategyType:
- return dc.rolloutRolling(d, rsList)
- }
- return fmt.Errorf("unexpected deployment strategy type: %s", d.Spec.Strategy.Type)
- }
|