Merge pull request #9307 from sseago/parallel-backup

Parallel backup processing

changelogs/unreleased/9307-sseago

@@ -0,0 +1 @@
+Concurrent backup processing

config/crd/v1/bases/velero.io_backups.yaml

@@ -594,6 +594,8 @@ spec:
                 description: Phase is the current state of the Backup.
                 enum:
                 - New
+                - Queued
+                - ReadyToStart
                 - FailedValidation
                 - InProgress
                 - WaitingForPluginOperations
@@ -625,6 +627,11 @@ spec:
                       filters that happen as items are processed.
                     type: integer
                 type: object
+              queuePosition:
+                description: |-
+                  QueuePosition is the position of the backup in the queue.
+                  Only relevant when Phase is "Queued"
+                type: integer
               startTimestamp:
                 description: |-
                   StartTimestamp records the time a backup was started.

pkg/apis/velero/v1/backup_types.go

@@ -288,7 +288,7 @@ const (
 
 // BackupPhase is a string representation of the lifecycle phase
 // of a Velero backup.
-// +kubebuilder:validation:Enum=New;FailedValidation;InProgress;WaitingForPluginOperations;WaitingForPluginOperationsPartiallyFailed;Finalizing;FinalizingPartiallyFailed;Completed;PartiallyFailed;Failed;Deleting
+// +kubebuilder:validation:Enum=New;Queued;ReadyToStart;FailedValidation;InProgress;WaitingForPluginOperations;WaitingForPluginOperationsPartiallyFailed;Finalizing;FinalizingPartiallyFailed;Completed;PartiallyFailed;Failed;Deleting
 type BackupPhase string
 
 const (
@@ -296,6 +296,12 @@ const (
 	// yet processed by the BackupController.
 	BackupPhaseNew BackupPhase = "New"
 
+	// BackupPhaseQueued means the backup has been added to the queue and is waiting for the Queue to move it out of the queue.
+	BackupPhaseQueued BackupPhase = "Queued"
+
+	// BackupPhaseReadyToStart means the backup has been pulled from the queue and is ready to start.
+	BackupPhaseReadyToStart BackupPhase = "ReadyToStart"
+
 	// BackupPhaseFailedValidation means the backup has failed
 	// the controller's validations and therefore will not run.
 	BackupPhaseFailedValidation BackupPhase = "FailedValidation"
@@ -371,6 +377,11 @@ type BackupStatus struct {
 	// +optional
 	Phase BackupPhase `json:"phase,omitempty"`
 
+	// QueuePosition is the position of the backup in the queue.
+	// Only relevant when Phase is "Queued"
+	// +optional
+	QueuePosition int `json:"queuePosition,omitempty"`
+
 	// ValidationErrors is a slice of all validation errors (if
 	// applicable).
 	// +optional

pkg/backup/backup.go

@@ -117,7 +117,6 @@ type kubernetesBackupper struct {
 	podCommandExecutor        podexec.PodCommandExecutor
 	podVolumeBackupperFactory podvolume.BackupperFactory
 	podVolumeTimeout          time.Duration
-	podVolumeContext          context.Context
 	defaultVolumesToFsBackup  bool
 	clientPageSize            int
 	uploaderType              string
@@ -365,12 +364,12 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 	}
 
 	var podVolumeCancelFunc context.CancelFunc
-	kb.podVolumeContext, podVolumeCancelFunc = context.WithTimeout(context.Background(), podVolumeTimeout)
+	podVolumeContext, podVolumeCancelFunc := context.WithTimeout(context.Background(), podVolumeTimeout)
 	defer podVolumeCancelFunc()
 
 	var podVolumeBackupper podvolume.Backupper
 	if kb.podVolumeBackupperFactory != nil {
-		podVolumeBackupper, err = kb.podVolumeBackupperFactory.NewBackupper(kb.podVolumeContext, log, backupRequest.Backup, kb.uploaderType)
+		podVolumeBackupper, err = kb.podVolumeBackupperFactory.NewBackupper(podVolumeContext, log, backupRequest.Backup, kb.uploaderType)
 		if err != nil {
 			log.WithError(errors.WithStack(err)).Debugf("Error from NewBackupper")
 			return errors.WithStack(err)
@@ -416,6 +415,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 		kbClient:                 kb.kbClient,
 		discoveryHelper:          kb.discoveryHelper,
 		podVolumeBackupper:       podVolumeBackupper,
+		podVolumeContext:         podVolumeContext,
 		podVolumeSnapshotTracker: podvolume.NewTracker(),
 		volumeSnapshotterCache:   NewVolumeSnapshotterCache(volumeSnapshotterGetter),
 		itemHookHandler: &hook.DefaultItemHookHandler{
@@ -597,7 +597,7 @@ func (kb *kubernetesBackupper) BackupWithResolvers(
 			log.Infof("Backing Up Item Block including %s %s/%s (%v items in block)", items[i].groupResource.String(), items[i].namespace, items[i].name, len(itemBlock.Items))
 
 			wg.Add(1)
-			backupRequest.ItemBlockChannel <- ItemBlockInput{
+			backupRequest.WorkerPool.GetInputChannel() <- ItemBlockInput{
 				itemBlock:  itemBlock,
 				returnChan: itemBlockReturn,
 			}
@@ -848,7 +848,7 @@ func (kb *kubernetesBackupper) handleItemBlockPostHooks(itemBlock *BackupItemBlo
 	log := itemBlock.Log
 
 	// the post hooks will not execute until all PVBs of the item block pods are processed
-	if err := kb.waitUntilPVBsProcessed(kb.podVolumeContext, log, itemBlock, hookPods); err != nil {
+	if err := kb.waitUntilPVBsProcessed(itemBlock.itemBackupper.podVolumeContext, log, itemBlock, hookPods); err != nil {
 		log.WithError(err).Error("failed to wait PVBs processed for the ItemBlock")
 		return
 	}

pkg/backup/backup_test.go

@@ -79,7 +79,7 @@ func TestBackedUpItemsMatchesTarballContents(t *testing.T) {
 		Backup:           defaultBackup().Result(),
 		SkippedPVTracker: NewSkipPVTracker(),
 		BackedUpItems:    NewBackedUpItemsMap(),
-		ItemBlockChannel: h.itemBlockPool.GetInputChannel(),
+		WorkerPool:       &h.itemBlockPool,
 	}
 
 	backupFile := bytes.NewBuffer([]byte{})
@@ -141,7 +141,7 @@ func TestBackupProgressIsUpdated(t *testing.T) {
 		Backup:           defaultBackup().Result(),
 		SkippedPVTracker: NewSkipPVTracker(),
 		BackedUpItems:    NewBackedUpItemsMap(),
-		ItemBlockChannel: h.itemBlockPool.GetInputChannel(),
+		WorkerPool:       &h.itemBlockPool,
 	}
 	backupFile := bytes.NewBuffer([]byte{})
 
@@ -881,7 +881,7 @@ func TestBackupOldResourceFiltering(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -1062,7 +1062,7 @@ func TestCRDInclusion(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -1161,7 +1161,7 @@ func TestBackupResourceCohabitation(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -1190,7 +1190,7 @@ func TestBackupUsesNewCohabitatingResourcesForEachBackup(t *testing.T) {
 		Backup:           defaultBackup().Result(),
 		SkippedPVTracker: NewSkipPVTracker(),
 		BackedUpItems:    NewBackedUpItemsMap(),
-		ItemBlockChannel: h.itemBlockPool.GetInputChannel(),
+		WorkerPool:       &h.itemBlockPool,
 	}
 	backup1File := bytes.NewBuffer([]byte{})
 
@@ -1206,7 +1206,7 @@ func TestBackupUsesNewCohabitatingResourcesForEachBackup(t *testing.T) {
 		Backup:           defaultBackup().Result(),
 		SkippedPVTracker: NewSkipPVTracker(),
 		BackedUpItems:    NewBackedUpItemsMap(),
-		ItemBlockChannel: h.itemBlockPool.GetInputChannel(),
+		WorkerPool:       &h.itemBlockPool,
 	}
 	backup2File := bytes.NewBuffer([]byte{})
 
@@ -1260,7 +1260,7 @@ func TestBackupResourceOrdering(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -1381,7 +1381,7 @@ func TestBackupItemActionsForSkippedPV(t *testing.T) {
 				Backup:           defaultBackup().SnapshotVolumes(false).Result(),
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			resPolicies: &resourcepolicies.ResourcePolicies{
 				Version: "v1",
@@ -1428,8 +1428,8 @@ func TestBackupItemActionsForSkippedPV(t *testing.T) {
 					},
 					includedPVs: map[string]struct{}{},
 				},
-				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				BackedUpItems: NewBackedUpItemsMap(),
+				WorkerPool:    itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVCs(
@@ -1679,7 +1679,7 @@ func TestBackupActionsRunForCorrectItems(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -1764,7 +1764,7 @@ func TestBackupWithInvalidActions(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -1918,7 +1918,7 @@ func TestBackupActionModifications(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -2178,7 +2178,7 @@ func TestBackupActionAdditionalItems(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -2439,7 +2439,7 @@ func TestItemBlockActionsRunForCorrectItems(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -2524,7 +2524,7 @@ func TestBackupWithInvalidItemBlockActions(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -2780,7 +2780,7 @@ func TestItemBlockActionRelatedItems(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -2948,7 +2948,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -2984,7 +2984,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3021,7 +3021,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3058,7 +3058,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3095,7 +3095,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3130,7 +3130,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3148,7 +3148,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				Backup:           defaultBackup().Result(),
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3169,7 +3169,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3188,7 +3188,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3210,7 +3210,7 @@ func TestBackupWithSnapshots(t *testing.T) {
 				},
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.PVs(
@@ -3344,7 +3344,7 @@ func TestBackupWithAsyncOperations(t *testing.T) {
 				Backup:           defaultBackup().Result(),
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.Pods(
@@ -3376,7 +3376,7 @@ func TestBackupWithAsyncOperations(t *testing.T) {
 				Backup:           defaultBackup().Result(),
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.Pods(
@@ -3408,7 +3408,7 @@ func TestBackupWithAsyncOperations(t *testing.T) {
 				Backup:           defaultBackup().Result(),
 				SkippedPVTracker: NewSkipPVTracker(),
 				BackedUpItems:    NewBackedUpItemsMap(),
-				ItemBlockChannel: itemBlockPool.GetInputChannel(),
+				WorkerPool:       itemBlockPool,
 			},
 			apiResources: []*test.APIResource{
 				test.Pods(
@@ -3494,7 +3494,7 @@ func TestBackupWithInvalidHooks(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -3968,7 +3968,7 @@ func TestBackupWithHooks(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile         = bytes.NewBuffer([]byte{})
 				podCommandExecutor = new(test.MockPodCommandExecutor)
@@ -4193,7 +4193,7 @@ func TestBackupWithPodVolume(t *testing.T) {
 					SnapshotLocations: []*velerov1.VolumeSnapshotLocation{tc.vsl},
 					SkippedPVTracker:  NewSkipPVTracker(),
 					BackedUpItems:     NewBackedUpItemsMap(),
-					ItemBlockChannel:  itemBlockPool.GetInputChannel(),
+					WorkerPool:        itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -5312,7 +5312,7 @@ func TestBackupNewResourceFiltering(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)
@@ -5477,7 +5477,7 @@ func TestBackupNamespaces(t *testing.T) {
 					Backup:           tc.backup,
 					SkippedPVTracker: NewSkipPVTracker(),
 					BackedUpItems:    NewBackedUpItemsMap(),
-					ItemBlockChannel: itemBlockPool.GetInputChannel(),
+					WorkerPool:       itemBlockPool,
 				}
 				backupFile = bytes.NewBuffer([]byte{})
 			)

pkg/backup/item_backupper.go

@@ -69,6 +69,7 @@ type itemBackupper struct {
 	kbClient                 kbClient.Client
 	discoveryHelper          discovery.Helper
 	podVolumeBackupper       podvolume.Backupper
+	podVolumeContext         context.Context
 	podVolumeSnapshotTracker *podvolume.Tracker
 	kubernetesBackupper      *kubernetesBackupper
 	volumeSnapshotterCache   *VolumeSnapshotterCache

pkg/backup/request.go

@@ -69,7 +69,7 @@ type Request struct {
 	ResPolicies               *resourcepolicies.Policies
 	SkippedPVTracker          *skipPVTracker
 	VolumesInformation        volume.BackupVolumesInformation
-	ItemBlockChannel          chan ItemBlockInput
+	WorkerPool                *ItemBlockWorkerPool
 }
 
 // BackupVolumesInformation contains the information needs by generating
@@ -103,3 +103,7 @@ func (r *Request) FillVolumesInformation() {
 	r.VolumesInformation.BackupOperations = *r.GetItemOperationsList()
 	r.VolumesInformation.BackupName = r.Backup.Name
 }
+
+func (r *Request) StopWorkerPool() {
+	r.WorkerPool.Stop()
+}

pkg/builder/backup_builder.go

@@ -222,6 +222,12 @@ func (b *BackupBuilder) Phase(phase velerov1api.BackupPhase) *BackupBuilder {
 	return b
 }
 
+// Phase sets the Backup's queue position.
+func (b *BackupBuilder) QueuePosition(queuePos int) *BackupBuilder {
+	b.object.Status.QueuePosition = queuePos
+	return b
+}
+
 // StorageLocation sets the Backup's storage location.
 func (b *BackupBuilder) StorageLocation(location string) *BackupBuilder {
 	b.object.Spec.StorageLocation = location

pkg/cmd/cli/install/install.go

@@ -89,6 +89,7 @@ type Options struct {
 	RepoMaintenanceJobConfigMap     string
 	NodeAgentConfigMap              string
 	ItemBlockWorkerCount            int
+	ConcurrentBackups               int
 	NodeAgentDisableHostPath        bool
 	kubeletRootDir                  string
 	ServerPriorityClassName         string
@@ -196,6 +197,12 @@ func (o *Options) BindFlags(flags *pflag.FlagSet) {
 		o.ItemBlockWorkerCount,
 		"Number of worker threads to process ItemBlocks. Default is one. Optional.",
 	)
+	flags.IntVar(
+		&o.ConcurrentBackups,
+		"concurrent-backups",
+		o.ConcurrentBackups,
+		"Number of backups to process concurrently. Default is one. Optional.",
+	)
 	flags.StringVar(
 		&o.ServerPriorityClassName,
 		"server-priority-class-name",
@@ -313,6 +320,7 @@ func (o *Options) AsVeleroOptions() (*install.VeleroOptions, error) {
 		RepoMaintenanceJobConfigMap:     o.RepoMaintenanceJobConfigMap,
 		NodeAgentConfigMap:              o.NodeAgentConfigMap,
 		ItemBlockWorkerCount:            o.ItemBlockWorkerCount,
+		ConcurrentBackups:               o.ConcurrentBackups,
 		KubeletRootDir:                  o.kubeletRootDir,
 		NodeAgentDisableHostPath:        o.NodeAgentDisableHostPath,
 		ServerPriorityClassName:         o.ServerPriorityClassName,

pkg/cmd/server/config/config.go

@@ -47,11 +47,13 @@ const (
 	defaultDisableInformerCache        = false
 
 	DefaultItemBlockWorkerCount = 1
+	DefaultConcurrentBackups    = 1
 )
 
 var (
 	// DisableableControllers is a list of controllers that can be disabled
 	DisableableControllers = []string{
+		constant.ControllerBackupQueue,
 		constant.ControllerBackup,
 		constant.ControllerBackupOperations,
 		constant.ControllerBackupDeletion,
@@ -174,6 +176,7 @@ type Config struct {
 	BackupRepoConfig               string
 	RepoMaintenanceJobConfig       string
 	ItemBlockWorkerCount           int
+	ConcurrentBackups              int
 }
 
 func GetDefaultConfig() *Config {
@@ -206,6 +209,7 @@ func GetDefaultConfig() *Config {
 		ScheduleSkipImmediately:        false,
 		CredentialsDirectory:           credentials.DefaultStoreDirectory(),
 		ItemBlockWorkerCount:           DefaultItemBlockWorkerCount,
+		ConcurrentBackups:              DefaultConcurrentBackups,
 	}
 
 	return config
@@ -261,4 +265,10 @@ func (c *Config) BindFlags(flags *pflag.FlagSet) {
 		c.ItemBlockWorkerCount,
 		"Number of worker threads to process ItemBlocks. Default is one. Optional.",
 	)
+	flags.IntVar(
+		&c.ConcurrentBackups,
+		"concurrent-backups",
+		c.ConcurrentBackups,
+		"Number of backups to process concurrently. Default is one. Optional.",
+	)
 }

pkg/cmd/server/server.go

@@ -581,6 +581,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		constant.ControllerSchedule:            {},
 		constant.ControllerServerStatusRequest: {},
 		constant.ControllerRestoreFinalizer:    {},
+		constant.ControllerBackupQueue:         {},
 	}
 
 	if s.config.RestoreOnly {
@@ -668,6 +669,7 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 			s.config.MaxConcurrentK8SConnections,
 			s.config.DefaultSnapshotMoveData,
 			s.config.ItemBlockWorkerCount,
+			s.config.ConcurrentBackups,
 			s.crClient,
 		).SetupWithManager(s.mgr); err != nil {
 			s.logger.Fatal(err, "unable to create controller", "controller", constant.ControllerBackup)
@@ -909,6 +911,18 @@ func (s *server) runControllers(defaultVolumeSnapshotLocations map[string]string
 		}
 	}
 
+	if _, ok := enabledRuntimeControllers[constant.ControllerBackupQueue]; ok {
+		if err := controller.NewBackupQueueReconciler(
+			s.mgr.GetClient(),
+			s.mgr.GetScheme(),
+			s.logger,
+			s.config.ConcurrentBackups,
+			backupTracker,
+		).SetupWithManager(s.mgr); err != nil {
+			s.logger.Fatal(err, "unable to create controller", "controller", constant.ControllerBackupQueue)
+		}
+	}
+
 	s.logger.Info("Server starting...")
 
 	if err := s.mgr.Start(s.ctx); err != nil {

pkg/cmd/util/output/backup_describer.go

@@ -75,6 +75,7 @@ func DescribeBackup(
 		case velerov1api.BackupPhaseFinalizing, velerov1api.BackupPhaseFinalizingPartiallyFailed:
 		case velerov1api.BackupPhaseInProgress:
 		case velerov1api.BackupPhaseNew:
+		case velerov1api.BackupPhaseQueued, velerov1api.BackupPhaseReadyToStart:
 		}
 
 		logsNote := ""
@@ -83,6 +84,9 @@ func DescribeBackup(
 		}
 
 		d.Printf("Phase:\t%s%s\n", phaseString, logsNote)
+		if phase == velerov1api.BackupPhaseQueued {
+			d.Printf("Queue position:\t%v\n", backup.Status.QueuePosition)
+		}
 
 		if backup.Spec.ResourcePolicy != nil {
 			d.Println()

pkg/cmd/util/output/backup_printer.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"regexp"
 	"sort"
+	"strconv"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -40,6 +41,7 @@ var (
 		{Name: "Created"},
 		{Name: "Expires"},
 		{Name: "Storage Location"},
+		{Name: "Queue Position"},
 		{Name: "Selector"},
 	}
 )
@@ -108,6 +110,7 @@ func printBackup(backup *velerov1api.Backup) []metav1.TableRow {
 		backup.Status.StartTimestamp,
 		humanReadableTimeFromNow(expiration),
 		backup.Spec.StorageLocation,
+		queuePosition(backup.Status.QueuePosition),
 		metav1.FormatLabelSelector(backup.Spec.LabelSelector),
 	)
 
@@ -127,3 +130,11 @@ func humanReadableTimeFromNow(when time.Time) string {
 		return fmt.Sprintf("%s ago", duration.ShortHumanDuration(now.Sub(when)))
 	}
 }
+
+func queuePosition(pos int) string {
+	if pos == 0 {
+		return ""
+	} else {
+		return strconv.Itoa(pos)
+	}
+}

pkg/constant/constant.go

@@ -1,6 +1,7 @@
 package constant
 
 const (
+	ControllerBackupQueue           = "backup-queue"
 	ControllerBackup                = "backup"
 	ControllerBackupOperations      = "backup-operations"
 	ControllerBackupDeletion        = "backup-deletion"

pkg/controller/backup_controller.go

@@ -36,7 +36,11 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/utils/clock"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	"github.com/vmware-tanzu/velero/internal/credentials"
 	"github.com/vmware-tanzu/velero/internal/resourcepolicies"
@@ -105,7 +109,7 @@ type backupReconciler struct {
 	defaultSnapshotMoveData     bool
 	globalCRClient              kbclient.Client
 	itemBlockWorkerCount        int
-	workerPool                  *pkgbackup.ItemBlockWorkerPool
+	concurrentBackups           int
 }
 
 func NewBackupReconciler(
@@ -132,6 +136,7 @@ func NewBackupReconciler(
 	maxConcurrentK8SConnections int,
 	defaultSnapshotMoveData bool,
 	itemBlockWorkerCount int,
+	concurrentBackups int,
 	globalCRClient kbclient.Client,
 ) *backupReconciler {
 	b := &backupReconciler{
@@ -159,8 +164,8 @@ func NewBackupReconciler(
 		maxConcurrentK8SConnections: maxConcurrentK8SConnections,
 		defaultSnapshotMoveData:     defaultSnapshotMoveData,
 		itemBlockWorkerCount:        itemBlockWorkerCount,
+		concurrentBackups:           max(concurrentBackups, 1),
 		globalCRClient:              globalCRClient,
-		workerPool:                  pkgbackup.StartItemBlockWorkerPool(ctx, itemBlockWorkerCount, logger),
 	}
 	b.updateTotalBackupMetric()
 	return b
@@ -168,7 +173,24 @@ func NewBackupReconciler(
 
 func (b *backupReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
-		For(&velerov1api.Backup{}).
+		For(&velerov1api.Backup{}, builder.WithPredicates(predicate.Funcs{
+			UpdateFunc: func(ue event.UpdateEvent) bool {
+				backup := ue.ObjectNew.(*velerov1api.Backup)
+				return backup.Status.Phase == velerov1api.BackupPhaseReadyToStart
+			},
+			CreateFunc: func(ce event.CreateEvent) bool {
+				return false
+			},
+			DeleteFunc: func(de event.DeleteEvent) bool {
+				return false
+			},
+			GenericFunc: func(ge event.GenericEvent) bool {
+				return false
+			},
+		})).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: b.concurrentBackups,
+		}).
 		Named(constant.ControllerBackup).
 		Complete(b)
 }
@@ -254,8 +276,8 @@ func (b *backupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 	// InProgress, we still need this check so we can return nil to indicate we've finished processing
 	// this key (even though it was a no-op).
 	switch original.Status.Phase {
-	case "", velerov1api.BackupPhaseNew:
-		// only process new backups
+	case velerov1api.BackupPhaseReadyToStart:
+		// only process ReadytToStart backups
 	default:
 		b.logger.WithFields(logrus.Fields{
 			"backup": kubeutil.NamespaceAndName(original),
@@ -265,7 +287,9 @@ func (b *backupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 	}
 
 	log.Debug("Preparing backup request")
-	request := b.prepareBackupRequest(original, log)
+	request := b.prepareBackupRequest(ctx, original, log)
+	// delete worker pool after reconcile
+	defer request.StopWorkerPool()
 	if len(request.Status.ValidationErrors) > 0 {
 		request.Status.Phase = velerov1api.BackupPhaseFailedValidation
 	} else {
@@ -299,6 +323,8 @@ func (b *backupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 		switch request.Status.Phase {
 		case velerov1api.BackupPhaseCompleted, velerov1api.BackupPhasePartiallyFailed, velerov1api.BackupPhaseFailed, velerov1api.BackupPhaseFailedValidation:
 			b.backupTracker.Delete(request.Namespace, request.Name)
+		case velerov1api.BackupPhaseWaitingForPluginOperations, velerov1api.BackupPhaseWaitingForPluginOperationsPartiallyFailed, velerov1api.BackupPhaseFinalizing, velerov1api.BackupPhaseFinalizingPartiallyFailed:
+			b.backupTracker.AddPostProcessing(request.Namespace, request.Name)
 		}
 	}()
 
@@ -347,12 +373,12 @@ func (b *backupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 	return ctrl.Result{}, nil
 }
 
-func (b *backupReconciler) prepareBackupRequest(backup *velerov1api.Backup, logger logrus.FieldLogger) *pkgbackup.Request {
+func (b *backupReconciler) prepareBackupRequest(ctx context.Context, backup *velerov1api.Backup, logger logrus.FieldLogger) *pkgbackup.Request {
 	request := &pkgbackup.Request{
 		Backup:           backup.DeepCopy(), // don't modify items in the cache
 		SkippedPVTracker: pkgbackup.NewSkipPVTracker(),
 		BackedUpItems:    pkgbackup.NewBackedUpItemsMap(),
-		ItemBlockChannel: b.workerPool.GetInputChannel(),
+		WorkerPool:       pkgbackup.StartItemBlockWorkerPool(ctx, b.itemBlockWorkerCount, logger),
 	}
 	request.VolumesInformation.Init()
 

pkg/controller/backup_controller_test.go

@@ -95,7 +95,11 @@ func (b *fakeBackupper) FinalizeBackup(
 }
 
 func defaultBackup() *builder.BackupBuilder {
-	return builder.ForBackup(velerov1api.DefaultNamespace, "backup-1")
+	return builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").Phase(velerov1api.BackupPhaseReadyToStart)
+}
+
+func namedBackup(name string) *builder.BackupBuilder {
+	return builder.ForBackup(velerov1api.DefaultNamespace, name).Phase(velerov1api.BackupPhaseReadyToStart)
 }
 
 func TestProcessBackupNonProcessedItems(t *testing.T) {
@@ -104,6 +108,16 @@ func TestProcessBackupNonProcessedItems(t *testing.T) {
 		key    string
 		backup *velerov1api.Backup
 	}{
+		{
+			name:   "New backup is not processed",
+			key:    "velero/backup-1",
+			backup: defaultBackup().Phase(velerov1api.BackupPhaseNew).Result(),
+		},
+		{
+			name:   "Queued backup is not processed",
+			key:    "velero/backup-1",
+			backup: defaultBackup().Phase(velerov1api.BackupPhaseQueued).Result(),
+		},
 		{
 			name:   "FailedValidation backup is not processed",
 			key:    "velero/backup-1",
@@ -135,9 +149,7 @@ func TestProcessBackupNonProcessedItems(t *testing.T) {
 				kbClient:   velerotest.NewFakeControllerRuntimeClient(t),
 				formatFlag: formatFlag,
 				logger:     logger,
-				workerPool: pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 			if test.backup != nil {
 				require.NoError(t, c.kbClient.Create(t.Context(), test.backup))
 			}
@@ -234,9 +246,7 @@ func TestProcessBackupValidationFailures(t *testing.T) {
 				clock:                 &clock.RealClock{},
 				formatFlag:            formatFlag,
 				metrics:               metrics.NewServerMetrics(),
-				workerPool:            pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 
 			require.NotNil(t, test.backup)
 			require.NoError(t, c.kbClient.Create(t.Context(), test.backup))
@@ -299,11 +309,10 @@ func TestBackupLocationLabel(t *testing.T) {
 				defaultBackupLocation: test.backupLocation.Name,
 				clock:                 &clock.RealClock{},
 				formatFlag:            formatFlag,
-				workerPool:            pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 
-			res := c.prepareBackupRequest(test.backup, logger)
+			res := c.prepareBackupRequest(ctx, test.backup, logger)
+			defer res.WorkerPool.Stop()
 			assert.NotNil(t, res)
 			assert.Equal(t, test.expectedBackupLocation, res.Labels[velerov1api.StorageLocationLabel])
 		})
@@ -331,7 +340,7 @@ func Test_prepareBackupRequest_BackupStorageLocation(t *testing.T) {
 	}{
 		{
 			name:                             "BackupLocation is specified in backup CR'spec and it can be found in ApiServer",
-			backup:                           builder.ForBackup("velero", "backup-1").Result(),
+			backup:                           defaultBackup().Result(),
 			backupLocationNameInBackup:       "test-backup-location",
 			backupLocationInAPIServer:        builder.ForBackupStorageLocation("velero", "test-backup-location").Result(),
 			defaultBackupLocationInAPIServer: builder.ForBackupStorageLocation("velero", "default-location").Result(),
@@ -340,7 +349,7 @@ func Test_prepareBackupRequest_BackupStorageLocation(t *testing.T) {
 		},
 		{
 			name:                             "BackupLocation is specified in backup CR'spec and it can't be found in ApiServer",
-			backup:                           builder.ForBackup("velero", "backup-1").Result(),
+			backup:                           defaultBackup().Result(),
 			backupLocationNameInBackup:       "test-backup-location",
 			backupLocationInAPIServer:        nil,
 			defaultBackupLocationInAPIServer: nil,
@@ -349,7 +358,7 @@ func Test_prepareBackupRequest_BackupStorageLocation(t *testing.T) {
 		},
 		{
 			name:                             "Using default BackupLocation and it can be found in ApiServer",
-			backup:                           builder.ForBackup("velero", "backup-1").Result(),
+			backup:                           defaultBackup().Result(),
 			backupLocationNameInBackup:       "",
 			backupLocationInAPIServer:        builder.ForBackupStorageLocation("velero", "test-backup-location").Result(),
 			defaultBackupLocationInAPIServer: builder.ForBackupStorageLocation("velero", "default-location").Result(),
@@ -358,7 +367,7 @@ func Test_prepareBackupRequest_BackupStorageLocation(t *testing.T) {
 		},
 		{
 			name:                             "Using default BackupLocation and it can't be found in ApiServer",
-			backup:                           builder.ForBackup("velero", "backup-1").Result(),
+			backup:                           defaultBackup().Result(),
 			backupLocationNameInBackup:       "",
 			backupLocationInAPIServer:        nil,
 			defaultBackupLocationInAPIServer: nil,
@@ -396,14 +405,13 @@ func Test_prepareBackupRequest_BackupStorageLocation(t *testing.T) {
 				defaultBackupTTL:      defaultBackupTTL.Duration,
 				clock:                 testclocks.NewFakeClock(now),
 				formatFlag:            formatFlag,
-				workerPool:            pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 
 			test.backup.Spec.StorageLocation = test.backupLocationNameInBackup
 
 			// Run
-			res := c.prepareBackupRequest(test.backup, logger)
+			res := c.prepareBackupRequest(ctx, test.backup, logger)
+			defer res.WorkerPool.Stop()
 
 			// Assert
 			if test.expectedSuccess {
@@ -472,11 +480,10 @@ func TestDefaultBackupTTL(t *testing.T) {
 				defaultBackupTTL: defaultBackupTTL.Duration,
 				clock:            testclocks.NewFakeClock(now),
 				formatFlag:       formatFlag,
-				workerPool:       pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 
-			res := c.prepareBackupRequest(test.backup, logger)
+			res := c.prepareBackupRequest(ctx, test.backup, logger)
+			defer res.WorkerPool.Stop()
 			assert.NotNil(t, res)
 			assert.Equal(t, test.expectedTTL, res.Spec.TTL)
 			assert.Equal(t, test.expectedExpiration, *res.Status.Expiration)
@@ -497,7 +504,7 @@ func TestPrepareBackupRequest_SetsVGSLabelKey(t *testing.T) {
 	}{
 		{
 			name: "backup with spec label key set",
-			backup: builder.ForBackup("velero", "backup-1").
+			backup: defaultBackup().
 				VolumeGroupSnapshotLabelKey("spec-key").
 				Result(),
 			serverFlagKey:    "server-key",
@@ -505,13 +512,13 @@ func TestPrepareBackupRequest_SetsVGSLabelKey(t *testing.T) {
 		},
 		{
 			name:             "backup with no spec key, uses server flag",
-			backup:           builder.ForBackup("velero", "backup-2").Result(),
+			backup:           namedBackup("backup-2").Result(),
 			serverFlagKey:    "server-key",
 			expectedLabelKey: "server-key",
 		},
 		{
 			name:             "backup with no spec or server flag, uses default",
-			backup:           builder.ForBackup("velero", "backup-3").Result(),
+			backup:           namedBackup("backup-3").Result(),
 			serverFlagKey:    velerov1api.DefaultVGSLabelKey,
 			expectedLabelKey: velerov1api.DefaultVGSLabelKey,
 		},
@@ -533,11 +540,10 @@ func TestPrepareBackupRequest_SetsVGSLabelKey(t *testing.T) {
 				defaultVGSLabelKey: test.serverFlagKey,
 				discoveryHelper:    discoveryHelper,
 				clock:              testclocks.NewFakeClock(now),
-				workerPool:         pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 
-			res := c.prepareBackupRequest(test.backup, logger)
+			res := c.prepareBackupRequest(ctx, test.backup, logger)
+			defer res.WorkerPool.Stop()
 			assert.NotNil(t, res)
 
 			assert.Equal(t, test.expectedLabelKey, res.Spec.VolumeGroupSnapshotLabelKey)
@@ -635,11 +641,10 @@ func TestDefaultVolumesToResticDeprecation(t *testing.T) {
 				clock:                    &clock.RealClock{},
 				formatFlag:               formatFlag,
 				defaultVolumesToFsBackup: test.globalVal,
-				workerPool:               pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 
-			res := c.prepareBackupRequest(test.backup, logger)
+			res := c.prepareBackupRequest(ctx, test.backup, logger)
+			defer res.WorkerPool.Stop()
 			assert.NotNil(t, res)
 			assert.NotNil(t, res.Spec.DefaultVolumesToFsBackup)
 			if test.expectRemap {
@@ -1390,7 +1395,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 		},
 		{
 			name: "backup with namespace-scoped and cluster-scoped resource filters",
-			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").
+			backup: defaultBackup().
 				ExcludedClusterScopedResources("clusterroles").
 				IncludedClusterScopedResources("storageclasses").
 				ExcludedNamespaceScopedResources("secrets").
@@ -1439,7 +1444,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 		},
 		{
 			name: "backup's include filter overlap with default exclude resources",
-			backup: builder.ForBackup(velerov1api.DefaultNamespace, "backup-1").
+			backup: defaultBackup().
 				ExcludedClusterScopedResources("clusterroles").
 				IncludedClusterScopedResources("storageclasses", "volumesnapshotcontents.snapshot.storage.k8s.io").
 				ExcludedNamespaceScopedResources("secrets").
@@ -1555,9 +1560,7 @@ func TestProcessBackupCompletions(t *testing.T) {
 				backupper:                backupper,
 				formatFlag:               formatFlag,
 				globalCRClient:           fakeGlobalClient,
-				workerPool:               pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 
 			pluginManager.On("GetBackupItemActionsV2").Return(nil, nil)
 			pluginManager.On("GetItemBlockActions").Return(nil, nil)
@@ -1763,9 +1766,7 @@ func TestValidateAndGetSnapshotLocations(t *testing.T) {
 				logger:                   logger,
 				defaultSnapshotLocations: test.defaultLocations,
 				kbClient:                 velerotest.NewFakeControllerRuntimeClient(t),
-				workerPool:               pkgbackup.StartItemBlockWorkerPool(t.Context(), 1, logger),
 			}
-			defer c.workerPool.Stop()
 
 			// set up a Backup object to represent what we expect to be passed to backupper.Backup()
 			backup := test.backup.DeepCopy()

pkg/controller/backup_queue_controller.go

@@ -0,0 +1,363 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"context"
+	"slices"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	corev1api "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/constant"
+	"github.com/vmware-tanzu/velero/pkg/util/collections"
+	"github.com/vmware-tanzu/velero/pkg/util/kube"
+)
+
+// backupQueueReconciler reconciles a Backup object
+type backupQueueReconciler struct {
+	client.Client
+	Scheme            *runtime.Scheme
+	logger            logrus.FieldLogger
+	concurrentBackups int
+	backupTracker     BackupTracker
+	frequency         time.Duration
+}
+
+const (
+	defaultQueuedBackupRecheckFrequency = time.Minute
+)
+
+// NewBackupQueueReconciler returns a new backupQueueReconciler
+func NewBackupQueueReconciler(
+	client client.Client,
+	scheme *runtime.Scheme,
+	logger logrus.FieldLogger,
+	concurrentBackups int,
+	backupTracker BackupTracker,
+) *backupQueueReconciler {
+	return &backupQueueReconciler{
+		Client:            client,
+		Scheme:            scheme,
+		logger:            logger,
+		concurrentBackups: max(concurrentBackups, 1),
+		backupTracker:     backupTracker,
+		frequency:         defaultQueuedBackupRecheckFrequency,
+	}
+}
+
+func queuePositionOrderFunc(objList client.ObjectList) client.ObjectList {
+	backupList := objList.(*velerov1api.BackupList)
+	slices.SortFunc(backupList.Items, func(backup1, backup2 velerov1api.Backup) int {
+		if backup1.Status.QueuePosition < backup2.Status.QueuePosition {
+			return -1
+		} else if backup1.Status.QueuePosition == backup2.Status.QueuePosition {
+			return 0
+		} else {
+			return 1
+		}
+	})
+	return backupList
+}
+
+// SetupWithManager adds the reconciler to the manager
+func (r *backupQueueReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	// For periodic requeue, only consider Queued backups, order by QueuePosition
+	gp := kube.NewGenericEventPredicate(func(object client.Object) bool {
+		backup := object.(*velerov1api.Backup)
+		return backup.Status.Phase == velerov1api.BackupPhaseQueued
+	})
+
+	s := kube.NewPeriodicalEnqueueSource(r.logger.WithField("controller", constant.ControllerBackupQueue), mgr.GetClient(), &velerov1api.BackupList{}, r.frequency, kube.PeriodicalEnqueueSourceOption{
+		Predicates: []predicate.Predicate{gp},
+		OrderFunc:  queuePositionOrderFunc,
+	})
+
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&velerov1api.Backup{}, builder.WithPredicates(predicate.Funcs{
+			UpdateFunc: func(ue event.UpdateEvent) bool {
+				backup := ue.ObjectNew.(*velerov1api.Backup)
+				return backup.Status.Phase == "" || backup.Status.Phase == velerov1api.BackupPhaseNew
+			},
+			CreateFunc: func(ce event.CreateEvent) bool {
+				backup := ce.Object.(*velerov1api.Backup)
+				return backup.Status.Phase == "" || backup.Status.Phase == velerov1api.BackupPhaseNew
+			},
+			DeleteFunc: func(de event.DeleteEvent) bool {
+				return false
+			},
+			GenericFunc: func(ge event.GenericEvent) bool {
+				return false
+			},
+		})).
+		Watches(
+			&velerov1api.Backup{},
+			handler.EnqueueRequestsFromMapFunc(r.findQueuedBackupsToRequeue),
+			builder.WithPredicates(predicate.Funcs{
+				UpdateFunc: func(ue event.UpdateEvent) bool {
+					oldBackup := ue.ObjectOld.(*velerov1api.Backup)
+					newBackup := ue.ObjectNew.(*velerov1api.Backup)
+					return oldBackup.Status.Phase == velerov1api.BackupPhaseInProgress &&
+						newBackup.Status.Phase != velerov1api.BackupPhaseInProgress ||
+						oldBackup.Status.Phase != velerov1api.BackupPhaseQueued &&
+							newBackup.Status.Phase == velerov1api.BackupPhaseQueued &&
+							r.backupTracker.RunningCount() < r.concurrentBackups
+				},
+				CreateFunc: func(event.CreateEvent) bool {
+					return false
+				},
+				DeleteFunc: func(de event.DeleteEvent) bool {
+					return false
+				},
+				GenericFunc: func(ge event.GenericEvent) bool {
+					return false
+				},
+			})).
+		WatchesRawSource(s).
+		Named(constant.ControllerBackupQueue).
+		Complete(r)
+}
+
+func (r *backupQueueReconciler) detectNamespaceConflict(ctx context.Context, backup *velerov1api.Backup, earlierBackups []velerov1api.Backup) (bool, string, []string, error) {
+	nsList := &corev1api.NamespaceList{}
+	if err := r.Client.List(ctx, nsList); err != nil {
+		return false, "", nil, err
+	}
+	var clusterNamespaces []string
+	for _, ns := range nsList.Items {
+		clusterNamespaces = append(clusterNamespaces, ns.Name)
+	}
+	foundConflict, conflictBackup := detectNSConflictsInternal(backup, earlierBackups, clusterNamespaces)
+	return foundConflict, conflictBackup, clusterNamespaces, nil
+}
+
+func detectNSConflictsInternal(backup *velerov1api.Backup, earlierBackups []velerov1api.Backup, clusterNamespaces []string) (bool, string) {
+	backupNamespaces := sets.NewString(namespacesForBackup(backup, clusterNamespaces)...)
+	for _, earlierBackup := range earlierBackups {
+		// This will never be true for the primary backup, but for the secondary
+		// runnability check for queued backups ahead of the current backup, we
+		// only care about backups ahead of it.
+		// Backup isn't earlier than this one, skip
+		if earlierBackup.Status.Phase == velerov1api.BackupPhaseQueued &&
+			earlierBackup.Status.QueuePosition >= backup.Status.QueuePosition {
+			continue
+		}
+		if backupNamespaces.HasAny(namespacesForBackup(&earlierBackup, clusterNamespaces)...) {
+			return true, earlierBackup.Name
+		}
+	}
+	return false, ""
+}
+
+// Returns true if there are backups ahead of the current backup that are runnable
+// This could happen if velero just reconciled the one earlier in the queue and rejected it
+// due to too many running backups, but a backup completed in between that reconcile and this one
+// so exit, as the recent completion has triggered another reconcile of all queued backups
+func (r *backupQueueReconciler) checkForEarlierRunnableBackups(backup *velerov1api.Backup, earlierBackups []velerov1api.Backup, clusterNamespaces []string) (bool, string) {
+	for _, earlierBackup := range earlierBackups {
+		// if this backup is queued and ahead of current backup, check for conflicts
+		if earlierBackup.Status.Phase != velerov1api.BackupPhaseQueued ||
+			earlierBackup.Status.QueuePosition >= backup.Status.QueuePosition {
+			continue
+		}
+		conflict, _ := detectNSConflictsInternal(&earlierBackup, earlierBackups, clusterNamespaces)
+		// !conflict means we've found an earlier backup that is currently runnable
+		// so current reconcile should exit to run this one
+		if !conflict {
+			return true, earlierBackup.Name
+		}
+	}
+	return false, ""
+}
+
+func namespacesForBackup(backup *velerov1api.Backup, clusterNamespaces []string) []string {
+	// Ignore error here. If a backup has invalid namespace wildcards, the backup controller
+	// will validate and fail it. Consider the ns list empty for conflict detection purposes.
+	nsList, err := collections.NewNamespaceIncludesExcludes().Includes(backup.Spec.IncludedNamespaces...).Excludes(backup.Spec.ExcludedNamespaces...).ActiveNamespaces(clusterNamespaces).ResolveNamespaceList()
+	if err != nil {
+		return []string{}
+	}
+	return nsList
+}
+func (r *backupQueueReconciler) getMaxQueuePosition(lister *queuedBackupsLister) int {
+	queuedBackups := lister.orderedQueued()
+	maxPos := 0
+	if len(queuedBackups) > 0 {
+		maxPos = queuedBackups[len(queuedBackups)-1].Status.QueuePosition
+	}
+	return maxPos
+}
+
+func (r *backupQueueReconciler) findQueuedBackupsToRequeue(ctx context.Context, obj client.Object) []reconcile.Request {
+	backup := obj.(*velerov1api.Backup)
+	requests := []reconcile.Request{}
+	allBackups := &velerov1api.BackupList{}
+	if err := r.Client.List(ctx, allBackups, &client.ListOptions{Namespace: backup.Namespace}); err != nil {
+		r.logger.WithError(err).Error("error listing backups")
+		return requests
+	}
+	backups := r.newQueuedBackupsLister(allBackups).orderedQueued()
+	for _, item := range backups {
+		requests = append(requests, reconcile.Request{
+			NamespacedName: types.NamespacedName{
+				Namespace: item.GetNamespace(),
+				Name:      item.GetName(),
+			},
+		})
+	}
+	return requests
+}
+
+// Reconcile reconciles a Backup object
+func (r *backupQueueReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	log := r.logger.WithField("backup", req.NamespacedName.String())
+
+	log.Debug("Getting backup")
+	backup := &velerov1api.Backup{}
+	if err := r.Get(ctx, req.NamespacedName, backup); err != nil {
+		log.WithError(err).Error("unable to get backup")
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+	switch backup.Status.Phase {
+	case "", velerov1api.BackupPhaseNew:
+		// queue new backup
+		allBackups := &velerov1api.BackupList{}
+		if err := r.Client.List(ctx, allBackups, &client.ListOptions{Namespace: backup.Namespace}); err != nil {
+			r.logger.WithError(err).Error("error listing backups")
+			return ctrl.Result{}, nil
+		}
+		lister := r.newQueuedBackupsLister(allBackups)
+		maxQueuePosition := r.getMaxQueuePosition(lister)
+		original := backup.DeepCopy()
+		backup.Status.Phase = velerov1api.BackupPhaseQueued
+		backup.Status.QueuePosition = maxQueuePosition + 1
+		log.Infof("Queueing backup %v, queue position %v", backup.Name, backup.Status.QueuePosition)
+		if err := kube.PatchResource(original, backup, r.Client); err != nil {
+			return ctrl.Result{}, errors.Wrapf(err, "error updating Backup status to %s", backup.Status.Phase)
+		}
+	case velerov1api.BackupPhaseQueued:
+		// handle queued backup
+		// Find backups ahead of this one (InProgress, ReadyToStart, or Queued with higher position)
+		allBackups := &velerov1api.BackupList{}
+		if err := r.Client.List(ctx, allBackups, &client.ListOptions{Namespace: backup.Namespace}); err != nil {
+			r.logger.WithError(err).Error("error listing backups")
+			return ctrl.Result{}, nil
+		}
+		lister := r.newQueuedBackupsLister(allBackups)
+		if r.backupTracker.RunningCount() >= r.concurrentBackups {
+			log.Debugf("%v concurrent backups are already running, leaving %v queued", r.concurrentBackups, backup.Name)
+			return ctrl.Result{}, nil
+		}
+		earlierBackups := lister.earlierThan(backup.Status.QueuePosition)
+		foundConflict, conflictBackup, clusterNamespaces, err := r.detectNamespaceConflict(ctx, backup, earlierBackups)
+		if err != nil {
+			log.WithError(err).Error("error listing namespaces")
+			return ctrl.Result{}, nil
+		}
+		if foundConflict {
+			log.Infof("Backup %v has namespace conflict with %v, leaving queued", backup.Name, conflictBackup)
+			return ctrl.Result{}, nil
+		}
+		foundEarlierRunnable, earlierRunnable := r.checkForEarlierRunnableBackups(backup, earlierBackups, clusterNamespaces)
+		if foundEarlierRunnable {
+			log.Infof("Earlier queued backup %v is runnable, leaving %v queued", earlierRunnable, backup.Name)
+			return ctrl.Result{}, nil
+		}
+		log.Infof("Dequeueing backup %v, moving to ReadyToStart", backup.Name)
+		original := backup.DeepCopy()
+		backup.Status.Phase = velerov1api.BackupPhaseReadyToStart
+		backup.Status.QueuePosition = 0
+		if err := kube.PatchResource(original, backup, r.Client); err != nil {
+			return ctrl.Result{}, errors.Wrapf(err, "error updating Backup status to %s", backup.Status.Phase)
+		}
+		r.backupTracker.AddReadyToStart(backup.Namespace, backup.Name)
+		log.Debug("Updating queuePosition for remaining queued backups")
+		queuedBackups := lister.orderedQueued()
+		newQueuePos := 1
+		for _, queuedBackup := range queuedBackups {
+			if queuedBackup.Name != backup.Name {
+				original := queuedBackup.DeepCopy()
+				queuedBackup.Status.QueuePosition = newQueuePos
+				if err := kube.PatchResource(original, &queuedBackup, r.Client); err != nil {
+					log.WithError(errors.Wrapf(err, "error updating Backup %s queuePosition to %v", queuedBackup.Name, newQueuePos))
+					return ctrl.Result{}, nil
+				}
+				newQueuePos++
+			}
+		}
+		return ctrl.Result{}, nil
+	default:
+		log.Debug("Backup is not New or Queued, skipping")
+		return ctrl.Result{}, nil
+	}
+
+	return ctrl.Result{}, nil
+}
+
+// queuedBackupsLister manages a list of all backups Queued, ReadyToStart, or InProgress
+// with methods to return specific subsets as needed
+type queuedBackupsLister struct {
+	backups *velerov1api.BackupList
+}
+
+func (r *backupQueueReconciler) newQueuedBackupsLister(backupList *velerov1api.BackupList) *queuedBackupsLister {
+	backups := []velerov1api.Backup{}
+	for _, backup := range backupList.Items {
+		if backup.Status.Phase == velerov1api.BackupPhaseQueued ||
+			backup.Status.Phase == velerov1api.BackupPhaseInProgress ||
+			backup.Status.Phase == velerov1api.BackupPhaseReadyToStart {
+			backups = append(backups, backup)
+		}
+	}
+	backupList.Items = backups
+	return &queuedBackupsLister{backupList}
+}
+
+func (l *queuedBackupsLister) earlierThan(queuePos int) []velerov1api.Backup {
+	backups := []velerov1api.Backup{}
+	for _, backup := range l.backups.Items {
+		// InProgress and ReadyToStart backups have QueuePosition==0
+		if backup.Status.QueuePosition < queuePos {
+			backups = append(backups, backup)
+		}
+	}
+	return backups
+}
+
+func (l *queuedBackupsLister) orderedQueued() []velerov1api.Backup {
+	var returnList []velerov1api.Backup
+	orderedBackupList := queuePositionOrderFunc(l.backups).(*velerov1api.BackupList)
+	for _, item := range orderedBackupList.Items {
+		if item.Status.Phase == velerov1api.BackupPhaseQueued {
+			returnList = append(returnList, item)
+		}
+	}
+	return returnList
+}

pkg/controller/backup_queue_controller_test.go

@@ -0,0 +1,241 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package controller
+
+import (
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	//metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	//"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func TestBackupQueueReconciler(t *testing.T) {
+	scheme := runtime.NewScheme()
+	velerov1api.AddToScheme(scheme)
+
+	tests := []struct {
+		name                string
+		priorBackups        []*velerov1api.Backup
+		namespaces          []string
+		backup              *velerov1api.Backup
+		concurrentBackups   int
+		expectError         bool
+		expectPhase         velerov1api.BackupPhase
+		expectQueuePosition int
+	}{
+		{
+			name:                "New Backup gets queued",
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 1,
+		},
+		{
+			name:        "InProgress Backup is ignored",
+			backup:      builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).Result(),
+			expectPhase: velerov1api.BackupPhaseInProgress,
+		},
+		{
+			name: "Second New Backup gets queued with queuePosition 2",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).Result(),
+			},
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-12").Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 2,
+		},
+		{
+			name:        "Queued Backup moves to ReadyToStart if no others are running",
+			backup:      builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseQueued).Result(),
+			expectPhase: velerov1api.BackupPhaseReadyToStart,
+		},
+		{
+			name: "Queued Backup remains queued if no spaces available",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).Result(),
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-12").Phase(velerov1api.BackupPhaseInProgress).Result(),
+			},
+			concurrentBackups:   2,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 1,
+		},
+		{
+			name: "Queued Backup remains queued if no spaces available including ReadyToStart",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).Result(),
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-12").Phase(velerov1api.BackupPhaseReadyToStart).Result(),
+			},
+			concurrentBackups:   2,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 1,
+		},
+		{
+			name: "Queued Backup remains queued if earlier runnable backup is also queued",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).Result(),
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-12").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).Result(),
+			},
+			concurrentBackups:   3,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(2).Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 2,
+		},
+		{
+			name: "Queued Backup remains queued if in conflict with running backup",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).IncludedNamespaces("foo").Result(),
+			},
+			namespaces:          []string{"foo"},
+			concurrentBackups:   3,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).IncludedNamespaces("foo").Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 1,
+		},
+		{
+			name: "Queued Backup remains queued if in conflict with ReadyToStart backup",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseReadyToStart).IncludedNamespaces("foo").Result(),
+			},
+			namespaces:          []string{"foo"},
+			concurrentBackups:   3,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).IncludedNamespaces("foo").Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 1,
+		},
+		{
+			name: "Queued Backup remains queued if in conflict with earlier queued backup",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).IncludedNamespaces("foo").Result(),
+			},
+			namespaces:          []string{"foo", "bar"},
+			concurrentBackups:   3,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(2).IncludedNamespaces("foo", "bar").Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 2,
+		},
+		{
+			name: "Queued Backup remains queued if earlier non-ns-conflict backup exists",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).IncludedNamespaces("bar").Result(),
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-12").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).IncludedNamespaces("foo").Result(),
+			},
+			namespaces:          []string{"foo", "bar", "baz"},
+			concurrentBackups:   3,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(2).IncludedNamespaces("baz").Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 2,
+		},
+		{
+			name: "Running all-namespace backup conflicts with queued one-namespace backup ",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).IncludedNamespaces("*").Result(),
+			},
+			namespaces:          []string{"foo", "bar"},
+			concurrentBackups:   3,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).IncludedNamespaces("foo").Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 1,
+		},
+		{
+			name: "Running one-namespace backup conflicts with queued all-namespace backup ",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).IncludedNamespaces("bar").Result(),
+			},
+			namespaces:          []string{"foo", "bar"},
+			concurrentBackups:   3,
+			backup:              builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).IncludedNamespaces("*").Result(),
+			expectPhase:         velerov1api.BackupPhaseQueued,
+			expectQueuePosition: 1,
+		},
+		{
+			name: "Queued Backup moves to ReadyToStart if running count < concurrentBackups",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseInProgress).Result(),
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-12").Phase(velerov1api.BackupPhaseInProgress).Result(),
+			},
+			concurrentBackups: 3,
+			backup:            builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).Result(),
+			expectPhase:       velerov1api.BackupPhaseReadyToStart,
+		},
+		{
+			name: "Queued Backup moves to ReadyToStart if running count < concurrentBackups and no ns conflict found",
+			priorBackups: []*velerov1api.Backup{
+				builder.ForBackup(velerov1api.DefaultNamespace, "backup-11").Phase(velerov1api.BackupPhaseReadyToStart).IncludedNamespaces("foo").Result(),
+			},
+			namespaces:        []string{"foo", "bar"},
+			concurrentBackups: 3,
+			backup:            builder.ForBackup(velerov1api.DefaultNamespace, "backup-20").Phase(velerov1api.BackupPhaseQueued).QueuePosition(1).IncludedNamespaces("bar").Result(),
+			expectPhase:       velerov1api.BackupPhaseReadyToStart,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if test.backup == nil {
+				return
+			}
+
+			backupTracker := NewBackupTracker()
+			initObjs := []runtime.Object{}
+			for _, priorBackup := range test.priorBackups {
+				initObjs = append(initObjs, priorBackup)
+				if priorBackup.Status.Phase == velerov1api.BackupPhaseReadyToStart {
+					backupTracker.AddReadyToStart(priorBackup.Namespace, priorBackup.Name)
+				} else if priorBackup.Status.Phase == velerov1api.BackupPhaseInProgress {
+					backupTracker.Add(priorBackup.Namespace, priorBackup.Name)
+				}
+			}
+			for _, ns := range test.namespaces {
+				initObjs = append(initObjs, builder.ForNamespace(ns).Result())
+			}
+			initObjs = append(initObjs, test.backup)
+
+			fakeClient := velerotest.NewFakeControllerRuntimeClient(t, initObjs...)
+			logger := logrus.New()
+			log := logger.WithField("controller", "backup-queue-test")
+			r := NewBackupQueueReconciler(fakeClient, scheme, log, test.concurrentBackups, backupTracker)
+			req := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: test.backup.Namespace, Name: test.backup.Name}}
+			res, err := r.Reconcile(t.Context(), req)
+			gotErr := err != nil
+			require.NoError(t, err)
+			assert.Equal(t, ctrl.Result{}, res)
+			assert.Equal(t, test.expectError, gotErr)
+			backupAfter := velerov1api.Backup{}
+			err = fakeClient.Get(t.Context(), types.NamespacedName{
+				Namespace: test.backup.Namespace,
+				Name:      test.backup.Name,
+			}, &backupAfter)
+
+			require.NoError(t, err)
+			assert.Equal(t, test.expectPhase, backupAfter.Status.Phase)
+			assert.Equal(t, test.expectQueuePosition, backupAfter.Status.QueuePosition)
+		})
+	}
+}

pkg/controller/backup_tracker.go

@@ -25,45 +25,90 @@ import (
 
 // BackupTracker keeps track of in-progress backups.
 type BackupTracker interface {
+	// Add informs the tracker that a backup is ReadyToStart.
+	AddReadyToStart(ns, name string)
 	// Add informs the tracker that a backup is in progress.
 	Add(ns, name string)
-	// Delete informs the tracker that a backup is no longer in progress.
+	// Add informs the tracker that a backup has moved beyond InProgress
+	AddPostProcessing(ns, name string)
+	// Delete informs the tracker that a backup has reached a terminal state.
 	Delete(ns, name string)
-	// Contains returns true if the tracker is tracking the backup.
+	// Contains returns true if backup is InProgress or post-InProgress
 	Contains(ns, name string) bool
+	// RunningCount returns the number of backups which are ReadyToStart or InProgress
+	RunningCount() int
 }
 
 type backupTracker struct {
-	lock    sync.RWMutex
-	backups sets.Set[string]
+	lock                sync.RWMutex
+	readyToStartBackups sets.Set[string]
+	inProgressBackups   sets.Set[string]
+	postProgressBackups sets.Set[string]
 }
 
 // NewBackupTracker returns a new BackupTracker.
 func NewBackupTracker() BackupTracker {
 	return &backupTracker{
-		backups: sets.New[string](),
+		readyToStartBackups: sets.New[string](),
+		inProgressBackups:   sets.New[string](),
+		postProgressBackups: sets.New[string](),
 	}
 }
 
+func (bt *backupTracker) AddReadyToStart(ns, name string) {
+	bt.lock.Lock()
+	defer bt.lock.Unlock()
+
+	bt.readyToStartBackups.Insert(backupTrackerKey(ns, name))
+}
+
 func (bt *backupTracker) Add(ns, name string) {
 	bt.lock.Lock()
 	defer bt.lock.Unlock()
 
-	bt.backups.Insert(backupTrackerKey(ns, name))
+	key := backupTrackerKey(ns, name)
+	bt.readyToStartBackups.Delete(key)
+	bt.inProgressBackups.Insert(key)
+}
+
+func (bt *backupTracker) AddPostProcessing(ns, name string) {
+	bt.lock.Lock()
+	defer bt.lock.Unlock()
+
+	key := backupTrackerKey(ns, name)
+	bt.readyToStartBackups.Delete(key)
+	bt.inProgressBackups.Delete(key)
+	bt.postProgressBackups.Insert(key)
 }
 
 func (bt *backupTracker) Delete(ns, name string) {
 	bt.lock.Lock()
 	defer bt.lock.Unlock()
 
-	bt.backups.Delete(backupTrackerKey(ns, name))
+	key := backupTrackerKey(ns, name)
+	bt.readyToStartBackups.Delete(key)
+	bt.inProgressBackups.Delete(key)
+	bt.postProgressBackups.Delete(key)
 }
 
+// Contains returns true if backup is InProgress or post-InProgress
+// ignores ReadyToStart, since this is used to determine whether
+// a backup is in progress and thus not able to be deleted now.
 func (bt *backupTracker) Contains(ns, name string) bool {
 	bt.lock.RLock()
 	defer bt.lock.RUnlock()
 
-	return bt.backups.Has(backupTrackerKey(ns, name))
+	key := backupTrackerKey(ns, name)
+	return bt.inProgressBackups.Has(key) || bt.postProgressBackups.Has(key)
+}
+
+// RunningCount returns the number of backups which are ReadyToStart or InProgress
+// used by queue controller to determine whether a new backup can be started.
+func (bt *backupTracker) RunningCount() int {
+	bt.lock.RLock()
+	defer bt.lock.RUnlock()
+
+	return bt.inProgressBackups.Len() + bt.readyToStartBackups.Len()
 }
 
 func backupTrackerKey(ns, name string) string {

pkg/install/deployment.go

@@ -59,6 +59,7 @@ type podTemplateConfig struct {
 	repoMaintenanceJobConfigMap     string
 	nodeAgentConfigMap              string
 	itemBlockWorkerCount            int
+	concurrentBackups               int
 	forWindows                      bool
 	kubeletRootDir                  string
 	nodeAgentDisableHostPath        bool
@@ -224,6 +225,12 @@ func WithItemBlockWorkerCount(itemBlockWorkerCount int) podTemplateOption {
 	}
 }
 
+func WithConcurrentBackups(concurrentBackups int) podTemplateOption {
+	return func(c *podTemplateConfig) {
+		c.concurrentBackups = concurrentBackups
+	}
+}
+
 func WithPriorityClassName(priorityClassName string) podTemplateOption {
 	return func(c *podTemplateConfig) {
 		c.priorityClassName = priorityClassName
@@ -337,6 +344,10 @@ func Deployment(namespace string, opts ...podTemplateOption) *appsv1api.Deployme
 		args = append(args, fmt.Sprintf("--item-block-worker-count=%d", c.itemBlockWorkerCount))
 	}
 
+	if c.concurrentBackups > 0 {
+		args = append(args, fmt.Sprintf("--concurrent-backups=%d", c.concurrentBackups))
+	}
+
 	deployment := &appsv1api.Deployment{
 		ObjectMeta: objectMeta(namespace, "velero"),
 		TypeMeta: metav1.TypeMeta{

pkg/install/resources.go

@@ -271,6 +271,7 @@ type VeleroOptions struct {
 	RepoMaintenanceJobConfigMap     string
 	NodeAgentConfigMap              string
 	ItemBlockWorkerCount            int
+	ConcurrentBackups               int
 	KubeletRootDir                  string
 	NodeAgentDisableHostPath        bool
 	ServerPriorityClassName         string
@@ -362,6 +363,7 @@ func AllResources(o *VeleroOptions) *unstructured.UnstructuredList {
 		WithPodResources(o.PodResources),
 		WithKeepLatestMaintenanceJobs(o.KeepLatestMaintenanceJobs),
 		WithItemBlockWorkerCount(o.ItemBlockWorkerCount),
+		WithConcurrentBackups(o.ConcurrentBackups),
 	}
 
 	if o.ServerPriorityClassName != "" {

site/content/docs/main/customize-installation.md

@@ -158,7 +158,7 @@ At installation, You could set resource requests and limits for the Velero pod,
 |Memory limit|512Mi|N/A|
 {{< /table >}}
   
-For Velero pod, through testing, the Velero maintainers have found these defaults work well when backing up and restoring 1000 or less resources.  
+For Velero pod, through testing, the Velero maintainers have found these defaults work well when backing up and restoring 1000 or less resources. If you are enabling concurrent backups and your backups tend to be large, you may need to increase these limits.
 For node-agent pod, by default it doesn't have CPU/memory request/limit, so that the backups/restores won't break due to resource throttling. The Velero maintainers have also done some [Performance Tests][13] to show the relationship of CPU/memory usage and the scale of data being backed up/restored.
 
 For repository maintenance job, it's no limit on resources by default. You could configure the job resource limitation based on target data to be backed up, some further settings please refer to [repository maintenance job][14].