mirror of
https://github.com/minio/minio.git
synced 2025-11-07 21:02:58 -05:00
migrate all bucket metadata into a single file (#9586)
this is a major overhaul by migrating off all bucket metadata related configs into a single object '.metadata.bin' this allows us for faster bootups across 1000's of buckets and as well as keeps the code simple enough for future work and additions. Additionally also fixes #9396, #9394
This commit is contained in:
Harshavardhana
GitHub
@@ -20,11 +20,10 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/url"
|
||||
"path"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -34,9 +33,6 @@ import (
|
||||
"github.com/minio/minio-go/v6/pkg/set"
|
||||
"github.com/minio/minio/cmd/crypto"
|
||||
"github.com/minio/minio/cmd/logger"
|
||||
bucketsse "github.com/minio/minio/pkg/bucket/encryption"
|
||||
"github.com/minio/minio/pkg/bucket/lifecycle"
|
||||
objectlock "github.com/minio/minio/pkg/bucket/object/lock"
|
||||
"github.com/minio/minio/pkg/bucket/policy"
|
||||
"github.com/minio/minio/pkg/event"
|
||||
"github.com/minio/minio/pkg/madmin"
|
||||
@@ -546,182 +542,46 @@ func (sys *NotificationSys) GetLocks(ctx context.Context) []*PeerLocks {
|
||||
return locksResp
|
||||
}
|
||||
|
||||
// SetBucketPolicy - calls SetBucketPolicy RPC call on all peers.
|
||||
func (sys *NotificationSys) SetBucketPolicy(ctx context.Context, bucketName string, bucketPolicy *policy.Policy) {
|
||||
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.SetBucketPolicy(bucketName, bucketPolicy)
|
||||
}, idx, *client.host)
|
||||
// LoadBucketMetadata - calls LoadBucketMetadata call on all peers
|
||||
func (sys *NotificationSys) LoadBucketMetadata(ctx context.Context, bucketName string) {
|
||||
ng := WithNPeers(len(sys.peerClients))
|
||||
for idx, client := range sys.peerClients {
|
||||
if client == nil {
|
||||
continue
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
client := client
|
||||
ng.Go(ctx, func() error {
|
||||
return client.LoadBucketMetadata(bucketName)
|
||||
}, idx, *client.host)
|
||||
}
|
||||
for _, nErr := range ng.Wait() {
|
||||
reqInfo := (&logger.ReqInfo{}).AppendTags("peerAddress", nErr.Host.String())
|
||||
if nErr.Err != nil {
|
||||
logger.LogIf(logger.SetReqInfo(ctx, reqInfo), nErr.Err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// DeleteBucket - calls DeleteBucket RPC call on all peers.
|
||||
func (sys *NotificationSys) DeleteBucket(ctx context.Context, bucketName string) {
|
||||
globalNotificationSys.RemoveNotification(bucketName)
|
||||
globalBucketObjectLockSys.Remove(bucketName)
|
||||
globalBucketQuotaSys.Remove(bucketName)
|
||||
globalPolicySys.Remove(bucketName)
|
||||
globalLifecycleSys.Remove(bucketName)
|
||||
// DeleteBucketMetadata - calls DeleteBucketMetadata call on all peers
|
||||
func (sys *NotificationSys) DeleteBucketMetadata(ctx context.Context, bucketName string) {
|
||||
globalBucketMetadataSys.Remove(bucketName)
|
||||
|
||||
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.DeleteBucket(bucketName)
|
||||
}, idx, *client.host)
|
||||
ng := WithNPeers(len(sys.peerClients))
|
||||
for idx, client := range sys.peerClients {
|
||||
if client == nil {
|
||||
continue
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
}
|
||||
|
||||
// RemoveBucketPolicy - calls RemoveBucketPolicy RPC call on all peers.
|
||||
func (sys *NotificationSys) RemoveBucketPolicy(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.RemoveBucketPolicy(bucketName)
|
||||
}, idx, *client.host)
|
||||
client := client
|
||||
ng.Go(ctx, func() error {
|
||||
return client.DeleteBucketMetadata(bucketName)
|
||||
}, idx, *client.host)
|
||||
}
|
||||
for _, nErr := range ng.Wait() {
|
||||
reqInfo := (&logger.ReqInfo{}).AppendTags("peerAddress", nErr.Host.String())
|
||||
if nErr.Err != nil {
|
||||
logger.LogIf(logger.SetReqInfo(ctx, reqInfo), nErr.Err)
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
}
|
||||
|
||||
// RemoveBucketObjectLockConfig - calls RemoveBucketObjectLockConfig RPC call on all peers.
|
||||
func (sys *NotificationSys) RemoveBucketObjectLockConfig(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.RemoveBucketObjectLockConfig(bucketName)
|
||||
}, idx, *client.host)
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
}
|
||||
|
||||
// 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) {
|
||||
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.SetBucketLifecycle(bucketName, bucketLifecycle)
|
||||
}, idx, *client.host)
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
}
|
||||
|
||||
// RemoveBucketLifecycle - calls RemoveLifecycle on all peers.
|
||||
func (sys *NotificationSys) RemoveBucketLifecycle(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.RemoveBucketLifecycle(bucketName)
|
||||
}, idx, *client.host)
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
}
|
||||
|
||||
// SetBucketSSEConfig - calls SetBucketSSEConfig on all peers.
|
||||
func (sys *NotificationSys) SetBucketSSEConfig(ctx context.Context, bucketName string,
|
||||
encConfig *bucketsse.BucketSSEConfig) {
|
||||
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.SetBucketSSEConfig(bucketName, encConfig)
|
||||
}, idx, *client.host)
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
}
|
||||
|
||||
// RemoveBucketSSEConfig - calls RemoveBucketSSEConfig on all peers.
|
||||
func (sys *NotificationSys) RemoveBucketSSEConfig(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.RemoveBucketSSEConfig(bucketName)
|
||||
}, idx, *client.host)
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
}
|
||||
|
||||
// PutBucketNotification - calls PutBucketNotification RPC call on all peers.
|
||||
func (sys *NotificationSys) PutBucketNotification(ctx context.Context, bucketName string, rulesMap event.RulesMap) {
|
||||
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.PutBucketNotification(bucketName, rulesMap)
|
||||
}, idx, *client.host)
|
||||
}
|
||||
ng.Wait()
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// AddRemoteTarget - adds event rules map, HTTP/PeerRPC client target to bucket name.
|
||||
@@ -753,14 +613,18 @@ func (sys *NotificationSys) AddRemoteTarget(bucketName string, target event.Targ
|
||||
func (sys *NotificationSys) load(buckets []BucketInfo, objAPI ObjectLayer) error {
|
||||
for _, bucket := range buckets {
|
||||
ctx := logger.SetReqInfo(GlobalContext, &logger.ReqInfo{BucketName: bucket.Name})
|
||||
config, err := readNotificationConfig(ctx, objAPI, bucket.Name)
|
||||
if err != nil && err != errNoSuchNotifications {
|
||||
if _, ok := err.(*event.ErrARNNotFound); ok {
|
||||
configData, err := globalBucketMetadataSys.GetConfig(bucket.Name, bucketNotificationConfig)
|
||||
if err != nil {
|
||||
if errors.Is(err, errConfigNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
if err == errNoSuchNotifications {
|
||||
config, err := event.ParseConfig(bytes.NewReader(configData), globalServerRegion, globalNotificationSys.targetList)
|
||||
if err != nil {
|
||||
if _, ok := err.(*event.ErrARNNotFound); !ok {
|
||||
logger.LogIf(ctx, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
sys.AddRulesMap(bucket.Name, config.ToRulesMap())
|
||||
@@ -904,46 +768,6 @@ func (sys *NotificationSys) Send(args eventArgs) {
|
||||
sys.targetList.Send(args.ToEvent(true), targetIDSet, sys.targetResCh)
|
||||
}
|
||||
|
||||
// PutBucketObjectLockConfig - put bucket object lock configuration to all peers.
|
||||
func (sys *NotificationSys) PutBucketObjectLockConfig(ctx context.Context, bucketName string, retention *objectlock.Retention) {
|
||||
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].PutBucketObjectLockConfig(bucketName, retention)
|
||||
}, index)
|
||||
}
|
||||
for i, err := range g.Wait() {
|
||||
if err != nil {
|
||||
logger.GetReqInfo(ctx).AppendTags("remotePeer", sys.peerClients[i].host.String())
|
||||
logger.LogIf(ctx, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -1387,30 +1211,3 @@ func sendEvent(args eventArgs) {
|
||||
|
||||
globalNotificationSys.Send(args)
|
||||
}
|
||||
|
||||
func readNotificationConfig(ctx context.Context, objAPI ObjectLayer, bucketName string) (*event.Config, error) {
|
||||
// Construct path to notification.xml for the given bucket.
|
||||
configFile := path.Join(bucketConfigPrefix, bucketName, bucketNotificationConfig)
|
||||
configData, err := readConfig(ctx, objAPI, configFile)
|
||||
if err != nil {
|
||||
if err == errConfigNotFound {
|
||||
err = errNoSuchNotifications
|
||||
}
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
config, err := event.ParseConfig(bytes.NewReader(configData), globalServerRegion, globalNotificationSys.targetList)
|
||||
logger.LogIf(ctx, err)
|
||||
return config, err
|
||||
}
|
||||
|
||||
func saveNotificationConfig(ctx context.Context, objAPI ObjectLayer, bucketName string, config *event.Config) error {
|
||||
data, err := xml.Marshal(config)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
configFile := path.Join(bucketConfigPrefix, bucketName, bucketNotificationConfig)
|
||||
return saveConfig(ctx, objAPI, configFile, data)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user