Chunk constraint and vulnerability report uploads (#442)

This should improve overall system scalability by eliminating some very long running xactions

Author: michaeljguarino

internal/controller/constraint_controller.go

@@ -3,7 +3,9 @@ package controller
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"strings"
+	"time"
 
 	"gopkg.in/yaml.v3"
 
@@ -26,6 +28,7 @@ import (
 const (
 	missingLabelError    = "missing label while attempting to map a constraint status resource"
 	bundleDataAnnotation = "policy.plural.sh/constraintData"
+	constraintChunkSize  = 15
 )
 
 type BundleData struct {
@@ -89,11 +92,15 @@ func (r *ConstraintReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 
 	logger.Info("recording constraint", "name", pca.Name)
 	r.Constraints[pca.Name] = *pca
-	res, err := r.ConsoleClient.UpsertConstraints(algorithms.MapValues[string, console.PolicyConstraintAttributes](r.Constraints))
-	if err != nil {
-		return ctrl.Result{}, err
+
+	for _, chunk := range lo.Chunk(algorithms.MapValues[string, console.PolicyConstraintAttributes](r.Constraints), constraintChunkSize) {
+		res, err := r.ConsoleClient.UpsertConstraints(chunk)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		logger.Info("upsert constraint", "number", *res.UpsertPolicyConstraints)
+		time.Sleep(time.Duration(rand.Int63n(int64(500 * time.Millisecond))))
 	}
-	logger.Info("upsert constraint", "number", *res.UpsertPolicyConstraints)
 	return ctrl.Result{}, nil
 }
 

internal/controller/vulnerabilityreports_controller.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"strings"
 	"time"
 
@@ -33,6 +34,8 @@ const (
 
 	// vulnerabilityJitter defines the random jitter duration added to vulnerability processing to prevent synchronized operations.
 	vulnerabilityJitter = 2 * time.Minute
+
+	reportChunkSize = 15
 )
 
 // VulnerabilityReportReconciler reconciles a Trivy VulnerabilityReport resource.
@@ -188,13 +191,18 @@ func (r *VulnerabilityReportReconciler) SetupWithManager(mgr ctrl.Manager) error
 	err := helpers.BackgroundPollUntilContextCancel(r.Ctx, reportUploadInterval, false, true, func(_ context.Context) (done bool, err error) {
 		if !r.reports.IsEmpty() {
 			apiReports := algorithms.Map(lo.Values(r.reports.Items()), func(s console.VulnerabilityReportAttributes) *console.VulnerabilityReportAttributes { return &s })
-			if _, err := r.ConsoleClient.UpsertVulnerabilityReports(apiReports); err != nil {
-				logger.Error(err, "unable to upsert vulnerability reports")
-			} else {
-				// Clear the reports map and allow dangling elements to be garbage collected
-				r.reports.Clear()
-				r.reports = cmap.New[console.VulnerabilityReportAttributes]()
-				logger.Info("upsert vulnerability reports")
+
+			for _, chunk := range lo.Chunk(apiReports, reportChunkSize) {
+				if _, err := r.ConsoleClient.UpsertVulnerabilityReports(chunk); err != nil {
+					logger.Error(err, "unable to upsert vulnerability reports")
+				} else {
+					// Clear the reports map and allow dangling elements to be garbage collected
+					r.reports.Clear()
+					r.reports = cmap.New[console.VulnerabilityReportAttributes]()
+					logger.Info("upsert vulnerability reports")
+				}
+
+				time.Sleep(time.Duration(rand.Int63n(int64(500 * time.Millisecond))))
 			}
 		}
 		return false, nil