chore(controller): spread out API sync requests

Rate-limit API sync requests on the client side to avoid hitting the
backend's rate limit.

Author: basti1302

internal/controller/api_sync_common.go

@@ -13,13 +13,16 @@ import (
 	"slices"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
 	dash0common "github.com/dash0hq/dash0-operator/api/operator/common"
 	dash0v1beta1 "github.com/dash0hq/dash0-operator/api/operator/v1beta1"
 	"github.com/dash0hq/dash0-operator/internal/resources"
 	"github.com/dash0hq/dash0-operator/internal/util"
 	"github.com/dash0hq/dash0-operator/internal/util/logd"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"sigs.k8s.io/controller-runtime/pkg/client"
+	"github.com/dash0hq/dash0-operator/internal/util/rate"
+	utilResources "github.com/dash0hq/dash0-operator/internal/util/resources"
 )
 
 type ApiConfig struct {
@@ -337,6 +340,7 @@ func synchronizeViaApiAndUpdateStatus(
 	dash0ApiResource *unstructured.Unstructured,
 	ownedResource client.Object,
 	action apiAction,
+	limiter *rate.CappedRateLimiter,
 	logger logd.Logger,
 ) {
 	preconditionChecksResult := validatePreconditionsAndPreprocess(
@@ -349,6 +353,18 @@ func synchronizeViaApiAndUpdateStatus(
 		return
 	}
 
+	// limiter.Wait deliberately slows down API sync requests to avoid running into the backend's rate limit,
+	// at the cost of keeping the associated goroutine. It also protects against an ever-growing backlog of API sync
+	// operations by dropping requests when the number of pending items exceeds the cap.
+	if proceed := limiter.Wait(
+		ctx,
+		apiSyncReconciler.KindDisplayName(),
+		utilResources.QualifiedResourceName(dash0ApiResource),
+		logger,
+	); !proceed {
+		return
+	}
+
 	resourceHasBeenDeleted := action == deleteAction
 
 	if preconditionChecksResult.syncDisabledViaLabel {

internal/controller/perses_dashboards_controller.go

@@ -32,6 +32,7 @@ import (
 	"github.com/dash0hq/dash0-operator/internal/selfmonitoringapiaccess"
 	"github.com/dash0hq/dash0-operator/internal/util"
 	"github.com/dash0hq/dash0-operator/internal/util/logd"
+	"github.com/dash0hq/dash0-operator/internal/util/rate"
 )
 
 type PersesDashboardCrdReconciler struct {
@@ -40,6 +41,7 @@ type PersesDashboardCrdReconciler struct {
 	leaderElectionAware       util.LeaderElectionAware
 	mgr                       ctrl.Manager
 	httpClient                *http.Client
+	limiter                   *rate.CappedRateLimiter
 	skipNameValidation        bool
 	persesDashboardReconciler *PersesDashboardReconciler
 	persesDashboardCrdExists  atomic.Bool
@@ -50,6 +52,7 @@ type PersesDashboardReconciler struct {
 	pseudoClusterUid           types.UID
 	queue                      *workqueue.Typed[ThirdPartyResourceSyncJob]
 	httpClient                 *http.Client
+	limiter                    *rate.CappedRateLimiter
 	defaultApiConfigs          selfmonitoringapiaccess.SynchronizedSlice[ApiConfig]
 	namespacedApiConfigs       selfmonitoringapiaccess.SynchronizedMapSlice[ApiConfig]
 	namespacedSyncEnabled      sync.Map
@@ -67,12 +70,14 @@ func NewPersesDashboardCrdReconciler(
 	queue *workqueue.Typed[ThirdPartyResourceSyncJob],
 	leaderElectionAware util.LeaderElectionAware,
 	httpClient *http.Client,
+	limiter *rate.CappedRateLimiter,
 ) *PersesDashboardCrdReconciler {
 	return &PersesDashboardCrdReconciler{
 		Client:              k8sClient,
 		queue:               queue,
 		leaderElectionAware: leaderElectionAware,
 		httpClient:          httpClient,
+		limiter:             limiter,
 	}
 }
 
@@ -126,6 +131,7 @@ func (r *PersesDashboardCrdReconciler) CreateThirdPartyResourceReconciler(pseudo
 		queue:                r.queue,
 		pseudoClusterUid:     pseudoClusterUid,
 		httpClient:           r.httpClient,
+		limiter:              r.limiter,
 		defaultApiConfigs:    *selfmonitoringapiaccess.NewSynchronizedSlice[ApiConfig](),
 		namespacedApiConfigs: *selfmonitoringapiaccess.NewSynchronizedMapSlice[ApiConfig](),
 	}
@@ -723,9 +729,6 @@ func (r *PersesDashboardReconciler) synchronizeNamespacedResources(
 				Object: dashboardResource,
 			}
 			r.Generic(ctx, evt, nil)
-
-			// stagger API requests a bit
-			time.Sleep(50 * time.Millisecond)
 		}
 		logger.Info(fmt.Sprintf("Triggering synchronization of dashboards in namespace %s has finished.", namespace))
 	}()

internal/controller/perses_dashboards_controller_test.go

@@ -27,6 +27,7 @@ import (
 	dash0v1beta1 "github.com/dash0hq/dash0-operator/api/operator/v1beta1"
 	"github.com/dash0hq/dash0-operator/internal/util"
 	"github.com/dash0hq/dash0-operator/internal/util/logd"
+	"github.com/dash0hq/dash0-operator/internal/util/rate"
 
 	"github.com/h2non/gock"
 	. "github.com/onsi/ginkgo/v2"
@@ -392,7 +393,11 @@ var _ = Describe(
 
 						Expect(persesDashboardCrdReconciler.SetupWithManager(ctx, mgr, k8sClient, logger)).To(Succeed())
 
-						StartProcessingThirdPartySynchronizationQueue(testQueuePersesDashboards, logger)
+						StartProcessingThirdPartySynchronizationQueue(
+							testQueuePersesDashboards,
+							rate.NewNoOpCappedRateLimiter(),
+							logger,
+						)
 					},
 				)
 
@@ -1018,6 +1023,7 @@ func createPersesDashboardCrdReconciler() *PersesDashboardCrdReconciler {
 		testQueuePersesDashboards,
 		&DummyLeaderElectionAware{Leader: true},
 		TestHTTPClient(),
+		rate.NewNoOpCappedRateLimiter(),
 	)
 
 	// We create the controller multiple times in tests, this option is required, otherwise the controller

internal/controller/prometheus_rules_controller.go

@@ -37,6 +37,7 @@ import (
 	"github.com/dash0hq/dash0-operator/internal/selfmonitoringapiaccess"
 	"github.com/dash0hq/dash0-operator/internal/util"
 	"github.com/dash0hq/dash0-operator/internal/util/logd"
+	"github.com/dash0hq/dash0-operator/internal/util/rate"
 )
 
 type PrometheusRuleCrdReconciler struct {
@@ -45,6 +46,7 @@ type PrometheusRuleCrdReconciler struct {
 	leaderElectionAware      util.LeaderElectionAware
 	mgr                      ctrl.Manager
 	httpClient               *http.Client
+	limiter                  *rate.CappedRateLimiter
 	skipNameValidation       bool
 	prometheusRuleReconciler *PrometheusRuleReconciler
 	prometheusRuleCrdExists  atomic.Bool
@@ -55,6 +57,7 @@ type PrometheusRuleReconciler struct {
 	pseudoClusterUid           types.UID
 	queue                      *workqueue.Typed[ThirdPartyResourceSyncJob]
 	httpClient                 *http.Client
+	limiter                    *rate.CappedRateLimiter
 	defaultApiConfigs          selfmonitoringapiaccess.SynchronizedSlice[ApiConfig]
 	namespacedApiConfigs       selfmonitoringapiaccess.SynchronizedMapSlice[ApiConfig]
 	namespacedSyncEnabled      sync.Map
@@ -92,12 +95,14 @@ func NewPrometheusRuleCrdReconciler(
 	queue *workqueue.Typed[ThirdPartyResourceSyncJob],
 	leaderElectionAware util.LeaderElectionAware,
 	httpClient *http.Client,
+	limiter *rate.CappedRateLimiter,
 ) *PrometheusRuleCrdReconciler {
 	return &PrometheusRuleCrdReconciler{
 		Client:              k8sClient,
 		queue:               queue,
 		leaderElectionAware: leaderElectionAware,
 		httpClient:          httpClient,
+		limiter:             limiter,
 	}
 }
 
@@ -151,6 +156,7 @@ func (r *PrometheusRuleCrdReconciler) CreateThirdPartyResourceReconciler(pseudoC
 		queue:                r.queue,
 		pseudoClusterUid:     pseudoClusterUid,
 		httpClient:           r.httpClient,
+		limiter:              r.limiter,
 		defaultApiConfigs:    *selfmonitoringapiaccess.NewSynchronizedSlice[ApiConfig](),
 		namespacedApiConfigs: *selfmonitoringapiaccess.NewSynchronizedMapSlice[ApiConfig](),
 	}
@@ -559,7 +565,7 @@ func (r *PrometheusRuleReconciler) MapResourceToHttpRequests(
 				allValidationIssues[checkRuleName] = validationIssues
 				// If a rule becomes temporarily invalid due to a bad edit, we do not want to delete it in Dash0
 				// (assuming it has been valid at some point before and has been synchronized). Instead, we keep the
-				// most recent valid state. To do that, we add its id to the list of ids we have seen (the list will
+				// most recent valid state. To do that, we add its id to the list of ids we have seen. The list will
 				// later be used to determine which of the checks in Dash0 need to be removed because they are no longer
 				// in the K8s resource.
 				originsInResource = append(originsInResource, checkRuleOriginNotUrlEncoded)
@@ -568,7 +574,7 @@ func (r *PrometheusRuleReconciler) MapResourceToHttpRequests(
 			if syncError != nil {
 				// If a rule cannot be synchronized temporarily, we do not want to delete it in Dash0 immediately
 				// (assuming the rule has been synchronized before at some point). Instead, we keep the
-				// most recent state. To do that, we add its id to the list of ids we have seen (the list will later
+				// most recent state. To do that, we add its id to the list of ids we have seen. The list will later
 				// be used to determine which of the checks in Dash0 need to be removed because they are no longer
 				// in the K8s resource.
 				allSynchronizationErrors[checkRuleName] = syncError.Error()
@@ -1133,9 +1139,6 @@ func (r *PrometheusRuleReconciler) synchronizeNamespacedResources(
 				Object: ruleResource,
 			}
 			r.Generic(ctx, evt, nil)
-
-			// stagger API requests a bit
-			time.Sleep(50 * time.Millisecond)
 		}
 		logger.Info(fmt.Sprintf("Triggering synchronization of check rules in namespace %s has finished.", namespace))
 	}()

internal/controller/prometheus_rules_controller_test.go

@@ -31,6 +31,7 @@ import (
 	dash0v1beta1 "github.com/dash0hq/dash0-operator/api/operator/v1beta1"
 	"github.com/dash0hq/dash0-operator/internal/util"
 	"github.com/dash0hq/dash0-operator/internal/util/logd"
+	"github.com/dash0hq/dash0-operator/internal/util/rate"
 
 	"github.com/h2non/gock"
 	. "github.com/onsi/ginkgo/v2"
@@ -397,7 +398,11 @@ var _ = Describe(
 
 						Expect(prometheusRuleCrdReconciler.SetupWithManager(ctx, mgr, k8sClient, logger)).To(Succeed())
 
-						StartProcessingThirdPartySynchronizationQueue(testQueuePrometheusRules, logger)
+						StartProcessingThirdPartySynchronizationQueue(
+							testQueuePrometheusRules,
+							rate.NewNoOpCappedRateLimiter(),
+							logger,
+						)
 					},
 				)
 
@@ -2215,6 +2220,7 @@ func createPrometheusRuleCrdReconciler() *PrometheusRuleCrdReconciler {
 		testQueuePrometheusRules,
 		&DummyLeaderElectionAware{Leader: true},
 		TestHTTPClient(),
+		rate.NewNoOpCappedRateLimiter(),
 	)
 
 	// We create the controller multiple times in tests, this option is required, otherwise the controller

internal/controller/synthetic_check_controller.go

@@ -34,13 +34,15 @@ import (
 	"github.com/dash0hq/dash0-operator/internal/selfmonitoringapiaccess"
 	"github.com/dash0hq/dash0-operator/internal/util"
 	"github.com/dash0hq/dash0-operator/internal/util/logd"
+	"github.com/dash0hq/dash0-operator/internal/util/rate"
 )
 
 type SyntheticCheckReconciler struct {
 	client.Client
 	pseudoClusterUid      types.UID
 	leaderElectionAware   util.LeaderElectionAware
 	httpClient            *http.Client
+	limiter               *rate.CappedRateLimiter
 	defaultApiConfigs     selfmonitoringapiaccess.SynchronizedSlice[ApiConfig]
 	namespacedApiConfigs  selfmonitoringapiaccess.SynchronizedMapSlice[ApiConfig]
 	initialSyncMutex      sync.Mutex
@@ -57,12 +59,14 @@ func NewSyntheticCheckReconciler(
 	pseudoClusterUid types.UID,
 	leaderElectionAware util.LeaderElectionAware,
 	httpClient *http.Client,
+	limiter *rate.CappedRateLimiter,
 ) *SyntheticCheckReconciler {
 	return &SyntheticCheckReconciler{
 		Client:               k8sClient,
 		pseudoClusterUid:     pseudoClusterUid,
 		leaderElectionAware:  leaderElectionAware,
 		httpClient:           httpClient,
+		limiter:              limiter,
 		defaultApiConfigs:    *selfmonitoringapiaccess.NewSynchronizedSlice[ApiConfig](),
 		namespacedApiConfigs: *selfmonitoringapiaccess.NewSynchronizedMapSlice[ApiConfig](),
 		namespacedSyncMutex:  *selfmonitoringapiaccess.NewNamespaceMutex(),
@@ -231,9 +235,8 @@ func (r *SyntheticCheckReconciler) maybeDoInitialSynchronizationOfAllResources(
 				},
 			}
 			_, _ = r.Reconcile(ctx, pseudoReconcileRequest)
-			// stagger API requests a bit
-			time.Sleep(50 * time.Millisecond)
 		}
+
 		logger.Info("Initial synchronization of synthetic checks has finished.")
 		r.initialSyncHasHappend.Store(true)
 	}()
@@ -284,8 +287,6 @@ func (r *SyntheticCheckReconciler) synchronizeNamespacedResources(
 				},
 			}
 			_, _ = r.Reconcile(ctx, pseudoReconcileRequest)
-			// stagger API requests a bit
-			time.Sleep(50 * time.Millisecond)
 		}
 		logger.Info(fmt.Sprintf("Synchronization of synthetic checks in namespace %s has finished.", namespace))
 	}()
@@ -345,6 +346,7 @@ func (r *SyntheticCheckReconciler) Reconcile(ctx context.Context, req reconcile.
 		unstructuredResource,
 		syntheticCheckResource,
 		action,
+		r.limiter,
 		logger,
 	)
 

internal/controller/synthetic_check_controller_test.go

@@ -22,6 +22,7 @@ import (
 	dash0v1alpha1 "github.com/dash0hq/dash0-operator/api/operator/v1alpha1"
 	"github.com/dash0hq/dash0-operator/internal/util"
 	"github.com/dash0hq/dash0-operator/internal/util/logd"
+	"github.com/dash0hq/dash0-operator/internal/util/rate"
 
 	"github.com/h2non/gock"
 	. "github.com/onsi/ginkgo/v2"
@@ -831,6 +832,7 @@ func createSyntheticCheckReconciler(clusterId string) *SyntheticCheckReconciler
 		types.UID(clusterId),
 		leaderElectionAware,
 		TestHTTPClient(),
+		rate.NewNoOpCappedRateLimiter(),
 	)
 	return syntheticCheckReconciler
 }

internal/controller/third_party_crd_common.go

@@ -33,6 +33,7 @@ import (
 	dash0v1beta1 "github.com/dash0hq/dash0-operator/api/operator/v1beta1"
 	"github.com/dash0hq/dash0-operator/internal/util"
 	"github.com/dash0hq/dash0-operator/internal/util/logd"
+	"github.com/dash0hq/dash0-operator/internal/util/rate"
 )
 
 // ThirdPartyCrdReconciler is an interface for reconcilers that act on CRDs of third-party resource types (i.e. when a
@@ -430,7 +431,11 @@ func upsertViaApi(
 	// attempts that are related to third-party resources types (Prometheus rules, Perses dashboards) end with writing
 	// to the Dash0 monitoring resource status in the same namespace. These write attempts will fail if they happen
 	// concurrently. Therefore, we add all synchronization attempts to one queue that is shared across resource types,
-	// and only process them one at a time.
+	// and only process them one at a time. This queueing is independent of the rate limiting which happens later in
+	// api_sync_common#synchronizeViaApiAndUpdateStatus via limiter.Wait(ctx), which spreads out API request for all
+	// types (not only third-party CRDs) to avoid running into Dash0 API rate limits. That rate limiting does not offer
+	// any guarantee that API sync operations (including updating the Dash0 monitoring resource's status) are handled
+	// sequentially.
 	thirdPartyResourceReconciler.Queue().Add(
 		ThirdPartyResourceSyncJob{
 			thirdPartyResourceReconciler: thirdPartyResourceReconciler,
@@ -456,6 +461,7 @@ func deleteViaApi(
 
 func StartProcessingThirdPartySynchronizationQueue(
 	thirdPartyResourceSynchronizationQueue *workqueue.Typed[ThirdPartyResourceSyncJob],
+	limiter *rate.CappedRateLimiter,
 	setupLog logd.Logger,
 ) {
 	setupLog.Info("Starting the Dash0 API resource synchronization queue.")
@@ -483,6 +489,7 @@ func StartProcessingThirdPartySynchronizationQueue(
 				item.dash0ApiResource,
 				nil,
 				item.action,
+				limiter,
 				logger,
 			)
 			logger.Info(