fix: replace panic, update panic (#338)

* fix: replace panic, update panic

* fix: replace panic, update panic

* fix: replace panic, update panic

* fix: replace panic, update panic

Author: ColdsteelRail

pkg/controllers/collaset/synccontrol/replace.go

@@ -131,6 +131,11 @@ func (r *RealSyncControl) replaceOriginPods(
 	successCount, err := controllerutils.SlowStartBatch(len(needReplaceOriginPods), controllerutils.SlowStartInitialBatchSize, false, func(i int, _ error) error {
 		originPod := needReplaceOriginPods[i]
 		originPodId, _ := collasetutils.GetPodInstanceID(originPod)
+		if ownedIDs[originPodId] == nil {
+			r.recorder.Eventf(originPod, corev1.EventTypeWarning, "OriginPodContext", "cannot found resource context id %d of origin pod %s/%s", originPodId, originPod.Namespace, originPod.Name)
+			return fmt.Errorf("cannot found context for replace origin pod %s/%s", originPod.Namespace, originPod.Name)
+		}
+
 		ownerRef := metav1.NewControllerRef(instance, instance.GroupVersionKind())
 		updatedPDs, err := resources.PDGetter.GetEffective(ctx, originPod)
 		replaceRevision := getReplaceRevision(originPod, resources)
@@ -155,7 +160,8 @@ func (r *RealSyncControl) replaceOriginPods(
 			logger.Info("replaceOriginPods", "try to reuse new pod resourceContext id", instanceId)
 		} else {
 			if availableContexts[i] == nil {
-				return fmt.Errorf("cannot found available context for replace new pod when replacing origin pod %s/%s", originPod.Namespace, originPod.Name)
+				r.recorder.Eventf(originPod, corev1.EventTypeWarning, "AvailableContext", "cannot found available context for replace new pod when replacing origin pod %s/%s", originPod.Namespace, originPod.Name)
+				return fmt.Errorf("cannot find available context for replace new pod when replacing origin pod %s/%s", originPod.Namespace, originPod.Name)
 			}
 			newPodContext = availableContexts[i]
 			// add replace pair-relation to podContexts for originPod and newPod

pkg/controllers/collaset/synccontrol/update.go

@@ -272,12 +272,12 @@ func decidePodToUpdateByPartition(
 	filteredPodInfos []*PodUpdateInfo) []*PodUpdateInfo {
 
 	replicas := ptr.Deref(cls.Spec.Replicas, 0)
+	currentPodCount := int32(len(filteredPodInfos))
 	partition := int32(0)
 	if cls.Spec.UpdateStrategy.RollingUpdate != nil && cls.Spec.UpdateStrategy.RollingUpdate.ByPartition != nil {
 		partition = ptr.Deref(cls.Spec.UpdateStrategy.RollingUpdate.ByPartition.Partition, 0)
 	}
 
-	//filteredPodInfos := getTargetsUpdatePods(podInfos)
 	// update all or not update any replicas
 	if partition == 0 {
 		return filteredPodInfos
@@ -290,7 +290,7 @@ func decidePodToUpdateByPartition(
 	ordered := orderByDefault(filteredPodInfos)
 	sort.Sort(ordered)
 	podToUpdate := ordered[:replicas-partition]
-	for i := replicas - partition; i < replicas; i++ {
+	for i := replicas - partition; i < int32Min(replicas, currentPodCount); i++ {
 		if ordered[i].PodDecorationChanged {
 			// separate pd and collaset update progress
 			filteredPodInfos[i].IsUpdatedRevision = true
@@ -907,3 +907,11 @@ func isPodUpdatedServiceAvailable(podInfo *PodUpdateInfo) (finished bool, msg st
 
 	return false, "pod not service available", nil
 }
+
+func int32Min(l, r int32) int32 {
+	if l < r {
+		return l
+	}
+
+	return r
+}