diff --git a/Gopkg.lock b/Gopkg.lock index cf1b45e365..04d3e9b288 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" @@ -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..89e64a49ce 100644 --- a/pkg/appsrv/response.go +++ b/pkg/appsrv/response.go @@ -2,9 +2,9 @@ package appsrv import ( "context" + "fmt" "net/http" - "fmt" "yunion.io/x/onecloud/pkg/httperrors" ) @@ -56,6 +56,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..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 @@ -113,6 +116,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 { @@ -437,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") } @@ -1006,6 +1020,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 +1080,15 @@ 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 + 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 +1100,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 +1170,9 @@ 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 { + self.SnapshotId = diskConfig.SnapshotId + self.DiskType = diskConfig.DiskType } if len(diskConfig.Fs) > 0 { self.FsFormat = diskConfig.Fs diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index 011e28c134..b9d046fda0 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" @@ -2113,3 +2113,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 352012aa0b..784bc461b9 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.SnapshotId) > 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", i+1, DISK_TYPE_SYS) + } if len(diskConfig.Backend) == 0 { diskConfig.Backend = rootStorageType } @@ -2344,6 +2350,9 @@ func (self *SGuest) CreateDisksOnHost(ctx context.Context, userCred mcclient.Tok return err } data.Add(jsonutils.NewString(disk.Id), fmt.Sprintf("disk.%d.id", idx)) + 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 473ddbcc6e..52752c8f66 100644 --- a/pkg/compute/models/hosts.go +++ b/pkg/compute/models/hosts.go @@ -649,7 +649,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/models/snapshots.go b/pkg/compute/models/snapshots.go index ed9c81a6c4..a4aa643515 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,20 @@ func (self *SSnapshot) GetHost() *SHost { return storage.GetMasterHost() } +func (self *SSnapshotManager) AddRefCount(snapshotId string, count int) { + iSnapshot, _ := self.FetchById(snapshotId) + 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) + } + } +} + func (self *SSnapshotManager) GetDiskSnapshotsByCreate(diskId, createdBy string) []SSnapshot { dest := make([]SSnapshot, 0) q := self.Query().SubQuery() @@ -343,6 +360,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/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/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 2305667475..2b9181df40 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,8 @@ 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)) + 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..1b04b596be --- /dev/null +++ b/pkg/hostimage/host_image_service.go @@ -0,0 +1,233 @@ +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() +}