Initialize only one retry timer for all sub-systems (#8913)

Also make sure that we create buckets on all zones
successfully, do not run quick heal buckets if not
running with expansion.
master
Harshavardhana 5 years ago committed by GitHub
parent 5d838edcef
commit d76160c245
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 30
      cmd/config.go
  2. 44
      cmd/iam.go
  3. 31
      cmd/lifecycle.go
  4. 39
      cmd/notification.go
  5. 29
      cmd/policy.go
  6. 41
      cmd/server-main.go
  7. 6
      cmd/xl-zones.go

@ -20,7 +20,6 @@ import (
"bytes" "bytes"
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"path" "path"
"sort" "sort"
"strings" "strings"
@ -28,7 +27,6 @@ import (
jsoniter "github.com/json-iterator/go" jsoniter "github.com/json-iterator/go"
"github.com/minio/minio/cmd/config" "github.com/minio/minio/cmd/config"
"github.com/minio/minio/cmd/logger"
"github.com/minio/minio/pkg/madmin" "github.com/minio/minio/pkg/madmin"
) )
@ -212,33 +210,7 @@ func (sys *ConfigSys) Init(objAPI ObjectLayer) error {
return errInvalidArgument return errInvalidArgument
} }
doneCh := make(chan struct{}) return initConfig(objAPI)
defer close(doneCh)
// Initializing configuration needs a retry mechanism for
// the following reasons:
// - Read quorum is lost just after the initialization
// of the object layer.
// - Write quorum not met when upgrading configuration
// version is needed.
retryTimerCh := newRetryTimerSimple(doneCh)
for {
select {
case <-retryTimerCh:
if err := initConfig(objAPI); err != nil {
if err == errDiskNotFound ||
strings.Contains(err.Error(), InsufficientReadQuorum{}.Error()) ||
strings.Contains(err.Error(), InsufficientWriteQuorum{}.Error()) {
logger.Info("Waiting for configuration to be initialized..")
continue
}
return err
}
return nil
case <-globalOSSignalCh:
return fmt.Errorf("Initializing config sub-system gracefully stopped")
}
}
} }
// NewConfigSys - creates new config system object. // NewConfigSys - creates new config system object.

@ -20,8 +20,6 @@ import (
"bytes" "bytes"
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"strings"
"sync" "sync"
"github.com/minio/minio-go/v6/pkg/set" "github.com/minio/minio-go/v6/pkg/set"
@ -363,44 +361,14 @@ func (sys *IAMSys) Init(objAPI ObjectLayer) error {
} }
sys.Unlock() sys.Unlock()
doneCh := make(chan struct{}) // Migrate IAM configuration
defer close(doneCh) if err := sys.doIAMConfigMigration(objAPI); err != nil {
return err
// Migrating IAM amd Loading IAM needs a retry mechanism for }
// the following reasons:
// - Read quorum is lost just after the initialization
// of the object layer.
retryCh := newRetryTimerSimple(doneCh)
for {
select {
case <-retryCh:
// Migrate IAM configuration
if err := sys.doIAMConfigMigration(objAPI); err != nil {
if err == errDiskNotFound ||
strings.Contains(err.Error(), InsufficientReadQuorum{}.Error()) ||
strings.Contains(err.Error(), InsufficientWriteQuorum{}.Error()) {
logger.Info("Waiting for IAM subsystem to be initialized..")
continue
}
return err
}
sys.store.watch(sys) sys.store.watch(sys)
if err := sys.store.loadAll(sys, objAPI); err != nil { return sys.store.loadAll(sys, objAPI)
if err == errDiskNotFound ||
strings.Contains(err.Error(), InsufficientReadQuorum{}.Error()) ||
strings.Contains(err.Error(), InsufficientWriteQuorum{}.Error()) {
logger.Info("Waiting for IAM subsystem to be initialized..")
continue
}
return err
}
return nil
case <-globalOSSignalCh:
return fmt.Errorf("Initializing IAM sub-system gracefully stopped")
}
}
} }
// DeletePolicy - deletes a canned policy from backend or etcd. // DeletePolicy - deletes a canned policy from backend or etcd.

@ -20,12 +20,9 @@ import (
"bytes" "bytes"
"context" "context"
"encoding/xml" "encoding/xml"
"fmt"
"path" "path"
"strings"
"sync" "sync"
"github.com/minio/minio/cmd/logger"
"github.com/minio/minio/pkg/bucket/lifecycle" "github.com/minio/minio/pkg/bucket/lifecycle"
) )
@ -131,32 +128,8 @@ func (sys *LifecycleSys) Init(buckets []BucketInfo, objAPI ObjectLayer) error {
return nil return nil
} }
doneCh := make(chan struct{}) // Load LifecycleSys once during boot.
defer close(doneCh) return sys.load(buckets, objAPI)
// Initializing lifecycle needs a retry mechanism for
// the following reasons:
// - Read quorum is lost just after the initialization
// of the object layer.
retryTimerCh := newRetryTimerSimple(doneCh)
for {
select {
case <-retryTimerCh:
// Load LifecycleSys once during boot.
if err := sys.load(buckets, objAPI); err != nil {
if err == errDiskNotFound ||
strings.Contains(err.Error(), InsufficientReadQuorum{}.Error()) ||
strings.Contains(err.Error(), InsufficientWriteQuorum{}.Error()) {
logger.Info("Waiting for lifecycle subsystem to be initialized..")
continue
}
return err
}
return nil
case <-globalOSSignalCh:
return fmt.Errorf("Initializing Lifecycle sub-system gracefully stopped")
}
}
} }
// Loads lifecycle policies for all buckets into LifecycleSys. // Loads lifecycle policies for all buckets into LifecycleSys.

