mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-28 07:50:44 +01:00
s3: answer NoSuchBucket for encryption and policy writes to a missing bucket (#12133)
* s3: answer NoSuchBucket for encryption and policy writes to a missing bucket PutBucketEncryption, DeleteBucketEncryption, and PutBucketPolicy surfaced the filer's bucket-miss as InternalError; GetBucketEncryption masked it as ServerSideEncryptionConfigurationNotFoundError. Map the miss to NoSuchBucket instead, matching AWS. Internal readers still treat a missing bucket as "no encryption configured" so uploads keep their behavior. * s3: wrap ErrBucketNotFound only for verified misses in bucket policy ops The filer lookup helpers in get/set/deleteBucketPolicy marked every LookupDirectoryEntry failure as bucket-not-found, so a transient filer outage answered NoSuchBucket for a live bucket. Route them through filer_pb.LookupEntry, which normalizes only true misses to ErrNotFound. * s3: answer NoSuchBucket when the policy delete's own write loses the bucket * s3: classify a filer NotFound on the policy UpdateEntry as ErrBucketNotFound * s3: require the bucket entry to exist on config patch transactions * s3: answer NoSuchBucket when tagging and CORS writes lose the bucket mid-request * s3: keep the existence check on no-op bucket config patches An unchanged config write produced an empty patch that early-returned before the transaction, so a bucket deleted between the read and the write still reported success. Ship the condition-only transaction so IF_EXISTS runs on its own. * s3: guard the iam step and cover failure-vs-miss classification Nil-guard s3a.iam in the bucket policy handlers so the handlers do not panic when no IAM server is wired (unit tests, embedded servers), treat a "precondition failed" response string the same as the error code when mapping a failed patch to ErrNotFound, and add handler-level tests that pin the miss/failure distinction: missing buckets and vanished mid-write buckets answer 404 NoSuchBucket while transient filer failures stay 500 InternalError.
This commit is contained in:
1 parent
4c9e62177d
commit
6536fac9bd
8 files changed
+514
-40
No files matched your search
@@ -13,6 +13,8 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
// ErrNoEncryptionConfig is returned when a bucket has no encryption configuration
|
||||
@@ -179,7 +181,7 @@ func (s3a *S3ApiServer) DeleteBucketEncryptionHandler(w http.ResponseWriter, r *
|
||||
func (s3a *S3ApiServer) GetBucketEncryptionConfig(bucket string) (*s3_pb.EncryptionConfiguration, error) {
|
||||
config, errCode := s3a.getEncryptionConfiguration(bucket)
|
||||
if errCode != s3err.ErrNone {
|
||||
if errCode == s3err.ErrNoSuchBucketEncryptionConfiguration {
|
||||
if errCode == s3err.ErrNoSuchBucketEncryptionConfiguration || errCode == s3err.ErrNoSuchBucket {
|
||||
return nil, ErrNoEncryptionConfig
|
||||
}
|
||||
return nil, fmt.Errorf("failed to get encryption configuration")
|
||||
@@ -187,6 +189,15 @@ func (s3a *S3ApiServer) GetBucketEncryptionConfig(bucket string) (*s3_pb.Encrypt
|
||||
return config, nil
|
||||
}
|
||||
|
||||
// isBucketNotFoundErr reports whether an error came from looking up a bucket
|
||||
// directory that does not exist (directly, via the negative cache, or as a gRPC
|
||||
// NotFound from the filer writing to the vanished entry).
|
||||
func isBucketNotFoundErr(err error) bool {
|
||||
return errors.Is(err, filer_pb.ErrNotFound) ||
|
||||
status.Code(err) == codes.NotFound ||
|
||||
strings.Contains(err.Error(), "bucket directory not found")
|
||||
}
|
||||
|
||||
// Internal methods following the bucket configuration pattern
|
||||
|
||||
// getEncryptionConfiguration retrieves encryption configuration with caching
|
||||
@@ -194,12 +205,8 @@ func (s3a *S3ApiServer) getEncryptionConfiguration(bucket string) (*s3_pb.Encryp
|
||||
// Get metadata using structured API
|
||||
metadata, err := s3a.GetBucketMetadata(bucket)
|
||||
if err != nil {
|
||||
// If the bucket directory is not found (e.g., during bucket recreation after
|
||||
// a partial delete), treat it as having no encryption configured rather than
|
||||
// failing the upload with an internal error.
|
||||
if errors.Is(err, filer_pb.ErrNotFound) || strings.Contains(err.Error(), "bucket directory not found") {
|
||||
glog.Warningf("getEncryptionConfiguration: bucket metadata not found for %s, treating as no encryption: %v", bucket, err)
|
||||
return nil, s3err.ErrNoSuchBucketEncryptionConfiguration
|
||||
if isBucketNotFoundErr(err) {
|
||||
return nil, s3err.ErrNoSuchBucket
|
||||
}
|
||||
glog.Errorf("getEncryptionConfiguration: failed to get bucket metadata for bucket %s: %v", bucket, err)
|
||||
return nil, s3err.ErrInternalError
|
||||
@@ -219,6 +226,9 @@ func (s3a *S3ApiServer) updateEncryptionConfiguration(bucket string, encryptionC
|
||||
// already invalidates the cache synchronously after successful update
|
||||
err := s3a.UpdateBucketEncryption(bucket, encryptionConfig)
|
||||
if err != nil {
|
||||
if isBucketNotFoundErr(err) {
|
||||
return s3err.ErrNoSuchBucket
|
||||
}
|
||||
glog.Errorf("updateEncryptionConfiguration: failed to update encryption config for bucket %s: %v", bucket, err)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
@@ -231,6 +241,9 @@ func (s3a *S3ApiServer) removeEncryptionConfiguration(bucket string) s3err.Error
|
||||
// Check if encryption configuration exists
|
||||
metadata, err := s3a.GetBucketMetadata(bucket)
|
||||
if err != nil {
|
||||
if isBucketNotFoundErr(err) {
|
||||
return s3err.ErrNoSuchBucket
|
||||
}
|
||||
glog.Errorf("removeEncryptionConfiguration: failed to get bucket metadata for bucket %s: %v", bucket, err)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
@@ -244,6 +257,9 @@ func (s3a *S3ApiServer) removeEncryptionConfiguration(bucket string) s3err.Error
|
||||
// already invalidates the cache synchronously after successful update
|
||||
err = s3a.ClearBucketEncryption(bucket)
|
||||
if err != nil {
|
||||
if isBucketNotFoundErr(err) {
|
||||
return s3err.ErrNoSuchBucket
|
||||
}
|
||||
glog.Errorf("removeEncryptionConfiguration: failed to remove encryption config for bucket %s: %v", bucket, err)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/s3_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
// Bucket-level configuration writes against a bucket that does not exist must
|
||||
// answer NoSuchBucket, not InternalError.
|
||||
func TestBucketEncryptionOpsOnMissingBucket(t *testing.T) {
|
||||
filerImpl := &ambiguousPutFiler{entries: map[string]*filer_pb.Entry{}}
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
|
||||
|
||||
if code := s3a.updateEncryptionConfiguration("missing-bucket", &s3_pb.EncryptionConfiguration{SseAlgorithm: EncryptionTypeAES256}); code != s3err.ErrNoSuchBucket {
|
||||
t.Fatalf("updateEncryptionConfiguration = %v, want %v", code, s3err.ErrNoSuchBucket)
|
||||
}
|
||||
if code := s3a.removeEncryptionConfiguration("missing-bucket"); code != s3err.ErrNoSuchBucket {
|
||||
t.Fatalf("removeEncryptionConfiguration = %v, want %v", code, s3err.ErrNoSuchBucket)
|
||||
}
|
||||
if _, code := s3a.getEncryptionConfiguration("missing-bucket"); code != s3err.ErrNoSuchBucket {
|
||||
t.Fatalf("getEncryptionConfiguration = %v, want %v", code, s3err.ErrNoSuchBucket)
|
||||
}
|
||||
|
||||
// Internal readers must keep treating a missing bucket as "no encryption".
|
||||
if _, err := s3a.GetBucketEncryptionConfig("missing-bucket"); !errors.Is(err, ErrNoEncryptionConfig) {
|
||||
t.Fatalf("GetBucketEncryptionConfig = %v, want %v", err, ErrNoEncryptionConfig)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPutBucketPolicyOnMissingBucket(t *testing.T) {
|
||||
filerImpl := &ambiguousPutFiler{entries: map[string]*filer_pb.Entry{}}
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, filerImpl))
|
||||
|
||||
policy := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3:GetObject","Resource":"arn:aws:s3:::missing-bucket/*"}]}`
|
||||
r := mux.SetURLVars(httptest.NewRequest(http.MethodPut, "/missing-bucket", strings.NewReader(policy)), map[string]string{"bucket": "missing-bucket"})
|
||||
rec := httptest.NewRecorder()
|
||||
s3a.PutBucketPolicyHandler(rec, r)
|
||||
if rec.Code != http.StatusNotFound || !strings.Contains(rec.Body.String(), "NoSuchBucket") {
|
||||
t.Fatalf("status=%d body=%s, want 404 NoSuchBucket", rec.Code, rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// A bucket deleted between the policy write's own lookup and its UpdateEntry
|
||||
// surfaces as a filer NotFound; the answer stays NoSuchBucket, not
|
||||
// InternalError.
|
||||
func TestBucketPolicyOpsOnVanishedUpdateEntry(t *testing.T) {
|
||||
const bucket = "vanished-update-bucket"
|
||||
policy := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3:GetObject","Resource":"arn:aws:s3:::vanished-update-bucket/*"}]}`
|
||||
newFiler := func() *ambiguousPutFiler {
|
||||
return &ambiguousPutFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
"/buckets/" + bucket: {
|
||||
Name: bucket,
|
||||
IsDirectory: true,
|
||||
Extended: map[string][]byte{BUCKET_POLICY_METADATA_KEY: []byte(policy)},
|
||||
},
|
||||
},
|
||||
updateErr: status.Error(codes.NotFound, "not found /buckets/"+bucket),
|
||||
updateFailKey: "/buckets/" + bucket,
|
||||
}
|
||||
}
|
||||
|
||||
t.Run("put", func(t *testing.T) {
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, newFiler()))
|
||||
r := mux.SetURLVars(httptest.NewRequest(http.MethodPut, "/"+bucket, strings.NewReader(policy)), map[string]string{"bucket": bucket})
|
||||
rec := httptest.NewRecorder()
|
||||
s3a.PutBucketPolicyHandler(rec, r)
|
||||
if rec.Code != http.StatusNotFound || !strings.Contains(rec.Body.String(), "NoSuchBucket") {
|
||||
t.Fatalf("status=%d body=%s, want 404 NoSuchBucket", rec.Code, rec.Body.String())
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("delete", func(t *testing.T) {
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, newFiler()))
|
||||
r := mux.SetURLVars(httptest.NewRequest(http.MethodDelete, "/"+bucket, nil), map[string]string{"bucket": bucket})
|
||||
rec := httptest.NewRecorder()
|
||||
s3a.DeleteBucketPolicyHandler(rec, r)
|
||||
if rec.Code != http.StatusNotFound || !strings.Contains(rec.Body.String(), "NoSuchBucket") {
|
||||
t.Fatalf("status=%d body=%s, want 404 NoSuchBucket", rec.Code, rec.Body.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// The bucket patch transaction carries an existence condition: a PATCH on an
|
||||
// absent entry is otherwise a server-side no-op, so without it a bucket
|
||||
// deleted between the metadata read and the patch would acknowledge a config
|
||||
// write that never landed. The failed condition reports NoSuchBucket.
|
||||
func TestBucketConfigWritesOnVanishedPatch(t *testing.T) {
|
||||
const bucket = "vanished-patch-bucket"
|
||||
bucketKey := "/buckets/" + bucket
|
||||
newFiler := func(sawCondition *bool) *ambiguousPutFiler {
|
||||
f := &ambiguousPutFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
bucketKey: {Name: bucket, IsDirectory: true},
|
||||
},
|
||||
}
|
||||
f.txnCheck = func(req *filer_pb.ObjectTransactionRequest) {
|
||||
for _, c := range req.GetCondition().GetClauses() {
|
||||
if c.GetKind() == filer_pb.WriteCondition_IF_EXISTS {
|
||||
*sawCondition = true
|
||||
}
|
||||
}
|
||||
delete(f.entries, bucketKey)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
t.Run("put bucket encryption", func(t *testing.T) {
|
||||
var sawCondition bool
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, newFiler(&sawCondition)))
|
||||
body := `<ServerSideEncryptionConfiguration>` +
|
||||
`<Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm>` +
|
||||
`</ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>`
|
||||
r := mux.SetURLVars(httptest.NewRequest(http.MethodPut, "/"+bucket, strings.NewReader(body)), map[string]string{"bucket": bucket})
|
||||
rec := httptest.NewRecorder()
|
||||
s3a.PutBucketEncryptionHandler(rec, r)
|
||||
if rec.Code != http.StatusNotFound || !strings.Contains(rec.Body.String(), "NoSuchBucket") {
|
||||
t.Fatalf("status=%d body=%s, want 404 NoSuchBucket", rec.Code, rec.Body.String())
|
||||
}
|
||||
if !sawCondition {
|
||||
t.Fatal("bucket patch missing the existence condition")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("update bucket config", func(t *testing.T) {
|
||||
var sawCondition bool
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, newFiler(&sawCondition)))
|
||||
if code := s3a.setBucketVersioningStatus(bucket, s3_constants.VersioningEnabled); code != s3err.ErrNoSuchBucket {
|
||||
t.Fatalf("setBucketVersioningStatus = %v, want %v", code, s3err.ErrNoSuchBucket)
|
||||
}
|
||||
if !sawCondition {
|
||||
t.Fatal("bucket patch missing the existence condition")
|
||||
}
|
||||
})
|
||||
|
||||
// Rewriting the stored status produces an empty patch; it must still carry
|
||||
// the existence condition rather than skip the transaction.
|
||||
t.Run("unchanged config value", func(t *testing.T) {
|
||||
var sawEmptyPatch bool
|
||||
f := &ambiguousPutFiler{
|
||||
entries: map[string]*filer_pb.Entry{
|
||||
bucketKey: {
|
||||
Name: bucket,
|
||||
IsDirectory: true,
|
||||
Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)},
|
||||
},
|
||||
},
|
||||
}
|
||||
f.txnCheck = func(req *filer_pb.ObjectTransactionRequest) {
|
||||
for _, c := range req.GetCondition().GetClauses() {
|
||||
if c.GetKind() == filer_pb.WriteCondition_IF_EXISTS && len(req.Mutations) == 0 {
|
||||
sawEmptyPatch = true
|
||||
}
|
||||
}
|
||||
delete(f.entries, bucketKey)
|
||||
}
|
||||
s3a := newPutTestServer(t, startFakeFiler(t, f))
|
||||
if code := s3a.setBucketVersioningStatus(bucket, s3_constants.VersioningEnabled); code != s3err.ErrNoSuchBucket {
|
||||
t.Fatalf("setBucketVersioningStatus = %v, want %v", code, s3err.ErrNoSuchBucket)
|
||||
}
|
||||
if !sawEmptyPatch {
|
||||
t.Fatal("empty patch skipped the existence-condition transaction")
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -537,6 +537,9 @@ func (s3a *S3ApiServer) updateBucketConfig(bucket string, updateFn func(*BucketC
|
||||
}
|
||||
glog.V(3).Infof("updateBucketConfig: patching %d/%d extended keys for bucket %s", len(set), len(del), bucket)
|
||||
if err := s3a.patchBucketEntry(bucket, &filer_pb.ObjectMutation{SetExtended: set, DeleteExtended: del}); err != nil {
|
||||
if isBucketNotFoundErr(err) {
|
||||
return s3err.ErrNoSuchBucket
|
||||
}
|
||||
glog.Errorf("updateBucketConfig: failed to patch bucket entry for %s: %v", bucket, err)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
@@ -582,25 +585,33 @@ func applyBucketConfigToExtended(config *BucketConfig, ext map[string][]byte) er
|
||||
// patchBucketEntry applies a field-level PATCH_EXTENDED mutation to the bucket's
|
||||
// entry via ObjectTransaction, routed to the bucket's owner filer so its per-path
|
||||
// lock serializes concurrent config writes cluster-wide rather than racing
|
||||
// whole-entry rewrites. A nil/empty mutation is a no-op.
|
||||
// whole-entry rewrites. The transaction requires the bucket entry to exist: a
|
||||
// PATCH on an absent entry is otherwise a silent no-op, which would acknowledge
|
||||
// a config write that never landed. An empty mutation still ships so the
|
||||
// existence condition runs on its own.
|
||||
func (s3a *S3ApiServer) patchBucketEntry(bucket string, m *filer_pb.ObjectMutation) error {
|
||||
if m == nil || (len(m.SetExtended) == 0 && len(m.DeleteExtended) == 0 && !m.SetContent) {
|
||||
return nil
|
||||
}
|
||||
dir := s3a.option.BucketsPath
|
||||
bucketPath := dir + "/" + bucket
|
||||
m.Type = filer_pb.ObjectMutation_PATCH_EXTENDED
|
||||
m.Directory = dir
|
||||
m.Name = bucket
|
||||
req := &filer_pb.ObjectTransactionRequest{
|
||||
LockKey: bucketPath,
|
||||
RouteKey: objectWriteRouteKeyPrefix + bucketPath,
|
||||
Mutations: []*filer_pb.ObjectMutation{m},
|
||||
Condition: clause(filer_pb.WriteCondition_IF_EXISTS),
|
||||
}
|
||||
if m != nil && (len(m.SetExtended) > 0 || len(m.DeleteExtended) > 0 || m.SetContent) {
|
||||
m.Type = filer_pb.ObjectMutation_PATCH_EXTENDED
|
||||
m.Directory = dir
|
||||
m.Name = bucket
|
||||
req.Mutations = []*filer_pb.ObjectMutation{m}
|
||||
}
|
||||
respErr := func(resp *filer_pb.ObjectTransactionResponse, err error) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// The only precondition on this request is the bucket's own existence,
|
||||
// so a failed condition is a missing bucket, not a policy violation.
|
||||
if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED || resp.Error == "precondition failed" {
|
||||
return fmt.Errorf("patch bucket %s: %w", bucket, filer_pb.ErrNotFound)
|
||||
}
|
||||
if resp.Error != "" {
|
||||
return fmt.Errorf("patch bucket %s: %s", bucket, resp.Error)
|
||||
}
|
||||
@@ -755,6 +766,9 @@ func (s3a *S3ApiServer) updateCORSConfiguration(bucket string, corsConfig *cors.
|
||||
// already invalidates the cache synchronously after successful update
|
||||
err := s3a.UpdateBucketCORS(bucket, corsConfig)
|
||||
if err != nil {
|
||||
if isBucketNotFoundErr(err) {
|
||||
return s3err.ErrNoSuchBucket
|
||||
}
|
||||
glog.Errorf("updateCORSConfiguration: failed to update CORS config for bucket %s: %v", bucket, err)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
@@ -769,6 +783,9 @@ func (s3a *S3ApiServer) removeCORSConfiguration(bucket string) s3err.ErrorCode {
|
||||
// already invalidates the cache synchronously after successful update
|
||||
err := s3a.ClearBucketCORS(bucket)
|
||||
if err != nil {
|
||||
if isBucketNotFoundErr(err) {
|
||||
return s3err.ErrNoSuchBucket
|
||||
}
|
||||
glog.Errorf("removeCORSConfiguration: failed to remove CORS config for bucket %s: %v", bucket, err)
|
||||
return s3err.ErrInternalError
|
||||
}
|
||||
|
||||
@@ -0,0 +1,203 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/gorilla/mux"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
// Bucket encryption and policy writes used to answer 500 InternalError for a
|
||||
// bucket that does not exist, because the failed filer lookup only surfaced
|
||||
// from inside the configuration update, which maps every error to an internal
|
||||
// one. S3 answers 404 NoSuchBucket.
|
||||
|
||||
func testPolicyBody(bucket string) string {
|
||||
return `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3:GetObject","Resource":"arn:aws:s3:::` + bucket + `/*"}]}`
|
||||
}
|
||||
|
||||
const testEncryptionBody = `<ServerSideEncryptionConfiguration>` +
|
||||
`<Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm>` +
|
||||
`</ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>`
|
||||
|
||||
func writeSubresource(t *testing.T, filer *fakeLookupFiler, method, bucket, body string, handler func(*S3ApiServer) http.HandlerFunc) (int, string) {
|
||||
t.Helper()
|
||||
s3a := newHeadBucketTestServer(t, filer)
|
||||
req := httptest.NewRequest(method, "/"+bucket, strings.NewReader(body))
|
||||
req = mux.SetURLVars(req, map[string]string{"bucket": bucket})
|
||||
rr := httptest.NewRecorder()
|
||||
handler(s3a)(rr, req)
|
||||
return rr.Code, rr.Body.String()
|
||||
}
|
||||
|
||||
// A missing entry is a definite answer, so every subresource write must report
|
||||
// NoSuchBucket rather than InternalError.
|
||||
func TestBucketSubresourceWritesOnMissingBucket(t *testing.T) {
|
||||
const bucket = "gone-bucket"
|
||||
|
||||
t.Run("put bucket encryption", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{}, http.MethodPut, bucket, testEncryptionBody,
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.PutBucketEncryptionHandler })
|
||||
if code != http.StatusNotFound || !strings.Contains(body, "NoSuchBucket") {
|
||||
t.Fatalf("missing bucket encryption put: %d %s, want 404 NoSuchBucket", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("delete bucket encryption", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{}, http.MethodDelete, bucket, "",
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.DeleteBucketEncryptionHandler })
|
||||
if code != http.StatusNotFound || !strings.Contains(body, "NoSuchBucket") {
|
||||
t.Fatalf("missing bucket encryption delete: %d %s, want 404 NoSuchBucket", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("put bucket policy", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{}, http.MethodPut, bucket, testPolicyBody(bucket),
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.PutBucketPolicyHandler })
|
||||
if code != http.StatusNotFound || !strings.Contains(body, "NoSuchBucket") {
|
||||
t.Fatalf("missing bucket policy put: %d %s, want 404 NoSuchBucket", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("delete bucket policy", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{}, http.MethodDelete, bucket, "",
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.DeleteBucketPolicyHandler })
|
||||
if code != http.StatusNotFound || !strings.Contains(body, "NoSuchBucket") {
|
||||
t.Fatalf("missing bucket policy delete: %d %s, want 404 NoSuchBucket", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("get bucket encryption", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{}, http.MethodGet, bucket, "",
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.GetBucketEncryptionHandler })
|
||||
if code != http.StatusNotFound || !strings.Contains(body, "NoSuchBucket") {
|
||||
t.Fatalf("missing bucket encryption get: %d %s, want 404 NoSuchBucket", code, body)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// On a live bucket with no encryption configured the answer stays
|
||||
// ServerSideEncryptionConfigurationNotFoundError: the miss classification must
|
||||
// only rewrite the missing-bucket case.
|
||||
func TestGetBucketEncryptionOnLiveBucketWithoutConfig(t *testing.T) {
|
||||
const bucket = "unencrypted-bucket"
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{
|
||||
entry: &filer_pb.Entry{Name: bucket, IsDirectory: true},
|
||||
}, http.MethodGet, bucket, "",
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.GetBucketEncryptionHandler })
|
||||
if code != http.StatusNotFound || !strings.Contains(body, "ServerSideEncryptionConfigurationNotFoundError") {
|
||||
t.Fatalf("live bucket encryption get: %d %s, want 404 ServerSideEncryptionConfigurationNotFoundError", code, body)
|
||||
}
|
||||
}
|
||||
|
||||
// A transient filer failure that arrives mid-write (after the entry's own
|
||||
// lookup passed) must still be InternalError, not a definite 404.
|
||||
func TestBucketSubresourceWritesOnFilerFailureMidRequest(t *testing.T) {
|
||||
const bucket = "flaky-bucket"
|
||||
entry := &filer_pb.Entry{Name: bucket, IsDirectory: true,
|
||||
Extended: map[string][]byte{BUCKET_POLICY_METADATA_KEY: []byte(testPolicyBody(bucket))}}
|
||||
filerError := status.Error(codes.Internal, "filer store unavailable")
|
||||
|
||||
t.Run("put bucket policy", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{entry: entry, updateErr: filerError},
|
||||
http.MethodPut, bucket, testPolicyBody(bucket),
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.PutBucketPolicyHandler })
|
||||
if code != http.StatusInternalServerError || !strings.Contains(body, "InternalError") {
|
||||
t.Fatalf("mid-write filer failure policy put: %d %s, want 500 InternalError", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("delete bucket policy", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{entry: entry, updateErr: filerError},
|
||||
http.MethodDelete, bucket, "",
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.DeleteBucketPolicyHandler })
|
||||
if code != http.StatusInternalServerError || !strings.Contains(body, "InternalError") {
|
||||
t.Fatalf("mid-write filer failure policy delete: %d %s, want 500 InternalError", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("put bucket encryption", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{entry: entry, txnErr: filerError},
|
||||
http.MethodPut, bucket, testEncryptionBody,
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.PutBucketEncryptionHandler })
|
||||
if code != http.StatusInternalServerError || !strings.Contains(body, "InternalError") {
|
||||
t.Fatalf("mid-write filer failure encryption put: %d %s, want 500 InternalError", code, body)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// A lookup that failed for another reason must not be reported as a missing
|
||||
// bucket: clients stop retrying a transient filer failure once they see a 404.
|
||||
func TestBucketSubresourceWritesOnFilerFailure(t *testing.T) {
|
||||
const bucket = "unreachable-bucket"
|
||||
filerError := status.Error(codes.Internal, "filer store unavailable")
|
||||
|
||||
t.Run("put bucket encryption", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{lookupErr: filerError}, http.MethodPut, bucket, testEncryptionBody,
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.PutBucketEncryptionHandler })
|
||||
if code != http.StatusInternalServerError || !strings.Contains(body, "InternalError") {
|
||||
t.Fatalf("filer failure encryption put: %d %s, want 500 InternalError", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("delete bucket encryption", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{lookupErr: filerError}, http.MethodDelete, bucket, "",
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.DeleteBucketEncryptionHandler })
|
||||
if code != http.StatusInternalServerError || !strings.Contains(body, "InternalError") {
|
||||
t.Fatalf("filer failure encryption delete: %d %s, want 500 InternalError", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("put bucket policy", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{lookupErr: filerError}, http.MethodPut, bucket, testPolicyBody(bucket),
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.PutBucketPolicyHandler })
|
||||
if code != http.StatusInternalServerError || !strings.Contains(body, "InternalError") {
|
||||
t.Fatalf("filer failure policy put: %d %s, want 500 InternalError", code, body)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("delete bucket policy", func(t *testing.T) {
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{lookupErr: filerError}, http.MethodDelete, bucket, "",
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.DeleteBucketPolicyHandler })
|
||||
if code != http.StatusInternalServerError || !strings.Contains(body, "InternalError") {
|
||||
t.Fatalf("filer failure policy delete: %d %s, want 500 InternalError", code, body)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// A bucket deleted between the handler's existence check and the policy
|
||||
// write's own lookup still reports NoSuchBucket: the write path classifies its
|
||||
// own miss.
|
||||
func TestDeleteBucketPolicyOnBucketVanishingMidRequest(t *testing.T) {
|
||||
const bucket = "vanishing-bucket"
|
||||
entry := &filer_pb.Entry{Name: bucket, IsDirectory: true,
|
||||
Extended: map[string][]byte{BUCKET_POLICY_METADATA_KEY: []byte(testPolicyBody(bucket))}}
|
||||
// The delete handler looks the bucket up twice (existence check, policy
|
||||
// read) before deleteBucketPolicy's own lookup fails.
|
||||
code, body := writeSubresource(t,
|
||||
&fakeLookupFiler{entry: entry, failAfter: 2, failErr: filer_pb.ErrNotFound},
|
||||
http.MethodDelete, bucket, "",
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.DeleteBucketPolicyHandler })
|
||||
if code != http.StatusNotFound || !strings.Contains(body, "NoSuchBucket") {
|
||||
t.Fatalf("vanishing bucket policy delete: %d %s, want 404 NoSuchBucket", code, body)
|
||||
}
|
||||
}
|
||||
|
||||
// A live bucket must get past the write: the fake filer applies the
|
||||
// UpdateEntry, and the handler's IAM step tolerates a nil iam.
|
||||
func TestPutBucketPolicyOnExistingBucket(t *testing.T) {
|
||||
const bucket = "live-bucket"
|
||||
code, body := writeSubresource(t, &fakeLookupFiler{
|
||||
entry: &filer_pb.Entry{Name: bucket, IsDirectory: true},
|
||||
}, http.MethodPut, bucket, testPolicyBody(bucket),
|
||||
func(s3a *S3ApiServer) http.HandlerFunc { return s3a.PutBucketPolicyHandler })
|
||||
|
||||
if code != http.StatusNoContent {
|
||||
t.Fatalf("existing bucket policy put failed: %d %s, want 204", code, body)
|
||||
}
|
||||
}
|
||||
@@ -114,8 +114,12 @@ func (s3a *S3ApiServer) PutBucketPolicyHandler(w http.ResponseWriter, r *http.Re
|
||||
|
||||
// Store bucket policy
|
||||
if err := s3a.setBucketPolicy(bucket, &policyDoc); err != nil {
|
||||
glog.Errorf("Failed to store bucket policy for %s: %v", bucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
if errors.Is(err, ErrBucketNotFound) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
|
||||
} else {
|
||||
glog.Errorf("Failed to store bucket policy for %s: %v", bucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@@ -129,7 +133,7 @@ func (s3a *S3ApiServer) PutBucketPolicyHandler(w http.ResponseWriter, r *http.Re
|
||||
}
|
||||
|
||||
// Update IAM integration with new bucket policy
|
||||
if s3a.iam.iamIntegration != nil {
|
||||
if s3a.iam != nil && s3a.iam.iamIntegration != nil {
|
||||
if err := s3a.updateBucketPolicyInIAM(bucket, &policyDoc); err != nil {
|
||||
glog.Errorf("Failed to update IAM with bucket policy: %v", err)
|
||||
// Don't fail the request, but log the warning
|
||||
@@ -171,8 +175,12 @@ func (s3a *S3ApiServer) DeleteBucketPolicyHandler(w http.ResponseWriter, r *http
|
||||
|
||||
// Delete bucket policy
|
||||
if err := s3a.deleteBucketPolicy(bucket); err != nil {
|
||||
glog.Errorf("Failed to delete bucket policy for %s: %v", bucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
if errors.Is(err, ErrBucketNotFound) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
|
||||
} else {
|
||||
glog.Errorf("Failed to delete bucket policy for %s: %v", bucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@@ -186,7 +194,7 @@ func (s3a *S3ApiServer) DeleteBucketPolicyHandler(w http.ResponseWriter, r *http
|
||||
}
|
||||
|
||||
// Update IAM integration to remove bucket policy
|
||||
if s3a.iam.iamIntegration != nil {
|
||||
if s3a.iam != nil && s3a.iam.iamIntegration != nil {
|
||||
if err := s3a.removeBucketPolicyFromIAM(bucket); err != nil {
|
||||
glog.Errorf("Failed to remove bucket policy from IAM: %v", err)
|
||||
// Don't fail the request, but log the warning
|
||||
@@ -198,23 +206,20 @@ func (s3a *S3ApiServer) DeleteBucketPolicyHandler(w http.ResponseWriter, r *http
|
||||
|
||||
// Helper functions for bucket policy storage and retrieval
|
||||
|
||||
// getBucketPolicy retrieves a bucket policy from filer metadata
|
||||
// getBucketPolicy retrieves the bucket policy from filer
|
||||
func (s3a *S3ApiServer) getBucketPolicy(bucket string) (*policy_engine.PolicyDocument, error) {
|
||||
|
||||
var policyDoc policy_engine.PolicyDocument
|
||||
err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
||||
resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
|
||||
resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: s3a.bucketRoot(bucket),
|
||||
Name: bucket,
|
||||
})
|
||||
if err != nil {
|
||||
// Return sentinel error for bucket not found
|
||||
return fmt.Errorf("%w: %v", ErrBucketNotFound, err)
|
||||
}
|
||||
|
||||
if resp.Entry == nil {
|
||||
return ErrPolicyNotFound
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
return fmt.Errorf("%w: %v", ErrBucketNotFound, err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
policyJSON, exists := resp.Entry.Extended[BUCKET_POLICY_METADATA_KEY]
|
||||
@@ -246,12 +251,15 @@ func (s3a *S3ApiServer) setBucketPolicy(bucket string, policyDoc *policy_engine.
|
||||
|
||||
return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
||||
// First, get the current entry to preserve other attributes
|
||||
resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
|
||||
resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: s3a.bucketRoot(bucket),
|
||||
Name: bucket,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("bucket not found: %v", err)
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
return fmt.Errorf("%w: %v", ErrBucketNotFound, err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
entry := resp.Entry
|
||||
@@ -262,11 +270,15 @@ func (s3a *S3ApiServer) setBucketPolicy(bucket string, policyDoc *policy_engine.
|
||||
// Set the bucket policy metadata
|
||||
entry.Extended[BUCKET_POLICY_METADATA_KEY] = policyJSON
|
||||
|
||||
// Update the entry with new metadata
|
||||
// Update the entry with new metadata; a bucket deleted between the
|
||||
// lookup above and this write surfaces as a filer NotFound.
|
||||
_, err = client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
|
||||
Directory: s3a.bucketRoot(bucket),
|
||||
Entry: entry,
|
||||
})
|
||||
if err != nil && isBucketNotFoundErr(err) {
|
||||
return fmt.Errorf("%w: %v", ErrBucketNotFound, err)
|
||||
}
|
||||
|
||||
return err
|
||||
})
|
||||
@@ -276,12 +288,15 @@ func (s3a *S3ApiServer) setBucketPolicy(bucket string, policyDoc *policy_engine.
|
||||
func (s3a *S3ApiServer) deleteBucketPolicy(bucket string) error {
|
||||
return s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
|
||||
// Get the current entry
|
||||
resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
|
||||
resp, err := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
|
||||
Directory: s3a.bucketRoot(bucket),
|
||||
Name: bucket,
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("bucket not found: %v", err)
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
return fmt.Errorf("%w: %v", ErrBucketNotFound, err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
entry := resp.Entry
|
||||
@@ -292,11 +307,15 @@ func (s3a *S3ApiServer) deleteBucketPolicy(bucket string) error {
|
||||
// Remove the bucket policy metadata
|
||||
delete(entry.Extended, BUCKET_POLICY_METADATA_KEY)
|
||||
|
||||
// Update the entry
|
||||
// Update the entry; a bucket deleted between the lookup above and this
|
||||
// write surfaces as a filer NotFound.
|
||||
_, err = client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
|
||||
Directory: s3a.bucketRoot(bucket),
|
||||
Entry: entry,
|
||||
})
|
||||
if err != nil && isBucketNotFoundErr(err) {
|
||||
return fmt.Errorf("%w: %v", ErrBucketNotFound, err)
|
||||
}
|
||||
|
||||
return err
|
||||
})
|
||||
|
||||
@@ -84,8 +84,12 @@ func (s3a *S3ApiServer) PutBucketTaggingHandler(w http.ResponseWriter, r *http.R
|
||||
|
||||
// Store bucket tags in metadata
|
||||
if err = s3a.UpdateBucketTags(bucket, tags); err != nil {
|
||||
glog.Errorf("PutBucketTagging UpdateBucketTags %s: %v", r.URL, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
if isBucketNotFoundErr(err) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
|
||||
} else {
|
||||
glog.Errorf("PutBucketTagging UpdateBucketTags %s: %v", r.URL, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@@ -105,8 +109,12 @@ func (s3a *S3ApiServer) DeleteBucketTaggingHandler(w http.ResponseWriter, r *htt
|
||||
|
||||
// Remove bucket tags from metadata
|
||||
if err := s3a.ClearBucketTags(bucket); err != nil {
|
||||
glog.Errorf("DeleteBucketTagging ClearBucketTags %s: %v", r.URL, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
if isBucketNotFoundErr(err) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
|
||||
} else {
|
||||
glog.Errorf("DeleteBucketTagging ClearBucketTags %s: %v", r.URL, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -21,10 +21,21 @@ type fakeLookupFiler struct {
|
||||
filer_pb.UnimplementedSeaweedFilerServer
|
||||
entry *filer_pb.Entry
|
||||
lookupErr error
|
||||
lookups int
|
||||
failAfter int
|
||||
failErr error
|
||||
deleted []string
|
||||
updateErr error
|
||||
txnResp *filer_pb.ObjectTransactionResponse
|
||||
txnErr error
|
||||
txnSeen []*filer_pb.ObjectTransactionRequest
|
||||
}
|
||||
|
||||
func (f *fakeLookupFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) {
|
||||
f.lookups++
|
||||
if f.failAfter > 0 && f.lookups > f.failAfter {
|
||||
return nil, f.failErr
|
||||
}
|
||||
if f.lookupErr != nil {
|
||||
return nil, f.lookupErr
|
||||
}
|
||||
@@ -37,6 +48,24 @@ func (f *fakeLookupFiler) DeleteEntry(ctx context.Context, req *filer_pb.DeleteE
|
||||
return &filer_pb.DeleteEntryResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *fakeLookupFiler) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
|
||||
if f.updateErr != nil {
|
||||
return nil, f.updateErr
|
||||
}
|
||||
return &filer_pb.UpdateEntryResponse{}, nil
|
||||
}
|
||||
|
||||
func (f *fakeLookupFiler) ObjectTransaction(ctx context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) {
|
||||
f.txnSeen = append(f.txnSeen, req)
|
||||
if f.txnErr != nil {
|
||||
return nil, f.txnErr
|
||||
}
|
||||
if f.txnResp != nil {
|
||||
return f.txnResp, nil
|
||||
}
|
||||
return &filer_pb.ObjectTransactionResponse{}, nil
|
||||
}
|
||||
|
||||
func newHeadBucketTestServer(t *testing.T, impl filer_pb.SeaweedFilerServer) *S3ApiServer {
|
||||
t.Helper()
|
||||
filers := []pb.ServerAddress{startFakeFiler(t, impl)}
|
||||
|
||||
@@ -140,6 +140,8 @@ type ambiguousPutFiler struct {
|
||||
respError string
|
||||
lookupErr error
|
||||
lookupFailKey string
|
||||
updateErr error
|
||||
updateFailKey string
|
||||
nextKey uint64
|
||||
// supersede, once, swaps in a concurrent write's entry as the routed
|
||||
// transaction commits — issue 11631's client retry. txnNoApply models a
|
||||
@@ -332,7 +334,11 @@ func (f *ambiguousPutFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream
|
||||
func (f *ambiguousPutFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.entries[req.Directory+"/"+req.Entry.Name] = proto.Clone(req.Entry).(*filer_pb.Entry)
|
||||
key := req.Directory + "/" + req.Entry.Name
|
||||
if f.updateErr != nil && key == f.updateFailKey {
|
||||
return nil, f.updateErr
|
||||
}
|
||||
f.entries[key] = proto.Clone(req.Entry).(*filer_pb.Entry)
|
||||
return &filer_pb.UpdateEntryResponse{}, nil
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user