add parallel snapshots

anton.voskresensky · anton.voskresensky · commit fb623c7bf00b · 2025-12-01T00:02:40.000+03:00
diff --git a/commands/snapshots.go b/commands/snapshots.go
@@ -362,7 +362,7 @@ func runSnapshot(cmd *cobra.Command, args []string) error {
 		}
 
 		if len(snapshotTasks) > 0 {
-			successful, failed := utils.CreateSnapshotsInParallel(client, snapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger)
+			successful, failed := utils.CreateSnapshotsInParallel(client, snapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger, true)
 			successfulSnapshots = append(successfulSnapshots, successful...)
 			failedSnapshots = append(failedSnapshots, failed...)
 		}
@@ -474,7 +474,7 @@ func runSnapshot(cmd *cobra.Command, args []string) error {
 				}
 			}
 			if len(repoSnapshotTasks) > 0 {
-				successful, failed := utils.CreateSnapshotsInParallel(client, repoSnapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger)
+				successful, failed := utils.CreateSnapshotsInParallel(client, repoSnapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger, true)
 				successfulSnapshots = append(successfulSnapshots, successful...)
 				failedSnapshots = append(failedSnapshots, failed...)
 			}
diff --git a/commands/snapshotsbackfill.go b/commands/snapshotsbackfill.go
@@ -621,7 +621,7 @@ func runSnapshotsBackfill(cmd *cobra.Command, args []string) error {
 			}
 
 			if len(snapshotTasks) > 0 {
-				successful, failed := utils.CreateSnapshotsInParallel(client, snapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger)
+				successful, failed := utils.CreateSnapshotsInParallel(client, snapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger, false)
 				successfulSnapshots = append(successfulSnapshots, successful...)
 				failedSnapshots = append(failedSnapshots, failed...)
 			}
@@ -733,7 +733,7 @@ func runSnapshotsBackfill(cmd *cobra.Command, args []string) error {
 					}
 				}
 				if len(repoSnapshotTasks) > 0 {
-					successful, failed := utils.CreateSnapshotsInParallel(client, repoSnapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger)
+					successful, failed := utils.CreateSnapshotsInParallel(client, repoSnapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger, false)
 					successfulSnapshots = append(successfulSnapshots, successful...)
 					failedSnapshots = append(failedSnapshots, failed...)
 				}
diff --git a/pkg/utils/snapshots.go b/pkg/utils/snapshots.go
@@ -166,7 +166,7 @@ type SnapshotTask struct {
 	Size         int64
 }
 
-func CreateSnapshotsInParallel(client *opensearch.Client, tasks []SnapshotTask, maxConcurrent int, madisonClient interface{}, logger *logging.Logger) ([]string, []string) {
+func CreateSnapshotsInParallel(client *opensearch.Client, tasks []SnapshotTask, maxConcurrent int, madisonClient interface{}, logger *logging.Logger, sortDescending bool) ([]string, []string) {
 	var successful []string
 	var failed []string
 	var mu sync.Mutex
@@ -175,7 +175,10 @@ func CreateSnapshotsInParallel(client *opensearch.Client, tasks []SnapshotTask,
 	sortedTasks := make([]SnapshotTask, len(tasks))
 	copy(sortedTasks, tasks)
 	sort.Slice(sortedTasks, func(i, j int) bool {
-		return sortedTasks[i].Size > sortedTasks[j].Size
+		if sortDescending {
+			return sortedTasks[i].Size > sortedTasks[j].Size
+		}
+		return sortedTasks[i].Size < sortedTasks[j].Size
 	})
 
 	taskChan := make(chan SnapshotTask, len(sortedTasks))

Original file line number	Diff line number	Diff line change
`@@ -362,7 +362,7 @@ func runSnapshot(cmd *cobra.Command, args []string) error {`
`362`	`362`	`}`
`363`	`363`
`364`	`364`	`if len(snapshotTasks) > 0 {`
`365`		`- successful, failed := utils.CreateSnapshotsInParallel(client, snapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger)`
	`365`	`+ successful, failed := utils.CreateSnapshotsInParallel(client, snapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger, true)`
`366`	`366`	`successfulSnapshots = append(successfulSnapshots, successful...)`
`367`	`367`	`failedSnapshots = append(failedSnapshots, failed...)`
`368`	`368`	`}`
`@@ -474,7 +474,7 @@ func runSnapshot(cmd *cobra.Command, args []string) error {`
`474`	`474`	`}`
`475`	`475`	`}`
`476`	`476`	`if len(repoSnapshotTasks) > 0 {`
`477`		`- successful, failed := utils.CreateSnapshotsInParallel(client, repoSnapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger)`
	`477`	`+ successful, failed := utils.CreateSnapshotsInParallel(client, repoSnapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger, true)`
`478`	`478`	`successfulSnapshots = append(successfulSnapshots, successful...)`
`479`	`479`	`failedSnapshots = append(failedSnapshots, failed...)`
`480`	`480`	`}`
Original file line number	Diff line number	Diff line change
`@@ -621,7 +621,7 @@ func runSnapshotsBackfill(cmd *cobra.Command, args []string) error {`
`621`	`621`	`}`
`622`	`622`
`623`	`623`	`if len(snapshotTasks) > 0 {`
`624`		`- successful, failed := utils.CreateSnapshotsInParallel(client, snapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger)`
	`624`	`+ successful, failed := utils.CreateSnapshotsInParallel(client, snapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger, false)`
`625`	`625`	`successfulSnapshots = append(successfulSnapshots, successful...)`
`626`	`626`	`failedSnapshots = append(failedSnapshots, failed...)`
`627`	`627`	`}`
`@@ -733,7 +733,7 @@ func runSnapshotsBackfill(cmd *cobra.Command, args []string) error {`
`733`	`733`	`}`
`734`	`734`	`}`
`735`	`735`	`if len(repoSnapshotTasks) > 0 {`
`736`		`- successful, failed := utils.CreateSnapshotsInParallel(client, repoSnapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger)`
	`736`	`+ successful, failed := utils.CreateSnapshotsInParallel(client, repoSnapshotTasks, cfg.GetMaxConcurrentSnapshots(), madisonClient, logger, false)`
`737`	`737`	`successfulSnapshots = append(successfulSnapshots, successful...)`
`738`	`738`	`failedSnapshots = append(failedSnapshots, failed...)`
`739`	`739`	`}`