Add PriorityClass setting to registry pods for CatalogSource via annotations

pkg/controller/registry/reconciler/configmap_test.go

@@ -204,8 +204,9 @@ func objectsForCatalogSource(catsrc *v1alpha1.CatalogSource) []runtime.Object {
 		if catsrc.Spec.Image != "" {
 			decorated := grpcCatalogSourceDecorator{catsrc}
 			objs = clientfake.AddSimpleGeneratedNames(
-				decorated.Pod(""),
+				decorated.Pod(catsrc.GetName()),
 				decorated.Service(),
+				decorated.ServiceAccount(),
 			)
 		}
 	}

pkg/controller/registry/reconciler/grpc.go

@@ -171,15 +171,16 @@ func (c *GrpcRegistryReconciler) currentUpdatePods(source grpcCatalogSourceDecor
 	return pods
 }
 
-func (c *GrpcRegistryReconciler) currentPodsWithCorrectImage(source grpcCatalogSourceDecorator) []*corev1.Pod {
+func (c *GrpcRegistryReconciler) currentPodsWithCorrectImageAndSpec(source grpcCatalogSourceDecorator, saName string) []*corev1.Pod {
 	pods, err := c.Lister.CoreV1().PodLister().Pods(source.GetNamespace()).List(labels.SelectorFromValidatedSet(source.Labels()))
 	if err != nil {
 		logrus.WithError(err).Warn("couldn't find pod in cache")
 		return nil
 	}
 	found := []*corev1.Pod{}
+	newPod := source.Pod(saName)
 	for _, p := range pods {
-		if p.Spec.Containers[0].Image == source.Spec.Image {
+		if p.Spec.Containers[0].Image == source.Spec.Image && podHashMatch(p, newPod) {
 			found = append(found, p)
 		}
 	}
@@ -192,11 +193,12 @@ func (c *GrpcRegistryReconciler) EnsureRegistryServer(catalogSource *v1alpha1.Ca
 
 	// if service status is nil, we force create every object to ensure they're created the first time
 	overwrite := source.Status.RegistryServiceStatus == nil
-	// recreate the pod if no existing pod is serving the latest image
-	overwritePod := overwrite || len(c.currentPodsWithCorrectImage(source)) == 0
 
 	//TODO: if any of these error out, we should write a status back (possibly set RegistryServiceStatus to nil so they get recreated)
 	sa, err := c.ensureSA(source)
+	// recreate the pod if no existing pod is serving the latest image or correct spec
+	overwritePod := overwrite || len(c.currentPodsWithCorrectImageAndSpec(source, sa.GetName())) == 0
+
 	if err != nil && !k8serror.IsAlreadyExists(err) {
 		return errors.Wrapf(err, "error ensuring service account: %s", source.GetName())
 	}
@@ -421,10 +423,9 @@ func (c *GrpcRegistryReconciler) removePods(pods []*corev1.Pod, namespace string
 // CheckRegistryServer returns true if the given CatalogSource is considered healthy; false otherwise.
 func (c *GrpcRegistryReconciler) CheckRegistryServer(catalogSource *v1alpha1.CatalogSource) (healthy bool, err error) {
 	source := grpcCatalogSourceDecorator{catalogSource}
-
 	// Check on registry resources
 	// TODO: add gRPC health check
-	if len(c.currentPodsWithCorrectImage(source)) < 1 ||
+	if len(c.currentPodsWithCorrectImageAndSpec(source, source.ServiceAccount().GetName())) < 1 ||
 		c.currentService(source) == nil {
 		healthy = false
 		return
@@ -478,3 +479,30 @@ func (c *GrpcRegistryReconciler) podFailed(pod *corev1.Pod) (bool, error) {
 	}
 	return false, nil
 }
+
+// podHashMatch will check the hash info in existing pod to ensure its
+// hash info matches the desired Service's hash.
+func podHashMatch(existing, new *corev1.Pod) bool {
+	labels := existing.GetLabels()
+	newLabels := new.GetLabels()
+	// If both new & existing pods don't have labels, consider it not matched
+	if len(labels) == 0 || len(newLabels) == 0 {
+		return false
+	}
+
+	existingPodSpecHash, ok := labels[PodHashLabelKey]
+	if !ok {
+		return false
+	}
+
+	newPodSpecHash, ok := newLabels[PodHashLabelKey]
+	if !ok {
+		return false
+	}
+
+	if existingPodSpecHash != newPodSpecHash {
+		return false
+	}
+
+	return true
+}

pkg/controller/registry/reconciler/grpc_test.go

@@ -339,6 +339,74 @@ func TestGrpcRegistryReconciler(t *testing.T) {
 	}
 }
 
+func TestRegistryPodPriorityClass(t *testing.T) {
+	now := func() metav1.Time { return metav1.Date(2018, time.January, 26, 20, 40, 0, 0, time.UTC) }
+
+	type cluster struct {
+		k8sObjs []runtime.Object
+	}
+	type in struct {
+		cluster cluster
+		catsrc  *v1alpha1.CatalogSource
+	}
+	tests := []struct {
+		testName      string
+		in            in
+		priorityclass string
+	}{
+		{
+			testName: "Grpc/WithValidPriorityClassAnnotation",
+			in: in{
+				catsrc: grpcCatalogSourceWithAnnotations(map[string]string{
+					"operatorframework.io/priorityclass": "system-cluster-critical",
+				}),
+			},
+			priorityclass: "system-cluster-critical",
+		},
+		{
+			testName: "Grpc/WithInvalidPriorityClassAnnotation",
+			in: in{
+				catsrc: grpcCatalogSourceWithAnnotations(map[string]string{
+					"operatorframework.io/priorityclass": "",
+				}),
+			},
+			priorityclass: "",
+		},
+		{
+			testName: "Grpc/WithNoPriorityClassAnnotation",
+			in: in{
+				catsrc: grpcCatalogSourceWithAnnotations(map[string]string{
+					"annotationkey": "annotationvalue",
+				}),
+			},
+			priorityclass: "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.testName, func(t *testing.T) {
+			stopc := make(chan struct{})
+			defer close(stopc)
+
+			factory, client := fakeReconcilerFactory(t, stopc, withNow(now), withK8sObjs(tt.in.cluster.k8sObjs...), withK8sClientOptions(clientfake.WithNameGeneration(t)))
+			rec := factory.ReconcilerForSource(tt.in.catsrc)
+
+			err := rec.EnsureRegistryServer(tt.in.catsrc)
+			require.NoError(t, err)
+
+			// Check for resource existence
+			decorated := grpcCatalogSourceDecorator{tt.in.catsrc}
+			pod := decorated.Pod(tt.in.catsrc.GetName())
+			listOptions := metav1.ListOptions{LabelSelector: labels.SelectorFromSet(labels.Set{CatalogSourceLabelKey: tt.in.catsrc.GetName()}).String()}
+			outPods, podErr := client.KubernetesInterface().CoreV1().Pods(pod.GetNamespace()).List(context.TODO(), listOptions)
+			require.NoError(t, podErr)
+			require.Len(t, outPods.Items, 1)
+			outPod := outPods.Items[0]
+			require.Equal(t, tt.priorityclass, outPod.Spec.PriorityClassName)
+			require.Equal(t, pod.GetLabels()[PodHashLabelKey], outPod.GetLabels()[PodHashLabelKey])
+		})
+	}
+}
+
 func TestGrpcRegistryChecker(t *testing.T) {
 	type cluster struct {
 		k8sObjs []runtime.Object

pkg/controller/registry/reconciler/reconciler.go

@@ -2,14 +2,18 @@
 package reconciler
 
 import (
+	"fmt"
+	"hash/fnv"
 	"strings"
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/rand"
 
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	controllerclient "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/controller-runtime/client"
+	hashutil "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/kubernetes/pkg/util/hash"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister"
 )
@@ -19,6 +23,10 @@ type nowFunc func() metav1.Time
 const (
 	// CatalogSourceLabelKey is the key for a label containing a CatalogSource name.
 	CatalogSourceLabelKey string = "olm.catalogSource"
+	// CatalogPriorityClassKey is the key of an annotation in default catalogsources
+	CatalogPriorityClassKey string = "operatorframework.io/priorityclass"
+	// PodHashLabelKey is the key of a label for podspec hash information
+	PodHashLabelKey = "olm.pod-spec-hash"
 )
 
 // RegistryEnsurer describes methods for ensuring a registry exists.
@@ -160,5 +168,25 @@ func Pod(source *v1alpha1.CatalogSource, name string, image string, saName strin
 			ServiceAccountName: saName,
 		},
 	}
+
+	// Set priorityclass if its annotation exists
+	if prio, ok := annotations[CatalogPriorityClassKey]; ok && prio != "" {
+		pod.Spec.PriorityClassName = prio
+	}
+
+	// Add PodSpec hash
+	// This hash info will be used to detect PodSpec changes
+	if labels == nil {
+		labels = make(map[string]string)
+	}
+	labels[PodHashLabelKey] = hashPodSpec(pod.Spec)
+	pod.SetLabels(labels)
 	return pod
 }
+
+// hashPodSpec calculates a hash given a copy of the pod spec
+func hashPodSpec(spec v1.PodSpec) string {
+	hasher := fnv.New32a()
+	hashutil.DeepHashObject(hasher, &spec)
+	return rand.SafeEncodeString(fmt.Sprint(hasher.Sum32()))
+}