uplog

Author: bachue

client/client.go

@@ -49,8 +49,12 @@ func SetAppName(userApp string) error {
 }
 
 func getUserAgentWithAppName(userApp string) string {
-	return fmt.Sprintf("QiniuGo/%s (%s; %s; %s) %s",
-		conf.Version, runtime.GOOS, runtime.GOARCH, userApp, runtime.Version())
+	userAgentPrefix := "QiniuGo/"
+	if testRuntime {
+		userAgentPrefix = "QiniuGo_Debug/"
+	}
+	return fmt.Sprintf("%s%s (%s; %s; %s) %s",
+		userAgentPrefix, conf.Version, runtime.GOOS, runtime.GOARCH, userApp, runtime.Version())
 }
 
 // --------------------------------------------------------------------

client/client_prod_env.go

@@ -0,0 +1,6 @@
+//go:build !unit && !integration
+// +build !unit,!integration
+
+package client
+
+const testRuntime = false

client/client_test_env.go

@@ -0,0 +1,6 @@
+//go:build unit || integration
+// +build unit integration
+
+package client
+
+const testRuntime = true

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/gofrs/flock v0.8.1
 	github.com/iancoleman/strcase v0.3.0
 	github.com/kr/pretty v0.3.0 // indirect
+	github.com/matishsiao/goInfo v0.0.0-20210923090445-da2e3fa8d45f
 	github.com/qiniu/dyn v1.3.0
 	github.com/rogpeppe/go-internal v1.8.0 // indirect
 	github.com/stretchr/testify v1.6.1

go.sum

@@ -32,6 +32,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/leodido/go-urn v1.2.1 h1:BqpAaACuzVSgi/VLzGZIobT2z4v53pjosyNd9Yv6n/w=
 github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY=
+github.com/matishsiao/goInfo v0.0.0-20210923090445-da2e3fa8d45f h1:B0OD7nYl2FPQEVrw8g2uyc1lGEzNbvrKh7fspGZcbvY=
+github.com/matishsiao/goInfo v0.0.0-20210923090445-da2e3fa8d45f/go.mod h1:aEt7p9Rvh67BYApmZwNDPpgircTO2kgdmDUoF/1QmwA=
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=

internal/clientv2/interceptor.go

@@ -5,13 +5,15 @@ import (
 )
 
 const (
-	InterceptorPriorityDefault     InterceptorPriority = 100
-	InterceptorPriorityRetryHosts  InterceptorPriority = 200
-	InterceptorPriorityRetrySimple InterceptorPriority = 300
-	InterceptorPrioritySetHeader   InterceptorPriority = 400
-	InterceptorPriorityNormal      InterceptorPriority = 500
-	InterceptorPriorityAuth        InterceptorPriority = 600
-	InterceptorPriorityDebug       InterceptorPriority = 700
+	InterceptorPriorityDefault        InterceptorPriority = 100
+	InterceptorPriorityRetryHosts     InterceptorPriority = 200
+	InterceptorPriorityRetrySimple    InterceptorPriority = 300
+	InterceptorPriorityUplog          InterceptorPriority = 310
+	InterceptorPriorityBufferResponse InterceptorPriority = 320
+	InterceptorPrioritySetHeader      InterceptorPriority = 400
+	InterceptorPriorityNormal         InterceptorPriority = 500
+	InterceptorPriorityAuth           InterceptorPriority = 600
+	InterceptorPriorityDebug          InterceptorPriority = 700
 )
 
 type InterceptorPriority int

internal/clientv2/interceptor_buffer_response.go

@@ -0,0 +1,23 @@
+package clientv2
+
+import "net/http"
+
+type bufferResponseInterceptor struct {
+}
+
+func NewBufferResponseInterceptor() Interceptor {
+	return bufferResponseInterceptor{}
+}
+
+func (interceptor bufferResponseInterceptor) Priority() InterceptorPriority {
+	return InterceptorPriorityBufferResponse
+}
+
+func (interceptor bufferResponseInterceptor) Intercept(req *http.Request, handler Handler) (resp *http.Response, err error) {
+	toBufferResponse := req.Context().Value(contextKeyBufferResponse{}) != nil
+	resp, err = handler(req)
+	if err == nil && toBufferResponse {
+		err = bufferResponse(resp)
+	}
+	return
+}

internal/clientv2/interceptor_retry_simple.go

@@ -70,10 +70,12 @@ func (c *SimpleRetryConfig) getRetryDecision(req *http.Request, resp *http.Respo
 		} else {
 			return retrier.DontRetry
 		}
-	} else if c.Retrier != nil {
-		return c.Retrier.Retry(req, resp, err, &retrier.RetrierOptions{Attempts: attempts})
 	} else {
-		return errorRetrier.Retry(req, resp, err, &retrier.RetrierOptions{Attempts: attempts})
+		r := errorRetrier
+		if c.Retrier != nil {
+			r = c.Retrier
+		}
+		return r.Retry(req, resp, err, &retrier.RetrierOptions{Attempts: attempts})
 	}
 }
 
