feat(authentication): per method cleanup process

cmd/flipt/main.go

@@ -29,6 +29,7 @@ import (
 	"github.com/phyber/negroni-gzip/gzip"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/cobra"
+	"go.flipt.io/flipt/internal/cleanup"
 	"go.flipt.io/flipt/internal/config"
 	"go.flipt.io/flipt/internal/info"
 	"go.flipt.io/flipt/internal/server"
@@ -41,6 +42,7 @@ import (
 	"go.flipt.io/flipt/internal/storage"
 	authstorage "go.flipt.io/flipt/internal/storage/auth"
 	authsql "go.flipt.io/flipt/internal/storage/auth/sql"
+	oplocksql "go.flipt.io/flipt/internal/storage/oplock/sql"
 	"go.flipt.io/flipt/internal/storage/sql"
 	"go.flipt.io/flipt/internal/storage/sql/mysql"
 	"go.flipt.io/flipt/internal/storage/sql/postgres"
@@ -455,7 +457,21 @@ func run(ctx context.Context, logger *zap.Logger) error {
 			otelgrpc.UnaryServerInterceptor(),
 		}
 
-		authenticationStore := authsql.NewStore(driver, sql.BuilderFor(db, driver), logger)
+		var (
+			sqlBuilder           = sql.BuilderFor(db, driver)
+			authenticationStore  = authsql.NewStore(driver, sqlBuilder, logger)
+			operationLockService = oplocksql.New(logger, driver, sqlBuilder)
+		)
+
+		if cfg.Authentication.ShouldRunCleanup() {
+			cleanupAuthService := cleanup.NewAuthenticationService(logger, operationLockService, authenticationStore, cfg.Authentication)
+			cleanupAuthService.Run(ctx)
+
+			shutdownFuncs = append(shutdownFuncs, func(context.Context) {
+				_ = cleanupAuthService.Stop()
+				logger.Info("cleanup service has been shutdown")
+			})
+		}
 
 		// only enable enforcement middleware if authentication required
 		if cfg.Authentication.Required {

config/migrations/cockroachdb/2_create_table_operation_lock.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS operation_lock;

config/migrations/cockroachdb/2_create_table_operation_lock.up.sql

@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS operation_lock (
+  operation VARCHAR(255) PRIMARY KEY UNIQUE NOT NULL,
+  version INTEGER DEFAULT 0 NOT NULL,
+  last_acquired_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  acquired_until TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);

config/migrations/mysql/3_create_table_operation_lock.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS operation_lock;

config/migrations/mysql/3_create_table_operation_lock.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE IF NOT EXISTS operation_lock (
+  operation VARCHAR(255) UNIQUE NOT NULL,
+  version INTEGER DEFAULT 0 NOT NULL,
+  last_acquired_at TIMESTAMP(6) DEFAULT CURRENT_TIMESTAMP(6),
+  acquired_until TIMESTAMP(6) DEFAULT CURRENT_TIMESTAMP(6),
+  PRIMARY KEY (`operation`)
+);

config/migrations/postgres/5_create_table_operation_lock.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS operation_lock;

config/migrations/postgres/5_create_table_operation_lock.up.sql

@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS operation_lock (
+  operation VARCHAR(255) PRIMARY KEY UNIQUE NOT NULL,
+  version INTEGER DEFAULT 0 NOT NULL,
+  last_acquired_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  acquired_until TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);

config/migrations/sqlite3/5_create_table_operation_lock.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS operation_lock;

config/migrations/sqlite3/5_create_table_operation_lock.up.sql

@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS operation_lock (
+  operation VARCHAR(255) PRIMARY KEY UNIQUE NOT NULL,
+  version INTEGER DEFAULT 0 NOT NULL,
+  last_acquired_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+  acquired_until TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);

errors/errors.go

@@ -5,6 +5,12 @@ import (
 	"fmt"
 )
 
+// As is a utility for one-lining errors.As statements.
+// e.g. cerr, match := errors.As[MyCustomError](err).
+func As[E error](err error) (e E, _ bool) {
+	return e, errors.As(err, &e)
+}
+
 // New creates a new error with errors.New
 func New(s string) error {
 	return errors.New(s)

internal/cleanup/cleanup.go

@@ -0,0 +1,114 @@
+package cleanup
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"go.flipt.io/flipt/internal/config"
+	authstorage "go.flipt.io/flipt/internal/storage/auth"
+	"go.flipt.io/flipt/internal/storage/oplock"
+	"go.flipt.io/flipt/rpc/flipt/auth"
+	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
+)
+
+const minCleanupInterval = 5 * time.Minute
+
+// AuthenticationService is configured to run background goroutines which
+// will clear out expired authentication tokens.
+type AuthenticationService struct {
+	logger *zap.Logger
+	lock   oplock.Service
+	store  authstorage.Store
+	config config.AuthenticationConfig
+
+	errgroup errgroup.Group
+	cancel   func()
+}
+
+// NewAuthenticationService constructs and configures a new instance of authentication service.
+func NewAuthenticationService(logger *zap.Logger, lock oplock.Service, store authstorage.Store, config config.AuthenticationConfig) *AuthenticationService {
+	return &AuthenticationService{
+		logger: logger,
+		lock:   lock,
+		store:  store,
+		config: config,
+		cancel: func() {},
+	}
+}
+
+func (s *AuthenticationService) schedules() map[auth.Method]config.AuthenticationCleanupSchedule {
+	schedules := map[auth.Method]config.AuthenticationCleanupSchedule{}
+	if s.config.Methods.Token.Cleanup != nil {
+		schedules[auth.Method_METHOD_TOKEN] = *s.config.Methods.Token.Cleanup
+	}
+
+	return schedules
+}
+
+// Run starts up a background goroutine per configure authentication method schedule.
+func (s *AuthenticationService) Run(ctx context.Context) {
+	ctx, s.cancel = context.WithCancel(ctx)
+
+	for method, schedule := range s.schedules() {
+		var (
+			method    = method
+			schedule  = schedule
+			operation = oplock.Operation(fmt.Sprintf("cleanup_auth_%s", method))
+		)
+
+		s.errgroup.Go(func() error {
+			// on the first attempt to run the cleanup authentication service
+			// we attempt to obtain the lock immediately. If the lock is already
+			// held the service should return false and return the current acquired
+			// current timestamp
+			acquiredUntil := time.Now().UTC()
+			for {
+				select {
+				case <-ctx.Done():
+					return nil
+				case <-time.After(time.Until(acquiredUntil)):
+				}
+
+				acquired, entry, err := s.lock.TryAcquire(ctx, operation, schedule.Interval)
+				if err != nil {
+					// ensure we dont go into hot loop when the operation lock service
+					// enters an error state by ensuring we sleep for at-least the minimum
+					// interval.
+					now := time.Now().UTC()
+					if acquiredUntil.Before(now) {
+						acquiredUntil = now.Add(minCleanupInterval)
+					}
+
+					s.logger.Warn("attempting to acquire lock", zap.Error(err))
+					continue
+				}
+
+				// update the next sleep target to current entries acquired until
+				acquiredUntil = entry.AcquiredUntil
+
+				if !acquired {
+					s.logger.Info("cleanup process not acquired", zap.Time("next_attempt", entry.AcquiredUntil))
+					continue
+				}
+
+				expiredBefore := time.Now().UTC().Add(-schedule.GracePeriod)
+				s.logger.Info("cleanup process deleting authentications", zap.Time("expired_before", expiredBefore))
+				if err := s.store.DeleteAuthentications(ctx, authstorage.Delete(
+					authstorage.WithMethod(method),
+					authstorage.WithExpiredBefore(expiredBefore),
+				)); err != nil {
+					s.logger.Error("attempting to delete expired authentications", zap.Error(err))
+				}
+			}
+		})
+	}
+}
+
+// Stop signals for the cleanup goroutines to cancel and waits for them to finish.
+func (s *AuthenticationService) Stop() error {
+	s.cancel()
+
+	return s.errgroup.Wait()
+}

internal/cleanup/cleanup_test.go

@@ -0,0 +1,95 @@
+package cleanup
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.flipt.io/flipt/internal/config"
+	authstorage "go.flipt.io/flipt/internal/storage/auth"
+	inmemauth "go.flipt.io/flipt/internal/storage/auth/memory"
+	inmemoplock "go.flipt.io/flipt/internal/storage/oplock/memory"
+	"go.flipt.io/flipt/rpc/flipt/auth"
+	"go.uber.org/zap/zaptest"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+func TestCleanup(t *testing.T) {
+	var (
+		ctx        = context.Background()
+		logger     = zaptest.NewLogger(t)
+		authstore  = inmemauth.NewStore()
+		lock       = inmemoplock.New()
+		authConfig = config.AuthenticationConfig{
+			Methods: config.AuthenticationMethods{
+				Token: config.AuthenticationMethodTokenConfig{
+					Enabled: true,
+					Cleanup: &config.AuthenticationCleanupSchedule{
+						Interval:    time.Second,
+						GracePeriod: 5 * time.Second,
+					},
+				},
+			},
+		}
+	)
+
+	// create an initial non-expiring token
+	clientToken, storedAuth, err := authstore.CreateAuthentication(
+		ctx,
+		&authstorage.CreateAuthenticationRequest{Method: auth.Method_METHOD_TOKEN},
+	)
+	require.NoError(t, err)
+
+	for i := 0; i < 5; i++ {
+		// run five instances of service
+		// it should be a safe operation given they share the same lock service
+		service := NewAuthenticationService(logger, lock, authstore, authConfig)
+		service.Run(ctx)
+		defer func() {
+			require.NoError(t, service.Stop())
+		}()
+	}
+
+	t.Run("ensure non-expiring token exists", func(t *testing.T) {
+		retrievedAuth, err := authstore.GetAuthenticationByClientToken(ctx, clientToken)
+		require.NoError(t, err)
+		assert.Equal(t, storedAuth, retrievedAuth)
+	})
+
+	t.Run("create an expiring token and ensure it exists", func(t *testing.T) {
+		clientToken, storedAuth, err = authstore.CreateAuthentication(
+			ctx,
+			&authstorage.CreateAuthenticationRequest{
+				Method:    auth.Method_METHOD_TOKEN,
+				ExpiresAt: timestamppb.New(time.Now().UTC().Add(5 * time.Second)),
+			},
+		)
+		require.NoError(t, err)
+
+		retrievedAuth, err := authstore.GetAuthenticationByClientToken(ctx, clientToken)
+		require.NoError(t, err)
+		assert.Equal(t, storedAuth, retrievedAuth)
+	})
+
+	t.Run("ensure grace period protects token from being deleted", func(t *testing.T) {
+		// token should still exist as it wont be deleted until
+		// expiry + grace period (5s + 5s == 10s)
+		time.Sleep(5 * time.Second)
+
+		retrievedAuth, err := authstore.GetAuthenticationByClientToken(ctx, clientToken)
+		require.NoError(t, err)
+		assert.Equal(t, storedAuth, retrievedAuth)
+
+		// ensure authentication is expired but still persisted
+		assert.True(t, retrievedAuth.ExpiresAt.AsTime().Before(time.Now().UTC()))
+	})
+
+	t.Run("once expiry and grace period ellapses ensure token is deleted", func(t *testing.T) {
+		time.Sleep(10 * time.Second)
+
+		_, err := authstore.GetAuthenticationByClientToken(ctx, clientToken)
+		require.Error(t, err, "resource not found")
+	})
+}

internal/config/authentication.go

@@ -1,8 +1,28 @@
 package config
 
-import "github.com/spf13/viper"
+import (
+	"strings"
+	"time"
 
-var _ defaulter = (*AuthenticationConfig)(nil)
+	"github.com/spf13/viper"
+	"go.flipt.io/flipt/rpc/flipt/auth"
+)
+
+var (
+	_                  defaulter = (*AuthenticationConfig)(nil)
+	stringToAuthMethod           = map[string]auth.Method{}
+)
+
+func init() {
+	for method, v := range auth.Method_value {
+		if auth.Method(v) == auth.Method_METHOD_NONE {
+			continue
+		}
+
+		name := strings.ToLower(strings.TrimPrefix(method, "METHOD_"))
+		stringToAuthMethod[name] = auth.Method(v)
+	}
+}
 
 // AuthenticationConfig configures Flipts authentication mechanisms
 type AuthenticationConfig struct {
@@ -11,30 +31,57 @@ type AuthenticationConfig struct {
 	// Else, authentication is not required and Flipt's APIs are not secured.
 	Required bool `json:"required,omitempty" mapstructure:"required"`
 
-	Methods struct {
-		Token AuthenticationMethodTokenConfig `json:"token,omitempty" mapstructure:"token"`
-	} `json:"methods,omitempty" mapstructure:"methods"`
+	Methods AuthenticationMethods `json:"methods,omitempty" mapstructure:"methods"`
+}
+
+// ShouldRunCleanup returns true if the cleanup background process should be started.
+// It returns true given at-least 1 method is enabled and it's associated schedule
+// has been configured (non-nil).
+func (c AuthenticationConfig) ShouldRunCleanup() bool {
+	return (c.Methods.Token.Enabled && c.Methods.Token.Cleanup != nil)
 }
 
-func (a *AuthenticationConfig) setDefaults(v *viper.Viper) []string {
+func (c *AuthenticationConfig) setDefaults(v *viper.Viper) []string {
+	token := map[string]any{
+		"enabled": false,
+	}
+
+	if v.GetBool("authentication.methods.token.enabled") {
+		token["cleanup"] = map[string]any{
+			"interval":     time.Hour,
+			"grace_period": 30 * time.Minute,
+		}
+	}
+
 	v.SetDefault("authentication", map[string]any{
 		"required": false,
 		"methods": map[string]any{
-			"token": map[string]any{
-				"enabled": false,
-			},
+			"token": token,
 		},
 	})
 
 	return nil
 }
 
+// AuthenticationMethods is a set of configuration for each authentication
+// method available for use within Flipt.
+type AuthenticationMethods struct {
+	Token AuthenticationMethodTokenConfig `json:"token,omitempty" mapstructure:"token"`
+}
+
 // AuthenticationMethodTokenConfig contains fields used to configure the authentication
 // method "token".
 // This authentication method supports the ability to create static tokens via the
 // /auth/v1/method/token prefix of endpoints.
 type AuthenticationMethodTokenConfig struct {
 	// Enabled designates whether or not static token authentication is enabled
 	// and whether Flipt will mount the "token" method APIs.
-	Enabled bool `json:"enabled,omitempty" mapstructure:"enabled"`
+	Enabled bool                           `json:"enabled,omitempty" mapstructure:"enabled"`
+	Cleanup *AuthenticationCleanupSchedule `json:"cleanup,omitempty" mapstructure:"cleanup"`
+}
+
+// AuthenticationCleanupSchedule is used to configure a cleanup goroutine.
+type AuthenticationCleanupSchedule struct {
+	Interval    time.Duration `json:"interval,omitempty" mapstructure:"interval"`
+	GracePeriod time.Duration `json:"gracePeriod,omitempty" mapstructure:"grace_period"`
 }