enhancement (CollaSet): fix linter

pkg/controllers/collaset/collaset_controller.go

@@ -118,12 +118,11 @@ func (r *CollaSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	if err := r.Get(ctx, req.NamespacedName, instance); err != nil {
 		if !errors.IsNotFound(err) {
 			klog.Error("fail to find CollaSet %s: %s", req, err)
-			collasetutils.ActiveExpectations.Delete(req.Namespace, req.Name)
 			return reconcile.Result{}, err
 		}
 
 		klog.Infof("CollaSet %s is deleted", req)
-		return ctrl.Result{}, nil
+		return ctrl.Result{}, collasetutils.ActiveExpectations.Delete(req.Namespace, req.Name)
 	}
 
 	// if expectation not satisfied, shortcut this reconciling till informer cache is updated.
@@ -134,7 +133,7 @@ func (r *CollaSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, nil
 	}
 
-	newStatus, err := DoReconcile(ctx, instance)
+	newStatus, err := DoReconcile(instance)
 	if err != nil {
 		return ctrl.Result{}, err
 	}
@@ -146,7 +145,7 @@ func (r *CollaSetReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	return ctrl.Result{}, nil
 }
 
-func DoReconcile(ctx context.Context, instance *appsv1alpha1.CollaSet) (*appsv1alpha1.CollaSetStatus, error) {
+func DoReconcile(instance *appsv1alpha1.CollaSet) (*appsv1alpha1.CollaSetStatus, error) {
 	newStatus := instance.Status.DeepCopy()
 
 	currentRevision, updatedRevision, revisions, collisionCount, _, err := revisionManager.ConstructRevisions(instance, false)

pkg/controllers/collaset/collaset_controller_test.go

@@ -20,9 +20,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"k8s.io/apimachinery/pkg/util/sets"
-	"kusionstack.io/kafed/pkg/controllers/collaset/synccontrol"
-	"kusionstack.io/kafed/pkg/controllers/utils/podopslifecycle"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -34,6 +31,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
@@ -47,7 +45,9 @@ import (
 
 	"kusionstack.io/kafed/apis"
 	appsv1alpha1 "kusionstack.io/kafed/apis/apps/v1alpha1"
+	"kusionstack.io/kafed/pkg/controllers/collaset/synccontrol"
 	collasetutils "kusionstack.io/kafed/pkg/controllers/collaset/utils"
+	"kusionstack.io/kafed/pkg/controllers/utils/podopslifecycle"
 	"kusionstack.io/kafed/pkg/utils/inject"
 )
 

pkg/controllers/collaset/revision.go

@@ -90,6 +90,6 @@ func (roa *revisionOwnerAdapter) GetCurrentRevision(obj metav1.Object) string {
 	return ips.Status.CurrentRevision
 }
 
-func (roa *revisionOwnerAdapter) IsInUsed(obj metav1.Object, controllerRevision string) bool {
+func (roa *revisionOwnerAdapter) IsInUsed(_ metav1.Object, _ string) bool {
 	return false
 }

pkg/controllers/collaset/synccontrol/scale.go

@@ -19,13 +19,12 @@ package synccontrol
 import (
 	"sort"
 
-	appsv1alpha1 "kusionstack.io/kafed/apis/apps/v1alpha1"
 	collasetutils "kusionstack.io/kafed/pkg/controllers/collaset/utils"
 	controllerutils "kusionstack.io/kafed/pkg/controllers/utils"
 	"kusionstack.io/kafed/pkg/controllers/utils/podopslifecycle"
 )
 
-func getPodsToDelete(filteredPods []*collasetutils.PodWrapper, ownedIDs map[int]*appsv1alpha1.ContextDetail, diff int) []*collasetutils.PodWrapper {
+func getPodsToDelete(filteredPods []*collasetutils.PodWrapper, diff int) []*collasetutils.PodWrapper {
 	sort.Sort(ActivePodsForDeletion(filteredPods))
 	start := len(filteredPods) - diff
 	if start < 0 {

pkg/controllers/collaset/synccontrol/sync_control.go

@@ -61,7 +61,7 @@ type RealSyncControl struct {
 }
 
 // SyncPods is used to reclaim Pod instance ID
-func (sc *RealSyncControl) SyncPods(instance *appsv1alpha1.CollaSet, filteredPods []*corev1.Pod, updatedRevision *appsv1.ControllerRevision, newStatus *appsv1alpha1.CollaSetStatus) (bool, []*collasetutils.PodWrapper, map[int]*appsv1alpha1.ContextDetail, error) {
+func (sc *RealSyncControl) SyncPods(instance *appsv1alpha1.CollaSet, filteredPods []*corev1.Pod, updatedRevision *appsv1.ControllerRevision, _ *appsv1alpha1.CollaSetStatus) (bool, []*collasetutils.PodWrapper, map[int]*appsv1alpha1.ContextDetail, error) {
 	// get owned IDs
 	var ownedIDs map[int]*appsv1alpha1.ContextDetail
 	var err error
@@ -174,17 +174,15 @@ func (sc *RealSyncControl) Scale(set *appsv1alpha1.CollaSet, podWrappers []*coll
 
 		if err != nil {
 			collasetutils.AddOrUpdateCondition(newStatus, appsv1alpha1.CollaSetScale, err, "ScaleOutFailed", err.Error())
-			return scaling, err
+			return succCount > 0, err
 		} else {
 			collasetutils.AddOrUpdateCondition(newStatus, appsv1alpha1.CollaSetScale, nil, "ScaleOut", "")
 		}
 
-		// TODO record pod current revision
-
-		return succCount != 0, err
+		return succCount > 0, err
 	} else if diff < 0 {
 		// chose the pods to scale in
-		podsToScaleIn := getPodsToDelete(podWrappers, ownedIDs, diff*-1)
+		podsToScaleIn := getPodsToDelete(podWrappers, diff*-1)
 		// filter out Pods need to trigger PodOpsLifecycle
 		podCh := make(chan *collasetutils.PodWrapper, len(podsToScaleIn))
 		for i := range podsToScaleIn {

pkg/controllers/collaset/synccontrol/update.go

@@ -85,7 +85,7 @@ func decidePodToUpdate(cls *appsv1alpha1.CollaSet, podInfos []*PodUpdateInfo) []
 	return decidePodToUpdateByPartition(cls, podInfos)
 }
 
-func decidePodToUpdateByLabel(cls *appsv1alpha1.CollaSet, podInfos []*PodUpdateInfo) (podToUpdate []*PodUpdateInfo) {
+func decidePodToUpdateByLabel(_ *appsv1alpha1.CollaSet, podInfos []*PodUpdateInfo) (podToUpdate []*PodUpdateInfo) {
 	for i := range podInfos {
 		if _, exist := podInfos[i].Labels[appsv1alpha1.CollaSetUpdateIndicateLabelKey]; exist {
 			podToUpdate = append(podToUpdate, podInfos[i])
@@ -182,10 +182,9 @@ func (u *InPlaceIfPossibleUpdater) AnalyseAndGetUpdatedPod(cls *appsv1alpha1.Col
 
 	// 2. compare current and updated pods. Only pod image and metadata are supported to update in-place
 	// TODO: use cache
-	inPlaceSetUpdateSupport := true
-	inPlaceSetUpdateSupport, onlyMetadataChanged = u.diffPod(currentPod, updatedPod)
+	inPlaceUpdateSupport, onlyMetadataChanged = u.diffPod(currentPod, updatedPod)
 	// 2.1 if pod has changes more than metadata and image
-	if !inPlaceSetUpdateSupport {
+	if !inPlaceUpdateSupport {
 		return false, onlyMetadataChanged, nil, nil
 	}
 
@@ -221,7 +220,7 @@ func (u *InPlaceIfPossibleUpdater) AnalyseAndGetUpdatedPod(cls *appsv1alpha1.Col
 
 		podStatusStr, err := json.Marshal(podStatus)
 		if err != nil {
-			return inPlaceSetUpdateSupport, onlyMetadataChanged, updatedPod, err
+			return inPlaceUpdateSupport, onlyMetadataChanged, updatedPod, err
 		}
 
 		if updatedPod.Annotations == nil {
@@ -329,23 +328,23 @@ func (u *InPlaceIfPossibleUpdater) GetPodUpdateFinishStatus(pod *corev1.Pod) (fi
 type InPlaceOnlyPodUpdater struct {
 }
 
-func (u *InPlaceOnlyPodUpdater) AnalyseAndGetUpdatedPod(cls *appsv1alpha1.CollaSet, updatedRevision *appsv1.ControllerRevision, podUpdateInfo *PodUpdateInfo) (inPlaceUpdateSupport bool, onlyMetadataChanged bool, updatedPod *corev1.Pod, err error) {
+func (u *InPlaceOnlyPodUpdater) AnalyseAndGetUpdatedPod(_ *appsv1alpha1.CollaSet, _ *appsv1.ControllerRevision, _ *PodUpdateInfo) (inPlaceUpdateSupport bool, onlyMetadataChanged bool, updatedPod *corev1.Pod, err error) {
 
 	return
 }
 
-func (u *InPlaceOnlyPodUpdater) GetPodUpdateFinishStatus(pod *corev1.Pod) (finished bool, msg string, err error) {
+func (u *InPlaceOnlyPodUpdater) GetPodUpdateFinishStatus(_ *corev1.Pod) (finished bool, msg string, err error) {
 	return
 }
 
 type RecreatePodUpdater struct {
 }
 
-func (u *RecreatePodUpdater) AnalyseAndGetUpdatedPod(cls *appsv1alpha1.CollaSet, updatedRevision *appsv1.ControllerRevision, podUpdateInfo *PodUpdateInfo) (inPlaceUpdateSupport bool, onlyMetadataChanged bool, updatedPod *corev1.Pod, err error) {
+func (u *RecreatePodUpdater) AnalyseAndGetUpdatedPod(_ *appsv1alpha1.CollaSet, _ *appsv1.ControllerRevision, _ *PodUpdateInfo) (inPlaceUpdateSupport bool, onlyMetadataChanged bool, updatedPod *corev1.Pod, err error) {
 	return false, false, nil, nil
 }
 
-func (u *RecreatePodUpdater) GetPodUpdateFinishStatus(pod *corev1.Pod) (finished bool, msg string, err error) {
+func (u *RecreatePodUpdater) GetPodUpdateFinishStatus(_ *corev1.Pod) (finished bool, msg string, err error) {
 	// Recreate policy alway treat Pod as update finished
 	return true, "", nil
 }

pkg/controllers/collaset/utils/lifecycle_adapter.go

@@ -51,7 +51,7 @@ func (a *CollaSetUpdateOpsLifecycleAdapter) AllowMultiType() bool {
 }
 
 // WhenBegin will be executed when begin a lifecycle
-func (a *CollaSetUpdateOpsLifecycleAdapter) WhenBegin(pod client.Object) (bool, error) {
+func (a *CollaSetUpdateOpsLifecycleAdapter) WhenBegin(_ client.Object) (bool, error) {
 	return false, nil
 }
 
@@ -103,6 +103,6 @@ func (a *CollaSetScaleInOpsLifecycleAdapter) WhenBegin(pod client.Object) (bool,
 }
 
 // WhenFinish will be executed when finish a lifecycle
-func (a *CollaSetScaleInOpsLifecycleAdapter) WhenFinish(pod client.Object) (bool, error) {
+func (a *CollaSetScaleInOpsLifecycleAdapter) WhenFinish(_ client.Object) (bool, error) {
 	return false, nil
 }

pkg/controllers/poddeletion/lifecycle_adapter.go

@@ -46,12 +46,12 @@ func (a *PodDeleteOpsLifecycleAdapter) AllowMultiType() bool {
 }
 
 // WhenBegin will be executed when begin a lifecycle
-func (a *PodDeleteOpsLifecycleAdapter) WhenBegin(pod client.Object) (bool, error) {
+func (a *PodDeleteOpsLifecycleAdapter) WhenBegin(_ client.Object) (bool, error) {
 	return false, nil
 }
 
 // WhenFinish will be executed when finish a lifecycle
-func (a *PodDeleteOpsLifecycleAdapter) WhenFinish(pod client.Object) (bool, error) {
+func (a *PodDeleteOpsLifecycleAdapter) WhenFinish(_ client.Object) (bool, error) {
 
 	return false, nil
 }

pkg/controllers/poddeletion/poddeletion_controller.go

@@ -19,10 +19,8 @@ package poddeletion
 import (
 	"context"
 	"fmt"
-	corev1 "k8s.io/api/core/v1"
-	"kusionstack.io/kafed/pkg/controllers/utils/expectations"
-	"kusionstack.io/kafed/pkg/controllers/utils/podopslifecycle"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
@@ -34,6 +32,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	"kusionstack.io/kafed/pkg/controllers/collaset/utils"
+	"kusionstack.io/kafed/pkg/controllers/utils/expectations"
+	"kusionstack.io/kafed/pkg/controllers/utils/podopslifecycle"
 )
 
 const (
@@ -88,12 +88,11 @@ func (r *PodDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	if err := r.Get(ctx, req.NamespacedName, instance); err != nil {
 		if !errors.IsNotFound(err) {
 			klog.Error("fail to find Pod %s: %s", req, err)
-			utils.ActiveExpectations.Delete(req.Namespace, req.Name)
 			return reconcile.Result{}, err
 		}
 
 		klog.Infof("Pod %s is deleted", req)
-		return ctrl.Result{}, nil
+		return ctrl.Result{}, utils.ActiveExpectations.Delete(req.Namespace, req.Name)
 	}
 
 	// if expectation not satisfied, shortcut this reconciling till informer cache is updated.

pkg/controllers/poddeletion/poddeletion_controller_test.go

@@ -19,29 +19,30 @@ package poddeletion
 import (
 	"context"
 	"fmt"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
-	appsv1alpha1 "kusionstack.io/kafed/apis/apps/v1alpha1"
-	"net/url"
-	"os"
-	"path/filepath"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-	"strings"
-	"testing"
-	"time"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
 	"kusionstack.io/kafed/apis"
+	appsv1alpha1 "kusionstack.io/kafed/apis/apps/v1alpha1"
 	"kusionstack.io/kafed/pkg/utils/inject"
 )
 

pkg/controllers/resourcecontext/expectation.go

@@ -18,6 +18,7 @@ package resourcecontext
 
 import (
 	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 
@@ -43,8 +44,10 @@ func (h *ExpectationEventHandler) Create(event.CreateEvent, workqueue.RateLimiti
 func (h *ExpectationEventHandler) Update(event.UpdateEvent, workqueue.RateLimitingInterface) {}
 
 // Delete is called in response to a delete event - e.g. Pod Deleted.
-func (h *ExpectationEventHandler) Delete(e event.DeleteEvent, q workqueue.RateLimitingInterface) {
-	activeExpectations.Delete(e.Object.GetNamespace(), e.Object.GetName())
+func (h *ExpectationEventHandler) Delete(e event.DeleteEvent, _ workqueue.RateLimitingInterface) {
+	if err := activeExpectations.Delete(e.Object.GetNamespace(), e.Object.GetName()); err != nil {
+		klog.Error("fail to delete expectation in ResourceContextController for %s/%s: %s", e.Object.GetNamespace(), e.Object.GetName(), err)
+	}
 }
 
 // Generic is called in response to an event of an unknown type or a synthetic event triggered as a cron or

pkg/controllers/resourcecontext/resourcecontext_controller.go

@@ -18,7 +18,6 @@ package resourcecontext
 
 import (
 	"context"
-	"kusionstack.io/kafed/pkg/controllers/utils/expectations"
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/tools/record"
@@ -32,6 +31,7 @@ import (
 
 	appsv1alpha1 "kusionstack.io/kafed/apis/apps/v1alpha1"
 	"kusionstack.io/kafed/pkg/controllers/collaset/utils"
+	"kusionstack.io/kafed/pkg/controllers/utils/expectations"
 )
 
 const (
@@ -92,12 +92,11 @@ func (r *ResourceContextReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	if err := r.Get(ctx, req.NamespacedName, instance); err != nil {
 		if !errors.IsNotFound(err) {
 			klog.Error("fail to find ResourceContext %s: %s", req, err)
-			utils.ActiveExpectations.Delete(req.Namespace, req.Name)
 			return reconcile.Result{}, err
 		}
 
 		klog.Infof("ResourceContext %s is deleted", req)
-		return ctrl.Result{}, nil
+		return ctrl.Result{}, utils.ActiveExpectations.Delete(req.Namespace, req.Name)
 	}
 
 	// if expectation not satisfied, shortcut this reconciling till informer cache is updated.

pkg/controllers/resourcecontext/resourcecontext_controller_test.go

@@ -19,17 +19,15 @@ package resourcecontext
 import (
 	"context"
 	"fmt"
-	"k8s.io/apimachinery/pkg/util/sets"
-	"kusionstack.io/kafed/pkg/controllers/collaset"
-	collasetutils "kusionstack.io/kafed/pkg/controllers/collaset/utils"
-	"kusionstack.io/kafed/pkg/controllers/poddeletion"
 	"net/url"
 	"os"
 	"path/filepath"
 	"strings"
 	"testing"
 	"time"
 
+	"k8s.io/apimachinery/pkg/util/sets"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -47,6 +45,9 @@ import (
 
 	"kusionstack.io/kafed/apis"
 	appsv1alpha1 "kusionstack.io/kafed/apis/apps/v1alpha1"
+	"kusionstack.io/kafed/pkg/controllers/collaset"
+	collasetutils "kusionstack.io/kafed/pkg/controllers/collaset/utils"
+	"kusionstack.io/kafed/pkg/controllers/poddeletion"
 	"kusionstack.io/kafed/pkg/utils/inject"
 )
 

pkg/controllers/utils/expectations/expectations_suite_test.go

@@ -1,5 +1,5 @@
 /*
-Copyright YEAR The Kubernetes Authors.
+Copyright 2023 The KusionStack Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

pkg/controllers/utils/expectations/resourceversion_expectation.go

@@ -92,8 +92,9 @@ func (r *ResourceVersionExpectation) ExpectUpdate(controllerKey string, resource
 	} else if exists {
 		exp.Set(resourceVersion)
 	} else {
-		r.SetExpectations(controllerKey, resourceVersion)
+		return r.SetExpectations(controllerKey, resourceVersion)
 	}
+
 	return nil
 }