Add etcd support to support STS on gateway mode (#6531)

cmd/iam.go

@@ -20,9 +20,12 @@ import (
 	"context"
 	"encoding/json"
 	"path"
+	"strings"
 	"sync"
 	"time"
 
+	etcd "github.com/coreos/etcd/clientv3"
+	"github.com/minio/minio-go/pkg/set"
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/auth"
 	"github.com/minio/minio/pkg/iam/policy"
@@ -64,25 +67,46 @@ func (sys *IAMSys) Init(objAPI ObjectLayer) error {
 		return errInvalidArgument
 	}
 
-	if err := sys.refresh(objAPI); err != nil {
-		return err
-	}
-
-	// Refresh IAMSys in background.
-	go func() {
-		ticker := time.NewTicker(globalRefreshIAMInterval)
-		defer ticker.Stop()
-		for {
-			select {
-			case <-globalServiceDoneCh:
-				return
-			case <-ticker.C:
-				logger.LogIf(context.Background(), sys.refresh(objAPI))
+	defer func() {
+		// Refresh IAMSys in background.
+		go func() {
+			ticker := time.NewTicker(globalRefreshIAMInterval)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-globalServiceDoneCh:
+					return
+				case <-ticker.C:
+					sys.refresh(objAPI)
+				}
 			}
-		}
+		}()
 	}()
-	return nil
 
+	doneCh := make(chan struct{})
+	defer close(doneCh)
+
+	// Initializing IAM 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 IAMSys once during boot.
+			if err := sys.refresh(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
+			}
+			return nil
+		}
+	}
 }
 
 // SetPolicy - sets policy to given user name.  If policy is empty,
@@ -99,13 +123,19 @@ func (sys *IAMSys) SetPolicy(accessKey string, p iampolicy.Policy) error {
 		return err
 	}
 
+	if globalEtcdClient != nil {
+		if err = saveConfigEtcd(context.Background(), globalEtcdClient, configFile, data); err != nil {
+			return err
+		}
+	} else {
+		if err = saveConfig(context.Background(), objectAPI, configFile, data); err != nil {
+			return err
+		}
+	}
+
 	sys.Lock()
 	defer sys.Unlock()
 
