Add an exception for fast CO updates

The added exception is for the case where
* a CO does not report Progressing, and
* CVO waits for the CO's update less than 2 minutes

Author: hongkailiu

pkg/monitortests/clusterversionoperator/legacycvomonitortests/operators.go

@@ -618,10 +618,12 @@ func testUpgradeOperatorProgressingStateTransitions(events monitorapi.Intervals)
 	var eventsInUpgradeWindows monitorapi.Intervals
 
 	var start, stop time.Time
+	COWaiting := map[string]monitorapi.Intervals{}
 	for _, event := range events {
 		if !isInUpgradeWindow(upgradeWindows, event) {
 			continue
 		}
+		updateCOWaiting(event, COWaiting)
 		eventsInUpgradeWindows = append(eventsInUpgradeWindows, event)
 		if start.IsZero() || event.From.Before(start) {
 			start = event.From
@@ -651,6 +653,16 @@ func testUpgradeOperatorProgressingStateTransitions(events monitorapi.Intervals)
 	}
 
 	except := func(co string, _ string) string {
+		intervals, ok := COWaiting[co]
+		if !ok {
+			// CO have not shown up in CVO Progressing message
+			return fmt.Sprintf("%s completing its update so fast that CVO did not recogize any waiting", co)
+		}
+		from, to := fromAndTo(intervals)
+		if d := to.Sub(from); d < 2*time.Minute {
+			// CO showed up in CVO Progressing message but the total duration is less than two minutes
+			return fmt.Sprintf("%s completing its update within less than two minutes: %s", co, d.String())
+		}
 		return ""
 	}
 
@@ -668,6 +680,9 @@ func testUpgradeOperatorProgressingStateTransitions(events monitorapi.Intervals)
 			exception = except(operatorName, "")
 			if exception != "" {
 				output = fmt.Sprintf("%s which is expected up to %s", output, exception)
+			} else {
+				from, to := fromAndTo(COWaiting[operatorName])
+				output = fmt.Sprintf("%s and CVO waited for it over 2 minutes from %s to %s", output, from.Format(time.RFC3339), to.Format(time.RFC3339))
 			}
 			mcTestCase.FailureOutput = &junitapi.FailureOutput{
 				Output: output,
@@ -835,6 +850,67 @@ func testUpgradeOperatorProgressingStateTransitions(events monitorapi.Intervals)
 	return ret
 }
 
+func fromAndTo(intervals monitorapi.Intervals) (time.Time, time.Time) {
+	var from, to time.Time
+	for _, interval := range intervals {
+		if from.IsZero() || interval.From.Before(from) {
+			from = interval.From
+		}
+		if to.IsZero() || interval.To.After(to) {
+			to = interval.To
+		}
+	}
+	return from, to
+}
+
+func updateCOWaiting(interval monitorapi.Interval, waiting map[string]monitorapi.Intervals) {
+	if waiting == nil {
+		return
+	}
+	if interval.Source != monitorapi.SourceOperatorState ||
+		interval.Locator.Type != monitorapi.LocatorTypeClusterVersion ||
+		interval.Locator.Keys[monitorapi.LocatorClusterVersionKey] != "version" {
+		return
+	}
+
+	c, ok := interval.Message.Annotations[monitorapi.AnnotationCondition]
+	if !ok {
+		return
+	}
+	if t := configv1.ClusterStatusConditionType(c); t != configv1.OperatorProgressing {
+		return
+	}
+
+	status, ok := interval.Message.Annotations[monitorapi.AnnotationStatus]
+	if !ok {
+		return
+	}
+	s := configv1.ConditionStatus(status)
+	if s != configv1.ConditionTrue {
+		return
+	}
+	operators := getOperatorsFromProgressingMessage(interval.Message.HumanMessage)
+	for o := range operators {
+		waiting[o] = append(waiting[o], interval)
+	}
+	return
+}
+
+const ProgressingWaitingCOsKey = "waiting on "
+
+func getOperatorsFromProgressingMessage(message string) sets.Set[string] {
+	ret := sets.New[string]()
+	// If CVO changes the message, we have to change here accordingly
+	// https://github.com/openshift/cluster-version-operator/blob/a26c85e0fc1651645b009ee8c84b50e629fcc299/pkg/cvo/status.go#L593
+	if i := strings.LastIndex(message, ProgressingWaitingCOsKey); i == -1 {
+		return nil
+	} else {
+		ret.Insert(strings.Split(message[i+len(ProgressingWaitingCOsKey):], ", ")...)
+
+	}
+	return ret
+}
+
 type startedStaged struct {
 	// OSUpdateStarted is the event Reason emitted by the machine config operator when a node begins extracting
 	// it's OS content.

pkg/monitortests/clusterversionoperator/legacycvomonitortests/operators_test.go

@@ -4,8 +4,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/openshift/origin/pkg/monitor/monitorapi"
+	configv1 "github.com/openshift/api/config/v1"
 	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	"github.com/openshift/origin/pkg/monitor/monitorapi"
 )
 
 // buildUpgradeInterval creates a standard upgrade interval using the given reason and eventTime.
@@ -192,3 +196,104 @@ func Test_isInUpgradeWindow(t *testing.T) {
 		})
 	}
 }
