Merge pull request #934 from wanyaoqi/bugfix/wyq/import-from-libvirt

一些bug修复

cmd/climc/shell/servers.go

@@ -18,9 +18,11 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"net"
 	"os"
 	"path/filepath"
 
+	yaml "gopkg.in/yaml.v2"
 	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
 
@@ -844,34 +846,100 @@ func init() {
 	})
 
 	type ServersImportFromLibvirtOptions struct {
-		CONFIG_FILE string `help:"JSON file describing servers from libvirt, e.g. 
-			{'hosts': 
-				[
-					{
-						'servers': [
-							{
-								'mac_ip': {'mac1': 'ip1', ...}
-							},
-						],
-						'xml_file_path': '/etc/libvirt/qemu',
-						'host_ip': '192.168.1.100',
-					},
-				],
-			}
-		"`
+		CONFIG_FILE string `help:"File Path describing servers from libvirt"`
 	}
+
+	type Servers struct {
+		Mac string `yaml:"mac"`
+		Ip  string `yaml:"ip"`
+	}
+
+	type Hosts struct {
+		HostIp      string    `yaml:"host_ip"`
+		XmlFilePath string    `yaml:"xml_file_path"`
+		Servers     []Servers `yaml:"servers"`
+	}
+
+	type LibvirtImportOptions struct {
+		Hosts []Hosts `yaml:"hosts"`
+	}
+
 	R(&ServersImportFromLibvirtOptions{}, "servers-import-from-libvirt", "Import servers from libvrt", func(s *mcclient.ClientSession, args *ServersImportFromLibvirtOptions) error {
-		rawConfig, err := ioutil.ReadFile(args.CONFIG_FILE)
+		var (
+			rawConfig []byte
+			err       error
+		)
+
+		rawConfig, err = ioutil.ReadFile(args.CONFIG_FILE)
 		if err != nil {
-			return fmt.Errorf("Read file error: %s", err)
+			return fmt.Errorf("Read config file %s error: %s", args.CONFIG_FILE, err)
 		}
 
-		config := &compute.SLibvirtImportConfig{}
-		err = json.Unmarshal(rawConfig, config)
-		if err != nil {
-			return fmt.Errorf("Parse config error %s", err)
+		var (
+			params []jsonutils.JSONObject
+			config = &compute.SLibvirtImportConfig{}
+		)
+
+		// Try parse as json first
+		{
+			err = json.Unmarshal(rawConfig, config)
+			if err != nil {
+				goto YAML
+			}
+			for i := 0; i < len(config.Hosts); i++ {
+				if nIp := net.ParseIP(config.Hosts[i].HostIp); nIp == nil {
+					return fmt.Errorf("Parse host ip %s failed", config.Hosts[i].HostIp)
+				}
+				for _, server := range config.Hosts[i].Servers {
+					for mac, ip := range server.MacIp {
+						if _, err := net.ParseMAC(mac); err != nil {
+							return fmt.Errorf("Parse mac %s error %s", mac, err)
+						}
+						if nIp := net.ParseIP(ip); nIp == nil {
+							return fmt.Errorf("Parse ip %s failed", ip)
+						}
+					}
+				}
+			}
+
+			goto REQUEST
 		}
-		params, err := jsonutils.Marshal(config.Hosts).GetArray()
+
+	YAML: // Try Parse as yaml
+		{
+			yamlConfig := &LibvirtImportOptions{}
+			err = yaml.Unmarshal(rawConfig, yamlConfig)
+			if err != nil {
+				return err
+			}
+
+			config.Hosts = make([]compute.SLibvirtHostConfig, len(yamlConfig.Hosts))
+			for i := 0; i < len(yamlConfig.Hosts); i++ {
+				if nIp := net.ParseIP(yamlConfig.Hosts[i].HostIp); nIp == nil {
+					return fmt.Errorf("Parse host ip %s failed", yamlConfig.Hosts[i].HostIp)
+				}
+				config.Hosts[i].HostIp = yamlConfig.Hosts[i].HostIp
+				config.Hosts[i].XmlFilePath = yamlConfig.Hosts[i].XmlFilePath
+				config.Hosts[i].Servers = make([]compute.SLibvirtServerConfig, len(yamlConfig.Hosts[i].Servers))
+				for j := 0; j < len(yamlConfig.Hosts[i].Servers); j++ {
+					config.Hosts[i].Servers[j].MacIp = make(map[string]string)
+					mac := yamlConfig.Hosts[i].Servers[j].Mac
+					_, err := net.ParseMAC(mac)
+					if err != nil {
+						return fmt.Errorf("Parse mac address %s error %s", mac, err)
+					}
+					ip := yamlConfig.Hosts[i].Servers[j].Ip
+					nIp := net.ParseIP(ip)
+					if len(nIp) == 0 {
+						return fmt.Errorf("Parse ip address %s failed", ip)
+					}
+					config.Hosts[i].Servers[j].MacIp[mac] = ip
+				}
+			}
+		}
+
+	REQUEST:
+		params, err = jsonutils.Marshal(config.Hosts).GetArray()
 		if err != nil {
 			return err
 		}

pkg/apis/compute/guest_const.go

@@ -72,13 +72,13 @@ const (
 	VM_REBUILD_ROOT       = "rebuild_root"
 	VM_REBUILD_ROOT_FAIL  = "rebuild_root_fail"
 
-	VM_START_SNAPSHOT  = "snapshot_start"
-	VM_SNAPSHOT        = "snapshot"
-	VM_SNAPSHOT_DELETE = "snapshot_delete"
-	VM_BLOCK_STREAM    = "block_stream"
-	VM_MIRROR_FAIL     = "mirror_failed"
-	VM_SNAPSHOT_SUCC   = "snapshot_succ"
-	VM_SNAPSHOT_FAILED = "snapshot_failed"
+	VM_START_SNAPSHOT    = "snapshot_start"
+	VM_SNAPSHOT          = "snapshot"
+	VM_SNAPSHOT_DELETE   = "snapshot_delete"
+	VM_BLOCK_STREAM      = "block_stream"
+	VM_BLOCK_STREAM_FAIL = "block_stream_fail"
+	VM_SNAPSHOT_SUCC     = "snapshot_succ"
+	VM_SNAPSHOT_FAILED   = "snapshot_failed"
 
 	VM_SYNCING_STATUS = "syncing"
 	VM_SYNC_CONFIG    = "sync_config"

pkg/cloudcommon/db/db_dispatcher.go

@@ -1245,7 +1245,7 @@ func reflectDispatcher(
 ) (jsonutils.JSONObject, error) {
 	result, err := reflectDispatcherInternal(
 		dispatcher, model, modelValue, ctx, userCred, operator, generalFuncName, funcPrefix, spec, query, data)
-	if err == nil && result == nil {
+	if model != nil && err == nil && result == nil {
 		return getItemDetails(dispatcher.modelManager, model, ctx, userCred, query)
 	} else {
 		return result, err

pkg/compute/models/guest_actions.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"net"
 	"net/http"
 	"strconv"
 	"strings"
@@ -2222,6 +2223,12 @@ func (self *SGuest) PerformStatus(ctx context.Context, userCred mcclient.TokenCr
 	if err != nil {
 		return nil, err
 	}
+
+	status, _ := data.GetString("status")
+	if len(self.BackupHostId) > 0 && status == api.VM_RUNNING {
+		self.SetMetadata(ctx, "__mirror_job_status", "ready", userCred)
+	}
+
 	if preStatus != self.Status && !self.isNotRunningStatus(preStatus) && self.isNotRunningStatus(self.Status) {
 		db.OpsLog.LogEvent(self, db.ACT_STOP, "", userCred)
 		if self.Status == api.VM_READY && !self.DisableDelete.Bool() && self.ShutdownBehavior == api.SHUTDOWN_TERMINATE {
@@ -2559,6 +2566,12 @@ func (self *SGuest) PerformSwitchToBackup(ctx context.Context, userCred mcclient
 	if len(self.BackupHostId) == 0 {
 		return nil, httperrors.NewBadRequestError("Guest no backup host")
 	}
+
+	mirrorJobStatus := self.GetMetadata("__mirror_job_status", userCred)
+	if mirrorJobStatus != "ready" {
+		return nil, httperrors.NewBadRequestError("Guest can't switch to backup, mirror job not ready")
+	}
+
 	oldStatus := self.Status
 	self.SetStatus(userCred, api.VM_SWITCH_TO_BACKUP, "Switch to backup")
 	deleteBackup := jsonutils.QueryBoolean(data, "delete_backup", false)
@@ -2669,6 +2682,21 @@ func (manager *SGuestManager) PerformBatchSetUserMetadata(ctx context.Context, u
 	return jsonutils.Marshal(guests), nil
 }
 
+func (self *SGuest) AllowPerformBlockStreamFailed(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool {
+	return db.IsAdminAllowPerform(userCred, self, "block-stream-failed")
+}
+
+func (self *SGuest) PerformBlockStreamFailed(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
+	if len(self.BackupHostId) > 0 {
+		self.SetMetadata(ctx, "__mirror_job_status", "failed", userCred)
+	}
+	if self.Status == api.VM_BLOCK_STREAM {
+		reason, _ := data.GetString("reason")
+		return nil, self.SetStatus(userCred, api.VM_BLOCK_STREAM_FAIL, reason)
+	}
+	return nil, nil
+}
+
 func (manager *SGuestManager) AllowPerformDirtyServerStart(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool {
 	return db.IsAdminAllowClassPerform(userCred, manager, "dirty-server-start")
 }
@@ -2853,18 +2881,6 @@ func (self *SGuest) StartCreateBackup(ctx context.Context, userCred mcclient.Tok
 	return nil
 }
 
-func (self *SGuest) AllowPerformMirrorJobFailed(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool {
-	return db.IsAdminAllowPerform(userCred, self, "mirror-job-failed")
-}
-
-func (self *SGuest) PerformMirrorJobFailed(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
-	if len(self.BackupHostId) == 0 {
-		return nil, nil
-	} else {
-		return nil, self.SetStatus(userCred, api.VM_MIRROR_FAIL, "OnSyncToBackup")
-	}
-}
-
 func (self *SGuest) AllowPerformSetExtraOption(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool {
 	return db.IsAdminAllowPerform(userCred, self, "set-extra-option")
 }
@@ -3097,6 +3113,14 @@ func (self *SGuest) importNics(ctx context.Context, userCred mcclient.TokenCrede
 		return httperrors.NewInputParameterError("Empty import nics")
 	}
 	for _, nic := range nics {
+		q := GuestnetworkManager.Query()
+		count := q.Filter(sqlchemy.OR(
+			sqlchemy.Equals(q.Field("mac_addr"), nic.Mac),
+			sqlchemy.Equals(q.Field("ip_addr"), nic.Ip)),
+		).Count()
+		if count > 0 {
+			return httperrors.NewInputParameterError("ip %s or mac %s has been registered", nic.Ip, nic.Mac)
+		}
 		net, err := NetworkManager.GetOnPremiseNetworkOfIP(nic.Ip, "", tristate.None)
 		if err != nil {
 			return httperrors.NewNotFoundError("Not found network by ip %s", nic.Ip)
@@ -3162,6 +3186,27 @@ func (manager *SGuestManager) PerformImportFromLibvirt(ctx context.Context, user
 		return nil, httperrors.NewInputParameterError("Invalid host ip %s", host.HostIp)
 	}
 
+	for _, server := range host.Servers {
+		for mac, ip := range server.MacIp {
+			_, err = net.ParseMAC(mac)
+			if err != nil {
+				return nil, httperrors.NewBadRequestError("Invalid server mac address %s", mac)
+			}
+			nIp := net.ParseIP(ip)
+			if nIp == nil {
+				return nil, httperrors.NewBadRequestError("Invalid server ip address %s", ip)
+			}
+			q := GuestnetworkManager.Query()
+			count := q.Filter(sqlchemy.OR(
+				sqlchemy.Equals(q.Field("mac_addr"), mac),
+				sqlchemy.Equals(q.Field("ip_addr"), ip)),
+			).Count()
+			if count > 0 {
+				return nil, httperrors.NewInputParameterError("ip %s or mac %s has been registered", mac, ip)
+			}
+		}
+	}
+
 	taskData := jsonutils.NewDict()
 	taskData.Set("xml_file_path", jsonutils.NewString(host.XmlFilePath))
 	taskData.Set("servers", jsonutils.Marshal(host.Servers))

pkg/compute/models/hosts.go

@@ -2821,7 +2821,7 @@ func (self *SHost) PerformOffline(ctx context.Context, userCred mcclient.TokenCr
 			return nil, err
 		}
 		db.OpsLog.LogEvent(self, db.ACT_OFFLINE, "", userCred)
-		logclient.AddActionLogWithContext(ctx, self, logclient.ACT_ONLINE, nil, userCred, true)
+		logclient.AddActionLogWithContext(ctx, self, logclient.ACT_OFFLINE, nil, userCred, true)
 		self.SyncAttachedStorageStatus()
 	}
 	return nil, nil

pkg/compute/tasks/guest_live_migrate_task.go

@@ -100,7 +100,7 @@ func (self *GuestMigrateTask) SaveScheduleResult(ctx context.Context, obj ISched
 
 	self.SetStage("OnCachedImageComplete", body)
 	// prepare disk for migration
-	if isLocalStorage {
+	if len(disk.TemplateId) > 0 && isLocalStorage {
 		targetStorageCache := targetHost.GetLocalStoragecache()
 		if targetStorageCache != nil {
 			err := targetStorageCache.StartImageCacheTask(
@@ -108,10 +108,10 @@ func (self *GuestMigrateTask) SaveScheduleResult(ctx context.Context, obj ISched
 			if err != nil {
 				self.TaskFailed(ctx, guest, err.Error())
 			}
+			return
 		}
-	} else {
-		self.OnSrcPrepareComplete(ctx, guest, nil)
 	}
+	self.OnSrcPrepareComplete(ctx, guest, nil)
 }
 
 // For local storage get disk info

pkg/hostman/downloader/downloader.go

@@ -19,6 +19,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"path/filepath"
 	"time"
 
 	"yunion.io/x/log"
@@ -61,6 +62,10 @@ func (d *SDownloadProvider) Start(
 	}
 
 	log.Infof("Downloader Start Transfer %s, compress %t", downloadFilePath, d.compress)
+	spath, err := filepath.EvalSymlinks(downloadFilePath)
+	if err == nil {
+		downloadFilePath = spath
+	}
 	fi, err := os.Open(downloadFilePath)
 	if err != nil {
 		log.Errorln(err)

pkg/hostman/guesthandlers/guesthandler.go

@@ -294,7 +294,7 @@ func guestDestPrepareMigrate(ctx context.Context, sid string, body jsonutils.JSO
 		}
 		disksBack, err := body.Get("disks_back")
 		if err != nil {
-			return nil, httperrors.NewMissingParameterError("disks_back")
+			params.DisksBackingFile = jsonutils.NewDict()
 		} else {
 			params.DisksBackingFile = disksBack
 		}

pkg/hostman/guestman/guestman.go

@@ -482,9 +482,6 @@ func (m *SGuestManager) DestPrepareMigrate(ctx context.Context, params interface
 			return nil, fmt.Errorf("Target storage %s not found", migParams.TargetStorageId)
 		}
 
-		// 可能可以不用?
-		// guest.CreateFromUrl(ctx, migParams.ServerUrl, migParams.Desc)
-
 		disks, _ := migParams.Desc.GetArray("disks")
 		for i, diskinfo := range disks {
 			var (
@@ -539,7 +536,6 @@ func (m *SGuestManager) DestPrepareMigrate(ctx context.Context, params interface
 			diskDesc.Set("path", jsonutils.NewString(disk.GetPath()))
 		}
 
-		// 可能可以不要
 		if err := guest.SaveDesc(migParams.Desc); err != nil {
 			log.Errorln(err)
 			return nil, err

pkg/hostman/guestman/libvirt.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"path"
+	"strings"
 	"unicode"
 
 	libvirtxml "github.com/libvirt/libvirt-go-xml"
@@ -251,7 +252,7 @@ func (m *SGuestManager) LibvirtDomainDiskToDiskConfig(
 
 		// XXX: Ignore backing file
 		var diskConfig = compute.SImportDisk{
-			AccessPath: disk.Source.File.File,
+			AccessPath: strings.Trim(disk.Source.File.File, "\""),
 			Index:      int(disk.Source.Index),
 		}
 		if disk.Target.Bus != "virtio" {

pkg/hostman/guestman/qemu-kvm.go

@@ -392,6 +392,16 @@ func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) {
 				}
 			}
 		}
+	} else if event.Event == `"BLOCK_JOB_ERROR"` {
+		params := jsonutils.NewDict()
+		params.Set("reason", jsonutils.NewString("BLOCK_JOB_ERROR"))
+		_, err := modules.Servers.PerformAction(
+			hostutils.GetComputeSession(context.Background()),
+			s.GetId(), "block-stream-failed", params,
+		)
+		if err != nil {
+			log.Errorf("Server %s perform block-stream-failed got error %s", s.GetId(), err)
+		}
 	}
 }
 

pkg/hostman/hostinfo/hostinfo.go

@@ -38,6 +38,7 @@ import (
 	"yunion.io/x/onecloud/pkg/hostman/isolated_device"
 	"yunion.io/x/onecloud/pkg/hostman/options"
 	"yunion.io/x/onecloud/pkg/hostman/storageman"
+	"yunion.io/x/onecloud/pkg/hostman/storageman/nbd"
 	"yunion.io/x/onecloud/pkg/hostman/system_service"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/modules"
@@ -221,6 +222,7 @@ func (h *SHostInfo) prepareEnv() error {
 	if err != nil {
 		log.Errorf("Failed to activate nbd device: %s", output)
 	}
+	nbd.Init()
 
 	if !winutils.CheckTool(options.HostOptions.ChntpwPath) {
 		return fmt.Errorf("Failed to find chntpw tool")

pkg/hostman/storageman/disk_base.go

@@ -120,8 +120,8 @@ func (d *SBaseDisk) DeployGuestFs(diskPath string, guestDesc *jsonutils.JSONDict
 
 func (d *SBaseDisk) ResizeFs(diskPath string) error {
 	disk := NewKVMGuestDisk(diskPath)
+	defer disk.Disconnect()
 	if disk.Connect() {
-		defer disk.Disconnect()
 		if err := disk.ResizePartition(); err != nil {
 			return err
 		}

pkg/hostman/storageman/disk_local.go

@@ -276,8 +276,8 @@ func (d *SLocalDisk) CreateRaw(ctx context.Context, sizeMB int, diskFormat, fsFo
 func (d *SLocalDisk) FormatFs(fsFormat, uuid string) {
 	log.Infof("Make disk %s fs %s", uuid, fsFormat)
 	gd := NewKVMGuestDisk(d.GetPath())
+	defer gd.Disconnect()
 	if gd.Connect() {
-		defer gd.Disconnect()
 		if err := gd.MakePartition(fsFormat); err == nil {
 			err = gd.FormatPartition(fsFormat, uuid)
 			if err != nil {

pkg/hostman/storageman/disk_rbd.go

@@ -207,8 +207,8 @@ func (d *SRBDDisk) CreateRaw(ctx context.Context, sizeMb int, diskFromat string,
 func (d *SRBDDisk) FormatFs(fsFormat, uuid string) {
 	log.Infof("Make disk %s fs %s", uuid, fsFormat)
 	gd := NewKVMGuestDisk(d.GetPath())
+	defer gd.Disconnect()
 	if gd.Connect() {
-		defer gd.Disconnect()
 		if err := gd.MakePartition(fsFormat); err == nil {
 			err = gd.FormatPartition(fsFormat, uuid)
 			if err != nil {

pkg/hostman/storageman/diskutils.go

@@ -20,6 +20,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"runtime/debug"
 	"strings"
 	"time"
 
@@ -69,7 +70,7 @@ func (d *SKVMGuestDisk) Connect() bool {
 		return false
 	}
 
-	d.connectionPrecheck()
+	pathType := d.connectionPrecheck()
 
 	var cmd []string
 	if strings.HasPrefix(d.imagePath, "rbd:") || d.getImageFormat() == "raw" {
@@ -109,9 +110,10 @@ func (d *SKVMGuestDisk) Connect() bool {
 		tried += 1
 	}
 
-	if !d.isNonLvmImagePath() {
-		hasLVM, err := d.setupLVMS()
-		if !hasLVM && err == nil {
+	if pathType == LVM_PATH {
+		d.setupLVMS()
+	} else if pathType == PATH_TYPE_UNKNOWN {
+		if hasLVM, err := d.setupLVMS(); !hasLVM && err == nil {
 			d.cacheNonLVMImagePath()
 		}
 	}
@@ -161,7 +163,7 @@ func (d *SKVMGuestDisk) findLVMPartitions(partDev string) string {
 	return findVgname(partDev)
 }
 
-func (d *SKVMGuestDisk) rootBackingFilePath() string {
+func (d *SKVMGuestDisk) rootImagePath() string {
 	if len(d.imageRootBackFilePath) > 0 {
 		return d.imageRootBackFilePath
 	}
@@ -183,29 +185,32 @@ func (d *SKVMGuestDisk) rootBackingFilePath() string {
 }
 
 func (d *SKVMGuestDisk) isNonLvmImagePath() bool {
-	pathType := lvmTool.GetPathType(d.rootBackingFilePath())
+	pathType := lvmTool.GetPathType(d.rootImagePath())
 	return pathType == NON_LVM_PATH
 }
 
 func (d *SKVMGuestDisk) cacheNonLVMImagePath() {
-	lvmTool.CacheNonLvmImagePath(d.rootBackingFilePath())
+	lvmTool.CacheNonLvmImagePath(d.rootImagePath())
 }
 
-func (d *SKVMGuestDisk) connectionPrecheck() {
-	pathType := lvmTool.GetPathType(d.rootBackingFilePath())
-	switch pathType {
-	case LVM_PATH:
-		lvmTool.Wait(d.rootBackingFilePath())
-		lvmTool.Add(d.rootBackingFilePath())
-	case NON_LVM_PATH:
-		return
-	case PATH_NOT_FOUND:
-		lvmTool.Add(d.rootBackingFilePath())
+func (d *SKVMGuestDisk) connectionPrecheck() int {
+	pathType := lvmTool.GetPathType(d.rootImagePath())
+	if pathType == LVM_PATH || pathType == PATH_TYPE_UNKNOWN {
+		lvmTool.Acquire(d.rootImagePath())
 	}
+	return pathType
 }
 
 func (d *SKVMGuestDisk) LvmDisconnectNotify() {
-	lvmTool.Signal(d.rootBackingFilePath())
+	defer func() {
+		if r := recover(); r != nil {
+			log.Errorf("Catch panic on LvmDisconnectNotify %v \n %s", r, debug.Stack())
+		}
+	}()
+	pathType := lvmTool.GetPathType(d.rootImagePath())
+	if pathType != NON_LVM_PATH {
+		lvmTool.Release(d.rootImagePath())
+	}
 }
 
 func (d *SKVMGuestDisk) setupLVMS() (bool, error) {
@@ -247,9 +252,9 @@ func (d *SKVMGuestDisk) PutdownLVMs() {
 
 func (d *SKVMGuestDisk) Disconnect() bool {
 	if len(d.nbdDev) > 0 {
+		defer d.LvmDisconnectNotify()
 		d.PutdownLVMs()
 		_, err := procutils.NewCommand(qemutils.GetQemuNbd(), "-d", d.nbdDev).Run()
-		d.LvmDisconnectNotify()
 		if err != nil {
 			log.Errorln(err.Error())
 			return false

pkg/hostman/storageman/lvmutils.go

@@ -30,105 +30,61 @@ import (
 )
 
 const (
-	PATH_NOT_FOUND = 0
-	LVM_PATH       = 1
-	NON_LVM_PATH   = 2
+	PATH_TYPE_UNKNOWN = 0
+	LVM_PATH          = 1
+	NON_LVM_PATH      = 2
 )
 
-type SLVMImageConnectUniqueTool struct {
-	cond     *sync.Cond
-	refCount int
-}
-
-func NewLVMImageConnectUniqueTool() *SLVMImageConnectUniqueTool {
-	return &SLVMImageConnectUniqueTool{
-		cond: sync.NewCond(&sync.Mutex{}),
-	}
-}
-
-func (t *SLVMImageConnectUniqueTool) AddRef() {
-	t.cond.L.Lock()
-	defer t.cond.L.Unlock()
-	t.refCount++
-}
-
-func (t *SLVMImageConnectUniqueTool) Wait() {
-	t.cond.L.Lock()
-	defer t.cond.L.Unlock()
-	t.refCount++
-	t.cond.Wait()
-}
-
-func (t *SLVMImageConnectUniqueTool) Signal() {
-	t.cond.L.Lock()
-	defer t.cond.L.Unlock()
-	t.refCount--
-	t.cond.Signal()
-}
-
-func (t *SLVMImageConnectUniqueTool) CanBeDestoryed() bool {
-	t.cond.L.Lock()
-	defer t.cond.L.Unlock()
-	return t.refCount == 0
-}
-
 type SLVMImageConnectUniqueToolSet struct {
-	lvms               map[string]*SLVMImageConnectUniqueTool
-	nonLVMImagePathSet map[string]struct{}
-	lock               *sync.Mutex
+	lvms    map[string]*sync.Mutex
+	nonLvms map[string]struct{}
+	lock    *sync.Mutex
 }
 
 func NewLVMImageConnectUniqueToolSet() *SLVMImageConnectUniqueToolSet {
 	return &SLVMImageConnectUniqueToolSet{
-		lvms:               make(map[string]*SLVMImageConnectUniqueTool),
-		nonLVMImagePathSet: make(map[string]struct{}),
-		lock:               new(sync.Mutex),
+		lvms:    make(map[string]*sync.Mutex),
+		nonLvms: make(map[string]struct{}),
+		lock:    new(sync.Mutex),
 	}
 }
 
-func (s *SLVMImageConnectUniqueToolSet) Add(imagePath string) {
-	s.lock.Lock()
-	if _, ok := s.lvms[imagePath]; !ok {
-		s.lvms[imagePath] = NewLVMImageConnectUniqueTool()
-	}
-	s.lock.Unlock()
-
-	s.lvms[imagePath].AddRef()
-}
-
 func (s *SLVMImageConnectUniqueToolSet) CacheNonLvmImagePath(imagePath string) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	s.nonLVMImagePathSet[imagePath] = struct{}{}
+	s.nonLvms[imagePath] = struct{}{}
 }
 
 func (s *SLVMImageConnectUniqueToolSet) GetPathType(imagePath string) int {
 	if _, ok := s.lvms[imagePath]; ok {
 		return LVM_PATH
-	} else if _, ok := s.nonLVMImagePathSet[imagePath]; ok {
+	} else if _, ok := s.nonLvms[imagePath]; ok {
 		return NON_LVM_PATH
 	} else {
-		return PATH_NOT_FOUND
+		return PATH_TYPE_UNKNOWN
 	}
 }
 
-func (s *SLVMImageConnectUniqueToolSet) Wait(imagePath string) {
-	if tool, ok := s.lvms[imagePath]; ok {
-		tool.Wait()
+func (s *SLVMImageConnectUniqueToolSet) Release(imagePath string) {
+	if _, ok := s.lvms[imagePath]; ok {
+		s.lvms[imagePath].Unlock()
 	}
+	s.lock.Lock()
+	if _, ok := s.nonLvms[imagePath]; ok {
+		delete(s.lvms, imagePath)
+	}
+	s.lock.Unlock()
 }
 
-func (s *SLVMImageConnectUniqueToolSet) Signal(imagePath string) {
-	if tool, ok := s.lvms[imagePath]; ok {
-		tool.Signal()
-
-		if tool.CanBeDestoryed() {
-			s.lock.Lock()
-			defer s.lock.Unlock()
-			delete(s.lvms, imagePath)
-		}
+func (s *SLVMImageConnectUniqueToolSet) Acquire(imagePath string) {
+	s.lock.Lock()
+	if _, ok := s.lvms[imagePath]; !ok {
+		s.lvms[imagePath] = new(sync.Mutex)
 	}
+	s.lock.Unlock()
+
+	s.lvms[imagePath].Lock()
 }
 
 type SKVMGuestLVMPartition struct {

pkg/hostman/storageman/nbd/nbdman.go

@@ -30,7 +30,7 @@ type SNBDManager struct {
 
 var nbdManager *SNBDManager
 
-func init() {
+func Init() {
 	nbdManager = NewNBDManager()
 }
 

pkg/hostman/storageman/nbd/nbdman_test.go

@@ -19,6 +19,7 @@ import (
 )
 
 func TestGetNBDManager(t *testing.T) {
+	Init()
 	nbdman := GetNBDManager()
 	t.Logf("Acquire nbd: %s, %v", nbdman.AcquireNbddev(), nbdman.nbdDevs)
 }

pkg/hostman/storageman/storage_local.go

@@ -232,9 +232,8 @@ func (s *SLocalStorage) saveToGlance(ctx context.Context, imageId, imagePath str
 	)
 
 	if err := func() error {
+		defer kvmDisk.Disconnect()
 		if kvmDisk.Connect() {
-			defer kvmDisk.Disconnect()
-
 			var err error
 			func() {
 				if root := kvmDisk.MountKvmRootfs(); root != nil {

pkg/hostman/storageman/storage_rbd.go

@@ -473,9 +473,8 @@ func (s *SRbdStorage) saveToGlance(ctx context.Context, imageId, imagePath strin
 	)
 
 	if err := func() error {
+		defer kvmDisk.Disconnect()
 		if kvmDisk.Connect() {
-			defer kvmDisk.Disconnect()
-
 			if root := kvmDisk.MountKvmRootfs(); root != nil {
 				defer kvmDisk.UmountKvmRootfs(root)