-	if err = saveConfig(context.Background(), objectAPI, configFile, data); err != nil {
-		return err
-	}
-
 	if p.IsEmpty() {
 		delete(sys.iamPolicyMap, accessKey)
 	} else {
@@ -128,13 +158,19 @@ func (sys *IAMSys) SaveTempPolicy(accessKey string, p iampolicy.Policy) error {
 		return err
 	}
 
+	if globalEtcdClient != nil {
+		if err = saveConfigEtcd(context.Background(), globalEtcdClient, configFile, data); err != nil {
+			return err
+		}
+	} else {
+		if err = saveConfig(context.Background(), objectAPI, configFile, data); err != nil {
+			return err
+		}
+	}
+
 	sys.Lock()
 	defer sys.Unlock()
 
-	if err = saveConfig(context.Background(), objectAPI, configFile, data); err != nil {
-		return err
-	}
-
 	if p.IsEmpty() {
 		delete(sys.iamPolicyMap, accessKey)
 	} else {
@@ -152,13 +188,17 @@ func (sys *IAMSys) DeletePolicy(accessKey string) error {
 		return errServerNotInitialized
 	}
 
+	var err error
 	configFile := pathJoin(iamConfigUsersPrefix, accessKey, iamPolicyFile)
+	if globalEtcdClient != nil {
+		err = deleteConfigEtcd(context.Background(), globalEtcdClient, configFile)
+	} else {
+		err = deleteConfig(context.Background(), objectAPI, configFile)
+	}
 
 	sys.Lock()
 	defer sys.Unlock()
 
-	err := objectAPI.DeleteObject(context.Background(), minioMetaBucket, configFile)
-
 	delete(sys.iamPolicyMap, accessKey)
 
 	return err
@@ -171,12 +211,17 @@ func (sys *IAMSys) DeleteUser(accessKey string) error {
 		return errServerNotInitialized
 	}
 
+	var err error
+	configFile := pathJoin(iamConfigUsersPrefix, accessKey, iamPolicyFile)
+	if globalEtcdClient != nil {
+		err = deleteConfigEtcd(context.Background(), globalEtcdClient, configFile)
+	} else {
+		err = deleteConfig(context.Background(), objectAPI, configFile)
+	}
+
 	sys.Lock()
 	defer sys.Unlock()
 
-	configFile := pathJoin(iamConfigUsersPrefix, accessKey, iamIdentityFile)
-	err := objectAPI.DeleteObject(context.Background(), minioMetaBucket, configFile)
-
 	delete(sys.iamUsersMap, accessKey)
 	return err
 }
@@ -188,19 +233,25 @@ func (sys *IAMSys) SetTempUser(accessKey string, cred auth.Credentials) error {
 		return errServerNotInitialized
 	}
 
-	sys.Lock()
-	defer sys.Unlock()
-
 	configFile := pathJoin(iamConfigSTSPrefix, accessKey, iamIdentityFile)
 	data, err := json.Marshal(cred)
 	if err != nil {
 		return err
 	}
 
-	if err = saveConfig(context.Background(), objectAPI, configFile, data); err != nil {
-		return err
+	if globalEtcdClient != nil {
+		if err = saveConfigEtcd(context.Background(), globalEtcdClient, configFile, data); err != nil {
+			return err
+		}
+	} else {
+		if err = saveConfig(context.Background(), objectAPI, configFile, data); err != nil {
+			return err
+		}
 	}
 
+	sys.Lock()
+	defer sys.Unlock()
+
 	sys.iamUsersMap[accessKey] = cred
 	return nil
 }
@@ -218,13 +269,19 @@ func (sys *IAMSys) SetUser(accessKey string, uinfo madmin.UserInfo) error {
 		return err
 	}
 
+	if globalEtcdClient != nil {
+		if err = saveConfigEtcd(context.Background(), globalEtcdClient, configFile, data); err != nil {
+			return err
+		}
+	} else {
+		if err = saveConfig(context.Background(), objectAPI, configFile, data); err != nil {
+			return err
+		}
+	}
+
 	sys.Lock()
 	defer sys.Unlock()
 
-	if err = saveConfig(context.Background(), objectAPI, configFile, data); err != nil {
-		return err
-	}
-
 	sys.iamUsersMap[accessKey] = auth.Credentials{
 		AccessKey: accessKey,
 		SecretKey: uinfo.SecretKey,
@@ -266,6 +323,76 @@ func (sys *IAMSys) IsAllowed(args iampolicy.Args) bool {
 	return args.IsOwner
 }
 
+var defaultContextTimeout = 5 * time.Minute
+
+// Similar to reloadUsers but updates users, policies maps from etcd server,
+func reloadEtcdUsers(prefix string, usersMap map[string]auth.Credentials, policyMap map[string]iampolicy.Policy) error {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultContextTimeout)
+	r, err := globalEtcdClient.Get(ctx, prefix, etcd.WithPrefix(), etcd.WithKeysOnly())
+	defer cancel()
+	if err != nil {
+		return err
+	}
+	// No users are created yet.
+	if r.Count == 0 {
+		return nil
+	}
+
+	users := set.NewStringSet()
+	for _, kv := range r.Kvs {
+		// Extract user by stripping off the `prefix` value as suffix,
+		// then strip off the remaining basename to obtain the prefix
+		// value, usually in the following form.
+		//
+		//  key := "config/iam/users/newuser/identity.json"
+		//  prefix := "config/iam/users/"
+		//  v := trim(trim(key, prefix), base(key)) == "newuser"
+		//
+		user := strings.TrimSuffix(strings.TrimSuffix(string(kv.Key), prefix), path.Base(string(kv.Key)))
+		if !users.Contains(user) {
+			users.Add(user)
+		}
+	}
+
+	// Reload config and policies for all users.
+	for _, user := range users.ToSlice() {
+		idFile := pathJoin(prefix, user, iamIdentityFile)
+		pFile := pathJoin(prefix, user, iamPolicyFile)
+		cdata, cerr := readConfigEtcd(ctx, globalEtcdClient, idFile)
+		pdata, perr := readConfigEtcd(ctx, globalEtcdClient, pFile)
+		if cerr != nil && cerr != errConfigNotFound {
+			return cerr
+		}
+		if perr != nil && perr != errConfigNotFound {
+			return perr
+		}
+		if cerr == errConfigNotFound && perr == errConfigNotFound {
+			continue
+		}
+		if cerr == nil {
+			var cred auth.Credentials
+			if err = json.Unmarshal(cdata, &cred); err != nil {
+				return err
+			}
+			cred.AccessKey = user
+			if cred.IsExpired() {
+				deleteConfigEtcd(ctx, globalEtcdClient, idFile)
+				deleteConfigEtcd(ctx, globalEtcdClient, pFile)
+				continue
+			}
+			usersMap[cred.AccessKey] = cred
+		}
+		if perr == nil {
+			var p iampolicy.Policy
+			if err = json.Unmarshal(pdata, &p); err != nil {
+				return err
+			}
+			policyMap[path.Base(prefix)] = p
+		}
+	}
+	return nil
+}
+
 // reloadUsers reads an updates users, policies from object layer into user and policy maps.
 func reloadUsers(objectAPI ObjectLayer, prefix string, usersMap map[string]auth.Credentials, policyMap map[string]iampolicy.Policy) error {
 	marker := ""
@@ -326,12 +453,20 @@ func (sys *IAMSys) refresh(objAPI ObjectLayer) error {
 	iamUsersMap := make(map[string]auth.Credentials)
 	iamPolicyMap := make(map[string]iampolicy.Policy)
 
-	if err := reloadUsers(objAPI, iamConfigUsersPrefix, iamUsersMap, iamPolicyMap); err != nil {
-		return err
-	}
-
-	if err := reloadUsers(objAPI, iamConfigSTSPrefix, iamUsersMap, iamPolicyMap); err != nil {
-		return err
+	if globalEtcdClient != nil {
+		if err := reloadEtcdUsers(iamConfigUsersPrefix, iamUsersMap, iamPolicyMap); err != nil {
+			return err
+		}
+		if err := reloadEtcdUsers(iamConfigSTSPrefix, iamUsersMap, iamPolicyMap); err != nil {
+			return err
+		}
+	} else {
+		if err := reloadUsers(objAPI, iamConfigUsersPrefix, iamUsersMap, iamPolicyMap); err != nil {
+			return err
+		}
+		if err := reloadUsers(objAPI, iamConfigSTSPrefix, iamUsersMap, iamPolicyMap); err != nil {
+			return err
+		}
 	}
 
 	sys.Lock()