fix on stream disk complete snapshots refcount not update

pkg/compute/models/guest_actions.go

@@ -3797,7 +3797,7 @@ func (self *SGuest) AllowPerformStreamDisksComplete(ctx context.Context, userCre
 func (self *SGuest) PerformStreamDisksComplete(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
 	for _, disk := range self.GetDisks() {
 		d := disk.GetDisk()
-		if len(d.SnapshotId) > 0 {
+		if len(d.SnapshotId) > 0 && d.GetMetadata("merge_snapshot", userCred) == "true" {
 			SnapshotManager.AddRefCount(d.SnapshotId, -1)
 			d.SetMetadata(ctx, "merge_snapshot", jsonutils.JSONFalse, userCred)
 		}

pkg/hostman/guestman/guestman.go

@@ -291,6 +291,11 @@ func (m *SGuestManager) LoadServer(sid string) {
 		log.Errorf("On load server error: %s", err)
 		return
 	}
+
+	if jsonutils.QueryBoolean(guest.Desc, "need_sync_stream_disks", false) {
+		go guest.sendStreamDisksComplete(context.Background())
+	}
+
 	m.CandidateServers[sid] = guest
 }
 

pkg/hostman/guestman/qemu-kvm.go

@@ -1271,6 +1271,7 @@ func (s *SKVMGuestInstance) streamDisksComplete(ctx context.Context) {
 		if jsonutils.QueryBoolean(disk, "merge_snapshot", false) {
 			d := disks[i].(*jsonutils.JSONDict)
 			d.Set("merge_snapshot", jsonutils.JSONFalse)
+			s.Desc.Set("need_sync_stream_disks", jsonutils.JSONTrue)
 		}
 	}
 	if err := s.SaveDesc(s.Desc); err != nil {
@@ -1279,10 +1280,24 @@ func (s *SKVMGuestInstance) streamDisksComplete(ctx context.Context) {
 	if err := s.delFlatFiles(ctx); err != nil {
 		log.Errorf("del flat files failed %s", err)
 	}
-	_, err := modules.Servers.PerformAction(hostutils.GetComputeSession(ctx),
-		s.Id, "stream-disks-complete", nil)
-	if err != nil {
-		log.Infof("stream disks complete sync error %s", err)
+	go s.sendStreamDisksComplete(ctx)
+}
+
+func (s *SKVMGuestInstance) sendStreamDisksComplete(ctx context.Context) {
+	for {
+		_, err := modules.Servers.PerformAction(hostutils.GetComputeSession(ctx),
+			s.Id, "stream-disks-complete", nil)
+		if err != nil {
+			log.Errorf("stream disks complete sync error %s", err)
+			time.Sleep(30 * time.Second)
+			continue
+		} else {
+			break
+		}
+	}
+	s.Desc.Remove("need_sync_stream_disks")
+	if err := s.SaveDesc(s.Desc); err != nil {
+		log.Errorf("save guest desc failed %s", err)
 	}
 }