investigating memory leak

core/clustersmngr/cluster/cluster.go

@@ -3,10 +3,12 @@ package cluster
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"os"
 	"time"
 
 	"github.com/weaveworks/weave-gitops/pkg/server/auth"
+	machnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/cli-utils/pkg/flowcontrol"
@@ -28,8 +30,9 @@ const (
 )
 
 var (
-	kubeClientTimeout        = getEnvDuration("WEAVE_GITOPS_KUBE_CLIENT_TIMEOUT", 30*time.Second)
-	DefaultKubeConfigOptions = []KubeConfigOption{WithFlowControl}
+	kubeClientTimeout = getEnvDuration("WEAVE_GITOPS_KUBE_CLIENT_TIMEOUT", 30*time.Second)
+	// DefaultKubeConfigOptions = []KubeConfigOption{WithFlowControl}
+	DefaultKubeConfigOptions = []KubeConfigOption{}
 )
 
 type KubeConfigOption func(*rest.Config) (*rest.Config, error)
@@ -77,6 +80,11 @@ func WithFlowControl(config *rest.Config) (*rest.Config, error) {
 	config.QPS = ClientQPS
 	config.Burst = ClientBurst
 
+	// From https://github.com/weaveworks/weave-gitops-enterprise/issues/3189
+	// Suggested in https://github.com/kubernetes/kubernetes/issues/118703#issuecomment-1595072383
+	// TODO: Revert or adapt when upstream fix is available
+	config.Proxy = machnet.NewProxierWithNoProxyCIDR(http.ProxyFromEnvironment)
+
 	return config, nil
 }
 

core/clustersmngr/cluster/single.go

