Add force backup option (#195)

Signed-off-by: wayblink <[email protected]>
zilliztech · Oct 16, 2023 · b616e60 · b616e60
1 parent 6d4dbbf
commit b616e60
Show file tree

Hide file tree

Showing 5 changed files with 256 additions and 224 deletions.
diff --git a/cmd/create.go b/cmd/create.go
@@ -18,6 +18,7 @@ var (
 	collectionNames string
 	databases       string
 	dbCollections   string
+	force           bool
 )
 
 var createBackupCmd = &cobra.Command{
@@ -57,6 +58,7 @@ var createBackupCmd = &cobra.Command{
 			BackupName:      backupName,
 			CollectionNames: collectionNameArr,
 			DbCollections:   utils.WrapDBCollections(dbCollections),
+			Force:           force,
 		})
 		fmt.Println(resp.GetCode(), "\n", resp.GetMsg())
 	},
@@ -66,7 +68,8 @@ func init() {
 	createBackupCmd.Flags().StringVarP(&backupName, "name", "n", "", "backup name, if unset will generate a name automatically")
 	createBackupCmd.Flags().StringVarP(&collectionNames, "colls", "c", "", "collectionNames to backup, use ',' to connect multiple collections")
 	createBackupCmd.Flags().StringVarP(&databases, "databases", "d", "", "databases to backup")
-	createBackupCmd.Flags().StringVarP(&dbCollections, "database_collections", "f", "", "databases and collections to backup, json format: {\"db1\":[\"c1\", \"c2\"],\"db2\":[]}")
+	createBackupCmd.Flags().StringVarP(&dbCollections, "database_collections", "a", "", "databases and collections to backup, json format: {\"db1\":[\"c1\", \"c2\"],\"db2\":[]}")
+	createBackupCmd.Flags().BoolVarP(&force, "force", "f", false, "force backup skip flush, should make sure data has been stored into disk when using it")
 
 	rootCmd.AddCommand(createBackupCmd)
 }
diff --git a/core/backup_impl_create_backup.go b/core/backup_impl_create_backup.go
@@ -390,64 +390,76 @@ func (b *BackupContext) executeCreateBackup(ctx context.Context, request *backup
 			}
 		}
 
-		// Flush
-		segmentEntitiesBeforeFlush, err := b.getMilvusClient().GetPersistentSegmentInfo(ctx, collection.GetCollectionName())
-		if err != nil {
-			return backupInfo, err
-		}
-		log.Info("GetPersistentSegmentInfo before flush from milvus",
-			zap.String("collectionName", collection.GetCollectionName()),
-			zap.Int("segmentNumBeforeFlush", len(segmentEntitiesBeforeFlush)))
-
-		newSealedSegmentIDs, flushedSegmentIDs, timeOfSeal, err := b.getMilvusClient().FlushV2(ctx, collection.GetCollectionName(), false)
-		if err != nil {
-			log.Error(fmt.Sprintf("fail to flush the collection: %s", collection.GetCollectionName()))
-			return backupInfo, err
-		}
-		log.Info("flush segments",
-			zap.String("collectionName", collection.GetCollectionName()),
-			zap.Int64s("newSealedSegmentIDs", newSealedSegmentIDs),
-			zap.Int64s("flushedSegmentIDs", flushedSegmentIDs),
-			zap.Int64("timeOfSeal", timeOfSeal))
-		collection.BackupTimestamp = utils.ComposeTS(timeOfSeal, 0)
-		collection.BackupPhysicalTimestamp = uint64(timeOfSeal)
-
-		flushSegmentIDs := append(newSealedSegmentIDs, flushedSegmentIDs...)
-		segmentEntitiesAfterFlush, err := b.getMilvusClient().GetPersistentSegmentInfo(ctx, collection.GetCollectionName())
-		if err != nil {
-			return backupInfo, err
-		}
-		log.Info("GetPersistentSegmentInfo after flush from milvus",
-			zap.String("collectionName", collection.GetCollectionName()),
-			zap.Int("segmentNumBeforeFlush", len(segmentEntitiesBeforeFlush)),
-			zap.Int("segmentNumAfterFlush", len(segmentEntitiesAfterFlush)))
-
 		// fill segments
 		filledSegments := make([]*entity.Segment, 0)
