core, client: improve log readability during the flush process

core/backup_impl_create_backup.go

@@ -398,7 +398,7 @@ func (b *BackupContext) backupCollectionPrepare(ctx context.Context, backupInfo
 			zap.Int("segmentNumBeforeFlush", len(segmentEntitiesBeforeFlush)))
 		flushResp, err := b.getMilvusClient().Flush(ctx, collectionBackup.GetDbName(), collectionBackup.GetCollectionName())
 		if err != nil {
-			log.Error("fail to flush the collection",
+			log.Warn("fail to flush the collection",
 				zap.String("databaseName", collectionBackup.GetDbName()),
 				zap.String("collectionName", collectionBackup.GetCollectionName()),
 				zap.Error(err))
@@ -455,7 +455,7 @@ func (b *BackupContext) backupCollectionPrepare(ctx context.Context, backupInfo
 		// Flush
 		segmentEntitiesBeforeFlush, err := b.getMilvusClient().GetPersistentSegmentInfo(ctx, collectionBackup.GetDbName(), collectionBackup.GetCollectionName())
 		if err != nil {
-			log.Error(fmt.Sprintf("fail to flush the collection: %s", collectionBackup.GetCollectionName()), zap.Error(err))
+			log.Warn(fmt.Sprintf("fail to flush the collection: %s", collectionBackup.GetCollectionName()), zap.Error(err))
 			return err
 		}
 		log.Info("GetPersistentSegmentInfo from milvus",

core/client/grpc.go

@@ -159,6 +159,8 @@ func (l *limiters) close() {
 var _ Grpc = (*GrpcClient)(nil)
 
 type GrpcClient struct {
+	logger *zap.Logger
+
 	conn *grpc.ClientConn
 	srv  milvuspb.MilvusServiceClient
 
@@ -254,6 +256,8 @@ func NewGrpc(cfg *paramtable.MilvusConfig) (*GrpcClient, error) {
 	srv := milvuspb.NewMilvusServiceClient(conn)
 
 	cli := &GrpcClient{
+		logger: log.L().With(zap.String("component", "grpc-client")),
+
 		conn: conn,
 		srv:  srv,
 
@@ -440,32 +444,44 @@ func (g *GrpcClient) Flush(ctx context.Context, db, collName string) (*milvuspb.
 	segmentIDs, has := resp.GetCollSegIDs()[collName]
 	ids := segmentIDs.GetData()
 	if has && len(ids) > 0 {
-		flushed := func() bool {
-			getFlushResp, err := g.srv.GetFlushState(ctx, &milvuspb.GetFlushStateRequest{
-				SegmentIDs:     ids,
-				FlushTs:        resp.GetCollFlushTs()[collName],
-				CollectionName: collName,
-			})
-			if err != nil {
-				// TODO max retry
-				return false
-			}
-			return getFlushResp.GetFlushed()
-		}
-		for !flushed() {
-			// respect context deadline/cancel
-			select {
-			case <-ctx.Done():
-				return nil, errors.New("deadline exceeded")
-			default:
-			}
-			time.Sleep(200 * time.Millisecond)
+		flushTS := resp.GetCollFlushTs()[collName]
+		if err := g.checkFlush(ctx, ids, flushTS, collName); err != nil {
+			return nil, fmt.Errorf("client: check flush failed: %w", err)
 		}
 	}
 
 	return resp, nil
 }
 
+func (g *GrpcClient) checkFlush(ctx context.Context, segIDs []int64, flushTS uint64, collName string) error {
+	start := time.Now()
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		resp, err := g.srv.GetFlushState(ctx, &milvuspb.GetFlushStateRequest{
+			SegmentIDs:     segIDs,
+			FlushTs:        flushTS,
+			CollectionName: collName,
+		})
+		if err != nil {
+			g.logger.Warn("get flush state failed, will retry", zap.Error(err))
+		}
+		if resp.GetFlushed() {
+			return nil
+		}
+
+		cost := time.Since(start)
+		if cost > 30*time.Minute {
+			g.logger.Warn("waiting for the flush to complete took too much time!",
+				zap.Duration("cost", cost),
+				zap.String("collection", collName))
+		}
+	}
+
+	return errors.New("client: into an dead end, should not reach here")
+}
+
 func (g *GrpcClient) ListCollections(ctx context.Context, db string) (*milvuspb.ShowCollectionsResponse, error) {
 	ctx = g.newCtxWithDB(ctx, db)
 	resp, err := g.srv.ShowCollections(ctx, &milvuspb.ShowCollectionsRequest{})