Enable status reporting for CatalogSource in OperatorPolicy

Following the schema of implementing multiple `handle` functions,
the CatalogSource status reporting logic utilizes the same
`updateStatus` function to facilitate the `conditions` and
`relatedObject` field updates on the related operator policy. The
controller directly checks if the CatalogSource object exists on the
cluster to propagate relevant health information up to the operator
policy. The specific field that is checked in the catalog source is
`.status.connectionState.lastObservedState`. Any value other than
`READY` is considered as failing.

ref: https://issues.redhat.com/browse/ACM-9285
Signed-off-by: Jason Zhang <jaszhang@redhat.com>

controllers/operatorpolicy_controller.go

@@ -35,6 +35,7 @@ import (
 
 const (
 	OperatorControllerName string = "operator-policy-controller"
+	CatalogSourceReady     string = "READY"
 )
 
 var (
@@ -58,6 +59,11 @@ var (
 		Version: "v1",
 		Kind:    "Deployment",
 	}
+	catalogSrcGVK = schema.GroupVersionKind{
+		Group:   "operators.coreos.com",
+		Version: "v1alpha1",
+		Kind:    "CatalogSource",
+	}
 )
 
 // OperatorPolicyReconciler reconciles a OperatorPolicy object
@@ -165,9 +171,83 @@ func (r *OperatorPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		return reconcile.Result{}, err
 	}
 
+	// handle catalogsource
+	if err := r.handleCatalogSource(ctx, policy, subscription); err != nil {
+		OpLog.Error(err, "Error handling CatalogSource")
+
+		return reconcile.Result{}, err
+	}
+
 	return reconcile.Result{}, nil
 }
 
