Merge pull request #717 in YUNIONIO/onecloud from ~WANYAOQI/onecloud:feature/wyq/server-create-disk-by-snapshot to release/2.4.0

* commit 'ad2de3384697067d40a68e06e7870c2a9e9612b3':
  fix code
  add snapshot reference count, fix code
  server create disk by snapshot, mount snapshot by fusefs

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",

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"

cmd/host-image/main.go

@@ -0,0 +1,7 @@
+package main
+
+import "yunion.io/x/onecloud/pkg/hostimage"
+
+func main() {
+	hostimage.StartService()
+}

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

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()

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

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
+}

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")

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
 }

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 {

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
 }
 

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
 }
 

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)
 }

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

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/<sid>", prefix), auth.Authenticate(getImage))
+	app.AddHandler("GET", fmt.Sprintf("%s/snapshots/<diskId>/<sid>", prefix), auth.Authenticate(getImage))
+	app.AddHandler("HEAD", fmt.Sprintf("%s/disks/<sid>", prefix), auth.Authenticate(getImageMeta))
+	app.AddHandler("HEAD", fmt.Sprintf("%s/snapshots/<diskId>/<sid>", 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["<sid>"]
+	var imagePath string
+	if diskId, ok := params["<diskId>"]; 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)
+}

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()
+}