fix snapshot create disk

pkg/hostman/guestman/guesttasks.go

@@ -546,7 +546,7 @@ func (s *SGuestStreamDisksTask) checkBlockDrives() {
 }
 
 func (s *SGuestStreamDisksTask) onBlockDrivesSucc(res *jsonutils.JSONArray) {
-	streamDevs := []string{}
+	s.streamDevs = []string{}
 	drvs, _ := res.GetArray()
 	for _, drv := range drvs {
 		device, err := drv.GetString("device")
@@ -555,7 +555,7 @@ func (s *SGuestStreamDisksTask) onBlockDrivesSucc(res *jsonutils.JSONArray) {
 			continue
 		}
 		inserted, err := drv.Get("inserted")
-		if err != nil && inserted.Contains("file") && inserted.Contains("backing_file") {
+		if err == nil && inserted.Contains("file") && inserted.Contains("backing_file") {
 			var stream = false
 			idx := device[len(device)-1] - '0'
 			for i := 0; i < len(s.disksIdx); i++ {
@@ -566,11 +566,11 @@ func (s *SGuestStreamDisksTask) onBlockDrivesSucc(res *jsonutils.JSONArray) {
 			if !stream {
 				continue
 			}
-			streamDevs = append(streamDevs, device)
+			s.streamDevs = append(s.streamDevs, device)
 		}
 	}
-	s.streamDevs = streamDevs
-	if len(streamDevs) == 0 {
+	log.Infof("Stream devices: %v", s.streamDevs)
+	if len(s.streamDevs) == 0 {
 		s.taskComplete()
 	} else {
 		s.SyncStatus()
@@ -583,32 +583,29 @@ func (s *SGuestStreamDisksTask) startDoBlockStream() {
 		dev := s.streamDevs[0]
 		s.streamDevs = s.streamDevs[1:]
 		s.Monitor.BlockStream(dev, s.startWaitBlockStream)
+	} else {
+		s.taskComplete()
 	}
 }
 
 func (s *SGuestStreamDisksTask) startWaitBlockStream(res string) {
-	if s.c == nil {
-		s.c = make(chan struct{})
-		for {
-			select {
-			case <-s.c:
-				s.c = nil
-				return
-			case <-time.After(time.Second * 1):
-				s.Monitor.GetBlockJobCounts(s.checkStreamJobs)
-			}
+	log.Infof("Block stream command res: %s", res)
+	s.c = make(chan struct{})
+	for {
+		select {
+		case <-s.c:
+			s.c = nil
+			return
+		case <-time.After(time.Second * 3):
+			s.Monitor.GetBlockJobCounts(s.checkStreamJobs)
 		}
 	}
 }
 
 func (s *SGuestStreamDisksTask) checkStreamJobs(jobs int) {
 	if jobs == 0 {
-		if len(s.streamDevs) == 0 {
-			close(s.c)
-			s.taskComplete()
-		} else {
-			close(s.c)
-		}
+		close(s.c)
+		s.startDoBlockStream()
 	}
 }
 

pkg/hostman/monitor/hmp.go

@@ -381,7 +381,7 @@ func (m *HmpMonitor) DriveMirror(callback StringCallback, drive, target, syncMod
 
 func (m *HmpMonitor) BlockStream(drive string, callback StringCallback) {
 	var (
-		speed = 30 // MB/s
+		speed = 30 // MB/s speed limit 31457280 bytes/s
 		cmd   = fmt.Sprintf("block_stream %s %d", drive, speed)
 	)
 	m.Query(cmd, callback)

pkg/hostman/monitor/qmp.go

@@ -691,7 +691,7 @@ func (m *QmpMonitor) DriveMirror(callback StringCallback, drive, target, syncMod
 
 func (m *QmpMonitor) BlockStream(drive string, callback StringCallback) {
 	var (
-		speed = 30 // MB/s
+		speed = 30 * 1024 * 1024 // qmp speed default unit is byte
 		cb    = func(res *Response) {
 			callback(m.actionResult(res))
 		}

pkg/hostman/storageman/disk_local.go

@@ -155,7 +155,7 @@ func (d *SLocalDisk) CreateFromImageFuse(ctx context.Context, url string) error
 
 	var (
 		localPath   = d.Storage.GetFuseTmpPath()
-		mntPath     = path.Join(d.Storage.GetFuseMountPath())
+		mntPath     = path.Join(d.Storage.GetFuseMountPath(), d.Id)
 		contentPath = path.Join(mntPath, "content")
 		newImg, err = qemuimg.NewQemuImage(d.getPath())
 	)

pkg/hostman/storageman/storage_base.go

@@ -271,11 +271,11 @@ func (s *SBaseStorage) CreateDiskFromSnpashot(ctx context.Context, disk IDisk, c
 	var (
 		// diskPath            = path.Join(s.Path, createParams.DiskId)
 		snapshotUrl, _      = createParams.DiskInfo.GetString("snapshot_url")
-		transferProtocol, _ = createParams.DiskInfo.GetString("url")
+		transferProtocol, _ = createParams.DiskInfo.GetString("protocol")
 	)
 
 	if len(snapshotUrl) == 0 || len(transferProtocol) == 0 {
-		return nil, fmt.Errorf("Create disk form snapshot missing params snapshot url or protocol")
+		return nil, fmt.Errorf("Create disk from snapshot missing params snapshot url or protocol")
 	}
 
 	if transferProtocol == "url" {

pkg/util/fuseutils/fuseutils.go

@@ -59,7 +59,7 @@ func MountFusefs(fetcherfsPath, url, tmpdir, token, mntpath string, blocksize in
 	opts += fmt.Sprintf(",token=%s", token)
 	opts += fmt.Sprintf(",blocksize=%d", blocksize)
 
-	var cmd = []string{fetcherfsPath, "-s", "-p", opts, mntpath}
+	var cmd = []string{fetcherfsPath, "-s", "-o", opts, mntpath}
 	log.Infof("%s", strings.Join(cmd, " "))
 	_, err := procutils.NewCommand(cmd[0], cmd[1:]...).Run()
 	if err != nil {