diff --git a/pkg/reconciler/autoscaling/hpa/hpa.go b/pkg/reconciler/autoscaling/hpa/hpa.go index cd8724a871bf..3451d8655f9f 100644 --- a/pkg/reconciler/autoscaling/hpa/hpa.go +++ b/pkg/reconciler/autoscaling/hpa/hpa.go @@ -78,7 +78,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error { // This is important because the copy we loaded from the informer's // cache may be stale and we don't want to overwrite a prior update // to status with this stale state. - } else if _, err = c.UpdateStatus(pa); err != nil { + } else if err = c.UpdateStatus(original, pa); err != nil { logger.Warnw("Failed to update pa status", zap.Error(err)) c.Recorder.Eventf(pa, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status for PA %q: %v", pa.Name, err) diff --git a/pkg/reconciler/autoscaling/kpa/kpa.go b/pkg/reconciler/autoscaling/kpa/kpa.go index e57f3cda267c..9775dc3a4002 100644 --- a/pkg/reconciler/autoscaling/kpa/kpa.go +++ b/pkg/reconciler/autoscaling/kpa/kpa.go @@ -89,7 +89,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error { // This is important because the copy we loaded from the informer's // cache may be stale and we don't want to overwrite a prior update // to status with this stale state. - } else if _, err = c.UpdateStatus(pa); err != nil { + } else if err = c.UpdateStatus(original, pa); err != nil { logger.Warnw("Failed to update pa status", zap.Error(err)) c.Recorder.Eventf(pa, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status for PA %q: %v", pa.Name, err) diff --git a/pkg/reconciler/autoscaling/reconciler.go b/pkg/reconciler/autoscaling/reconciler.go index fb5ea3798bdb..1ebc03271a53 100644 --- a/pkg/reconciler/autoscaling/reconciler.go +++ b/pkg/reconciler/autoscaling/reconciler.go @@ -138,18 +138,24 @@ func (c *Base) ReconcileMetric(ctx context.Context, pa *pav1alpha1.PodAutoscaler } // UpdateStatus updates the status of the given PodAutoscaler. -func (c *Base) UpdateStatus(desired *pav1alpha1.PodAutoscaler) (*pav1alpha1.PodAutoscaler, error) { - pa, err := c.PALister.PodAutoscalers(desired.Namespace).Get(desired.Name) - if err != nil { - return nil, err - } - // If there's nothing to update, just return. - if reflect.DeepEqual(pa.Status, desired.Status) { - return pa, nil - } - // Don't modify the informers copy - existing := pa.DeepCopy() - existing.Status = desired.Status +func (c *Base) UpdateStatus(existing *pav1alpha1.PodAutoscaler, desired *pav1alpha1.PodAutoscaler) error { + existing = existing.DeepCopy() + return reconciler.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = c.ServingClientSet.AutoscalingV1alpha1().PodAutoscalers(desired.Namespace).Get(desired.Name, metav1.GetOptions{}) + if err != nil { + return err + } + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } - return c.ServingClientSet.AutoscalingV1alpha1().PodAutoscalers(pa.Namespace).UpdateStatus(existing) + existing.Status = desired.Status + _, err = c.ServingClientSet.AutoscalingV1alpha1().PodAutoscalers(existing.Namespace).UpdateStatus(existing) + return err + }) } diff --git a/pkg/reconciler/certificate/certificate.go b/pkg/reconciler/certificate/certificate.go index 1dbbf6323724..f6b248221b3a 100644 --- a/pkg/reconciler/certificate/certificate.go +++ b/pkg/reconciler/certificate/certificate.go @@ -100,7 +100,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error { // This is important because the copy we loaded from the informer's // cache may be stale and we don't want to overwrite a prior update // to status with this stale state. - } else if _, err := c.updateStatus(knCert); err != nil { + } else if err := c.updateStatus(original, knCert); err != nil { logger.Warnw("Failed to update certificate status", zap.Error(err)) c.Recorder.Eventf(knCert, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status for Certificate %s: %v", key, err) @@ -173,18 +173,24 @@ func (c *Reconciler) reconcileCMCertificate(ctx context.Context, knCert *v1alpha return cmCert, nil } -func (c *Reconciler) updateStatus(desired *v1alpha1.Certificate) (*v1alpha1.Certificate, error) { - cert, err := c.knCertificateLister.Certificates(desired.Namespace).Get(desired.Name) - if err != nil { - return nil, err - } - // If there's nothing to update, just return. - if reflect.DeepEqual(cert.Status, desired.Status) { - return cert, nil - } - // Don't modify the informers copy - existing := cert.DeepCopy() - existing.Status = desired.Status +func (c *Reconciler) updateStatus(existing *v1alpha1.Certificate, desired *v1alpha1.Certificate) error { + existing = existing.DeepCopy() + return reconciler.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = c.ServingClientSet.NetworkingV1alpha1().Certificates(desired.Namespace).Get(desired.Name, metav1.GetOptions{}) + if err != nil { + return err + } + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } - return c.ServingClientSet.NetworkingV1alpha1().Certificates(existing.Namespace).UpdateStatus(existing) + existing.Status = desired.Status + _, err = c.ServingClientSet.NetworkingV1alpha1().Certificates(existing.Namespace).UpdateStatus(existing) + return err + }) } diff --git a/pkg/reconciler/configuration/configuration.go b/pkg/reconciler/configuration/configuration.go index 3e5a8f64aa3c..0390b20b6a13 100644 --- a/pkg/reconciler/configuration/configuration.go +++ b/pkg/reconciler/configuration/configuration.go @@ -87,7 +87,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error { // This is important because the copy we loaded from the informer's // cache may be stale and we don't want to overwrite a prior update // to status with this stale state. - } else if _, err = c.updateStatus(config); err != nil { + } else if err = c.updateStatus(original, config); err != nil { logger.Warnw("Failed to update configuration status", zap.Error(err)) c.Recorder.Eventf(config, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status: %v", err) return err @@ -341,17 +341,24 @@ func (c *Reconciler) createRevision(ctx context.Context, config *v1alpha1.Config return created, nil } -func (c *Reconciler) updateStatus(desired *v1alpha1.Configuration) (*v1alpha1.Configuration, error) { - config, err := c.configurationLister.Configurations(desired.Namespace).Get(desired.Name) - if err != nil { - return nil, err - } - // If there's nothing to update, just return. - if reflect.DeepEqual(config.Status, desired.Status) { - return config, nil - } - // Don't modify the informers copy - existing := config.DeepCopy() - existing.Status = desired.Status - return c.ServingClientSet.ServingV1alpha1().Configurations(desired.Namespace).UpdateStatus(existing) +func (c *Reconciler) updateStatus(existing *v1alpha1.Configuration, desired *v1alpha1.Configuration) error { + existing = existing.DeepCopy() + return reconciler.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = c.ServingClientSet.ServingV1alpha1().Configurations(desired.Namespace).Get(desired.Name, metav1.GetOptions{}) + if err != nil { + return err + } + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } + + existing.Status = desired.Status + _, err = c.ServingClientSet.ServingV1alpha1().Configurations(desired.Namespace).UpdateStatus(existing) + return err + }) } diff --git a/pkg/reconciler/ingress/ingress.go b/pkg/reconciler/ingress/ingress.go index 1bef17039834..6b5e64796300 100644 --- a/pkg/reconciler/ingress/ingress.go +++ b/pkg/reconciler/ingress/ingress.go @@ -130,7 +130,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, key string) error { // cache may be stale and we don't want to overwrite a prior update // to status with this stale state. } else { - if _, err = r.updateStatus(ingress); err != nil { + if err = r.updateStatus(original, ingress); err != nil { logger.Warnw("Failed to update Ingress status", zap.Error(err)) r.Recorder.Eventf(ingress, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status for Ingress %q: %v", ingress.GetName(), err) @@ -306,20 +306,26 @@ func (r *Reconciler) reconcileDeletion(ctx context.Context, ia *v1alpha1.Ingress // Update the Status of the Ingress. Caller is responsible for checking // for semantic differences before calling. -func (r *Reconciler) updateStatus(desired *v1alpha1.Ingress) (*v1alpha1.Ingress, error) { - ingress, err := r.ingressLister.Ingresses(desired.GetNamespace()).Get(desired.GetName()) - if err != nil { - return nil, err - } +func (r *Reconciler) updateStatus(existing *v1alpha1.Ingress, desired *v1alpha1.Ingress) error { + existing = existing.DeepCopy() + return reconciler.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = r.ServingClientSet.NetworkingV1alpha1().Ingresses(desired.GetNamespace()).Get(desired.GetName(), metav1.GetOptions{}) + if err != nil { + return err + } + } - // If there's nothing to update, just return. - if reflect.DeepEqual(ingress.Status, desired.Status) { - return ingress, nil - } - // Don't modify the informers copy - existing := ingress.DeepCopy() - existing.Status = desired.Status - return r.ServingClientSet.NetworkingV1alpha1().Ingresses(existing.GetNamespace()).UpdateStatus(existing) + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } + + existing.Status = desired.Status + _, err = r.ServingClientSet.NetworkingV1alpha1().Ingresses(existing.GetNamespace()).UpdateStatus(existing) + return err + }) } func (r *Reconciler) ensureFinalizer(ia *v1alpha1.Ingress) error { diff --git a/pkg/reconciler/metric/metric.go b/pkg/reconciler/metric/metric.go index 6cd8f2a90944..d753b232397d 100644 --- a/pkg/reconciler/metric/metric.go +++ b/pkg/reconciler/metric/metric.go @@ -19,6 +19,7 @@ package metric import ( "context" "fmt" + "reflect" "go.uber.org/zap" "knative.dev/serving/pkg/apis/autoscaling/v1alpha1" @@ -32,6 +33,7 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/equality" apierrs "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/tools/cache" ) @@ -81,7 +83,7 @@ func (r *reconciler) Reconcile(ctx context.Context, key string) error { if !equality.Semantic.DeepEqual(original.Status, metric.Status) { // Change of status, need to update the object. - if uErr := r.updateStatus(metric); uErr != nil { + if uErr := r.updateStatus(original, metric); uErr != nil { logger.Warnw("Failed to update metric status", zap.Error(uErr)) r.Recorder.Eventf(metric, corev1.EventTypeWarning, "UpdateFailed", "Failed to update metric status: %v", uErr) @@ -102,18 +104,24 @@ func (r *reconciler) reconcileCollection(ctx context.Context, metric *v1alpha1.M return nil } -func (r *reconciler) updateStatus(m *v1alpha1.Metric) error { - ex, err := r.metricLister.Metrics(m.Namespace).Get(m.Name) - if err != nil { - // If something deleted metric while we were reconciling ¯\(°_o)/¯. +func (r *reconciler) updateStatus(existing *v1alpha1.Metric, desired *v1alpha1.Metric) error { + existing = existing.DeepCopy() + return rbase.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = r.ServingClientSet.AutoscalingV1alpha1().Metrics(desired.Namespace).Get(desired.Name, metav1.GetOptions{}) + if err != nil { + return err + } + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } + + existing.Status = desired.Status + _, err = r.ServingClientSet.AutoscalingV1alpha1().Metrics(existing.Namespace).UpdateStatus(existing) return err - } - if equality.Semantic.DeepEqual(ex.Status, m.Status) { - // no-op - return nil - } - ex = ex.DeepCopy() - ex.Status = m.Status - _, err = r.ServingClientSet.AutoscalingV1alpha1().Metrics(ex.Namespace).UpdateStatus(ex) - return err + }) } diff --git a/pkg/reconciler/retry.go b/pkg/reconciler/retry.go new file mode 100644 index 000000000000..26e99b74a40b --- /dev/null +++ b/pkg/reconciler/retry.go @@ -0,0 +1,32 @@ +/* +Copyright 2019 The Knative Authors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package reconciler + +import ( + "k8s.io/client-go/util/retry" +) + +// RetryUpdateConflicts retries the inner function if it returns conflict errors. +// This can be used to retry status updates without constantly reenqueuing keys. +func RetryUpdateConflicts(updater func(int) error) error { + attempts := 0 + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + err := updater(attempts) + attempts++ + return err + }) +} diff --git a/pkg/reconciler/retry_test.go b/pkg/reconciler/retry_test.go new file mode 100644 index 000000000000..7166ac978b7a --- /dev/null +++ b/pkg/reconciler/retry_test.go @@ -0,0 +1,80 @@ +/* +Copyright 2019 The Knative Authors + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package reconciler + +import ( + "errors" + "testing" + + apierrs "k8s.io/apimachinery/pkg/api/errors" + + v1alpha1 "knative.dev/serving/pkg/apis/autoscaling/v1alpha1" +) + +func TestRetryUpdateConflicts(t *testing.T) { + errAny := errors.New("foo") + errConflict := apierrs.NewConflict(v1alpha1.Resource("foo"), "bar", errAny) + + tests := []struct { + name string + returns []error + want error + wantAttempts int + }{{ + name: "all good", + returns: []error{nil}, + want: nil, + wantAttempts: 1, + }, { + name: "not retry on non-conflict error", + returns: []error{errAny}, + want: errAny, + wantAttempts: 1, + }, { + name: "retry up to 5 times on conflicts", + returns: []error{errConflict, errConflict, errConflict, errConflict, errConflict, errConflict}, + want: errConflict, + wantAttempts: 5, + }, { + name: "eventually succeed", + returns: []error{errConflict, errConflict, nil}, + want: nil, + wantAttempts: 3, + }, { + name: "eventually fail", + returns: []error{errConflict, errConflict, errAny}, + want: errAny, + wantAttempts: 3, + }} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + attempts := 0 + got := RetryUpdateConflicts(func(i int) error { + attempts++ + return test.returns[i] + }) + + if got != test.want { + t.Errorf("RetryUpdateConflicts() = %v, want %v", got, test.want) + } + if attempts != test.wantAttempts { + t.Errorf("attempts = %d, want %d", attempts, test.wantAttempts) + } + }) + } +} diff --git a/pkg/reconciler/revision/revision.go b/pkg/reconciler/revision/revision.go index 6eed164a37ae..14a68672eb6a 100644 --- a/pkg/reconciler/revision/revision.go +++ b/pkg/reconciler/revision/revision.go @@ -28,6 +28,7 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/equality" apierrs "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/sets" appsv1listers "k8s.io/client-go/listers/apps/v1" corev1listers "k8s.io/client-go/listers/core/v1" @@ -104,7 +105,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error { // This is important because the copy we loaded from the informer's // cache may be stale and we don't want to overwrite a prior update // to status with this stale state. - } else if _, err = c.updateStatus(rev); err != nil { + } else if err = c.updateStatus(original, rev); err != nil { logger.Warnw("Failed to update revision status", zap.Error(err)) c.Recorder.Eventf(rev, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status for Revision %q: %v", rev.Name, err) @@ -231,17 +232,24 @@ func (c *Reconciler) updateRevisionLoggingURL( "${REVISION_UID}", uid, -1) } -func (c *Reconciler) updateStatus(desired *v1alpha1.Revision) (*v1alpha1.Revision, error) { - rev, err := c.revisionLister.Revisions(desired.Namespace).Get(desired.Name) - if err != nil { - return nil, err - } - // If there's nothing to update, just return. - if reflect.DeepEqual(rev.Status, desired.Status) { - return rev, nil - } - // Don't modify the informers copy - existing := rev.DeepCopy() - existing.Status = desired.Status - return c.ServingClientSet.ServingV1alpha1().Revisions(desired.Namespace).UpdateStatus(existing) +func (c *Reconciler) updateStatus(existing *v1alpha1.Revision, desired *v1alpha1.Revision) error { + existing = existing.DeepCopy() + return reconciler.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = c.ServingClientSet.ServingV1alpha1().Revisions(desired.Namespace).Get(desired.Name, metav1.GetOptions{}) + if err != nil { + return err + } + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } + + existing.Status = desired.Status + _, err = c.ServingClientSet.ServingV1alpha1().Revisions(desired.Namespace).UpdateStatus(existing) + return err + }) } diff --git a/pkg/reconciler/route/reconcile_resources.go b/pkg/reconciler/route/reconcile_resources.go index 57b29cd2f8ed..7d085c735ea6 100644 --- a/pkg/reconciler/route/reconcile_resources.go +++ b/pkg/reconciler/route/reconcile_resources.go @@ -36,6 +36,7 @@ import ( netv1alpha1 "knative.dev/serving/pkg/apis/networking/v1alpha1" "knative.dev/serving/pkg/apis/serving" "knative.dev/serving/pkg/apis/serving/v1alpha1" + "knative.dev/serving/pkg/reconciler" "knative.dev/serving/pkg/reconciler/route/config" "knative.dev/serving/pkg/reconciler/route/resources" "knative.dev/serving/pkg/reconciler/route/traffic" @@ -187,21 +188,26 @@ func (c *Reconciler) updatePlaceholderServices(ctx context.Context, route *v1alp return eg.Wait() } -// Update the Status of the route. Caller is responsible for checking -// for semantic differences before calling. -func (c *Reconciler) updateStatus(desired *v1alpha1.Route) (*v1alpha1.Route, error) { - route, err := c.routeLister.Routes(desired.Namespace).Get(desired.Name) - if err != nil { - return nil, err - } - // If there's nothing to update, just return. - if reflect.DeepEqual(route.Status, desired.Status) { - return route, nil - } - // Don't modify the informers copy - existing := route.DeepCopy() - existing.Status = desired.Status - return c.ServingClientSet.ServingV1alpha1().Routes(desired.Namespace).UpdateStatus(existing) +func (c *Reconciler) updateStatus(existing *v1alpha1.Route, desired *v1alpha1.Route) error { + existing = existing.DeepCopy() + return reconciler.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = c.ServingClientSet.ServingV1alpha1().Routes(desired.Namespace).Get(desired.Name, metav1.GetOptions{}) + if err != nil { + return err + } + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } + + existing.Status = desired.Status + _, err = c.ServingClientSet.ServingV1alpha1().Routes(desired.Namespace).UpdateStatus(existing) + return err + }) } // Update the lastPinned annotation on revisions we target so they don't get GC'd. diff --git a/pkg/reconciler/route/route.go b/pkg/reconciler/route/route.go index 93a0f0c8b3f9..b04df2c86b7d 100644 --- a/pkg/reconciler/route/route.go +++ b/pkg/reconciler/route/route.go @@ -120,7 +120,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error { // This is important because the copy we loaded from the informer's // cache may be stale and we don't want to overwrite a prior update // to status with this stale state. - } else if _, err = c.updateStatus(route); err != nil { + } else if err = c.updateStatus(original, route); err != nil { logger.Warnw("Failed to update route status", zap.Error(err)) c.Recorder.Eventf(route, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status for Route %q: %v", route.Name, err) diff --git a/pkg/reconciler/serverlessservice/serverlessservice.go b/pkg/reconciler/serverlessservice/serverlessservice.go index bfd28340f4de..0e375d64d3a8 100644 --- a/pkg/reconciler/serverlessservice/serverlessservice.go +++ b/pkg/reconciler/serverlessservice/serverlessservice.go @@ -94,7 +94,7 @@ func (r *reconciler) Reconcile(ctx context.Context, key string) error { r.Recorder.Eventf(sks, corev1.EventTypeWarning, "UpdateFailed", "InternalError: %v", reconcileErr.Error()) } if !equality.Semantic.DeepEqual(sks.Status, original.Status) { - if _, err := r.updateStatus(sks, logger); err != nil { + if err := r.updateStatus(original, sks); err != nil { r.Recorder.Eventf(sks, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status: %v", err) return err } @@ -127,18 +127,26 @@ func (r *reconciler) reconcile(ctx context.Context, sks *netv1alpha1.ServerlessS return nil } -func (r *reconciler) updateStatus(sks *netv1alpha1.ServerlessService, logger *zap.SugaredLogger) (*netv1alpha1.ServerlessService, error) { - original, err := r.sksLister.ServerlessServices(sks.Namespace).Get(sks.Name) - if err != nil { - return nil, err - } - if reflect.DeepEqual(original.Status, sks.Status) { - return original, nil - } - logger.Debugf("StatusDiff: %s", cmp.Diff(original.Status, sks.Status)) - original = original.DeepCopy() - original.Status = sks.Status - return r.ServingClientSet.NetworkingV1alpha1().ServerlessServices(sks.Namespace).UpdateStatus(original) +func (r *reconciler) updateStatus(existing *netv1alpha1.ServerlessService, desired *netv1alpha1.ServerlessService) error { + existing = existing.DeepCopy() + return rbase.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = r.ServingClientSet.NetworkingV1alpha1().ServerlessServices(desired.Namespace).Get(desired.Name, metav1.GetOptions{}) + if err != nil { + return err + } + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } + + existing.Status = desired.Status + _, err = r.ServingClientSet.NetworkingV1alpha1().ServerlessServices(existing.Namespace).UpdateStatus(existing) + return err + }) } func (r *reconciler) reconcilePublicService(ctx context.Context, sks *netv1alpha1.ServerlessService) error { diff --git a/pkg/reconciler/service/service.go b/pkg/reconciler/service/service.go index cb4f236bcff9..3e819f884647 100644 --- a/pkg/reconciler/service/service.go +++ b/pkg/reconciler/service/service.go @@ -101,7 +101,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, key string) error { // cache may be stale and we don't want to overwrite a prior update // to status with this stale state. - } else if _, uErr := c.updateStatus(service, logger); uErr != nil { + } else if uErr := c.updateStatus(original, service, logger); uErr != nil { logger.Warnw("Failed to update service status", zap.Error(uErr)) c.Recorder.Eventf(service, corev1.EventTypeWarning, "UpdateFailed", "Failed to update status for Service %q: %v", service.Name, uErr) @@ -273,28 +273,32 @@ func (c *Reconciler) checkRoutesNotReady(config *v1alpha1.Configuration, logger } } -func (c *Reconciler) updateStatus(desired *v1alpha1.Service, logger *zap.SugaredLogger) (*v1alpha1.Service, error) { - service, err := c.serviceLister.Services(desired.Namespace).Get(desired.Name) - if err != nil { - return nil, err - } - // If there's nothing to update, just return. - if reflect.DeepEqual(service.Status, desired.Status) { - return service, nil - } - becomesReady := desired.Status.IsReady() && !service.Status.IsReady() - // Don't modify the informers copy. - existing := service.DeepCopy() - existing.Status = desired.Status - - svc, err := c.ServingClientSet.ServingV1alpha1().Services(desired.Namespace).UpdateStatus(existing) - if err == nil && becomesReady { - duration := time.Since(svc.ObjectMeta.CreationTimestamp.Time) - logger.Infof("Service became ready after %v", duration) - c.StatsReporter.ReportServiceReady(service.Namespace, service.Name, duration) - } +func (c *Reconciler) updateStatus(existing *v1alpha1.Service, desired *v1alpha1.Service, logger *zap.SugaredLogger) error { + existing = existing.DeepCopy() + return reconciler.RetryUpdateConflicts(func(attempts int) (err error) { + // The first iteration tries to use the informer's state, subsequent attempts fetch the latest state via API. + if attempts > 0 { + existing, err = c.ServingClientSet.ServingV1alpha1().Services(desired.Namespace).Get(desired.Name, metav1.GetOptions{}) + if err != nil { + return err + } + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(existing.Status, desired.Status) { + return nil + } - return svc, err + becomesReady := desired.Status.IsReady() && !existing.Status.IsReady() + existing.Status = desired.Status + _, err = c.ServingClientSet.ServingV1alpha1().Services(desired.Namespace).UpdateStatus(existing) + if err == nil && becomesReady { + duration := time.Since(existing.ObjectMeta.CreationTimestamp.Time) + logger.Infof("Service became ready after %v", duration) + c.StatsReporter.ReportServiceReady(existing.Namespace, existing.Name, duration) + } + return err + }) } func (c *Reconciler) createConfiguration(service *v1alpha1.Service) (*v1alpha1.Configuration, error) {