fix: align instanceset2 revision status (#10488)

Author: leon-ape

cmd/manager/main.go

@@ -73,6 +73,7 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/instanceset"
 	"github.com/apecloud/kubeblocks/pkg/controller/multicluster"
+	"github.com/apecloud/kubeblocks/pkg/controller/revisionmap"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	"github.com/apecloud/kubeblocks/pkg/metrics"
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
@@ -149,7 +150,7 @@ func init() {
 	viper.SetDefault(constant.CfgKeyClusterDefaultResources, `{"zero":true}`)
 	viper.SetDefault(constant.CfgKeyOperationZeroResourceForUnset, true)
 	viper.SetDefault(constant.KubernetesClusterDomainEnv, constant.DefaultDNSDomain)
-	viper.SetDefault(instanceset.MaxPlainRevisionCount, 1024)
+	viper.SetDefault(revisionmap.MaxPlainRevisionCount, 1024)
 	viper.SetDefault(instanceset.FeatureGateIgnorePodVerticalScaling, false)
 	viper.SetDefault(intctrlutil.FeatureGateEnableRuntimeMetrics, false)
 	viper.SetDefault(constant.FeatureGateIgnoreConfigTemplateDefaultMode, false)

pkg/controller/instanceset/instance_util.go

@@ -28,11 +28,9 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/klauspost/compress/zstd"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
@@ -61,19 +59,6 @@ type instanceTemplateExt struct {
 	VolumeClaimTemplates []corev1.PersistentVolumeClaim
 }
 
-var (
-	reader *zstd.Decoder
-	writer *zstd.Encoder
-)
-
-func init() {
-	var err error
-	reader, err = zstd.NewReader(nil)
-	runtime.Must(err)
-	writer, err = zstd.NewWriter(nil)
-	runtime.Must(err)
-}
-
 // parseParentNameAndOrdinal parses parent (instance template) Name and ordinal from the give instance name.
 // -1 will be returned if no numeric suffix contained.
 func parseParentNameAndOrdinal(s string) (string, int) {

pkg/controller/instanceset/revision_util.go

@@ -20,14 +20,12 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package instanceset
 
 import (
-	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"hash"
 	"hash/fnv"
 	"strconv"
 
-	jsoniter "github.com/json-iterator/go"
 	apps "k8s.io/api/apps/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -38,8 +36,7 @@ import (
 
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
-	"github.com/apecloud/kubeblocks/pkg/lru"
-	viper "github.com/apecloud/kubeblocks/pkg/viperx"
+	"github.com/apecloud/kubeblocks/pkg/controller/revisionmap"
 )
 
 // controllerRevisionHashLabel is the label used to indicate the hash value of a controllerRevision's Data.
@@ -49,8 +46,6 @@ var codecs = serializer.NewCodecFactory(model.GetScheme())
 var patchCodec = codecs.LegacyCodec(workloads.SchemeGroupVersion)
 var controllerKind = apps.SchemeGroupVersion.WithKind("StatefulSet")
 
-var jsonIter = jsoniter.ConfigCompatibleWithStandardLibrary
-
 func newRevision(its *workloads.InstanceSet) (*apps.ControllerRevision, error) {
 	patch, err := getPatch(its)
 	if err != nil {
@@ -162,46 +157,10 @@ func deepHashObject(hasher hash.Hash, objectToWrite interface{}) {
 	fmt.Fprintf(hasher, "%v", dump.ForHash(objectToWrite))
 }
 
-var revisionsCache = lru.New(1024)
-
 func GetRevisions(revisions map[string]string) (map[string]string, error) {
-	if revisions == nil {
-		return nil, nil
-	}
-	revisionsStr, ok := revisions[revisionsZSTDKey]
-	if !ok {
-		return revisions, nil
-	}
-	if revisionsInCache, ok := revisionsCache.Get(revisionsStr); ok {
-		return revisionsInCache.(map[string]string), nil
-	}
-	revisionsData, err := base64.StdEncoding.DecodeString(revisionsStr)
-	if err != nil {
-		return nil, err
-	}
-	revisionsJSON, err := reader.DecodeAll(revisionsData, nil)
-	if err != nil {
-		return nil, err
-	}
-	updateRevisions := make(map[string]string)
-
-	if err = jsonIter.Unmarshal(revisionsJSON, &updateRevisions); err != nil {
-		return nil, err
-	}
-	revisionsCache.Put(revisionsStr, updateRevisions)
-	return updateRevisions, nil
+	return revisionmap.Decode(revisions)
 }
 
 func buildRevisions(updateRevisions map[string]string) (map[string]string, error) {
-	maxPlainRevisionCount := viper.GetInt(MaxPlainRevisionCount)
-	if len(updateRevisions) <= maxPlainRevisionCount {
-		return updateRevisions, nil
-	}
-	revisionsJSON, err := jsonIter.Marshal(updateRevisions)
-	if err != nil {
-		return nil, err
-	}
-	revisionsData := writer.EncodeAll(revisionsJSON, nil)
-	revisionsStr := base64.StdEncoding.EncodeToString(revisionsData)
-	return map[string]string{revisionsZSTDKey: revisionsStr}, nil
+	return revisionmap.Encode(updateRevisions)
 }

pkg/controller/instanceset/suite_test.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/go-logr/logr"
 	"github.com/golang/mock/gomock"
+	"github.com/klauspost/compress/zstd"
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -170,6 +171,10 @@ func mockCompressedInstanceTemplates(ns, name string) (*corev1.ConfigMap, string
 	if err != nil {
 		return nil, "", err
 	}
+	writer, err := zstd.NewWriter(nil)
+	if err != nil {
+		return nil, "", err
+	}
 	templateData := writer.EncodeAll(templateByte, nil)
 	templateName := fmt.Sprintf("template-ref-%s", name)
 	templateObj := builder.NewConfigMapBuilder(ns, templateName).

pkg/controller/instanceset/types.go

@@ -31,13 +31,8 @@ const (
 )
 
 const (
-	// MaxPlainRevisionCount specified max number of plain revision stored in status.updateRevisions.
-	// All revisions will be compressed if exceeding this value.
-	MaxPlainRevisionCount = "MAX_PLAIN_REVISION_COUNT"
-
 	templateRefAnnotationKey = "kubeblocks.io/template-ref"
 	templateRefDataKey       = "instances"
-	revisionsZSTDKey         = "zstd"
 
 	FeatureGateIgnorePodVerticalScaling = "IGNORE_POD_VERTICAL_SCALING"
 

pkg/controller/instanceset2/instance_util.go

@@ -38,6 +38,7 @@ import (
 	"github.com/apecloud/kubeblocks/pkg/controller/instancetemplate"
 	"github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/apecloud/kubeblocks/pkg/controller/revisionmap"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 )
 
@@ -325,6 +326,37 @@ func getHeadlessSvcName(itsName string) string {
 	return strings.Join([]string{itsName, "headless"}, "-")
 }
 
+func buildDesiredInstancesByName(tree *kubebuilderx.ObjectTree, its *workloads.InstanceSet) (map[string]*workloads.Instance, []string, error) {
+	itsExt, err := instancetemplate.BuildInstanceSetExt(its, tree)
+	if err != nil {
+		return nil, nil, err
+	}
+	nameBuilder, err := instancetemplate.NewPodNameBuilder(itsExt, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	nameMap, err := nameBuilder.BuildInstanceName2TemplateMap()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	names := make([]string, 0, len(nameMap))
+	for name := range nameMap {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	desired := make(map[string]*workloads.Instance, len(names))
+	for _, name := range names {
+		inst, err := buildInstanceByTemplate(tree, name, nameMap[name], its)
+		if err != nil {
+			return nil, nil, err
+		}
+		desired[name] = inst
+	}
+	return desired, names, nil
+}
+
 // func mergeInPlaceFields(src, dst *corev1.PodTemplateSpec) {
 //	mergeMap(&src.Annotations, &dst.Annotations)
 //	mergeMap(&src.Labels, &dst.Labels)
@@ -393,12 +425,17 @@ func getHeadlessSvcName(itsName string) string {
 //	return requests, limits
 // }
 
-func isInstanceUpdated(its *workloads.InstanceSet, inst *workloads.Instance) bool {
-	generation, ok := inst.Annotations[constant.KubeBlocksGenerationKey]
-	if !ok {
+func isInstanceUpdated(its *workloads.InstanceSet, inst, desired *workloads.Instance) bool {
+	updateRevisions, err := revisionmap.Decode(its.Status.UpdateRevisions)
+	if err != nil {
 		return false
 	}
-	if strconv.FormatInt(its.Generation, 10) != generation {
+	return isInstanceUpdatedWithRevisions(inst, buildCurrentInstanceRevision(inst, desired), updateRevisions)
+}
+
+func isInstanceUpdatedWithRevisions(inst *workloads.Instance, currentRevision string, updateRevisions map[string]string) bool {
+	updateRevision, ok := updateRevisions[inst.Name]
+	if !ok || currentRevision != updateRevision {
 		return false
 	}
 	return inst.Generation == inst.Status.ObservedGeneration && inst.Status.UpToDate

pkg/controller/instanceset2/reconciler_revision_update.go

@@ -25,6 +25,7 @@ import (
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
 	"github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/apecloud/kubeblocks/pkg/controller/revisionmap"
 )
 
 func NewRevisionUpdateReconciler() kubebuilderx.Reconciler {
@@ -45,19 +46,37 @@ func (r *revisionUpdateReconciler) PreCondition(tree *kubebuilderx.ObjectTree) *
 func (r *revisionUpdateReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilderx.Result, error) {
 	its, _ := tree.GetRoot().(*workloads.InstanceSet)
 
-	updatedReplicas := r.calculateUpdatedReplicas(its, tree.List(&workloads.Instance{}))
+	desiredInstances, names, err := buildDesiredInstancesByName(tree, its)
+	if err != nil {
+		return kubebuilderx.Continue, err
+	}
+
+	updateRevisions := make(map[string]string, len(names))
+	for _, name := range names {
+		updateRevisions[name] = buildInstanceRevision(desiredInstances[name])
+	}
+	revisions, err := revisionmap.Encode(updateRevisions)
+	if err != nil {
+		return kubebuilderx.Continue, err
+	}
+	its.Status.UpdateRevisions = revisions
+	if len(names) > 0 {
+		its.Status.UpdateRevision = updateRevisions[names[len(names)-1]]
+	}
+
+	updatedReplicas := r.calculateUpdatedReplicas(its, tree.List(&workloads.Instance{}), desiredInstances)
 	its.Status.UpdatedReplicas = updatedReplicas
 
 	its.Status.ObservedGeneration = its.Generation
 
 	return kubebuilderx.Continue, nil
 }
 
-func (r *revisionUpdateReconciler) calculateUpdatedReplicas(its *workloads.InstanceSet, instances []client.Object) int32 {
+func (r *revisionUpdateReconciler) calculateUpdatedReplicas(its *workloads.InstanceSet, instances []client.Object, desiredInstances map[string]*workloads.Instance) int32 {
 	updatedReplicas := int32(0)
 	for i := range instances {
 		inst, _ := instances[i].(*workloads.Instance)
-		if isInstanceUpdated(its, inst) {
+		if isInstanceUpdated(its, inst, desiredInstances[inst.Name]) {
 			updatedReplicas++
 		}
 	}

pkg/controller/instanceset2/reconciler_status.go

@@ -29,9 +29,11 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
+	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/instancetemplate"
 	"github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/apecloud/kubeblocks/pkg/controller/revisionmap"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 )
 
@@ -53,6 +55,11 @@ func (r *statusReconciler) PreCondition(tree *kubebuilderx.ObjectTree) *kubebuil
 func (r *statusReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilderx.Result, error) {
 	its, _ := tree.GetRoot().(*workloads.InstanceSet)
 
+	desiredInstances, _, err := buildDesiredInstancesByName(tree, its)
+	if err != nil {
+		return kubebuilderx.Continue, err
+	}
+
 	instances := tree.List(&workloads.Instance{})
 	var instanceList []*workloads.Instance
 	for _, object := range instances {
@@ -65,7 +72,11 @@ func (r *statusReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 	readyReplicas, availableReplicas := int32(0), int32(0)
 	notReadyNames := sets.New[string]()
 	notAvailableNames := sets.New[string]()
-	// currentRevisions := map[string]string{}
+	currentRevisions := map[string]string{}
+	updateRevisions, err := revisionmap.Decode(its.Status.UpdateRevisions)
+	if err != nil {
+		return kubebuilderx.Continue, err
+	}
 
 	template2TemplatesStatus := map[string]*workloads.InstanceTemplateStatus{}
 	template2TotalReplicas := map[string]int32{}
@@ -78,7 +89,7 @@ func (r *statusReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 	}
 
 	for _, inst := range instanceList {
-		templateName := inst.Labels[instancetemplate.TemplateNameLabelKey]
+		templateName := getInstanceTemplateName(inst)
 		if template2TemplatesStatus[templateName] == nil {
 			template2TemplatesStatus[templateName] = &workloads.InstanceTemplateStatus{
 				Name: templateName,
@@ -100,8 +111,9 @@ func (r *statusReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 				notAvailableNames.Insert(inst.Name)
 			}
 		}
+		currentRevisions[inst.Name] = buildCurrentInstanceRevision(inst, desiredInstances[inst.Name])
 		if !intctrlutil.IsInstanceTerminating(inst) {
-			if isInstanceUpdated(its, inst) {
+			if isInstanceUpdatedWithRevisions(inst, currentRevisions[inst.Name], updateRevisions) {
 				updatedReplicas++
 				template2TemplatesStatus[templateName].UpdatedReplicas++
 			} else {
@@ -115,15 +127,15 @@ func (r *statusReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 	its.Status.AvailableReplicas = availableReplicas
 	its.Status.CurrentReplicas = currentReplicas
 	its.Status.UpdatedReplicas = updatedReplicas
-	// its.Status.CurrentRevisions, _ = buildRevisions(currentRevisions)
+	its.Status.CurrentRevisions, _ = revisionmap.Encode(currentRevisions)
 	its.Status.TemplatesStatus = buildTemplatesStatus(template2TemplatesStatus)
 	// all pods have been updated
 	totalReplicas := int32(1)
 	if its.Spec.Replicas != nil {
 		totalReplicas = *its.Spec.Replicas
 	}
 	if its.Status.Replicas == totalReplicas && its.Status.UpdatedReplicas == totalReplicas {
-		// its.Status.CurrentRevision = its.Status.UpdateRevision
+		its.Status.CurrentRevision = its.Status.UpdateRevision
 		its.Status.CurrentReplicas = totalReplicas
 	}
 	for idx, templateStatus := range its.Status.TemplatesStatus {
@@ -165,6 +177,16 @@ func (r *statusReconciler) Reconcile(tree *kubebuilderx.ObjectTree) (kubebuilder
 	return kubebuilderx.Continue, nil
 }
 
+func getInstanceTemplateName(inst *workloads.Instance) string {
+	if inst.Labels == nil {
+		return ""
+	}
+	if templateName := inst.Labels[instancetemplate.TemplateNameLabelKey]; templateName != "" {
+		return templateName
+	}
+	return inst.Labels[constant.KBAppInstanceTemplateLabelKey]
+}
+
 func buildConditionMessageWithNames(instanceNames []string) ([]byte, error) {
 	baseSort(instanceNames, func(i int) (string, int) {
 		return parseParentNameAndOrdinal(instanceNames[i])