s3: auto-enforce bucket quota read-only both ways (#9774)

* s3: auto-enforce bucket quota read-only both ways

Quota read-only only ever flipped when an admin re-ran
s3.bucket.quota.enforce, so a bucket that went over quota stayed
read-only forever even after usage dropped back under.

Fold enforcement into the per-minute, leader-locked bucket-size loop
the s3 gateway already runs for metrics: it now flips each bucket's
read-only flag to match its quota in both directions, rewriting
filer.conf only when a flag actually changes. The set/clear decision
lives in one shared FilerConf.ApplyBucketQuotaReadOnly helper so the
shell command and the gateway can't drift.

* only manage read-only when a quota is set, never clobber manual locks

* trim comments

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 {

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)
+}

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)
 						}
 					}
 				}

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