feat(glance): support backend storage s3

pkg/image/drivers/s3/doc.go

@@ -0,0 +1 @@
+package s3 // import "yunion.io/x/onecloud/pkg/image/drivers/s3"

pkg/image/drivers/s3/minio.go

@@ -0,0 +1,111 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package s3
+
+import (
+	"fmt"
+
+	"github.com/minio/minio-go"
+
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/apis/image"
+)
+
+type Err string
+
+func (e Err) Error() string {
+	return string(e)
+}
+
+const ErrClientNotInit = Err("s3 client not init")
+
+var client *S3Client
+
+// Bucket is image upload bucket name
+type S3Client struct {
+	*minio.Client
+	bucket   string
+	endpoint string
+}
+
+func (c *S3Client) Location(filePath string) string {
+	return fmt.Sprintf("%s%s", image.S3Prefix, filePath)
+}
+
+func Init(endpoint, accessKey, secretKey, bucket string, useSSL bool) error {
+	if client != nil {
+		return nil
+	}
+	minioClient, err := minio.New(endpoint, accessKey, secretKey, useSSL)
+	if err != nil {
+		return errors.Wrap(err, "new minio client")
+	}
+	client = &S3Client{
+		Client:   minioClient,
+		bucket:   bucket,
+		endpoint: endpoint,
+	}
+	err = ensureBucket()
+	if err != nil {
+		return errors.Wrap(err, "ensure bucket")
+	}
+	return nil
+}
+
+func ensureBucket() error {
+	exists, err := client.BucketExists(client.bucket)
+	if err != nil {
+		return errors.Wrap(err, "call bucket exists")
+	}
+	if !exists {
+		if err = client.MakeBucket(client.bucket, ""); err != nil {
+			return errors.Wrap(err, "call make bucket")
+		}
+	}
+	return nil
+}
+
+func Put(filePath, objName string) (string, error) {
+	if client == nil {
+		return "", ErrClientNotInit
+	}
+
+	size, err := client.FPutObject(client.bucket, objName, filePath, minio.PutObjectOptions{})
+	if err != nil {
+		return "", errors.Wrap(err, "put object")
+	}
+	log.Debugf("put object %s size %d", objName, size)
+	return client.Location(objName), nil
+}
+
+func Get(fileName string) (*minio.Object, error) {
+	if client == nil {
+		return nil, ErrClientNotInit
+	}
+	obj, err := client.GetObject(client.bucket, fileName, minio.GetObjectOptions{})
+	if err != nil {
+		return nil, errors.Wrapf(err, "get object %s", fileName)
+	}
+	return obj, nil
+}
+
+func Remove(fileName string) error {
+	if client == nil {
+		return ErrClientNotInit
+	}
+	return client.RemoveObject(client.bucket, fileName)
+}

pkg/image/models/image_subs.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 
 	"yunion.io/x/log"
 