@@ -91,7 +93,6 @@ func (interceptor *simpleRetryInterceptor) Intercept(req *http.Request, handler
 	if interceptor == nil || req == nil {
 		return interceptor.callHandler(req, &retrier.RetrierOptions{Attempts: 0}, handler)
 	}
-	toBufferResponse := req.Context().Value(contextKeyBufferResponse{}) != nil
 
 	interceptor.config.init()
 
@@ -105,10 +106,6 @@ func (interceptor *simpleRetryInterceptor) Intercept(req *http.Request, handler
 		reqBefore := cloneReq(req)
 		resp, err = interceptor.callHandler(req, &retrier.RetrierOptions{Attempts: i}, handler)
 
-		if err == nil && toBufferResponse {
-			err = bufferResponse(resp)
-		}
-
 		retryDecision := interceptor.config.getRetryDecision(reqBefore, resp, err, i)
 		if retryDecision == retrier.DontRetry {
 			interceptor.feedbackGood(req, hostname, chosenIPs)

internal/io/compatible.go

@@ -56,3 +56,7 @@ func (r *sizedReadSeekCloserFromReader) Seek(offset int64, whence int) (int64, e
 func (r *sizedReadSeekCloserFromReader) Close() error {
 	return nil
 }
+
+func (r *sizedReadSeekCloserFromReader) DetectLength() (int64, error) {
+	return r.size, nil
+}

internal/io/nopcloser.go

@@ -5,6 +5,10 @@ import (
 	"io"
 )
 
+type KnownLength interface {
+	DetectLength() (int64, error)
+}
+
 type ReadSeekableNopCloser struct {
 	r io.ReadSeeker
 }
@@ -21,6 +25,25 @@ func (nc ReadSeekableNopCloser) Seek(offset int64, whence int) (int64, error) {
 	return nc.r.Seek(offset, whence)
 }
 
+func (nc ReadSeekableNopCloser) DetectLength() (int64, error) {
+	if kl, ok := nc.r.(KnownLength); ok {
+		return kl.DetectLength()
+	}
+	cur, err := nc.r.Seek(0, io.SeekCurrent)
+	if err != nil {
+		return 0, err
+	}
+	length, err := nc.r.Seek(0, io.SeekEnd)
+	if err != nil {
+		return 0, err
+	}
+	_, err = nc.r.Seek(cur, io.SeekStart)
+	if err != nil {
+		return 0, err
+	}
+	return length, nil
+}
+
 func (nc ReadSeekableNopCloser) Close() error {
 	return nil
 }
@@ -46,6 +69,10 @@ func (nc *BytesNopCloser) Seek(offset int64, whence int) (int64, error) {
 	return nc.r.Seek(offset, whence)
 }
 
+func (nc *BytesNopCloser) DetectLength() (int64, error) {
+	return nc.Size(), nil
+}
+
 func (nc *BytesNopCloser) Size() int64 {
 	return nc.r.Size()
 }

storage/resume_uploader.go

@@ -16,6 +16,7 @@ import (
 	"github.com/qiniu/go-sdk/v7/storagev2/apis"
 	"github.com/qiniu/go-sdk/v7/storagev2/http_client"
 	"github.com/qiniu/go-sdk/v7/storagev2/region"
+	"github.com/qiniu/go-sdk/v7/storagev2/uplog"
 )
 
 // ResumeUploader 表示一个分片上传的对象
@@ -135,17 +136,22 @@ func (p *ResumeUploader) rput(ctx context.Context, ret interface{}, upToken stri
 	extra.init()
 
 	var (
+		bucket      string
 		recorderKey string
 		fileInfo    os.FileInfo = nil
 	)
 	if fileDetails != nil {
 		fileInfo = fileDetails.fileInfo
 	}
 
+	if _, bucket, err = getAkBucketFromUploadToken(upToken); err != nil {
+		return
+	}
+
 	recorderKey = getRecorderKey(extra.Recorder, upToken, key, "v1", blockSize, fileDetails)
 
 	return uploadByWorkers(
-		newResumeUploaderImpl(p, key, hasKey, upToken, makeEndpointsFromUpHost(extra.UpHost), fileInfo, extra, ret, recorderKey),
+		newResumeUploaderImpl(p, bucket, key, hasKey, upToken, makeEndpointsFromUpHost(extra.UpHost), fileInfo, extra, ret, recorderKey),
 		ctx, newSizedChunkReader(f, fsize, blockSize))
 }
 
@@ -155,8 +161,14 @@ func (p *ResumeUploader) rputWithoutSize(ctx context.Context, ret interface{}, u
 	}
 	extra.init()
 
+	var bucket string
+
+	if _, bucket, err = getAkBucketFromUploadToken(upToken); err != nil {
+		return
+	}
+
 	return uploadByWorkers(
-		newResumeUploaderImpl(p, key, hasKey, upToken, makeEndpointsFromUpHost(extra.UpHost), nil, extra, ret, ""),
+		newResumeUploaderImpl(p, bucket, key, hasKey, upToken, makeEndpointsFromUpHost(extra.UpHost), nil, extra, ret, ""),
 		ctx, newUnsizedChunkReader(r, 1<<blockBits))
 }
 
@@ -211,6 +223,7 @@ type (
 	resumeUploaderImpl struct {
 		cfg         *Config
 		storage     *apis.Storage
+		bucket      string
 		key         string
 		hasKey      bool
 		upToken     string
@@ -241,7 +254,7 @@ type (
 	}
 )
 
-func newResumeUploaderImpl(resumeUploader *ResumeUploader, key string, hasKey bool, upToken string, upEndpoints region.EndpointsProvider, fileInfo os.FileInfo, extra *RputExtra, ret interface{}, recorderKey string) *resumeUploaderImpl {
+func newResumeUploaderImpl(resumeUploader *ResumeUploader, bucket, key string, hasKey bool, upToken string, upEndpoints region.EndpointsProvider, fileInfo os.FileInfo, extra *RputExtra, ret interface{}, recorderKey string) *resumeUploaderImpl {
 	opts := http_client.Options{
 		BasicHTTPClient:     resumeUploader.Client.Client,
 		UseInsecureProtocol: !resumeUploader.Cfg.UseHTTPS,
@@ -261,6 +274,7 @@ func newResumeUploaderImpl(resumeUploader *ResumeUploader, key string, hasKey bo
 	}
 	return &resumeUploaderImpl{
 		cfg:         resumeUploader.Cfg,
+		bucket:      bucket,
 		key:         key,
 		hasKey:      hasKey,
 		upToken:     upToken,
@@ -279,17 +293,16 @@ func newResumeUploaderImpl(resumeUploader *ResumeUploader, key string, hasKey bo
 	}
 }
 
-func (impl *resumeUploaderImpl) initUploader(ctx context.Context) ([]int64, error) {
-	var recovered []int64
+func (impl *resumeUploaderImpl) initUploader(ctx context.Context) (recovered []int64, recoveredSize int64, err error) {
 	if impl.extra.Recorder != nil && len(impl.recorderKey) > 0 {
 		if recorderData, err := impl.extra.Recorder.Get(impl.recorderKey); err == nil {
-			recovered = impl.recover(ctx, recorderData)
+			recovered, recoveredSize = impl.recover(ctx, recorderData)
 			if len(recovered) == 0 {
 				impl.deleteUploadRecordIfNeed(nil, true)
 			}
 		}
 	}
-	return recovered, nil
+	return
 }
 
 func (impl *resumeUploaderImpl) uploadChunk(ctx context.Context, c chunk) error {
@@ -384,21 +397,37 @@ func (impl *resumeUploaderImpl) final(ctx context.Context) error {
 	return err
 }
 
+func (impl *resumeUploaderImpl) version() uplog.UpApiVersion {
+	return uplog.UpApiVersionV1
+}
+
+func (impl *resumeUploaderImpl) getUpToken() string {
+	return impl.upToken
+}
+
+func (impl *resumeUploaderImpl) getBucket() string {
+	return impl.bucket
+}
+
+func (impl *resumeUploaderImpl) getKey() (string, bool) {
+	return impl.key, impl.hasKey
+}
+
 func (impl *resumeUploaderImpl) deleteUploadRecordIfNeed(err error, force bool) {
 	// 无效删除之前的记录
 	if force || (isContextExpiredError(err) && impl.extra.Recorder != nil && len(impl.recorderKey) > 0) {
 		_ = impl.extra.Recorder.Delete(impl.recorderKey)
 	}
 }
 
-func (impl *resumeUploaderImpl) recover(ctx context.Context, recoverData []byte) (recovered []int64) {
+func (impl *resumeUploaderImpl) recover(ctx context.Context, recoverData []byte) (recovered []int64, recoveredSize int64) {
 	var recoveryInfo resumeUploaderRecoveryInfo
 	if err := json.Unmarshal(recoverData, &recoveryInfo); err != nil {
-		return nil
+		return
 	}
 	if impl.fileInfo == nil || recoveryInfo.FileSize != impl.fileInfo.Size() ||
 		recoveryInfo.ModTimeStamp != impl.fileInfo.ModTime().UnixNano() {
-		return nil
+		return
 	}
 	if recoveryInfo.RecorderVersion != uploadRecordVersion {
 		return
@@ -407,17 +436,18 @@ func (impl *resumeUploaderImpl) recover(ctx context.Context, recoverData []byte)
 	for _, c := range recoveryInfo.Contexts {
 		if isUploadContextExpired(c.ExpiredAt) {
 			// 有一个过期，最终其实都会无效，重传最后之前没过期的也可能会过期
-			return nil
+			return nil, 0
 		}
 
 		impl.fileSize += int64(c.ChunkSize)
+		recoveredSize += int64(c.ChunkSize)
 		impl.extra.Progresses = append(impl.extra.Progresses, BlkputRet{
 			blkIdx: c.Idx, fileOffset: c.Offset, chunkSize: c.ChunkSize, Ctx: c.Ctx, ExpiredAt: c.ExpiredAt,
 		})
 		recovered = append(recovered, c.Offset)
 	}
 
-	return recovered
+	return recovered, recoveredSize
 }
 
 func (impl *resumeUploaderImpl) save(ctx context.Context) {