diff --git a/weed/filer/filer_conf.go b/weed/filer/filer_conf.go index 6ddef02e8..f80c63f63 100644 --- a/weed/filer/filer_conf.go +++ b/weed/filer/filer_conf.go @@ -233,6 +233,35 @@ func ClonePathConf(src *filer_pb.FilerConf_PathConf) *filer_pb.FilerConf_PathCon } } +// ApplyBucketQuotaReadOnly sets read-only when usedSize exceeds quota and clears it +// once back under, reporting whether the flag changed. A non-positive quota is left +// untouched so a manually locked bucket is never reopened. +func (fc *FilerConf) ApplyBucketQuotaReadOnly(locationPrefix string, usedSize, quota float64) (readOnly, changed bool) { + if quota <= 0 { + return fc.MatchStorageRule(locationPrefix).ReadOnly, false + } + + locConf := ClonePathConf(fc.MatchStorageRule(locationPrefix)) + locConf.LocationPrefix = locationPrefix + wasReadOnly := locConf.ReadOnly + + if wasReadOnly { + if usedSize < quota { + locConf.ReadOnly = false + } + } else { + if usedSize > quota { + locConf.ReadOnly = true + } + } + + if locConf.ReadOnly == wasReadOnly { + return wasReadOnly, false + } + fc.SetLocationConf(locConf) + return locConf.ReadOnly, true +} + func (fc *FilerConf) GetCollectionTtls(collection string) (ttls map[string]string) { ttls = make(map[string]string) fc.rules.Walk(func(key []byte, value *filer_pb.FilerConf_PathConf) bool { diff --git a/weed/filer/filer_conf_test.go b/weed/filer/filer_conf_test.go index 121ea7e18..e0d430a98 100644 --- a/weed/filer/filer_conf_test.go +++ b/weed/filer/filer_conf_test.go @@ -128,3 +128,36 @@ func TestClonePathConfNil(t *testing.T) { assert.NotNil(t, clone, "ClonePathConf(nil) should return a non-nil empty PathConf") assert.Equal(t, "", clone.LocationPrefix, "ClonePathConf(nil) should return empty PathConf") } + +func TestApplyBucketQuotaReadOnly(t *testing.T) { + const prefix = "/buckets/b/" + + // over quota: flips to read-only + fc := NewFilerConf() + readOnly, changed := fc.ApplyBucketQuotaReadOnly(prefix, 150, 100) + assert.True(t, changed) + assert.True(t, readOnly) + assert.True(t, fc.MatchStorageRule(prefix).ReadOnly) + + // still over quota: no change + _, changed = fc.ApplyBucketQuotaReadOnly(prefix, 150, 100) + assert.False(t, changed) + + // back under quota: flips to writable + readOnly, changed = fc.ApplyBucketQuotaReadOnly(prefix, 50, 100) + assert.True(t, changed) + assert.False(t, readOnly) + assert.False(t, fc.MatchStorageRule(prefix).ReadOnly) + + // quota disabled leaves the flag untouched, so manual locks survive + fc = NewFilerConf() + fc.ApplyBucketQuotaReadOnly(prefix, 150, 100) + readOnly, changed = fc.ApplyBucketQuotaReadOnly(prefix, 150, -1) + assert.False(t, changed) + assert.True(t, readOnly) + + // under quota and not read-only: no rule churn + fc = NewFilerConf() + _, changed = fc.ApplyBucketQuotaReadOnly(prefix, 50, 100) + assert.False(t, changed) +} diff --git a/weed/s3api/bucket_size_metrics.go b/weed/s3api/bucket_size_metrics.go index d6012275e..b6881cb4d 100644 --- a/weed/s3api/bucket_size_metrics.go +++ b/weed/s3api/bucket_size_metrics.go @@ -1,6 +1,7 @@ package s3api import ( + "bytes" "context" "fmt" "io" @@ -8,6 +9,7 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -93,7 +95,7 @@ func (s3a *S3ApiServer) collectAndUpdateBucketSizeMetrics(ctx context.Context) { } // Get list of buckets - buckets, err := s3a.listBucketNames(ctx) + buckets, err := s3a.listBuckets(ctx) if err != nil { glog.V(2).Infof("Failed to list buckets for size metrics: %v", err) return @@ -101,16 +103,61 @@ func (s3a *S3ApiServer) collectAndUpdateBucketSizeMetrics(ctx context.Context) { // Map collections to buckets and update metrics for _, bucket := range buckets { - collection := s3a.getCollectionName(bucket) + collection := s3a.getCollectionName(bucket.Name) if info, found := collectionInfos[collection]; found { - stats.UpdateBucketSizeMetrics(bucket, info.Size, info.PhysicalSize, info.FileCount) + stats.UpdateBucketSizeMetrics(bucket.Name, info.Size, info.PhysicalSize, info.FileCount) glog.V(3).Infof("Updated bucket size metrics: bucket=%s, logicalSize=%.0f, physicalSize=%.0f, objects=%.0f", - bucket, info.Size, info.PhysicalSize, info.FileCount) + bucket.Name, info.Size, info.PhysicalSize, info.FileCount) } else { // Bucket exists but no collection data (empty bucket) - stats.UpdateBucketSizeMetrics(bucket, 0, 0, 0) + stats.UpdateBucketSizeMetrics(bucket.Name, 0, 0, 0) } } + + s3a.enforceBucketQuotas(ctx, buckets, collectionInfos) +} + +// enforceBucketQuotas flips each bucket's read-only flag to match its quota, +// rewriting filer.conf only when a flag changes. +func (s3a *S3ApiServer) enforceBucketQuotas(ctx context.Context, buckets []*filer_pb.Entry, collectionInfos map[string]*CollectionInfo) { + if len(s3a.option.Filers) == 0 { + return + } + + fc, err := filer.ReadFilerConfFromFilers(s3a.option.Filers, s3a.option.GrpcDialOption, nil) + if err != nil { + glog.V(1).Infof("read filer.conf for quota enforcement: %v", err) + return + } + + changed := false + for _, bucket := range buckets { + var size float64 + if info, found := collectionInfos[s3a.getCollectionName(bucket.Name)]; found { + size = info.Size + } + locPrefix := s3a.option.BucketsPath + "/" + bucket.Name + "/" + readOnly, flipped := fc.ApplyBucketQuotaReadOnly(locPrefix, size, float64(bucket.Quota)) + if flipped { + changed = true + glog.V(0).Infof("bucket %s quota enforcement: readOnly=%v (size=%.0f quota=%d)", bucket.Name, readOnly, size, bucket.Quota) + } + } + + if !changed { + return + } + + var buf bytes.Buffer + if err := fc.ToText(&buf); err != nil { + glog.Errorf("serialize filer.conf for quota enforcement: %v", err) + return + } + if err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return filer.SaveInsideFiler(ctx, client, filer.DirectoryEtcSeaweedFS, filer.FilerConfName, buf.Bytes()) + }); err != nil { + glog.Errorf("save filer.conf for quota enforcement: %v", err) + } } // collectCollectionInfoFromMaster queries the master for topology info and extracts collection sizes. @@ -147,9 +194,9 @@ func (s3a *S3ApiServer) collectCollectionInfoFromMaster(ctx context.Context) (ma return collectionInfos, nil } -// listBucketNames returns a list of all bucket names using pagination -func (s3a *S3ApiServer) listBucketNames(ctx context.Context) ([]string, error) { - var buckets []string +// listBuckets returns all bucket directory entries using pagination. +func (s3a *S3ApiServer) listBuckets(ctx context.Context) ([]*filer_pb.Entry, error) { + var buckets []*filer_pb.Entry err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { lastFileName := "" @@ -181,7 +228,7 @@ func (s3a *S3ApiServer) listBucketNames(ctx context.Context) ([]string, error) { if resp.Entry.IsDirectory { // Skip .uploads and other hidden directories if !strings.HasPrefix(resp.Entry.Name, ".") { - buckets = append(buckets, resp.Entry.Name) + buckets = append(buckets, resp.Entry) } } } diff --git a/weed/shell/command_s3_bucket_quota_check.go b/weed/shell/command_s3_bucket_quota_check.go index 100c70f17..c72f043d5 100644 --- a/weed/shell/command_s3_bucket_quota_check.go +++ b/weed/shell/command_s3_bucket_quota_check.go @@ -105,41 +105,17 @@ func (c *commandS3BucketQuotaEnforce) Do(args []string, commandEnv *CommandEnv, func (c *commandS3BucketQuotaEnforce) processEachBucket(fc *filer.FilerConf, filerBucketsPath string, entry *filer_pb.Entry, writer io.Writer, collectionSize float64) (hasConfChanges bool) { locPrefix := filerBucketsPath + "/" + entry.Name + "/" - existingConf := fc.MatchStorageRule(locPrefix) - - // Create a mutable copy for modification - locConf := filer.ClonePathConf(existingConf) - locConf.LocationPrefix = locPrefix - - if entry.Quota > 0 { - if locConf.ReadOnly { - if collectionSize < float64(entry.Quota) { - locConf.ReadOnly = false - hasConfChanges = true - } - } else { - if collectionSize > float64(entry.Quota) { - locConf.ReadOnly = true - hasConfChanges = true - } - } - } else { - if locConf.ReadOnly { - locConf.ReadOnly = false - hasConfChanges = true - } - } + readOnly, hasConfChanges := fc.ApplyBucketQuotaReadOnly(locPrefix, collectionSize, float64(entry.Quota)) if hasConfChanges { fmt.Fprintf(writer, " %s\tsize:%.0f", entry.Name, collectionSize) fmt.Fprintf(writer, "\tquota:%d\tusage:%.2f%%", entry.Quota, collectionSize*100/float64(entry.Quota)) fmt.Fprintln(writer) - if locConf.ReadOnly { + if readOnly { fmt.Fprintf(writer, " changing bucket %s to read only!\n", entry.Name) } else { fmt.Fprintf(writer, " changing bucket %s to writable.\n", entry.Name) } - fc.SetLocationConf(locConf) } return