Merge pull request #47084 from janetkuo/daemonset-history-followup

Automatic merge from submit-queue (batch tested with PRs 47084, 46016, 46372)

Update adoption/release of DaemonSet controller history, and wait for history store sync

**What this PR does / why we need it**: 
~Depends on #47075, so that DaemonSet controller can update history's controller ref. Ignore that commit when reviewing.~ (merged)

**Which issue this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close that issue when PR gets merged)*: #46981 

**Special notes for your reviewer**: @kubernetes/sig-apps-bugs 

**Release note**:

```release-note
NONE
```

pkg/controller/BUILD

@@ -25,6 +25,7 @@ go_library(
         "//pkg/api/v1/pod:go_default_library",
         "//pkg/api/v1/ref:go_default_library",
         "//pkg/api/validation:go_default_library",
+        "//pkg/apis/apps/v1beta1:go_default_library",
         "//pkg/apis/authentication/v1:go_default_library",
         "//pkg/apis/extensions/v1beta1:go_default_library",
         "//pkg/client/clientset_generated/clientset:go_default_library",

pkg/controller/controller_ref_manager.go

@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/kubernetes/pkg/api/v1"
+	appsv1beta1 "k8s.io/kubernetes/pkg/apis/apps/v1beta1"
 	extensions "k8s.io/kubernetes/pkg/apis/extensions/v1beta1"
 )
 
@@ -394,3 +395,121 @@ func RecheckDeletionTimestamp(getObject func() (metav1.Object, error)) func() er
 		return nil
 	}
 }
+
+// ControllerRevisionControllerRefManager is used to manage controllerRef of ControllerRevisions.
+// Three methods are defined on this object 1: Classify 2: AdoptControllerRevision and
+// 3: ReleaseControllerRevision which are used to classify the ControllerRevisions into appropriate
+// categories and accordingly adopt or release them. See comments on these functions
+// for more details.
+type ControllerRevisionControllerRefManager struct {
+	baseControllerRefManager
+	controllerKind schema.GroupVersionKind
+	crControl      ControllerRevisionControlInterface
+}
+
+// NewControllerRevisionControllerRefManager returns a ControllerRevisionControllerRefManager that exposes
+// methods to manage the controllerRef of ControllerRevisions.
+//
+// The canAdopt() function can be used to perform a potentially expensive check
+// (such as a live GET from the API server) prior to the first adoption.
+// It will only be called (at most once) if an adoption is actually attempted.
+// If canAdopt() returns a non-nil error, all adoptions will fail.
+//
+// NOTE: Once canAdopt() is called, it will not be called again by the same
+//       ControllerRevisionControllerRefManager instance. Create a new instance if it
+//       makes sense to check canAdopt() again (e.g. in a different sync pass).
+func NewControllerRevisionControllerRefManager(
+	crControl ControllerRevisionControlInterface,
+	controller metav1.Object,
+	selector labels.Selector,
+	controllerKind schema.GroupVersionKind,
+	canAdopt func() error,
+) *ControllerRevisionControllerRefManager {
+	return &ControllerRevisionControllerRefManager{
+		baseControllerRefManager: baseControllerRefManager{
+			controller:   controller,
+			selector:     selector,
+			canAdoptFunc: canAdopt,
+		},
+		controllerKind: controllerKind,
+		crControl:      crControl,
+	}
+}
+
+// ClaimControllerRevisions tries to take ownership of a list of ControllerRevisions.
+//
+// It will reconcile the following:
+//   * Adopt orphans if the selector matches.
+//   * Release owned objects if the selector no longer matches.
+//
+// A non-nil error is returned if some form of reconciliation was attemped and
+// failed. Usually, controllers should try again later in case reconciliation
+// is still needed.
+//
+// If the error is nil, either the reconciliation succeeded, or no
+// reconciliation was necessary. The list of ControllerRevisions that you now own is
+// returned.
+func (m *ControllerRevisionControllerRefManager) ClaimControllerRevisions(histories []*appsv1beta1.ControllerRevision) ([]*appsv1beta1.ControllerRevision, error) {
+	var claimed []*appsv1beta1.ControllerRevision
+	var errlist []error
+
+	match := func(obj metav1.Object) bool {
+		return m.selector.Matches(labels.Set(obj.GetLabels()))
+	}
+	adopt := func(obj metav1.Object) error {
+		return m.AdoptControllerRevision(obj.(*appsv1beta1.ControllerRevision))
+	}
+	release := func(obj metav1.Object) error {
+		return m.ReleaseControllerRevision(obj.(*appsv1beta1.ControllerRevision))
+	}
+
+	for _, h := range histories {
+		ok, err := m.claimObject(h, match, adopt, release)
+		if err != nil {
+			errlist = append(errlist, err)
+			continue
+		}
+		if ok {
+			claimed = append(claimed, h)
+		}
+	}
+	return claimed, utilerrors.NewAggregate(errlist)
+}
+
+// AdoptControllerRevision sends a patch to take control of the ControllerRevision. It returns the error if
+// the patching fails.
+func (m *ControllerRevisionControllerRefManager) AdoptControllerRevision(history *appsv1beta1.ControllerRevision) error {
+	if err := m.canAdopt(); err != nil {
+		return fmt.Errorf("can't adopt ControllerRevision %v/%v (%v): %v", history.Namespace, history.Name, history.UID, err)
+	}
+	// Note that ValidateOwnerReferences() will reject this patch if another
+	// OwnerReference exists with controller=true.
+	addControllerPatch := fmt.Sprintf(
+		`{"metadata":{"ownerReferences":[{"apiVersion":"%s","kind":"%s","name":"%s","uid":"%s","controller":true,"blockOwnerDeletion":true}],"uid":"%s"}}`,
+		m.controllerKind.GroupVersion(), m.controllerKind.Kind,
+		m.controller.GetName(), m.controller.GetUID(), history.UID)
+	return m.crControl.PatchControllerRevision(history.Namespace, history.Name, []byte(addControllerPatch))
+}
+
+// ReleaseControllerRevision sends a patch to free the ControllerRevision from the control of its controller.
+// It returns the error if the patching fails. 404 and 422 errors are ignored.
+func (m *ControllerRevisionControllerRefManager) ReleaseControllerRevision(history *appsv1beta1.ControllerRevision) error {
+	glog.V(2).Infof("patching ControllerRevision %s_%s to remove its controllerRef to %s/%s:%s",
+		history.Namespace, history.Name, m.controllerKind.GroupVersion(), m.controllerKind.Kind, m.controller.GetName())
+	deleteOwnerRefPatch := fmt.Sprintf(`{"metadata":{"ownerReferences":[{"$patch":"delete","uid":"%s"}],"uid":"%s"}}`, m.controller.GetUID(), history.UID)
+	err := m.crControl.PatchControllerRevision(history.Namespace, history.Name, []byte(deleteOwnerRefPatch))
+	if err != nil {
+		if errors.IsNotFound(err) {
+			// If the ControllerRevision no longer exists, ignore it.
+			return nil
+		}
+		if errors.IsInvalid(err) {
+			// Invalid error will be returned in two cases: 1. the ControllerRevision
+			// has no owner reference, 2. the uid of the ControllerRevision doesn't
+			// match, which means the ControllerRevision is deleted and then recreated.
+			// In both cases, the error can be ignored.
+			return nil
+		}
+	}
+	return err
+}

