Merge pull request #2524 from aristanetworks/mkelly-limit-gc-size

Limit the amount of objects we attempt to GC on each cycle

storage/kubernetes/client.go

@@ -15,6 +15,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"net/url"
 	"os"
 	"path"
 	"strconv"
@@ -82,7 +83,9 @@ func offlineTokenName(userID string, connID string, h func() hash.Hash) string {
 	return strings.TrimRight(encoding.EncodeToString(hash.Sum(nil)), "=")
 }
 
-func (cli *client) urlFor(apiVersion, namespace, resource, name string) string {
+func (cli *client) urlForWithParams(
+	apiVersion, namespace, resource, name string, params url.Values,
+) string {
 	basePath := "apis/"
 	if apiVersion == "v1" {
 		basePath = "api/"
@@ -94,10 +97,22 @@ func (cli *client) urlFor(apiVersion, namespace, resource, name string) string {
 	} else {
 		p = path.Join(basePath, apiVersion, resource, name)
 	}
-	if strings.HasSuffix(cli.baseURL, "/") {
+
-		return cli.baseURL + p
+	encodedParams := params.Encode()
+	paramsSuffix := ""
+	if len(encodedParams) > 0 {
+		paramsSuffix = "?" + encodedParams
 	}
-	return cli.baseURL + "/" + p
+
+	if strings.HasSuffix(cli.baseURL, "/") {
+		return cli.baseURL + p + paramsSuffix
+	}
+
+	return cli.baseURL + "/" + p + paramsSuffix
+}
+
+func (cli *client) urlFor(apiVersion, namespace, resource, name string) string {
+	return cli.urlForWithParams(apiVersion, namespace, resource, name, url.Values{})
 }
 
 // Define an error interface so we can get at the underlying status code if it's
@@ -163,8 +178,7 @@ func (cli *client) get(resource, name string, v interface{}) error {
 	return cli.getResource(cli.apiVersion, cli.namespace, resource, name, v)
 }
 
-func (cli *client) getResource(apiVersion, namespace, resource, name string, v interface{}) error {
+func (cli *client) getURL(url string, v interface{}) error {
-	url := cli.urlFor(apiVersion, namespace, resource, name)
 	resp, err := cli.client.Get(url)
 	if err != nil {
 		return err
@@ -176,6 +190,17 @@ func (cli *client) getResource(apiVersion, namespace, resource, name string, v i
 	return json.NewDecoder(resp.Body).Decode(v)
 }
 
+func (cli *client) getResource(apiVersion, namespace, resource, name string, v interface{}) error {
+	return cli.getURL(cli.urlFor(apiVersion, namespace, resource, name), v)
+}
+
+func (cli *client) listN(resource string, v interface{}, n int) error {
+	params := url.Values{}
+	params.Add("limit", fmt.Sprintf("%d", n))
+	u := cli.urlForWithParams(cli.apiVersion, cli.namespace, resource, "", params)
+	return cli.getURL(u, v)
+}
+
 func (cli *client) list(resource string, v interface{}) error {
 	return cli.get(resource, "", v)
 }

storage/kubernetes/storage.go

@@ -40,6 +40,10 @@ const (
 	resourceDeviceToken     = "devicetokens"
 )
 
+const (
+	gcResultLimit = 500
+)
+
 // Config values for the Kubernetes storage type.
 type Config struct {
 	InCluster      bool   `json:"inCluster"`
@@ -599,7 +603,7 @@ func (cli *client) UpdateConnector(id string, updater func(a storage.Connector)
 
 func (cli *client) GarbageCollect(now time.Time) (result storage.GCResult, err error) {
 	var authRequests AuthRequestList
-	if err := cli.list(resourceAuthRequest, &authRequests); err != nil {
+	if err := cli.listN(resourceAuthRequest, &authRequests, gcResultLimit); err != nil {
 		return result, fmt.Errorf("failed to list auth requests: %v", err)
 	}
 
@@ -618,7 +622,7 @@ func (cli *client) GarbageCollect(now time.Time) (result storage.GCResult, err e
 	}
 
 	var authCodes AuthCodeList
-	if err := cli.list(resourceAuthCode, &authCodes); err != nil {
+	if err := cli.listN(resourceAuthCode, &authCodes, gcResultLimit); err != nil {
 		return result, fmt.Errorf("failed to list auth codes: %v", err)
 	}
 
@@ -633,7 +637,7 @@ func (cli *client) GarbageCollect(now time.Time) (result storage.GCResult, err e
 	}
 
 	var deviceRequests DeviceRequestList
-	if err := cli.list(resourceDeviceRequest, &deviceRequests); err != nil {
+	if err := cli.listN(resourceDeviceRequest, &deviceRequests, gcResultLimit); err != nil {
 		return result, fmt.Errorf("failed to list device requests: %v", err)
 	}
 
@@ -648,7 +652,7 @@ func (cli *client) GarbageCollect(now time.Time) (result storage.GCResult, err e
 	}
 
 	var deviceTokens DeviceTokenList
-	if err := cli.list(resourceDeviceToken, &deviceTokens); err != nil {
+	if err := cli.listN(resourceDeviceToken, &deviceTokens, gcResultLimit); err != nil {
 		return result, fmt.Errorf("failed to list device tokens: %v", err)
 	}