storage/kubernetes: add garbage collection method
This commit is contained in:
Eric Chiang
parent
9ce05ecf73
commit
558059ee58
5 changed files with 39 additions and 163 deletions
|
|
@ -20,7 +20,6 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/gtank/cryptopasta"
|
||||
"golang.org/x/net/context"
|
||||
yaml "gopkg.in/yaml.v2"
|
||||
|
||||
"github.com/coreos/dex/storage"
|
||||
|
|
@ -35,9 +34,6 @@ type client struct {
|
|||
|
||||
now func() time.Time
|
||||
|
||||
// If not nil, the cancel function for stopping garbage colletion.
|
||||
cancel context.CancelFunc
|
||||
|
||||
// BUG: currently each third party API group can only have one resource in it,
|
||||
// so for each resource this storage uses, it need a unique API group.
|
||||
//
|
||||
|
|
|
|||
|
|
@ -1,58 +0,0 @@
|
|||
package kubernetes
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"golang.org/x/net/context"
|
||||
)
|
||||
|
||||
// gc begins the gc process for Kubernetes.
|
||||
func (cli *client) gc(ctx context.Context, every time.Duration) {
|
||||
handleErr := func(err error) { log.Println(err.Error()) }
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(every):
|
||||
}
|
||||
|
||||
// TODO(ericchiang): On failures, run garbage collection more often.
|
||||
log.Println("kubernetes: running garbage collection")
|
||||
cli.gcAuthRequests(handleErr)
|
||||
cli.gcAuthCodes(handleErr)
|
||||
log.Printf("kubernetes: garbage collection finished, next run at %s", cli.now().Add(every))
|
||||
}
|
||||
}
|
||||
|
||||
func (cli *client) gcAuthRequests(handleErr func(error)) {
|
||||
var authRequests AuthRequestList
|
||||
if err := cli.list(resourceAuthRequest, &authRequests); err != nil {
|
||||
handleErr(fmt.Errorf("failed to list auth requests: %v", err))
|
||||
return
|
||||
}
|
||||
for _, authRequest := range authRequests.AuthRequests {
|
||||
if cli.now().After(authRequest.Expiry) {
|
||||
if err := cli.delete(resourceAuthRequest, authRequest.ObjectMeta.Name); err != nil {
|
||||
handleErr(fmt.Errorf("failed to detele auth request: %v", err))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (cli *client) gcAuthCodes(handleErr func(error)) {
|
||||
var authCodes AuthCodeList
|
||||
if err := cli.list(resourceAuthCode, &authCodes); err != nil {
|
||||
handleErr(fmt.Errorf("failed to list auth codes: %v", err))
|
||||
return
|
||||
}
|
||||
for _, authCode := range authCodes.AuthCodes {
|
||||
if cli.now().After(authCode.Expiry) {
|
||||
if err := cli.delete(resourceAuthCode, authCode.ObjectMeta.Name); err != nil {
|
||||
handleErr(fmt.Errorf("failed to delete auth code: %v", err))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,88 +0,0 @@
|
|||
package kubernetes
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/coreos/dex/storage"
|
||||
)
|
||||
|
||||
func muster(t *testing.T) func(err error) {
|
||||
return func(err error) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGCAuthRequests(t *testing.T) {
|
||||
cli := loadClient(t)
|
||||
must := muster(t)
|
||||
|
||||
now := time.Now()
|
||||
cli.now = func() time.Time { return now }
|
||||
|
||||
expiredID := storage.NewID()
|
||||
goodID := storage.NewID()
|
||||
|
||||
must(cli.CreateAuthRequest(storage.AuthRequest{
|
||||
ID: expiredID,
|
||||
Expiry: now.Add(-time.Second),
|
||||
}))
|
||||
|
||||
must(cli.CreateAuthRequest(storage.AuthRequest{
|
||||
ID: goodID,
|
||||
Expiry: now.Add(time.Second),
|
||||
}))
|
||||
|
||||
handleErr := func(err error) { t.Error(err.Error()) }
|
||||
cli.gcAuthRequests(handleErr)
|
||||
|
||||
if _, err := cli.GetAuthRequest(goodID); err != nil {
|
||||
t.Errorf("failed to get good auth ID: %v", err)
|
||||
}
|
||||
_, err := cli.GetAuthRequest(expiredID)
|
||||
switch {
|
||||
case err == nil:
|
||||
t.Errorf("gc did not remove expired auth request")
|
||||
case err == storage.ErrNotFound:
|
||||
default:
|
||||
t.Errorf("expected storage.ErrNotFound, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGCAuthCodes(t *testing.T) {
|
||||
cli := loadClient(t)
|
||||
must := muster(t)
|
||||
|
||||
now := time.Now()
|
||||
cli.now = func() time.Time { return now }
|
||||
|
||||
expiredID := storage.NewID()
|
||||
goodID := storage.NewID()
|
||||
|
||||
must(cli.CreateAuthCode(storage.AuthCode{
|
||||
ID: expiredID,
|
||||
Expiry: now.Add(-time.Second),
|
||||
}))
|
||||
|
||||
must(cli.CreateAuthCode(storage.AuthCode{
|
||||
ID: goodID,
|
||||
Expiry: now.Add(time.Second),
|
||||
}))
|
||||
|
||||
handleErr := func(err error) { t.Error(err.Error()) }
|
||||
cli.gcAuthCodes(handleErr)
|
||||
|
||||
if _, err := cli.GetAuthCode(goodID); err != nil {
|
||||
t.Errorf("failed to get good auth ID: %v", err)
|
||||
}
|
||||
_, err := cli.GetAuthCode(expiredID)
|
||||
switch {
|
||||
case err == nil:
|
||||
t.Errorf("gc did not remove expired auth request")
|
||||
case err == storage.ErrNotFound:
|
||||
default:
|
||||
t.Errorf("expected storage.ErrNotFound, got %v", err)
|
||||
}
|
||||
}
|
||||
|
|
@ -3,12 +3,12 @@ package kubernetes
|
|||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
homedir "github.com/mitchellh/go-homedir"
|
||||
"golang.org/x/net/context"
|
||||
|
||||
"github.com/coreos/dex/storage"
|
||||
"github.com/coreos/dex/storage/kubernetes/k8sapi"
|
||||
|
|
@ -46,14 +46,6 @@ func (c *Config) Open() (storage.Storage, error) {
|
|||
return nil, err
|
||||
}
|
||||
|
||||
// start up garbage collection
|
||||
gcFrequency := c.GCFrequency
|
||||
if gcFrequency == 0 {
|
||||
gcFrequency = 600
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cli.cancel = cancel
|
||||
go cli.gc(ctx, time.Duration(gcFrequency)*time.Second)
|
||||
return cli, nil
|
||||
}
|
||||
|
||||
|
|
@ -93,9 +85,6 @@ func (c *Config) open() (*client, error) {
|
|||
}
|
||||
|
||||
func (cli *client) Close() error {
|
||||
if cli.cancel != nil {
|
||||
cli.cancel()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -291,3 +280,40 @@ func (cli *client) UpdateAuthRequest(id string, updater func(a storage.AuthReque
|
|||
newReq.ObjectMeta = req.ObjectMeta
|
||||
return cli.put(resourceAuthRequest, id, newReq)
|
||||
}
|
||||
|
||||
func (cli *client) GarbageCollect(now time.Time) (result storage.GCResult, err error) {
|
||||
var authRequests AuthRequestList
|
||||
if err := cli.list(resourceAuthRequest, &authRequests); err != nil {
|
||||
return result, fmt.Errorf("failed to list auth requests: %v", err)
|
||||
}
|
||||
|
||||
var delErr error
|
||||
for _, authRequest := range authRequests.AuthRequests {
|
||||
if now.After(authRequest.Expiry) {
|
||||
if err := cli.delete(resourceAuthRequest, authRequest.ObjectMeta.Name); err != nil {
|
||||
log.Printf("failed to delete auth request: %v", err)
|
||||
delErr = fmt.Errorf("failed to delete auth request: %v", err)
|
||||
}
|
||||
result.AuthRequests++
|
||||
}
|
||||
}
|
||||
if delErr != nil {
|
||||
return result, delErr
|
||||
}
|
||||
|
||||
var authCodes AuthCodeList
|
||||
if err := cli.list(resourceAuthCode, &authCodes); err != nil {
|
||||
return result, fmt.Errorf("failed to list auth codes: %v", err)
|
||||
}
|
||||
|
||||
for _, authCode := range authCodes.AuthCodes {
|
||||
if now.After(authCode.Expiry) {
|
||||
if err := cli.delete(resourceAuthCode, authCode.ObjectMeta.Name); err != nil {
|
||||
log.Printf("failed to delete auth code %v", err)
|
||||
delErr = fmt.Errorf("failed to delete auth code: %v", err)
|
||||
}
|
||||
result.AuthCodes++
|
||||
}
|
||||
}
|
||||
return result, delErr
|
||||
}
|
||||
|
|
|
|||
|
|
@ -74,7 +74,7 @@ func TestURLFor(t *testing.T) {
|
|||
|
||||
func TestStorage(t *testing.T) {
|
||||
client := loadClient(t)
|
||||
conformance.RunTestSuite(t, func() storage.Storage {
|
||||
conformance.RunTests(t, func() storage.Storage {
|
||||
for _, resource := range []string{
|
||||
resourceAuthCode,
|
||||
resourceAuthRequest,
|
||||
|
|
|
|||
Reference in a new issue