From 30e95092751f96a9d4ded2be108232424eff9817 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Fri, 26 Apr 2019 17:49:17 +0800 Subject: [PATCH] fix snapshot create disk --- pkg/hostman/guestman/guesttasks.go | 39 ++++++++++++-------------- pkg/hostman/monitor/hmp.go | 2 +- pkg/hostman/monitor/qmp.go | 2 +- pkg/hostman/storageman/disk_local.go | 2 +- pkg/hostman/storageman/storage_base.go | 4 +-- pkg/util/fuseutils/fuseutils.go | 2 +- 6 files changed, 24 insertions(+), 27 deletions(-) diff --git a/pkg/hostman/guestman/guesttasks.go b/pkg/hostman/guestman/guesttasks.go index 6189f20092..0946f8afcb 100644 --- a/pkg/hostman/guestman/guesttasks.go +++ b/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() } } diff --git a/pkg/hostman/monitor/hmp.go b/pkg/hostman/monitor/hmp.go index 461b4e551a..d0ae0455ec 100644 --- a/pkg/hostman/monitor/hmp.go +++ b/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) diff --git a/pkg/hostman/monitor/qmp.go b/pkg/hostman/monitor/qmp.go index a176052a94..046d909163 100644 --- a/pkg/hostman/monitor/qmp.go +++ b/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)) } diff --git a/pkg/hostman/storageman/disk_local.go b/pkg/hostman/storageman/disk_local.go index 60fc91de49..64a9cb486a 100644 --- a/pkg/hostman/storageman/disk_local.go +++ b/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()) ) diff --git a/pkg/hostman/storageman/storage_base.go b/pkg/hostman/storageman/storage_base.go index 78116ae815..619434ef2c 100644 --- a/pkg/hostman/storageman/storage_base.go +++ b/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" { diff --git a/pkg/util/fuseutils/fuseutils.go b/pkg/util/fuseutils/fuseutils.go index 2fa397b767..50989067bb 100644 --- a/pkg/util/fuseutils/fuseutils.go +++ b/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 {