[TT-10826] self trim oAuth sorted set (#5907)

<!-- Provide a general summary of your changes in the Title above -->

Add a background job to self trim oAuth sorted set

https://tyktech.atlassian.net/browse/TT-10826

<!-- Why is this change required? What problem does it solve? -->

<!-- Please describe in detail how you tested your changes -->
<!-- Include details of your testing environment, and the tests -->
<!-- you ran to see how your change affects other areas of the code,
etc. -->
<!-- This information is helpful for reviewers and QA. -->

<!-- What types of changes does your code introduce? Put an `x` in all
the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Refactoring or add test (improvements in base code or adds test
coverage to functionality)

<!-- Go over all the following points, and put an `x` in all the boxes
that apply -->
<!-- If there are no documentation updates required, mark the item as
checked. -->
<!-- Raise up any additional concerns not covered by the checklist. -->

- [ ] I ensured that the documentation is up to date
- [ ] I explained why this PR updates go.mod in detail with reasoning
why it's required
- [ ] I would like a code coverage CI quality gate exception and have
explained why

---------

Co-authored-by: Tit Petric <tit.petric@monotek.net>
Co-authored-by: Tit Petric <tit@tyk.io>

(cherry picked from commit ee5dc29)

bin/run-benchmarks.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+set -e
+
+echo "Running benchmarks"
+
+
+go test -json -benchtime 30s -run='^$' -bench BenchmarkPurgeLapsedOAuthTokens github.com/TykTechnologies/tyk/gateway

cli/linter/linter.go

@@ -23,7 +23,7 @@ func Run(schm string, paths []string) (string, []string, error) {
 	schemaLoader := schema.NewBytesLoader([]byte(schm))
 
 	var orig map[string]interface{}
-	f, err := os.Open(conf.OriginalPath)
+	f, err := os.Open(conf.Private.OriginalPath)
 	if err != nil {
 		return "", nil, err
 	}
@@ -46,11 +46,11 @@ func Run(schm string, paths []string) (string, []string, error) {
 	}
 
 	// ensure it's well formatted and the keys are all lowercase
-	if err := config.WriteConf(conf.OriginalPath, &conf); err != nil {
+	if err := config.WriteConf(conf.Private.OriginalPath, &conf); err != nil {
 		return "", nil, err
 	}
 
-	return conf.OriginalPath, resultWarns(result), nil
+	return conf.Private.OriginalPath, resultWarns(result), nil
 }
 
 type stringFormat func(string) bool

config/config.go

@@ -557,12 +557,16 @@ func (pwl *PortsWhiteList) Decode(value string) error {
 
 // Config is the configuration object used by Tyk to set up various parameters.
 type Config struct {
+<<<<<<< HEAD
 	// OriginalPath is the path to the config file that is read. If
 	// none was found, it's the path to the default config file that
 	// was written.
 	OriginalPath string `json:"-"`
 
 	// Force your Gateway to work only on a specifc domain name. Can be overriden by API custom domain.
+=======
+	// Force your Gateway to work only on a specific domain name. Can be overridden by API custom domain.
+>>>>>>> ee5dc29b... [TT-10826] self trim oAuth sorted set (#5907)
 	HostName string `json:"hostname"`
 
 	// If your machine has mulitple network devices or IPs you can force the Gateway to use the IP address you want.
@@ -1015,6 +1019,24 @@ type Config struct {
 
 	// Skip TLS verification for JWT JWKs url validation
 	JWTSSLInsecureSkipVerify bool `json:"jwt_ssl_insecure_skip_verify"`
+<<<<<<< HEAD
+=======
+
+	// ResourceSync configures mitigation strategy in case sync fails.
+	ResourceSync ResourceSyncConfig `json:"resource_sync"`
+
+	Private Private `json:"-"`
+}
+
+type ResourceSyncConfig struct {
+	// RetryAttempts defines the number of retries that the Gateway
+	// should perform during a resource sync (APIs or policies), defaulting
+	// to zero which means no retries are attempted.
+	RetryAttempts int `json:"retry_attempts"`
+
+	// Interval configures the interval in seconds between each retry on a resource sync error.
+	Interval int `json:"interval"`
+>>>>>>> ee5dc29b... [TT-10826] self trim oAuth sorted set (#5907)
 }
 
 type TykError struct {
@@ -1176,7 +1198,12 @@ func Load(paths []string, conf *Config) error {
 		f, err := os.Open(path)
 		if err == nil {
 			r = f
+<<<<<<< HEAD
 			conf.OriginalPath = path
+=======
+			defer r.Close()
+			conf.Private.OriginalPath = filename
+>>>>>>> ee5dc29b... [TT-10826] self trim oAuth sorted set (#5907)
 			break
 		}
 		if os.IsNotExist(err) {

config/config_test.go

@@ -113,7 +113,7 @@ func TestConfigFiles(t *testing.T) {
 	if _, err := os.Stat(path2); err == nil {
 		t.Fatalf("Load with no configs wrote too many default config files")
 	}
-	if conf.OriginalPath != path1 {
+	if conf.Private.OriginalPath != path1 {
 		t.Fatalf("OriginalPath was not set properly")
 	}
 
@@ -122,7 +122,7 @@ func TestConfigFiles(t *testing.T) {
 	if err := Load(paths, conf); err != nil {
 		t.Fatalf("Load with an existing config errored")
 	}
-	if conf.OriginalPath != path1 {
+	if conf.Private.OriginalPath != path1 {
 		t.Fatalf("OriginalPath was not set properly")
 	}
 
@@ -134,7 +134,7 @@ func TestConfigFiles(t *testing.T) {
 	if _, err := os.Stat(path1); err == nil {
 		t.Fatalf("Load with a config wrote a default config file")
 	}
-	if conf.OriginalPath != path2 {
+	if conf.Private.OriginalPath != path2 {
 		t.Fatalf("OriginalPath was not set properly")
 	}
 

config/private.go

@@ -0,0 +1,22 @@
+package config
+
+import "time"
+
+// Private contains configurations which are private, adding it to be part of config without exposing to customers.
+type Private struct {
+	// OAuthTokensPurgeInterval specifies the interval at which lapsed tokens get purged.
+	OAuthTokensPurgeInterval int `json:"-"`
+	// OriginalPath is the path to the config file that is read. If
+	// none was found, it's the path to the default config file that
+	// was written.
+	OriginalPath string `json:"-"`
+}
+
+// GetOAuthTokensPurgeInterval returns purge interval for lapsed OAuth tokens.
+func (p Private) GetOAuthTokensPurgeInterval() time.Duration {
+	if p.OAuthTokensPurgeInterval != 0 {
+		return time.Second * time.Duration(p.OAuthTokensPurgeInterval)
+	}
+
+	return time.Hour
+}

config/private_test.go

@@ -0,0 +1,20 @@
+package config
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPrivate_GetOAuthTokensPurgeInterval(t *testing.T) {
+	t.Run("default value", func(t *testing.T) {
+		p := Private{}
+		assert.Equal(t, time.Hour, p.GetOAuthTokensPurgeInterval())
+	})
+
+	t.Run("custom value", func(t *testing.T) {
+		p := Private{OAuthTokensPurgeInterval: 5}
+		assert.Equal(t, time.Second*5, p.GetOAuthTokensPurgeInterval())
+	})
+}

config/util.go

@@ -0,0 +1,83 @@
+package config
+
+import (
+	"errors"
+	"os"
+	"path"
+	"path/filepath"
+	"runtime"
+	"strconv"
+)
+
+// New produces a new config object by parsing
+// the default configuration for the values.
+func New() (*Config, error) {
+	cfg := new(Config)
+
+	cfgFile, err := findFile("tyk.conf")
+	if err != nil {
+		// Return cfg filled with environment
+		// if we don't have a config file.
+		if errors.Is(err, os.ErrNotExist) {
+			err := FillEnv(cfg)
+			return cfg, err
+		}
+
+		return nil, err
+	}
+
+	if err := Load([]string{cfgFile}, cfg); err != nil {
+		return nil, err
+	}
+
+	return cfg, nil
+}
+
+func findFile(filename string) (string, error) {
+	// Get folder in which current file lives
+	_, testFile, _, _ := runtime.Caller(0)
+	// Strip the filename and produce the dir
+	currentDir := path.Dir(testFile)
+
+	// Traverse the current directory and its parent directories
+	for {
+		// Check if the file exists in the current directory
+		filePath := filepath.Join(currentDir, filename)
+		_, err := os.Stat(filePath)
+		if err == nil {
+			// File found
+			return filePath, nil
+		}
+
+		// Move to the parent directory
+		parentDir := filepath.Dir(currentDir)
+		if parentDir == currentDir {
+			// No more parent directories remaining
+			break
+		}
+
+		currentDir = parentDir
+	}
+
+	// File not found
+	return "", os.ErrNotExist
+}
+
+// HostAddrs returns a sanitized list of hosts to connect to.
+func (config *StorageOptionsConf) HostAddrs() (addrs []string) {
+	if len(config.Addrs) != 0 {
+		addrs = config.Addrs
+	} else {
+		for h, p := range config.Hosts {
+			addr := h + ":" + p
+			addrs = append(addrs, addr)
+		}
+	}
+
+	if len(addrs) == 0 && config.Port != 0 {
+		addr := config.Host + ":" + strconv.Itoa(config.Port)
+		addrs = append(addrs, addr)
+	}
+
+	return addrs
+}

config/util_test.go

@@ -0,0 +1,63 @@
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHostAddrs(t *testing.T) {
+	tests := []struct {
+		name   string
+		config StorageOptionsConf
+		want   []string
+	}{
+		{
+			name:   "empty",
+			config: StorageOptionsConf{},
+		},
+		{
+			name: "addrs",
+			config: StorageOptionsConf{
+				Addrs: []string{"host1:1234", "host2:5678"},
+			},
+			want: []string{"host1:1234", "host2:5678"},
+		},
+		{
+			name: "hosts map",
+			config: StorageOptionsConf{
+				Hosts: map[string]string{
+					"host1": "1234",
+					"host2": "5678",
+				},
+			},
+			want: []string{"host1:1234", "host2:5678"},
+		},
+		{
+			name: "addrs and host maps",
+			config: StorageOptionsConf{
+				Addrs: []string{"host1:1234", "host2:5678"},
+				Hosts: map[string]string{
+					"host3": "1234",
+					"host4": "5678",
+				},
+			},
+			want: []string{"host1:1234", "host2:5678"},
+		},
+		{
+			name: "host and port",
+			config: StorageOptionsConf{
+				Host: "localhost",
+				Port: 6379,
+			},
+			want: []string{"localhost:6379"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.config.HostAddrs()
+			assert.ElementsMatch(t, tt.want, got)
+		})
+	}
+}

gateway/api_test.go

@@ -2043,7 +2043,7 @@ func TestOrgKeyHandler_LastUpdated(t *testing.T) {
 	}...)
 }
 
-func TestPurgeOAuthClientTokens(t *testing.T) {
+func TestPurgeOAuthClientTokensEndpoint(t *testing.T) {
 	conf := func(globalConf *config.Config) {
 		// set tokens to be expired after 1 second
 		globalConf.OauthTokenExpire = 1

gateway/oauth_manager.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"encoding/base64"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"math"
 	"net/http"
@@ -19,12 +18,22 @@ import (
 	"github.com/sirupsen/logrus"
 	"golang.org/x/crypto/bcrypt"
 
+<<<<<<< HEAD
 	"strconv"
 
 	"github.com/TykTechnologies/tyk/internal/uuid"
 
 	"github.com/TykTechnologies/tyk/headers"
 	tykerrors "github.com/TykTechnologies/tyk/internal/errors"
+=======
+	"github.com/TykTechnologies/tyk/internal/errors"
+	"github.com/TykTechnologies/tyk/internal/uuid"
+	"github.com/TykTechnologies/tyk/request"
+
+	"strconv"
+
+	"github.com/TykTechnologies/tyk/header"
+>>>>>>> ee5dc29b... [TT-10826] self trim oAuth sorted set (#5907)
 	"github.com/TykTechnologies/tyk/storage"
 	"github.com/TykTechnologies/tyk/user"
 )
@@ -1195,10 +1204,22 @@ func (gw *Gateway) purgeLapsedOAuthTokens() error {
 	}
 
 	redisCluster := &storage.RedisCluster{KeyPrefix: "", HashKeys: false, RedisController: gw.RedisController}
+
+	ok, err := redisCluster.Lock("oauth-purge-lock", time.Minute)
+	if err != nil {
+		log.WithError(err).Error("error acquiring lock to purge oauth tokens")
+		return err
+	}
+
+	if !ok {
+		log.Info("oauth tokens purge lock not acquired, purging in background")
+		return nil
+	}
+
 	keys, err := redisCluster.ScanKeys(oAuthClientTokensKeyPattern)
 
 	if err != nil {
-		log.WithError(err).Debug("error while scanning for tokens")
+		log.WithError(err).Error("error while scanning for tokens")
 		return err
 	}
 
@@ -1224,7 +1245,7 @@ func (gw *Gateway) purgeLapsedOAuthTokens() error {
 	close(errs)
 
 	combinedErr := &multierror.Error{
-		ErrorFormat: tykerrors.Formatter,
+		ErrorFormat: errors.Formatter,
 	}
 
 	for err := range errs {