pkg/controller/controller_utils.go

@@ -401,6 +401,26 @@ func (r RealRSControl) PatchReplicaSet(namespace, name string, data []byte) erro
 	return err
 }
 
+// TODO: merge the controller revision interface in controller_history.go with this one
+// ControllerRevisionControlInterface is an interface that knows how to patch
+// ControllerRevisions, as well as increment or decrement them. It is used
+// by the daemonset controller to ease testing of actions that it takes.
+type ControllerRevisionControlInterface interface {
+	PatchControllerRevision(namespace, name string, data []byte) error
+}
+
+// RealControllerRevisionControl is the default implementation of ControllerRevisionControlInterface.
+type RealControllerRevisionControl struct {
+	KubeClient clientset.Interface
+}
+
+var _ ControllerRevisionControlInterface = &RealControllerRevisionControl{}
+
+func (r RealControllerRevisionControl) PatchControllerRevision(namespace, name string, data []byte) error {
+	_, err := r.KubeClient.AppsV1beta1().ControllerRevisions(namespace).Patch(name, types.StrategicMergePatchType, data)
+	return err
+}
+
 // PodControlInterface is an interface that knows how to add or delete pods
 // created as an interface to allow testing.
 type PodControlInterface interface {

pkg/controller/daemon/daemoncontroller.go

@@ -86,6 +86,7 @@ type DaemonSetsController struct {
 	kubeClient    clientset.Interface
 	eventRecorder record.EventRecorder
 	podControl    controller.PodControlInterface
+	crControl     controller.ControllerRevisionControlInterface
 
 	// An dsc is temporarily suspended after creating/deleting these many replicas.
 	// It resumes normal action after observing the watch events for them.
@@ -138,6 +139,9 @@ func NewDaemonSetsController(daemonSetInformer extensionsinformers.DaemonSetInfo
 			KubeClient: kubeClient,
 			Recorder:   eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "daemon-set"}),
 		},
