backend: caching: implemented caching for single-cluster

backend/cmd/headlamp.go

@@ -59,12 +59,11 @@ import (
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
+	"golang.org/x/oauth2"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/clientcmd/api"
-
-	"golang.org/x/oauth2"
 )
 
 type HeadlampConfig struct {
@@ -101,6 +100,8 @@ const (
 	TokenCacheFileName = "headlamp-token-cache"
 )
 
+var k8scache = cache.New[string]()
+
 type clientConfig struct {
 	Clusters                []Cluster `json:"clusters"`
 	IsDynamicClusterEnabled bool      `json:"isDynamicClusterEnabled"`
@@ -1350,35 +1351,11 @@ func (c *HeadlampConfig) handleError(w http.ResponseWriter, ctx context.Context,
 	http.Error(w, err.Error(), status)
 }
 
-// handleClusterAPI handles cluster API requests. It is responsible for
-// all the requests made to /clusters/{clusterName}/{api:.*} endpoint.
-// It parses the request and creates a proxy request to the cluster.
-// That proxy is saved in the cache with the context key.
-func handleClusterAPI(c *HeadlampConfig, router *mux.Router) { //nolint:funlen
-	router.PathPrefix("/clusters/{clusterName}/{api:.*}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		start := time.Now()
-		ctx := r.Context()
-
-		ctx, span := telemetry.CreateSpan(ctx, r, "cluster-api", "handleClusterAPI",
-			attribute.String("cluster", mux.Vars(r)["clusterName"]),
-		)
-		defer span.End()
-
-		c.telemetryHandler.RecordRequestCount(ctx, r, attribute.String("cluster", mux.Vars(r)["clusterName"]))
-		c.telemetryHandler.RecordEvent(span, "Cluster API request started")
-
-		// A deferred function to record duration metrics & log the request completion
-		defer recordRequestCompletion(c, ctx, start, r)
-
-		contextKey, err := c.getContextKeyForRequest(r)
-		if err != nil {
-			c.handleError(w, ctx, span, err, "failed to get context key", http.StatusBadRequest)
-			return
-		}
-
-		kContext, err := c.KubeConfigStore.GetContext(contextKey)
+func clusterRequestHandler(c *HeadlampConfig) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx, span, contextKey, kContext, err := GetContextKeyAndKContext(w, r, c)
 		if err != nil {
-			c.handleError(w, ctx, span, err, "failed to get context", http.StatusNotFound)
+			c.handleError(w, ctx, span, err, "error while retrieving context and kContext", http.StatusBadRequest)
 			return
 		}
 
@@ -1427,6 +1404,24 @@ func handleClusterAPI(c *HeadlampConfig, router *mux.Router) { //nolint:funlen
 	})
 }
 
+// handleClusterAPI handles cluster API requests. It is responsible for
+// all the requests made to /clusters/{clusterName}/{api:.*} endpoint.
+// It parses the request and creates a proxy request to the cluster.
+// That proxy is saved in the cache with the context key.
+func handleClusterAPI(c *HeadlampConfig, router *mux.Router) {
+	clusterAPIrequest := clusterRequestHandler(c)
+
+	handler := clusterAPIrequest
+
+	if !c.CacheEnabled {
+		cacheMiddleware := CacheMiddleWare(c)
+
+		handler = cacheMiddleware(handler)
+	}
+
+	router.PathPrefix("/clusters/{clusterName}/{api:.*}").Handler(handler)
+}
+
 func recordRequestCompletion(c *HeadlampConfig, ctx context.Context,
 	start time.Time, r *http.Request,
 ) {
@@ -1524,6 +1519,8 @@ func (c *HeadlampConfig) getClusters() []Cluster {
 	}
 
 	for _, context := range contexts {
+		context := context
+
 		if context.Error != "" {
 			clusters = append(clusters, Cluster{
 				Name:  context.Name,
@@ -1578,6 +1575,8 @@ func parseCustomNameClusters(contexts []kubeconfig.Context) ([]Cluster, []error)
 	var setupErrors []error
 
 	for _, context := range contexts {
+		context := context
+
 		info := context.KubeContext.Extensions["headlamp_info"]
 		if info != nil {
 			// Convert the runtime.Unknown object to a byte slice
@@ -2038,6 +2037,8 @@ func (c *HeadlampConfig) updateCustomContextToCache(config *api.Config, clusterN
 	}
 
 	for _, context := range contexts {
+		context := context
+
 		// Remove the old context from the store
 		if err := c.KubeConfigStore.RemoveContext(clusterName); err != nil {
 			logger.Log(logger.LevelError, nil, err, "Removing context from the store")

backend/cmd/server.go

@@ -17,15 +17,23 @@ limitations under the License.
 package main
 
 import (
+	"context"
+	"errors"
+	"net/http"
 	"os"
 	"strings"
 
+	"github.com/gorilla/mux"
 	"github.com/kubernetes-sigs/headlamp/backend/pkg/cache"
 	"github.com/kubernetes-sigs/headlamp/backend/pkg/config"
 	"github.com/kubernetes-sigs/headlamp/backend/pkg/headlampconfig"
+	"github.com/kubernetes-sigs/headlamp/backend/pkg/k8cache"
 	"github.com/kubernetes-sigs/headlamp/backend/pkg/kubeconfig"
 	"github.com/kubernetes-sigs/headlamp/backend/pkg/logger"
 	"github.com/kubernetes-sigs/headlamp/backend/pkg/plugins"
+	"github.com/kubernetes-sigs/headlamp/backend/pkg/telemetry"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 func main() {
@@ -50,6 +58,7 @@ func main() {
 			UseInCluster:          conf.InCluster,
 			KubeConfigPath:        conf.KubeConfigPath,
 			SkippedKubeContexts:   conf.SkippedKubeContexts,
+			CacheEnabled:          conf.CacheEnabled,
 			ListenAddr:            conf.ListenAddr,
 			Port:                  conf.Port,
 			DevMode:               conf.DevMode,
@@ -86,6 +95,102 @@ func main() {
 	})
 }
 
+// GetContextKeyAndContext returns Kcontext , ContextKey for using these in CacheMiddleWare function.
+// It also return span and ctx that will help while using handleError function.
+func GetContextKeyAndKContext(w http.ResponseWriter,
+	r *http.Request, c *HeadlampConfig) (context.Context,
+	trace.Span, string, *kubeconfig.Context, error,
+) {
+	ctx := r.Context()
+	ctx, span := telemetry.CreateSpan(ctx, r, "cluster-api", "handleClusterAPI",
+		attribute.String("cluster", mux.Vars(r)["clusterName"]),
+	)
+
+	defer span.End()
+
+	contextKey, err := c.getContextKeyForRequest(r)
+	if err != nil {
+		c.handleError(w, ctx, span, err, "failed to get context Key:", http.StatusBadRequest)
+		return nil, nil, "", nil, err
+	}
+
+	kContext, err := c.KubeConfigStore.GetContext(contextKey)
+	if err != nil {
+		c.handleError(w, ctx, span, err, "failed to get context", http.StatusNotFound)
+		return nil, nil, "", nil, err
+	}
+
+	return ctx, span, contextKey, kContext, nil
+}
+
+// CacheMiddleWare is Middleware for Caching purpose. It involves generating key for a request,
+// authorizing user , store resource data in cache and returns data if key is present.
+func CacheMiddleWare(c *HeadlampConfig) mux.MiddlewareFunc {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			ctx, span, contextKey, kContext, err := GetContextKeyAndKContext(w, r, c)
+			if err != nil {
+				c.handleError(w, ctx, span, err, "failed to get context and Kcontext", http.StatusNotFound)
+				return
+			}
+
+			if kContext.Error != "" {
+				c.handleError(w, ctx, span, errors.New(kContext.Error), "context has error", http.StatusBadRequest)
+				return
+			}
+
+			rcw := k8cache.CreateResponseCapture(w)
+
+			key, err := k8cache.GenerateKey(r.URL, contextKey)
+			if err != nil {
+				c.handleError(w, ctx, span, err, "failed to generate key ", http.StatusBadRequest)
+				return
+			}
+
+			isAllowed, authErr := k8cache.IsAllowed(r.URL, kContext, w, r)
+			if authErr != nil {
+				k8cache.StoreAfterAuthError(k8scache, isAllowed, next, key, w, r, rcw)
+
+				return
+			} else if !isAllowed {
+				err := k8cache.ReturnAuthErrorResponse(w, r, contextKey)
+				if err != nil {
+					c.handleError(w, ctx, span, err, "error while returning to client", http.StatusInternalServerError)
+				}
+
+				return
+			}
+
+			served, err := k8cache.LoadFromCache(k8scache, isAllowed, key, w, r)
+			if err != nil {
+				c.handleError(w, ctx, span, errors.New(kContext.Error), "failed to load from cache", http.StatusServiceUnavailable)
+			}
+
+			if served {
+				c.telemetryHandler.RecordEvent(span, "Served from cache")
+				return
+			}
+
+			go k8cache.CheckForChanges(k8scache, contextKey, r, rcw)
+			next.ServeHTTP(rcw, r)
+
+			// // Here after making changes the request goes to CheckAndPurge to check whether
+			// // the method is any these POST, PUT, PATCH, DELETE. If yes then it
+			// // is proceed for purge the stale the staled stored requests.
+			// err = k8cache.CheckAndPurge(w, r, k8scache, next, rcw, isAllowed)
+			// if err != nil {
+			// 	c.handleError(w, ctx, span, err, "error while purging data", http.StatusInternalServerError)
+			// }
+
+			err = k8cache.RequestK8ClusterAPIAndStore(k8scache, r.URL, rcw, r, key)
+			if err != nil {
+				c.handleError(w, ctx, span, errors.New(kContext.Error), "error while storing into cache", http.StatusBadRequest)
+				return
+			}
+		})
+	}
+}
+
 func runListPlugins() {
 	conf, err := config.Parse(os.Args[2:])
 	if err != nil {

backend/pkg/cache/cache.go

@@ -32,6 +32,7 @@ type Cache[T any] interface {
 	Get(ctx context.Context, key string) (T, error)
 	GetAll(ctx context.Context, selectFunc Matcher) (map[string]T, error)
 	UpdateTTL(ctx context.Context, key string, ttl time.Duration) error
+	// Size(ctx context.Context) int
 }
 
 // Matcher is a function that returns true if the key matches.
@@ -169,3 +170,20 @@ func (c *cache[T]) UpdateTTL(ctx context.Context, key string, ttl time.Duration)
 
 	return nil
 }
+
+// Size returns the number of unexpired items in the cache.
+func (c *cache[T]) Size(ctx context.Context) int {
+	c.lock.RLock()
+	defer c.lock.RUnlock()
+
+	count := 0
+	now := time.Now()
+
+	for _, value := range c.store {
+		if value.expiresAt.IsZero() || value.expiresAt.After(now) {
+			count++
+		}
+	}
+
+	return count
+}

backend/pkg/config/config.go

@@ -21,6 +21,7 @@ const defaultPort = 4466
 
 type Config struct {
 	InCluster                 bool   `koanf:"in-cluster"`
+	CacheEnabled              bool   `koanf:"cache-enabled"`
 	DevMode                   bool   `koanf:"dev"`
 	InsecureSsl               bool   `koanf:"insecure-ssl"`
 	EnableHelm                bool   `koanf:"enable-helm"`
@@ -237,6 +238,7 @@ func flagset() *flag.FlagSet {
 	f := flag.NewFlagSet("config", flag.ContinueOnError)
 
 	f.Bool("in-cluster", false, "Set when running from a k8s cluster")
+	f.Bool("cache-enabled", false, "To cache k8s resources")
 	f.Bool("dev", false, "Allow connections from other origins")
 	f.Bool("insecure-ssl", false, "Accept/Ignore all server SSL certificates")
 	f.Bool("enable-dynamic-clusters", false, "Enable dynamic clusters, which stores stateless clusters in the frontend.")

backend/pkg/headlampconfig/headlampConfig.go

@@ -9,6 +9,7 @@ type HeadlampCFG struct {
 	UseInCluster          bool
 	ListenAddr            string
 	DevMode               bool
+	CacheEnabled          bool
 	Insecure              bool
 	EnableHelm            bool
 	EnableDynamicClusters bool