JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

satellite/metainfo: set user_agent in bucket_metainfos on bucket recreation

Browse Source 
Before this change, if a user creates a bucket with a user_agent attributed then deletes and recreates it, the row in bucket_metainfos
will not have the user_agent. This is because we skip setting the field
in bucket_metainfos if the bucket already exists in value_attributions.
This can be problematic, as we return the bucket's user agent during the
ListBuckets operation, and the client may be expecting this value to be
populated.

This change ensures the bucket table user_agent is set when (re)creating a bucket. To avoid decreasing BeginObject performance, which also
updates attribution, a flag has been added to determine whether to
make sure the buckets table is updated: `forceBucketUpdate`.

Change-Id: Iada2f233b327b292ad9f98c73ea76a1b0113c926
This commit is contained in:
Cameron 2023-06-28 17:17:01 -04:00 committed by Storj Robot
parent 99128ab551
commit e8fcdc10a4
4 changed files with 197 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
satellite/metainfo/attribution.go
View File

@ -4,6 +4,7 @@
package metainfo
import (
"bytes"
"context"
"sync"
@ -24,9 +25,11 @@ import (
const MaxUserAgentLength = 500
// ensureAttribution ensures that the bucketName has the partner information specified by project-level user agent, or header user agent.
// If `forceBucketUpdate` is true, then the buckets table will be updated if necessary (needed for bucket creation). Otherwise, it is sufficient
// to only ensure the attribution exists in the value attributions db.
//
// Assumes that the user has permissions sufficient for authenticating.
func (endpoint *Endpoint) ensureAttribution(ctx context.Context, header *pb.RequestHeader, keyInfo *console.APIKeyInfo, bucketName, projectUserAgent []byte) (err error) {
func (endpoint *Endpoint) ensureAttribution(ctx context.Context, header *pb.RequestHeader, keyInfo *console.APIKeyInfo, bucketName, projectUserAgent []byte, forceBucketUpdate bool) (err error) {
defer mon.Task()(&ctx)(&err)
if header == nil {
@ -36,13 +39,15 @@ func (endpoint *Endpoint) ensureAttribution(ctx context.Context, header *pb.Requ
return nil
}
if conncache := drpccache.FromContext(ctx); conncache != nil {
cache := conncache.LoadOrCreate(attributionCheckCacheKey{},
func() interface{} {
return &attributionCheckCache{}
}).(*attributionCheckCache)
if !cache.needsCheck(string(bucketName)) {
return nil
if !forceBucketUpdate {
if conncache := drpccache.FromContext(ctx); conncache != nil {
cache := conncache.LoadOrCreate(attributionCheckCacheKey{},
func() interface{} {
return &attributionCheckCache{}
}).(*attributionCheckCache)
if !cache.needsCheck(string(bucketName)) {
return nil
}
}
}
@ -62,7 +67,7 @@ func (endpoint *Endpoint) ensureAttribution(ctx context.Context, header *pb.Requ
return err
}
err = endpoint.tryUpdateBucketAttribution(ctx, header, keyInfo.ProjectID, bucketName, userAgent)
err = endpoint.tryUpdateBucketAttribution(ctx, header, keyInfo.ProjectID, bucketName, userAgent, forceBucketUpdate)
if errs2.IsRPC(err, rpcstatus.NotFound) || errs2.IsRPC(err, rpcstatus.AlreadyExists) {
return nil
}
@ -110,7 +115,7 @@ func TrimUserAgent(userAgent []byte) ([]byte, error) {
return userAgent, nil
}
func (endpoint *Endpoint) tryUpdateBucketAttribution(ctx context.Context, header *pb.RequestHeader, projectID uuid.UUID, bucketName []byte, userAgent []byte) (err error) {
func (endpoint *Endpoint) tryUpdateBucketAttribution(ctx context.Context, header *pb.RequestHeader, projectID uuid.UUID, bucketName []byte, userAgent []byte, forceBucketUpdate bool) (err error) {
defer mon.Task()(&ctx)(&err)
if header == nil {
@ -118,26 +123,17 @@ func (endpoint *Endpoint) tryUpdateBucketAttribution(ctx context.Context, header
}
// check if attribution is set for given bucket
_, err = endpoint.attributions.Get(ctx, projectID, bucketName)
attrInfo, err := endpoint.attributions.Get(ctx, projectID, bucketName)
if err == nil {
// bucket has already an attribution, no need to update
return nil
}
if !attribution.ErrBucketNotAttributed.Has(err) {
// try only to set the attribution, when it's missing
if !forceBucketUpdate {
// bucket has already an attribution, no need to update
return nil
}
} else if !attribution.ErrBucketNotAttributed.Has(err) {
endpoint.log.Error("error while getting attribution from DB", zap.Error(err))
return rpcstatus.Error(rpcstatus.Internal, err.Error())
}
empty, err := endpoint.isBucketEmpty(ctx, projectID, bucketName)
if err != nil {
endpoint.log.Error("internal", zap.Error(err))
return rpcstatus.Error(rpcstatus.Internal, Error.Wrap(err).Error())
}
if !empty {
return rpcstatus.Errorf(rpcstatus.AlreadyExists, "bucket %q is not empty, Partner %q cannot be attributed", bucketName, userAgent)
}
// checks if bucket exists before updates it or makes a new entry
bucket, err := endpoint.buckets.GetBucket(ctx, bucketName, projectID)
if err != nil {
@ -147,8 +143,36 @@ func (endpoint *Endpoint) tryUpdateBucketAttribution(ctx context.Context, header
endpoint.log.Error("error while getting bucket", zap.ByteString("bucketName", bucketName), zap.Error(err))
return rpcstatus.Error(rpcstatus.Internal, "unable to set bucket attribution")
}
if bucket.UserAgent != nil {
return rpcstatus.Errorf(rpcstatus.AlreadyExists, "bucket %q already has attribution, Partner %q cannot be attributed", bucketName, userAgent)
if attrInfo != nil {
// bucket user agent and value attributions user agent already set
if bytes.Equal(bucket.UserAgent, attrInfo.UserAgent) {
return nil
}
// make sure bucket user_agent matches value_attribution
userAgent = attrInfo.UserAgent
}
empty, err := endpoint.isBucketEmpty(ctx, projectID, bucketName)
if err != nil {
endpoint.log.Error("internal", zap.Error(err))
return rpcstatus.Error(rpcstatus.Internal, Error.Wrap(err).Error())
}
if !empty {
return rpcstatus.Errorf(rpcstatus.AlreadyExists, "bucket %q is not empty, Partner %q cannot be attributed", bucketName, userAgent)
}
if attrInfo == nil {
// update attribution table
_, err = endpoint.attributions.Insert(ctx, &attribution.Info{
ProjectID: projectID,
BucketName: bucketName,
UserAgent: userAgent,
})
if err != nil {
endpoint.log.Error("error while inserting attribution to DB", zap.Error(err))
return rpcstatus.Error(rpcstatus.Internal, err.Error())
}
}
// update bucket information
@ -159,17 +183,6 @@ func (endpoint *Endpoint) tryUpdateBucketAttribution(ctx context.Context, header
return rpcstatus.Error(rpcstatus.Internal, "unable to set bucket attribution")
}
// update attribution table
_, err = endpoint.attributions.Insert(ctx, &attribution.Info{
ProjectID: projectID,
BucketName: bucketName,
UserAgent: userAgent,
})
if err != nil {
endpoint.log.Error("error while inserting attribution to DB", zap.Error(err))
return rpcstatus.Error(rpcstatus.Internal, err.Error())
}
return nil
}

143
satellite/metainfo/attribution_test.go
View File

@ -381,3 +381,146 @@ func TestBucketAttributionConcurrentUpload(t *testing.T) {
require.Equal(t, []byte(config.UserAgent), attributionInfo.UserAgent)
})
}
func TestAttributionDeletedBucketRecreated(t *testing.T) {
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1, StorageNodeCount: 0, UplinkCount: 1,
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
satellite := planet.Satellites[0]
upl := planet.Uplinks[0]
proj := upl.Projects[0].ID
bucket := "testbucket"
ua1 := []byte("minio")
ua2 := []byte("not minio")
require.NoError(t, satellite.DB.Console().Projects().UpdateUserAgent(ctx, proj, ua1))
require.NoError(t, upl.CreateBucket(ctx, satellite, bucket))
b, err := satellite.DB.Buckets().GetBucket(ctx, []byte(bucket), proj)
require.NoError(t, err)
require.Equal(t, ua1, b.UserAgent)
// test recreate with same user agent
require.NoError(t, upl.DeleteBucket(ctx, satellite, bucket))
require.NoError(t, upl.CreateBucket(ctx, satellite, bucket))
b, err = satellite.DB.Buckets().GetBucket(ctx, []byte(bucket), proj)
require.NoError(t, err)
require.Equal(t, ua1, b.UserAgent)
// test recreate with different user agent
// should still have original user agent
require.NoError(t, upl.DeleteBucket(ctx, satellite, bucket))
upl.Config.UserAgent = string(ua2)
require.NoError(t, upl.CreateBucket(ctx, satellite, bucket))
require.NoError(t, err)
require.Equal(t, ua1, b.UserAgent)
})
}
func TestAttributionBeginObject(t *testing.T) {
testplanet.Run(t, testplanet.Config{
SatelliteCount: 1, StorageNodeCount: 0, UplinkCount: 1,
}, func(t *testing.T, ctx *testcontext.Context, planet *testplanet.Planet) {
satellite := planet.Satellites[0]
upl := planet.Uplinks[0]
proj := upl.Projects[0].ID
ua := []byte("minio")
tests := []struct {
name string
vaAttrBefore, bktAttrBefore, bktAttrAfter bool
}{
// test for existence of user_agent in buckets table given the different possibilities of preconditions of user_agent
// in value_attributions and bucket_metainfos to make sure nothing breaks and outcome is expected.
{
name: "attribution exists in VA and bucket",
vaAttrBefore: true,
bktAttrBefore: true,
bktAttrAfter: true,
},
{
name: "attribution exists in VA and NOT bucket",
vaAttrBefore: true,
bktAttrBefore: false,
bktAttrAfter: false,
},
{
name: "attribution exists in bucket and NOT VA",
vaAttrBefore: false,
bktAttrBefore: true,
bktAttrAfter: true,
},
{
name: "attribution exists in neither VA nor buckets",
vaAttrBefore: false,
bktAttrBefore: false,
bktAttrAfter: true,
},
}
for i, tt := range tests {
t.Run(tt.name, func(*testing.T) {
bucketName := fmt.Sprintf("bucket-%d", i)
var expectedBktUA []byte
var config uplink.Config
if tt.bktAttrBefore || tt.vaAttrBefore {
config.UserAgent = string(ua)
}
if tt.bktAttrAfter {
expectedBktUA = ua
}
p, err := config.OpenProject(ctx, upl.Access[satellite.ID()])
require.NoError(t, err)
_, err = p.CreateBucket(ctx, bucketName)
require.NoError(t, err)
require.NoError(t, p.Close())
if !tt.bktAttrBefore && tt.vaAttrBefore {
// remove user agent from bucket
err = satellite.API.DB.Buckets().UpdateUserAgent(ctx, proj, bucketName, nil)
require.NoError(t, err)
}
_, err = satellite.API.DB.Attribution().Get(ctx, proj, []byte(bucketName))
if !tt.bktAttrBefore && !tt.vaAttrBefore {
require.Error(t, err)
} else {
require.NoError(t, err)
}
b, err := satellite.API.DB.Buckets().GetBucket(ctx, []byte(bucketName), proj)
require.NoError(t, err)
if !tt.bktAttrBefore {
require.Nil(t, b.UserAgent)
} else {
require.Equal(t, expectedBktUA, b.UserAgent)
}
config.UserAgent = string(ua)
p, err = config.OpenProject(ctx, upl.Access[satellite.ID()])
require.NoError(t, err)
upload, err := p.UploadObject(ctx, bucketName, fmt.Sprintf("foobar-%d", i), nil)
require.NoError(t, err)
_, err = upload.Write([]byte("content"))
require.NoError(t, err)
err = upload.Commit()
require.NoError(t, err)
attr, err := satellite.API.DB.Attribution().Get(ctx, proj, []byte(bucketName))
require.NoError(t, err)
require.Equal(t, ua, attr.UserAgent)
b, err = satellite.API.DB.Buckets().GetBucket(ctx, []byte(bucketName), proj)
require.NoError(t, err)
require.Equal(t, expectedBktUA, b.UserAgent)
})
}
})
}

4
satellite/metainfo/endpoint_bucket.go
View File

@ -83,7 +83,7 @@ func (endpoint *Endpoint) CreateBucket(ctx context.Context, req *pb.BucketCreate
return nil, rpcstatus.Error(rpcstatus.Internal, err.Error())
} else if exists {
// When the bucket exists, try to set the attribution.
if err := endpoint.ensureAttribution(ctx, req.Header, keyInfo, req.GetName(), nil); err != nil {
if err := endpoint.ensureAttribution(ctx, req.Header, keyInfo, req.GetName(), nil, true); err != nil {
return nil, err
}
return nil, rpcstatus.Error(rpcstatus.AlreadyExists, "bucket already exists")
@ -119,7 +119,7 @@ func (endpoint *Endpoint) CreateBucket(ctx context.Context, req *pb.BucketCreate
}
// Once we have created the bucket, we can try setting the attribution.
if err := endpoint.ensureAttribution(ctx, req.Header, keyInfo, req.GetName(), project.UserAgent); err != nil {
if err := endpoint.ensureAttribution(ctx, req.Header, keyInfo, req.GetName(), project.UserAgent, true); err != nil {
return nil, err
}

2
satellite/metainfo/endpoint_object.go
View File

@ -95,7 +95,7 @@ func (endpoint *Endpoint) BeginObject(ctx context.Context, req *pb.ObjectBeginRe
return nil, rpcstatus.Error(rpcstatus.Internal, err.Error())
}
if err := endpoint.ensureAttribution(ctx, req.Header, keyInfo, req.Bucket, nil); err != nil {
if err := endpoint.ensureAttribution(ctx, req.Header, keyInfo, req.Bucket, nil, false); err != nil {
return nil, err
}