diff --git a/pkg/clusterconditions/mock/mock.go b/pkg/clusterconditions/mock/mock.go index f75cf5694..30edc8740 100644 --- a/pkg/clusterconditions/mock/mock.go +++ b/pkg/clusterconditions/mock/mock.go @@ -44,7 +44,7 @@ type Mock struct { } // Valid returns an error popped from ValidQueue. -func (m *Mock) Valid(ctx context.Context, condition *configv1.ClusterCondition) error { +func (m *Mock) Valid(_ context.Context, condition *configv1.ClusterCondition) error { m.Calls = append(m.Calls, Call{ When: time.Now(), Method: "Valid", @@ -61,7 +61,7 @@ func (m *Mock) Valid(ctx context.Context, condition *configv1.ClusterCondition) } // Match returns an error popped from MatchQueue. -func (m *Mock) Match(ctx context.Context, condition *configv1.ClusterCondition) (bool, error) { +func (m *Mock) Match(_ context.Context, condition *configv1.ClusterCondition) (bool, error) { m.Calls = append(m.Calls, Call{ When: time.Now(), Method: "Match", diff --git a/pkg/clusterconditions/promql/promql.go b/pkg/clusterconditions/promql/promql.go index 7a75046e8..19f20b689 100644 --- a/pkg/clusterconditions/promql/promql.go +++ b/pkg/clusterconditions/promql/promql.go @@ -50,7 +50,7 @@ func NewPromQL(kubeClient kubernetes.Interface) *cache.Cache { }, QueryTimeout: 5 * time.Minute, }, - MinBetweenMatches: 10 * time.Minute, + MinBetweenMatches: 1 * time.Second, MinForCondition: time.Hour, Expiration: 24 * time.Hour, } diff --git a/pkg/cvo/availableupdates.go b/pkg/cvo/availableupdates.go index 152ea968a..f016a526d 100644 --- a/pkg/cvo/availableupdates.go +++ b/pkg/cvo/availableupdates.go @@ -45,52 +45,91 @@ func (optr *Operator) syncAvailableUpdates(ctx context.Context, config *configv1 } // updates are only checked at most once per minimumUpdateCheckInterval or if the generation changes - u := optr.getAvailableUpdates() - if u == nil { + optrAvailableUpdates := optr.getAvailableUpdates() + needFreshFetch := true + if optrAvailableUpdates == nil { klog.V(2).Info("First attempt to retrieve available updates") - } else if !u.RecentlyChanged(optr.minimumUpdateCheckInterval) { - klog.V(2).Infof("Retrieving available updates again, because more than %s has elapsed since %s", optr.minimumUpdateCheckInterval, u.LastAttempt.Format(time.RFC3339)) - } else if channel != u.Channel { - klog.V(2).Infof("Retrieving available updates again, because the channel has changed from %q to %q", u.Channel, channel) - } else if desiredArch != u.Architecture { - klog.V(2).Infof("Retrieving available updates again, because the architecture has changed from %q to %q", u.Architecture, desiredArch) - } else if upstream == u.Upstream || (upstream == optr.defaultUpstreamServer && u.Upstream == "") { - klog.V(2).Infof("Available updates were recently retrieved, with less than %s elapsed since %s, will try later.", optr.minimumUpdateCheckInterval, u.LastAttempt.Format(time.RFC3339)) - return nil + optrAvailableUpdates = &availableUpdates{} + } else if !optrAvailableUpdates.RecentlyChanged(optr.minimumUpdateCheckInterval) { + klog.V(2).Infof("Retrieving available updates again, because more than %s has elapsed since %s", optr.minimumUpdateCheckInterval, optrAvailableUpdates.LastAttempt.Format(time.RFC3339)) + } else if channel != optrAvailableUpdates.Channel { + klog.V(2).Infof("Retrieving available updates again, because the channel has changed from %q to %q", optrAvailableUpdates.Channel, channel) + } else if desiredArch != optrAvailableUpdates.Architecture { + klog.V(2).Infof("Retrieving available updates again, because the architecture has changed from %q to %q", optrAvailableUpdates.Architecture, desiredArch) + } else if upstream == optrAvailableUpdates.Upstream || (upstream == optr.defaultUpstreamServer && optrAvailableUpdates.Upstream == "") { + needsConditionalUpdateEval := false + for _, conditionalUpdate := range optrAvailableUpdates.ConditionalUpdates { + if recommended := meta.FindStatusCondition(conditionalUpdate.Conditions, "Recommended"); recommended == nil { + needsConditionalUpdateEval = true + break + } else if recommended.Status != metav1.ConditionTrue && recommended.Status != metav1.ConditionFalse { + needsConditionalUpdateEval = true + break + } + } + if !needsConditionalUpdateEval { + klog.V(2).Infof("Available updates were recently retrieved, with less than %s elapsed since %s, will try later.", optr.minimumUpdateCheckInterval, optrAvailableUpdates.LastAttempt.Format(time.RFC3339)) + return nil + } + needFreshFetch = false } else { - klog.V(2).Infof("Retrieving available updates again, because the upstream has changed from %q to %q", u.Upstream, config.Spec.Upstream) + klog.V(2).Infof("Retrieving available updates again, because the upstream has changed from %q to %q", optrAvailableUpdates.Upstream, config.Spec.Upstream) } - transport, err := optr.getTransport() - if err != nil { - return err - } + if needFreshFetch { + transport, err := optr.getTransport() + if err != nil { + return err + } - userAgent := optr.getUserAgent() + userAgent := optr.getUserAgent() - current, updates, conditionalUpdates, condition := calculateAvailableUpdatesStatus(ctx, string(config.Spec.ClusterID), - transport, userAgent, upstream, desiredArch, currentArch, channel, optr.release.Version, optr.conditionRegistry) + current, updates, conditionalUpdates, condition := calculateAvailableUpdatesStatus(ctx, string(config.Spec.ClusterID), + transport, userAgent, upstream, desiredArch, currentArch, channel, optr.release.Version, optr.conditionRegistry) - if usedDefaultUpstream { - upstream = "" + // Populate conditions on conditional updates from operator state + for i := range optrAvailableUpdates.ConditionalUpdates { + for j := range conditionalUpdates { + if optrAvailableUpdates.ConditionalUpdates[i].Release.Image == conditionalUpdates[j].Release.Image { + conditionalUpdates[j].Conditions = optrAvailableUpdates.ConditionalUpdates[i].Conditions + break + } + } + } + + if usedDefaultUpstream { + upstream = "" + } + + optrAvailableUpdates.Upstream = upstream + optrAvailableUpdates.Channel = channel + optrAvailableUpdates.Architecture = desiredArch + optrAvailableUpdates.Current = current + optrAvailableUpdates.Updates = updates + optrAvailableUpdates.ConditionalUpdates = conditionalUpdates + optrAvailableUpdates.ConditionRegistry = optr.conditionRegistry + optrAvailableUpdates.Condition = condition } - au := &availableUpdates{ - Upstream: upstream, - Channel: config.Spec.Channel, - Architecture: desiredArch, - Current: current, - Updates: updates, - ConditionalUpdates: conditionalUpdates, - ConditionRegistry: optr.conditionRegistry, - Condition: condition, + optrAvailableUpdates.evaluateConditionalUpdates(ctx) + + queueKey := optr.queueKey() + for _, conditionalUpdate := range optrAvailableUpdates.ConditionalUpdates { + if recommended := meta.FindStatusCondition(conditionalUpdate.Conditions, "Recommended"); recommended == nil { + klog.Warningf("Requeue available-update evaluation, because %q lacks a Recommended condition", conditionalUpdate.Release.Version) + optr.availableUpdatesQueue.AddAfter(queueKey, time.Second) + break + } else if recommended.Status != metav1.ConditionTrue && recommended.Status != metav1.ConditionFalse { + klog.V(2).Infof("Requeue available-update evaluation, because %q is %s=%s: %s: %s", conditionalUpdate.Release.Version, recommended.Type, recommended.Status, recommended.Reason, recommended.Message) + optr.availableUpdatesQueue.AddAfter(queueKey, time.Second) + break + } } - au.evaluateConditionalUpdates(ctx) - optr.setAvailableUpdates(au) + optr.setAvailableUpdates(optrAvailableUpdates) - // requeue - optr.queue.Add(optr.queueKey()) + // queue optr.sync() to update ClusterVersion status + optr.queue.Add(queueKey) return nil } @@ -190,7 +229,37 @@ func (optr *Operator) setAvailableUpdates(u *availableUpdates) { func (optr *Operator) getAvailableUpdates() *availableUpdates { optr.statusLock.Lock() defer optr.statusLock.Unlock() - return optr.availableUpdates + + if optr.availableUpdates == nil { + return nil + } + + u := &availableUpdates{ + Upstream: optr.availableUpdates.Upstream, + Channel: optr.availableUpdates.Channel, + Architecture: optr.availableUpdates.Architecture, + LastAttempt: optr.availableUpdates.LastAttempt, + LastSyncOrConfigChange: optr.availableUpdates.LastSyncOrConfigChange, + Current: *optr.availableUpdates.Current.DeepCopy(), + ConditionRegistry: optr.availableUpdates.ConditionRegistry, // intentionally not a copy, to preserve cache state + Condition: optr.availableUpdates.Condition, + } + + if optr.availableUpdates.Updates != nil { + u.Updates = make([]configv1.Release, 0, len(optr.availableUpdates.Updates)) + for _, update := range optr.availableUpdates.Updates { + u.Updates = append(u.Updates, *update.DeepCopy()) + } + } + + if optr.availableUpdates.ConditionalUpdates != nil { + u.ConditionalUpdates = make([]configv1.ConditionalUpdate, 0, len(optr.availableUpdates.ConditionalUpdates)) + for _, conditionalUpdate := range optr.availableUpdates.ConditionalUpdates { + u.ConditionalUpdates = append(u.ConditionalUpdates, *conditionalUpdate.DeepCopy()) + } + } + + return u } // getArchitecture returns the currently determined cluster architecture. @@ -321,12 +390,22 @@ func (u *availableUpdates) evaluateConditionalUpdates(ctx context.Context) { Reason: "AsExpected", Message: "The update is recommended, because none of the conditional update risks apply to this cluster.", }) - u.Updates = append(u.Updates, conditionalUpdate.Release) + u.addUpdate(conditionalUpdate.Release) } u.ConditionalUpdates[i].Conditions = conditionalUpdate.Conditions } } +func (u *availableUpdates) addUpdate(release configv1.Release) { + for _, update := range u.Updates { + if update.Image == release.Image { + return + } + } + + u.Updates = append(u.Updates, release) +} + func (u *availableUpdates) removeUpdate(image string) { for i, update := range u.Updates { if update.Image == image { diff --git a/pkg/cvo/availableupdates_test.go b/pkg/cvo/availableupdates_test.go new file mode 100644 index 000000000..117ec2161 --- /dev/null +++ b/pkg/cvo/availableupdates_test.go @@ -0,0 +1,226 @@ +package cvo + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/util/workqueue" + + configv1 "github.com/openshift/api/config/v1" + "github.com/openshift/cluster-version-operator/pkg/clusterconditions" + "github.com/openshift/cluster-version-operator/pkg/clusterconditions/always" + "github.com/openshift/cluster-version-operator/pkg/clusterconditions/mock" +) + +// notFoundProxyLister is a stub for ProxyLister +type notFoundProxyLister struct{} + +func (n notFoundProxyLister) List(labels.Selector) ([]*configv1.Proxy, error) { + return nil, nil +} + +func (n notFoundProxyLister) Get(name string) (*configv1.Proxy, error) { + return nil, errors.NewNotFound(schema.GroupResource{Group: configv1.GroupName, Resource: "proxy"}, name) +} + +type notFoundConfigMapLister struct{} + +func (n notFoundConfigMapLister) List(labels.Selector) ([]*corev1.ConfigMap, error) { + return nil, nil +} + +func (n notFoundConfigMapLister) Get(name string) (*corev1.ConfigMap, error) { + return nil, errors.NewNotFound(schema.GroupResource{Group: "", Resource: "configmap"}, name) +} + +// osusWithSingleConditionalEdge helper returns: +// 1. mock osus server that serves a simple conditional path between two versions. +// 2. mock condition that always evaluates to match +// 3. expected []ConditionalUpdate data after evaluation of the data served by mock osus server +// (assuming the mock condition (2) was used) +// 4. current version of the cluster that would issue the request to the mock osus server +func osusWithSingleConditionalEdge() (*httptest.Server, clusterconditions.Condition, []configv1.ConditionalUpdate, string) { + from := "4.5.5" + to := "4.5.6" + osus := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprintf(w, `{ + "nodes": [{"version": "%s", "payload": "payload/%s"}, {"version": "%s", "payload": "payload/%s"}], + "conditionalEdges": [ + { + "edges": [{"from": "%s", "to": "%s"}], + "risks": [ + { + "url": "https://example.com/%s", + "name": "FourFiveSix", + "message": "Four Five Five is just fine", + "matchingRules": [{"type": "PromQL", "promql": { "promql": "this is a query"}}] + } + ] + } + ] +} +`, from, from, to, to, from, to, to) + })) + + updates := []configv1.ConditionalUpdate{ + { + Release: configv1.Release{Version: to, Image: "payload/" + to}, + Risks: []configv1.ConditionalUpdateRisk{ + { + URL: "https://example.com/" + to, + Name: "FourFiveSix", + Message: "Four Five Five is just fine", + MatchingRules: []configv1.ClusterCondition{ + { + Type: "PromQL", + PromQL: &configv1.PromQLClusterCondition{PromQL: "this is a query"}, + }, + }, + }, + }, + Conditions: []metav1.Condition{ + { + Type: "Recommended", + Status: metav1.ConditionFalse, + Reason: "FourFiveSix", + Message: "Four Five Five is just fine https://example.com/" + to, + }, + }, + }, + } + mockPromql := &mock.Mock{ + ValidQueue: []error{nil}, + MatchQueue: []mock.MatchResult{{Match: true, Error: nil}}, + } + + return osus, mockPromql, updates, from +} + +func newOperator(url, version string, promqlMock clusterconditions.Condition) (*availableUpdates, *Operator) { + currentRelease := configv1.Release{Version: version, Image: "payload/" + version} + registry := clusterconditions.NewConditionRegistry() + registry.Register("Always", &always.Always{}) + registry.Register("PromQL", promqlMock) + operator := &Operator{ + defaultUpstreamServer: url, + architecture: "amd64", + proxyLister: notFoundProxyLister{}, + cmConfigManagedLister: notFoundConfigMapLister{}, + conditionRegistry: registry, + queue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()), + release: currentRelease, + } + availableUpdates := &availableUpdates{ + Architecture: "amd64", + Current: configv1.Release{Version: version, Image: "payload/" + version}, + } + return availableUpdates, operator +} + +var cvFixture = &configv1.ClusterVersion{ + Spec: configv1.ClusterVersionSpec{ + ClusterID: "897f0a22-33ca-4106-a2c4-29b75250255a", + Channel: "channel", + }, +} + +var availableUpdatesCmpOpts = []cmp.Option{ + cmpopts.IgnoreTypes(time.Time{}), + cmpopts.IgnoreInterfaces(struct { + clusterconditions.ConditionRegistry + }{}), +} + +func TestSyncAvailableUpdates(t *testing.T) { + fakeOsus, mockPromql, expectedConditionalUpdates, version := osusWithSingleConditionalEdge() + defer fakeOsus.Close() + expectedAvailableUpdates, optr := newOperator(fakeOsus.URL, version, mockPromql) + expectedAvailableUpdates.ConditionalUpdates = expectedConditionalUpdates + expectedAvailableUpdates.Channel = cvFixture.Spec.Channel + expectedAvailableUpdates.Condition = configv1.ClusterOperatorStatusCondition{ + Type: configv1.RetrievedUpdates, + Status: configv1.ConditionTrue, + } + + err := optr.syncAvailableUpdates(context.Background(), cvFixture) + + if err != nil { + t.Fatalf("syncAvailableUpdates() unexpected error: %v", err) + } + if diff := cmp.Diff(expectedAvailableUpdates, optr.availableUpdates, availableUpdatesCmpOpts...); diff != "" { + t.Fatalf("available updates differ from expected:\n%s", diff) + } +} + +func TestSyncAvailableUpdates_ConditionalUpdateRecommendedConditions(t *testing.T) { + testCases := []struct { + name string + modifyOriginalState func(condition *metav1.Condition) + expectTimeChange bool + }{ + { + name: "lastTransitionTime is not updated when nothing changes", + modifyOriginalState: func(condition *metav1.Condition) {}, + }, + { + name: "lastTransitionTime is not updated when changed but status is identical", + modifyOriginalState: func(condition *metav1.Condition) { + condition.Reason = "OldReason" + condition.Message = "This message should be changed to something else" + }, + }, + { + name: "lastTransitionTime is updated when status changes", + modifyOriginalState: func(condition *metav1.Condition) { + condition.Status = metav1.ConditionUnknown + }, + expectTimeChange: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + fakeOsus, mockPromql, conditionalUpdates, version := osusWithSingleConditionalEdge() + defer fakeOsus.Close() + availableUpdates, optr := newOperator(fakeOsus.URL, version, mockPromql) + optr.availableUpdates = availableUpdates + optr.availableUpdates.ConditionalUpdates = conditionalUpdates + expectedConditions := []metav1.Condition{{}} + conditionalUpdates[0].Conditions[0].DeepCopyInto(&expectedConditions[0]) + tc.modifyOriginalState(&optr.availableUpdates.ConditionalUpdates[0].Conditions[0]) + + err := optr.syncAvailableUpdates(context.Background(), cvFixture) + + if err != nil { + t.Fatalf("syncAvailableUpdates() unexpected error: %v", err) + } + if optr.availableUpdates == nil || len(optr.availableUpdates.ConditionalUpdates) == 0 { + t.Fatalf("syncAvailableUpdates() did not properly set available updates") + } + if diff := cmp.Diff(expectedConditions, optr.availableUpdates.ConditionalUpdates[0].Conditions, cmpopts.IgnoreTypes(time.Time{})); diff != "" { + t.Errorf("conditions on conditional updates differ from expected:\n%s", diff) + } + timeBefore := expectedConditions[0].LastTransitionTime + timeAfter := optr.availableUpdates.ConditionalUpdates[0].Conditions[0].LastTransitionTime + + if tc.expectTimeChange && timeBefore == timeAfter { + t.Errorf("lastTransitionTime was not updated as expected: before=%s after=%s", timeBefore, timeAfter) + } + if !tc.expectTimeChange && timeBefore != timeAfter { + t.Errorf("lastTransitionTime was updated but was not expected to: before=%s after=%s", timeBefore, timeAfter) + } + }) + } +}