-		segmentDict := utils.ArrayToMap(flushSegmentIDs)
-		for _, seg := range segmentEntitiesAfterFlush {
-			sid := seg.ID
-			if _, ok := segmentDict[sid]; ok {
-				delete(segmentDict, sid)
-				filledSegments = append(filledSegments, seg)
-			} else {
-				log.Debug("this may be new segments after flush, skip it", zap.Int64("id", sid))
+		if !request.GetForce() {
+			// Flush
+			segmentEntitiesBeforeFlush, err := b.getMilvusClient().GetPersistentSegmentInfo(ctx, collection.GetCollectionName())
+			if err != nil {
+				return backupInfo, err
 			}
-		}
-		for _, seg := range segmentEntitiesBeforeFlush {
-			sid := seg.ID
-			if _, ok := segmentDict[sid]; ok {
-				delete(segmentDict, sid)
+			log.Info("GetPersistentSegmentInfo before flush from milvus",
+				zap.String("collectionName", collection.GetCollectionName()),
+				zap.Int("segmentNumBeforeFlush", len(segmentEntitiesBeforeFlush)))
+			newSealedSegmentIDs, flushedSegmentIDs, timeOfSeal, err := b.getMilvusClient().FlushV2(ctx, collection.GetCollectionName(), false)
+			if err != nil {
+				log.Error(fmt.Sprintf("fail to flush the collection: %s", collection.GetCollectionName()))
+				return backupInfo, err
+			}
+			log.Info("flush segments",
+				zap.String("collectionName", collection.GetCollectionName()),
+				zap.Int64s("newSealedSegmentIDs", newSealedSegmentIDs),
+				zap.Int64s("flushedSegmentIDs", flushedSegmentIDs),
+				zap.Int64("timeOfSeal", timeOfSeal))
+			collection.BackupTimestamp = utils.ComposeTS(timeOfSeal, 0)
+			collection.BackupPhysicalTimestamp = uint64(timeOfSeal)
+
+			flushSegmentIDs := append(newSealedSegmentIDs, flushedSegmentIDs...)
+			segmentEntitiesAfterFlush, err := b.getMilvusClient().GetPersistentSegmentInfo(ctx, collection.GetCollectionName())
+			if err != nil {
+				return backupInfo, err
+			}
+			log.Info("GetPersistentSegmentInfo after flush from milvus",
+				zap.String("collectionName", collection.GetCollectionName()),
+				zap.Int("segmentNumBeforeFlush", len(segmentEntitiesBeforeFlush)),
+				zap.Int("segmentNumAfterFlush", len(segmentEntitiesAfterFlush)))
+			segmentDict := utils.ArrayToMap(flushSegmentIDs)
+			for _, seg := range segmentEntitiesAfterFlush {
+				sid := seg.ID
+				if _, ok := segmentDict[sid]; ok {
+					delete(segmentDict, sid)
+					filledSegments = append(filledSegments, seg)
+				} else {
+					log.Debug("this may be new segments after flush, skip it", zap.Int64("id", sid))
+				}
+			}
+			for _, seg := range segmentEntitiesBeforeFlush {
+				sid := seg.ID
+				if _, ok := segmentDict[sid]; ok {
+					delete(segmentDict, sid)
+					filledSegments = append(filledSegments, seg)
+				} else {
+					log.Debug("this may be old segments before flush, skip it", zap.Int64("id", sid))
+				}
+			}
+			if len(segmentDict) > 0 {
+				// very rare situation, segments return in flush doesn't exist in either segmentEntitiesBeforeFlush and segmentEntitiesAfterFlush
+				errorMsg := "Segment return in Flush not exist in GetPersistentSegmentInfo. segment ids: " + fmt.Sprint(utils.MapKeyArray(segmentDict))
+				log.Warn(errorMsg)
+			}
+		} else {
+			// Flush
+			segmentEntitiesBeforeFlush, err := b.getMilvusClient().GetPersistentSegmentInfo(ctx, collection.GetCollectionName())
+			if err != nil {
+				return backupInfo, err
+			}
+			log.Info("GetPersistentSegmentInfo from milvus",
+				zap.String("collectionName", collection.GetCollectionName()),
+				zap.Int("segmentNum", len(segmentEntitiesBeforeFlush)))
+			for _, seg := range segmentEntitiesBeforeFlush {
 				filledSegments = append(filledSegments, seg)
-			} else {
-				log.Debug("this may be old segments before flush, skip it", zap.Int64("id", sid))
 			}
 		}
-		if len(segmentDict) > 0 {
-			// very rare situation, segments return in flush doesn't exist in either segmentEntitiesBeforeFlush and segmentEntitiesAfterFlush
-			errorMsg := "Segment return in Flush not exist in GetPersistentSegmentInfo. segment ids: " + fmt.Sprint(utils.MapKeyArray(segmentDict))
-			log.Warn(errorMsg)
-		}
 
 		if err != nil {
 			collection.StateCode = backuppb.BackupTaskStateCode_BACKUP_FAIL

diff --git a/core/backup_impl_restore_backup.go b/core/backup_impl_restore_backup.go
@@ -560,7 +560,13 @@ func (b *BackupContext) executeBulkInsert(ctx context.Context, coll string, part
 		zap.String("partition", partition),
 		zap.Strings("files", files),
 		zap.Int64("endTime", endTime))
-	taskId, err := b.getMilvusClient().BulkInsert(ctx, coll, partition, files, gomilvus.IsBackup(), gomilvus.WithEndTs(endTime))
+	var taskId int64
+	var err error
+	if endTime == 0 {
+		taskId, err = b.getMilvusClient().BulkInsert(ctx, coll, partition, files, gomilvus.IsBackup())
+	} else {
+		taskId, err = b.getMilvusClient().BulkInsert(ctx, coll, partition, files, gomilvus.IsBackup(), gomilvus.WithEndTs(endTime))
+	}
 	if err != nil {
 		log.Error("fail to bulk insert",
 			zap.Error(err),

diff --git a/core/proto/backup.proto b/core/proto/backup.proto
@@ -147,6 +147,8 @@ message CreateBackupRequest {
   bool async = 4;
   // database and collections to backup. A json string. To support database. 2023.7.7
   google.protobuf.Value db_collections = 5;
+  // force backup skip flush, Should make sure data has been stored into disk when using it
+  bool force = 6;
 }
 
 /**