@@ -3,10 +3,12 @@ package cluster
 import (
 	"fmt"
 	"net"
+	"net/http"
 
 	"github.com/weaveworks/weave-gitops/pkg/kube"
 	"github.com/weaveworks/weave-gitops/pkg/server/auth"
 	apiruntime "k8s.io/apimachinery/pkg/runtime"
+	machnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -60,6 +62,11 @@ func getClientFromConfig(config *rest.Config, scheme *apiruntime.Scheme) (client
 		return nil, fmt.Errorf("could not create RESTMapper from config: %w", err)
 	}
 
+	// From https://github.com/weaveworks/weave-gitops-enterprise/issues/3189
+	// Suggested in https://github.com/kubernetes/kubernetes/issues/118703#issuecomment-1595072383
+	// TODO: Revert or adapt when upstream fix is available
+	config.Proxy = machnet.NewProxierWithNoProxyCIDR(http.ProxyFromEnvironment)
+
 	client, err := client.New(config, client.Options{
 		Scheme: scheme,
 		Mapper: mapper,

core/clustersmngr/factory.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"os"
 	"sync"
 	"time"
 
@@ -34,25 +33,9 @@ const (
 )
 
 var (
-	usersClientsTTL = getEnvDuration("WEAVE_GITOPS_USERS_CLIENTS_TTL", 30*time.Minute)
+	usersClientsTTL = 24 * time.Hour
 )
 
-func getEnvDuration(key string, defaultDuration time.Duration) time.Duration {
-	val := os.Getenv(key)
-	if val == "" {
-		return defaultDuration
-	}
-
-	d, err := time.ParseDuration(val)
-
-	// on error return the default duration
-	if err != nil {
-		return defaultDuration
-	}
-
-	return d
-}
-
 var (
 	opsUpdateClusters = prometheus.NewCounter(
 		prometheus.CounterOpts{
@@ -228,7 +211,7 @@ func NewClustersManager(fetchers []ClusterFetcher, nsChecker nsaccess.Checker, l
 		clusters:                   &Clusters{},
 		clustersNamespaces:         &ClustersNamespaces{},
 		usersNamespaces:            &UsersNamespaces{Cache: ttlcache.New(userNamespaceResolution)},
-		usersClients:               &UsersClients{Cache: ttlcache.New(usersClientResolution)},
+		usersClients:               &UsersClients{Cache: ttlcache.New(usersClientResolution), log: logger},
 		log:                        logger,
 		initialClustersLoad:        make(chan bool),
 		watchers:                   []*ClustersWatcher{},
@@ -609,11 +592,15 @@ func (cf *clustersManager) getOrCreateClient(user *auth.UserPrincipal, cluster c
 		}
 		isServer = true
 	}
+	cf.log.Info("getOrCreateClient", "user", user.ID, "cluster", cluster.GetName())
 
 	if client, found := cf.usersClients.Get(user, cluster.GetName()); found {
+		cf.log.Info("found cached client")
 		return client, nil
 	}
 
+	cf.log.Info("client not found in cache so creating ", "user", user.ID, "cluster", cluster.GetName())
+
 	var (
 		client client.Client
 		err    error
@@ -622,16 +609,19 @@ func (cf *clustersManager) getOrCreateClient(user *auth.UserPrincipal, cluster c
 	if isServer {
 		opsCreateServerClient.WithLabelValues(cluster.GetName()).Inc()
 		client, err = cluster.GetServerClient()
+		cf.log.Info("created server client", "user", user.ID, "cluster", cluster.GetName())
 	} else {
 		opsCreateUserClient.WithLabelValues(cluster.GetName()).Inc()
 		client, err = cluster.GetUserClient(user)
+		cf.log.Info("created user client", "user", user.ID, "cluster", cluster.GetName())
 	}
 
 	if err != nil {
 		return nil, fmt.Errorf("failed creating client for cluster=%s: %w", cluster.GetName(), err)
 	}
 
 	cf.usersClients.Set(user, cluster.GetName(), client)
+	cf.log.Info("set client", "user", user.ID, "cluster", cluster.GetName())
 
 	return client, nil
 }

core/clustersmngr/factory_caches.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	"github.com/cheshir/ttlcache"
+	"github.com/go-logr/logr"
 	"github.com/weaveworks/weave-gitops/core/clustersmngr/cluster"
 	"github.com/weaveworks/weave-gitops/pkg/server/auth"
 	v1 "k8s.io/api/core/v1"
@@ -155,21 +156,36 @@ func (un UsersNamespaces) cacheKey(user *auth.UserPrincipal, cluster string) uin
 
 type UsersClients struct {
 	Cache *ttlcache.Cache
+	log   logr.Logger
 }
 
 func (uc *UsersClients) cacheKey(user *auth.UserPrincipal, clusterName string) uint64 {
 	return ttlcache.StringKey(fmt.Sprintf("%s:%s-%s", user.ID, user.Hash(), clusterName))
 }
 
 func (uc *UsersClients) Set(user *auth.UserPrincipal, clusterName string, client client.Client) {
-	uc.Cache.Set(uc.cacheKey(user, clusterName), client, usersClientsTTL)
+	cacheKey := uc.cacheKey(user, clusterName)
+	uc.log.Info("set cached connection", "user", user, "cluster", clusterName, "cacheKey", cacheKey, "ttl", usersClientsTTL)
+
+	uc.Cache.Set(cacheKey, client, usersClientsTTL)
+
+	if _, found := uc.Cache.Get(cacheKey); found {
+		uc.log.Info("found after set")
+	} else {
+		uc.log.Info("not found after set")
+	}
 }
 
 func (uc *UsersClients) Get(user *auth.UserPrincipal, clusterName string) (client.Client, bool) {
-	if val, found := uc.Cache.Get(uc.cacheKey(user, clusterName)); found {
+	cacheKey := uc.cacheKey(user, clusterName)
+	uc.log.Info("get cached connection", "user", user, "cluster", clusterName, "cacheKey", cacheKey)
+
+	if val, found := uc.Cache.Get(cacheKey); found {
+		uc.log.Info("client found")
 		return val.(client.Client), true
 	}
 
+	uc.log.Info("client not found")
 	return nil, false
 }
 

pkg/services/crd/fetcher.go

@@ -11,7 +11,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 )
 
-const watchCRDsFrequency = 30 * time.Second
+const watchCRDsFrequency = 10 * time.Minute
 
 type Fetcher interface {
 	IsAvailable(clusterName, name string) bool
@@ -67,7 +67,6 @@ func (s *defaultFetcher) UpdateCRDList() {
 
 	for clusterName, client := range client.ClientsPool().Clients() {
 		crdList := &v1.CustomResourceDefinitionList{}
-
 		s.crds[clusterName] = []v1.CustomResourceDefinition{}
 
 		err := client.List(ctx, crdList)