ClustersClientsPool Middleware

core/clustersmngr/clustersmngr.go

@@ -0,0 +1,92 @@
+package clustersmngr
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/weaveworks/weave-gitops/pkg/kube"
+	"github.com/weaveworks/weave-gitops/pkg/server/auth"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+//go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate
+
+type key int
+
+const (
+	// Clusters Clients Pool context key
+	ClustersClientsPoolCtxKey key = iota
+)
+
+var (
+	scheme = kube.CreateScheme()
+)
+
+// Cluster defines a leaf cluster
+type Cluster struct {
+	// Name defines the cluster name
+	Name string `yaml:"name"`
+	// Server defines cluster api address
+	Server string `yaml:"server"`
+
+	// SecretRef defines secret name that holds the cluster Bearer Token
+	SecretRef string `yaml:"secretRef"`
+	// BearerToken cluster access token read from SecretRef
+	BearerToken string
+
+	// TLSConfig holds configuration for TLS connection with the cluster values read from SecretRef
+	TLSConfig rest.TLSClientConfig
+}
+
+//ClusterFetcher fetches all leaf clusters
+//counterfeiter:generate . ClusterFetcher
+type ClusterFetcher interface {
+	Fetch(ctx context.Context) ([]Cluster, error)
+}
+
+// ClientsPool stores all clients to the leaf clusters
+type ClientsPool interface {
+	Add(user *auth.UserPrincipal, cluster Cluster) error
+	Clients() map[string]client.Client
+}
+
+type clientsPool struct {
+	clients map[string]client.Client
+}
+
+// NewClustersClientsPool initializes a new ClientsPool
+func NewClustersClientsPool() ClientsPool {
+	return &clientsPool{
+		clients: map[string]client.Client{},
+	}
+}
+
+// Add adds a cluster client to the clients pool with the given user impersonation
+func (cp *clientsPool) Add(user *auth.UserPrincipal, cluster Cluster) error {
+	config := &rest.Config{
+		Host:            cluster.Server,
+		BearerToken:     cluster.BearerToken,
+		TLSClientConfig: cluster.TLSConfig,
+		Impersonate: rest.ImpersonationConfig{
+			UserName: user.ID,
+			Groups:   user.Groups,
+		},
+	}
+
+	leafClient, err := client.New(config, client.Options{
+		Scheme: scheme,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to create leaf client: %w", err)
+	}
+
+	cp.clients[cluster.Name] = leafClient
+
+	return nil
+}
+
+// Clients returns the clusters clients
+func (cp *clientsPool) Clients() map[string]client.Client {
+	return cp.clients
+}

core/clustersmngr/middleware.go

@@ -0,0 +1,50 @@
+package clustersmngr
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/weaveworks/weave-gitops/pkg/server/auth"
+)
+
+// WithClustersClients creates clusters client for provided user in the context
+func WithClustersClients(clustersFetcher ClusterFetcher, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		user := auth.Principal(r.Context())
+		if user == nil {
+			next.ServeHTTP(w, r)
+			return
+		}
+
+		clusters, err := clustersFetcher.Fetch(r.Context())
+		if err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+			fmt.Fprintln(w, "failed fetching clusters list: %w", err)
+			return
+		}
+
+		clientsPool := NewClustersClientsPool()
+		for _, c := range clusters {
+			if err := clientsPool.Add(user, c); err != nil {
+				w.WriteHeader(http.StatusInternalServerError)
+				fmt.Fprintf(w, "failed adding cluster client to the pool: %s", err)
+				return
+			}
+		}
+
+		ctx := context.WithValue(r.Context(), ClustersClientsPoolCtxKey, clientsPool)
+
+		next.ServeHTTP(w, r.WithContext(ctx))
+	})
+}
+
+// ClientsPoolFromCtx returns the ClusterClients pool stored in the context
+func ClientsPoolFromCtx(ctx context.Context) ClientsPool {
+	pool, ok := ctx.Value(ClustersClientsPoolCtxKey).(*clientsPool)
+	if ok {
+		return pool
+	}
+
+	return nil
+}

core/clustersmngr/middleware_test.go

