address comments

Signed-off-by: Zhiying Lin <[email protected]>

Author: zhiying-lin

apis/placement/v1/resourcesnapshot_types.go

@@ -51,6 +51,9 @@ import (
 //
 // Each snapshot (excluding the first snapshot) MUST have the following annotations:
 //   - `SubindexOfResourceSnapshotAnnotation` to store the subindex of resource snapshot in the group.
+//
+// Snapshot may have the following annotations to indicate the time of next resourceSnapshot candidate detected by the controller:
+//   - `NextResourceSnapshotCandidateDetectionTime` to store the time of next resourceSnapshot candidate detected by the controller.
 type ClusterResourceSnapshot struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`

apis/placement/v1beta1/resourcesnapshot_types.go

@@ -40,6 +40,9 @@ const (
 	// SubindexOfResourceSnapshotAnnotation is the annotation to store the subindex of resource snapshot in the group.
 	SubindexOfResourceSnapshotAnnotation = fleetPrefix + "subindex-of-resource-snapshot"
 
+	// NextResourceSnapshotCandidateDetectionTimeAnnotation is the annotation to store the time of next resourceSnapshot candidate detected by the controller.
+	NextResourceSnapshotCandidateDetectionTimeAnnotation = fleetPrefix + "next-resource-snapshot-candidate-detection-time"
+
 	// ResourceSnapshotNameFmt is resourcePolicySnapshot name format: {CRPName}-{resourceIndex}-snapshot.
 	ResourceSnapshotNameFmt = "%s-%d-snapshot"
 

charts/hub-agent/README.md

@@ -40,4 +40,4 @@ _See [helm install](https://helm.sh/docs/helm/helm_install/) for command documen
 | `ConcurrentResourceChangeSyncs`    | Max concurrent resourceChange reconcilers                                                                           | `20`                                             |
 | `logFileMaxSize`                  | Max log file size before rotation                                                                                   | `1000000`                                        |
 | `MaxFleetSizeSupported`            | Max number of member clusters supported                                                                             | `100`                                            |
-| `resourceSnapshotCreationInterval` | Interval for resource snapshot creation                                                                             | `30s`                                             |
+| `resourceSnapshotCreationInterval` | Interval for resource snapshot creation                                                                             | `1m`                                             |

charts/hub-agent/values.yaml

@@ -18,7 +18,7 @@ enableGuardRail: true
 webhookClientConnectionType: service
 forceDeleteWaitTime: 15m0s
 clusterUnhealthyThreshold: 3m0s
-resourceSnapshotCreationInterval: 30s
+resourceSnapshotCreationInterval: 1m
 
 namespace:
   fleet-system

cmd/hubagent/options/options.go

@@ -125,7 +125,7 @@ func NewOptions() *Options {
 		EnableStagedUpdateRunAPIs:        true,
 		EnablePprof:                      false,
 		PprofPort:                        6065,
-		ResourceSnapshotCreationInterval: 30 * time.Second,
+		ResourceSnapshotCreationInterval: 1 * time.Minute,
 	}
 }
 
@@ -172,7 +172,7 @@ func (o *Options) AddFlags(flags *flag.FlagSet) {
 	flags.BoolVar(&o.EnablePprof, "enable-pprof", false, "If set, the pprof profiling is enabled.")
 	flags.IntVar(&o.PprofPort, "pprof-port", 6065, "The port for pprof profiling.")
 	flags.BoolVar(&o.DenyModifyMemberClusterLabels, "deny-modify-member-cluster-labels", false, "If set, users not in the system:masters cannot modify member cluster labels.")
-	flags.DurationVar(&o.ResourceSnapshotCreationInterval, "resource-snapshot-creation-interval", 30*time.Second, "The interval at which resource snapshots are created.")
+	flags.DurationVar(&o.ResourceSnapshotCreationInterval, "resource-snapshot-creation-interval", 1*time.Minute, "The interval at which resource snapshots are created.")
 
 	o.RateLimiterOpts.AddFlags(flags)
 }

pkg/controllers/clusterresourceplacement/controller.go

@@ -536,12 +536,13 @@ func (r *Reconciler) getOrCreateClusterResourceSnapshot(ctx context.Context, crp
 	if latestResourceSnapshot != nil && latestResourceSnapshotHash != resourceHash && latestResourceSnapshot.Labels[fleetv1beta1.IsLatestSnapshotLabel] == strconv.FormatBool(true) {
 		// When the latest resource snapshot without the isLastest label, it means it fails to create the new
 		// resource snapshot in the last reconcile and we don't need to check and delay the request.
-		if since := time.Since(latestResourceSnapshot.CreationTimestamp.Time); since < r.ResourceSnapshotCreationInterval {
-			// If the latest resource snapshot is created less than configured the resourceSnapshotCreationInterval,
-			//  requeue the request to avoid too frequent update.
-			klog.V(2).InfoS("The latest resource snapshot is just created, skipping the update", "clusterResourcePlacement", crpKObj,
-				"clusterResourceSnapshot", klog.KObj(latestResourceSnapshot), "creationTime", latestResourceSnapshot.CreationTimestamp, "configuredResourceSnapshotCreationInterval", r.ResourceSnapshotCreationInterval, "afterDuration", r.ResourceSnapshotCreationInterval-since)
-			return ctrl.Result{Requeue: true, RequeueAfter: r.ResourceSnapshotCreationInterval - since}, latestResourceSnapshot, nil
+		res, error := r.shouldCreateNewResourceSnapshotNow(ctx, latestResourceSnapshot)
+		if error != nil {
+			return ctrl.Result{}, nil, error
+		}
+		if res.Requeue {
+			// If the latest resource snapshot is not ready to be updated, we requeue the request.
+			return res, latestResourceSnapshot, nil
 		}
 
 		// set the latest label to false first to make sure there is only one or none active resource snapshot
@@ -587,6 +588,48 @@ func (r *Reconciler) getOrCreateClusterResourceSnapshot(ctx context.Context, crp
 	return ctrl.Result{}, latestResourceSnapshot, nil
 }
 
+// shouldCreateNewResourceSnapshotNow checks whether it is ready to create the new resource snapshot to avoid too frequent creation
+// based on the configured ResourceSnapshotCreationInterval.
+func (r *Reconciler) shouldCreateNewResourceSnapshotNow(ctx context.Context, latestResourceSnapshot *fleetv1beta1.ClusterResourceSnapshot) (ctrl.Result, error) {
+	// We reserve half of the resourceSnapshotCreationInterval to allow the controller to bundle all the resource changes into one snapshot.
+	// For example, if the interval is 1m, the first resource change will be captured starting from 30s.
+	// And then the next 30s will be used to caputre all the resource changes into one snapshot.
+	snapshotKObj := klog.KObj(latestResourceSnapshot)
+	now := time.Now()
+	half := r.ResourceSnapshotCreationInterval / 2
+	if since := now.Sub(latestResourceSnapshot.CreationTimestamp.Time); since < half {
+		// If the latest resource snapshot is created less than configured the resourceSnapshotCreationInterval,
+		//  requeue the request to avoid too frequent update.
+		klog.V(2).InfoS("The latest resource snapshot is just created, skipping the update",
+			"clusterResourceSnapshot", snapshotKObj, "creationTime", latestResourceSnapshot.CreationTimestamp, "configuredResourceSnapshotCreationInterval", r.ResourceSnapshotCreationInterval, "afterDuration", half-since)
+		return ctrl.Result{Requeue: true, RequeueAfter: half - since}, nil
+	}
+	nextResourceSnapshotCandidateDetectionTime, err := annotations.ExtractNextResourceSnapshotCandidateDetectionTimeFromResourceSnapshot(latestResourceSnapshot)
+	if nextResourceSnapshotCandidateDetectionTime.IsZero() || err != nil {
+		if err != nil {
+			klog.ErrorS(controller.NewUnexpectedBehaviorError(err), "Failed to get the NextResourceSnapshotCandidateDetectionTimeAnnotation", "clusterResourceSnapshot", snapshotKObj)
+		}
+		// If the annoation is not set, set next resource snapshot candidate detection time is now.
+		if latestResourceSnapshot.Annotations == nil {
+			latestResourceSnapshot.Annotations = make(map[string]string)
+		}
+		latestResourceSnapshot.Annotations[fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation] = now.Format(time.RFC3339)
+		if err := r.Client.Update(ctx, latestResourceSnapshot); err != nil {
+			klog.ErrorS(err, "Failed to update the NextResourceSnapshotCandidateDetectionTimeAnnotation", "clusterResourceSnapshot", snapshotKObj)
+			return ctrl.Result{}, controller.NewUpdateIgnoreConflictError(err)
+		}
+		klog.V(2).InfoS("Set the next-resource-snapshot-candidate-detection-time to now", "clusterResourceSnapshot", snapshotKObj, "nextResourceSnapshotCandidateDetectionTime", now)
+		return ctrl.Result{Requeue: true, RequeueAfter: half}, nil
+	}
+	if delay := now.Sub(nextResourceSnapshotCandidateDetectionTime); delay < half {
+		// If the next resource snapshot candidate detection time is not reached, we requeue the request to avoid too frequent update.
+		klog.V(2).InfoS("The next resource snapshot candidate has not reached the half of the ResourceSnapshotCreationInterval, skipping the creation",
+			"clusterResourceSnapshot", snapshotKObj, "nextResourceSnapshotCandidateDetectionTime", nextResourceSnapshotCandidateDetectionTime, "configuredResourceSnapshotCreationInterval", r.ResourceSnapshotCreationInterval, "afterDuration", half-delay)
+		return ctrl.Result{Requeue: true, RequeueAfter: half - delay}, nil
+	}
+	return ctrl.Result{}, nil
+}
+
 // buildMasterClusterResourceSnapshot builds and returns the master cluster resource snapshot for the latest resource snapshot index and selected resources.
 func buildMasterClusterResourceSnapshot(latestResourceSnapshotIndex, resourceSnapshotCount, envelopeObjCount int, crpName, resourceHash string, selectedResources []fleetv1beta1.ResourceContent) *fleetv1beta1.ClusterResourceSnapshot {
 	return &fleetv1beta1.ClusterResourceSnapshot{

pkg/controllers/clusterresourceplacement/controller_test.go

@@ -4477,3 +4477,106 @@ func TestDetermineRolloutStateForCRPWithExternalRolloutStrategy(t *testing.T) {
 		})
 	}
 }
+
+func TestShouldCreateNewResourceSnapshotNow(t *testing.T) {
+	interval := 60 * time.Second
+	half := interval / 2
+	now := time.Now()
+
+	cases := []struct {
+		name            string
+		creationTime    time.Time
+		annotationValue string
+		wantAnnoation   bool
+		wantRequeue     bool
+	}{
+		{
+			name:         "recently created resource snapshot",
+			creationTime: now.Add(-half + 5*time.Second),
+			wantRequeue:  true,
+		},
+		{
+			name:         "existing resource snapshot without annotation",
+			creationTime: now.Add(-2 * half),
+			// no annotation → sets it and requeues
+			annotationValue: "",
+			wantAnnoation:   true,
+			wantRequeue:     true,
+		},
+		{
+			name:            "existing resource snapshot with expired annotation",
+			creationTime:    now.Add(-2 * half),
+			annotationValue: now.Add(-half).Format(time.RFC3339),
+			// annotation far in past → no requeue
+			wantAnnoation: true,
+			wantRequeue:   false,
+		},
+		{
+			name:            "existing resource snapshot with invalid annotation",
+			creationTime:    now.Add(-2 * half),
+			annotationValue: "invalid-value",
+			// annotation far in past → no requeue
+			wantAnnoation: true,
+			wantRequeue:   true,
+		},
+		{
+			name:            "existing resource snapshot with recently added annotation",
+			creationTime:    now.Add(-2 * half),
+			annotationValue: now.Add(-half + 5*time.Second).Format(time.RFC3339),
+			// annotation exists but within half interval → requeue
+			wantAnnoation: true,
+			wantRequeue:   true,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			// initialize a snapshot with given creation time and annotation
+			snapshot := &fleetv1beta1.ClusterResourceSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "test-snapshot",
+					CreationTimestamp: metav1.Time{Time: tc.creationTime},
+					Annotations:       map[string]string{},
+				},
+			}
+			if tc.annotationValue != "" {
+				snapshot.Annotations[fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation] = tc.annotationValue
+			}
+
+			// use fake client seeded with the snapshot
+			scheme := serviceScheme(t)
+			client := fake.NewClientBuilder().
+				WithScheme(scheme).
+				WithRuntimeObjects(snapshot.DeepCopy()).
+				Build()
+
+			r := &Reconciler{
+				Client:                           client,
+				ResourceSnapshotCreationInterval: interval,
+			}
+
+			ctx := context.Background()
+			if err := client.Get(ctx, types.NamespacedName{Name: snapshot.Name}, snapshot); err != nil {
+				t.Fatalf("Failed to get snapshot: %v", err)
+			}
+			got, err := r.shouldCreateNewResourceSnapshotNow(ctx, snapshot)
+			if err != nil {
+				t.Fatalf("shouldCreateNewResourceSnapshotNow() failed: %v", err)
+			}
+			if got.Requeue != tc.wantRequeue {
+				t.Errorf("shouldCreateNewResourceSnapshotNow() = %v, want %v", got.Requeue, tc.wantRequeue)
+			}
+			if tc.wantRequeue {
+				if got.RequeueAfter <= 0 || got.RequeueAfter > half {
+					t.Errorf("shouldCreateNewResourceSnapshotNow() = RequeueAfter got %v, want in (0, %v], ", got.RequeueAfter, half)
+				}
+			}
+			if err := client.Get(ctx, types.NamespacedName{Name: snapshot.Name}, snapshot); err != nil {
+				t.Fatalf("failed to get snapshot after shouldCreateNewResourceSnapshotNow: %v", err)
+			}
+			if gotAnnotation := len(snapshot.Annotations[fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation]) != 0; tc.wantAnnoation != gotAnnotation {
+				t.Errorf("shouldCreateNewResourceSnapshotNow() = annotation %v, want %v", snapshot.Annotations[fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation], tc.wantAnnoation)
+			}
+		})
+	}
+}

pkg/utils/annotations/annotations.go

@@ -20,6 +20,7 @@ package annotations
 import (
 	"fmt"
 	"strconv"
+	"time"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -98,3 +99,17 @@ func ExtractNumberOfEnvelopeObjFromResourceSnapshot(snapshot *fleetv1beta1.Clust
 	}
 	return envelopeObjCount, nil
 }
+
+// ExtractNextResourceSnapshotCandidateDetectionTimeFromResourceSnapshot extracts the next resource snapshot candidate detection time from the annotations of a clusterResourceSnapshot.
+// If the annotation does not exist, it returns 0 duration.
+func ExtractNextResourceSnapshotCandidateDetectionTimeFromResourceSnapshot(snapshot *fleetv1beta1.ClusterResourceSnapshot) (time.Time, error) {
+	nextDetectionTimeStr, ok := snapshot.Annotations[fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation]
+	if !ok {
+		return time.Time{}, nil
+	}
+	nextDetectionTime, err := time.Parse(time.RFC3339, nextDetectionTimeStr)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("invalid annotation %s: %s is not a valid RFC3339 time: %w", fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation, nextDetectionTimeStr, err)
+	}
+	return nextDetectionTime, nil
+}

pkg/utils/annotations/annotations_test.go

@@ -18,6 +18,7 @@ package annotations
 
 import (
 	"testing"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -394,3 +395,59 @@ func TestExtractNumberOfEnvelopeObjFromResourceSnapshot(t *testing.T) {
 		})
 	}
 }
+
+func TestExtractNextResourceSnapshotCandidateDetectionTimeFromResourceSnapshot(t *testing.T) {
+	validTime := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
+	testCases := []struct {
+		name      string
+		snapshot  *fleetv1beta1.ClusterResourceSnapshot
+		want      time.Time
+		wantError bool
+	}{
+		{
+			name: "valid annotation",
+			snapshot: &fleetv1beta1.ClusterResourceSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: snapshotName,
+					Annotations: map[string]string{
+						fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation: validTime.Format(time.RFC3339),
+					},
+				},
+			},
+			want: validTime,
+		},
+		{
+			name: "no annotation means no next detection time",
+			snapshot: &fleetv1beta1.ClusterResourceSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: snapshotName,
+				},
+			},
+			want: time.Time{},
+		},
+		{
+			name: "invalid annotation format",
+			snapshot: &fleetv1beta1.ClusterResourceSnapshot{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: snapshotName,
+					Annotations: map[string]string{
+						fleetv1beta1.NextResourceSnapshotCandidateDetectionTimeAnnotation: "invalid-time",
+					},
+				},
+			},
+			wantError: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := ExtractNextResourceSnapshotCandidateDetectionTimeFromResourceSnapshot(tc.snapshot)
+			if gotErr := err != nil; gotErr != tc.wantError {
+				t.Fatalf("ExtractNextResourceSnapshotCandidateDetectionTimeFromResourceSnapshot() got err %v, want err %v", err, tc.wantError)
+			}
+			if !tc.wantError && got != tc.want {
+				t.Errorf("ExtractNextResourceSnapshotCandidateDetectionTimeFromResourceSnapshot() got %s, want %s", got, tc.want)
+			}
+		})
+	}
+}

test/e2e/README.md

@@ -14,6 +14,9 @@ test suites, follow the steps below:
     # Use a different path if the local set up is different.
     export KUBECONFIG=~/.kube/config
     export OUTPUT_TYPE=type=docker
+    # optional, used to test the placement features with custom configurations
+    # It defaults to 0m.
+    # export RESOURCE_SNAPSHOT_CREATION_INTERVAL=1m
     ./setup.sh ${number of member clusters}
     ```
 
@@ -25,10 +28,38 @@ test suites, follow the steps below:
     * Upload the images to the `Kind` clusters
     * Install the agent images
 
-3. Run the command below to start running the tests:
+3. Run the command below to run non custom configuration e2e tests with the following command:
 
     ```sh
-   ginkgo -v -p .
+   ginkgo --label-filter="!custom" -v -p .
+   ```
+
+   or run the custom configuration e2e tests with the following command
+   ```sh
+   ginkgo --label-filter="custom" -v -p .
+   ```
+
+   or create a launch.json in your vscode workspace.
+   ```yaml
+   {
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Launch E2E Tests",
+            "type": "go",
+            "request": "launch",
+            "mode": "test",
+            "program": "${workspaceFolder}/test/e2e",
+            "args": [],
+            "env": {
+                "KUBECONFIG": "~/.kube/config",
+                #"RESOURCE_SNAPSHOT_CREATION_INTERVAL": "1m",
+            },
+            "buildFlags": "-tags=e2e",
+            "showLog": true
+        }
+    ]
+   }
    ```
 
 ## Access the `Kind` clusters