opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-01 15:07:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: host cache image directly from object storage (#22317)

Browse Source 
Co-authored-by: Qiu Jian <qiujian@yunionyun.com>
This commit is contained in:
Jian Qiu
2025-03-22 08:33:42 +08:00
committed by GitHub
parent 79ae3be805
commit a778fb9dd5
14 changed files with 201 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2 -2
View File
@@ -93,12 +93,12 @@ require (
k8s.io/cri-api v0.22.17
k8s.io/klog/v2 v2.20.0
moul.io/http2curl/v2 v2.3.0
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20250317132726-d9536ae12f8c
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20250321092740-6d5a1a097a47
yunion.io/x/executor v0.0.0-20241205080005-48f5b1212256
yunion.io/x/jsonutils v1.0.1-0.20240930100528-1671a2d0d22f
yunion.io/x/log v1.0.1-0.20240305175729-7cf2d6cd5a91
yunion.io/x/ovsdb v0.0.0-20230306173834-f164f413a900
yunion.io/x/pkg v1.10.4-0.20250128110515-2cde6f625882
yunion.io/x/pkg v1.10.4-0.20250321091808-805c8df74399
yunion.io/x/s3cli v0.0.0-20241221171442-1c11599d28e1
yunion.io/x/sqlchemy v1.1.3-0.20250307104108-2f8e0af883e9
yunion.io/x/structarg v0.0.0-20231017124457-df4d5009457c
go.sum
+4 -4
View File
@@ -1376,8 +1376,8 @@ sigs.k8s.io/structured-merge-diff/v4 v4.0.1/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK
sigs.k8s.io/yaml v1.1.0/go.mod h1:UJmg0vDUVViEyp3mgSv9WPwZCDxu4rQW1olrI1uml+o=
sigs.k8s.io/yaml v1.2.0 h1:kr/MCeFWJWTwyaHoR9c8EjH9OumOmoF9YGiZd7lFm/Q=
sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20250317132726-d9536ae12f8c h1:v/fzLoIK8c5SBrirCeG+VpO+jOagbi3QftrOdsaQJAE=
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20250317132726-d9536ae12f8c/go.mod h1:KQ/jWx7bZlmjCE711KEWuvHW/dzpdr/UTlBjjutkj0Y=
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20250321092740-6d5a1a097a47 h1:dWSgfCr60VzeW9MLX0aBwvDRERrciuwZm3EQNe1Y2ws=
yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20250321092740-6d5a1a097a47/go.mod h1:krsxghCPvP3xuDKObVtkJr9TZ9bhbcY1GApmsrIui34=
yunion.io/x/executor v0.0.0-20241205080005-48f5b1212256 h1:kLKQ6zbgPDQflRwoHFAjxNChcbhXIFgsUVLkJwiXu/8=
yunion.io/x/executor v0.0.0-20241205080005-48f5b1212256/go.mod h1:Uxuou9WQIeJXNpy7t2fPLL0BYLvLiMvGQwY7Qc6aSws=
yunion.io/x/jsonutils v0.0.0-20190625054549-a964e1e8a051/go.mod h1:4N0/RVzsYL3kH3WE/H1BjUQdFiWu50JGCFQuuy+Z634=
@@ -1391,8 +1391,8 @@ yunion.io/x/ovsdb v0.0.0-20230306173834-f164f413a900 h1:Hu/4ERvoWaN6aiFs4h4/yvVB
yunion.io/x/ovsdb v0.0.0-20230306173834-f164f413a900/go.mod h1:0vLkNEhlmA64HViPBAnSTUMrx5QP1CLsxXmxDKQ80tc=
yunion.io/x/pkg v0.0.0-20190620104149-945c25821dbf/go.mod h1:t6rEGG2sQ4J7DhFxSZVOTjNd0YO/KlfWQyK1W4tog+E=
yunion.io/x/pkg v0.0.0-20190628082551-f4033ba2ea30/go.mod h1:t6rEGG2sQ4J7DhFxSZVOTjNd0YO/KlfWQyK1W4tog+E=
yunion.io/x/pkg v1.10.4-0.20250128110515-2cde6f625882 h1:Nl4qngZ4mrVe2xsMVrCHLe9t+nY/KpaBXO9a0sTbg4M=
yunion.io/x/pkg v1.10.4-0.20250128110515-2cde6f625882/go.mod h1:0Bwxqd9MA3ACi119/l02FprY/o9gHahmYC2bsSbnVpM=
yunion.io/x/pkg v1.10.4-0.20250321091808-805c8df74399 h1:tm4mYslBe/tWKF950GSkpqKU5l7UWaYfm07zG2Tlpyo=
yunion.io/x/pkg v1.10.4-0.20250321091808-805c8df74399/go.mod h1:0Bwxqd9MA3ACi119/l02FprY/o9gHahmYC2bsSbnVpM=
yunion.io/x/s3cli v0.0.0-20241221171442-1c11599d28e1 h1:1KJ3YYinydPHpDEQRXdr/T8SYcKZ5Er+m489H+PnaQ4=
yunion.io/x/s3cli v0.0.0-20241221171442-1c11599d28e1/go.mod h1:0iFKpOs1y4lbCxeOmq3Xx/0AcQoewVPwj62eRluioEo=
yunion.io/x/sqlchemy v1.1.3-0.20250307104108-2f8e0af883e9 h1:KU8BikBBUJlSkA1Ib01htqaDdzt20K+ezNkTZD8Rxjk=
pkg/hostman/container/prober/prober.go
+3 -3
View File
@@ -104,7 +104,7 @@ func (pb *prober) probe(probeType apis.ContainerProbeType, pod IPod, container *
log.Warningf(msg)
} else {
msg = fmt.Sprintf("%s probe for %q succeeded", probeType, ctrName)
log.Debugf(msg)
//log.Debugf(msg)
}
return results.NewSuccess(msg), nil
}
@@ -127,7 +127,7 @@ func (pb *prober) runProbeWithRetries(probeType apis.ContainerProbeType, p *apis
func (pb *prober) runProbe(probeType apis.ContainerProbeType, p *apis.ContainerProbe, pod IPod, container *hostapi.ContainerDesc) (probe.Result, string, error) {
timeout := time.Duration(p.TimeoutSeconds) * time.Second
if p.Exec != nil {
log.Debugf("Exec-Probe Pod: %v, Container: %v, Command: %v", pod.GetDesc().Name, container.Name, p.Exec.Command)
// log.Debugf("Exec-Probe Pod: %v, Container: %v, Command: %v", pod.GetDesc().Name, container.Name, p.Exec.Command)
return pb.exec.Probe(pb.newExecInContainer(pod, container, p.Exec.Command, timeout), strings.Join(p.Exec.Command, " "))
}
if p.TCPSocket != nil {
@@ -144,7 +144,7 @@ func (pb *prober) runProbe(probeType apis.ContainerProbeType, p *apis.ContainerP
return probe.Unknown, "", errors.Errorf("not found guest ip")
}
}
log.Debugf("TCP-Probe Host: %v, Port: %v, Timeout: %v", host, port, timeout)
// log.Debugf("TCP-Probe Host: %v, Port: %v, Timeout: %v", host, port, timeout)
return pb.tcp.Probe(host, port, timeout)
}
errMsg := fmt.Sprintf("Failed to find probe builder for pod %v, container: %v", pod.GetName(), container.Name)
pkg/hostman/guestman/pod/pleg/generic.go
+1 -1
View File
@@ -131,7 +131,7 @@ func (g *GenericPLEG) updateRelistTime(timestamp time.Time) {
// relist queries the container runtime for list of pods/containers, compare
// with the internal pods/containers, and generates events accordingly.
func (g *GenericPLEG) relist() {
log.Debugf("GenericPLEG: Relisting")
// log.Debugf("GenericPLEG: Relisting")
timestamp := g.clock.Now()
pkg/hostman/hostinfo/hostinfo.go
+1 -1
View File
@@ -2709,7 +2709,7 @@ func (h *SHostInfo) getProcessesPids(processesPrefix []string) (map[string]strin
segs := strings.Split(cmdline, "\x00")
if utils.IsInStringArray(segs[0], processesPrefix) {
res[segs[0]] = f.Name()
log.Debugf("getProcessesPids append %s %s", segs[0], f.Name())
// log.Debugf("getProcessesPids append %s %s", segs[0], f.Name())
}
}
}
pkg/hostman/hostinfo/hostpinger/hostpinger.go
+1 -1
View File
@@ -131,7 +131,7 @@ func (p *SHostPingTask) payload() api.SHostPingInput {
}
func (p *SHostPingTask) ping(div int, hostId string) error {
log.Debugf("ping region at %d...", div)
// log.Debugf("ping region at %d...", div)
res, err := modules.Hosts.PerformAction(hostutils.GetComputeSession(context.Background()),
hostId, "ping", jsonutils.Marshal(p.payload()))
if err != nil {
pkg/hostman/storageman/remotefile/remotefile.go
+39 -1
View File
@@ -61,6 +61,8 @@ type SRemoteFile struct {
chksum string
format string
name string
s3Info *S3RemoteFileInfo
}
func NewRemoteFile(
@@ -107,7 +109,7 @@ func (r *SRemoteFile) Fetch(callback func(progress, progressMbps float64, totalS
}
return nil
}
log.Infof("Fetch remote file %q to %q", r.downloadUrl, r.tmpPath)
log.Infof("Fetch remote file %q %q to %q", r.url, r.downloadUrl, r.tmpPath)
return r.fetch("", callback)
}
@@ -210,7 +212,24 @@ func (r *SRemoteFile) download(getData bool, preChksum string, callback func(pro
return r.downloadInternal(getData, preChksum, callback)
}
func (r *SRemoteFile) downloadS3(callback func(progress, progressMbps float64, totalSizeMb int64)) error {
defer func() {
// clear s3 info
r.s3Info = nil
}()
err := r.s3Info.download(r.ctx, r.tmpPath, callback)
if err != nil {
return errors.Wrap(err, "download s3")
}
return nil
}
func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
if getData && r.s3Info != nil {
return r.downloadS3(callback)
}
var header = http.Header{}
header.Set("X-Auth-Token", auth.GetTokenString())
if len(preChksum) > 0 {
@@ -321,4 +340,23 @@ func (r *SRemoteFile) setProperties(header http.Header) {
if name := header.Get("X-Image-Meta-Name"); len(name) > 0 {
r.name = name
}
if s3Url := header.Get("X-Image-Meta-S3_info_url"); len(s3Url) > 0 {
r.s3Info = &S3RemoteFileInfo{}
r.s3Info.Url = s3Url
if s3AccessKey := header.Get("X-Image-Meta-S3_info_access_key"); len(s3AccessKey) > 0 {
r.s3Info.AcessKey = s3AccessKey
}
if s3Secret := header.Get("X-Image-Meta-S3_info_secret"); len(s3Secret) > 0 {
r.s3Info.Secret = s3Secret
}
if s3Key := header.Get("X-Image-Meta-S3_info_key"); len(s3Key) > 0 {
r.s3Info.Key = s3Key
}
if s3SignVer := header.Get("X-Image-Meta-S3_info_sign_ver"); len(s3SignVer) > 0 {
r.s3Info.SignVer = s3SignVer
}
if s3Bucket := header.Get("X-Image-Meta-S3_info_bucket"); len(s3Bucket) > 0 {
r.s3Info.Bucket = s3Bucket
}
}
}
pkg/hostman/storageman/remotefile/s3remotefile.go
+64
View File
@@ -0,0 +1,64 @@
// 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 remotefile
import (
"context"
"os"
"yunion.io/x/cloudmux/pkg/cloudprovider"
"yunion.io/x/cloudmux/pkg/multicloud/objectstore"
"yunion.io/x/log"
"yunion.io/x/pkg/errors"
)
type S3RemoteFileInfo struct {
Bucket string `json:"bucket"`
AcessKey string `json:"access_key"`
Secret string `json:"secret"`
Url string `json:"url"`
Key string `json:"key"`
SignVer string `json:"sign_ver"`
}
func (info *S3RemoteFileInfo) download(ctx context.Context, localPath string, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
log.Infof("start s3 download url: %s key: %s to %s", info.Url, info.Key, localPath)
cfg := objectstore.NewObjectStoreClientConfig(info.Url, info.AcessKey, info.Secret)
if len(info.SignVer) > 0 {
cfg.SignVersion(objectstore.S3SignVersion(info.SignVer))
}
minioClient, err := objectstore.NewObjectStoreClient(cfg)
if err != nil {
return errors.Wrap(err, "new minio client")
}
bucket, err := minioClient.GetIBucketByName(info.Bucket)
if err != nil {
return errors.Wrap(err, "get bucket")
}
fi, err := os.Create(localPath)
if err != nil {
return errors.Wrap(err, "create file")
}
defer fi.Close()
_, err = cloudprovider.DownloadObjectParallelWithProgress(ctx, bucket, info.Key, nil, fi, 0, 0, false, 10, callback)
if err != nil {
return errors.Wrap(err, "download object")
}
return nil
}
pkg/image/drivers/s3/minio.go
+14 -2
View File
@@ -19,6 +19,7 @@ import (
"fmt"
"io"
"os"
"strings"
"time"
"yunion.io/x/cloudmux/pkg/cloudprovider"
@@ -54,11 +55,22 @@ func (c *S3Client) getBucket() (cloudprovider.ICloudBucket, error) {
return c.osc.GetIBucketByName(c.bucket)
}
func GetEndpoint(endpoint string, useSSL bool) string {
if !strings.HasPrefix(endpoint, "http://") && !strings.HasPrefix(endpoint, "https://") {
prefix := "http://"
if useSSL {
prefix = "https://"
}
endpoint = prefix + endpoint
}
return endpoint
}
func Init(endpoint, accessKey, secretKey, bucket string, useSSL bool, signVer string) error {
if client != nil {
return nil
}
cfg := objectstore.NewObjectStoreClientConfig(endpoint, accessKey, secretKey)
cfg := objectstore.NewObjectStoreClientConfig(GetEndpoint(endpoint, useSSL), accessKey, secretKey)
if len(signVer) > 0 {
cfg.SignVersion(objectstore.S3SignVersion(signVer))
}
@@ -142,7 +154,7 @@ func Get(ctx context.Context, fileName string) (int64, io.ReadCloser, error) {
if err != nil {
return 0, nil, errors.Wrap(err, "bucket.ListObject")
}
if len(result.Objects) == 0 {
if len(result.Objects) == 0 || result.Objects[0].GetKey() != fileName {
return 0, nil, errors.Wrapf(errors.ErrNotFound, "no such object %s", fileName)
}
pkg/image/models/images.go
+11 -1
View File
@@ -53,6 +53,7 @@ import (
deployapi "yunion.io/x/onecloud/pkg/hostman/hostdeployer/apis"
"yunion.io/x/onecloud/pkg/hostman/hostdeployer/deployclient"
"yunion.io/x/onecloud/pkg/httperrors"
"yunion.io/x/onecloud/pkg/image/drivers/s3"
"yunion.io/x/onecloud/pkg/image/options"
"yunion.io/x/onecloud/pkg/mcclient"
"yunion.io/x/onecloud/pkg/mcclient/auth"
@@ -388,6 +389,15 @@ func (self *SImage) GetExtraDetailsHeaders(ctx context.Context, userCred mcclien
headers[fmt.Sprintf("%s%s", modules.IMAGE_META, "auto_delete_at")] = timeutils.FullIsoTime(pendingDeletedAt)
}
if strings.HasPrefix(self.Location, api.S3Prefix) {
headers[fmt.Sprintf("%s%s", modules.IMAGE_META, "s3_info_url")] = s3.GetEndpoint(options.Options.S3Endpoint, options.Options.S3UseSSL)
headers[fmt.Sprintf("%s%s", modules.IMAGE_META, "s3_info_access_key")] = options.Options.S3AccessKey
headers[fmt.Sprintf("%s%s", modules.IMAGE_META, "s3_info_secret")] = options.Options.S3SecretKey
headers[fmt.Sprintf("%s%s", modules.IMAGE_META, "s3_info_bucket")] = options.Options.S3BucketName
headers[fmt.Sprintf("%s%s", modules.IMAGE_META, "s3_info_key")] = imagePathToName(self.Location)
headers[fmt.Sprintf("%s%s", modules.IMAGE_META, "s3_info_sign_ver")] = options.Options.S3SignVersion
}
return headers
}
@@ -511,7 +521,7 @@ func (self *SImage) saveImageFromStream(localPath string, reader io.Reader, tota
}
defer fp.Close()
lastSaveTime := time.Now()
return streamutils.StreamPipe(reader, fp, calChecksum, func(saved int64) {
return streamutils.StreamPipe(reader, fp, calChecksum, func(saved int64, _ int64) {
now := time.Now()
if now.Sub(lastSaveTime) > 5*time.Second {
self.saveSize(saved, totalSize)
pkg/image/service/service.go
+2 -11
View File
@@ -19,7 +19,6 @@ import (
"fmt"
"os"
"path/filepath"
"strings"
"time"
execlient "yunion.io/x/executor/client"
@@ -195,16 +194,8 @@ func hasVmwareAccount() (bool, error) {
}
func initS3() {
url := options.Options.S3Endpoint
if !strings.HasPrefix(url, "http://") && !strings.HasPrefix(url, "https://") {
prefix := "http://"
if options.Options.S3UseSSL {
prefix = "https://"
}
url = prefix + url
}
err := s3.Init(
url,
options.Options.S3Endpoint,
options.Options.S3AccessKey,
options.Options.S3SecretKey,
options.Options.S3BucketName,
@@ -249,7 +240,7 @@ func initS3() {
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()
"-o", fmt.Sprintf("passwd_file=/tmp/s3-pass,use_path_request_style,url=%s", s3.GetEndpoint(options.Options.S3Endpoint, options.Options.S3UseSSL))).Output()
if err != nil {
log.Fatalf("failed mount s3fs %s %s", err, out)
}
vendor/modules.txt
Vendored
+2 -2
View File
@@ -1785,7 +1785,7 @@ sigs.k8s.io/structured-merge-diff/v4/value
# sigs.k8s.io/yaml v1.2.0
## explicit; go 1.12
sigs.k8s.io/yaml
# yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20250317132726-d9536ae12f8c
# yunion.io/x/cloudmux v0.3.10-0-alpha.1.0.20250321092740-6d5a1a097a47
## explicit; go 1.21
yunion.io/x/cloudmux/pkg/apis
yunion.io/x/cloudmux/pkg/apis/billing
@@ -1882,7 +1882,7 @@ yunion.io/x/log/hooks
yunion.io/x/ovsdb/cli_util
yunion.io/x/ovsdb/schema/ovn_nb
yunion.io/x/ovsdb/types
# yunion.io/x/pkg v1.10.4-0.20250128110515-2cde6f625882
# yunion.io/x/pkg v1.10.4-0.20250321091808-805c8df74399
## explicit; go 1.18
yunion.io/x/pkg/appctx
yunion.io/x/pkg/errors
vendor/yunion.io/x/cloudmux/pkg/cloudprovider/objectstore.go
Generated Vendored
+53 -5
View File
@@ -1052,7 +1052,7 @@ func (ow *sOffsetWriter) Write(p []byte) (int, error) {
}
func calculateRateMbps(sizeBytes int64, duration time.Duration) float64 {
return float64(sizeBytes*8*int64(time.Second)) / float64(duration) / 1000 / 1000
return float64(sizeBytes*8) / (float64(duration) / float64(time.Second)) / 1000 / 1000
}
type downloadPartOfMultipartJob struct {
@@ -1065,12 +1065,13 @@ type downloadPartOfMultipartJob struct {
debug bool
segSizes []int64
errs []error
callback func(saved int64, written int64)
}
func downloadPartOfMultipartWorker(wg *sync.WaitGroup, queue chan downloadPartOfMultipartJob) {
defer wg.Done()
for job := range queue {
sz, err := downloadPartOfMultipart(job.ctx, job.bucket, job.key, job.rangeOpt, job.output, job.partIndex, job.debug)
sz, err := downloadPartOfMultipart(job.ctx, job.bucket, job.key, job.rangeOpt, job.output, job.partIndex, job.debug, job.callback)
if err != nil {
job.errs = append(job.errs, err)
} else {
@@ -1079,7 +1080,7 @@ func downloadPartOfMultipartWorker(wg *sync.WaitGroup, queue chan downloadPartOf
}
}
func downloadPartOfMultipart(ctx context.Context, bucket ICloudBucket, key string, rangeOpt *SGetObjectRange, output io.Writer, partIndex int, debug bool) (int64, error) {
func downloadPartOfMultipart(ctx context.Context, bucket ICloudBucket, key string, rangeOpt *SGetObjectRange, output io.Writer, partIndex int, debug bool, callback func(saved int64, written int64)) (int64, error) {
partSize := rangeOpt.SizeBytes()
var startAt time.Time
if debug {
@@ -1091,7 +1092,7 @@ func downloadPartOfMultipart(ctx context.Context, bucket ICloudBucket, key strin
return 0, errors.Wrap(err, "bucket.GetObject")
}
defer stream.Close()
prop, err := streamutils.StreamPipe(stream, output, false, nil)
prop, err := streamutils.StreamPipe(stream, output, false, callback)
if err != nil {
return 0, errors.Wrap(err, "StreamPipe")
}
@@ -1104,6 +1105,47 @@ func downloadPartOfMultipart(ctx context.Context, bucket ICloudBucket, key strin
}
func DownloadObjectParallel(ctx context.Context, bucket ICloudBucket, key string, rangeOpt *SGetObjectRange, output io.WriterAt, outputOffset int64, blocksz int64, debug bool, parallel int) (int64, error) {
return DownloadObjectParallelWithProgress(ctx, bucket, key, rangeOpt, output, outputOffset, blocksz, debug, parallel, nil)
}
type sDownloadProgresser struct {
totalSize int64
progress int64
startTime time.Time
reportTime time.Time
callback func(progress float64, progressMbps float64, totalSizeMb int64)
}
func newDownloadProgresser(totalSize int64, callback func(progress float64, progressMbps float64, totalSizeMb int64)) *sDownloadProgresser {
return &sDownloadProgresser{
totalSize: totalSize,
startTime: time.Now(),
reportTime: time.Now(),
callback: callback,
}
}
func (p *sDownloadProgresser) Progress(_ int64, written int64) {
p.progress += written
duration := time.Since(p.startTime)
progress := float64(p.progress) / float64(p.totalSize)
progressMbps := calculateRateMbps(p.progress, duration)
if p.callback != nil {
p.callback(progress, progressMbps, p.totalSize/1000/1000)
}
if time.Since(p.reportTime) > time.Second*5 {
p.reportTime = time.Now()
log.Infof("Download progress: %d/%d, %f%%, %fMbps", p.progress, p.totalSize, progress*100, progressMbps)
}
}
func (p *sDownloadProgresser) Summary() {
duration := time.Since(p.startTime)
rateMbps := calculateRateMbps(p.progress, duration)
log.Infof("End of download %d: downloaded %d takes %f seconds at %fMbps", p.totalSize, p.progress, float64(duration)/float64(time.Second), rateMbps)
}
func DownloadObjectParallelWithProgress(ctx context.Context, bucket ICloudBucket, key string, rangeOpt *SGetObjectRange, output io.WriterAt, outputOffset int64, blocksz int64, debug bool, parallel int, callback func(progress float64, progressMbps float64, totalSizeMb int64)) (int64, error) {
obj, err := GetIObject(bucket, key)
if err != nil {
return 0, errors.Wrap(err, "GetIObject")
@@ -1131,11 +1173,16 @@ func DownloadObjectParallel(ctx context.Context, bucket ICloudBucket, key string
}
sizeBytes = rangeOpt.SizeBytes()
}
progresser := newDownloadProgresser(sizeBytes, callback)
defer progresser.Summary()
progressCallback := progresser.Progress
if sizeBytes < blocksz {
if debug {
log.Debugf("too small, download object in one shot")
}
size, err := downloadPartOfMultipart(ctx, bucket, key, rangeOpt, newWriter(output, outputOffset), 0, true)
size, err := downloadPartOfMultipart(ctx, bucket, key, rangeOpt, newWriter(output, outputOffset), 0, true, progressCallback)
if err != nil {
return 0, errors.Wrap(err, "downloadPartOfMultipart")
}
@@ -1186,6 +1233,7 @@ func DownloadObjectParallel(ctx context.Context, bucket ICloudBucket, key string
debug: debug,
segSizes: segSizes,
errs: errs,
callback: progressCallback,
}
queue <- job
}
vendor/yunion.io/x/pkg/util/streamutils/streamutils.go
Generated Vendored
+4 -4
View File
@@ -73,7 +73,7 @@ func (s *sXZReadAheadReader) Read(buf []byte) (int, error) {
}
}
func StreamPipe(upstream io.Reader, writer io.Writer, CalChecksum bool, callback func(saved int64)) (*SStreamProperty, error) {
func StreamPipe(upstream io.Reader, writer io.Writer, CalChecksum bool, callback func(savedTotal int64, savedOnce int64)) (*SStreamProperty, error) {
sp := SStreamProperty{}
var md5sum hash.Hash
@@ -103,6 +103,9 @@ func StreamPipe(upstream io.Reader, writer io.Writer, CalChecksum bool, callback
n, err := reader.Read(buf)
if n > 0 {
sp.Size += int64(n)
if callback != nil {
callback(sp.Size, int64(n))
}
if CalChecksum {
md5sum.Write(buf[:n])
}
@@ -114,9 +117,6 @@ func StreamPipe(upstream io.Reader, writer io.Writer, CalChecksum bool, callback
}
offset += m
}
if callback != nil {
callback(sp.Size)
}
}
if err != nil {
if err == io.EOF {