@ -716,42 +716,11 @@ func (sys *NotificationSys) Init(buckets []BucketInfo, objAPI ObjectLayer) error
} }
} }
doneCh := make(chan struct{}) if err := sys.load(buckets, objAPI); err != nil {
defer close(doneCh) return err
// Initializing notification needs a retry mechanism for
// the following reasons:
// - Read quorum is lost just after the initialization
// of the object layer.
retryTimerCh := newRetryTimerSimple(doneCh)
for {
select {
case <-retryTimerCh:
if err := sys.load(buckets, objAPI); err != nil {
if err == errDiskNotFound ||
strings.Contains(err.Error(), InsufficientReadQuorum{}.Error()) ||
strings.Contains(err.Error(), InsufficientWriteQuorum{}.Error()) {
logger.Info("Waiting for notification subsystem to be initialized..")
continue
}
return err
}
// Initializing bucket retention config needs a retry mechanism if
// read quorum is lost just after the initialization of the object layer.
if err := sys.initBucketObjectLockConfig(objAPI); err != nil {
if err == errDiskNotFound ||
strings.Contains(err.Error(), InsufficientReadQuorum{}.Error()) ||
strings.Contains(err.Error(), InsufficientWriteQuorum{}.Error()) {
logger.Info("Waiting for bucket retention configuration to be initialized..")
continue
}
return err
}
return nil
case <-globalOSSignalCh:
return fmt.Errorf("Initializing Notification sub-system gracefully stopped")
}
} }
return sys.initBucketObjectLockConfig(objAPI)
} }
// AddRulesMap - adds rules map for bucket name. // AddRulesMap - adds rules map for bucket name.

@ -23,7 +23,6 @@ import (
"fmt" "fmt"
"net/http" "net/http"
"path" "path"
"strings"
"sync" "sync"
jsoniter "github.com/json-iterator/go" jsoniter "github.com/json-iterator/go"
@ -133,32 +132,8 @@ func (sys *PolicySys) Init(buckets []BucketInfo, objAPI ObjectLayer) error {
return nil return nil
} }
doneCh := make(chan struct{}) // Load PolicySys once during boot.
defer close(doneCh) return sys.load(buckets, objAPI)
// Initializing policy needs a retry mechanism for
// the following reasons:
// - Read quorum is lost just after the initialization
// of the object layer.
retryTimerCh := newRetryTimerSimple(doneCh)
for {
select {
case <-retryTimerCh:
// Load PolicySys once during boot.
if err := sys.load(buckets, objAPI); err != nil {
if err == errDiskNotFound ||
strings.Contains(err.Error(), InsufficientReadQuorum{}.Error()) ||
strings.Contains(err.Error(), InsufficientWriteQuorum{}.Error()) {
logger.Info("Waiting for policy subsystem to be initialized..")
continue
}
return err
}
return nil
case <-globalOSSignalCh:
return fmt.Errorf("Initializing Policy sub-system gracefully stopped")
}
}
} }
// NewPolicySys - creates new policy system. // NewPolicySys - creates new policy system.

@ -206,11 +206,44 @@ func initSafeMode(buckets []BucketInfo) (err error) {
// sub-systems, make sure that we do not move the above codeblock elsewhere. // sub-systems, make sure that we do not move the above codeblock elsewhere.
// Validate and initialize all subsystems. // Validate and initialize all subsystems.
if err = initAllSubsystems(buckets, newObject); err != nil { doneCh := make(chan struct{})
return err defer close(doneCh)
// Initializing sub-systems needs a retry mechanism for
// the following reasons:
// - Read quorum is lost just after the initialization
// of the object layer.
// - Write quorum not met when upgrading configuration
// version is needed, migration is needed etc.
retryTimerCh := newRetryTimerSimple(doneCh)
for {
rquorum := InsufficientReadQuorum{}
wquorum := InsufficientWriteQuorum{}
var err error
select {
case n := <-retryTimerCh:
if err = initAllSubsystems(buckets, newObject); err != nil {
if errors.Is(err, errDiskNotFound) ||
errors.As(err, &rquorum) ||
errors.As(err, &wquorum) {
if n < 5 {
logger.Info("Waiting for all sub-systems to be initialized..")
} else {
logger.Info("Waiting for all sub-systems to be initialized.. %v", err)
}
continue
}
return err
}
return nil
case <-globalOSSignalCh:
if err == nil {
return errors.New("Initializing sub-systems stopped gracefully")
}
return fmt.Errorf("Unable to initialize sub-systems: %w", err)
}
} }
return nil
} }
func initAllSubsystems(buckets []BucketInfo, newObject ObjectLayer) (err error) { func initAllSubsystems(buckets []BucketInfo, newObject ObjectLayer) (err error) {

@ -49,7 +49,7 @@ func (z *xlZones) quickHealBuckets(ctx context.Context) {
return return
} }
for _, bucket := range bucketsInfo { for _, bucket := range bucketsInfo {
z.HealBucket(ctx, bucket.Name, false, false) z.MakeBucketWithLocation(ctx, bucket.Name, "")
} }
} }
@ -77,7 +77,9 @@ func newXLZones(endpointZones EndpointZones) (ObjectLayer, error) {
return nil, err return nil, err
} }
} }
z.quickHealBuckets(context.Background()) if !z.SingleZone() {
z.quickHealBuckets(context.Background())
}
return z, nil return z, nil
} }

Loading…
Cancel
Save