+		crControl: controller.RealControllerRevisionControl{
+			KubeClient: kubeClient,
+		},
 		burstReplicas: BurstReplicas,
 		expectations:  controller.NewControllerExpectations(),
 		queue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "daemonset"),
@@ -217,7 +221,7 @@ func (dsc *DaemonSetsController) Run(workers int, stopCh <-chan struct{}) {
 	glog.Infof("Starting daemon sets controller")
 	defer glog.Infof("Shutting down daemon sets controller")
 
-	if !controller.WaitForCacheSync("daemon sets", stopCh, dsc.podStoreSynced, dsc.nodeStoreSynced, dsc.dsStoreSynced) {
+	if !controller.WaitForCacheSync("daemon sets", stopCh, dsc.podStoreSynced, dsc.nodeStoreSynced, dsc.historyStoreSynced, dsc.dsStoreSynced) {
 		return
 	}
 

pkg/controller/daemon/daemoncontroller_test.go

@@ -274,6 +274,7 @@ type daemonSetsController struct {
 	*DaemonSetsController
 
 	dsStore      cache.Store
+	historyStore cache.Store
 	podStore     cache.Store
 	nodeStore    cache.Store
 	fakeRecorder *record.FakeRecorder
@@ -297,6 +298,7 @@ func newTestController(initialObjects ...runtime.Object) (*daemonSetsController,
 	manager.podStoreSynced = alwaysReady
 	manager.nodeStoreSynced = alwaysReady
 	manager.dsStoreSynced = alwaysReady
+	manager.historyStoreSynced = alwaysReady
 	podControl := newFakePodControl()
 	manager.podControl = podControl
 	podControl.podStore = informerFactory.Core().V1().Pods().Informer().GetStore()
@@ -304,6 +306,7 @@ func newTestController(initialObjects ...runtime.Object) (*daemonSetsController,
 	return &daemonSetsController{
 		manager,
 		informerFactory.Extensions().V1beta1().DaemonSets().Informer().GetStore(),
+		informerFactory.Apps().V1beta1().ControllerRevisions().Informer().GetStore(),
 		informerFactory.Core().V1().Pods().Informer().GetStore(),
 		informerFactory.Core().V1().Nodes().Informer().GetStore(),
 		fakeRecorder,

pkg/controller/daemon/update.go

@@ -79,6 +79,7 @@ func (dsc *DaemonSetsController) rollingUpdate(ds *extensions.DaemonSet, hash st
 	return dsc.syncNodes(ds, oldPodsToDelete, []string{}, hash)
 }
 
+// constructHistory returns current history and a list of old histories of a given DaemonSet.
 func (dsc *DaemonSetsController) constructHistory(ds *extensions.DaemonSet) (cur *apps.ControllerRevision, old []*apps.ControllerRevision, err error) {
 	var histories []*apps.ControllerRevision
 	var currentHistories []*apps.ControllerRevision
@@ -272,27 +273,37 @@ func (dsc *DaemonSetsController) dedupCurHistories(ds *extensions.DaemonSet, cur
 	return keepCur, nil
 }
 
-// controlledHistories returns all ControllerRevisions controlled by the given DaemonSet
+// controlledHistories returns all ControllerRevisions controlled by the given DaemonSet.
+// This also reconciles ControllerRef by adopting/orphaning.
 // Note that returned histories are pointers to objects in the cache.
 // If you want to modify one, you need to deep-copy it first.
 func (dsc *DaemonSetsController) controlledHistories(ds *extensions.DaemonSet) ([]*apps.ControllerRevision, error) {
-	var result []*apps.ControllerRevision
 	selector, err := metav1.LabelSelectorAsSelector(ds.Spec.Selector)
 	if err != nil {
 		return nil, err
 	}
-	histories, err := dsc.historyLister.List(selector)
+
+	// List all histories to include those that don't match the selector anymore
+	// but have a ControllerRef pointing to the controller.
+	histories, err := dsc.historyLister.List(labels.Everything())
 	if err != nil {
 		return nil, err
 	}
-	for _, history := range histories {
-		// Skip history that doesn't belong to the DaemonSet
-		if controllerRef := controller.GetControllerOf(history); controllerRef == nil || controllerRef.UID != ds.UID {
-			continue
+	// If any adoptions are attempted, we should first recheck for deletion with
+	// an uncached quorum read sometime after listing Pods (see #42639).
+	canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
+		fresh, err := dsc.kubeClient.ExtensionsV1beta1().DaemonSets(ds.Namespace).Get(ds.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
 		}
-		result = append(result, history)
+		if fresh.UID != ds.UID {
+			return nil, fmt.Errorf("original DaemonSet %v/%v is gone: got uid %v, wanted %v", ds.Namespace, ds.Name, fresh.UID, ds.UID)
 		}
-	return result, nil
+		return fresh, nil
+	})
+	// Use ControllerRefManager to adopt/orphan as needed.
+	cm := controller.NewControllerRevisionControllerRefManager(dsc.crControl, ds, selector, controllerKind, canAdoptFunc)
+	return cm.ClaimControllerRevisions(histories)
 }
 
 // Match check if the given DaemonSet's template matches the template stored in the given history.

test/e2e/daemon_set.go

@@ -387,6 +387,8 @@ var _ = framework.KubeDescribe("Daemon set [Serial]", func() {
 		Expect(err).NotTo(HaveOccurred())
 		err = wait.PollImmediate(dsRetryPeriod, dsRetryTimeout, checkDaemonSetPodsOrphaned(c, ns, label))
 		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet pods to be orphaned")
+		err = wait.PollImmediate(dsRetryPeriod, dsRetryTimeout, checkDaemonSetHistoryOrphaned(c, ns, label))
+		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet history to be orphaned")
 		err = wait.PollImmediate(dsRetryPeriod, dsRetryTimeout, checkDaemonSetDeleted(f, ns, ds.Name))
 		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet to be deleted")
 
@@ -402,7 +404,9 @@ var _ = framework.KubeDescribe("Daemon set [Serial]", func() {
 
 		By(fmt.Sprintf("Wait for all pods to be adopted by DaemonSet %s", newDSName))
 		err = wait.PollImmediate(dsRetryPeriod, dsRetryTimeout, checkDaemonSetPodsAdopted(c, ns, newDS.UID, label))
-		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet pods to be orphaned")
+		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet pods to be adopted")
+		err = wait.PollImmediate(dsRetryPeriod, dsRetryTimeout, checkDaemonSetHistoryAdopted(c, ns, newDS.UID, label))
+		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet history to be adopted")
 
 		By(fmt.Sprintf("Make sure no daemon pod updated its template generation %d", templateGeneration))
 		err = checkDaemonPodsTemplateGeneration(c, ns, label, fmt.Sprint(templateGeneration))
@@ -418,6 +422,8 @@ var _ = framework.KubeDescribe("Daemon set [Serial]", func() {
 		Expect(err).NotTo(HaveOccurred())
 		err = wait.PollImmediate(dsRetryPeriod, dsRetryTimeout, checkDaemonSetPodsOrphaned(c, ns, label))
 		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet pods to be orphaned")
+		err = wait.PollImmediate(dsRetryPeriod, dsRetryTimeout, checkDaemonSetHistoryOrphaned(c, ns, label))
+		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet history to be orphaned")
 		err = wait.PollImmediate(dsRetryPeriod, dsRetryTimeout, checkDaemonSetDeleted(f, ns, newDSName))
 		Expect(err).NotTo(HaveOccurred(), "error waiting for DaemonSet to be deleted")
 
@@ -705,6 +711,19 @@ func checkDaemonSetPodsOrphaned(c clientset.Interface, ns string, label map[stri
 	}
 }
 
+func checkDaemonSetHistoryOrphaned(c clientset.Interface, ns string, label map[string]string) func() (bool, error) {
+	return func() (bool, error) {
+		histories := listDaemonHistories(c, ns, label)
+		for _, history := range histories.Items {
+			// This history is orphaned only when controller ref is cleared
+			if controllerRef := controller.GetControllerOf(&history); controllerRef != nil {
+				return false, nil
+			}
+		}
+		return true, nil
+	}
+}
+
 func checkDaemonSetPodsAdopted(c clientset.Interface, ns string, dsUID types.UID, label map[string]string) func() (bool, error) {
 	return func() (bool, error) {
 		pods := listDaemonPods(c, ns, label)
@@ -718,6 +737,19 @@ func checkDaemonSetPodsAdopted(c clientset.Interface, ns string, dsUID types.UID
 	}
 }
 
+func checkDaemonSetHistoryAdopted(c clientset.Interface, ns string, dsUID types.UID, label map[string]string) func() (bool, error) {
+	return func() (bool, error) {
+		histories := listDaemonHistories(c, ns, label)
+		for _, history := range histories.Items {
+			// This history is adopted only when its controller ref is update
+			if controllerRef := controller.GetControllerOf(&history); controllerRef == nil || controllerRef.UID != dsUID {
+				return false, nil
+			}
+		}
+		return true, nil
+	}
+}
+
 func checkDaemonSetPodsNameMatch(c clientset.Interface, ns, prefix string, label map[string]string) func() (bool, error) {
 	return func() (bool, error) {
 		if err := checkDaemonSetPodsName(c, ns, prefix, label); err != nil {