aboutsummaryrefslogtreecommitdiff
path: root/weed
diff options
context:
space:
mode:
Diffstat (limited to 'weed')
-rw-r--r--weed/mq/client/cmd/weed_sub/subscriber.go2
-rw-r--r--weed/s3api/filer_multipart.go4
-rw-r--r--weed/s3api/s3_constants/header.go7
-rw-r--r--weed/s3api/s3api_object_handlers.go17
-rw-r--r--weed/server/filer_server.go2
-rw-r--r--weed/server/filer_server_handlers_read.go22
-rw-r--r--weed/server/volume_server_handlers.go2
-rw-r--r--weed/shell/command_volume_balance_test.go19
-rw-r--r--weed/stats/metrics_names.go1
-rw-r--r--weed/util/constants.go2
10 files changed, 64 insertions, 14 deletions
diff --git a/weed/mq/client/cmd/weed_sub/subscriber.go b/weed/mq/client/cmd/weed_sub/subscriber.go
index 6d0e49be1..adcdda04c 100644
--- a/weed/mq/client/cmd/weed_sub/subscriber.go
+++ b/weed/mq/client/cmd/weed_sub/subscriber.go
@@ -52,7 +52,7 @@ func main() {
})
subscriber.SetCompletionFunc(func() {
- glog.V(0).Infof("done recived %d messages", counter)
+ glog.V(0).Infof("done received %d messages", counter)
})
if err := subscriber.Subscribe(); err != nil {
diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go
index e9cd6a0c4..43f982897 100644
--- a/weed/s3api/filer_multipart.go
+++ b/weed/s3api/filer_multipart.go
@@ -103,7 +103,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
if len(entries) == 0 {
entryName, dirName := s3a.getEntryNameAndDir(input)
if entry, _ := s3a.getEntry(dirName, entryName); entry != nil && entry.Extended != nil {
- if uploadId, ok := entry.Extended[s3_constants.X_SeaweedFS_Header_Upload_Id]; ok && *input.UploadId == string(uploadId) {
+ if uploadId, ok := entry.Extended[s3_constants.SeaweedFSUploadId]; ok && *input.UploadId == string(uploadId) {
return &CompleteMultipartUploadResult{
CompleteMultipartUploadOutput: s3.CompleteMultipartUploadOutput{
Location: aws.String(fmt.Sprintf("http://%s%s/%s", s3a.option.Filer.ToHttpAddress(), urlEscapeObject(dirName), urlPathEscape(entryName))),
@@ -222,7 +222,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(input *s3.CompleteMultipartUploa
if entry.Extended == nil {
entry.Extended = make(map[string][]byte)
}
- entry.Extended[s3_constants.X_SeaweedFS_Header_Upload_Id] = []byte(*input.UploadId)
+ entry.Extended[s3_constants.SeaweedFSUploadId] = []byte(*input.UploadId)
for k, v := range pentry.Extended {
if k != "key" {
entry.Extended[k] = v
diff --git a/weed/s3api/s3_constants/header.go b/weed/s3api/s3_constants/header.go
index 8e4a2f8c7..765b2a0f1 100644
--- a/weed/s3api/s3_constants/header.go
+++ b/weed/s3api/s3_constants/header.go
@@ -38,8 +38,9 @@ const (
AmzObjectTaggingDirective = "X-Amz-Tagging-Directive"
AmzTagCount = "x-amz-tagging-count"
- X_SeaweedFS_Header_Directory_Key = "x-seaweedfs-is-directory-key"
- X_SeaweedFS_Header_Upload_Id = "X-Seaweedfs-Upload-Id"
+ SeaweedFSIsDirectoryKey = "X-Seaweedfs-Is-Directory-Key"
+ SeaweedFSPartNumber = "X-Seaweedfs-Part-Number"
+ SeaweedFSUploadId = "X-Seaweedfs-Upload-Id"
// S3 ACL headers
AmzCannedAcl = "X-Amz-Acl"
@@ -48,6 +49,8 @@ const (
AmzAclWrite = "X-Amz-Grant-Write"
AmzAclReadAcp = "X-Amz-Grant-Read-Acp"
AmzAclWriteAcp = "X-Amz-Grant-Write-Acp"
+
+ AmzMpPartsCount = "X-Amz-Mp-Parts-Count"
)
// Non-Standard S3 HTTP request constants
diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go
index f32dcefeb..dd3a16bf8 100644
--- a/weed/s3api/s3api_object_handlers.go
+++ b/weed/s3api/s3api_object_handlers.go
@@ -370,6 +370,9 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
if _, ok := s3_constants.PassThroughHeaders[strings.ToLower(k)]; ok {
proxyReq.Header[k] = v
}
+ if k == "partNumber" {
+ proxyReq.Header[s3_constants.SeaweedFSPartNumber] = v
+ }
}
for header, values := range r.Header {
proxyReq.Header[header] = values
@@ -411,7 +414,7 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
}
TimeToFirstByte(r.Method, start, r)
- if resp.Header.Get(s3_constants.X_SeaweedFS_Header_Directory_Key) == "true" {
+ if resp.Header.Get(s3_constants.SeaweedFSIsDirectoryKey) == "true" {
responseStatusCode := responseFn(resp, w)
s3err.PostLog(r, responseStatusCode, s3err.ErrNone)
return
@@ -429,6 +432,18 @@ func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, des
return
}
+ if resp.StatusCode == http.StatusBadRequest {
+ resp_body, _ := io.ReadAll(resp.Body)
+ switch string(resp_body) {
+ case "InvalidPart":
+ s3err.WriteErrorResponse(w, r, s3err.ErrInvalidPart)
+ default:
+ s3err.WriteErrorResponse(w, r, s3err.ErrInvalidRequest)
+ }
+ resp.Body.Close()
+ return
+ }
+
setUserMetadataKeyToLowercase(resp)
responseStatusCode := responseFn(resp, w)
diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go
index 9880afee0..ec1dff445 100644
--- a/weed/server/filer_server.go
+++ b/weed/server/filer_server.go
@@ -177,7 +177,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete")
v.SetDefault("filer.options.buckets_folder", "/buckets")
fs.filer.DirBucketsPath = v.GetString("filer.options.buckets_folder")
- // TODO deprecated, will be be removed after 2020-12-31
+ // TODO deprecated, will be removed after 2020-12-31
// replaced by https://github.com/seaweedfs/seaweedfs/wiki/Path-Specific-Configuration
// fs.filer.FsyncBuckets = v.GetStringSlice("filer.options.buckets_fsync")
isFresh := fs.filer.LoadConfiguration(v)
diff --git a/weed/server/filer_server_handlers_read.go b/weed/server/filer_server_handlers_read.go
index 83411051d..123b7a494 100644
--- a/weed/server/filer_server_handlers_read.go
+++ b/weed/server/filer_server_handlers_read.go
@@ -3,6 +3,8 @@ package weed_server
import (
"bytes"
"context"
+ "encoding/base64"
+ "encoding/hex"
"errors"
"fmt"
"io"
@@ -132,7 +134,7 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
return
}
// inform S3 API this is a user created directory key object
- w.Header().Set(s3_constants.X_SeaweedFS_Header_Directory_Key, "true")
+ w.Header().Set(s3_constants.SeaweedFSIsDirectoryKey, "true")
}
if isForDirectory && entry.Attr.Mime != s3_constants.FolderMimeType {
@@ -158,7 +160,22 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
return
}
- etag := filer.ETagEntry(entry)
+ var etag string
+ if partNumber, errNum := strconv.Atoi(r.Header.Get(s3_constants.SeaweedFSPartNumber)); errNum == nil {
+ if len(entry.Chunks) < partNumber {
+ stats.FilerHandlerCounter.WithLabelValues(stats.ErrorReadChunk).Inc()
+ w.WriteHeader(http.StatusBadRequest)
+ w.Write([]byte("InvalidPart"))
+ return
+ }
+ w.Header().Set(s3_constants.AmzMpPartsCount, strconv.Itoa(len(entry.Chunks)))
+ partChunk := entry.GetChunks()[partNumber-1]
+ md5, _ := base64.StdEncoding.DecodeString(partChunk.ETag)
+ etag = hex.EncodeToString(md5)
+ r.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", partChunk.Offset, uint64(partChunk.Offset)+partChunk.Size-1))
+ } else {
+ etag = filer.ETagEntry(entry)
+ }
w.Header().Set("Accept-Ranges", "bytes")
// mime type
@@ -207,7 +224,6 @@ func (fs *FilerServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
filename := entry.Name()
adjustPassthroughHeaders(w, r, filename)
-
totalSize := int64(entry.Size())
if r.Method == "HEAD" {
diff --git a/weed/server/volume_server_handlers.go b/weed/server/volume_server_handlers.go
index 4325b0074..85fb9ba81 100644
--- a/weed/server/volume_server_handlers.go
+++ b/weed/server/volume_server_handlers.go
@@ -97,7 +97,7 @@ func (vs *VolumeServer) privateStoreHandler(w http.ResponseWriter, r *http.Reque
}
}()
- // processs uploads
+ // processes uploads
stats.WriteRequest()
vs.guard.WhiteList(vs.PostHandler)(w, r)
diff --git a/weed/shell/command_volume_balance_test.go b/weed/shell/command_volume_balance_test.go
index d533269a4..b8af4fb98 100644
--- a/weed/shell/command_volume_balance_test.go
+++ b/weed/shell/command_volume_balance_test.go
@@ -1,10 +1,10 @@
package shell
import (
- "testing"
-
+ "fmt"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/stretchr/testify/assert"
+ "testing"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
@@ -271,3 +271,18 @@ func TestVolumeSelection(t *testing.T) {
assert.Equal(t, 378, len(vids))
}
+
+func TestDeleteEmptySelection(t *testing.T) {
+ topologyInfo := parseOutput(topoData)
+
+ eachDataNode(topologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
+ for _, diskInfo := range dn.DiskInfos {
+ for _, v := range diskInfo.VolumeInfos {
+ if v.Size <= super_block.SuperBlockSize && v.ModifiedAtSecond > 0 {
+ fmt.Printf("empty volume %d from %s\n", v.Id, dn.Id)
+ }
+ }
+ }
+ })
+
+}
diff --git a/weed/stats/metrics_names.go b/weed/stats/metrics_names.go
index c0a6e99be..15f0ad24f 100644
--- a/weed/stats/metrics_names.go
+++ b/weed/stats/metrics_names.go
@@ -41,6 +41,7 @@ const (
ErrorWriteEntry = "write.entry.failed"
RepeatErrorUploadContent = "upload.content.repeat.failed"
ErrorChunkAssign = "chunkAssign.failed"
+ ErrorReadChunk = "read.chunk.failed"
ErrorReadCache = "read.cache.failed"
ErrorReadStream = "read.stream.failed"
diff --git a/weed/util/constants.go b/weed/util/constants.go
index 94f30ffb7..d49c09d19 100644
--- a/weed/util/constants.go
+++ b/weed/util/constants.go
@@ -7,7 +7,7 @@ import (
const HttpStatusCancelled = 499
var (
- VERSION_NUMBER = fmt.Sprintf("%.02f", 3.64)
+ VERSION_NUMBER = fmt.Sprintf("%.02f", 3.65)
VERSION = sizeLimit + " " + VERSION_NUMBER
COMMIT = ""
)
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-25 07:49:18 +0000