+func (r *OperatorPolicyReconciler) handleCatalogSource(
+	ctx context.Context,
+	policy *policyv1beta1.OperatorPolicy,
+	subscription *operatorv1alpha1.Subscription,
+) error {
+	watcher := opPolIdentifier(policy.Namespace, policy.Name)
+
+	var catalogName string
+	var catalogNS string
+
+	if subscription == nil {
+		sub, err := buildSubscription(policy)
+		if err != nil {
+			return fmt.Errorf("error building Subscription: %w", err)
+		}
+
+		catalogName = sub.Spec.CatalogSource
+		catalogNS = sub.Spec.CatalogSourceNamespace
+	} else {
+		catalogName = subscription.Spec.CatalogSource
+		catalogNS = subscription.Spec.CatalogSourceNamespace
+	}
+
+	// Check if CatalogSource exists
+	foundCatalogSrc, err := r.DynamicWatcher.Get(watcher, catalogSrcGVK,
+		catalogNS, catalogName)
+	if err != nil {
+		return fmt.Errorf("error getting CatalogSource: %w", err)
+	}
+
+	isMissing := foundCatalogSrc == nil
+	isUnhealthy := isMissing
+
+	if !isMissing {
+		// CatalogSource is found, initiate health check
+		catalogSrcUnstruct := foundCatalogSrc.DeepCopy()
+		catalogSrc := new(operatorv1alpha1.CatalogSource)
+
+		err := runtime.DefaultUnstructuredConverter.
+			FromUnstructured(catalogSrcUnstruct.Object, catalogSrc)
+		if err != nil {
+			return fmt.Errorf("error converting the retrieved CatalogSource to the Go type: %w", err)
+		}
+
+		if catalogSrc.Status.GRPCConnectionState == nil {
+			// Unknown State
+			err := r.updateStatus(ctx, policy, catalogSourceUnknownCond, catalogSrcUnknownObj(catalogName, catalogNS))
+			if err != nil {
+				return fmt.Errorf("error retrieving the status for a CatalogSource: %w", err)
+			}
+
+			return nil
+		}
+
+		CatalogSrcState := catalogSrc.Status.GRPCConnectionState.LastObservedState
+		isUnhealthy = (CatalogSrcState != CatalogSourceReady)
+	}
+
+	err = r.updateStatus(ctx, policy, catalogSourceFindCond(isUnhealthy, isMissing),
+		catalogSourceObj(catalogName, catalogNS, isUnhealthy, isMissing))
+	if err != nil {
+		return fmt.Errorf("error updating the status for a CatalogSource: %w", err)
+	}
+
+	return nil
+}
+
 func (r *OperatorPolicyReconciler) handleOpGroup(ctx context.Context, policy *policyv1beta1.OperatorPolicy) error {
 	watcher := opPolIdentifier(policy.Namespace, policy.Name)
 
@@ -453,7 +533,7 @@ func (r *OperatorPolicyReconciler) handleSubscription(
 		// For now, just mark it as compliant
 		err := r.updateStatus(ctx, policy, matchesCond("Subscription"), matchedObj(foundSub))
 		if err != nil {
-			return nil, fmt.Errorf("error updating the status for an OperatorGroup that matches: %w", err)
+			return nil, fmt.Errorf("error updating the status for a Subscription that matches: %w", err)
 		}
 
 		return mergedSub, nil

controllers/operatorpolicy_status.go

@@ -226,6 +226,17 @@ func calculateComplianceCondition(policy *policyv1beta1.OperatorPolicy) metav1.C
 	}
 
 	// FUTURE: check additional conditions
+	idx, cond = policy.Status.GetCondition(catalogSrcConditionType)
+	if idx == -1 {
+		messages = append(messages, "the status of the CatalogSource is unknown")
+		foundNonCompliant = true
+	} else {
+		messages = append(messages, cond.Message)
+
+		if cond.Status != metav1.ConditionFalse {
+			foundNonCompliant = true
+		}
+	}
 
 	if foundNonCompliant {
 		return metav1.Condition{
@@ -305,6 +316,7 @@ const (
 	subConditionType        = "SubscriptionCompliant"
 	csvConditionType        = "ClusterServiceVersionCompliant"
 	deploymentConditionType = "DeploymentCompliant"
+	catalogSrcConditionType = "CatalogSourcesUnhealthy"
 )
 
 func condType(kind string) string {
@@ -468,6 +480,88 @@ var noDeploymentsCond = metav1.Condition{
 	Message: "The ClusterServiceVersion is missing, thus meaning there are no relevant deployments",
 }
 
+// catalogSourceFindCond is a conditionally compliant condition with reason
+// based on the `isUnhealthy` and `isMissing` parameters
+func catalogSourceFindCond(isUnhealthy bool, isMissing bool) metav1.Condition {
+	status := metav1.ConditionFalse
+	reason := "CatalogSourcesFound"
+	message := "CatalogSource was found"
+
+	if isUnhealthy {
+		status = metav1.ConditionTrue
+		reason = "CatalogSourcesFoundUnhealthy"
+		message = "CatalogSource was found but is unhealthy"
+	}
+
+	if isMissing {
+		status = metav1.ConditionTrue
+		reason = "CatalogSourcesNotFound"
+		message = "CatalogSource was not found"
+	}
+
+	return metav1.Condition{
+		Type:    "CatalogSourcesUnhealthy",
+		Status:  status,
+		Reason:  reason,
+		Message: message,
+	}
+}
+
+// catalogSourceUnknownCond is a NonCompliant condition
+var catalogSourceUnknownCond = metav1.Condition{
+	Type:    "CatalogSourcesUnknownState",
+	Status:  metav1.ConditionTrue,
+	Reason:  "LastObservedUnknown",
+	Message: "Could not determine last observed state of CatalogSource",
+}
+
+// catalogSourceObj returns a conditionally compliant RelatedObject with reason based on the
+// `isUnhealthy` and `isMissing` parameters
+func catalogSourceObj(catalogName string, catalogNS string, isUnhealthy bool, isMissing bool) policyv1.RelatedObject {
+	compliance := string(policyv1.Compliant)
+	reason := reasonWantFoundExists
+
+	if isUnhealthy {
+		compliance = string(policyv1.NonCompliant)
+		reason = reasonWantFoundExists + " but is unhealthy"
+	}
+
+	if isMissing {
+		compliance = string(policyv1.NonCompliant)
+		reason = reasonWantFoundDNE
+	}
+
+	return policyv1.RelatedObject{
+		Object: policyv1.ObjectResource{
+			Kind:       catalogSrcGVK.Kind,
+			APIVersion: catalogSrcGVK.GroupVersion().String(),
+			Metadata: policyv1.ObjectMetadata{
+				Name:      catalogName,
+				Namespace: catalogNS,
+			},
+		},
+		Compliant: compliance,
+		Reason:    reason,
+	}
+}
+
+// catalogSrcUnknownObj returns a NonCompliant RelatedObject with
+// reason = 'Resource found but current state is unknown'
+func catalogSrcUnknownObj(catalogName string, catalogNS string) policyv1.RelatedObject {
+	return policyv1.RelatedObject{
+		Object: policyv1.ObjectResource{
+			Kind:       catalogSrcGVK.Kind,
+			APIVersion: catalogSrcGVK.GroupVersion().String(),
+			Metadata: policyv1.ObjectMetadata{
+				Name:      catalogName,
+				Namespace: catalogNS,
+			},
+		},
+		Compliant: string(policyv1.NonCompliant),
+		Reason:    "Resource found but current state is unknown",
+	}
+}
+
 // missingWantedObj returns a NonCompliant RelatedObject with reason = 'Resource not found but should exist'
 func missingWantedObj(obj client.Object) policyv1.RelatedObject {
 	return policyv1.RelatedObject{