pkg/{ansible,helm}/controller: cached default manager client (#1047)

* pkg/{ansible,helm}/controller: change the default manager client.

The new default client for Ansible and Helm with use the cache for the
reader interface.

* pkg/ansible/controller: read directly from API server in status updaters

* CHANGELOG.md: add Ansible/Helm cached reader change

Author: Shawn Hurley

CHANGELOG.md

@@ -40,6 +40,7 @@
 - Updated CRD generation for non-Go operators to use valid structural schema. ([#2275](https://github.com/operator-framework/operator-sdk/issues/2275))
 - Replace Role verb `"*"` with list of verb strings in generated files so the Role is compatible with OpenShift and Kubernetes. ([#2175](https://github.com/operator-framework/operator-sdk/pull/2175))
 - **Breaking change:** An existing CSV's `spec.customresourcedefinitions.owned` is now always overwritten except for each `name`, `version`, and `kind` on invoking [`olm-catalog gen-csv`](https://github.com/operator-framework/operator-sdk/blob/d147bb3/doc/cli/operator-sdk_olm-catalog_gen-csv.md) when Go API code [annotations](https://github.com/operator-framework/operator-sdk/blob/d147bb3/doc/user/olm-catalog/csv-annotations.md) are present. ([#1162](https://github.com/operator-framework/operator-sdk/pull/1162))
+- Ansible and Helm operator reconcilers use a cached client for reads instead of the default unstructured client. ([#1047](https://github.com/operator-framework/operator-sdk/pull/1047))
 
 ### Deprecated
 

hack/tests/e2e-ansible.sh

@@ -172,8 +172,8 @@ load_image_if_kind "$METRICS_TEST_IMAGE"
 
 OPERATORDIR="$(pwd)"
 
-deploy_operator
 trap_add 'remove_operator' EXIT
+deploy_operator
 test_operator
 remove_operator
 

pkg/ansible/controller/controller.go

@@ -65,6 +65,7 @@ func Add(mgr manager.Manager, options Options) *controller.Controller {
 		EventHandlers:   eventHandlers,
 		ReconcilePeriod: options.ReconcilePeriod,
 		ManageStatus:    options.ManageStatus,
+		APIReader:       mgr.GetAPIReader(),
 	}
 
 	scheme := mgr.GetScheme()

pkg/ansible/controller/reconcile.go

@@ -55,6 +55,7 @@ type AnsibleOperatorReconciler struct {
 	GVK             schema.GroupVersionKind
 	Runner          runner.Runner
 	Client          client.Client
+	APIReader       client.Reader
 	EventHandlers   []events.EventHandler
 	ReconcilePeriod time.Duration
 	ManageStatus    bool
@@ -84,7 +85,8 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 		duration, err := time.ParseDuration(ds)
 		if err != nil {
 			// Should attempt to update to a failed condition
-			if errmark := r.markError(u, request.NamespacedName, fmt.Sprintf("Unable to parse reconcile period annotation: %v", err)); errmark != nil {
+			errmark := r.markError(u, request.NamespacedName, fmt.Sprintf("Unable to parse reconcile period annotation: %v", err))
+			if errmark != nil {
 				logger.Error(errmark, "Unable to mark error annotation")
 			}
 			logger.Error(err, "Unable to parse reconcile period annotation")
@@ -124,10 +126,10 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 	}
 
 	if r.ManageStatus {
-		err = r.markRunning(u, request.NamespacedName)
-		if err != nil {
-			logger.Error(err, "Unable to update the status to mark cr as running")
-			return reconcileResult, err
+		errmark := r.markRunning(u, request.NamespacedName)
+		if errmark != nil {
+			logger.Error(errmark, "Unable to update the status to mark cr as running")
+			return reconcileResult, errmark
 		}
 	}
 
@@ -140,7 +142,8 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 
 	kc, err := kubeconfig.Create(ownerRef, "http://localhost:8888", u.GetNamespace())
 	if err != nil {
-		if errmark := r.markError(u, request.NamespacedName, "Unable to run reconciliation"); errmark != nil {
+		errmark := r.markError(u, request.NamespacedName, "Unable to run reconciliation")
+		if errmark != nil {
 			logger.Error(errmark, "Unable to mark error to run reconciliation")
 		}
 		logger.Error(err, "Unable to generate kubeconfig")
@@ -153,7 +156,8 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 	}()
 	result, err := r.Runner.Run(ident, u, kc.Name())
 	if err != nil {
-		if errmark := r.markError(u, request.NamespacedName, "Unable to run reconciliation"); errmark != nil {
+		errmark := r.markError(u, request.NamespacedName, "Unable to run reconciliation")
+		if errmark != nil {
 			logger.Error(errmark, "Unable to mark error to run reconciliation")
 		}
 		logger.Error(err, "Unable to run ansible runner")
@@ -193,13 +197,13 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 		return reconcileResult, eventErr
 	}
 
-	// Need to get the unstructured object after ansible
-	// this needs to hit the API
-	err = r.Client.Get(context.TODO(), request.NamespacedName, u)
-	if apierrors.IsNotFound(err) {
-		return reconcile.Result{}, nil
-	}
+	// Need to get the unstructured object after the Ansible runner finishes.
+	// This needs to hit the API server to retrieve updates.
+	err = r.APIReader.Get(context.TODO(), request.NamespacedName, u)
 	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return reconcile.Result{}, nil
+		}
 		return reconcile.Result{}, err
 	}
 
@@ -225,27 +229,24 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 		}
 	}
 	if r.ManageStatus {
-		err = r.markDone(u, request.NamespacedName, statusEvent, failureMessages)
-		if err != nil {
-			logger.Error(err, "Failed to mark status done")
+		errmark := r.markDone(u, request.NamespacedName, statusEvent, failureMessages)
+		if errmark != nil {
+			logger.Error(errmark, "Failed to mark status done")
 		}
+		return reconcileResult, errmark
 	}
-	return reconcileResult, err
+	return reconcileResult, nil
 }
 
 func (r *AnsibleOperatorReconciler) markRunning(u *unstructured.Unstructured, namespacedName types.NamespacedName) error {
-	// Get the latest resource to prevent updating a stale status
-	err := r.Client.Get(context.TODO(), namespacedName, u)
-	if err != nil {
+	// Get the latest resource to prevent updating a stale status.
+	if err := r.APIReader.Get(context.TODO(), namespacedName, u); err != nil {
 		return err
 	}
-	statusInterface := u.Object["status"]
-	statusMap, _ := statusInterface.(map[string]interface{})
-	crStatus := ansiblestatus.CreateFromMap(statusMap)
+	crStatus := getStatus(u)
 
 	// If there is no current status add that we are working on this resource.
 	errCond := ansiblestatus.GetCondition(crStatus, ansiblestatus.FailureConditionType)
-
 	if errCond != nil {
 		errCond.Status = v1.ConditionFalse
 		ansiblestatus.SetCondition(&crStatus, *errCond)
@@ -262,34 +263,26 @@ func (r *AnsibleOperatorReconciler) markRunning(u *unstructured.Unstructured, na
 	)
 	ansiblestatus.SetCondition(&crStatus, *c)
 	u.Object["status"] = crStatus.GetJSONMap()
-	err = r.Client.Status().Update(context.TODO(), u)
-	if err != nil {
-		return err
-	}
-	return nil
+
+	return r.Client.Status().Update(context.TODO(), u)
 }
 
 // markError - used to alert the user to the issues during the validation of a reconcile run.
 // i.e Annotations that could be incorrect
 func (r *AnsibleOperatorReconciler) markError(u *unstructured.Unstructured, namespacedName types.NamespacedName, failureMessage string) error {
 	logger := logf.Log.WithName("markError")
+	// Immediately update metrics with failed reconciliation, since Get()
+	// may fail.
 	metrics.ReconcileFailed(r.GVK.String())
-	// Get the latest resource to prevent updating a stale status
-	err := r.Client.Get(context.TODO(), namespacedName, u)
-	if apierrors.IsNotFound(err) {
-		logger.Info("Resource not found, assuming it was deleted")
-		return nil
-	}
-	if err != nil {
+	// Get the latest resource to prevent updating a stale status.
+	if err := r.APIReader.Get(context.TODO(), namespacedName, u); err != nil {
+		if apierrors.IsNotFound(err) {
+			logger.Info("Resource not found, assuming it was deleted")
+			return nil
+		}
 		return err
 	}
-	statusInterface := u.Object["status"]
-	statusMap, ok := statusInterface.(map[string]interface{})
-	// If the map is not available create one.
-	if !ok {
-		statusMap = map[string]interface{}{}
-	}
-	crStatus := ansiblestatus.CreateFromMap(statusMap)
+	crStatus := getStatus(u)
 
 	sc := ansiblestatus.GetCondition(crStatus, ansiblestatus.RunningConditionType)
 	if sc != nil {
@@ -313,27 +306,26 @@ func (r *AnsibleOperatorReconciler) markError(u *unstructured.Unstructured, name
 
 func (r *AnsibleOperatorReconciler) markDone(u *unstructured.Unstructured, namespacedName types.NamespacedName, statusEvent eventapi.StatusJobEvent, failureMessages eventapi.FailureMessages) error {
 	logger := logf.Log.WithName("markDone")
-	// Get the latest resource to prevent updating a stale status
-	err := r.Client.Get(context.TODO(), namespacedName, u)
-	if apierrors.IsNotFound(err) {
-		logger.Info("Resource not found, assuming it was deleted")
-		return nil
-	}
-	if err != nil {
+	// Get the latest resource to prevent updating a stale status.
+	if err := r.APIReader.Get(context.TODO(), namespacedName, u); err != nil {
+		if apierrors.IsNotFound(err) {
+			logger.Info("Resource not found, assuming it was deleted")
+			return nil
+		}
 		return err
 	}
-	statusInterface := u.Object["status"]
-	statusMap, _ := statusInterface.(map[string]interface{})
-	crStatus := ansiblestatus.CreateFromMap(statusMap)
+	crStatus := getStatus(u)
 
 	runSuccessful := len(failureMessages) == 0
 	ansibleStatus := ansiblestatus.NewAnsibleResultFromStatusJobEvent(statusEvent)
 
 	if !runSuccessful {
 		metrics.ReconcileFailed(r.GVK.String())
 		sc := ansiblestatus.GetCondition(crStatus, ansiblestatus.RunningConditionType)
-		sc.Status = v1.ConditionFalse
-		ansiblestatus.SetCondition(&crStatus, *sc)
+		if sc != nil {
+			sc.Status = v1.ConditionFalse
+			ansiblestatus.SetCondition(&crStatus, *sc)
+		}
 		c := ansiblestatus.NewCondition(
 			ansiblestatus.FailureConditionType,
 			v1.ConditionTrue,
@@ -369,3 +361,14 @@ func contains(l []string, s string) bool {
 	}
 	return false
 }
+
+// getStatus returns u's "status" block as a status.Status.
+func getStatus(u *unstructured.Unstructured) ansiblestatus.Status {
+	statusInterface := u.Object["status"]
+	statusMap, ok := statusInterface.(map[string]interface{})
+	// If the map is not available create one.
+	if !ok {
+		statusMap = map[string]interface{}{}
+	}
+	return ansiblestatus.CreateFromMap(statusMap)
+}

pkg/ansible/controller/reconcile_test.go

@@ -487,6 +487,7 @@ func TestReconcile(t *testing.T) {
 				GVK:             tc.GVK,
 				Runner:          tc.Runner,
 				Client:          tc.Client,
+				APIReader:       tc.Client,
 				EventHandlers:   tc.EventHandlers,
 				ReconcilePeriod: tc.ReconcilePeriod,
 				ManageStatus:    tc.ManageStatus,

pkg/ansible/run.go

@@ -21,25 +21,28 @@ import (
 	"runtime"
 
 	"github.com/operator-framework/operator-sdk/pkg/ansible/controller"
+	aoflags "github.com/operator-framework/operator-sdk/pkg/ansible/flags"
+	proxy "github.com/operator-framework/operator-sdk/pkg/ansible/proxy"
 	"github.com/operator-framework/operator-sdk/pkg/ansible/proxy/controllermap"
 	"github.com/operator-framework/operator-sdk/pkg/ansible/runner"
 	"github.com/operator-framework/operator-sdk/pkg/ansible/watches"
 	"github.com/operator-framework/operator-sdk/pkg/k8sutil"
+	kubemetrics "github.com/operator-framework/operator-sdk/pkg/kube-metrics"
 	"github.com/operator-framework/operator-sdk/pkg/leader"
 	"github.com/operator-framework/operator-sdk/pkg/metrics"
+	sdkVersion "github.com/operator-framework/operator-sdk/version"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
-
-	aoflags "github.com/operator-framework/operator-sdk/pkg/ansible/flags"
-	proxy "github.com/operator-framework/operator-sdk/pkg/ansible/proxy"
-	kubemetrics "github.com/operator-framework/operator-sdk/pkg/kube-metrics"
-	sdkVersion "github.com/operator-framework/operator-sdk/version"
-	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	logf "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 var (
@@ -79,6 +82,17 @@ func Run(flags *aoflags.AnsibleOperatorFlags) error {
 	mgr, err := manager.New(cfg, manager.Options{
 		Namespace:          namespace,
 		MetricsBindAddress: fmt.Sprintf("%s:%d", metricsHost, metricsPort),
+		NewClient: func(cache cache.Cache, config *rest.Config, options client.Options) (client.Client, error) {
+			c, err := client.New(config, options)
+			if err != nil {
+				return nil, err
+			}
+			return &client.DelegatingClient{
+				Reader:       cache,
+				Writer:       c,
+				StatusClient: c,
+			}, nil
+		},
 	})
 	if err != nil {
 		log.Error(err, "Failed to create a new manager.")

pkg/helm/run.go

@@ -34,6 +34,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -80,6 +83,17 @@ func Run(flags *hoflags.HelmOperatorFlags) error {
 	mgr, err := manager.New(cfg, manager.Options{
 		Namespace:          namespace,
 		MetricsBindAddress: fmt.Sprintf("%s:%d", metricsHost, metricsPort),
+		NewClient: func(cache cache.Cache, config *rest.Config, options crclient.Options) (crclient.Client, error) {
+			c, err := crclient.New(config, options)
+			if err != nil {
+				return nil, err
+			}
+			return &crclient.DelegatingClient{
+				Reader:       cache,
+				Writer:       c,
+				StatusClient: c,
+			}, nil
+		},
 	})
 	if err != nil {
 		log.Error(err, "Failed to create a new manager.")