Merge pull request #13341 from ioito/feat/qx-spare-file

feat(host): transfer sparse file over network
This commit is contained in:
Zexi Li
2022-01-27 15:33:41 +08:00
committed by GitHub
14 changed files with 645 additions and 89 deletions
+33 -43
View File
@@ -25,6 +25,9 @@ import (
"yunion.io/x/log"
"yunion.io/x/pkg/errors"
"yunion.io/x/onecloud/pkg/util/pb"
"yunion.io/x/onecloud/pkg/util/sparsefile"
)
const (
@@ -37,13 +40,14 @@ type SDownloadProvider struct {
w http.ResponseWriter
rateLimit int
compress bool
sparse bool
}
func NewDownloadProvider(w http.ResponseWriter, compress bool, rateLimit int) *SDownloadProvider {
func NewDownloadProvider(w http.ResponseWriter, compress, sparse bool, rateLimit int) *SDownloadProvider {
if rateLimit <= 0 {
rateLimit = DEFAULT_RATE_LIMIT
}
return &SDownloadProvider{w, rateLimit, compress}
return &SDownloadProvider{w: w, rateLimit: rateLimit, compress: compress, sparse: sparse}
}
func (d *SDownloadProvider) Start(
@@ -63,7 +67,7 @@ func (d *SDownloadProvider) Start(
d.w.Header().Add(k, headers.Get(k))
}
log.Infof("Downloader Start Transfer %s, compress %t", downloadFilePath, d.compress)
log.Infof("Downloader Start Transfer %s, compress %t sparse %t rateLimit: %dMiB/s", downloadFilePath, d.compress, d.sparse, d.rateLimit)
spath, err := filepath.EvalSymlinks(downloadFilePath)
if err == nil {
downloadFilePath = spath
@@ -78,16 +82,24 @@ func (d *SDownloadProvider) Start(
if err != nil {
return errors.Wrapf(err, "fi.Stat")
}
d.w.Header().Set("Content-Length", fmt.Sprintf("%d", stat.Size()))
var (
end = false
chunk = make([]byte, CHUNK_SIZE)
writer io.Writer = d.w
startTime = time.Now()
sendBytes = 0
writeChunk []byte
)
var reader io.Reader
reader = fi
size := stat.Size()
if d.sparse {
sparse, err := sparsefile.NewSparseFileReader(fi)
if err != nil {
return errors.Wrapf(err, "NewSparseFileReader")
}
size = sparse.Size()
d.w.Header().Set("X-Sparse-Header", fmt.Sprintf("%d", sparse.HeaderSize()))
reader = sparse
}
d.w.Header().Set("Content-Length", fmt.Sprintf("%d", size))
var writer io.Writer = d.w
if d.compress {
zw, err := zlib.NewWriterLevel(d.w, COMPRESS_LEVEL)
@@ -100,40 +112,18 @@ func (d *SDownloadProvider) Start(
defer zw.Flush() // it's cool
}
for !end {
size, err := fi.Read(chunk)
if err != nil {
if err != io.EOF {
log.Errorln(err)
return err
} else {
end = true
}
}
pb := pb.NewProxyReader(reader, size)
pb.SetRateLimit(d.rateLimit)
pb.SetRefreshRate(time.Second * 10)
pb.SetCallback(func() {
log.Infof("transfer %s rate: %.2f MiB p/s percent: %.2f%%", downloadFilePath, pb.Rate(), pb.Percent())
})
writeChunk = chunk[:size]
if size, err = writer.Write(writeChunk); err != nil {
log.Errorln(err)
return err
} else {
sendBytes += size
timeDur := time.Now().Sub(startTime)
exceptDur := float64(sendBytes) / 1000.0 / 1000.0 / float64(d.rateLimit)
if exceptDur > timeDur.Seconds() {
time.Sleep(time.Duration(exceptDur-timeDur.Seconds()) * time.Second)
}
}
_, err = io.Copy(writer, pb)
if err != nil {
return errors.Wrapf(err, "io.Copy")
}
// if d.compress {
// zw := writer.(*zlib.Writer)
// zw.Flush()
// }
sendMb := float64(sendBytes) / 1000.0 / 1000.0
timeDur := time.Now().Sub(startTime)
log.Infof("Send data: %fMB rate: %fMB/sec", sendMb, sendMb/timeDur.Seconds())
if onDownloadComplete != nil {
onDownloadComplete()
}
+16 -7
View File
@@ -84,6 +84,10 @@ func isCompress(r *http.Request) bool {
return r.Header.Get("X-Compress-Content") == "zlib"
}
func isSparse(r *http.Request) bool {
return r.Header.Get("X-Sparse-Content") == "true"
}
func download(ctx context.Context, w http.ResponseWriter, r *http.Request) {
var (
params, _, _ = appsrv.FetchEnv(ctx, w, r)
@@ -91,11 +95,12 @@ func download(ctx context.Context, w http.ResponseWriter, r *http.Request) {
action = params["<action>"]
rateLimit = options.HostOptions.BandwidthLimit
compress = isCompress(r)
sparse = isSparse(r)
)
switch action {
case "images":
hand := NewImageCacheDownloadProvider(w, compress, rateLimit, id)
hand := NewImageCacheDownloadProvider(w, compress, sparse, rateLimit, id)
if !fileutils2.Exists(hand.downloadFilePath()) {
httperrors.NotFoundError(ctx, w, "Image cache %s not found", id)
} else {
@@ -104,7 +109,7 @@ func download(ctx context.Context, w http.ResponseWriter, r *http.Request) {
}
}
case "servers":
hand := NewGuestDownloadProvider(w, compress, rateLimit, id)
hand := NewGuestDownloadProvider(w, compress, sparse, rateLimit, id)
if !fileutils2.Exists(hand.fullPath()) {
httperrors.NotFoundError(ctx, w, "Guest %s not found", id)
} else {
@@ -142,8 +147,9 @@ func diskDownload(ctx context.Context, w http.ResponseWriter, r *http.Request) {
hostutils.Response(ctx, w, err)
} else {
var compress = isCompress(r)
var sparse = isSparse(r)
hand := NewImageDownloadProvider(w,
compress, options.HostOptions.BandwidthLimit, disk, "")
compress, sparse, options.HostOptions.BandwidthLimit, disk, "")
if err := hand.Start(); err != nil {
hostutils.Response(ctx, w, err)
}
@@ -156,8 +162,9 @@ func diskHead(ctx context.Context, w http.ResponseWriter, r *http.Request) {
hostutils.Response(ctx, w, err)
} else {
var compress = isCompress(r)
var sparse = isSparse(r)
hand := NewImageDownloadProvider(w,
compress, options.HostOptions.BandwidthLimit, disk, "")
compress, sparse, options.HostOptions.BandwidthLimit, disk, "")
if err := hand.HandlerHead(); err != nil {
hostutils.Response(ctx, w, err)
}
@@ -187,8 +194,9 @@ func snapshotDownload(ctx context.Context, w http.ResponseWriter, r *http.Reques
hostutils.Response(ctx, w, err)
} else {
var compress = isCompress(r)
var sparse = isSparse(r)
hand := NewSnapshotDownloadProvider(w,
compress, options.HostOptions.BandwidthLimit, snapshotPath)
compress, sparse, options.HostOptions.BandwidthLimit, snapshotPath)
if err := hand.Start(); err != nil {
hostutils.Response(ctx, w, err)
}
@@ -201,8 +209,9 @@ func snapshotHead(ctx context.Context, w http.ResponseWriter, r *http.Request) {
hostutils.Response(ctx, w, err)
} else {
var compress = isCompress(r)
var sparse = isSparse(r)
hand := NewSnapshotDownloadProvider(w,
compress, options.HostOptions.BandwidthLimit, snapshotPath)
compress, sparse, options.HostOptions.BandwidthLimit, snapshotPath)
if err := hand.HandlerHead(); err != nil {
hostutils.Response(ctx, w, err)
}
@@ -215,7 +224,7 @@ func imageCacheHead(ctx context.Context, w http.ResponseWriter, r *http.Request)
rateLimit := options.HostOptions.BandwidthLimit
compress := isCompress(r)
hand := NewImageCacheDownloadProvider(w, compress, rateLimit, imageId)
hand := NewImageCacheDownloadProvider(w, compress, false, rateLimit, imageId)
if err := hand.HandlerHead(); err != nil {
hostutils.Response(ctx, w, err)
+2 -2
View File
@@ -32,10 +32,10 @@ type SGuestDownloadProvider struct {
}
func NewGuestDownloadProvider(
w http.ResponseWriter, compress bool, rateLimit int, sid string,
w http.ResponseWriter, compress, sparse bool, rateLimit int, sid string,
) *SGuestDownloadProvider {
return &SGuestDownloadProvider{
SDownloadProvider: NewDownloadProvider(w, compress, rateLimit),
SDownloadProvider: NewDownloadProvider(w, compress, sparse, rateLimit),
serverId: sid,
}
}
+2 -2
View File
@@ -33,9 +33,9 @@ type SImageDownloadProvider struct {
compressFormat string
}
func NewImageDownloadProvider(w http.ResponseWriter, compress bool, rateLimit int, disk storageman.IDisk, compressFormat string) *SImageDownloadProvider {
func NewImageDownloadProvider(w http.ResponseWriter, compress, sparse bool, rateLimit int, disk storageman.IDisk, compressFormat string) *SImageDownloadProvider {
return &SImageDownloadProvider{
SDownloadProvider: NewDownloadProvider(w, compress, rateLimit),
SDownloadProvider: NewDownloadProvider(w, compress, sparse, rateLimit),
disk: disk,
compressFormat: compressFormat,
}
@@ -30,10 +30,10 @@ type SImageCacheDownloadProvider struct {
}
func NewImageCacheDownloadProvider(
w http.ResponseWriter, compress bool, rateLimit int, imageId string,
w http.ResponseWriter, compress, sparse bool, rateLimit int, imageId string,
) *SImageCacheDownloadProvider {
return &SImageCacheDownloadProvider{
SDownloadProvider: NewDownloadProvider(w, compress, rateLimit),
SDownloadProvider: NewDownloadProvider(w, compress, sparse, rateLimit),
imageId: imageId,
}
}
@@ -26,10 +26,10 @@ type SSnapshotDownloadProvider struct {
}
func NewSnapshotDownloadProvider(
w http.ResponseWriter, compress bool, rateLimit int, snapshotPath string,
w http.ResponseWriter, compress, sparse bool, rateLimit int, snapshotPath string,
) *SSnapshotDownloadProvider {
return &SSnapshotDownloadProvider{
SDownloadProvider: NewDownloadProvider(w, compress, rateLimit),
SDownloadProvider: NewDownloadProvider(w, compress, sparse, rateLimit),
snapshotPath: snapshotPath,
}
}
+21 -31
View File
@@ -31,6 +31,8 @@ import (
"yunion.io/x/onecloud/pkg/mcclient/auth"
"yunion.io/x/onecloud/pkg/util/fileutils2"
"yunion.io/x/onecloud/pkg/util/httputils"
"yunion.io/x/onecloud/pkg/util/pb"
"yunion.io/x/onecloud/pkg/util/sparsefile"
)
type SImageDesc struct {
@@ -194,6 +196,7 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback
if r.compress {
header.Set("X-Compress-Content", "zlib")
}
header.Set("X-Sparse-Content", "true")
if len(r.extraHeaders) > 0 {
for k, v := range r.extraHeaders {
header.Set(k, v)
@@ -214,6 +217,7 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback
return errors.Wrapf(err, "request %s %s", method, url)
}
totalSize, _ := strconv.ParseInt(resp.Header.Get("Content-Length"), 10, 64)
sparseHeader, _ := strconv.ParseInt(resp.Header.Get("X-Sparse-Header"), 10, 64)
defer resp.Body.Close()
if resp.StatusCode < 300 {
if getData {
@@ -234,38 +238,24 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback
defer zlibRC.Close()
reader = zlibRC
}
var finishChan = make(chan struct{})
go func() {
defer recover()
preSizeMb := int64(0)
for {
select {
case <-time.After(1 * time.Second):
info, err := fi.Stat()
if err != nil {
log.Errorf("failed stat file %s", r.tmpPath)
return
}
percentInfo, percent := "", 0.0
if totalSize > 0 {
percent = float64(info.Size()) / float64(totalSize) * 100
percentInfo = fmt.Sprintf("(%.2f%%)", percent)
}
log.Infof("written file %s size %dM%s", r.tmpPath, info.Size()/1024/1024, percentInfo)
if callback != nil && percent > 0 {
callback(percent, float64(info.Size()-preSizeMb)/1024/1024, totalSize/1024/1024)
}
preSizeMb = info.Size()
case <-finishChan:
if callback != nil {
callback(100, 0, totalSize/1024/1024)
}
return
}
var writer io.Writer = fi
if sparseHeader > 0 {
writer = sparsefile.NewSparseFileWriter(fi, sparseHeader, totalSize)
}
pb := pb.NewProxyReader(reader, totalSize)
pb.SetCallback(func() {
if callback != nil {
go func() {
callback(pb.Percent(), pb.Rate(), totalSize/1024/1024)
}()
}
}()
_, err = io.Copy(fi, reader)
close(finishChan)
log.Infof("written file %s rate: %.2f MiB p/s percent: %.2f%%", r.tmpPath, pb.Rate(), pb.Percent())
})
_, err = io.Copy(writer, pb)
if err != nil {
return errors.Wrapf(err, "io.Copy to tmpPath %s from reader", r.tmpPath)
}
+1
View File
@@ -0,0 +1 @@
package pb // import "yunion.io/x/onecloud/pkg/util/pb"
+134
View File
@@ -0,0 +1,134 @@
// 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 pb
import (
"context"
"io"
"time"
"golang.org/x/time/rate"
)
const (
BURSTS = 1024 * 1024 * 1024
)
type SProxyReader struct {
reader io.Reader
size int64
cur int64
lastCur int64
finish chan bool
debug bool
ticker *time.Ticker
start time.Time
callback func()
ctx context.Context
refreshRate time.Duration
// Mbps
rateLimit *rate.Limiter
}
func NewProxyReader(reader io.Reader, size int64) *SProxyReader {
return &SProxyReader{
reader: reader,
size: size,
finish: make(chan bool),
ctx: context.Background(),
refreshRate: time.Second * 1,
}
}
func (self *SProxyReader) SetRateLimit(mb int) {
self.rateLimit = rate.NewLimiter(rate.Limit(mb*1024*1024), BURSTS)
self.rateLimit.AllowN(time.Now(), BURSTS)
}
// 设置刷新频率
// 仅在读取数据前生效
func (self *SProxyReader) SetRefreshRate(rate time.Duration) {
if rate > time.Second && self.cur == 0 {
self.refreshRate = rate
}
}
func (self *SProxyReader) SetCallback(callback func()) {
self.callback = callback
}
func (self *SProxyReader) Percent() float64 {
return float64(self.cur) / float64(self.size) * 100.0
}
func (self *SProxyReader) AvgRate() float64 {
return float64(self.cur) / float64(1024) / float64(1024) / float64(time.Now().Sub(self.start).Seconds())
}
func (self *SProxyReader) Rate() float64 {
return float64(self.cur-self.lastCur) / float64(1024) / float64(1024) / float64(self.refreshRate.Seconds())
}
func (self *SProxyReader) Read(p []byte) (n int, err error) {
defer func() {
if err != nil {
self.finish <- true
close(self.finish)
}
}()
if self.start.IsZero() {
self.start = time.Now()
self.ticker = time.NewTicker(self.refreshRate)
go self.refresh(self.finish)
}
n, err = self.reader.Read(p)
if err != nil {
return n, err
}
if self.rateLimit != nil {
err = self.rateLimit.WaitN(self.ctx, n)
if err != nil {
return n, err
}
}
self.cur += int64(n)
return n, nil
}
func (self *SProxyReader) refresh(finishChan chan bool) {
defer self.ticker.Stop()
for {
select {
case <-self.ticker.C:
if self.callback != nil {
self.callback()
}
self.lastCur = self.cur
case finished := <-finishChan:
if finished {
return
}
}
}
}
+1
View File
@@ -0,0 +1 @@
package sparsefile // import "yunion.io/x/onecloud/pkg/util/sparsefile"
+26
View File
@@ -0,0 +1,26 @@
// 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.
//go:build !linux
// +build !linux
package sparsefile
import (
"os"
)
func detectHoles(file *os.File) ([]sSparseHole, error) {
return []sSparseHole{}, nil
}
+60
View File
@@ -0,0 +1,60 @@
// 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.
//go:build linux
// +build linux
package sparsefile
import (
"io"
"os"
"syscall"
"golang.org/x/sys/unix"
)
const (
SEEK_DATA = 3
SEEK_HOLE = 4
)
func detectHoles(file *os.File) ([]sSparseHole, error) {
holes := []sSparseHole{}
offset := int64(0)
for {
start, err := unix.Seek(int(file.Fd()), offset, SEEK_HOLE)
if err != nil {
if e, ok := err.(syscall.Errno); ok && e == syscall.ENXIO {
break
}
return nil, err
}
end, err := unix.Seek(int(file.Fd()), start, SEEK_DATA)
if err != nil {
if e, ok := err.(syscall.Errno); ok && e == syscall.ENXIO {
end, _ = file.Seek(0, io.SeekEnd)
if end > start {
holes = append(holes, sSparseHole{Offset: start, Length: end - start})
}
break
}
return nil, err
}
offset = end
holes = append(holes, sSparseHole{Offset: start, Length: end - start})
}
return holes, nil
}
+224
View File
@@ -0,0 +1,224 @@
// 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 sparsefile
import (
"bytes"
"encoding/json"
"io"
"os"
"yunion.io/x/pkg/errors"
)
type sSparseHole struct {
Offset int64
Length int64
}
type SparseFileReader struct {
file *os.File
holes []sSparseHole
header []byte
size int64
headerSize int64
realReadLen int64
}
func (self *SparseFileReader) Close() error {
return self.file.Close()
}
func (self *SparseFileReader) HeaderSize() int64 {
return self.headerSize
}
func (self *SparseFileReader) GetHoles() []sSparseHole {
return self.holes
}
func (self *SparseFileReader) Size() int64 {
holeSize := int64(0)
for _, hole := range self.holes {
holeSize += hole.Length
}
return self.size - holeSize + self.headerSize
}
func (self *SparseFileReader) Read(p []byte) (int, error) {
if len(self.header) > 0 {
reader := bytes.NewReader(self.header)
n, err := reader.Read(p)
if err != nil {
return n, err
}
self.header = self.header[n:]
return n, nil
}
for _, hole := range self.holes {
if self.realReadLen > hole.Offset {
continue
}
if self.realReadLen < hole.Offset {
body := io.LimitReader(self.file, hole.Offset-self.realReadLen)
n, err := body.Read(p)
if err != nil {
return n, err
}
self.realReadLen += int64(n)
return n, nil
} else if self.realReadLen == hole.Offset {
n, err := self.file.Seek(hole.Length, io.SeekCurrent)
if err != nil {
return int(n), err
}
self.realReadLen += int64(n)
}
}
return self.file.Read(p)
}
func (self *SparseFileReader) probeHoles() error {
var err error
self.holes, err = detectHoles(self.file)
if err != nil {
return err
}
if len(self.holes) > 0 {
self.header, err = json.Marshal(self.holes)
if err != nil {
return errors.Wrapf(err, "json.Marshal")
}
self.headerSize = int64(len(self.header))
}
return nil
}
func NewSparseFileReader(file *os.File) (*SparseFileReader, error) {
ret := &SparseFileReader{file: file, holes: []sSparseHole{}, realReadLen: 0}
stat, err := ret.file.Stat()
if err != nil {
return nil, errors.Wrapf(err, "Stat")
}
ret.size = stat.Size()
err = ret.probeHoles()
if err != nil {
return nil, err
}
_, err = ret.file.Seek(0, io.SeekStart)
return ret, err
}
type SparseFileWrite struct {
f *os.File
headerSize int64
size int64
header []byte
holes []sSparseHole
bodyWriteLen int64
readed int
}
func (self *SparseFileWrite) Close() error {
return self.f.Close()
}
type zero struct{}
func (zero) Read(p []byte) (int, error) {
for index := range p {
p[index] = 0
}
return len(p), nil
}
func (self *SparseFileWrite) initHeader() error {
err := json.Unmarshal(self.header, &self.holes)
if err != nil {
return errors.Wrapf(err, "unmarshal header")
}
for _, h := range self.holes {
self.size += h.Length
}
return nil
}
func (self *SparseFileWrite) Write(p []byte) (int, error) {
if len(self.header) < int(self.headerSize) {
n := int(self.headerSize) - len(self.header)
if len(p) >= n {
self.header = append(self.header, p[:n]...)
err := self.initHeader()
if err != nil {
return n, err
}
self.readed = n
} else {
self.header = append(self.header, p...)
return len(p), nil
}
}
if self.readed == len(p) {
self.readed = 0
return len(p), nil
}
for _, hole := range self.holes {
if self.bodyWriteLen > hole.Offset {
continue
}
if self.bodyWriteLen < hole.Offset {
data := p[self.readed:]
if len(p[self.readed:]) > int(hole.Offset-self.bodyWriteLen) {
data = p[self.readed : hole.Offset-self.bodyWriteLen]
}
n, err := self.f.Write(data)
if err != nil {
return n, err
}
self.readed += n
self.bodyWriteLen += int64(n)
if len(p) == self.readed {
self.readed = 0
return len(p), nil
}
} else if self.bodyWriteLen == hole.Offset {
n, err := self.f.Seek(hole.Length, io.SeekCurrent)
if err != nil {
return int(n), err
}
self.bodyWriteLen += int64(n)
}
}
return self.f.Write(p)
}
func NewSparseFileWriter(f *os.File, headerSize int64, size int64) *SparseFileWrite {
return &SparseFileWrite{
f: f,
headerSize: headerSize,
header: []byte{},
holes: []sSparseHole{},
size: size,
}
}