Fix truncate history bug (#364)

Author: DeliangFan

controllers/flinkcluster_observer.go

@@ -552,15 +552,11 @@ func (observer *ClusterStateObserver) truncateHistory(observed *ObservedClusterS
 		historyLimit = 10
 	}
 
-	history := make([]*appsv1.ControllerRevision, 0, len(revisions))
-	historyLen := len(history)
-	if historyLen <= historyLimit {
-		return nil
-	}
+	nonLiveHistory := getNonLiveHistory(revisions, historyLimit)
+
 	// delete any non-live history to maintain the revision limit.
-	history = history[:(historyLen - historyLimit)]
-	for i := 0; i < len(history); i++ {
-		if err := observer.history.DeleteControllerRevision(history[i]); err != nil {
+	for i := 0; i < len(nonLiveHistory); i++ {
+		if err := observer.history.DeleteControllerRevision(nonLiveHistory[i]); err != nil {
 			return err
 		}
 	}

controllers/flinkcluster_util.go

@@ -519,3 +519,17 @@ func getUpdateState(observed ObservedClusterState) UpdateState {
 	}
 	return UpdateStateUpdating
 }
+
+func getNonLiveHistory(revisions []*appsv1.ControllerRevision, historyLimit int) []*appsv1.ControllerRevision {
+
+	history := append([]*appsv1.ControllerRevision{}, revisions...)
+	nonLiveHistory := make([]*appsv1.ControllerRevision, 0)
+
+	historyLen := len(history)
+	if historyLen <= historyLimit {
+		return nonLiveHistory
+	}
+
+	nonLiveHistory = append(nonLiveHistory, history[:(historyLen-historyLimit)]...)
+	return nonLiveHistory
+}

controllers/flinkcluster_util_test.go

@@ -553,3 +553,18 @@ func TestGetFlinkAPIBaseURL(t *testing.T) {
 	apiBaseURL = getFlinkAPIBaseURL(&cluster)
 	assert.Equal(t, apiBaseURL, "http://mycluster-jobmanager.default.svc.my.domain:8004")
 }
+
+func TestGetNonLiveHistory(t *testing.T) {
+	revison0 := appsv1.ControllerRevision{Revision: int64(0)}
+	revison1 := appsv1.ControllerRevision{Revision: int64(1)}
+	revisions := []*appsv1.ControllerRevision{&revison0, &revison1}
+
+	historyLimit := 1
+	nonLiveHistory := getNonLiveHistory(revisions, historyLimit)
+	assert.Equal(t, len(nonLiveHistory), 1)
+	assert.Equal(t, nonLiveHistory[0].Revision, int64(0))
+
+	historyLimit = 3
+	nonLiveHistory = getNonLiveHistory(revisions, historyLimit)
+	assert.Equal(t, len(nonLiveHistory), 0)
+}