+
+func Test_getOperatorsFromProgressingMessage(t *testing.T) {
+	tests := []struct {
+		name    string
+		message string
+		expect  sets.Set[string]
+	}{
+		{
+			name:    "unknown message	",
+			message: "bar foo",
+		},
+		{
+			name:    "single CO",
+			message: "working towards ${VERSION}: 106 of 841 done (12% complete), waiting on single",
+			expect:  sets.New[string]("single"),
+		},
+		{
+			name:    "multiple COs",
+			message: "working towards ${VERSION}: 106 of 841 done (12% complete), waiting on etcd, kube-apiserver",
+			expect:  sets.New[string]("kube-apiserver", "etcd"),
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			actual := getOperatorsFromProgressingMessage(tt.message)
+			assert.Equal(t, tt.expect, actual)
+		})
+	}
+}
+
+func Test_updateCOWaiting(t *testing.T) {
+	now := time.Now()
+	next := 3 * time.Hour
+	interval := func(m string, start time.Time, d time.Duration) monitorapi.Interval {
+		return monitorapi.NewInterval(monitorapi.SourceOperatorState, monitorapi.Warning).
+			Locator(monitorapi.NewLocator().ClusterVersion(&configv1.ClusterVersion{
+				ObjectMeta: metav1.ObjectMeta{Name: "version"}})).
+			Message(monitorapi.NewMessage().Reason("reason").
+				HumanMessage(m).
+				WithAnnotation(monitorapi.AnnotationCondition, string(configv1.OperatorProgressing)).
+				WithAnnotation(monitorapi.AnnotationStatus, string(configv1.ConditionTrue))).
+			Display().
+			Build(start, start.Add(d))
+	}
+
+	tests := []struct {
+		name    string
+		message string
+		d       time.Duration
+		waiting map[string]monitorapi.Intervals
+		expect  map[string]time.Duration
+	}{
+		{
+			name:    "happy case",
+			d:       time.Hour,
+			message: "working towards ${VERSION}: 106 of 841 done (12% complete), waiting on etcd, kube-apiserver",
+			waiting: map[string]monitorapi.Intervals{},
+			expect:  map[string]time.Duration{"etcd": time.Hour, "kube-apiserver": time.Hour},
+		},
+		{
+			name:    "incremental one",
+			d:       2 * time.Minute,
+			message: "working towards ${VERSION}: 106 of 841 done (12% complete), waiting on etcd, kube-apiserver",
+			waiting: map[string]monitorapi.Intervals{"etcd": {interval("some", now, 3*time.Minute)}},
+			expect:  map[string]time.Duration{"etcd": next + 2*time.Minute, "kube-apiserver": 2 * time.Minute},
+		},
+		{
+			name:    "incremental all",
+			d:       2 * time.Minute,
+			message: "working towards ${VERSION}: 106 of 841 done (12% complete), waiting on etcd, kube-apiserver",
+			waiting: map[string]monitorapi.Intervals{"etcd": {interval("some", now, 3*time.Minute)}, "kube-apiserver": {interval("some", now, 6*time.Minute)}},
+			expect:  map[string]time.Duration{"etcd": next + 2*time.Minute, "kube-apiserver": next + 2*time.Minute},
+		},
+		{
+			name:    "unknown message",
+			message: "unknown message",
+			waiting: map[string]monitorapi.Intervals{},
+			expect:  map[string]time.Duration{},
+		},
+	}
+	for _, tt := range tests {
+		i := monitorapi.NewInterval(monitorapi.SourceOperatorState, monitorapi.Warning).
+			Locator(monitorapi.NewLocator().ClusterVersion(&configv1.ClusterVersion{
+				ObjectMeta: metav1.ObjectMeta{Name: "version"}})).
+			Message(monitorapi.NewMessage().Reason("reason").
+				HumanMessage(tt.message).
+				WithAnnotation(monitorapi.AnnotationCondition, string(configv1.OperatorProgressing)).
+				WithAnnotation(monitorapi.AnnotationStatus, string(configv1.ConditionTrue))).
+			Display().
+			Build(now.Add(next), now.Add(next+tt.d))
+		t.Run(tt.name, func(t *testing.T) {
+			updateCOWaiting(i, tt.waiting)
+			actual := map[string]time.Duration{}
+			for co, intervals := range tt.waiting {
+				from, to := fromAndTo(intervals)
+				actual[co] = to.Sub(from)
+			}
+			assert.Equal(t, tt.expect, actual)
+		})
+	}
+}