@@ -153,7 +154,6 @@ func (self *SImageSubformat) Save(image *SImage) error {
 		return err
 	}
 	_, err = db.Update(self, func() error {
-		self.Status = api.IMAGE_STATUS_ACTIVE
 		self.Location = fmt.Sprintf("%s%s", LocalFilePrefix, location)
 		self.Checksum = checksum
 		self.FastHash = fastHash
@@ -174,7 +174,7 @@ func (self *SImageSubformat) SaveTorrent() error {
 	if self.TorrentStatus != api.IMAGE_STATUS_QUEUED {
 		return nil // httperrors.NewInvalidStatusError("cannot save torrent in status %s", self.Status)
 	}
-	imgPath := self.getLocalLocation()
+	imgPath := self.GetLocalLocation()
 	torrentPath := filepath.Join(options.Options.TorrentStoreDir, fmt.Sprintf("%s.torrent", filepath.Base(imgPath)))
 	_, err := db.Update(self, func() error {
 		self.TorrentStatus = api.IMAGE_STATUS_SAVING
@@ -209,7 +209,7 @@ func (self *SImageSubformat) SaveTorrent() error {
 	return nil
 }
 
-func (self *SImageSubformat) getLocalLocation() string {
+func (self *SImageSubformat) GetLocalLocation() string {
 	if len(self.Location) > len(LocalFilePrefix) {
 		return self.Location[len(LocalFilePrefix):]
 	}
@@ -244,14 +244,7 @@ func (self *SImageSubformat) RemoveFiles() error {
 			return err
 		}
 	}
-	location = self.getLocalLocation()
-	if len(location) > 0 && fileutils2.IsFile(location) {
-		err := os.Remove(location)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
+	return RemoveImage(self.Location)
 }
 
 type SImageSubformatDetails struct {
@@ -290,7 +283,7 @@ func (self *SImageSubformat) GetDetails() SImageSubformatDetails {
 }
 
 func (self *SImageSubformat) isActive(useFast bool) bool {
-	return isActive(self.getLocalLocation(), self.Size, self.Checksum, self.FastHash, useFast)
+	return isActive(self.GetLocalLocation(), self.Size, self.Checksum, self.FastHash, useFast)
 }
 
 func (self *SImageSubformat) isTorrentActive() bool {
@@ -314,36 +307,38 @@ func (self *SImageSubformat) setTorrentStatus(status string) error {
 }
 
 func (self *SImageSubformat) checkStatus(useFast bool) {
-	if self.isActive(useFast) {
-		if self.Status != api.IMAGE_STATUS_ACTIVE {
-			self.setStatus(api.IMAGE_STATUS_ACTIVE)
-		}
-		if len(self.FastHash) == 0 {
-			fastHash, err := fileutils2.FastCheckSum(self.getLocalLocation())
-			if err != nil {
-				log.Errorf("checkStatus fileutils2.FastChecksum fail %s", err)
-			} else {
-				_, err := db.Update(self, func() error {
-					self.FastHash = fastHash
-					return nil
-				})
+	if strings.HasPrefix(self.Location, LocalFilePrefix) {
+		if self.isActive(useFast) {
+			if self.Status != api.IMAGE_STATUS_ACTIVE {
+				self.setStatus(api.IMAGE_STATUS_ACTIVE)
+			}
+			if len(self.FastHash) == 0 {
+				fastHash, err := fileutils2.FastCheckSum(self.GetLocalLocation())
 				if err != nil {
-					log.Errorf("checkStatus save FastHash fail %s", err)
+					log.Errorf("checkStatus fileutils2.FastChecksum fail %s", err)
+				} else {
+					_, err := db.Update(self, func() error {
+						self.FastHash = fastHash
+						return nil
+					})
+					if err != nil {
+						log.Errorf("checkStatus save FastHash fail %s", err)
+					}
 				}
 			}
+		} else {
+			if self.Status != api.IMAGE_STATUS_QUEUED {
+				self.setStatus(api.IMAGE_STATUS_QUEUED)
+			}
 		}
-	} else {
-		if self.Status != api.IMAGE_STATUS_QUEUED {
-			self.setStatus(api.IMAGE_STATUS_QUEUED)
-		}
-	}
-	if self.isTorrentActive() {
-		if self.TorrentStatus != api.IMAGE_STATUS_ACTIVE {
-			self.setTorrentStatus(api.IMAGE_STATUS_ACTIVE)
-		}
-	} else {
-		if self.TorrentStatus != api.IMAGE_STATUS_QUEUED {
-			self.setTorrentStatus(api.IMAGE_STATUS_QUEUED)
+		if self.isTorrentActive() {
+			if self.TorrentStatus != api.IMAGE_STATUS_ACTIVE {
+				self.setTorrentStatus(api.IMAGE_STATUS_ACTIVE)
+			}
+		} else {
+			if self.TorrentStatus != api.IMAGE_STATUS_QUEUED {
+				self.setTorrentStatus(api.IMAGE_STATUS_QUEUED)
+			}
 		}
 	}
 }

pkg/image/models/images.go

@@ -21,7 +21,7 @@ import (
 	"math"
 	"net/http"
 	"os"
-	"os/exec"
+	"path"
 	"path/filepath"
 	"strconv"
 	"strings"
@@ -50,6 +50,7 @@ import (
 	"yunion.io/x/onecloud/pkg/mcclient/modules"
 	"yunion.io/x/onecloud/pkg/util/fileutils2"
 	"yunion.io/x/onecloud/pkg/util/logclient"
+	"yunion.io/x/onecloud/pkg/util/procutils"
 	"yunion.io/x/onecloud/pkg/util/qemuimg"
 	"yunion.io/x/onecloud/pkg/util/rbacutils"
 	"yunion.io/x/onecloud/pkg/util/streamutils"
@@ -57,7 +58,7 @@ import (
 )
 
 const (
-	LocalFilePrefix = "file://"
+	LocalFilePrefix = api.LocalFilePrefix
 )
 
 type SImageManager struct {
@@ -204,7 +205,7 @@ func (manager *SImageManager) IsCustomizedGetDetailsBody() bool {
 }
 
 func (self *SImage) CustomizedGetDetailsBody(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (jsonutils.JSONObject, error) {
-	filePath := self.getLocalLocation()
+	filePath := self.Location
 	status := self.Status
 
 	if self.IsGuestImage.IsFalse() {
@@ -212,13 +213,18 @@ func (self *SImage) CustomizedGetDetailsBody(ctx context.Context, userCred mccli
 		if len(formatStr) > 0 {
 			subimg := ImageSubformatManager.FetchSubImage(self.Id, formatStr)
 			if subimg != nil {
-				isTorrent := jsonutils.QueryBoolean(query, "torrent", false)
-				if !isTorrent {
-					filePath = subimg.getLocalLocation()
-					status = subimg.Status
+				if strings.HasPrefix(subimg.Location, api.LocalFilePrefix) {
+					isTorrent := jsonutils.QueryBoolean(query, "torrent", false)
+					if !isTorrent {
+						filePath = subimg.Location
+						status = subimg.Status
+					} else {
+						filePath = subimg.getLocalTorrentLocation()
+						status = subimg.TorrentStatus
+					}
 				} else {
-					filePath = subimg.getLocalTorrentLocation()
-					status = subimg.TorrentStatus
+					filePath = subimg.Location
+					status = subimg.Status
 				}
 			} else {
 				return nil, httperrors.NewNotFoundError("format %s not found", formatStr)
@@ -234,22 +240,16 @@ func (self *SImage) CustomizedGetDetailsBody(ctx context.Context, userCred mccli
 		return nil, httperrors.NewInvalidStatusError("empty file path")
 	}
 
+	size, rc, err := GetImage(filePath)
+	if err != nil {
+		return nil, errors.Wrap(err, "get image")
+	}
+	defer rc.Close()
+
 	appParams := appsrv.AppContextGetParams(ctx)
+	appParams.Response.Header().Set("Content-Length", strconv.FormatInt(size, 10))
 
-	fstat, err := os.Stat(filePath)
-	if err != nil {
-		return nil, errors.Wrap(err, "os.Stat")
-	}
-
-	appParams.Response.Header().Set("Content-Length", strconv.FormatInt(fstat.Size(), 10))
-
-	fp, err := os.Open(filePath)
-	if err != nil {
-		return nil, errors.Wrap(err, "os.Open")
-	}
-	defer fp.Close()
-
-	_, err = streamutils.StreamPipe(fp, appParams.Response, false, nil)
+	_, err = streamutils.StreamPipe(rc, appParams.Response, false, nil)
 	if err != nil {
 		return nil, httperrors.NewGeneralError(err)
 	}
@@ -516,7 +516,7 @@ func (self *SImage) SaveImageFromStream(reader io.Reader, calChecksum bool) erro
 		}
 	}
 
-	db.Update(self, func() error {
+	_, err = db.Update(self, func() error {
 		self.Size = sp.Size
 		if calChecksum {
 			self.Checksum = sp.CheckSum
@@ -531,6 +531,9 @@ func (self *SImage) SaveImageFromStream(reader io.Reader, calChecksum bool) erro
 		}
 		return nil
 	})
+	if err != nil {
+		return err
+	}
 
 	return nil
 }
@@ -621,7 +624,7 @@ func (self *SImage) ValidateUpdateData(ctx context.Context, userCred mcclient.To
 			return nil, httperrors.NewInvalidStatusError("cannot upload in status %s", self.Status)
 		}
 		if minDiskSize, err := data.Int("min_disk"); err == nil {
-			img, err := qemuimg.NewQemuImage(self.getLocalLocation())
+			img, err := qemuimg.NewQemuImage(self.GetLocalLocation())
 			if err != nil {
 				return nil, errors.Wrap(err, "open image")
 			}
@@ -818,6 +821,15 @@ func (self *SImage) StartImageConvertTask(ctx context.Context, userCred mcclient
 	return nil
 }
 
+func (self *SImage) StartPutImageTask(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
+	task, err := taskman.TaskManager.NewTask(ctx, "PutImageTask", self, userCred, nil, parentTaskId, "", nil)
+	if err != nil {
+		return err
+	}
+	task.ScheduleRun(nil)
+	return nil
+}
+
 func (self *SImage) AllowPerformCancelDelete(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool {
 	return db.IsAdminAllowPerform(userCred, self, "cancel-delete") && self.IsGuestImage.IsFalse()
 }
@@ -993,7 +1005,7 @@ func (self *SImage) GetImageType() api.TImageType {
 	}
 }
 
-func (self *SImage) newSubformat(ctx context.Context, format qemuimg.TImageFormat, migrate bool) error {
+func (self *SImage) NewSubformat(ctx context.Context, format qemuimg.TImageFormat, migrate bool) error {
 	subformat := &SImageSubformat{}
 	subformat.SetModelManager(ImageSubformatManager, subformat)
 
@@ -1004,7 +1016,7 @@ func (self *SImage) newSubformat(ctx context.Context, format qemuimg.TImageForma
 		subformat.Size = self.Size
 		subformat.Checksum = self.Checksum
 		subformat.FastHash = self.FastHash
-		subformat.Status = api.IMAGE_STATUS_ACTIVE
+		subformat.Status = self.Status
 		subformat.Location = self.Location
 	} else {
 		subformat.Status = api.IMAGE_STATUS_QUEUED
@@ -1037,25 +1049,24 @@ func (self *SImage) MigrateSubImage(ctx context.Context) error {
 	}
 	if self.GetImageType() != api.ImageTypeISO && imgInst.IsSparse() && utils.IsInStringArray(self.DiskFormat, options.Options.TargetImageFormats) {
 		// need to convert again
-		return self.newSubformat(ctx, qemuimg.String2ImageFormat(self.DiskFormat), false)
+		return self.NewSubformat(ctx, qemuimg.String2ImageFormat(self.DiskFormat), false)
 	} else {
-		localPath := self.getLocalLocation()
+		localPath := self.GetLocalLocation()
 		if !strings.HasSuffix(localPath, fmt.Sprintf(".%s", self.DiskFormat)) {
 			newLocalpath := fmt.Sprintf("%s.%s", localPath, self.DiskFormat)
-			cmd := exec.Command("mv", "-f", localPath, newLocalpath)
-			err := cmd.Run()
+			out, err := procutils.NewCommand("mv", "-f", localPath, newLocalpath).Output()
 			if err != nil {
-				return err
+				return errors.Wrapf(err, "rename file failed %s", out)
 			}
 			_, err = db.Update(self, func() error {
-				self.Location = fmt.Sprintf("%s%s", LocalFilePrefix, newLocalpath)
+				self.Location = self.GetNewLocation(newLocalpath)
 				return nil
 			})
 			if err != nil {
 				return err
 			}
 		}
-		return self.newSubformat(ctx, qemuimg.String2ImageFormat(self.DiskFormat), true)
+		return self.NewSubformat(ctx, qemuimg.String2ImageFormat(self.DiskFormat), true)
 	}
 }
 
@@ -1072,7 +1083,7 @@ func (self *SImage) MakeSubImages(ctx context.Context) error {
 			// need to create a record
 			subformat := ImageSubformatManager.FetchSubImage(self.Id, format)
 			if subformat == nil {
-				err := self.newSubformat(ctx, qemuimg.String2ImageFormat(format), false)
+				err := self.NewSubformat(ctx, qemuimg.String2ImageFormat(format), false)
 				if err != nil {
 					return err
 				}
@@ -1088,6 +1099,9 @@ func (self *SImage) ConvertAllSubformats() error {
 		if !utils.IsInStringArray(subimgs[i].Format, options.Options.TargetImageFormats) {
 			continue
 		}
+		if self.DiskFormat == subimgs[i].Format {
+			continue
+		}
 		err := subimgs[i].DoConvert(self)
 		if err != nil {
 			return err
@@ -1096,15 +1110,36 @@ func (self *SImage) ConvertAllSubformats() error {
 	return nil
 }
 
-func (self *SImage) getLocalLocation() string {
-	if len(self.Location) > len(LocalFilePrefix) {
-		return self.Location[len(LocalFilePrefix):]
+func (self *SImage) GetLocalLocation() string {
+	if strings.HasPrefix(self.Location, api.LocalFilePrefix) {
+		return self.Location[len(api.LocalFilePrefix):]
+	} else if strings.HasPrefix(self.Location, api.S3Prefix) {
+		return path.Join(options.Options.S3MountPoint, self.Location[len(api.S3Prefix):])
+	} else {
+		return ""
+	}
+}
+
+func (self *SImage) GetPrefix() string {
+	if strings.HasPrefix(self.Location, api.LocalFilePrefix) {
+		return api.LocalFilePrefix
+	} else if strings.HasPrefix(self.Location, api.S3Prefix) {
+		return api.S3Prefix
+	} else {
+		return api.LocalFilePrefix
+	}
+}
+
+func (self *SImage) GetNewLocation(newLocalPath string) string {
+	if strings.HasPrefix(self.Location, api.S3Prefix) {
+		return api.S3Prefix + path.Base(newLocalPath)
+	} else {
+		return api.LocalFilePrefix + newLocalPath
 	}
-	return ""
 }
 
 func (self *SImage) getQemuImage() (*qemuimg.SQemuImage, error) {
-	return qemuimg.NewQemuImageWithIOLevel(self.getLocalLocation(), qemuimg.IONiceIdle)
+	return qemuimg.NewQemuImageWithIOLevel(self.GetLocalLocation(), qemuimg.IONiceIdle)
 }
 
 func (self *SImage) StopTorrents() {
@@ -1121,16 +1156,8 @@ func (self *SImage) seedTorrents() {
 	}
 }
 
-func (self *SImage) RemoveFiles() error {
-	subimgs := ImageSubformatManager.GetAllSubImages(self.Id)
-	for i := 0; i < len(subimgs); i += 1 {
-		subimgs[i].StopTorrent()
-		err := subimgs[i].RemoveFiles()
-		if err != nil {
-			return err
-		}
-	}
-	filePath := self.getLocalLocation()
+func (self *SImage) RemoveFile() error {
+	filePath := self.GetLocalLocation()
 	if len(filePath) == 0 {
 		filePath = self.GetPath("")
 	}
@@ -1140,6 +1167,22 @@ func (self *SImage) RemoveFiles() error {
 	return nil
 }
 
+func (self *SImage) Remove() error {
+	subimgs := ImageSubformatManager.GetAllSubImages(self.Id)
+	for i := 0; i < len(subimgs); i += 1 {
+		err := subimgs[i].RemoveFiles()
+		if err != nil {
+			return err
+		}
+	}
+	if strings.HasPrefix(self.Location, LocalFilePrefix) {
+		return self.RemoveFile()
+	} else {
+		return RemoveImage(self.Location)
+	}
+
+}
+
 func (manager *SImageManager) getAllAliveImages() []SImage {
 	images := make([]SImage, 0)
 	q := manager.Query().NotIn("status", api.ImageDeadStatus)
@@ -1302,54 +1345,57 @@ func (self *SImage) IsIso() (error, bool) {
 }
 
 func (self *SImage) isActive(useFast bool) bool {
-	return isActive(self.getLocalLocation(), self.Size, self.Checksum, self.FastHash, useFast)
+	return isActive(self.GetLocalLocation(), self.Size, self.Checksum, self.FastHash, useFast)
 }
 
 func (self *SImage) DoCheckStatus(ctx context.Context, userCred mcclient.TokenCredential, useFast bool) {
 	if utils.IsInStringArray(self.Status, api.ImageDeadStatus) {
 		return
 	}
-	if self.isActive(useFast) {
-		if self.Status != api.IMAGE_STATUS_ACTIVE {
-			self.SetStatus(userCred, api.IMAGE_STATUS_ACTIVE, "check active")
-		}
-		if len(self.FastHash) == 0 {
-			fastHash, err := fileutils2.FastCheckSum(self.getLocalLocation())
-			if err != nil {
-				log.Errorf("DoCheckStatus fileutils2.FastChecksum fail %s", err)
-			} else {
-				_, err := db.Update(self, func() error {
-					self.FastHash = fastHash
-					return nil
-				})
+	if IsCheckStatusEnabled(self) {
+		if self.isActive(useFast) {
+			if self.Status != api.IMAGE_STATUS_ACTIVE {
+				self.SetStatus(userCred, api.IMAGE_STATUS_ACTIVE, "check active")
+			}
+			if len(self.FastHash) == 0 {
+				fastHash, err := fileutils2.FastCheckSum(self.GetLocalLocation())
 				if err != nil {
-					log.Errorf("DoCheckStatus save FastHash fail %s", err)
+					log.Errorf("DoCheckStatus fileutils2.FastChecksum fail %s", err)
+				} else {
+					_, err := db.Update(self, func() error {
+						self.FastHash = fastHash
+						return nil
+					})
+					if err != nil {
+						log.Errorf("DoCheckStatus save FastHash fail %s", err)
+					}
 				}
 			}
-		}
-		img, err := qemuimg.NewQemuImage(self.getLocalLocation())
-		if err == nil {
-			format := string(img.Format)
-			virtualSizeMB := int32(img.SizeBytes / 1024 / 1024)
-			if (len(format) > 0 && self.DiskFormat != format) || (virtualSizeMB > 0 && self.MinDiskMB != virtualSizeMB) {
-				db.Update(self, func() error {
-					if len(format) > 0 {
-						self.DiskFormat = format
-					}
-					if virtualSizeMB > 0 && self.MinDiskMB < virtualSizeMB {
-						self.MinDiskMB = virtualSizeMB
-					}
-					return nil
-				})
+			img, err := qemuimg.NewQemuImage(self.GetLocalLocation())
+			if err == nil {
+				format := string(img.Format)
+				virtualSizeMB := int32(img.SizeBytes / 1024 / 1024)
+				if (len(format) > 0 && self.DiskFormat != format) || (virtualSizeMB > 0 && self.MinDiskMB != virtualSizeMB) {
+					db.Update(self, func() error {
+						if len(format) > 0 {
+							self.DiskFormat = format
+						}
+						if virtualSizeMB > 0 && self.MinDiskMB < virtualSizeMB {
+							self.MinDiskMB = virtualSizeMB
+						}
+						return nil
+					})
+				}
+			} else {
+				log.Warningf("fail to check image size of %s(%s)", self.Id, self.Name)
 			}
 		} else {
-			log.Warningf("fail to check image size of %s(%s)", self.Id, self.Name)
-		}
-	} else {
-		if self.Status != api.IMAGE_STATUS_QUEUED {
-			self.SetStatus(userCred, api.IMAGE_STATUS_QUEUED, "check inactive")
+			if self.Status != api.IMAGE_STATUS_QUEUED {
+				self.SetStatus(userCred, api.IMAGE_STATUS_QUEUED, "check inactive")
+			}
 		}
 	}
+
 	needConvert := false
 	subimgs := ImageSubformatManager.GetAllSubImages(self.Id)
 	// for image the part of a guest image, convert is not necessary.
@@ -1358,7 +1404,7 @@ func (self *SImage) DoCheckStatus(ctx context.Context, userCred mcclient.TokenCr
 	}
 	for i := 0; i < len(subimgs); i += 1 {
 		subimgs[i].checkStatus(useFast)
-		if (subimgs[i].Status != api.IMAGE_STATUS_ACTIVE || subimgs[i].TorrentStatus != api.IMAGE_STATUS_ACTIVE) && utils.IsInStringArray(subimgs[i].Format, options.Options.TargetImageFormats) {
+		if subimgs[i].Status != api.IMAGE_STATUS_ACTIVE && utils.IsInStringArray(subimgs[i].Format, options.Options.TargetImageFormats) {
 			needConvert = true
 		}
 	}
@@ -1368,6 +1414,9 @@ func (self *SImage) DoCheckStatus(ctx context.Context, userCred mcclient.TokenCr
 			self.StartImageConvertTask(ctx, userCred, "")
 		} else if options.Options.EnableTorrentService {
 			self.seedTorrents()
+		} else {
+			log.Infof("Image %s put to specific storage", self.Name)
+			self.StartPutImageTask(ctx, userCred, "")
 		}
 	}
 }

pkg/image/models/storage.go

@@ -0,0 +1,158 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package models
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"strings"
+
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/apis/image"
+	"yunion.io/x/onecloud/pkg/image/drivers/s3"
+	"yunion.io/x/onecloud/pkg/util/procutils"
+)
+
+var local Storage = &LocalStorage{}
+var s3Instance Storage = &S3Storage{}
+var storage Storage
+
+func GetStorage() Storage {
+	return storage
+}
+
+func GetImage(location string) (int64, io.ReadCloser, error) {
+	switch {
+	case strings.HasPrefix(location, image.S3Prefix):
+		return s3Instance.GetImage(location[len(image.S3Prefix):])
+	case strings.HasPrefix(location, image.LocalFilePrefix):
+		return local.GetImage(location[len(image.LocalFilePrefix):])
+	default:
+		return local.GetImage(location)
+	}
+}
+
+func RemoveImage(location string) error {
+	switch {
+	case strings.HasPrefix(location, image.S3Prefix):
+		return s3Instance.RemoveImage(location[len(image.S3Prefix):])
+	case strings.HasPrefix(location, image.LocalFilePrefix):
+		return local.RemoveImage(location[len(image.LocalFilePrefix):])
+	default:
+		return local.RemoveImage(location)
+	}
+}
+
+func IsCheckStatusEnabled(img *SImage) bool {
+	switch {
+	case strings.HasPrefix(img.Location, image.S3Prefix):
+		return s3Instance.IsCheckStatusEnabled()
+	case strings.HasPrefix(img.Location, image.LocalFilePrefix):
+		return local.IsCheckStatusEnabled()
+	default:
+		return local.IsCheckStatusEnabled()
+	}
+}
+
+func Init(storageBackend string) {
+	switch storageBackend {
+	case "local":
+		storage = &LocalStorage{}
+	case "s3":
+		storage = &S3Storage{}
+	default:
+		storage = &LocalStorage{}
+	}
+}
+
+type Storage interface {
+	SaveImage(string) (string, error)
+	CleanTempfile(string) error
+	GetImage(string) (int64, io.ReadCloser, error)
+	RemoveImage(string) error
+
+	IsCheckStatusEnabled() bool
+}
+
+type LocalStorage struct{}
+
+func (s *LocalStorage) SaveImage(imagePath string) (string, error) {
+	return fmt.Sprintf("%s%s", LocalFilePrefix, imagePath), nil
+}
+
+func (s *LocalStorage) CleanTempfile(filePath string) error {
+	return nil
+}
+
+func (s *LocalStorage) GetImage(imagePath string) (int64, io.ReadCloser, error) {
+	fstat, err := os.Stat(imagePath)
+	if err != nil {
+		return -1, nil, errors.Wrapf(err, "stat file %s", imagePath)
+	}
+	f, err := os.Open(imagePath)
+	if err != nil {
+		return -1, nil, errors.Wrapf(err, "open file %s", imagePath)
+	}
+	return fstat.Size(), f, nil
+}
+
+func (s *LocalStorage) IsCheckStatusEnabled() bool {
+	return true
+}
+
+func (s *LocalStorage) RemoveImage(imagePath string) error {
+	return os.Remove(imagePath)
+}
+
+type S3Storage struct{}
+
+func imagePathToName(imagePath string) string {
+	segs := strings.Split(imagePath, "/")
+	return segs[len(segs)-1]
+}
+
+func (s *S3Storage) SaveImage(imagePath string) (string, error) {
+	return s3.Put(imagePath, imagePathToName(imagePath))
+}
+
+func (s *S3Storage) CleanTempfile(filePath string) error {
+	out, err := procutils.NewCommand("rm", "-f", filePath).Output()
+	if err != nil {
+		return errors.Wrapf(err, "rm %s failed %s", filePath, out)
+	}
+	return nil
+}
+
+func (s *S3Storage) GetImage(imagePath string) (int64, io.ReadCloser, error) {
+	obj, err := s3.Get(imagePathToName(imagePath))
+	if err != nil {
+		return -1, nil, errors.Wrap(err, "s3 get image")
+	}
+	objInfo, err := obj.Stat()
+	if err != nil {
+		return -1, nil, errors.Wrap(err, "s3 obj stat")
+	}
+	return objInfo.Size, obj, nil
+}
+
+func (s *S3Storage) IsCheckStatusEnabled() bool {
+	return false
+}
+
+func (s *S3Storage) RemoveImage(fileName string) error {
+	return s3.Remove(fileName)
+}

pkg/image/options/options.go

@@ -41,6 +41,15 @@ type SImageOptions struct {
 	TorrentClientPath string `help:"path to torrent executable" default:"/opt/yunion/bin/torrent"`
 
 	DeployServerSocketPath string `help:"Deploy server listen socket path" default:"/var/run/onecloud/deploy.sock"`
+
+	StorageDriver string `help:"image backend storage" default:"local" choices:"s3|local"`
+
+	S3AccessKey  string `help:"s3 access key"`
+	S3SecretKey  string `help:"s3 secret key"`
+	S3Endpoint   string `help:"s3 endpoint"`
+	S3UseSSL     bool   `help:"s3 access use ssl"`
+	S3BucketName string `help:"s3 bucket name" default:"onecloud-images"`
+	S3MountPoint string `help:"s3fs mount point" default:"/opt/cloud/workspace/data/glance/s3images"`
 }
 
 var (

pkg/image/service/service.go

@@ -15,6 +15,7 @@
 package service
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
 	"time"
@@ -30,12 +31,14 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
 	common_options "yunion.io/x/onecloud/pkg/cloudcommon/options"
 	"yunion.io/x/onecloud/pkg/hostman/hostdeployer/deployclient"
+	"yunion.io/x/onecloud/pkg/image/drivers/s3"
 	"yunion.io/x/onecloud/pkg/image/models"
 	"yunion.io/x/onecloud/pkg/image/options"
 	_ "yunion.io/x/onecloud/pkg/image/policy"
 	_ "yunion.io/x/onecloud/pkg/image/tasks"
 	"yunion.io/x/onecloud/pkg/image/torrent"
 	"yunion.io/x/onecloud/pkg/util/fileutils2"
+	"yunion.io/x/onecloud/pkg/util/procutils"
 )
 
 func StartService() {
@@ -97,6 +100,10 @@ func StartService() {
 	common_options.StartOptionManager(opts, opts.ConfigSyncPeriodSeconds, api.SERVICE_TYPE, api.SERVICE_VERSION, options.OnOptionsChange)
 
 	go models.CheckImages()
+	models.Init(options.Options.StorageDriver)
+	if options.Options.StorageDriver == "s3" {
+		initS3()
+	}
 
 	if len(options.Options.DeployServerSocketPath) > 0 {
 		log.Infof("deploy server socket path: %s", options.Options.DeployServerSocketPath)
@@ -121,5 +128,50 @@ func StartService() {
 		if options.Options.EnableTorrentService {
 			torrent.StopTorrents()
 		}
+		if options.Options.StorageDriver == "s3" {
+			procutils.NewCommand("umount", options.Options.S3MountPoint).Run()
+		}
 	})
 }
+
+func initS3() {
+	err := s3.Init(
+		options.Options.S3Endpoint,
+		options.Options.S3AccessKey,
+		options.Options.S3SecretKey,
+		options.Options.S3BucketName,
+		options.Options.S3UseSSL,
+	)
+	if err != nil {
+		log.Fatalf("failed init s3 client %s", err)
+	}
+	func() {
+		fd, err := os.OpenFile("/tmp/s3-pass", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
+		if err != nil {
+			log.Fatalf("failed open s3 pass file %s", err)
+		}
+		defer fd.Close()
+		_, err = fd.WriteString(fmt.Sprintf("%s:%s", options.Options.S3AccessKey, options.Options.S3SecretKey))
+		if err != nil {
+			log.Fatalf("failed write s3 pass file")
+		}
+	}()
+	if !fileutils2.Exists(options.Options.S3MountPoint) {
+		err := os.MkdirAll(options.Options.S3MountPoint, 0755)
+		if err != nil {
+			log.Fatalf("fail to create %s: %s", options.Options.S3MountPoint, err)
+		}
+	}
+
+	prefix := "http://"
+	if options.Options.S3UseSSL {
+		prefix = "https://"
+	}
+	url := prefix + options.Options.S3Endpoint
+	out, err := procutils.NewCommand("s3fs",
+		options.Options.S3BucketName, options.Options.S3MountPoint,
+		"-o", fmt.Sprintf("passwd_file=/tmp/s3-pass,use_path_request_style,url=%s", url)).Output()
+	if err != nil {
+		log.Fatalf("failed mount s3fs %s %s", err, out)
+	}
+}

pkg/image/tasks/guest_image_delete_task.go

@@ -78,7 +78,7 @@ func (self *GuestImageDeleteTask) startDelete(ctx context.Context, guestImage *m
 		self.taskFailed(ctx, guestImage, jsonutils.NewString(err.Error()))
 	}
 	for i := range images {
-		err := images[i].RemoveFiles()
+		err := images[i].Remove()
 		if err != nil {
 			self.taskFailed(ctx, guestImage, jsonutils.NewString(fmt.Sprintf("fail to remove %s: %s", images[i].GetPath(""), err)))
 			return

pkg/image/tasks/image_convert_task.go

@@ -16,9 +16,10 @@ package tasks
 
 import (
 	"context"
-	"fmt"
+	"strings"
 
 	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
 
 	api "yunion.io/x/onecloud/pkg/apis/image"
 	"yunion.io/x/onecloud/pkg/appsrv"
@@ -27,6 +28,7 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/notifyclient"
 	"yunion.io/x/onecloud/pkg/image/models"
 	"yunion.io/x/onecloud/pkg/mcclient/modules/notify"
+	"yunion.io/x/onecloud/pkg/util/procutils"
 )
 
 type ImageConvertTask struct {
@@ -35,43 +37,107 @@ type ImageConvertTask struct {
 
 func init() {
 	convertWorker := appsrv.NewWorkerManager("ImageConvertTaskWorkerManager", 2, 512, true)
+	putWorker := appsrv.NewWorkerManager("PutImageTaskWorkerManager", 4, 512, true)
 	taskman.RegisterTaskAndWorker(ImageConvertTask{}, convertWorker)
+	taskman.RegisterTaskAndWorker(PutImageTask{}, putWorker)
 }
 
 func (self *ImageConvertTask) OnInit(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) {
 	image := obj.(*models.SImage)
-
+	self.Params.Set("old_status", jsonutils.NewString(image.Status))
 	self.SetStage("OnConvertComplete", nil)
 	taskman.LocalTaskRun(self, func() (jsonutils.JSONObject, error) {
-		imgOldStatus := image.Status
 		image.SetStatus(self.UserCred, api.IMAGE_STATUS_CONVERTING, "start convert")
-		err := image.ConvertAllSubformats()
-		var msg string
-		if err != nil {
-			msg = fmt.Sprintf("convert failed: %s", err)
-		} else {
-			msg = fmt.Sprintf("convert success")
-		}
-
-		image.SetStatus(self.UserCred, api.IMAGE_STATUS_ACTIVE, msg)
-		if imgOldStatus != api.IMAGE_STATUS_ACTIVE {
-			kwargs := jsonutils.NewDict()
-			kwargs.Set("name", jsonutils.NewString(image.GetName()))
-			osType, err := models.ImagePropertyManager.GetProperty(image.Id, api.IMAGE_OS_TYPE)
-			if err == nil {
-				kwargs.Set("os_type", jsonutils.NewString(osType.Value))
-			}
-			notifyclient.SystemNotifyWithCtx(ctx, notify.NotifyPriorityNormal, notifyclient.IMAGE_ACTIVED, kwargs)
-			notifyclient.NotifyImportantWithCtx(ctx, []string{self.UserCred.GetUserId()}, false, notifyclient.IMAGE_ACTIVED, kwargs)
-		}
-		return nil, err
+		return nil, image.ConvertAllSubformats()
 	})
 }
 
-func (self *ImageConvertTask) OnConvertComplete(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) {
+func (self *ImageConvertTask) OnConvertComplete(ctx context.Context, image *models.SImage, data jsonutils.JSONObject) {
+	image.StartPutImageTask(ctx, self.UserCred, "")
 	self.SetStageComplete(ctx, nil)
 }
 
-func (self *ImageConvertTask) OnConvertCompleteFailed(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) {
+func (self *ImageConvertTask) OnConvertCompleteFailed(ctx context.Context, image *models.SImage, data jsonutils.JSONObject) {
+	image.StartPutImageTask(ctx, self.UserCred, "")
 	self.SetStageFailed(ctx, data)
 }
+
+type PutImageTask struct {
+	taskman.STask
+}
+
+func (self *PutImageTask) OnInit(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) {
+	image := obj.(*models.SImage)
+	oldStatus := image.Status
+	if strings.HasPrefix(image.Location, models.LocalFilePrefix) {
+		imagePath := image.GetLocalLocation()
+		image.SetStatus(self.UserCred, api.IMAGE_STATUS_SAVING, "save image to specific storage")
+		location, err := models.GetStorage().SaveImage(imagePath)
+		if err != nil {
+			log.Errorf("Failed save image to specific storage %s", err)
+		} else if location != image.Location {
+			_, err = db.Update(image, func() error {
+				image.Location = location
+				return nil
+			})
+			if err != nil {
+				log.Errorf("failed update image location %s", err)
+			} else {
+				if err = procutils.NewCommand("rm", "-f", imagePath).Run(); err != nil {
+					log.Errorf("failed remove file %s: %s", imagePath, err)
+				}
+			}
+		}
+	}
+	image.SetStatus(self.UserCred, api.IMAGE_STATUS_ACTIVE, "save image to specific storage complete")
+	if oldStatus != api.IMAGE_STATUS_ACTIVE {
+		kwargs := jsonutils.NewDict()
+		kwargs.Set("name", jsonutils.NewString(image.GetName()))
+		osType, err := models.ImagePropertyManager.GetProperty(image.Id, api.IMAGE_OS_TYPE)
+		if err == nil {
+			kwargs.Set("os_type", jsonutils.NewString(osType.Value))
+		}
+		notifyclient.SystemNotifyWithCtx(ctx, notify.NotifyPriorityNormal, notifyclient.IMAGE_ACTIVED, kwargs)
+		notifyclient.NotifyImportantWithCtx(ctx, []string{self.UserCred.GetUserId()}, false, notifyclient.IMAGE_ACTIVED, kwargs)
+	}
+
+	subimgs := models.ImageSubformatManager.GetAllSubImages(image.Id)
+	for i := 0; i < len(subimgs); i++ {
+		if !strings.HasPrefix(subimgs[i].Location, models.LocalFilePrefix) {
+			continue
+		}
+		if subimgs[i].Format == image.DiskFormat {
+			_, err := db.Update(&subimgs[i], func() error {
+				subimgs[i].Location = image.Location
+				subimgs[i].Status = api.IMAGE_STATUS_ACTIVE
+				return nil
+			})
+			if err != nil {
+				log.Errorf("failed update subimg %s", err)
+			}
+		} else {
+			imagePath := subimgs[i].GetLocalLocation()
+			location, err := models.GetStorage().SaveImage(imagePath)
+			if err != nil {
+				log.Errorf("Failed save image to sepcific storage %s", err)
+			} else if subimgs[i].Location != location {
+				_, err := db.Update(&subimgs[i], func() error {
+					subimgs[i].Location = location
+					return nil
+				})
+				if err != nil {
+					log.Errorf("failed update subimg %s", err)
+				}
+				if err = procutils.NewCommand("rm", "-f", imagePath).Run(); err != nil {
+					log.Errorf("failed remove file %s: %s", imagePath, err)
+				}
+			}
+			db.Update(&subimgs[i], func() error {
+				subimgs[i].Status = api.IMAGE_STATUS_ACTIVE
+				return nil
+			})
+		}
+	}
+
+	self.SetStageComplete(ctx, nil)
+}

pkg/image/tasks/image_delete_task.go

@@ -62,9 +62,9 @@ func (self *ImageDeleteTask) startPendingDeleteImage(ctx context.Context, image
 }
 
 func (self *ImageDeleteTask) startDeleteImage(ctx context.Context, image *models.SImage) {
-	err := image.RemoveFiles()
+	err := image.Remove()
 	if err != nil {
-		msg := fmt.Sprintf("fail to remove %s %s", image.GetPath(""), err)
+		msg := fmt.Sprintf("fail to remove %s %s", image.Name, err)
 		log.Errorf(msg)
 		self.SetStageFailed(ctx, jsonutils.NewString(msg))
 		return

pkg/image/tasks/image_probe_task.go

@@ -172,16 +172,7 @@ func (self *ImageProbeTask) OnProbeFailed(ctx context.Context, image *models.SIm
 	db.OpsLog.LogEvent(image, db.ACT_PROBE_FAIL, reason, self.UserCred)
 	logclient.AddActionLogWithContext(ctx, image, logclient.ACT_IMAGE_PROBE, reason, self.UserCred, false)
 
-	if jsonutils.QueryBoolean(self.Params, "do_convert", false) {
-		self.SetStage("OnConvertComplete", nil)
-		if err := image.StartImageConvertTask(ctx, self.UserCred, self.GetId()); err != nil {
-			image.SetStatus(self.UserCred, api.IMAGE_STATUS_ACTIVE, "")
-			self.SetStageFailed(ctx, jsonutils.NewString(err.Error()))
-		}
-	} else {
-		image.SetStatus(self.UserCred, api.IMAGE_STATUS_ACTIVE, "")
-		self.SetStageFailed(ctx, reason)
-	}
+	self.OnProbe(ctx, image)
 }
 
 func (self *ImageProbeTask) OnProbeSuccess(ctx context.Context, image *models.SImage) {
@@ -190,15 +181,24 @@ func (self *ImageProbeTask) OnProbeSuccess(ctx context.Context, image *models.SI
 	logclient.AddActionLogWithContext(
 		ctx, image, logclient.ACT_IMAGE_PROBE, "Image Probe Success", self.UserCred, true)
 
+	self.OnProbe(ctx, image)
+}
+
+func (self *ImageProbeTask) OnProbe(ctx context.Context, image *models.SImage) {
+	self.SetStage("OnConvertComplete", nil)
 	if jsonutils.QueryBoolean(self.Params, "do_convert", false) {
-		self.SetStage("OnConvertComplete", nil)
 		if err := image.StartImageConvertTask(ctx, self.UserCred, self.GetId()); err != nil {
-			image.SetStatus(self.UserCred, api.IMAGE_STATUS_ACTIVE, "")
-			self.SetStageFailed(ctx, jsonutils.NewString(err.Error()))
+			log.Errorf("start image convert task failed %s", err)
+			if err := image.StartPutImageTask(ctx, self.UserCred, self.GetId()); err != nil {
+				image.SetStatus(self.UserCred, api.IMAGE_STATUS_KILLED, err.Error())
+				self.SetStageFailed(ctx, jsonutils.NewString(err.Error()))
+			}
 		}
 	} else {
-		image.SetStatus(self.UserCred, api.IMAGE_STATUS_ACTIVE, "")
-		self.SetStageComplete(ctx, nil)
+		if err := image.StartPutImageTask(ctx, self.UserCred, self.GetId()); err != nil {
+			image.SetStatus(self.UserCred, api.IMAGE_STATUS_KILLED, err.Error())
+			self.SetStageFailed(ctx, jsonutils.NewString(err.Error()))
+		}
 	}
 }