pkg/cvo: allow sync task for a manifest to be requeued on special failures

lib/manifest.go

@@ -2,11 +2,11 @@ package lib
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
 	"io"
 	"os"
 
+	"github.com/pkg/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -43,7 +43,7 @@ func (m *Manifest) UnmarshalJSON(in []byte) error {
 	m.Raw = append(m.Raw[0:0], in...)
 	udi, _, err := scheme.Codecs.UniversalDecoder().Decode(in, nil, &unstructured.Unstructured{})
 	if err != nil {
-		return fmt.Errorf("unable to decode manifest: %v", err)
+		return errors.Wrapf(err, "unable to decode manifest")
 	}
 	ud, ok := udi.(*unstructured.Unstructured)
 	if !ok {
@@ -66,14 +66,14 @@ func ManifestsFromFiles(files []string) ([]Manifest, error) {
 	for _, file := range files {
 		file, err := os.Open(file)
 		if err != nil {
-			errs = append(errs, fmt.Errorf("error opening %s: %v", file.Name(), err))
+			errs = append(errs, errors.Wrapf(err, "error opening %s", file.Name()))
 			continue
 		}
 		defer file.Close()
 
 		ms, err := ParseManifests(file)
 		if err != nil {
-			errs = append(errs, fmt.Errorf("error parsing %s: %v", file.Name(), err))
+			errs = append(errs, errors.Wrapf(err, "error parsing %s", file.Name()))
 			continue
 		}
 		manifests = append(manifests, ms...)
@@ -98,7 +98,7 @@ func ParseManifests(r io.Reader) ([]Manifest, error) {
 			if err == io.EOF {
 				return manifests, nil
 			}
-			return manifests, fmt.Errorf("error parsing: %v", err)
+			return manifests, errors.Wrapf(err, "error parsing")
 		}
 		m.Raw = bytes.TrimSpace(m.Raw)
 		if len(m.Raw) == 0 || bytes.Equal(m.Raw, []byte("null")) {

pkg/cvo/image.go

@@ -1,13 +1,17 @@
 package cvo
 
-import "fmt"
+import (
+	"fmt"
+
+	"github.com/pkg/errors"
+)
 
 // ImageForShortName returns the image using the updatepayload embedded in
 // the Operator.
 func ImageForShortName(name string) (string, error) {
 	up, err := loadUpdatePayload(defaultUpdatePayloadDir, "")
 	if err != nil {
-		return "", fmt.Errorf("error loading update payload from %q: %v", defaultUpdatePayloadDir, err)
+		return "", errors.Wrapf(err, "error loading update payload from %q", defaultUpdatePayloadDir)
 	}
 
 	for _, tag := range up.imageRef.Spec.Tags {

pkg/cvo/internal/dynamicclient/client.go

@@ -1,7 +1,6 @@
 package dynamicclient
 
 import (
-	"fmt"
 	"sync"
 	"time"
 
@@ -68,7 +67,7 @@ func (c *resourceClientFactory) getResourceClient(gvk schema.GroupVersionKind, n
 		gvr, namespaced, err = gvkToGVR(gvk, c.restMapper)
 	}
 	if err != nil {
-		return nil, errors.WithMessage(err, fmt.Sprintf("failed to get resource type: %v", err))
+		return nil, errors.Wrapf(err, "failed to get resource type")
 	}
 
 	// sometimes manifests of non-namespaced resources
@@ -87,7 +86,7 @@ func gvkToGVR(gvk schema.GroupVersionKind, restMapper *restmapper.DeferredDiscov
 		return nil, false, err
 	}
 	if err != nil {
-		return nil, false, errors.WithMessage(err, fmt.Sprintf("failed to get the resource REST mapping for GroupVersionKind(%s): ", gvk.String()))
+		return nil, false, errors.Wrapf(err, "failed to get the resource REST mapping for GroupVersionKind(%s)", gvk.String())
 	}
 
 	return &mapping.Resource, mapping.Scope.Name() == meta.RESTScopeNameNamespace, nil

pkg/cvo/render.go

@@ -8,6 +8,7 @@ import (
 	"path/filepath"
 	"text/template"
 
+	"github.com/pkg/errors"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
 )
@@ -102,12 +103,12 @@ type manifestRenderConfig struct {
 func renderManifest(config manifestRenderConfig, manifestBytes []byte) ([]byte, error) {
 	tmpl, err := template.New("manifest").Parse(string(manifestBytes))
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse manifest: %v", err)
+		return nil, errors.Wrapf(err, "failed to parse manifest")
 	}
 
 	buf := new(bytes.Buffer)
 	if err := tmpl.Execute(buf, config); err != nil {
-		return nil, fmt.Errorf("failed to execute template: %v", err)
+		return nil, errors.Wrapf(err, "failed to execute template")
 	}
 
 	return buf.Bytes(), nil

pkg/cvo/sync.go

@@ -12,13 +12,30 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/rest"
 
 	"github.com/openshift/cluster-version-operator/lib"
 	"github.com/openshift/cluster-version-operator/lib/resourcebuilder"
 	cvv1 "github.com/openshift/cluster-version-operator/pkg/apis/config.openshift.io/v1"
 	"github.com/openshift/cluster-version-operator/pkg/cvo/internal"
 )
 
+const (
+	// RequeueOnErrorAnnotationKey is key for annotation on a manifests object that instructs CVO to requeue on specific errors.
+	// The value is comma separated list of causes that forces requeue.
+	RequeueOnErrorAnnotationKey = "v1.cluster-version-operator.operators.openshift.io/requeue-on-error"
+
+	// RequeueOnErrorCauseNoMatch is used when no match is found for object in api.
+	// This maps to https://godoc.org/k8s.io/apimachinery/pkg/api/meta#NoKindMatchError and https://godoc.org/k8s.io/apimachinery/pkg/api/meta#NoResourceMatchError .
+	// https://godoc.org/k8s.io/apimachinery/pkg/api/meta#IsNoMatchError is used as a check.
+	RequeueOnErrorCauseNoMatch = "NoMatch"
+)
+
+// This is used to map the know causes to their check.
+var requeueOnErrorCauseToCheck = map[string]func(error) bool{
+	RequeueOnErrorCauseNoMatch: meta.IsNoMatchError,
+}
+
 // loadUpdatePayload reads the payload from disk or remote, as necessary.
 func (optr *Operator) loadUpdatePayload(config *cvv1.ClusterVersion) (*updatePayload, error) {
 	payloadDir, err := optr.updatePayloadDir(config)
@@ -38,66 +55,141 @@ func (optr *Operator) syncUpdatePayload(config *cvv1.ClusterVersion, payload *up
 	if len(version) == 0 {
 		version = payload.releaseImage
 	}
-	for i, manifest := range payload.manifests {
-		metricPayload.WithLabelValues(version, "pending").Set(float64(len(payload.manifests) - i))
-		metricPayload.WithLabelValues(version, "applied").Set(float64(i))
-		taskName := taskName(&manifest, i+1, len(payload.manifests))
-		glog.V(4).Infof("Running sync for %s", taskName)
-		glog.V(6).Infof("Manifest: %s", string(manifest.Raw))
 
-		ov, ok := getOverrideForManifest(config.Spec.Overrides, manifest)
+	total := len(payload.manifests)
+	done := 0
+	var tasks []*syncTask
+	for i := range payload.manifests {
+		tasks = append(tasks, &syncTask{
+			index:    i + 1,
+			total:    total,
+			manifest: &payload.manifests[i],
+		})
+	}
+
+	for i := 0; i < len(tasks); i++ {
+		task := tasks[i]
+		setAppliedAndPending(version, total, done)
+		glog.V(4).Infof("Running sync for %s", task)
+		glog.V(6).Infof("Manifest: %s", string(task.manifest.Raw))
+
+		ov, ok := getOverrideForManifest(config.Spec.Overrides, task.manifest)
 		if ok && ov.Unmanaged {
-			glog.V(4).Infof("Skipping %s as unmanaged", taskName)
+			glog.V(4).Infof("Skipping %s as unmanaged", task)
 			continue
 		}
 
-		var lastErr error
-		if err := wait.ExponentialBackoff(wait.Backoff{
-			Duration: time.Second * 10,
-			Factor:   1.3,
-			Steps:    3,
-		}, func() (bool, error) {
-			// build resource builder for manifest
-			var b resourcebuilder.Interface
-			var err error
-			if resourcebuilder.Mapper.Exists(manifest.GVK) {
-				b, err = resourcebuilder.New(resourcebuilder.Mapper, optr.restConfig, manifest)
-			} else {
-				b, err = internal.NewGenericBuilder(optr.restConfig, manifest)
-			}
-			if err != nil {
-				utilruntime.HandleError(fmt.Errorf("error creating resourcebuilder for %s: %v", taskName, err))
-				lastErr = err
-				metricPayloadErrors.WithLabelValues(version).Inc()
-				return false, nil
-			}
-			// run builder for the manifest
-			if err := b.Do(); err != nil {
-				utilruntime.HandleError(fmt.Errorf("error running apply for %s: %v", taskName, err))
-				lastErr = err
-				metricPayloadErrors.WithLabelValues(version).Inc()
-				return false, nil
-			}
-			return true, nil
-		}); err != nil {
-			reason, cause := reasonForPayloadSyncError(lastErr)
-			if len(cause) > 0 {
-				cause = ": " + cause
-			}
-			return &updateError{
-				Reason:  reason,
-				Message: fmt.Sprintf("Could not update %s%s", taskName, cause),
+		if err := task.Run(version, optr.restConfig); err != nil {
+			cause := errors.Cause(err)
+			if task.requeued == 0 && shouldRequeueOnErr(cause, task.manifest) {
+				task.requeued++
+				tasks = append(tasks, task)
+				continue
 			}
+			return err
 		}
+		done++
+		glog.V(4).Infof("Done syncing for %s", task)
+	}
+	setAppliedAndPending(version, total, done)
+	return nil
+}
 
-		glog.V(4).Infof("Done syncing for %s", taskName)
+type syncTask struct {
+	index    int
+	total    int
+	manifest *lib.Manifest
+	requeued int
+}
+
+func (st *syncTask) String() string {
+	ns := st.manifest.Object().GetNamespace()
+	if len(ns) == 0 {
+		return fmt.Sprintf("%s %q (%s, %d of %d)", strings.ToLower(st.manifest.GVK.Kind), st.manifest.Object().GetName(), st.manifest.GVK.GroupVersion().String(), st.index, st.total)
+	}
+	return fmt.Sprintf("%s \"%s/%s\" (%s, %d of %d)", strings.ToLower(st.manifest.GVK.Kind), ns, st.manifest.Object().GetName(), st.manifest.GVK.GroupVersion().String(), st.index, st.total)
+}
+
+func (st *syncTask) Run(version string, rc *rest.Config) error {
+	var lastErr error
+	if err := wait.ExponentialBackoff(wait.Backoff{
+		Duration: time.Second * 10,
+		Factor:   1.3,
+		Steps:    3,
+	}, func() (bool, error) {
+		// build resource builder for manifest
+		var b resourcebuilder.Interface
+		var err error
+		if resourcebuilder.Mapper.Exists(st.manifest.GVK) {
+			b, err = resourcebuilder.New(resourcebuilder.Mapper, rc, *st.manifest)
+		} else {
+			b, err = internal.NewGenericBuilder(rc, *st.manifest)
+		}
+		if err != nil {
+			utilruntime.HandleError(errors.Wrapf(err, "error creating resourcebuilder for %s", st))
+			lastErr = err
+			metricPayloadErrors.WithLabelValues(version).Inc()
+			return false, nil
+		}
+		// run builder for the manifest
+		if err := b.Do(); err != nil {
+			utilruntime.HandleError(errors.Wrapf(err, "error running apply for %s", st))
+			lastErr = err
+			metricPayloadErrors.WithLabelValues(version).Inc()
+			return false, nil
+		}
+		return true, nil
+	}); err != nil {
+		reason, cause := reasonForPayloadSyncError(lastErr)
+		if len(cause) > 0 {
+			cause = ": " + cause
+		}
+		return &updateError{
+			cause:   lastErr,
+			Reason:  reason,
+			Message: fmt.Sprintf("Could not update %s%s", st, cause),
+		}
 	}
-	metricPayload.WithLabelValues(version, "applied").Set(float64(len(payload.manifests)))
-	metricPayload.WithLabelValues(version, "pending").Set(0)
 	return nil
 }
 
+func shouldRequeueOnErr(err error, manifest *lib.Manifest) bool {
+	ok, errs := hasRequeueOnErrorAnnotation(manifest.Object().GetAnnotations())
+	if !ok {
+		return false
+	}
+	cause := errors.Cause(err)
+
+	should := false
+	for _, e := range errs {
+		if ef, ok := requeueOnErrorCauseToCheck[e]; ok {
+			if ef(cause) {
+				should = true
+				break
+			}
+		}
+	}
+	return should
+}
+
+func hasRequeueOnErrorAnnotation(annos map[string]string) (bool, []string) {
+	if annos == nil {
+		return false, nil
+	}
+	errs, ok := annos[RequeueOnErrorAnnotationKey]
+	if !ok {
+		return false, nil
+	}
+	return ok, strings.Split(errs, ",")
+}
+
+func setAppliedAndPending(version string, total, done int) {
+	metricPayload.WithLabelValues(version, "pending").Set(float64(total - done))
+	metricPayload.WithLabelValues(version, "applied").Set(float64(done))
+}
+
 type updateError struct {
+	cause   error
 	Reason  string
 	Message string
 }
@@ -106,6 +198,10 @@ func (e *updateError) Error() string {
 	return e.Message
 }
 
+func (e *updateError) Cause() error {
+	return e.cause
+}
+
 // reasonForUpdateError provides a succint explanation of a known error type for use in a human readable
 // message during update. Since all objects in the payload should be successfully applied, messages
 // should direct the reader (likely a cluster administrator) to a possible cause in their own config.
@@ -171,16 +267,8 @@ func summaryForReason(reason string) string {
 	return "an unknown error has occurred"
 }
 
-func taskName(manifest *lib.Manifest, index, total int) string {
-	ns := manifest.Object().GetNamespace()
-	if len(ns) == 0 {
-		return fmt.Sprintf("%s %q (%s, %d of %d)", strings.ToLower(manifest.GVK.Kind), manifest.Object().GetName(), manifest.GVK.GroupVersion().String(), index, total)
-	}
-	return fmt.Sprintf("%s \"%s/%s\" (%s, %d of %d)", strings.ToLower(manifest.GVK.Kind), ns, manifest.Object().GetName(), manifest.GVK.GroupVersion().String(), index, total)
-}
-
 // getOverrideForManifest returns the override and true when override exists for manifest.
-func getOverrideForManifest(overrides []cvv1.ComponentOverride, manifest lib.Manifest) (cvv1.ComponentOverride, bool) {
+func getOverrideForManifest(overrides []cvv1.ComponentOverride, manifest *lib.Manifest) (cvv1.ComponentOverride, bool) {
 	for idx, ov := range overrides {
 		kind, namespace, name := manifest.GVK.Kind, manifest.Object().GetNamespace(), manifest.Object().GetName()
 		if ov.Kind == kind &&