Merge pull request #279 from versity/ben/iam_cache

feat: move local iam cache to a more generic cache mechanism

auth/iam.go

@@ -16,6 +16,7 @@ package auth
 
 import (
 	"errors"
+	"time"
 )
 
 // Account is a gateway IAM account
@@ -33,6 +34,7 @@ type IAMService interface {
 	GetUserAccount(access string) (Account, error)
 	DeleteUserAccount(access string) error
 	ListUserAccounts() ([]Account, error)
+	Shutdown() error
 }
 
 var ErrNoSuchUser = errors.New("user not found")
@@ -47,18 +49,36 @@ type Opts struct {
 	LDAPAccessAtr  string
 	LDAPSecretAtr  string
 	LDAPRoleAtr    string
+	CacheDisable   bool
+	CacheTTL       int
+	CachePrune     int
 }
 
 func New(o *Opts) (IAMService, error) {
+	var svc IAMService
+	var err error
+
 	switch {
 	case o.Dir != "":
-		return NewInternal(o.Dir)
+		svc, err = NewInternal(o.Dir)
 	case o.LDAPServerURL != "":
-		return NewLDAPService(o.LDAPServerURL, o.LDAPBindDN, o.LDAPPassword,
+		svc, err = NewLDAPService(o.LDAPServerURL, o.LDAPBindDN, o.LDAPPassword,
 			o.LDAPQueryBase, o.LDAPAccessAtr, o.LDAPSecretAtr, o.LDAPRoleAtr,
 			o.LDAPObjClasses)
 	default:
 		// if no iam options selected, default to the single user mode
 		return IAMServiceSingle{}, nil
 	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if o.CacheDisable {
+		return svc, nil
+	}
+
+	return NewCache(svc,
+		time.Duration(o.CacheTTL)*time.Second,
+		time.Duration(o.CachePrune)*time.Second), nil
 }

auth/iam_cache.go

@@ -0,0 +1,179 @@
+// Copyright 2023 Versity Software
+// This file is licensed under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package auth
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+)
+
+// IAMCache is an in memory cache of the IAM accounts
+// with expiration. This helps to alleviate the load on
+// the real IAM service if the gateway is handling
+// many requests. This forwards account updates to the
+// underlying service, and returns cached results while
+// the in memory account is not expired.
+type IAMCache struct {
+	service  IAMService
+	iamcache *icache
+	cancel   context.CancelFunc
+}
+
+var _ IAMService = &IAMCache{}
+
+type item struct {
+	value Account
+	exp   time.Time
+}
+
+type icache struct {
+	sync.RWMutex
+	expire time.Duration
+	items  map[string]item
+}
+
+func (i *icache) set(k string, v Account) {
+	cpy := v
+	i.Lock()
+	i.items[k] = item{
+		exp:   time.Now().Add(i.expire),
+		value: cpy,
+	}
+	i.Unlock()
+}
+
+func (i *icache) get(k string) (Account, bool) {
+	i.RLock()
+	v, ok := i.items[k]
+	i.RUnlock()
+	if !ok || !v.exp.After(time.Now()) {
+		return Account{}, false
+	}
+	return v.value, true
+}
+
+func (i *icache) Delete(k string) {
+	i.Lock()
+	delete(i.items, k)
+	i.Unlock()
+}
+
+func (i *icache) gcCache(ctx context.Context, interval time.Duration) {
+	for {
+		if ctx.Err() != nil {
+			break
+		}
+
+		now := time.Now()
+
+		i.Lock()
+		// prune expired entries
+		for k, v := range i.items {
+			if now.After(v.exp) {
+				delete(i.items, k)
+			}
+		}
+		i.Unlock()
+
+		// sleep for the clean interval or context cancelation,
+		// whichever comes first
+		select {
+		case <-ctx.Done():
+		case <-time.After(interval):
+		}
+	}
+}
+
+// NewCache initializes an IAM cache for the provided service. The expireTime
+// is the duration a cache entry can be valid, and the cleanupInterval is
+// how often to scan cache and cleanup expired entries.
+func NewCache(service IAMService, expireTime, cleanupInterval time.Duration) *IAMCache {
+	i := &IAMCache{
+		service: service,
+		iamcache: &icache{
+			items:  make(map[string]item),
+			expire: expireTime,
+		},
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go i.iamcache.gcCache(ctx, cleanupInterval)
+	i.cancel = cancel
+
+	return i
+}
+
+// CreateAccount send create to IAM service and creates an account cache entry
+func (c *IAMCache) CreateAccount(account Account) error {
+	err := c.service.CreateAccount(account)
+	if err != nil {
+		return err
+	}
+
+	// we need a copy of account to be able to store beyond the
+	// lifetime of the request, otherwise Fiber will reuse and corrupt
+	// these entries
+	acct := Account{
+		Access: strings.Clone(account.Access),
+		Secret: strings.Clone(account.Secret),
+		Role:   strings.Clone(account.Role),
+	}
+
+	c.iamcache.set(acct.Access, acct)
+	return nil
+}
+
+// GetUserAccount retrieves the cache account if it is in the cache and not
+// expired. Otherwise retrieves from underlying IAM service and caches
+// result for the expire duration.
+func (c *IAMCache) GetUserAccount(access string) (Account, error) {
+	acct, found := c.iamcache.get(access)
+	if found {
+		return acct, nil
+	}
+
+	a, err := c.service.GetUserAccount(access)
+	if err != nil {
+		return Account{}, err
+	}
+
+	c.iamcache.set(access, a)
+	return a, nil
+}
+
+// DeleteUserAccount deletes account from IAM service and cache
+func (c *IAMCache) DeleteUserAccount(access string) error {
+	err := c.service.DeleteUserAccount(access)
+	if err != nil {
+		return err
+	}
+
+	c.iamcache.Delete(access)
+	return nil
+}
+
+// ListUserAccounts is a passthrough to the underlying service and
+// does not make use of the cache
+func (c *IAMCache) ListUserAccounts() ([]Account, error) {
+	return c.service.ListUserAccounts()
+}
+
+// Shutdown graceful termination of service
+func (c *IAMCache) Shutdown() error {
+	c.cancel()
+	return nil
+}

auth/iam_internal.go

@@ -18,11 +18,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"hash/crc32"
 	"io/fs"
 	"os"
 	"path/filepath"
-	"sync"
+	"sort"
 	"time"
 )
 
@@ -31,20 +30,9 @@ const (
 	iamBackupFile = "users.json.backup"
 )
 
-var (
-	cacheDuration = 5 * time.Minute
-)
-
 // IAMServiceInternal manages the internal IAM service
 type IAMServiceInternal struct {
-	path string
-
-	mu        sync.RWMutex
-	accts     iAMConfig
-	serial    uint32
-	iamcache  []byte
-	iamvalid  bool
-	iamexpire time.Time
+	dir string
 }
 
 // UpdateAcctFunc accepts the current data and returns the new data to be stored
@@ -58,9 +46,9 @@ type iAMConfig struct {
 var _ IAMService = &IAMServiceInternal{}
 
 // NewInternal creates a new instance for the Internal IAM service
-func NewInternal(path string) (*IAMServiceInternal, error) {
+func NewInternal(dir string) (*IAMServiceInternal, error) {
 	i := &IAMServiceInternal{
-		path: path,
+		dir: dir,
 	}
 
 	err := i.initIAM()
@@ -68,29 +56,16 @@ func NewInternal(path string) (*IAMServiceInternal, error) {
 		return nil, fmt.Errorf("init iam: %w", err)
 	}
 
-	err = i.updateCache()
-	if err != nil {
-		return nil, fmt.Errorf("refresh iam cache: %w", err)
-	}
-
 	return i, nil
 }
 
 // CreateAccount creates a new IAM account. Returns an error if the account
 // already exists.
 func (s *IAMServiceInternal) CreateAccount(account Account) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	return s.storeIAM(func(data []byte) ([]byte, error) {
-		var conf iAMConfig
-
-		if len(data) > 0 {
-			if err := json.Unmarshal(data, &conf); err != nil {
-				return nil, fmt.Errorf("failed to parse iam: %w", err)
-			}
-		} else {
-			conf = iAMConfig{AccessAccounts: map[string]Account{}}
+		conf, err := parseIAM(data)
+		if err != nil {
+			return nil, fmt.Errorf("get iam data: %w", err)
 		}
 
 		_, ok := conf.AccessAccounts[account.Access]
@@ -103,7 +78,6 @@ func (s *IAMServiceInternal) CreateAccount(account Account) error {
 		if err != nil {
 			return nil, fmt.Errorf("failed to serialize iam: %w", err)
 		}
-		s.accts = conf
 
 		return b, nil
 	})
@@ -112,25 +86,12 @@ func (s *IAMServiceInternal) CreateAccount(account Account) error {
 // GetUserAccount retrieves account info for the requested user. Returns
 // ErrNoSuchUser if the account does not exist.
 func (s *IAMServiceInternal) GetUserAccount(access string) (Account, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-
-	data, err := s.getIAM()
+	conf, err := s.getIAM()
 	if err != nil {
 		return Account{}, fmt.Errorf("get iam data: %w", err)
 	}
 
-	serial := crc32.ChecksumIEEE(data)
-	if serial != s.serial {
-		s.mu.RUnlock()
-		err := s.updateCache()
-		s.mu.RLock()
-		if err != nil {
-			return Account{}, fmt.Errorf("refresh iam cache: %w", err)
-		}
-	}
-
-	acct, ok := s.accts.AccessAccounts[access]
+	acct, ok := conf.AccessAccounts[access]
 	if !ok {
 		return Account{}, ErrNoSuchUser
 	}
@@ -138,47 +99,13 @@ func (s *IAMServiceInternal) GetUserAccount(access string) (Account, error) {
 	return acct, nil
 }
 
-// updateCache must be called with no locks held
-func (s *IAMServiceInternal) updateCache() error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
-	data, err := s.getIAM()
-	if err != nil {
-		return fmt.Errorf("get iam data: %w", err)
-	}
-
-	serial := crc32.ChecksumIEEE(data)
-
-	if len(data) > 0 {
-		if err := json.Unmarshal(data, &s.accts); err != nil {
-			return fmt.Errorf("failed to parse the config file: %w", err)
-		}
-	} else {
-		s.accts.AccessAccounts = make(map[string]Account)
-	}
-
-	s.serial = serial
-
-	return nil
-}
-
 // DeleteUserAccount deletes the specified user account. Does not check if
 // account exists.
 func (s *IAMServiceInternal) DeleteUserAccount(access string) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	return s.storeIAM(func(data []byte) ([]byte, error) {
-		if len(data) == 0 {
-			// empty config, do nothing
-			return data, nil
-		}
-
-		var conf iAMConfig
-
-		if err := json.Unmarshal(data, &conf); err != nil {
-			return nil, fmt.Errorf("failed to parse iam: %w", err)
+		conf, err := parseIAM(data)
+		if err != nil {
+			return nil, fmt.Errorf("get iam data: %w", err)
 		}
 
 		delete(conf.AccessAccounts, access)
@@ -188,49 +115,46 @@ func (s *IAMServiceInternal) DeleteUserAccount(access string) error {
 			return nil, fmt.Errorf("failed to serialize iam: %w", err)
 		}
 
-		s.accts = conf
-
 		return b, nil
 	})
 }
 
 // ListUserAccounts lists all the user accounts stored.
-func (s *IAMServiceInternal) ListUserAccounts() (accs []Account, err error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-
-	data, err := s.getIAM()
+func (s *IAMServiceInternal) ListUserAccounts() ([]Account, error) {
+	conf, err := s.getIAM()
 	if err != nil {
 		return []Account{}, fmt.Errorf("get iam data: %w", err)
 	}
 
-	serial := crc32.ChecksumIEEE(data)
-	if serial != s.serial {
-		s.mu.RUnlock()
-		err := s.updateCache()
-		s.mu.RLock()
-		if err != nil {
-			return []Account{}, fmt.Errorf("refresh iam cache: %w", err)
-		}
+	keys := make([]string, 0, len(conf.AccessAccounts))
+	for k := range conf.AccessAccounts {
+		keys = append(keys, k)
 	}
+	sort.Strings(keys)
 
-	for access, usr := range s.accts.AccessAccounts {
+	var accs []Account
+	for _, k := range keys {
 		accs = append(accs, Account{
-			Access: access,
-			Secret: usr.Secret,
-			Role:   usr.Role,
+			Access: k,
+			Secret: conf.AccessAccounts[k].Secret,
+			Role:   conf.AccessAccounts[k].Role,
 		})
 	}
 
 	return accs, nil
 }
 
+// Shutdown graceful termination of service
+func (s *IAMServiceInternal) Shutdown() error {
+	return nil
+}
+
 const (
 	iamMode = 0600
 )
 
 func (s *IAMServiceInternal) initIAM() error {
-	fname := filepath.Join(s.path, iamFile)
+	fname := filepath.Join(s.dir, iamFile)
 
 	_, err := os.ReadFile(fname)
 	if errors.Is(err, fs.ErrNotExist) {
@@ -247,15 +171,22 @@ func (s *IAMServiceInternal) initIAM() error {
 	return nil
 }
 
-func (s *IAMServiceInternal) getIAM() ([]byte, error) {
-	if !s.iamvalid || !s.iamexpire.After(time.Now()) {
-		err := s.refreshIAM()
-		if err != nil {
-			return nil, err
-		}
+func (s *IAMServiceInternal) getIAM() (iAMConfig, error) {
+	b, err := s.readIAMData()
+	if err != nil {
+		return iAMConfig{}, err
 	}
 
-	return s.iamcache, nil
+	return parseIAM(b)
+}
+
+func parseIAM(b []byte) (iAMConfig, error) {
+	var conf iAMConfig
+	if err := json.Unmarshal(b, &conf); err != nil {
+		return iAMConfig{}, fmt.Errorf("failed to parse the config file: %w", err)
+	}
+
+	return conf, nil
 }
 
 const (
@@ -263,7 +194,7 @@ const (
 	maxretry = 300
 )
 
-func (s *IAMServiceInternal) refreshIAM() error {
+func (s *IAMServiceInternal) readIAMData() ([]byte, error) {
 	// We are going to be racing with other running gateways without any
 	// coordination. So we might find the file does not exist at times.
 	// For this case we need to retry for a while assuming the other gateway
@@ -273,7 +204,7 @@ func (s *IAMServiceInternal) refreshIAM() error {
 	retries := 0
 
 	for {
-		b, err := os.ReadFile(filepath.Join(s.path, iamFile))
+		b, err := os.ReadFile(filepath.Join(s.dir, iamFile))
 		if errors.Is(err, fs.ErrNotExist) {
 			// racing with someone else updating
 			// keep retrying after backoff
@@ -282,19 +213,14 @@ func (s *IAMServiceInternal) refreshIAM() error {
 				time.Sleep(backoff)
 				continue
 			}
-			return fmt.Errorf("read iam file: %w", err)
+			return nil, fmt.Errorf("read iam file: %w", err)
 		}
 		if err != nil {
-			return err
+			return nil, err
 		}
 
-		s.iamcache = b
-		s.iamvalid = true
-		s.iamexpire = time.Now().Add(cacheDuration)
-		break
+		return b, nil
 	}
-
-	return nil
 }
 
 func (s *IAMServiceInternal) storeIAM(update UpdateAcctFunc) error {
@@ -314,7 +240,7 @@ func (s *IAMServiceInternal) storeIAM(update UpdateAcctFunc) error {
 	// write the file.
 
 	retries := 0
-	fname := filepath.Join(s.path, iamFile)
+	fname := filepath.Join(s.dir, iamFile)
 
 	for {
 		b, err := os.ReadFile(fname)
@@ -361,7 +287,7 @@ func (s *IAMServiceInternal) storeIAM(update UpdateAcctFunc) error {
 		// can go wrong, but the remove should barrier other gateways
 		// from trying to write backup at the same time. Only one
 		// gateway will successfully remove the file.
-		os.WriteFile(filepath.Join(s.path, iamBackupFile), b, iamMode)
+		os.WriteFile(filepath.Join(s.dir, iamBackupFile), b, iamMode)
 
 		b, err = update(b)
 		if err != nil {
@@ -377,9 +303,6 @@ func (s *IAMServiceInternal) storeIAM(update UpdateAcctFunc) error {
 			return err
 		}
 
-		s.iamcache = b
-		s.iamvalid = true
-		s.iamexpire = time.Now().Add(cacheDuration)
 		break
 	}
 
@@ -387,9 +310,9 @@ func (s *IAMServiceInternal) storeIAM(update UpdateAcctFunc) error {
 }
 
 func (s *IAMServiceInternal) writeTempFile(b []byte) error {
-	fname := filepath.Join(s.path, iamFile)
+	fname := filepath.Join(s.dir, iamFile)
 
-	f, err := os.CreateTemp(s.path, iamFile)
+	f, err := os.CreateTemp(s.dir, iamFile)
 	if err != nil {
 		return fmt.Errorf("create temp file: %w", err)
 	}

auth/iam_ldap.go

@@ -126,3 +126,8 @@ func (ld *LdapIAMService) ListUserAccounts() ([]Account, error) {
 
 	return result, nil
 }
+
+// Shutdown graceful termination of service
+func (ld *LdapIAMService) Shutdown() error {
+	return ld.conn.Close()
+}

auth/iam_single.go

@@ -19,6 +19,8 @@ import "fmt"
 // IAMServiceSingle manages the single tenant (root-only) IAM service
 type IAMServiceSingle struct{}
 
+var _ IAMService = &IAMServiceSingle{}
+
 // CreateAccount not valid in single tenant mode
 func (IAMServiceSingle) CreateAccount(account Account) error {
 	return fmt.Errorf("create user not valid in single tenant mode")
@@ -38,3 +40,8 @@ func (IAMServiceSingle) DeleteUserAccount(access string) error {
 func (IAMServiceSingle) ListUserAccounts() ([]Account, error) {
 	return []Account{}, nil
 }
+
+// Shutdown graceful termination of service
+func (IAMServiceSingle) Shutdown() error {
+	return nil
+}

cmd/versitygw/main.go

@@ -48,6 +48,9 @@ var (
 	ldapURL, ldapBindDN, ldapPassword      string
 	ldapQueryBase, ldapObjClasses          string
 	ldapAccessAtr, ldapSecAtr, ldapRoleAtr string
+	iamCacheDisable                        bool
+	iamCacheTTL                            int
+	iamCachePrune                          int
 )
 
 var (
@@ -256,6 +259,23 @@ func initFlags() []cli.Flag {
 			Usage:       "ldap server user role attribute name",
 			Destination: &ldapRoleAtr,
 		},
+		&cli.BoolFlag{
+			Name:        "iam-cache-disable",
+			Usage:       "disable local iam cache",
+			Destination: &iamCacheDisable,
+		},
+		&cli.IntFlag{
+			Name:        "iam-cache-ttl",
+			Usage:       "local iam cache entry ttl (seconds)",
+			Value:       120,
+			Destination: &iamCacheTTL,
+		},
+		&cli.IntFlag{
+			Name:        "iam-cache-prune",
+			Usage:       "local iam cache cleanup interval (seconds)",
+			Value:       3600,
+			Destination: &iamCachePrune,
+		},
 	}
 }
 
@@ -328,6 +348,9 @@ func runGateway(ctx *cli.Context, be backend.Backend) error {
 		LDAPAccessAtr:  ldapAccessAtr,
 		LDAPSecretAtr:  ldapSecAtr,
 		LDAPRoleAtr:    ldapRoleAtr,
+		CacheDisable:   iamCacheDisable,
+		CacheTTL:       iamCacheTTL,
+		CachePrune:     iamCachePrune,
 	})
 	if err != nil {
 		return fmt.Errorf("setup iam: %w", err)
@@ -387,13 +410,21 @@ Loop:
 			}
 		}
 	}
+	saveErr := err
 
 	be.Shutdown()
+
+	err = iam.Shutdown()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "shutdown iam: %v\n", err)
+	}
+
 	if logger != nil {
-		lerr := logger.Shutdown()
-		if lerr != nil {
-			fmt.Fprintf(os.Stderr, "shutdown logger: %v\n", lerr)
+		err := logger.Shutdown()
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "shutdown logger: %v\n", err)
 		}
 	}
-	return err
+
+	return saveErr
 }

s3api/controllers/iam_moq_test.go

@@ -30,6 +30,9 @@ var _ auth.IAMService = &IAMServiceMock{}
 //			ListUserAccountsFunc: func() ([]auth.Account, error) {
 //				panic("mock out the ListUserAccounts method")
 //			},
+//			ShutdownFunc: func() error {
+//				panic("mock out the Shutdown method")
+//			},
 //		}
 //
 //		// use mockedIAMService in code that requires auth.IAMService
@@ -49,6 +52,9 @@ type IAMServiceMock struct {
 	// ListUserAccountsFunc mocks the ListUserAccounts method.
 	ListUserAccountsFunc func() ([]auth.Account, error)
 
+	// ShutdownFunc mocks the Shutdown method.
+	ShutdownFunc func() error
+
 	// calls tracks calls to the methods.
 	calls struct {
 		// CreateAccount holds details about calls to the CreateAccount method.
@@ -69,11 +75,15 @@ type IAMServiceMock struct {
 		// ListUserAccounts holds details about calls to the ListUserAccounts method.
 		ListUserAccounts []struct {
 		}
+		// Shutdown holds details about calls to the Shutdown method.
+		Shutdown []struct {
+		}
 	}
 	lockCreateAccount     sync.RWMutex
 	lockDeleteUserAccount sync.RWMutex
 	lockGetUserAccount    sync.RWMutex
 	lockListUserAccounts  sync.RWMutex
+	lockShutdown          sync.RWMutex
 }
 
 // CreateAccount calls CreateAccountFunc.
@@ -198,3 +208,30 @@ func (mock *IAMServiceMock) ListUserAccountsCalls() []struct {
 	mock.lockListUserAccounts.RUnlock()
 	return calls
 }
+
+// Shutdown calls ShutdownFunc.
+func (mock *IAMServiceMock) Shutdown() error {
+	if mock.ShutdownFunc == nil {
+		panic("IAMServiceMock.ShutdownFunc: method is nil but IAMService.Shutdown was just called")
+	}
+	callInfo := struct {
+	}{}
+	mock.lockShutdown.Lock()
+	mock.calls.Shutdown = append(mock.calls.Shutdown, callInfo)
+	mock.lockShutdown.Unlock()
+	return mock.ShutdownFunc()
+}
+
+// ShutdownCalls gets all the calls that were made to Shutdown.
+// Check the length with:
+//
+//	len(mockedIAMService.ShutdownCalls())
+func (mock *IAMServiceMock) ShutdownCalls() []struct {
+} {
+	var calls []struct {
+	}
+	mock.lockShutdown.RLock()
+	calls = mock.calls.Shutdown
+	mock.lockShutdown.RUnlock()
+	return calls
+}