@@ -0,0 +1,66 @@
+package clustersmngr_test
+
+import (
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	. "github.com/onsi/gomega"
+
+	"github.com/weaveworks/weave-gitops/core/clustersmngr"
+	"github.com/weaveworks/weave-gitops/core/clustersmngr/clustersmngrfakes"
+	"github.com/weaveworks/weave-gitops/pkg/server/auth"
+)
+
+func TestWithClustersClientsMiddleware(t *testing.T) {
+	cluster := makeLeafCluster(t)
+	clustersFetcher := &clustersmngrfakes.FakeClusterFetcher{}
+	clustersFetcher.FetchReturns([]clustersmngr.Cluster{cluster}, nil)
+
+	g := NewGomegaWithT(t)
+
+	defaultHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
+	middleware := func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			clientsPool := clustersmngr.ClientsPoolFromCtx(r.Context())
+
+			g.Expect(clientsPool.Clients()).To(HaveKey(cluster.Name))
+
+			next.ServeHTTP(w, r)
+		})
+	}(defaultHandler)
+
+	middleware = clustersmngr.WithClustersClients(clustersFetcher, middleware)
+	middleware = authMiddleware(middleware)
+
+	req := httptest.NewRequest(http.MethodGet, "http://www.foo.com/", nil)
+	res := httptest.NewRecorder()
+	middleware.ServeHTTP(res, req)
+
+	g.Expect(res).To(HaveHTTPStatus(http.StatusOK))
+}
+
+func TestWithClustersClientsMiddlewareFailsToFetchCluster(t *testing.T) {
+	defaultHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})
+
+	clustersFetcher := &clustersmngrfakes.FakeClusterFetcher{}
+	clustersFetcher.FetchReturns(nil, errors.New("error"))
+
+	middleware := clustersmngr.WithClustersClients(clustersFetcher, defaultHandler)
+	middleware = authMiddleware(middleware)
+
+	req := httptest.NewRequest(http.MethodGet, "http://www.foo.com/", nil)
+	res := httptest.NewRecorder()
+	middleware.ServeHTTP(res, req)
+
+	g := NewGomegaWithT(t)
+
+	g.Expect(res).To(HaveHTTPStatus(http.StatusInternalServerError))
+}
+
+func authMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		next.ServeHTTP(w, r.WithContext(auth.WithPrincipal(r.Context(), &auth.UserPrincipal{ID: "user@weave.gitops", Groups: []string{"developers"}})))
+	})
+}

core/clustersmngr/single_fetcher.go

@@ -0,0 +1,28 @@
+package clustersmngr
+
+import (
+	"context"
+
+	"k8s.io/client-go/rest"
+)
+
+type singleClusterFetcher struct {
+	restConfig *rest.Config
+}
+
+func NewSingleClusterFetcher(config *rest.Config) (ClusterFetcher, error) {
+	return singleClusterFetcher{
+		restConfig: config,
+	}, nil
+}
+
+func (cf singleClusterFetcher) Fetch(ctx context.Context) ([]Cluster, error) {
+	return []Cluster{
+		{
+			Name:        "Default",
+			Server:      cf.restConfig.Host,
+			BearerToken: cf.restConfig.BearerToken,
+			TLSConfig:   cf.restConfig.TLSClientConfig,
+		},
+	}, nil
+}

core/clustersmngr/single_fetcher_test.go

@@ -0,0 +1,29 @@
+package clustersmngr_test
+
+import (
+	"context"
+	"testing"
+
+	. "github.com/onsi/gomega"
+	"github.com/weaveworks/weave-gitops/core/clustersmngr"
+	"k8s.io/client-go/rest"
+)
+
+func TestSingleFetcher(t *testing.T) {
+	config := &rest.Config{
+		Host:        "my-host",
+		BearerToken: "my-token",
+	}
+
+	g := NewGomegaWithT(t)
+
+	fetcher, err := clustersmngr.NewSingleClusterFetcher(config)
+	g.Expect(err).To(BeNil())
+
+	clusters, err := fetcher.Fetch(context.TODO())
+	g.Expect(err).To(BeNil())
+
+	g.Expect(clusters[0].Name).To(Equal("Default"))
+	g.Expect(clusters[0].Server).To(Equal(config.Host))
+	g.Expect(clusters[0].BearerToken).To(Equal(config.BearerToken))
+}