Skip cache invalidation for SLTs (#260)

Co-authored-by: aalexand <[email protected]>

Author: aalexandru

cmd/apiserver/apiserver.go

@@ -129,6 +129,12 @@ func main() {
 			log.Errorf("Failed to delete message: %s", err.Error())
 			return
 		}
+
+		if val, ok := msg.MessageAttributes[sqs.MessageAttributeSkipCacheInvalidation]; ok && *val.StringValue == "true" {
+			log.Debugf("Skipping cache invalidation")
+			return
+		}
+
 		log.Debugf("Invalidating clusters cache")
 		err = cacheManager.Invalidate(context.Background(), store.WithInvalidateTags([]string{"clusters"}))
 		if err != nil {

local/sqs/sqs.go

@@ -93,6 +93,10 @@ func main() {
 						DataType:    aws.String("String"),
 						StringValue: aws.String(cluster.Spec.Name),
 					},
+					"SkipCacheInvalidation": {
+						DataType:    aws.String("Bool"),
+						StringValue: aws.String(fmt.Sprintf("%t", false)),
+					},
 				},
 				MessageBody: aws.String(string(data)),
 			},

pkg/client/controllers/cluster_controller.go

@@ -46,6 +46,9 @@ type ClusterReconciler struct {
 const (
 	// HashAnnotation ...
 	HashAnnotation = "registry.ethos.adobe.com/hash"
+
+	// SkipCacheInvalidationAnnotation ...
+	SkipCacheInvalidationAnnotation = "registry.ethos.adobe.com/skip-cache-invalidation"
 )
 
 //+kubebuilder:rbac:groups=registry.ethos.adobe.com,resources=clusters,verbs=get;list;watch;create;update;patch;delete
@@ -89,10 +92,18 @@ func (r *ClusterReconciler) ReconcileCreateUpdate(instance *registryv1.Cluster,
 	if annotations == nil {
 		annotations = make(map[string]string, 1)
 	}
+
 	annotations[HashAnnotation] = hash
+
+	skipCacheInvalidation := false
+	if _, ok := annotations[SkipCacheInvalidationAnnotation]; ok {
+		delete(annotations, SkipCacheInvalidationAnnotation)
+		skipCacheInvalidation = true
+	}
+
 	instance.SetAnnotations(annotations)
 
-	err := r.enqueue(instance)
+	err := r.enqueue(instance, skipCacheInvalidation)
 	if err != nil {
 		r.Log.Error(err, "error enqueuing message")
 		return ctrl.Result{}, err
@@ -132,7 +143,7 @@ func (r *ClusterReconciler) eventFilters() predicate.Predicate {
 	}
 }
 
-func (r *ClusterReconciler) enqueue(instance *registryv1.Cluster) error {
+func (r *ClusterReconciler) enqueue(instance *registryv1.Cluster, skipCacheInvalidation bool) error {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
@@ -160,6 +171,10 @@ func (r *ClusterReconciler) enqueue(instance *registryv1.Cluster) error {
 					DataType:    aws.String("String"),
 					StringValue: aws.String(instance.Spec.Name),
 				},
+				"SkipCacheInvalidation": {
+					DataType:    aws.String("String"),
+					StringValue: aws.String(fmt.Sprintf("%t", skipCacheInvalidation)),
+				},
 			},
 			MessageBody: aws.String(string(obj)),
 		},
@@ -182,12 +197,13 @@ func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 }
 
 // hashCluster returns a SHA256 hash of the Cluster object, after removing the ResourceVersion,
-// ManagedFields and hashCluster annotation
+// ManagedFields and hash/no-cache annotation
 func hashCluster(instance *registryv1.Cluster) string {
 	clone := instance.DeepCopyObject().(*registryv1.Cluster)
 
 	annotations := clone.GetAnnotations()
 	delete(annotations, HashAnnotation)
+	delete(annotations, SkipCacheInvalidationAnnotation)
 	clone.SetAnnotations(annotations)
 
 	clone.SetResourceVersion("")

pkg/sqs/event.go

@@ -19,8 +19,9 @@ import (
 )
 
 const (
-	MessageAttributeType        = "Type"
-	MessageAttributeClusterName = "ClusterName"
+	MessageAttributeType                  = "Type"
+	MessageAttributeClusterName           = "ClusterName"
+	MessageAttributeSkipCacheInvalidation = "SkipCacheInvalidation"
 
 	// ClusterUpdateEvent refers to an update of the Cluster object that
 	// is sent by the client controller. This event is sent to the SQS queue and

test/slt/checks/update/update.go

@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/adobe/cluster-registry/pkg/client/controllers"
 	"time"
 
 	h "github.com/adobe/cluster-registry/test/slt/helpers"
@@ -124,6 +125,14 @@ func updateCrd(namespace string) (string, string, error) {
 	// Remove immutable Kubernetes field
 	(*cluster).ObjectMeta.ManagedFields = []metav1.ManagedFieldsEntry{}
 
+	// Set SkipCacheInvalidationAnnotation to true
+	annotations := (*cluster).GetAnnotations()
+	if annotations == nil {
+		annotations = make(map[string]string, 1)
+	}
+	annotations[controllers.SkipCacheInvalidationAnnotation] = ""
+	(*cluster).SetAnnotations(annotations)
+
 	data, err := json.Marshal(*cluster)
 	if err != nil {
 		return "", "", fmt.Errorf("could not marshal updated CRD: %s", err.Error())
@@ -153,8 +162,7 @@ func checkAPIforUpdate(url, clusterName, tagSLTValue, jwtToken string) error {
 	if cluster.Tags == nil {
 		return errors.New("tags field is empty")
 	} else if tagSLTValue != cluster.Tags[tagSLT] {
-		return fmt.Errorf("the 'Tags' field is not what expected. The "+
-			"value is '%s', expected '%s'.", cluster.Tags[tagSLT], tagSLTValue)
+		return fmt.Errorf("the 'Tags' field is not what expected. The value is '%s', expected '%s'", cluster.Tags[tagSLT], tagSLTValue)
 	}
 
 	return nil
@@ -174,7 +182,7 @@ func Run(config TestConfig, jwtToken string) (int, error) {
 	for nrOfTries <= maxNrOfTries {
 		// Give to the CR client time to push to the SQS queue and for the API to read
 		// from the queue and update the DB. By local tests it takes around 11s
-		time.Sleep(11 * time.Second)
+		time.Sleep(30 * time.Second)
 
 		logger.Infof("checking the API for the update (check %d/%d)...",
 			nrOfTries, maxNrOfTries)

test/slt/helpers/helpers.go

@@ -127,14 +127,14 @@ func reqGet(endpoint, bearer string) (*[]byte, int, error) {
 func GetCluster(url, clusterName, jwtToken string) (*cr.ClusterSpec, error) {
 	var cluster cr.ClusterSpec
 
-	endpoint := fmt.Sprintf("%s/api/v1/clusters/%s", url, clusterName)
+	endpoint := fmt.Sprintf("%s/api/v2/clusters/%s", url, clusterName)
 	bearer := "Bearer " + jwtToken
 
 	start := time.Now()
 	body, respCode, err := reqGet(endpoint, bearer)
 	timeTook := float64(time.Since(start).Seconds())
 	metrics.EgressReqDuration.WithLabelValues(
-		"/api/v1/clusters/[cluster]",
+		"/api/v2/clusters/[cluster]",
 		"GET",
 		strconv.Itoa(respCode)).Observe(timeTook)
 	if err != nil {
@@ -156,14 +156,14 @@ func GetCluster(url, clusterName, jwtToken string) (*cr.ClusterSpec, error) {
 func GetClusters(url, perPageLimit, pageNr, jwtToken string) (*ClusterList, error) {
 	var clusters ClusterList
 
-	endpoint := fmt.Sprintf("%s/api/v1/clusters?offset=%s&limit=%s", url, pageNr, perPageLimit)
+	endpoint := fmt.Sprintf("%s/api/v2/clusters?offset=%s&limit=%s", url, pageNr, perPageLimit)
 	bearer := "Bearer " + jwtToken
 
 	start := time.Now()
 	body, respCode, err := reqGet(endpoint, bearer)
 	timeTook := float64(time.Since(start).Seconds())
 	metrics.EgressReqDuration.WithLabelValues(
-		"/api/v1/clusters",
+		"/api/v2/clusters",
 		"GET",
 		strconv.Itoa(respCode)).Observe(timeTook)
 	if err != nil {

test/slt/release.sh

@@ -24,7 +24,7 @@ IMAGE_SLT="${IMAGE_SLT:-"${default_image_name}"}"
 IMAGE_SLT="${IMAGE_SLT}${IMAGE_SUFFIX}"
 
 
-printf "Realeasing image %s...\n\n" "${IMAGE_SLT}:${TAG}"
+printf "Releasing image %s...\n\n" "${IMAGE_SLT}:${TAG}"
 
 make -C "${ROOT_DIR}" --always-make build-slt \
     TAG="${TAG}" \