From 1b4a998bcb1284da444e4360efd2b76c133945ec Mon Sep 17 00:00:00 2001 From: nikhiljindal Date: Mon, 7 Nov 2016 00:03:27 -0800 Subject: [PATCH] Updating federation service controller to support cascading deletion --- .../service/servicecontroller.go | 197 +++++++++++++++++- 1 file changed, 186 insertions(+), 11 deletions(-) diff --git a/federation/pkg/federation-controller/service/servicecontroller.go b/federation/pkg/federation-controller/service/servicecontroller.go index 9929b1684f6..a02bd26505f 100644 --- a/federation/pkg/federation-controller/service/servicecontroller.go +++ b/federation/pkg/federation-controller/service/servicecontroller.go @@ -28,6 +28,9 @@ import ( federationcache "k8s.io/kubernetes/federation/client/cache" fedclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_clientset" "k8s.io/kubernetes/federation/pkg/dnsprovider" + "k8s.io/kubernetes/federation/pkg/federation-controller/util" + fedutil "k8s.io/kubernetes/federation/pkg/federation-controller/util" + "k8s.io/kubernetes/federation/pkg/federation-controller/util/deletionhelper" "k8s.io/kubernetes/pkg/api" "k8s.io/kubernetes/pkg/api/errors" v1 "k8s.io/kubernetes/pkg/api/v1" @@ -36,6 +39,7 @@ import ( kubeclientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset" "k8s.io/kubernetes/pkg/client/record" "k8s.io/kubernetes/pkg/controller" + "k8s.io/kubernetes/pkg/conversion" pkgruntime "k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/util/runtime" "k8s.io/kubernetes/pkg/util/sets" @@ -68,6 +72,10 @@ const ( KubeAPIBurst = 30 maxNoOfClusters = 100 + + updateTimeout = 30 * time.Second + allClustersKey = "ALL_CLUSTERS" + clusterAvailableDelay = time.Second * 20 ) type cachedService struct { @@ -114,6 +122,7 @@ type ServiceController struct { serviceStore cache.StoreToServiceLister // Watches changes to all services serviceController *cache.Controller + federatedInformer fedutil.FederatedInformer // A store of services, populated by the serviceController clusterStore federationcache.StoreToClusterLister // Watches changes to all services @@ -133,6 +142,12 @@ type ServiceController struct { serviceWorkerMap map[string]bool // channel for worker to signal that it is going out of existence serviceWorkerDoneChan chan string + + // For triggering all services reconciliation. This is used when + // a new cluster becomes available. + clusterDeliverer *util.DelayingDeliverer + + deletionHelper *deletionhelper.DeletionHelper } // New returns a new service controller to keep DNS provider service resources @@ -162,6 +177,7 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface, queue: workqueue.New(), knownClusterSet: make(sets.String), } + s.clusterDeliverer = util.NewDelayingDeliverer() s.serviceStore.Indexer, s.serviceController = cache.NewIndexerInformer( &cache.ListWatch{ ListFunc: func(options v1.ListOptions) (pkgruntime.Object, error) { @@ -224,6 +240,66 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface, }, ) + clusterLifecycle := fedutil.ClusterLifecycleHandlerFuncs{ + ClusterAvailable: func(cluster *v1beta1.Cluster) { + s.clusterDeliverer.DeliverAfter(allClustersKey, nil, clusterAvailableDelay) + }, + } + fedInformerFactory := func(cluster *v1beta1.Cluster, targetClient kubeclientset.Interface) (cache.Store, cache.ControllerInterface) { + return cache.NewInformer( + &cache.ListWatch{ + ListFunc: func(options v1.ListOptions) (pkgruntime.Object, error) { + return targetClient.Core().Services(v1.NamespaceAll).List(options) + }, + WatchFunc: func(options v1.ListOptions) (watch.Interface, error) { + return targetClient.Core().Services(v1.NamespaceAll).Watch(options) + }, + }, + &v1.Service{}, + controller.NoResyncPeriodFunc(), + // Trigger reconciliation whenever something in federated cluster is changed. In most cases it + // would be just confirmation that some service operation succeeded. + util.NewTriggerOnAllChanges( + func(obj pkgruntime.Object) { + // TODO: Use this to enque services. + }, + )) + } + + s.federatedInformer = fedutil.NewFederatedInformer(federationClient, fedInformerFactory, &clusterLifecycle) + + federatedUpdater := fedutil.NewFederatedUpdater(s.federatedInformer, + func(client kubeclientset.Interface, obj pkgruntime.Object) error { + svc := obj.(*v1.Service) + _, err := client.Core().Services(svc.Namespace).Create(svc) + return err + }, + func(client kubeclientset.Interface, obj pkgruntime.Object) error { + svc := obj.(*v1.Service) + _, err := client.Core().Services(svc.Namespace).Update(svc) + return err + }, + func(client kubeclientset.Interface, obj pkgruntime.Object) error { + svc := obj.(*v1.Service) + err := client.Core().Services(svc.Namespace).Delete(svc.Name, &v1.DeleteOptions{}) + return err + }) + + s.deletionHelper = deletionhelper.NewDeletionHelper( + s.hasFinalizerFunc, + s.removeFinalizerFunc, + s.addFinalizerFunc, + // objNameFunc + func(obj pkgruntime.Object) string { + service := obj.(*v1.Service) + return service.Name + }, + updateTimeout, + s.eventRecorder, + s.federatedInformer, + federatedUpdater, + ) + s.endpointWorkerMap = make(map[string]bool) s.serviceWorkerMap = make(map[string]bool) s.endpointWorkerDoneChan = make(chan string, maxNoOfClusters) @@ -231,6 +307,54 @@ func New(federationClient fedclientset.Interface, dns dnsprovider.Interface, return s } +// Returns true if the given object has the given finalizer in its ObjectMeta. +func (s *ServiceController) hasFinalizerFunc(obj pkgruntime.Object, finalizer string) bool { + service := obj.(*v1.Service) + for i := range service.ObjectMeta.Finalizers { + if string(service.ObjectMeta.Finalizers[i]) == finalizer { + return true + } + } + return false +} + +// Removes the finalizer from the given objects ObjectMeta. +// Assumes that the given object is a service. +func (s *ServiceController) removeFinalizerFunc(obj pkgruntime.Object, finalizer string) (pkgruntime.Object, error) { + service := obj.(*v1.Service) + newFinalizers := []string{} + hasFinalizer := false + for i := range service.ObjectMeta.Finalizers { + if string(service.ObjectMeta.Finalizers[i]) != finalizer { + newFinalizers = append(newFinalizers, service.ObjectMeta.Finalizers[i]) + } else { + hasFinalizer = true + } + } + if !hasFinalizer { + // Nothing to do. + return obj, nil + } + service.ObjectMeta.Finalizers = newFinalizers + service, err := s.federationClient.Core().Services(service.Namespace).Update(service) + if err != nil { + return nil, fmt.Errorf("failed to remove finalizer %s from service %s: %v", finalizer, service.Name, err) + } + return service, nil +} + +// Adds the given finalizer to the given objects ObjectMeta. +// Assumes that the given object is a service. +func (s *ServiceController) addFinalizerFunc(obj pkgruntime.Object, finalizer string) (pkgruntime.Object, error) { + service := obj.(*v1.Service) + service.ObjectMeta.Finalizers = append(service.ObjectMeta.Finalizers, finalizer) + service, err := s.federationClient.Core().Services(service.Namespace).Update(service) + if err != nil { + return nil, fmt.Errorf("failed to add finalizer %s to service %s: %v", finalizer, service.Name, err) + } + return service, nil +} + // obj could be an *api.Service, or a DeletionFinalStateUnknown marker item. func (s *ServiceController) enqueueService(obj interface{}) { key, err := controller.KeyFunc(obj) @@ -257,6 +381,10 @@ func (s *ServiceController) Run(workers int, stopCh <-chan struct{}) error { return err } defer runtime.HandleCrash() + s.federatedInformer.Start() + s.clusterDeliverer.StartWithHandler(func(_ *util.DelayingDelivererItem) { + // TODO: Use this new clusterDeliverer to reconcile services in new clusters. + }) go s.serviceController.Run(stopCh) go s.clusterController.Run(stopCh) for i := 0; i < workers; i++ { @@ -268,6 +396,7 @@ func (s *ServiceController) Run(workers int, stopCh <-chan struct{}) error { <-stopCh glog.Infof("Shutting down Federation Service Controller") s.queue.ShutDown() + s.federatedInformer.Stop() return nil } @@ -361,20 +490,16 @@ func (s *ServiceController) processServiceForCluster(cachedService *cachedServic // should be retried. func (s *ServiceController) updateFederationService(key string, cachedService *cachedService) (error, bool) { // Clone federation service, and create them in underlying k8s cluster - clone, err := api.Scheme.DeepCopy(cachedService.lastState) - if err != nil { - return err, !retryable - } - service, ok := clone.(*v1.Service) - if !ok { - return fmt.Errorf("Unexpected service cast error : %v\n", service), !retryable + desiredService := &v1.Service{ + ObjectMeta: util.DeepCopyRelevantObjectMeta(cachedService.lastState.ObjectMeta), + Spec: *(util.DeepCopyApiTypeOrPanic(&cachedService.lastState.Spec).(*v1.ServiceSpec)), } // handle available clusters one by one var hasErr bool for clusterName, cache := range s.clusterCache.clientMap { go func(cache *clusterCache, clusterName string) { - err = s.processServiceForCluster(cachedService, clusterName, service, cache.clientset) + err := s.processServiceForCluster(cachedService, clusterName, desiredService, cache.clientset) if err != nil { hasErr = true } @@ -382,7 +507,7 @@ func (s *ServiceController) updateFederationService(key string, cachedService *c } if hasErr { // detail error has been dumped inside the loop - return fmt.Errorf("Service %s/%s was not successfully updated to all clusters", service.Namespace, service.Name), retryable + return fmt.Errorf("Service %s/%s was not successfully updated to all clusters", desiredService.Namespace, desiredService.Name), retryable } return nil, !retryable } @@ -812,18 +937,25 @@ func (s *ServiceController) syncService(key string) error { glog.V(4).Infof("Finished syncing service %q (%v)", key, time.Now().Sub(startTime)) }() // obj holds the latest service info from apiserver - obj, exists, err := s.serviceStore.Indexer.GetByKey(key) + objFromStore, exists, err := s.serviceStore.Indexer.GetByKey(key) if err != nil { glog.Errorf("Unable to retrieve service %v from store: %v", key, err) s.queue.Add(key) return err } - if !exists { // service absence in store means watcher caught the deletion, ensure LB info is cleaned glog.Infof("Service has been deleted %v", key) err, retryDelay = s.processServiceDeletion(key) } + // Create a copy before modifying the obj to prevent race condition with + // other readers of obj from store. + obj, err := conversion.NewCloner().DeepCopy(objFromStore) + if err != nil { + glog.Errorf("Error in deep copying service %v retrieved from store: %v", key, err) + s.queue.Add(key) + return err + } if exists { service, ok := obj.(*v1.Service) @@ -857,6 +989,29 @@ func (s *ServiceController) processServiceUpdate(cachedService *cachedService, s cachedService.rwlock.Lock() defer cachedService.rwlock.Unlock() + if service.DeletionTimestamp != nil { + if err := s.delete(service); err != nil { + glog.Errorf("Failed to delete %s: %v", service, err) + s.eventRecorder.Eventf(service, api.EventTypeNormal, "DeleteFailed", + "Service delete failed: %v", err) + return err, cachedService.nextRetryDelay() + } + return nil, doNotRetry + } + + glog.V(3).Infof("Ensuring delete object from underlying clusters finalizer for service: %s", + service.Name) + // Add the required finalizers before creating a service in underlying clusters. + updatedServiceObj, err := s.deletionHelper.EnsureFinalizers(service) + if err != nil { + glog.Errorf("Failed to ensure delete object from underlying clusters finalizer in service %s: %v", + service.Name, err) + return err, cachedService.nextRetryDelay() + } + service = updatedServiceObj.(*v1.Service) + + glog.V(3).Infof("Syncing service %s in underlying clusters", service.Name) + // Update the cached service (used above for populating synthetic deletes) // alway trust service, which is retrieve from serviceStore, which keeps the latest service info getting from apiserver // if the same service is changed before this go routine finished, there will be another queue entry to handle that. @@ -885,6 +1040,26 @@ func (s *ServiceController) processServiceUpdate(cachedService *cachedService, s return nil, doNotRetry } +// delete deletes the given service or returns error if the deletion was not complete. +func (s *ServiceController) delete(service *v1.Service) error { + glog.V(3).Infof("Handling deletion of service: %v", *service) + _, err := s.deletionHelper.HandleObjectInUnderlyingClusters(service) + if err != nil { + return err + } + + err = s.federationClient.Core().Services(service.Namespace).Delete(service.Name, nil) + if err != nil { + // Its all good if the error is not found error. That means it is deleted already and we do not have to do anything. + // This is expected when we are processing an update as a result of service finalizer deletion. + // The process that deleted the last finalizer is also going to delete the service and we do not have to do anything. + if !errors.IsNotFound(err) { + return fmt.Errorf("failed to delete service: %v", err) + } + } + return nil +} + // processServiceDeletion returns an error if processing the service deletion failed, along with a time.Duration // indicating whether processing should be retried; zero means no-retry; otherwise // we should retry in that Duration.