Merge pull request #34 from cybozu-go/add-sequential-creation-update

Implement sequential creation and update of NodeImageSets

Author: zeroalphat

cmd/imageprefetch-controller/main.go

@@ -36,7 +36,8 @@ func init() {
 }
 
 type Config struct {
-	imagePullNodeLimit int
+	imagePullNodeLimit                 int
+	maxConcurrentNodeImageSetCreations int
 }
 
 func main() {
@@ -59,6 +60,8 @@ func main() {
 		"If set, HTTP/2 will be enabled for the metrics and webhook servers")
 	flag.IntVar(&config.imagePullNodeLimit, "image-pull-node-limit", 1,
 		"The maximum number of nodes that can pull images concurrently.")
+	flag.IntVar(&config.maxConcurrentNodeImageSetCreations, "max-concurrent-nodeimageset-creations", 1,
+		"The maximum number of NodeImageSets to create or update concurrently.")
 	opts := zap.Options{
 		Development: true,
 	}
@@ -135,9 +138,10 @@ func main() {
 	}
 
 	if err = (&controller.ImagePrefetchReconciler{
-		Client:             mgr.GetClient(),
-		Scheme:             mgr.GetScheme(),
-		ImagePullNodeLimit: config.imagePullNodeLimit,
+		Client:                             mgr.GetClient(),
+		Scheme:                             mgr.GetScheme(),
+		ImagePullNodeLimit:                 config.imagePullNodeLimit,
+		MaxConcurrentNodeImageSetCreations: config.maxConcurrentNodeImageSetCreations,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "ImagePrefetch")
 		os.Exit(1)

internal/controller/imageprefetch_controller.go

@@ -40,8 +40,9 @@ import (
 // ImagePrefetchReconciler reconciles a ImagePrefetch object
 type ImagePrefetchReconciler struct {
 	client.Client
-	Scheme             *runtime.Scheme
-	ImagePullNodeLimit int
+	Scheme                             *runtime.Scheme
+	ImagePullNodeLimit                 int
+	MaxConcurrentNodeImageSetCreations int
 }
 
 // +kubebuilder:rbac:groups=ofen.cybozu.io,resources=imageprefetches,verbs=get;list;watch;create;update;patch;delete
@@ -273,8 +274,22 @@ func scoreNode(node corev1.Node, zoneCount map[string]int) int {
 func (r *ImagePrefetchReconciler) createOrUpdateNodeImageSet(ctx context.Context, imgPrefetch *ofenv1.ImagePrefetch, selectedNodes []string) error {
 	logger := log.FromContext(ctx)
 
+	var nodeImageSetList ofenv1.NodeImageSetList
+	if err := r.List(ctx, &nodeImageSetList, client.MatchingLabels(map[string]string{
+		constants.OwnerImagePrefetchNamespace: imgPrefetch.Namespace,
+		constants.OwnerImagePrefetchName:      imgPrefetch.Name,
+	})); err != nil {
+		return fmt.Errorf("failed to list NodeImageSets: %w", err)
+	}
+
+	maxNodeImageSetsToProcess := getMaxNodeImageSetsToCreate(nodeImageSetList, r.MaxConcurrentNodeImageSetCreations, len(selectedNodes), imgPrefetch.Generation)
 	selectNodes := map[string]struct{}{}
 	for i, nodeName := range selectedNodes {
+		// Limit the number of NodeImageSets created or updated in one reconciliation loop
+		if i >= maxNodeImageSetsToProcess {
+			return nil
+		}
+
 		selectNodes[nodeName] = struct{}{}
 		nodeImageSetName, err := getNodeImageSetName(imgPrefetch, nodeName)
 		if err != nil {
@@ -308,8 +323,8 @@ func (r *ImagePrefetchReconciler) createOrUpdateNodeImageSet(ctx context.Context
 	}
 
 	// Delete unnecessary NodeImageSets
-	nodeImageSetList := &ofenv1.NodeImageSetList{}
-	if err := r.List(ctx, nodeImageSetList, client.MatchingLabels(map[string]string{
+	nodeImageSetList = ofenv1.NodeImageSetList{}
+	if err := r.List(ctx, &nodeImageSetList, client.MatchingLabels(map[string]string{
 		constants.OwnerImagePrefetchNamespace: imgPrefetch.Namespace,
 		constants.OwnerImagePrefetchName:      imgPrefetch.Name,
 	})); err != nil {
@@ -333,6 +348,23 @@ func (r *ImagePrefetchReconciler) createOrUpdateNodeImageSet(ctx context.Context
 	return nil
 }
 
+func getMaxNodeImageSetsToCreate(currentNodeImageSets ofenv1.NodeImageSetList, maxConcurrentNodeImageSetCreations int, desiredNodeImageSetsCount int, generation int64) int {
+	var readyNodeImageSetsCount = 0
+
+	for _, nodeImageSet := range currentNodeImageSets.Items {
+		if meta.IsStatusConditionTrue(nodeImageSet.Status.Conditions, ofenv1.ConditionImageAvailable) &&
+			nodeImageSet.Status.ImagePrefetchGeneration == generation &&
+			len(nodeImageSet.Spec.Images) == len(nodeImageSet.Status.ContainerImageStatuses) {
+			readyNodeImageSetsCount++
+		}
+	}
+
+	if readyNodeImageSetsCount < desiredNodeImageSetsCount {
+		return readyNodeImageSetsCount + maxConcurrentNodeImageSetCreations
+	}
+	return desiredNodeImageSetsCount
+}
+
 func (r *ImagePrefetchReconciler) applyNodeImageSet(ctx context.Context, nodeImageSet *ofenv1apply.NodeImageSetApplyConfiguration, name string) error {
 	obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(nodeImageSet)
 	if err != nil {