From 6a6806e65b9bfc0048608c2a4d2555c3336a2c4b Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Thu, 6 Dec 2018 13:31:59 +0800 Subject: [PATCH 1/3] server create disk by snapshot, mount snapshot by fusefs --- Gopkg.lock | 1 + Gopkg.toml | 3 + cmd/host-image/main.go | 7 + pkg/appsrv/response.go | 11 + pkg/compute/hostdrivers/kvm.go | 16 ++ pkg/compute/models/disks.go | 101 ++++++-- pkg/compute/models/guests.go | 10 + pkg/compute/models/hosts.go | 2 +- pkg/compute/options/options.go | 2 + pkg/compute/tasks/guest_create_disk_task.go | 10 +- pkg/hostimage/host_image_service.go | 234 ++++++++++++++++++ pkg/hostimage/image.go | 134 ++++++++++ .../x/pkg/util/osprofile/osprofile.go | 2 +- vendor/yunion.io/x/sqlchemy/update.go | 14 +- 14 files changed, 510 insertions(+), 37 deletions(-) create mode 100644 cmd/host-image/main.go create mode 100644 pkg/hostimage/host_image_service.go create mode 100644 pkg/hostimage/image.go diff --git a/Gopkg.lock b/Gopkg.lock index 7c2cfbfbda..6752a09d06 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -1450,6 +1450,7 @@ "github.com/miekg/dns", "github.com/moul/http2curl", "github.com/nelsonken/cos-go-sdk-v5/cos", + "github.com/pierrec/lz4", "github.com/serialx/hashring", "github.com/stretchr/testify/assert", "github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common", diff --git a/Gopkg.toml b/Gopkg.toml index 64685308fd..fcf3971f99 100644 --- a/Gopkg.toml +++ b/Gopkg.toml @@ -24,6 +24,9 @@ # go-tests = true # unused-packages = true +[[constraint]] + name = "github.com/pierrec/lz4" + version = "2.0.7" [[constraint]] name = "github.com/360EntSecGroup-Skylar/excelize" diff --git a/cmd/host-image/main.go b/cmd/host-image/main.go new file mode 100644 index 0000000000..058ac20184 --- /dev/null +++ b/cmd/host-image/main.go @@ -0,0 +1,7 @@ +package main + +import "yunion.io/x/onecloud/pkg/hostimage" + +func main() { + hostimage.StartService() +} diff --git a/pkg/appsrv/response.go b/pkg/appsrv/response.go index f5ce722db6..ac611719af 100644 --- a/pkg/appsrv/response.go +++ b/pkg/appsrv/response.go @@ -5,6 +5,7 @@ import ( "net/http" "fmt" + "yunion.io/x/onecloud/pkg/httperrors" ) @@ -56,6 +57,16 @@ func (w *responseWriterChannel) WriteHeader(status int) { <-w.statusResp } +// implent http.Flusher +func (w *responseWriterChannel) Flush() { + if w.isClosed { + return + } + if f, ok := w.backend.(http.Flusher); ok { + f.Flush() + } +} + func (w *responseWriterChannel) wait(ctx context.Context, workerChan chan *SWorker) interface{} { var err error var worker *SWorker diff --git a/pkg/compute/hostdrivers/kvm.go b/pkg/compute/hostdrivers/kvm.go index 8b6cdde8f4..83e33fc4eb 100644 --- a/pkg/compute/hostdrivers/kvm.go +++ b/pkg/compute/hostdrivers/kvm.go @@ -120,6 +120,22 @@ func (self *SKVMHostDriver) RequestUncacheImage(ctx context.Context, host *model func (self *SKVMHostDriver) RequestAllocateDiskOnStorage(ctx context.Context, host *models.SHost, storage *models.SStorage, disk *models.SDisk, task taskman.ITask, content *jsonutils.JSONDict) error { header := task.GetTaskRequestHeader() + if snapshotId, err := content.GetString("snapshot"); err == nil { + iSnapshot, _ := models.SnapshotManager.FetchById(snapshotId) + snapshot := iSnapshot.(*models.SSnapshot) + snapshotStorage := models.StorageManager.FetchStorageById(snapshot.StorageId) + snapshotHost := snapshotStorage.GetMasterHost() + if options.Options.SnapshotCreateDiskProtocol == "url" { + content.Set("snapshot_url", + jsonutils.NewString(fmt.Sprintf("%s/download/snapshots/%s/%s/%s", + snapshotHost.ManagerUri, snapshotStorage.Id, snapshot.DiskId, snapshot.Id))) + content.Set("snapshot_out_of_chain", jsonutils.NewBool(snapshot.OutOfChain)) + } else if options.Options.SnapshotCreateDiskProtocol == "fuse" { + content.Set("snapshot_url", jsonutils.NewString(fmt.Sprintf("%s/snapshots/%s/%s", + snapshotHost.GetFetchUrl(), snapshot.DiskId, snapshot.Id))) + } + content.Set("protocol", jsonutils.NewString(options.Options.SnapshotCreateDiskProtocol)) + } url := fmt.Sprintf("/disks/%s/create/%s", storage.Id, disk.Id) body := jsonutils.NewDict() diff --git a/pkg/compute/models/disks.go b/pkg/compute/models/disks.go index 1f43ae83e7..f80c298bd5 100644 --- a/pkg/compute/models/disks.go +++ b/pkg/compute/models/disks.go @@ -113,6 +113,15 @@ func (manager *SDiskManager) GetContextManager() []db.IModelManager { return []db.IModelManager{StorageManager} } +func (manager *SDiskManager) FetchDiskById(diskId string) *SDisk { + disk, err := manager.FetchById(diskId) + if err != nil { + log.Errorf("FetchById fail %s", err) + return nil + } + return disk.(*SDisk) +} + func (manager *SDiskManager) ListItemFilter(ctx context.Context, q *sqlchemy.SQuery, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (*sqlchemy.SQuery, error) { queryDict, ok := query.(*jsonutils.JSONDict) if !ok { @@ -1006,6 +1015,10 @@ func totalDiskSize(projectId string, active tristate.TriState, ready tristate.Tr type SDiskConfig struct { ImageId string + + SnapshotId string + DiskType string // sys, data, swap + // ImageDiskFormat string SizeMb int // MB Fs string // file system @@ -1062,28 +1075,16 @@ func parseDiskInfo(ctx context.Context, userCred mcclient.TokenCredential, info diskConfig.SizeMb = -1 } else if utils.IsInStringArray(p, STORAGE_TYPES) { diskConfig.Backend = p + } else if strings.HasPrefix(p, "snapshot-") { + // HACK: use snapshot creat disk format snapshot-id + // example: snapshot-3140cecb-ccc4-4865-abae-3a5ba8c69d9b + log.Errorln("The snapshot XXXXXXX Create disk", p[len("snapshot-"):]) + if err := fillDiskConfigBySnapshot(userCred, &diskConfig, p[len("snapshot-"):]); err != nil { + return nil, err + } } else if len(p) > 0 { - if userCred == nil { - diskConfig.ImageId = p - } else { - image, err := CachedimageManager.getImageInfo(ctx, userCred, p, false) - if err != nil { - log.Errorf("getImageInfo fail %s", err) - return nil, err - } - if image.Status != IMAGE_STATUS_ACTIVE { - return nil, httperrors.NewInvalidStatusError("Image status is not active") - } - diskConfig.ImageId = image.Id - diskConfig.ImageProperties = image.Properties - if len(diskConfig.Format) == 0 { - diskConfig.Format = image.DiskFormat - } - // diskConfig.ImageDiskFormat = image.DiskFormat - CachedimageManager.ImageAddRefCount(image.Id) - if diskConfig.SizeMb == 0 { - diskConfig.SizeMb = image.MinDisk // MB - } + if err := fillDiskConfigByImage(ctx, userCred, &diskConfig, p); err != nil { + return nil, err } } } @@ -1095,6 +1096,60 @@ func parseDiskInfo(ctx context.Context, userCred mcclient.TokenCredential, info return &diskConfig, nil } +func fillDiskConfigBySnapshot(userCred mcclient.TokenCredential, diskConfig *SDiskConfig, snapshotId string) error { + iSnapshot, err := SnapshotManager.FetchByIdOrName(userCred, snapshotId) + if err != nil { + if err == sql.ErrNoRows { + return httperrors.NewNotFoundError("Snapshot %s not found", snapshotId) + } + return err + } + var snapshot = iSnapshot.(*SSnapshot) + if storage := StorageManager.FetchStorageById(snapshot.StorageId); storage == nil { + return httperrors.NewBadRequestError("Snapshot %s storage %s not found, is public cloud?", + snapshotId, snapshot.StorageId) + } else { + if disk := DiskManager.FetchDiskById(snapshot.DiskId); disk != nil { + diskConfig.Fs = disk.FsFormat + if len(diskConfig.Format) == 0 { + diskConfig.Format = disk.DiskFormat + } + } + diskConfig.SnapshotId = snapshot.Id + diskConfig.DiskType = snapshot.DiskType + diskConfig.SizeMb = snapshot.Size + diskConfig.Backend = storage.StorageType + } + return nil +} + +func fillDiskConfigByImage(ctx context.Context, userCred mcclient.TokenCredential, + diskConfig *SDiskConfig, imageId string) error { + if userCred == nil { + diskConfig.ImageId = imageId + } else { + image, err := CachedimageManager.getImageInfo(ctx, userCred, imageId, false) + if err != nil { + log.Errorf("getImageInfo fail %s", err) + return err + } + if image.Status != IMAGE_STATUS_ACTIVE { + return httperrors.NewInvalidStatusError("Image status is not active") + } + diskConfig.ImageId = image.Id + diskConfig.ImageProperties = image.Properties + if len(diskConfig.Format) == 0 { + diskConfig.Format = image.DiskFormat + } + // diskConfig.ImageDiskFormat = image.DiskFormat + CachedimageManager.ImageAddRefCount(image.Id) + if diskConfig.SizeMb == 0 { + diskConfig.SizeMb = image.MinDisk // MB + } + } + return nil +} + func parseIsoInfo(ctx context.Context, userCred mcclient.TokenCredential, info string) (string, error) { image, err := CachedimageManager.getImageInfo(ctx, userCred, info, false) if err != nil { @@ -1111,6 +1166,10 @@ func (self *SDisk) fetchDiskInfo(diskConfig *SDiskConfig) { if len(diskConfig.ImageId) > 0 { self.TemplateId = diskConfig.ImageId self.DiskType = DISK_TYPE_SYS + } else if len(diskConfig.SnapshotId) > 0 { + // XXX: HACK reuse template id as snapshot id + // self.TemplateId = "snapshot-" + diskConfig.SnapshotId + self.DiskType = diskConfig.DiskType } if len(diskConfig.Fs) > 0 { self.FsFormat = diskConfig.Fs diff --git a/pkg/compute/models/guests.go b/pkg/compute/models/guests.go index 352012aa0b..de39aae35e 100644 --- a/pkg/compute/models/guests.go +++ b/pkg/compute/models/guests.go @@ -689,6 +689,9 @@ func (manager *SGuestManager) ValidateCreateData(ctx context.Context, userCred m if err != nil { return nil, httperrors.NewInputParameterError("Invalid root image: %s", err) } + if len(diskConfig.ImageId) > 0 && diskConfig.DiskType != DISK_TYPE_SYS { + return nil, httperrors.NewBadRequestError("Snapshot error: disk index 0 but disk type is %s", diskConfig.DiskType) + } if len(diskConfig.Backend) == 0 { diskConfig.Backend = STORAGE_LOCAL @@ -777,6 +780,9 @@ func (manager *SGuestManager) ValidateCreateData(ctx context.Context, userCred m if err != nil { return nil, httperrors.NewInputParameterError("parse disk description error %s", err) } + if diskConfig.DiskType == DISK_TYPE_SYS { + return nil, httperrors.NewBadRequestError("Snapshot error: disk index %d > 0 but disk type is %s", idx, DISK_TYPE_SYS) + } if len(diskConfig.Backend) == 0 { diskConfig.Backend = rootStorageType } @@ -2344,6 +2350,10 @@ func (self *SGuest) CreateDisksOnHost(ctx context.Context, userCred mcclient.Tok return err } data.Add(jsonutils.NewString(disk.Id), fmt.Sprintf("disk.%d.id", idx)) + log.Errorln("CreateDisksOnHost XXXXXXXXXX", diskConfig.SnapshotId) + if len(diskConfig.SnapshotId) > 0 { + data.Add(jsonutils.NewString(diskConfig.SnapshotId), fmt.Sprintf("disk.%d.snapshot", idx)) + } } return nil } diff --git a/pkg/compute/models/hosts.go b/pkg/compute/models/hosts.go index 9fa926df40..8dfdfe95b4 100644 --- a/pkg/compute/models/hosts.go +++ b/pkg/compute/models/hosts.go @@ -629,7 +629,7 @@ func (self *SHost) GetFetchUrl() string { port = 80 } } - return fmt.Sprintf("%s://%s:%d", managerUrl.Scheme, managerUrl.Host, port+40000) + return fmt.Sprintf("%s://%s:%d", managerUrl.Scheme, strings.Split(managerUrl.Host, ":")[0], port+40000) } func (self *SHost) GetAttachedStorages(storageType string) []SStorage { diff --git a/pkg/compute/options/options.go b/pkg/compute/options/options.go index 4f35356546..3e369f6527 100644 --- a/pkg/compute/options/options.go +++ b/pkg/compute/options/options.go @@ -74,6 +74,8 @@ type ComputeOptions struct { NfsDefaultImageCacheDir string `default:"image_cache"` + SnapshotCreateDiskProtocol string `help:"Snapshot create disk protocol" choices:"url|fuse" default:"fuse"` + cloudcommon.DBOptions } diff --git a/pkg/compute/tasks/guest_create_disk_task.go b/pkg/compute/tasks/guest_create_disk_task.go index 2305667475..264689b650 100644 --- a/pkg/compute/tasks/guest_create_disk_task.go +++ b/pkg/compute/tasks/guest_create_disk_task.go @@ -3,10 +3,10 @@ package tasks import ( "context" "fmt" + "time" "yunion.io/x/jsonutils" - "time" "yunion.io/x/log" "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" @@ -66,11 +66,9 @@ func (self *KVMGuestCreateDiskTask) OnKvmDiskPrepared(ctx context.Context, obj d } disk := iDisk.(*models.SDisk) if disk.Status == models.DISK_INIT { - snapInfo, err := self.Params.GetString(fmt.Sprintf("disk.%d.snapshot", diskIndex)) - if err != nil { - snapInfo = "" - } - err = disk.StartDiskCreateTask(ctx, self.UserCred, false, snapInfo, self.GetTaskId()) + snapshotId, _ := self.Params.GetString(fmt.Sprintf("disk.%d.snapshot", diskIndex)) + log.Errorln("XXXXXXXXXXXXXX", snapshotId) + err = disk.StartDiskCreateTask(ctx, self.UserCred, false, snapshotId, self.GetTaskId()) if err != nil { self.SetStageFailed(ctx, err.Error()) return diff --git a/pkg/hostimage/host_image_service.go b/pkg/hostimage/host_image_service.go new file mode 100644 index 0000000000..fb8a2846f6 --- /dev/null +++ b/pkg/hostimage/host_image_service.go @@ -0,0 +1,234 @@ +package hostimage + +import ( + "context" + "fmt" + "net/http" + "os" + "path" + "strconv" + "strings" + "time" + + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/appctx" + "yunion.io/x/onecloud/pkg/appsrv" + "yunion.io/x/onecloud/pkg/cloudcommon" + "yunion.io/x/onecloud/pkg/cloudcommon/consts" + "yunion.io/x/onecloud/pkg/httperrors" + "yunion.io/x/onecloud/pkg/mcclient/auth" + + "github.com/pierrec/lz4" +) + +type SHostImageOptions struct { + cloudcommon.Options + LocalImagePath []string `help:"Local Image Paths"` + SnapshotDirSuffix string `help:"Snapshot dir name equal diskId concat snapshot dir suffix" default:"_snap"` +} + +var HostImageOptions SHostImageOptions + +func StartService() { + consts.SetServiceType("host-image") + cloudcommon.ParseOptions(&HostImageOptions, &HostImageOptions.Options, os.Args, "host.conf") + HostImageOptions.Port += 40000 + cloudcommon.InitAuth(&HostImageOptions.Options, func() { + log.Infof("Auth complete!!") + }) + app := cloudcommon.InitApp(&HostImageOptions.Options) + initHandlers(app, "") + cloudcommon.ServeForever(app, &HostImageOptions.Options) +} + +func initHandlers(app *appsrv.Application, prefix string) { + app.AddHandler("GET", fmt.Sprintf("%s/disks/", prefix), auth.Authenticate(getImage)) + app.AddHandler("GET", fmt.Sprintf("%s/snapshots//", prefix), auth.Authenticate(getImage)) + app.AddHandler("HEAD", fmt.Sprintf("%s/disks/", prefix), auth.Authenticate(getImageMeta)) + app.AddHandler("HEAD", fmt.Sprintf("%s/snapshots//", prefix), auth.Authenticate(getImageMeta)) +} + +func getDiskPath(diskId string) string { + for _, imagePath := range HostImageOptions.LocalImagePath { + diskPath := path.Join(imagePath, diskId) + if _, err := os.Stat(diskPath); !os.IsNotExist(err) { + return diskPath + } + } + return "" +} + +func getSnapshotPath(diskId, snapshotId string) string { + for _, imagePath := range HostImageOptions.LocalImagePath { + diskPath := path.Join(imagePath, "snapshots", + diskId+HostImageOptions.SnapshotDirSuffix, snapshotId) + if _, err := os.Stat(diskPath); !os.IsNotExist(err) { + return diskPath + } + } + return "" +} + +func inputCheck(ctx context.Context) (string, error) { + var userCred = auth.FetchUserCredential(ctx, nil) + if !userCred.HasSystemAdminPrivelege() { + return "", httperrors.NewForbiddenError("System admin only") + + } + + var params = appctx.AppContextParams(ctx) + var sid = params[""] + var imagePath string + if diskId, ok := params[""]; ok { + imagePath = getSnapshotPath(diskId, sid) + } else { + imagePath = getDiskPath(sid) + } + if len(imagePath) == 0 { + return "", httperrors.NewNotFoundError("Disk not found") + } + return imagePath, nil +} + +func parseRange(reqRange string) (int64, int64, error) { + if !strings.HasPrefix(reqRange, "bytes=") { + return 0, 0, httperrors.NewInputParameterError("Invalid range header") + } + reqRange = reqRange[len("bytes="):] + ranges := strings.Split(reqRange, "-") + if len(ranges) != 2 { + return 0, 0, httperrors.NewInputParameterError("Invalid range header") + } + startPos, err := strconv.ParseInt(ranges[0], 10, 0) + if err != nil { + return 0, 0, httperrors.NewInputParameterError("Invalid range header") + } + endPos, err := strconv.ParseInt(ranges[1], 10, 0) + if err != nil { + return 0, 0, httperrors.NewInputParameterError("Invalid range header") + } + return startPos, endPos, nil +} + +func getImage(ctx context.Context, w http.ResponseWriter, r *http.Request) { + imagePath, err := inputCheck(ctx) + if err != nil { + httperrors.GeneralServerError(w, err) + return + } + + var f IImage + var startPos, endPos int64 + var rateLimit int64 = -1 + + if r.Header.Get("X-Read-File") == "true" { + f = &SFile{} + } else { + f = &SQcow2Image{} + } + if err = f.Open(imagePath, true); err != nil { + log.Errorf("Open image error: %s", err) + httperrors.GeneralServerError(w, err) + return + } + defer f.Close() // Remenber close fd + + endPos = f.Length() - 1 + reqRange := r.Header.Get("Range") + if len(reqRange) > 0 { + startPos, endPos, err = parseRange(reqRange) + if err != nil { + log.Errorf("Parse range error: %s", err) + httperrors.GeneralServerError(w, err) + return + } + } + + strRateLimit := r.Header.Get("X-Rate-Limit-Mbps") + if len(strRateLimit) > 0 { + rateLimit, err = strconv.ParseInt(strRateLimit, 10, 0) + if err != nil { + log.Errorf("Parse ratelimit error: %s", err) + httperrors.InvalidInputError(w, "Invaild rate limit header") + return + } + } + + streamHeader(w, f, startPos, endPos) + startStream(w, f, startPos, endPos, rateLimit) +} + +func streamHeader(w http.ResponseWriter, f IImage, startPos, endPos int64) { + var statusCode = http.StatusOK + w.Header().Set("Content-Type", "application/octet-stream") + if startPos > 0 || endPos < f.Length()-1 { + statusCode = http.StatusPartialContent + w.Header().Set("Content-Range", + fmt.Sprintf("bytes %d-%d/%d", startPos, endPos, f.Length())) + } + w.WriteHeader(statusCode) +} + +func startStream(w http.ResponseWriter, f IImage, startPos, endPos, rateLimit int64) { + var CHUNK_SIZE int64 = 4 * 1024 + var readSize int64 = CHUNK_SIZE + var sendBytes int64 + var lz4Writer = lz4.NewWriter(w) + var startTime = time.Now() + + for startPos < endPos { + if endPos-startPos < CHUNK_SIZE { + readSize = endPos - startPos + 1 + } + buf, total := f.Read(startPos, readSize) + if total < 0 { + log.Errorf("Read image error: %d", total) + goto fail + } + startPos += readSize + wSize, err := lz4Writer.Write(buf) + if err != nil { + log.Errorf("lz4Write error: %s", err) + goto fail + } + sendBytes += int64(wSize) + if rateLimit > 0 { + tmDelta := time.Now().Sub(startTime) + tms := tmDelta.Seconds() + vtmDelta := float64(sendBytes*8) / float64(1024.0*1024.0*rateLimit) + if vtmDelta > tms { + time.Sleep(time.Duration(vtmDelta - tms)) + } + } + } + +fail: + if err := lz4Writer.Close(); err != nil { + log.Errorf("lz4 Close error: %s", err) + } +} + +func getImageMeta(ctx context.Context, w http.ResponseWriter, r *http.Request) { + imagePath, err := inputCheck(ctx) + if err != nil { + httperrors.GeneralServerError(w, err) + return + } + + var f IImage + if r.Header.Get("X-Read-File") == "true" { + f = &SFile{} + } else { + f = &SQcow2Image{} + } + if err = f.Open(imagePath, true); err != nil { + httperrors.GeneralServerError(w, err) + return + } + defer f.Close() // Remenber close fd + + w.Header().Set("Content-Length", fmt.Sprintf("%d", f.Length())) + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Accept-Ranges", "bytes") + w.WriteHeader(200) +} diff --git a/pkg/hostimage/image.go b/pkg/hostimage/image.go new file mode 100644 index 0000000000..c5631f8b91 --- /dev/null +++ b/pkg/hostimage/image.go @@ -0,0 +1,134 @@ +package hostimage + +/* +#cgo pkg-config: glib-2.0 zlib +#cgo CFLAGS: -I/home/yunion/rpmbuild/SOURCES/qemu/src -I/home/yunion/rpmbuild/SOURCES/qemu/src/include +#cgo LDFLAGS: -laio -lqemuio -lpthread -L /home/yunion/rpmbuild/SOURCES/qemu/src + +#include "libqemuio.h" +#include "qemu/osdep.h" +*/ +import "C" + +import ( + "fmt" + "io" + "os" + "unsafe" +) + +func init() { + C.qemuio_init() +} + +func ReadQcow2(qemuioBlk *C.struct_QemuioBlk, offset int64, count int64) ([]byte, int64) { + if qemuioBlk == nil || offset < 0 || count < 0 { + return nil, -1 + } + b := make([]byte, count) + var total = C.int64_t(0) + ret := C.read_qcow2(qemuioBlk, unsafe.Pointer(&b[0]), C.int64_t(offset), C.int64_t(count), &total) + if ret < 0 { + return nil, int64(ret) + } else { + return b, int64(total) + } +} + +func OpenQcow2(imagePath string, readonly bool) *C.struct_QemuioBlk { + return C.open_qcow2(C.CString(imagePath), C.bool(readonly)) +} + +func Qcow2GetLenth(qemuioBlk *C.struct_QemuioBlk) int64 { + return int64(C.qcow2_get_length(qemuioBlk)) +} + +func CloseQcow2(qemuioBlk *C.struct_QemuioBlk) { + C.close_qcow2(qemuioBlk) +} + +type IImage interface { + // Open image file and its backing file (if have) + Open(imagePath string, readonly bool) error + + // Close may not really close image file handle, just reudce ref count + Close() + + // If return number < 0 indicate read failed + Read(offset, count int64) ([]byte, int64) + + // Get image file length, not file actual length, it's image virtual size + Length() int64 +} + +type SQcow2Image struct { + fd *C.struct_QemuioBlk +} + +func (img *SQcow2Image) Open(imagePath string, readonly bool) error { + fd := OpenQcow2(imagePath, readonly) + if fd == nil { + return fmt.Errorf("Open image %s failed", imagePath) + } else { + img.fd = fd + return nil + } +} + +func (img *SQcow2Image) Read(offset, count int64) ([]byte, int64) { + return ReadQcow2(img.fd, offset, count) +} + +func (img *SQcow2Image) Close() { + CloseQcow2(img.fd) +} + +func (img *SQcow2Image) Length() int64 { + return Qcow2GetLenth(img.fd) +} + +type SFile struct { + fd *os.File +} + +func (f *SFile) Open(imagePath string, readonly bool) error { + var mode = os.O_RDWR + if readonly { + mode = os.O_RDONLY + } + fd, err := os.OpenFile(imagePath, mode, 0644) + if err != nil { + return err + } else { + f.fd = fd + return nil + } +} + +func (f *SFile) Read(offset, count int64) ([]byte, int64) { + buf := make([]byte, count) + var readCount int64 = 0 + for readCount < count { + cnt, err := f.fd.Read(buf[readCount:]) + readCount += int64(cnt) + if err == io.EOF { + return buf[0:readCount], readCount + } + if err != nil { + return nil, -1 + } + } + return buf, readCount +} + +func (f *SFile) Close() { + f.fd.Close() +} + +func (f *SFile) Length() int64 { + stat, e := f.fd.Stat() + if e != nil { + return -1 + } + return stat.Size() +} diff --git a/vendor/yunion.io/x/pkg/util/osprofile/osprofile.go b/vendor/yunion.io/x/pkg/util/osprofile/osprofile.go index 09cb355365..7d7c92d87f 100644 --- a/vendor/yunion.io/x/pkg/util/osprofile/osprofile.go +++ b/vendor/yunion.io/x/pkg/util/osprofile/osprofile.go @@ -106,7 +106,7 @@ func GetOSProfileFromImageProperties(imgProp map[string]string, hypervisor strin } var imgHypers []string imgHyperStr, ok := imgProp["hypervisor"] - if ok && len(imgHyperStr) > 0 { + if ok { imgHypers = strings.Split(imgHyperStr, ",") } else { imgHypers = []string{} diff --git a/vendor/yunion.io/x/sqlchemy/update.go b/vendor/yunion.io/x/sqlchemy/update.go index 3a6512d2fb..4a7e253e9f 100644 --- a/vendor/yunion.io/x/sqlchemy/update.go +++ b/vendor/yunion.io/x/sqlchemy/update.go @@ -85,14 +85,12 @@ func (us *SUpdateSession) saveUpdate(dt interface{}) (map[string]SUpdateDiff, er k := c.Name() of := ofields[k] nf := fields[k] - if !gotypes.IsNil(of) { - if c.IsPrimary() && !c.IsZero(of) { // skip update primary key - primaries[k] = of - continue - } else if c.IsKeyIndex() && !c.IsZero(of) { - keyIndexes[k] = of - continue - } + if c.IsPrimary() && !c.IsZero(of) { // skip update primary key + primaries[k] = of + continue + } else if c.IsKeyIndex() && !c.IsZero(of) { + keyIndexes[k] = of + continue } nc, ok := c.(*SIntegerColumn) if ok && nc.IsAutoVersion { From 86b3b38f41362b81994aee7c108fa13fc37ca4b6 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Tue, 11 Dec 2018 20:31:36 +0800 Subject: [PATCH 2/3] add snapshot reference count, fix code --- pkg/appsrv/response.go | 3 +-- pkg/compute/models/disks.go | 9 ++++++--- pkg/compute/models/guest_actions.go | 17 ++++++++++++++++- pkg/compute/models/guestdisks.go | 6 ++++++ pkg/compute/models/guests.go | 1 - pkg/compute/models/snapshots.go | 17 +++++++++++++++++ pkg/compute/tasks/disk_delete_task.go | 3 +++ pkg/compute/tasks/guest_create_disk_task.go | 1 - pkg/hostimage/host_image_service.go | 1 - 9 files changed, 49 insertions(+), 9 deletions(-) diff --git a/pkg/appsrv/response.go b/pkg/appsrv/response.go index ac611719af..89e64a49ce 100644 --- a/pkg/appsrv/response.go +++ b/pkg/appsrv/response.go @@ -2,9 +2,8 @@ package appsrv import ( "context" - "net/http" - "fmt" + "net/http" "yunion.io/x/onecloud/pkg/httperrors" ) diff --git a/pkg/compute/models/disks.go b/pkg/compute/models/disks.go index f80c298bd5..a95aae1ff8 100644 --- a/pkg/compute/models/disks.go +++ b/pkg/compute/models/disks.go @@ -100,6 +100,9 @@ type SDisk struct { // # backing template id and type TemplateId string `width:"256" charset:"ascii" nullable:"true" list:"user"` // Column(VARCHAR(ID_LENGTH, charset='ascii'), nullable=True) + // backing snapshot id + SnapshotId string `width:"256" charset:"ascii" nullable:"true" list:"user"` + // # file system FsFormat string `width:"32" charset:"ascii" nullable:"true" list:"user"` // Column(VARCHAR(32, charset='ascii'), nullable=True) // # disk type, OS, SWAP, DAT @@ -446,6 +449,8 @@ func (self *SDisk) StartAllocate(ctx context.Context, host *SHost, storage *SSto content.Add(jsonutils.NewInt(int64(self.DiskSize)), "size") if len(snapshot) > 0 { content.Add(jsonutils.NewString(snapshot), "snapshot") + SnapshotManager.AddRefCount(self.SnapshotId, 1) + self.SetMetadata(ctx, "merge_snapshot", jsonutils.JSONTrue, userCred) } else if len(templateId) > 0 { content.Add(jsonutils.NewString(templateId), "image_id") } @@ -1078,7 +1083,6 @@ func parseDiskInfo(ctx context.Context, userCred mcclient.TokenCredential, info } else if strings.HasPrefix(p, "snapshot-") { // HACK: use snapshot creat disk format snapshot-id // example: snapshot-3140cecb-ccc4-4865-abae-3a5ba8c69d9b - log.Errorln("The snapshot XXXXXXX Create disk", p[len("snapshot-"):]) if err := fillDiskConfigBySnapshot(userCred, &diskConfig, p[len("snapshot-"):]); err != nil { return nil, err } @@ -1167,8 +1171,7 @@ func (self *SDisk) fetchDiskInfo(diskConfig *SDiskConfig) { self.TemplateId = diskConfig.ImageId self.DiskType = DISK_TYPE_SYS } else if len(diskConfig.SnapshotId) > 0 { - // XXX: HACK reuse template id as snapshot id - // self.TemplateId = "snapshot-" + diskConfig.SnapshotId + self.SnapshotId = diskConfig.SnapshotId self.DiskType = diskConfig.DiskType } if len(diskConfig.Fs) > 0 { diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index 19f0397e3e..dac61120c5 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -8,6 +8,7 @@ import ( "net/http" "strconv" "strings" + "time" "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/cloudcommon/db/lockman" @@ -22,7 +23,6 @@ import ( "yunion.io/x/onecloud/pkg/util/logclient" "yunion.io/x/onecloud/pkg/util/seclib2" - "time" "yunion.io/x/jsonutils" "yunion.io/x/log" "yunion.io/x/onecloud/pkg/util/billing" @@ -2112,3 +2112,18 @@ func (self *SGuest) doSaveRenewInfo(userCred mcclient.TokenCredential, bc *billi db.OpsLog.LogEvent(self, db.ACT_RENEW, self.GetShortDesc(), userCred) return nil } + +func (self *SGuest) AllowPerformStreamDisksComplete(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool { + return db.IsAdminAllowPerform(userCred, self, "stream-disks-complete") +} + +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 { + SnapshotManager.AddRefCount(d.SnapshotId, -1) + d.SetMetadata(ctx, "merge_snapshot", jsonutils.JSONFalse, userCred) + } + } + return nil, nil +} diff --git a/pkg/compute/models/guestdisks.go b/pkg/compute/models/guestdisks.go index ab86de0496..e6000e0917 100644 --- a/pkg/compute/models/guestdisks.go +++ b/pkg/compute/models/guestdisks.go @@ -166,6 +166,12 @@ func (self *SGuestdisk) GetJsonDescAtHost(host *SHost) jsonutils.JSONObject { if len(tid) > 0 { desc.Add(jsonutils.NewString(tid), "template_id") } + if len(disk.SnapshotId) > 0 { + needMerge := disk.GetMetadata("merge_snapshot", nil) + if needMerge == "true" { + desc.Set("merge_snapshot", jsonutils.JSONTrue) + } + } fs := disk.GetFsFormat() if len(fs) > 0 { desc.Add(jsonutils.NewString(fs), "fs") diff --git a/pkg/compute/models/guests.go b/pkg/compute/models/guests.go index de39aae35e..1420bcc813 100644 --- a/pkg/compute/models/guests.go +++ b/pkg/compute/models/guests.go @@ -2350,7 +2350,6 @@ func (self *SGuest) CreateDisksOnHost(ctx context.Context, userCred mcclient.Tok return err } data.Add(jsonutils.NewString(disk.Id), fmt.Sprintf("disk.%d.id", idx)) - log.Errorln("CreateDisksOnHost XXXXXXXXXX", diskConfig.SnapshotId) if len(diskConfig.SnapshotId) > 0 { data.Add(jsonutils.NewString(diskConfig.SnapshotId), fmt.Sprintf("disk.%d.snapshot", idx)) } diff --git a/pkg/compute/models/snapshots.go b/pkg/compute/models/snapshots.go index ed9c81a6c4..2a68fdc766 100644 --- a/pkg/compute/models/snapshots.go +++ b/pkg/compute/models/snapshots.go @@ -49,6 +49,9 @@ type SSnapshot struct { FakeDeleted bool `nullable:"false" default:"false" index:"true"` DiskType string `width:"32" charset:"ascii" nullable:"true" list:"user"` + // create disk from snapshot, snapshot as disk backing file + RefCount int `nullable:"false" default:"0" list:"user"` + CloudregionId string `width:"36" charset:"ascii" nullable:"true" list:"user"` } @@ -246,6 +249,17 @@ func (self *SSnapshot) GetHost() *SHost { return storage.GetMasterHost() } +func (self *SSnapshotManager) AddRefCount(snapshotId string, count int) { + iSnapshot, _ := self.FetchById(snapshotId) + snapshot := iSnapshot.(*SSnapshot) + if snapshot != nil { + self.TableSpec().Update(snapshot, func() error { + snapshot.RefCount += count + return nil + }) + } +} + func (self *SSnapshotManager) GetDiskSnapshotsByCreate(diskId, createdBy string) []SSnapshot { dest := make([]SSnapshot, 0) q := self.Query().SubQuery() @@ -343,6 +357,9 @@ func (self *SSnapshot) StartSnapshotDeleteTask(ctx context.Context, userCred mcc } func (self *SSnapshot) ValidateDeleteCondition(ctx context.Context) error { + if self.RefCount > 0 { + return fmt.Errorf("Snapshot reference(by disk) count > 0, can not delete") + } return nil } diff --git a/pkg/compute/tasks/disk_delete_task.go b/pkg/compute/tasks/disk_delete_task.go index 3eac8a0371..ea433d2204 100644 --- a/pkg/compute/tasks/disk_delete_task.go +++ b/pkg/compute/tasks/disk_delete_task.go @@ -93,6 +93,9 @@ func (self *DiskDeleteTask) OnGuestDiskDeleteComplete(ctx context.Context, obj d disk := obj.(*models.SDisk) self.CleanHostSchedCache(disk) db.OpsLog.LogEvent(disk, db.ACT_DELOCATE, disk.GetShortDesc(), self.UserCred) + if len(disk.SnapshotId) > 0 && disk.GetMetadata("merge_snapshot", nil) == "true" { + models.SnapshotManager.AddRefCount(disk.SnapshotId, -1) + } disk.RealDelete(ctx, self.UserCred) self.SetStageComplete(ctx, nil) } diff --git a/pkg/compute/tasks/guest_create_disk_task.go b/pkg/compute/tasks/guest_create_disk_task.go index 264689b650..2b9181df40 100644 --- a/pkg/compute/tasks/guest_create_disk_task.go +++ b/pkg/compute/tasks/guest_create_disk_task.go @@ -67,7 +67,6 @@ func (self *KVMGuestCreateDiskTask) OnKvmDiskPrepared(ctx context.Context, obj d disk := iDisk.(*models.SDisk) if disk.Status == models.DISK_INIT { snapshotId, _ := self.Params.GetString(fmt.Sprintf("disk.%d.snapshot", diskIndex)) - log.Errorln("XXXXXXXXXXXXXX", snapshotId) err = disk.StartDiskCreateTask(ctx, self.UserCred, false, snapshotId, self.GetTaskId()) if err != nil { self.SetStageFailed(ctx, err.Error()) diff --git a/pkg/hostimage/host_image_service.go b/pkg/hostimage/host_image_service.go index fb8a2846f6..1b04b596be 100644 --- a/pkg/hostimage/host_image_service.go +++ b/pkg/hostimage/host_image_service.go @@ -73,7 +73,6 @@ func inputCheck(ctx context.Context) (string, error) { var userCred = auth.FetchUserCredential(ctx, nil) if !userCred.HasSystemAdminPrivelege() { return "", httperrors.NewForbiddenError("System admin only") - } var params = appctx.AppContextParams(ctx) From ad2de3384697067d40a68e06e7870c2a9e9612b3 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Wed, 12 Dec 2018 11:38:09 +0800 Subject: [PATCH 3/3] fix code --- Gopkg.lock | 4 ++-- pkg/compute/models/guests.go | 4 ++-- pkg/compute/models/snapshots.go | 9 ++++++--- .../x/pkg/util/osprofile/osprofile.go | 2 +- .../yunion.io/x/pkg/util/secrules/secrules.go | 19 +++++-------------- vendor/yunion.io/x/sqlchemy/update.go | 14 ++++++++------ 6 files changed, 24 insertions(+), 28 deletions(-) diff --git a/Gopkg.lock b/Gopkg.lock index 6752a09d06..38b16f9f6d 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -1368,11 +1368,11 @@ [[projects]] branch = "master" - digest = "1:2fdf064ae928c1b311e67ec51e856ce655abc39e8eacf5e6b3f7bd0417fac0a6" + digest = "1:04973e1902449b00dd7f3a9ad0b2b892c637cc8bc2e0dc569a82026fe1c4a4b3" name = "yunion.io/x/sqlchemy" packages = ["."] pruneopts = "UT" - revision = "e22221d5efcc667e68b0fdeed19958e788c3ad63" + revision = "998e91b54b0b9441f21dc9c09036f875a02ef8c5" [[projects]] branch = "master" diff --git a/pkg/compute/models/guests.go b/pkg/compute/models/guests.go index 1420bcc813..784bc461b9 100644 --- a/pkg/compute/models/guests.go +++ b/pkg/compute/models/guests.go @@ -689,7 +689,7 @@ func (manager *SGuestManager) ValidateCreateData(ctx context.Context, userCred m if err != nil { return nil, httperrors.NewInputParameterError("Invalid root image: %s", err) } - if len(diskConfig.ImageId) > 0 && diskConfig.DiskType != DISK_TYPE_SYS { + if len(diskConfig.SnapshotId) > 0 && diskConfig.DiskType != DISK_TYPE_SYS { return nil, httperrors.NewBadRequestError("Snapshot error: disk index 0 but disk type is %s", diskConfig.DiskType) } @@ -781,7 +781,7 @@ func (manager *SGuestManager) ValidateCreateData(ctx context.Context, userCred m return nil, httperrors.NewInputParameterError("parse disk description error %s", err) } if diskConfig.DiskType == DISK_TYPE_SYS { - return nil, httperrors.NewBadRequestError("Snapshot error: disk index %d > 0 but disk type is %s", idx, DISK_TYPE_SYS) + return nil, httperrors.NewBadRequestError("Snapshot error: disk index %d > 0 but disk type is %s", i+1, DISK_TYPE_SYS) } if len(diskConfig.Backend) == 0 { diskConfig.Backend = rootStorageType diff --git a/pkg/compute/models/snapshots.go b/pkg/compute/models/snapshots.go index 2a68fdc766..a4aa643515 100644 --- a/pkg/compute/models/snapshots.go +++ b/pkg/compute/models/snapshots.go @@ -251,12 +251,15 @@ func (self *SSnapshot) GetHost() *SHost { func (self *SSnapshotManager) AddRefCount(snapshotId string, count int) { iSnapshot, _ := self.FetchById(snapshotId) - snapshot := iSnapshot.(*SSnapshot) - if snapshot != nil { - self.TableSpec().Update(snapshot, func() error { + if iSnapshot != nil { + snapshot := iSnapshot.(*SSnapshot) + _, err := self.TableSpec().Update(snapshot, func() error { snapshot.RefCount += count return nil }) + if err != nil { + log.Errorf("Snapshot add refence count error: %s", err) + } } } diff --git a/vendor/yunion.io/x/pkg/util/osprofile/osprofile.go b/vendor/yunion.io/x/pkg/util/osprofile/osprofile.go index 7d7c92d87f..09cb355365 100644 --- a/vendor/yunion.io/x/pkg/util/osprofile/osprofile.go +++ b/vendor/yunion.io/x/pkg/util/osprofile/osprofile.go @@ -106,7 +106,7 @@ func GetOSProfileFromImageProperties(imgProp map[string]string, hypervisor strin } var imgHypers []string imgHyperStr, ok := imgProp["hypervisor"] - if ok { + if ok && len(imgHyperStr) > 0 { imgHypers = strings.Split(imgHyperStr, ",") } else { imgHypers = []string{} diff --git a/vendor/yunion.io/x/pkg/util/secrules/secrules.go b/vendor/yunion.io/x/pkg/util/secrules/secrules.go index 037bb62d45..6437ff1ae9 100644 --- a/vendor/yunion.io/x/pkg/util/secrules/secrules.go +++ b/vendor/yunion.io/x/pkg/util/secrules/secrules.go @@ -127,21 +127,12 @@ func ParseSecurityRule(pattern string) (*SecurityRule, error) { return nil, ErrInvalidAction } } else if status == SEG_IP { - // NOTE regutils.MatchCIDR actually also matches IP address without prefix length if regutils.MatchCIDR(seg) { - if idx := strings.Index(seg, "/"); idx > -1 { - if _, ipnet, err := net.ParseCIDR(seg); err != nil { - return nil, ErrInvalidNet - } else { - rule.IPNet = ipnet - } - } else if ip := net.ParseIP(seg); ip != nil { - rule.IPNet = &net.IPNet{ - IP: ip, - Mask: net.CIDRMask(32, 32), - } - } else { - return nil, ErrInvalidIPAddr + _, rule.IPNet, _ = net.ParseCIDR(seg) + } else if regutils.MatchIPAddr(seg) { + rule.IPNet = &net.IPNet{ + IP: net.ParseIP(seg), + Mask: net.CIDRMask(32, 32), } } else { rule.IPNet = &net.IPNet{ diff --git a/vendor/yunion.io/x/sqlchemy/update.go b/vendor/yunion.io/x/sqlchemy/update.go index 4a7e253e9f..3a6512d2fb 100644 --- a/vendor/yunion.io/x/sqlchemy/update.go +++ b/vendor/yunion.io/x/sqlchemy/update.go @@ -85,12 +85,14 @@ func (us *SUpdateSession) saveUpdate(dt interface{}) (map[string]SUpdateDiff, er k := c.Name() of := ofields[k] nf := fields[k] - if c.IsPrimary() && !c.IsZero(of) { // skip update primary key - primaries[k] = of - continue - } else if c.IsKeyIndex() && !c.IsZero(of) { - keyIndexes[k] = of - continue + if !gotypes.IsNil(of) { + if c.IsPrimary() && !c.IsZero(of) { // skip update primary key + primaries[k] = of + continue + } else if c.IsKeyIndex() && !c.IsZero(of) { + keyIndexes[k] = of + continue + } } nc, ok := c.(*SIntegerColumn) if ok && nc.IsAutoVersion {