Add filter labels to catalog pod cache

Add catalogsource and update labels to pod cache to filter out
the pods that are not managed by catalog-operator in OLM. The
intention is to reduce the amount of cache for pod resources.

Signed-off-by: Vu Dinh <[email protected]>

Author: dinhxuanvu

pkg/controller/bundle/bundle_unpacker.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/operator-framework/operator-registry/pkg/api"
 	"github.com/operator-framework/operator-registry/pkg/configmap"
+	"github.com/sirupsen/logrus"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
@@ -39,6 +40,7 @@ const (
 	// The time duration should be in the same format as accepted by time.ParseDuration()
 	// e.g 1m30s
 	BundleUnpackTimeoutAnnotationKey = "operatorframework.io/bundle-unpack-timeout"
+	BundleUnpackPodLabel             = "job-name"
 )
 
 type BundleUnpackResult struct {
@@ -225,6 +227,7 @@ type Unpacker interface {
 }
 
 type ConfigMapUnpacker struct {
+	logger        *logrus.Logger
 	opmImage      string
 	utilImage     string
 	client        kubernetes.Interface
@@ -272,6 +275,12 @@ func WithUtilImage(utilImage string) ConfigMapUnpackerOption {
 	}
 }
 
+func WithLogger(logger *logrus.Logger) ConfigMapUnpackerOption {
+	return func(unpacker *ConfigMapUnpacker) {
+		unpacker.logger = logger
+	}
+}
+
 func WithClient(client kubernetes.Interface) ConfigMapUnpackerOption {
 	return func(unpacker *ConfigMapUnpacker) {
 		unpacker.client = client
@@ -505,9 +514,10 @@ func (c *ConfigMapUnpacker) UnpackBundle(lookup *operatorsv1alpha1.BundleLookup,
 func (c *ConfigMapUnpacker) pendingContainerStatusMessages(job *batchv1.Job) (string, error) {
 	containerStatusMessages := []string{}
 	// List pods for unpack job
-	podLabel := map[string]string{"job-name": job.GetName()}
-	pods, listErr := c.podLister.Pods(job.GetNamespace()).List(k8slabels.SelectorFromSet(podLabel))
+	podLabel := map[string]string{BundleUnpackPodLabel: job.GetName()}
+	pods, listErr := c.podLister.Pods(job.GetNamespace()).List(k8slabels.SelectorFromValidatedSet(podLabel))
 	if listErr != nil {
+		c.logger.Errorf("Failed to list pods for job(%s): %v", job.GetName(), listErr)
 		return "", fmt.Errorf("Failed to list pods for job(%s): %v", job.GetName(), listErr)
 	}
 

pkg/controller/operators/catalog/operator.go

@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/selection"
 	utilclock "k8s.io/apimachinery/pkg/util/clock"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -310,10 +311,32 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	op.lister.CoreV1().RegisterServiceLister(metav1.NamespaceAll, serviceInformer.Lister())
 	sharedIndexInformers = append(sharedIndexInformers, serviceInformer.Informer())
 
-	// Wire Pods
-	podInformer := k8sInformerFactory.Core().V1().Pods()
-	op.lister.CoreV1().RegisterPodLister(metav1.NamespaceAll, podInformer.Lister())
-	sharedIndexInformers = append(sharedIndexInformers, podInformer.Informer())
+	// Wire Pods for CatalogSource
+	catsrcReq, err := labels.NewRequirement(reconciler.CatalogSourceLabelKey, selection.Exists, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	csPodLabels := labels.NewSelector()
+	csPodLabels = csPodLabels.Add(*catsrcReq)
+	csPodInformer := informers.NewSharedInformerFactoryWithOptions(op.opClient.KubernetesInterface(), resyncPeriod(), informers.WithTweakListOptions(func(options *metav1.ListOptions) {
+		options.LabelSelector = csPodLabels.String()
+	})).Core().V1().Pods()
+	op.lister.CoreV1().RegisterPodLister(metav1.NamespaceAll, csPodInformer.Lister())
+	sharedIndexInformers = append(sharedIndexInformers, csPodInformer.Informer())
+
+	// Wire Pods for BundleUnpack job
+	buReq, err := labels.NewRequirement(bundle.BundleUnpackPodLabel, selection.Exists, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	buPodLabels := labels.NewSelector()
+	buPodLabels = buPodLabels.Add(*buReq)
+	buPodInformer := informers.NewSharedInformerFactoryWithOptions(op.opClient.KubernetesInterface(), resyncPeriod(), informers.WithTweakListOptions(func(options *metav1.ListOptions) {
+		options.LabelSelector = buPodLabels.String()
+	})).Core().V1().Pods()
+	sharedIndexInformers = append(sharedIndexInformers, buPodInformer.Informer())
 
 	// Wire ConfigMaps
 	configMapInformer := k8sInformerFactory.Core().V1().ConfigMaps()
@@ -344,11 +367,12 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 
 	// Setup the BundleUnpacker
 	op.bundleUnpacker, err = bundle.NewConfigmapUnpacker(
+		bundle.WithLogger(op.logger),
 		bundle.WithClient(op.opClient.KubernetesInterface()),
 		bundle.WithCatalogSourceLister(catsrcInformer.Lister()),
 		bundle.WithConfigMapLister(configMapInformer.Lister()),
 		bundle.WithJobLister(jobInformer.Lister()),
-		bundle.WithPodLister(podInformer.Lister()),
+		bundle.WithPodLister(buPodInformer.Lister()),
 		bundle.WithRoleLister(roleInformer.Lister()),
 		bundle.WithRoleBindingLister(roleBindingInformer.Lister()),
 		bundle.WithOPMImage(configmapRegistryImage),