mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
s3: support object tagging
* GetObjectTagging * PutObjectTagging * DeleteObjectTagging
This commit is contained in:
parent
9ab98fa912
commit
f781cce500
82
test/s3/basic/object_tagging_test.go
Normal file
82
test/s3/basic/object_tagging_test.go
Normal file
|
@ -0,0 +1,82 @@
|
|||
package basic
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/service/s3"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestObjectTagging(t *testing.T) {
|
||||
|
||||
input := &s3.PutObjectInput{
|
||||
Bucket: aws.String("theBucket"),
|
||||
Key: aws.String("testDir/testObject"),
|
||||
}
|
||||
|
||||
svc.PutObject(input)
|
||||
|
||||
printTags()
|
||||
|
||||
setTags()
|
||||
|
||||
printTags()
|
||||
|
||||
clearTags()
|
||||
|
||||
printTags()
|
||||
|
||||
}
|
||||
|
||||
func printTags() {
|
||||
response, err := svc.GetObjectTagging(
|
||||
&s3.GetObjectTaggingInput{
|
||||
Bucket: aws.String("theBucket"),
|
||||
Key: aws.String("testDir/testObject"),
|
||||
})
|
||||
|
||||
fmt.Println("printTags")
|
||||
if err != nil {
|
||||
fmt.Println(err.Error())
|
||||
}
|
||||
|
||||
fmt.Println(response.TagSet)
|
||||
}
|
||||
|
||||
func setTags() {
|
||||
|
||||
response, err := svc.PutObjectTagging(&s3.PutObjectTaggingInput{
|
||||
Bucket: aws.String("theBucket"),
|
||||
Key: aws.String("testDir/testObject"),
|
||||
Tagging: &s3.Tagging{
|
||||
TagSet: []*s3.Tag{
|
||||
{
|
||||
Key: aws.String("kye2"),
|
||||
Value: aws.String("value2"),
|
||||
},
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
fmt.Println("setTags")
|
||||
if err != nil {
|
||||
fmt.Println(err.Error())
|
||||
}
|
||||
|
||||
fmt.Println(response.String())
|
||||
}
|
||||
|
||||
func clearTags() {
|
||||
|
||||
response, err := svc.DeleteObjectTagging(&s3.DeleteObjectTaggingInput{
|
||||
Bucket: aws.String("theBucket"),
|
||||
Key: aws.String("testDir/testObject"),
|
||||
})
|
||||
|
||||
fmt.Println("clearTags")
|
||||
if err != nil {
|
||||
fmt.Println(err.Error())
|
||||
}
|
||||
|
||||
fmt.Println(response.String())
|
||||
}
|
104
weed/s3api/filer_util_tags.go
Normal file
104
weed/s3api/filer_util_tags.go
Normal file
|
@ -0,0 +1,104 @@
|
|||
package s3api
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
)
|
||||
|
||||
const(
|
||||
S3TAG_PREFIX = "s3-"
|
||||
)
|
||||
|
||||
func (s3a *S3ApiServer) getTags(parentDirectoryPath string, entryName string) (tags map[string]string, err error) {
|
||||
|
||||
err = s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
||||
|
||||
resp, err := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: parentDirectoryPath,
|
||||
Name: entryName,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tags = make(map[string]string)
|
||||
for k, v := range resp.Entry.Extended {
|
||||
if strings.HasPrefix(k, S3TAG_PREFIX) {
|
||||
tags[k[len(S3TAG_PREFIX):]] = string(v)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) setTags(parentDirectoryPath string, entryName string, tags map[string]string) (err error) {
|
||||
|
||||
return s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
||||
|
||||
resp, err := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: parentDirectoryPath,
|
||||
Name: entryName,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for k, _ := range resp.Entry.Extended {
|
||||
if strings.HasPrefix(k, S3TAG_PREFIX) {
|
||||
delete(resp.Entry.Extended, k)
|
||||
}
|
||||
}
|
||||
|
||||
if resp.Entry.Extended == nil {
|
||||
resp.Entry.Extended = make(map[string][]byte)
|
||||
}
|
||||
for k, v := range tags {
|
||||
resp.Entry.Extended[S3TAG_PREFIX+k] = []byte(v)
|
||||
}
|
||||
|
||||
return filer_pb.UpdateEntry(client, &filer_pb.UpdateEntryRequest{
|
||||
Directory: parentDirectoryPath,
|
||||
Entry: resp.Entry,
|
||||
IsFromOtherCluster: false,
|
||||
Signatures: nil,
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) rmTags(parentDirectoryPath string, entryName string) (err error) {
|
||||
|
||||
return s3a.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
||||
|
||||
resp, err := filer_pb.LookupEntry(client, &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: parentDirectoryPath,
|
||||
Name: entryName,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
hasDeletion := false
|
||||
for k, _ := range resp.Entry.Extended {
|
||||
if strings.HasPrefix(k, S3TAG_PREFIX) {
|
||||
delete(resp.Entry.Extended, k)
|
||||
hasDeletion = true
|
||||
}
|
||||
}
|
||||
|
||||
if !hasDeletion {
|
||||
return nil
|
||||
}
|
||||
|
||||
return filer_pb.UpdateEntry(client, &filer_pb.UpdateEntryRequest{
|
||||
Directory: parentDirectoryPath,
|
||||
Entry: resp.Entry,
|
||||
IsFromOtherCluster: false,
|
||||
Signatures: nil,
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
}
|
117
weed/s3api/s3api_object_tagging_handlers.go
Normal file
117
weed/s3api/s3api_object_tagging_handlers.go
Normal file
|
@ -0,0 +1,117 @@
|
|||
package s3api
|
||||
|
||||
import (
|
||||
"encoding/xml"
|
||||
"fmt"
|
||||
"github.com/chrislusf/seaweedfs/weed/glog"
|
||||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/chrislusf/seaweedfs/weed/s3api/s3err"
|
||||
"github.com/chrislusf/seaweedfs/weed/util"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// GetObjectTaggingHandler - GET object tagging
|
||||
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObjectTagging.html
|
||||
func (s3a *S3ApiServer) GetObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
bucket, object := getBucketAndObject(r)
|
||||
|
||||
target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
|
||||
dir, name := target.DirAndName()
|
||||
|
||||
tags, err := s3a.getTags(dir, name)
|
||||
if err != nil {
|
||||
if err == filer_pb.ErrNotFound {
|
||||
glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
|
||||
writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
|
||||
} else {
|
||||
glog.Errorf("GetObjectTaggingHandler %s: %v", r.URL, err)
|
||||
writeErrorResponse(w, s3err.ErrInternalError, r.URL)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
writeSuccessResponseXML(w, encodeResponse(FromTags(tags)))
|
||||
|
||||
}
|
||||
|
||||
// PutObjectTaggingHandler Put object tagging
|
||||
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectTagging.html
|
||||
func (s3a *S3ApiServer) PutObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
bucket, object := getBucketAndObject(r)
|
||||
|
||||
target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
|
||||
dir, name := target.DirAndName()
|
||||
|
||||
tagging := &Tagging{}
|
||||
input, err := ioutil.ReadAll(io.LimitReader(r.Body, r.ContentLength))
|
||||
if err != nil {
|
||||
glog.Errorf("PutObjectTaggingHandler read input %s: %v", r.URL, err)
|
||||
writeErrorResponse(w, s3err.ErrInternalError, r.URL)
|
||||
return
|
||||
}
|
||||
if err = xml.Unmarshal(input, tagging); err != nil {
|
||||
glog.Errorf("PutObjectTaggingHandler Unmarshal %s: %v", r.URL, err)
|
||||
writeErrorResponse(w, s3err.ErrMalformedXML, r.URL)
|
||||
return
|
||||
}
|
||||
tags := tagging.ToTags()
|
||||
if len(tags) > 10 {
|
||||
glog.Errorf("PutObjectTaggingHandler tags %s: %d tags more than 10", r.URL, len(tags))
|
||||
writeErrorResponse(w, s3err.ErrInvalidTag, r.URL)
|
||||
return
|
||||
}
|
||||
for k, v := range tags {
|
||||
if len(k) > 128 {
|
||||
glog.Errorf("PutObjectTaggingHandler tags %s: tag key %s longer than 128", r.URL, k)
|
||||
writeErrorResponse(w, s3err.ErrInvalidTag, r.URL)
|
||||
return
|
||||
}
|
||||
if len(v) > 256 {
|
||||
glog.Errorf("PutObjectTaggingHandler tags %s: tag value %s longer than 256", r.URL, v)
|
||||
writeErrorResponse(w, s3err.ErrInvalidTag, r.URL)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if err = s3a.setTags(dir, name, tagging.ToTags()); err != nil {
|
||||
if err == filer_pb.ErrNotFound {
|
||||
glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
|
||||
writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
|
||||
} else {
|
||||
glog.Errorf("PutObjectTaggingHandler setTags %s: %v", r.URL, err)
|
||||
writeErrorResponse(w, s3err.ErrInternalError, r.URL)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
|
||||
}
|
||||
|
||||
// DeleteObjectTaggingHandler Delete object tagging
|
||||
// API reference: https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjectTagging.html
|
||||
func (s3a *S3ApiServer) DeleteObjectTaggingHandler(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
bucket, object := getBucketAndObject(r)
|
||||
|
||||
target := util.FullPath(fmt.Sprintf("%s/%s%s", s3a.option.BucketsPath, bucket, object))
|
||||
dir, name := target.DirAndName()
|
||||
|
||||
err := s3a.rmTags(dir, name)
|
||||
if err != nil {
|
||||
if err == filer_pb.ErrNotFound {
|
||||
glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err)
|
||||
writeErrorResponse(w, s3err.ErrNoSuchKey, r.URL)
|
||||
} else {
|
||||
glog.Errorf("DeleteObjectTaggingHandler %s: %v", r.URL, err)
|
||||
writeErrorResponse(w, s3err.ErrInternalError, r.URL)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
|
@ -68,6 +68,13 @@ func (s3a *S3ApiServer) registerRouter(router *mux.Router) {
|
|||
// ListMultipartUploads
|
||||
bucket.Methods("GET").HandlerFunc(track(s3a.iam.Auth(s3a.ListMultipartUploadsHandler, ACTION_WRITE), "GET")).Queries("uploads", "")
|
||||
|
||||
// GetObjectTagging
|
||||
bucket.Methods("GET").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.GetObjectTaggingHandler, ACTION_WRITE), "GET")).Queries("tagging", "")
|
||||
// PutObjectTagging
|
||||
bucket.Methods("PUT").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.PutObjectTaggingHandler, ACTION_WRITE), "PUT")).Queries("tagging", "")
|
||||
// DeleteObjectTagging
|
||||
bucket.Methods("DELETE").Path("/{object:.+}").HandlerFunc(track(s3a.iam.Auth(s3a.DeleteObjectTaggingHandler, ACTION_WRITE), "DELETE")).Queries("tagging", "")
|
||||
|
||||
// CopyObject
|
||||
bucket.Methods("PUT").Path("/{object:.+}").HeadersRegexp("X-Amz-Copy-Source", ".*?(\\/|%2F).*?").HandlerFunc(track(s3a.iam.Auth(s3a.CopyObjectHandler, ACTION_WRITE), "COPY"))
|
||||
// PutObject
|
||||
|
|
|
@ -61,6 +61,7 @@ const (
|
|||
ErrInternalError
|
||||
ErrInvalidCopyDest
|
||||
ErrInvalidCopySource
|
||||
ErrInvalidTag
|
||||
ErrAuthHeaderEmpty
|
||||
ErrSignatureVersionNotSupported
|
||||
ErrMalformedPOSTRequest
|
||||
|
@ -188,6 +189,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
|
|||
Description: "Copy Source must mention the source bucket and key: sourcebucket/sourcekey.",
|
||||
HTTPStatusCode: http.StatusBadRequest,
|
||||
},
|
||||
ErrInvalidTag: {
|
||||
Code: "InvalidArgument",
|
||||
Description: "The Tag value you have provided is invalid",
|
||||
HTTPStatusCode: http.StatusBadRequest,
|
||||
},
|
||||
ErrMalformedXML: {
|
||||
Code: "MalformedXML",
|
||||
Description: "The XML you provided was not well-formed or did not validate against our published schema.",
|
||||
|
|
38
weed/s3api/tags.go
Normal file
38
weed/s3api/tags.go
Normal file
|
@ -0,0 +1,38 @@
|
|||
package s3api
|
||||
|
||||
import (
|
||||
"encoding/xml"
|
||||
)
|
||||
|
||||
type Tag struct {
|
||||
Key string `xml:"Key"`
|
||||
Value string `xml:"Value"`
|
||||
}
|
||||
|
||||
type TagSet struct {
|
||||
Tag []Tag `xml:"Tag"`
|
||||
}
|
||||
|
||||
type Tagging struct {
|
||||
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ Tagging"`
|
||||
TagSet TagSet `xml:"TagSet"`
|
||||
}
|
||||
|
||||
func (t *Tagging) ToTags() map[string]string {
|
||||
output := make(map[string]string)
|
||||
for _, tag := range t.TagSet.Tag {
|
||||
output[tag.Key] = tag.Value
|
||||
}
|
||||
return output
|
||||
}
|
||||
|
||||
func FromTags(tags map[string]string) (t *Tagging) {
|
||||
t = &Tagging{}
|
||||
for k, v := range tags {
|
||||
t.TagSet.Tag = append(t.TagSet.Tag, Tag{
|
||||
Key: k,
|
||||
Value: v,
|
||||
})
|
||||
}
|
||||
return
|
||||
}
|
50
weed/s3api/tags_test.go
Normal file
50
weed/s3api/tags_test.go
Normal file
|
@ -0,0 +1,50 @@
|
|||
package s3api
|
||||
|
||||
import (
|
||||
"encoding/xml"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestXMLUnmarshall(t *testing.T) {
|
||||
|
||||
input := `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<Tagging xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
|
||||
<TagSet>
|
||||
<Tag>
|
||||
<Key>key1</Key>
|
||||
<Value>value1</Value>
|
||||
</Tag>
|
||||
</TagSet>
|
||||
</Tagging>
|
||||
`
|
||||
|
||||
tags := &Tagging{}
|
||||
|
||||
xml.Unmarshal([]byte(input), tags)
|
||||
|
||||
assert.Equal(t, len(tags.TagSet.Tag), 1)
|
||||
assert.Equal(t, tags.TagSet.Tag[0].Key, "key1")
|
||||
assert.Equal(t, tags.TagSet.Tag[0].Value, "value1")
|
||||
|
||||
}
|
||||
|
||||
func TestXMLMarshall(t *testing.T) {
|
||||
tags := &Tagging{
|
||||
TagSet: TagSet{
|
||||
[]Tag{
|
||||
{
|
||||
Key: "key1",
|
||||
Value: "value1",
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
actual := string(encodeResponse(tags))
|
||||
|
||||
expected := `<?xml version="1.0" encoding="UTF-8"?>
|
||||
<Tagging xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><TagSet><Tag><Key>key1</Key><Value>value1</Value></Tag></TagSet></Tagging>`
|
||||
assert.Equal(t, expected, actual)
|
||||
|
||||
}
|
Loading…
Reference in a new issue