DVO-191: [quickfix] revert threading (#314)

* reset reconcicler

* revert PR issue

* Suspend benchmark due to reverted changes

* Restore later changes that were undone by the revert

Author: ncaak

pkg/controller/generic_reconciler.go

@@ -44,7 +44,6 @@ type GenericReconciler struct {
 	cmWatcher             *configmap.Watcher
 	validationEngine      validations.Interface
 	apiResources          []metav1.APIResource
-	resourceGVKs          []schema.GroupVersionKind
 }
 
 // NewGenericReconciler returns a GenericReconciler struct
@@ -163,27 +162,10 @@ func (gr *GenericReconciler) reconcileEverything(ctx context.Context) error {
 	once.Do(func() {
 		apiResources, err := reconcileResourceList(gr.discovery, gr.client.Scheme())
 		if err != nil {
-			gr.logger.Error(
-				err,
-				"cannot read API resources",
-			)
+			gr.logger.Error(err, "retrieving API resources to reconcile")
 			return
 		}
 		gr.apiResources = apiResources
-		gvks := getNamespacedResourcesGVK(gr.apiResources)
-		// sorting GVKs is very important for getting the consistent results
-		// when trying to match the 'app' label values. We must be sure that
-		// resources from the group apps/v1 are processed between first.
-		sort.SliceStable(gvks, func(i, j int) bool {
-			f := gvks[i]
-			s := gvks[j]
-			// sort resource by Kind in the same group
-			if f.Group == s.Group {
-				return f.Kind < s.Kind
-			}
-			return f.Group < s.Group
-		})
-		gr.resourceGVKs = gvks
 	})
 
 	for i, resource := range gr.apiResources {
@@ -198,7 +180,8 @@ func (gr *GenericReconciler) reconcileEverything(ctx context.Context) error {
 		return fmt.Errorf("getting watched namespaces: %w", err)
 	}
 
-	errNR := gr.processNamespacedResources(ctx, namespaces)
+	gvkResources := gr.getNamespacedResourcesGVK(gr.apiResources)
+	errNR := gr.processNamespacedResources(ctx, gvkResources, namespaces)
 	if errNR != nil {
 		return fmt.Errorf("processing namespace scoped resources: %w", errNR)
 	}
@@ -208,27 +191,26 @@ func (gr *GenericReconciler) reconcileEverything(ctx context.Context) error {
 	return nil
 }
 
-type unstructuredWithSelector struct {
-	unstructured *unstructured.Unstructured
-	selector     labels.Selector
-}
-
-type groupOfObjects struct {
-	objects []*unstructured.Unstructured
-	label   string
-}
-
 // groupAppObjects iterates over provided GroupVersionKind in given namespace
 // and returns map of objects grouped by their "app" label
 func (gr *GenericReconciler) groupAppObjects(ctx context.Context,
-	namespace string, ch chan groupOfObjects) {
-	defer close(ch)
+	namespace string, gvks []schema.GroupVersionKind) (map[string][]*unstructured.Unstructured, error) {
 	relatedObjects := make(map[string][]*unstructured.Unstructured)
-	labelToLabelSet := make(map[string]*labels.Set)
 
-	var objectsWithNonEmptySelector []*unstructuredWithSelector
+	// sorting GVKs is very important for getting the consistent results
+	// when trying to match the 'app' label values. We must be sure that
+	// resources from the group apps/v1 are processed between first.
+	sort.Slice(gvks, func(i, j int) bool {
+		f := gvks[i]
+		s := gvks[j]
+		// sort resource by Kind in the same group
+		if f.Group == s.Group {
+			return f.Kind < s.Kind
+		}
+		return f.Group < s.Group
+	})
 
-	for _, gvk := range gr.resourceGVKs {
+	for _, gvk := range gvks {
 		list := unstructured.UnstructuredList{}
 		listOptions := &client.ListOptions{
 			Limit:     gr.listLimit,
@@ -238,20 +220,15 @@ func (gr *GenericReconciler) groupAppObjects(ctx context.Context,
 		for {
 
 			if err := gr.client.List(ctx, &list, listOptions); err != nil {
-				continue
+				return nil, fmt.Errorf("listing %s: %w", gvk.String(), err)
 			}
 
 			for i := range list.Items {
 				obj := &list.Items[i]
 				unstructured.RemoveNestedField(obj.Object, "metadata", "managedFields")
 				unstructured.RemoveNestedField(obj.Object, "status")
-				processResourceLabels(obj, relatedObjects, labelToLabelSet)
-				sel, err := getLabelSelector(obj)
-				if err != nil || sel == labels.Nothing() {
-					continue
-				}
-				objectsWithNonEmptySelector = append(objectsWithNonEmptySelector,
-					&unstructuredWithSelector{unstructured: obj, selector: sel})
+				processResourceLabels(obj, relatedObjects)
+				gr.processResourceSelectors(obj, relatedObjects)
 			}
 
 			listContinue := list.GetContinue()
@@ -261,63 +238,72 @@ func (gr *GenericReconciler) groupAppObjects(ctx context.Context,
 			listOptions.Continue = listContinue
 		}
 	}
-	for label := range relatedObjects {
-		labelsSet := labelToLabelSet[label]
-		for _, o := range objectsWithNonEmptySelector {
-			if o.selector.Matches(labelsSet) {
-				relatedObjects[label] = append(relatedObjects[label], o.unstructured)
-			}
-		}
-		ch <- groupOfObjects{label: label, objects: relatedObjects[label]}
-	}
-}
-
-func getLabelSelector(obj *unstructured.Unstructured) (labels.Selector, error) {
-	labelSelector := utils.GetLabelSelector(obj)
-	return metav1.LabelSelectorAsSelector(labelSelector)
+	return relatedObjects, nil
 }
 
 // processResourceLabels reads resource labels and if the labels
 // are not empty then format them into string and put the string value
 // as key and the object as a value into "relatedObjects" map
 func processResourceLabels(obj *unstructured.Unstructured,
-	relatedObjects map[string][]*unstructured.Unstructured, labelSetMapping map[string]*labels.Set) {
+	relatedObjects map[string][]*unstructured.Unstructured) {
 
 	objLabels := utils.GetLabels(obj)
 	if len(objLabels) == 0 {
 		return
 	}
 	labelsString := labels.FormatLabels(objLabels)
 	relatedObjects[labelsString] = append(relatedObjects[labelsString], obj)
-	labelSetMapping[labelsString] = &objLabels
+}
+
+// processResourceSelectors reads resource selector and then tries to match
+// the selector to known labels (keys in the relatedObjects map). If a match is found then
+// the object is added to the corresponding group (values in the relatedObjects map).
+func (gr *GenericReconciler) processResourceSelectors(obj *unstructured.Unstructured,
+	relatedObjects map[string][]*unstructured.Unstructured) {
+	labelSelector := utils.GetLabelSelector(obj)
+	selector, err := metav1.LabelSelectorAsSelector(labelSelector)
+	if err != nil {
+		gr.logger.Error(err, "cannot convert label selector for object", obj.GetKind(), obj.GetName())
+		return
+	}
+
+	if selector == labels.Nothing() {
+		return
+	}
+
+	for k := range relatedObjects {
+		labelsSet, err := labels.ConvertSelectorToLabelsMap(k)
+		if err != nil {
+			gr.logger.Error(err, "cannot convert selector to labels map for", obj.GetKind(), obj.GetName())
+			continue
+		}
+		if selector.Matches(labelsSet) {
+			relatedObjects[k] = append(relatedObjects[k], obj)
+		}
+	}
 }
 
 func (gr *GenericReconciler) processNamespacedResources(
-	ctx context.Context, namespaces *[]namespace) error {
+	ctx context.Context, gvks []schema.GroupVersionKind, namespaces *[]namespace) error {
 
-	var wg sync.WaitGroup
-	wg.Add(len(*namespaces))
 	for _, ns := range *namespaces {
-		namespace := ns.name
-		go func() {
-			ch := make(chan groupOfObjects)
-			go gr.groupAppObjects(ctx, namespace, ch)
-
-			for groupOfObjects := range ch {
-				gr.logger.Info("reconcileNamespaceResources",
-					"Reconciling group of", len(groupOfObjects.objects),
-					"objects with labels", groupOfObjects.label,
-					"in the namespace", namespace)
-				err := gr.reconcileGroupOfObjects(ctx, groupOfObjects.objects, namespace)
-				if err != nil {
-					gr.logger.Error(err, "error reconciling group of ",
-						len(groupOfObjects.objects), "objects ", "in the namespace", namespace)
-				}
+		relatedObjects, err := gr.groupAppObjects(ctx, ns.name, gvks)
+		if err != nil {
+			return err
+		}
+		for label, objects := range relatedObjects {
+			gr.logger.Info("reconcileNamespaceResources",
+				"Reconciling group of", len(objects), "objects with labels", label,
+				"in the namespace", ns.name)
+			err := gr.reconcileGroupOfObjects(ctx, objects, ns.name)
+			if err != nil {
+				return fmt.Errorf(
+					"reconciling related objects with labels '%s': %w", label, err,
+				)
 			}
-			wg.Done()
-		}()
+		}
 	}
-	wg.Wait()
+
 	return nil
 }
 
@@ -411,7 +397,7 @@ func (gr *GenericReconciler) handleResourceDeletions() {
 }
 
 // getNamespacedResourcesGVK filters APIResources and returns the ones within a namespace
-func getNamespacedResourcesGVK(resources []metav1.APIResource) []schema.GroupVersionKind {
+func (gr GenericReconciler) getNamespacedResourcesGVK(resources []metav1.APIResource) []schema.GroupVersionKind {
 	namespacedResources := make([]schema.GroupVersionKind, 0)
 	for _, resource := range resources {
 		if resource.Namespaced {

pkg/controller/generic_reconciler_test.go

@@ -18,7 +18,6 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/util/rand"
 	kubefake "k8s.io/client-go/kubernetes/fake"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	clifake "sigs.k8s.io/controller-runtime/pkg/client/fake"
@@ -611,20 +610,15 @@ func TestGroupAppObjects(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			// create testing reconciler
-			gr, err := createTestReconciler(nil, tt.gvks, tt.objs)
+			gr, err := createTestReconciler(nil, tt.objs)
+			assert.NoError(t, err)
+			groupMap, err := gr.groupAppObjects(context.Background(), tt.namespace, tt.gvks)
 			assert.NoError(t, err)
-			ch := make(chan groupOfObjects)
-			go gr.groupAppObjects(context.Background(), tt.namespace, ch)
 
-			resultMap := make(map[string][]string)
-			for groupOfObjects := range ch {
-				actualNames := unstructuredToNames(groupOfObjects.objects)
-				resultMap[groupOfObjects.label] = actualNames
-			}
 			for expectedLabel, expectedNames := range tt.expectedNames {
-				actualNames, ok := resultMap[expectedLabel]
+				objects, ok := groupMap[expectedLabel]
 				assert.True(t, ok, "can't find label %s", expectedLabel)
-				//actualNames := unstructuredToNames(objects)
+				actualNames := unstructuredToNames(objects)
 				for _, expectedName := range expectedNames {
 					assert.Contains(t, actualNames, expectedName,
 						"can't find %s for label value %s", expectedName, expectedLabel)
@@ -692,7 +686,7 @@ func TestUnstructuredToTyped(t *testing.T) {
 			err := v1.AddToScheme(tt.scheme)
 			assert.NoError(t, err)
 
-			gr, err := createTestReconciler(tt.scheme, nil, nil)
+			gr, err := createTestReconciler(tt.scheme, nil)
 			assert.NoError(t, err)
 			o, err := gr.unstructuredToTyped(tt.u)
 			if tt.expectedError == nil {
@@ -742,8 +736,11 @@ func TestGetNamespacedResourcesGVK(t *testing.T) {
 
 	for _, ut := range unitTests {
 		t.Run(ut.name, func(t *testing.T) {
+			// Given
+			gr := GenericReconciler{}
+
 			// When
-			test := getNamespacedResourcesGVK(ut.arg)
+			test := gr.getNamespacedResourcesGVK(ut.arg)
 
 			// Assert
 			assert.Equal(t, ut.result, test)
@@ -864,12 +861,12 @@ func TestProcessNamespacedResources(t *testing.T) {
 			err = policyv1.AddToScheme(sch)
 			assert.NoError(t, err)
 
-			testReconciler, err := createTestReconciler(sch, tt.gvks, tt.objects)
+			testReconciler, err := createTestReconciler(sch, tt.objects)
 			assert.NoError(t, err)
 
 			// set some namespaces to be watched
 			testReconciler.watchNamespaces.setCache(tt.namespaces)
-			err = testReconciler.processNamespacedResources(context.Background(), tt.namespaces)
+			err = testReconciler.processNamespacedResources(context.Background(), tt.gvks, tt.namespaces)
 			assert.NoError(t, err)
 			for _, o := range tt.objects {
 				vr, ok := testReconciler.objectValidationCache.retrieve(o)
@@ -984,9 +981,9 @@ func TestHandleResourceDeletions(t *testing.T) {
 	for _, testCase := range tests {
 		tt := testCase
 		t.Run(tt.name, func(t *testing.T) {
-			testReconciler, err := createTestReconciler(nil, nil, nil)
+			testReconciler, err := createTestReconciler(nil, nil)
 			assert.NoError(t, err)
-			testReconciler.watchNamespaces.setCache(&tt.testNamespaces) // nolint:gosec
+			testReconciler.watchNamespaces.setCache(&tt.testNamespaces)
 
 			// store the test objects in the caches
 			for _, co := range tt.testCurrentObjects {
@@ -1019,13 +1016,12 @@ func TestHandleResourceDeletions(t *testing.T) {
 
 func TestListLimit(t *testing.T) {
 	os.Setenv(EnvResorucesPerListQuery, "2")
-	testReconciler, err := createTestReconciler(runtime.NewScheme(), nil, nil)
+	testReconciler, err := createTestReconciler(runtime.NewScheme(), nil)
 	assert.NoError(t, err)
 	assert.Equal(t, int64(2), testReconciler.listLimit)
 }
 
-func createTestReconciler(scheme *runtime.Scheme, gvks []schema.GroupVersionKind,
-	objects []client.Object) (*GenericReconciler, error) {
+func createTestReconciler(scheme *runtime.Scheme, objects []client.Object) (*GenericReconciler, error) {
 	cliBuilder := clifake.NewClientBuilder()
 	if scheme != nil {
 		cliBuilder.WithScheme(scheme)
@@ -1040,14 +1036,12 @@ func createTestReconciler(scheme *runtime.Scheme, gvks []schema.GroupVersionKind
 	if err != nil {
 		return nil, err
 	}
-	testGenericReconciler, err := NewGenericReconciler(client, cli.Discovery(), &configmap.Watcher{}, ve)
-	if err != nil {
-		return nil, err
-	}
-	testGenericReconciler.resourceGVKs = gvks
-	return testGenericReconciler, nil
+	return NewGenericReconciler(client, cli.Discovery(), &configmap.Watcher{}, ve)
 }
 
+/* Benchmark used reverted changes in d80ec1f. Preserving it just in case the changes come back in near future.
+// TODO :: Check the usefulness of preserving this later on the road
+
 // BenchmarkGroupAppObjects measures the performance of grouping Kubernetes objects based on their labels.
 // The benchmark focuses on a scenario where a Reconciler needs to group different objects based on the 'app' label.
 // # Benchmark configuration:
@@ -1148,3 +1142,4 @@ func generateDeployments(count int, namespace string) []client.Object {
 
 	return objects
 }
+*/

pkg/controller/validationscache.go

@@ -1,15 +1,11 @@
 package controller
 
 import (
-	"sync"
-
 	"github.com/app-sre/deployment-validation-operator/pkg/validations"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-var lock sync.Mutex
-
 type validationKey struct {
 	group, version, kind, namespace, name string
 	uid                                   types.UID
@@ -72,8 +68,6 @@ func (vc *validationCache) has(key validationKey) bool {
 // constraint: cached outcomes will be updated in-place for a given object and
 // consecutive updates will not preserve previous state.
 func (vc *validationCache) store(obj client.Object, outcome validations.ValidationOutcome) {
-	lock.Lock()
-	defer lock.Unlock()
 	key := newValidationKey(obj)
 	(*vc)[key] = newValidationResource(
 		newResourceversionVal(obj.GetResourceVersion()),
@@ -98,8 +92,6 @@ func (vc *validationCache) remove(obj client.Object) {
 
 // removeKey deletes a key, and its value, from the instance
 func (vc *validationCache) removeKey(key validationKey) {
-	lock.Lock()
-	defer lock.Unlock()
 	delete(*vc, key)
 }