Implement weighted queues

api/v1alpha1/types.go

@@ -43,6 +43,13 @@ type ResourceGraphDefinitionSpec struct {
 	//
 	// +kubebuilder:validation:Optional
 	DefaultServiceAccounts map[string]string `json:"defaultServiceAccounts,omitempty"`
+	// The weight used to determine priority when queuing reconciliation.
+	//
+	// +kubebuilder:default=100
+	// +kubebuilder:validation:Maximum=1000
+	// +kubebuilder:validation:Minimum=1
+	// +kubebuilder:validation:Optional
+	Weight int `json:"weight,omitempty"`
 }
 
 // Schema represents the attributes that define an instance of

config/crd/bases/kro.run_resourcegraphdefinitions.yaml

@@ -145,6 +145,12 @@ spec:
                 - apiVersion
                 - kind
                 type: object
+              weight:
+                default: 100
+                description: The weight used to determine priority when queuing reconciliation.
+                maximum: 1000
+                minimum: 1
+                type: integer
             required:
             - schema
             type: object

examples/kubernetes/webapp/rg.yaml

@@ -3,6 +3,7 @@ kind: ResourceGraphDefinition
 metadata:
   name: webapp.kro.run
 spec:
+  weight: 150
   schema:
     apiVersion: v1alpha1
     kind: WebApp
@@ -68,7 +69,7 @@ spec:
 
   - id: service
     includeWhen:
-    - ${schema.spec.service.enabled}  
+    - ${schema.spec.service.enabled}
     template:
       apiVersion: v1
       kind: Service
@@ -86,7 +87,7 @@ spec:
 
   - id: ingress
     includeWhen:
-    - ${schema.spec.ingress.enabled}  
+    - ${schema.spec.ingress.enabled}
     template:
       apiVersion: networking.k8s.io/v1
       kind: Ingress

helm/crds/kro.run_resourcegraphdefinitions.yaml

@@ -145,6 +145,12 @@ spec:
                 - apiVersion
                 - kind
                 type: object
+              weight:
+                default: 100
+                description: The weight used to determine priority when queuing reconciliation.
+                maximum: 1000
+                minimum: 1
+                type: integer
             required:
             - schema
             type: object

pkg/controller/instance/controller.go

@@ -47,6 +47,9 @@ type ReconcileConfig struct {
 	// TODO(a-hilaly): need to define think the different deletion policies we need to
 	// support.
 	DeletionPolicy string
+
+	// Instance controller events are queued in a weighted queue.
+	Weight int
 }
 
 // Controller manages the reconciliation of a single instance of a ResourceGraphDefinition,

pkg/controller/resourcegraphdefinition/controller_reconcile.go

@@ -63,7 +63,7 @@ func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinition(ctx
 	controller := r.setupMicroController(gvr, processedRGD, rgd.Spec.DefaultServiceAccounts, graphExecLabeler)
 
 	log.V(1).Info("reconciling resource graph definition micro controller")
-	if err := r.reconcileResourceGraphDefinitionMicroController(ctx, &gvr, controller.Reconcile); err != nil {
+	if err := r.reconcileResourceGraphDefinitionMicroController(ctx, &gvr, controller.Reconcile, rgd.Spec.Weight); err != nil {
 		return processedRGD.TopologicalOrder, resourcesInfo, err
 	}
 
@@ -141,8 +141,8 @@ func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionCRD(
 }
 
 // reconcileResourceGraphDefinitionMicroController starts the microcontroller for handling the resources
-func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionMicroController(ctx context.Context, gvr *schema.GroupVersionResource, handler dynamiccontroller.Handler) error {
-	err := r.dynamicController.StartServingGVK(ctx, *gvr, handler)
+func (r *ResourceGraphDefinitionReconciler) reconcileResourceGraphDefinitionMicroController(ctx context.Context, gvr *schema.GroupVersionResource, handler dynamiccontroller.Handler, weight int) error {
+	err := r.dynamicController.StartServingGVK(ctx, *gvr, handler, weight)
 	if err != nil {
 		return newMicroControllerError(err)
 	}