Merge pull request #1 from rejoshed/feature/client-caching

Use client cache to reduce the number of client connections

Author: wongni

config/rbac/role.yaml

@@ -6,6 +6,14 @@ metadata:
   creationTimestamp: null
   name: manager-role
 rules:
+- apiGroups:
+  - ""
+  resources:
+  - configmaps
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resources:

controllers/cloudstackcluster_controller.go

@@ -39,6 +39,7 @@ import (
 // RBAC permissions used in all reconcilers. Events and Secrets.
 // "" empty string as the api group indicates core kubernetes objects. "*" indicates all objects.
 // +kubebuilder:rbac:groups="",resources=secrets;,verbs=get;list;watch
+// +kubebuilder:rbac:groups="",resources=configmaps;,verbs=get;list;watch
 // +kubebuilder:rbac:groups="",resources=events,verbs=get;list;watch;create;update;patch
 
 // RBAC permissions for CloudStackCluster.

controllers/utils/failuredomains.go

@@ -135,8 +135,12 @@ func (c *CloudClientImplementation) AsFailureDomainUser(fdSpec *infrav1.CloudSta
 			return ctrl.Result{}, errors.Wrapf(err, "getting ACSEndpoint secret with ref: %v", fdSpec.ACSEndpoint)
 		}
 
+		clientConfig := &corev1.ConfigMap{}
+		key = client.ObjectKey{Name: cloud.ClientConfigMapName, Namespace: cloud.ClientConfigMapNamespace}
+		_ = c.K8sClient.Get(c.RequestCtx, key, clientConfig)
+
 		var err error
-		if c.CSClient, err = cloud.NewClientFromK8sSecret(endpointCredentials); err != nil {
+		if c.CSClient, err = cloud.NewClientFromK8sSecret(endpointCredentials, clientConfig); err != nil {
 			return ctrl.Result{}, errors.Wrapf(err, "parsing ACSEndpoint secret with ref: %v", fdSpec.ACSEndpoint)
 		}
 

go.mod

@@ -3,6 +3,7 @@ module sigs.k8s.io/cluster-api-provider-cloudstack
 go 1.16
 
 require (
+	github.com/ReneKroon/ttlcache v1.7.0 // indirect
 	github.com/apache/cloudstack-go/v2 v2.13.0
 	github.com/go-logr/logr v1.2.3
 	github.com/golang/mock v1.6.0

go.sum

@@ -70,6 +70,8 @@ github.com/NYTimes/gziphandler v1.1.1/go.mod h1:n/CVRwUEOgIxrgPvAQhUUr9oeUtvrhMo
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0=
 github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE=
+github.com/ReneKroon/ttlcache v1.7.0 h1:8BkjFfrzVFXyrqnMtezAaJ6AHPSsVV10m6w28N/Fgkk=
+github.com/ReneKroon/ttlcache v1.7.0/go.mod h1:8BGGzdumrIjWxdRx8zpK6L3oGMWvIXdvB2GD1cfvd+I=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
@@ -600,6 +602,7 @@ go.starlark.net v0.0.0-20200306205701-8dd3e2ee1dd5/go.mod h1:nmDLcffg48OtT/PSW0H
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/goleak v0.10.0/go.mod h1:VCZuO8V8mFPlL0F5J5GK1rtHV3DrFcQ1R8ryq7FK0aI=
 go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/goleak v1.1.12 h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA=

pkg/cloud/client.go

@@ -18,14 +18,18 @@ package cloud
 
 import (
 	"bytes"
+	"fmt"
 	"io"
 	"os"
+	"sync"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 
 	"gopkg.in/yaml.v3"
 	"sigs.k8s.io/cluster-api-provider-cloudstack/pkg/metrics"
 
+	"github.com/ReneKroon/ttlcache"
 	"github.com/apache/cloudstack-go/v2/cloudstack"
 	"github.com/pkg/errors"
 )
@@ -66,6 +70,14 @@ type SecretConfig struct {
 	StringData Config            `yaml:"stringData"`
 }
 
+var clientCache *ttlcache.Cache
+var cacheMutex sync.Mutex
+
+const ClientConfigMapName = "capc-client-config"
+const ClientConfigMapNamespace = "capc-system"
+const ClientCacheTTLKey = "client-cache-ttl"
+const DefaultClientCacheTTL = time.Duration(1 * time.Hour)
+
 // UnmarshalAllSecretConfigs parses a yaml document for each secret.
 func UnmarshalAllSecretConfigs(in []byte, out *[]SecretConfig) error {
 	r := bytes.NewReader(in)
@@ -84,7 +96,8 @@ func UnmarshalAllSecretConfigs(in []byte, out *[]SecretConfig) error {
 	return nil
 }
 
-func NewClientFromK8sSecret(endpointSecret *corev1.Secret) (Client, error) {
+// NewClientFromK8sSecret returns a client from a k8s secret
+func NewClientFromK8sSecret(endpointSecret *corev1.Secret, clientConfig *corev1.ConfigMap) (Client, error) {
 	endpointSecretStrings := map[string]string{}
 	for k, v := range endpointSecret.Data {
 		endpointSecretStrings[k] = string(v)
@@ -93,19 +106,19 @@ func NewClientFromK8sSecret(endpointSecret *corev1.Secret) (Client, error) {
 	if err != nil {
 		return nil, err
 	}
-	return NewClientFromBytesConfig(bytes)
+	return NewClientFromBytesConfig(bytes, clientConfig)
 }
 
 // NewClientFromBytesConfig returns a client from a bytes array that unmarshals to a yaml config.
-func NewClientFromBytesConfig(conf []byte) (Client, error) {
+func NewClientFromBytesConfig(conf []byte, clientConfig *corev1.ConfigMap) (Client, error) {
 	r := bytes.NewReader(conf)
 	dec := yaml.NewDecoder(r)
 	var config Config
 	if err := dec.Decode(&config); err != nil {
 		return nil, err
 	}
 
-	return NewClientFromConf(config)
+	return NewClientFromConf(config, clientConfig)
 }
 
 // NewClientFromYamlPath returns a client from a yaml config at path.
@@ -129,11 +142,23 @@ func NewClientFromYamlPath(confPath string, secretName string) (Client, error) {
 		return nil, errors.Errorf("config with secret name %s not found", secretName)
 	}
 
-	return NewClientFromConf(conf)
+	return NewClientFromConf(conf, nil)
 }
 
-// Creates a new Cloud Client form a map of strings to strings.
-func NewClientFromConf(conf Config) (Client, error) {
+// NewClientFromConf creates a new Cloud Client form a map of strings to strings.
+func NewClientFromConf(conf Config, clientConfig *corev1.ConfigMap) (Client, error) {
+	cacheMutex.Lock()
+	defer cacheMutex.Unlock()
+
+	if clientCache == nil {
+		clientCache = newClientCache(clientConfig)
+	}
+
+	clientCacheKey := generateClientCacheKey(conf)
+	if client, exists := clientCache.Get(clientCacheKey); exists {
+		return client.(Client), nil
+	}
+
 	verifySSL := true
 	if conf.VerifySSL == "false" {
 		verifySSL = false
@@ -146,6 +171,8 @@ func NewClientFromConf(conf Config) (Client, error) {
 	c.cs = cloudstack.NewAsyncClient(conf.APIUrl, conf.APIKey, conf.SecretKey, verifySSL)
 	c.csAsync = cloudstack.NewClient(conf.APIUrl, conf.APIKey, conf.SecretKey, verifySSL)
 	c.customMetrics = metrics.NewCustomMetrics()
+	clientCache.Set(clientCacheKey, c)
+
 	return c, nil
 }
 
@@ -163,11 +190,38 @@ func (c *client) NewClientInDomainAndAccount(domain string, account string) (Cli
 	c.config.APIKey = user.APIKey
 	c.config.SecretKey = user.SecretKey
 
-	return NewClientFromConf(c.config)
+	return NewClientFromConf(c.config, nil)
 }
 
-// Create a client from a CloudStack-Go API client. Mostly used for testing.
+// NewClientFromCSAPIClient creates a client from a CloudStack-Go API client. Mostly used for testing.
 func NewClientFromCSAPIClient(cs *cloudstack.CloudStackClient) Client {
 	c := &client{cs: cs, csAsync: cs, customMetrics: metrics.NewCustomMetrics()}
 	return c
 }
+
+// generateClientCacheKey generates a cache key from a Config
+func generateClientCacheKey(conf Config) string {
+	return fmt.Sprintf("%+v", conf)
+}
+
+// newClientCache returns a new instance of client cache
+func newClientCache(clientConfig *corev1.ConfigMap) *ttlcache.Cache {
+	clientCache := ttlcache.NewCache()
+	clientCache.SetTTL(GetClientCacheTTL(clientConfig))
+	clientCache.SkipTtlExtensionOnHit(false)
+	return clientCache
+}
+
+// GetClientCacheTTL returns a client cache TTL duration from the passed config map
+func GetClientCacheTTL(clientConfig *corev1.ConfigMap) time.Duration {
+	var cacheTTL time.Duration
+	if clientConfig != nil {
+		if ttl, exists := clientConfig.Data[ClientCacheTTLKey]; exists {
+			cacheTTL, _ = time.ParseDuration(ttl)
+		}
+	}
+	if cacheTTL == 0 {
+		cacheTTL = DefaultClientCacheTTL
+	}
+	return cacheTTL
+}

pkg/cloud/client_test.go

@@ -18,6 +18,9 @@ package cloud_test
 
 import (
 	"os"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -36,17 +39,18 @@ var _ = Describe("Instance", func() {
 	var ()
 
 	BeforeEach(func() {
-		// This test fixture is useful for development, but the actual method of parsing is confinded to the client's
-		// new client method. The parsing used here is more of a schema, and we don't need to test another library's
-		// abilities to parse said schema.
-		Skip("Dev test suite.")
 	})
 
 	AfterEach(func() {
 	})
 
 	Context("When fetching a YAML config.", func() {
 		It("Handles the positive case.", func() {
+			// This test fixture is useful for development, but the actual method of parsing is confinded to the client's
+			// new client method. The parsing used here is more of a schema, and we don't need to test another library's
+			// abilities to parse said schema.
+			Skip("Dev test suite.")
+
 			// Create a real cloud client.
 			var connectionErr error
 			_, connectionErr = helpers.NewCSClient()
@@ -56,4 +60,97 @@ var _ = Describe("Instance", func() {
 			Ω(connectionErr).ShouldNot(HaveOccurred())
 		})
 	})
+
+	Context("GetClientCacheTTL", func() {
+		It("Returns the default TTL when a nil is passed", func() {
+			result := cloud.GetClientCacheTTL(nil)
+			Ω(result).Should(Equal(cloud.DefaultClientCacheTTL))
+		})
+
+		It("Returns the default TTL when an empty config map is passed", func() {
+			clientConfig := &corev1.ConfigMap{}
+			result := cloud.GetClientCacheTTL(clientConfig)
+			Ω(result).Should(Equal(cloud.DefaultClientCacheTTL))
+		})
+
+		It("Returns the default TTL when the TTL key does not exist", func() {
+			clientConfig := &corev1.ConfigMap{}
+			clientConfig.Data = map[string]string{}
+			clientConfig.Data[cloud.ClientCacheTTLKey+"XXXX"] = "1m5s"
+			result := cloud.GetClientCacheTTL(clientConfig)
+			Ω(result).Should(Equal(cloud.DefaultClientCacheTTL))
+		})
+
+		It("Returns the default TTL when the TTL value is invalid", func() {
+			clientConfig := &corev1.ConfigMap{}
+			clientConfig.Data = map[string]string{}
+			clientConfig.Data[cloud.ClientCacheTTLKey] = "5mXXX"
+			result := cloud.GetClientCacheTTL(clientConfig)
+			Ω(result).Should(Equal(cloud.DefaultClientCacheTTL))
+		})
+
+		It("Returns the TTL from the input clientConfig map", func() {
+			clientConfig := &corev1.ConfigMap{}
+			clientConfig.Data = map[string]string{}
+			clientConfig.Data[cloud.ClientCacheTTLKey] = "5m10s"
+			expected, _ := time.ParseDuration("5m10s")
+			result := cloud.GetClientCacheTTL(clientConfig)
+			Ω(result).Should(Equal(expected))
+		})
+	})
+
+	Context("NewClientFromConf", func() {
+		clientConfig := &corev1.ConfigMap{}
+
+		BeforeEach(func() {
+			clientConfig.Data = map[string]string{}
+			clientConfig.Data[cloud.ClientCacheTTLKey] = "100ms"
+		})
+
+		It("Returns a new client", func() {
+			config := cloud.Config{
+				APIUrl: "http://1.1.1.1",
+			}
+			result, err := cloud.NewClientFromConf(config, clientConfig)
+			Ω(err).ShouldNot(HaveOccurred())
+			Ω(result).ShouldNot(BeNil())
+		})
+
+		It("Returns a new client for a different config", func() {
+			config1 := cloud.Config{
+				APIUrl: "http://2.2.2.2",
+			}
+			config2 := cloud.Config{
+				APIUrl: "http://3.3.3.3",
+			}
+			result1, _ := cloud.NewClientFromConf(config1, clientConfig)
+			result2, _ := cloud.NewClientFromConf(config2, clientConfig)
+			Ω(result1).ShouldNot(Equal(result2))
+		})
+
+		It("Returns a cached client for the same config", func() {
+			config1 := cloud.Config{
+				APIUrl: "http://4.4.4.4",
+			}
+			config2 := cloud.Config{
+				APIUrl: "http://4.4.4.4",
+			}
+			result1, _ := cloud.NewClientFromConf(config1, clientConfig)
+			result2, _ := cloud.NewClientFromConf(config2, clientConfig)
+			Ω(result1).Should(Equal(result2))
+		})
+
+		It("Returns a new client after cache expiration", func() {
+			config1 := cloud.Config{
+				APIUrl: "http://5.5.5.5",
+			}
+			config2 := cloud.Config{
+				APIUrl: "http://5.5.5.5",
+			}
+			result1, _ := cloud.NewClientFromConf(config1, clientConfig)
+			time.Sleep(150 * time.Millisecond)
+			result2, _ := cloud.NewClientFromConf(config2, clientConfig)
+			Ω(result1).ShouldNot(Equal(result2))
+		})
+	})
 })