Add API's for managing bucket quota (#9379)

This PR allows setting a "hard" or "fifo" quota
restriction at the bucket level. Buckets that
have reached the FIFO quota configured, will
automatically be cleaned up in FIFO manner until
bucket usage drops to configured quota.
If a bucket is configured with a "hard" quota
ceiling, all further writes are disallowed.

cmd/notification.go

@@ -567,6 +567,7 @@ func (sys *NotificationSys) SetBucketPolicy(ctx context.Context, bucketName stri
 func (sys *NotificationSys) DeleteBucket(ctx context.Context, bucketName string) {
 	globalNotificationSys.RemoveNotification(bucketName)
 	globalBucketObjectLockConfig.Remove(bucketName)
+	globalBucketQuotaSys.Remove(bucketName)
 	globalPolicySys.Remove(bucketName)
 	globalLifecycleSys.Remove(bucketName)
 
@@ -619,6 +620,23 @@ func (sys *NotificationSys) RemoveBucketObjectLockConfig(ctx context.Context, bu
 	}()
 }
 
+// RemoveBucketQuotaConfig - calls RemoveBucketQuotaConfig RPC call on all peers.
+func (sys *NotificationSys) RemoveBucketQuotaConfig(ctx context.Context, bucketName string) {
+	go func() {
+		ng := WithNPeers(len(sys.peerClients))
+		for idx, client := range sys.peerClients {
+			if client == nil {
+				continue
+			}
+			client := client
+			ng.Go(ctx, func() error {
+				return client.RemoveBucketQuotaConfig(bucketName)
+			}, idx, *client.host)
+		}
+		ng.Wait()
+	}()
+}
+
 // SetBucketLifecycle - calls SetBucketLifecycle on all peers.
 func (sys *NotificationSys) SetBucketLifecycle(ctx context.Context, bucketName string,
 	bucketLifecycle *lifecycle.Lifecycle) {
@@ -906,6 +924,26 @@ func (sys *NotificationSys) PutBucketObjectLockConfig(ctx context.Context, bucke
 	}
 }
 
+// PutBucketQuotaConfig - put bucket quota configuration to all peers.
+func (sys *NotificationSys) PutBucketQuotaConfig(ctx context.Context, bucketName string, q madmin.BucketQuota) {
+	g := errgroup.WithNErrs(len(sys.peerClients))
+	for index, client := range sys.peerClients {
+		if client == nil {
+			continue
+		}
+		index := index
+		g.Go(func() error {
+			return sys.peerClients[index].PutBucketQuotaConfig(bucketName, q)
+		}, index)
+	}
+	for i, err := range g.Wait() {
+		if err != nil {
+			logger.GetReqInfo(ctx).AppendTags("remotePeer", sys.peerClients[i].host.String())
+			logger.LogIf(ctx, err)
+		}
+	}
+}
+
 // NetOBDInfo - Net OBD information
 func (sys *NotificationSys) NetOBDInfo(ctx context.Context) madmin.ServerNetOBDInfo {
 	var sortedGlobalEndpoints []string