feat: moved to multicluster-provider

cmd/gateway.go

@@ -13,6 +13,7 @@ import (
 	"github.com/platform-mesh/golang-commons/traces"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/spf13/cobra"
+	"golang.org/x/sync/errgroup"
 	ctrl "sigs.k8s.io/controller-runtime"
 
 	"github.com/platform-mesh/kubernetes-graphql-gateway/gateway/manager"
@@ -65,7 +66,7 @@ func initializeSentry(ctx context.Context, log *logger.Logger) error {
 		defaultCfg.Image.Name, defaultCfg.Image.Tag,
 	)
 	if err != nil {
-		log.Fatal().Err(err).Msg("Sentry init failed")
+		return fmt.Errorf("sentry init failed: %w", err)
 	}
 
 	defer openmfpcontext.Recover(log)
@@ -76,14 +77,14 @@ func initializeTracing(ctx context.Context, log *logger.Logger) (func(ctx contex
 	if defaultCfg.Tracing.Enabled {
 		shutdown, err := traces.InitProvider(ctx, defaultCfg.Tracing.Collector)
 		if err != nil {
-			log.Fatal().Err(err).Msg("unable to start gRPC-Sidecar TracerProvider")
+			return nil, fmt.Errorf("unable to start gRPC-Sidecar TracerProvider: %w", err)
 		}
 		return shutdown, nil
 	}
 
 	shutdown, err := traces.InitLocalProvider(ctx, defaultCfg.Tracing.Collector, false)
 	if err != nil {
-		log.Fatal().Err(err).Msg("unable to start local TracerProvider")
+		return nil, fmt.Errorf("unable to start local TracerProvider: %w", err)
 	}
 	return shutdown, nil
 }
@@ -121,61 +122,65 @@ func createServers(gatewayInstance http.Handler) (*http.Server, *http.Server, *h
 	return mainServer, metricsServer, healthServer
 }
 
-func shutdownServers(ctx context.Context, log *logger.Logger, mainServer, metricsServer, healthServer *http.Server) {
+func shutdownServers(ctx context.Context, log *logger.Logger, servers ...*http.Server) {
 	log.Info().Msg("Shutting down HTTP servers...")
 
-	if err := mainServer.Shutdown(ctx); err != nil {
-		log.Error().Err(err).Msg("Main HTTP server shutdown failed")
-	}
-
-	if err := metricsServer.Shutdown(ctx); err != nil {
-		log.Error().Err(err).Msg("Metrics HTTP server shutdown failed")
-	}
-
-	if err := healthServer.Shutdown(ctx); err != nil {
-		log.Error().Err(err).Msg("Health HTTP server shutdown failed")
+	for _, srv := range servers {
+		if err := srv.Shutdown(ctx); err != nil {
+			log.Error().Err(err).Str("addr", srv.Addr).Msg("HTTP server shutdown failed")
+		}
 	}
 }
 
 func runServers(ctx context.Context, log *logger.Logger, gatewayInstance http.Handler) error {
 	mainServer, metricsServer, healthServer := createServers(gatewayInstance)
 
-	// Start main server (GraphQL)
-	go func() {
+	eg, egCtx := errgroup.WithContext(ctx)
+
+	eg.Go(func() error {
 		log.Info().Str("addr", mainServer.Addr).Msg("Starting main HTTP server")
 		if err := mainServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-			log.Error().Err(err).Msg("Error starting main HTTP server")
+			return fmt.Errorf("main server error: %w", err)
 		}
-	}()
+		return nil
+	})
 
-	// Start metrics server
-	go func() {
+	eg.Go(func() error {
 		log.Info().Str("addr", metricsServer.Addr).Msg("Starting metrics HTTP server")
 		if err := metricsServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-			log.Error().Err(err).Msg("Error starting metrics HTTP server")
+			return fmt.Errorf("metrics server error: %w", err)
 		}
-	}()
+		return nil
+	})
 
-	// Start health server
-	go func() {
+	eg.Go(func() error {
 		log.Info().Str("addr", healthServer.Addr).Msg("Starting health HTTP server")
 		if err := healthServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-			log.Error().Err(err).Msg("Error starting health HTTP server")
+			return fmt.Errorf("health server error: %w", err)
 		}
-	}()
+		return nil
+	})
 
-	// Wait for shutdown signal
-	<-ctx.Done()
+	eg.Go(func() error {
+		<-egCtx.Done()
+		log.Info().Msg("Shutdown signal received")
 
-	shutdownCtx, cancel := context.WithTimeout(context.Background(), defaultCfg.ShutdownTimeout)
-	defer cancel()
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), defaultCfg.ShutdownTimeout)
+		defer cancel()
 
-	shutdownServers(shutdownCtx, log, mainServer, metricsServer, healthServer)
+		shutdownServers(shutdownCtx, log, mainServer, metricsServer, healthServer)
 
-	if closer, ok := gatewayInstance.(interface{ Close() error }); ok {
-		if err := closer.Close(); err != nil {
-			log.Error().Err(err).Msg("Error closing gateway services")
+		if closer, ok := gatewayInstance.(interface{ Close() error }); ok {
+			if err := closer.Close(); err != nil {
+				log.Error().Err(err).Msg("Error closing gateway services")
+			}
 		}
+
+		return nil
+	})
+
+	if err := eg.Wait(); err != nil {
+		return err
 	}
 
 	log.Info().Msg("Server shut down successfully")

cmd/listener.go

@@ -8,6 +8,7 @@ import (
 	kcpcore "github.com/kcp-dev/kcp/sdk/apis/core/v1alpha1"
 	kcptenancy "github.com/kcp-dev/kcp/sdk/apis/tenancy/v1alpha1"
 	"github.com/spf13/cobra"
+	"golang.org/x/sync/errgroup"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -77,7 +78,9 @@ var listenCmd = &cobra.Command{
 	Run: func(cmd *cobra.Command, args []string) {
 		log.Info().Str("LogLevel", log.GetLevel().String()).Msg("Starting the Listener...")
 
-		ctx := ctrl.SetupSignalHandler()
+		signalCtx := ctrl.SetupSignalHandler()
+		eg, ctx := errgroup.WithContext(signalCtx)
+
 		restCfg := ctrl.GetConfigOrDie()
 
 		mgrOpts := ctrl.Options{
@@ -104,24 +107,19 @@ var listenCmd = &cobra.Command{
 			OpenAPIDefinitionsPath: appCfg.OpenApiDefinitionsPath,
 		}
 
-		// Create the appropriate reconciler based on configuration
-		var reconcilerInstance reconciler.CustomReconciler
+		var reconcilerInstance reconciler.ControllerProvider
 		if appCfg.EnableKcp {
-			kcpReconciler, err := kcp.NewKCPReconciler(appCfg, reconcilerOpts, log)
+			kcpManager, err := kcp.NewKCPManager(appCfg, reconcilerOpts, log)
 			if err != nil {
-				log.Fatal().Err(err).Msg("unable to create KCP reconciler")
+				log.Fatal().Err(err).Msg("unable to create KCP manager")
 			}
+			reconcilerInstance = kcpManager
 
-			// Start virtual workspace watching if path is configured
 			if appCfg.Listener.VirtualWorkspacesConfigPath != "" {
-				go func() {
-					if err := kcpReconciler.StartVirtualWorkspaceWatching(ctx, appCfg.Listener.VirtualWorkspacesConfigPath); err != nil {
-						log.Fatal().Err(err).Msg("failed to start virtual workspace watching")
-					}
-				}()
+				eg.Go(func() error {
+					return kcpManager.StartVirtualWorkspaceWatching(ctx, appCfg.Listener.VirtualWorkspacesConfigPath)
+				})
 			}
-
-			reconcilerInstance = kcpReconciler
 		} else {
 			ioHandler, err := workspacefile.NewIOHandler(appCfg.OpenApiDefinitionsPath)
 			if err != nil {
@@ -134,37 +132,33 @@ var listenCmd = &cobra.Command{
 			}
 		}
 
-		// Setup reconciler with its own manager and start everything
-		if err := startManagerWithReconciler(ctx, reconcilerInstance); err != nil {
-			log.Fatal().Err(err).Msg("failed to start manager with reconciler")
+		eg.Go(func() error {
+			return startManagerWithReconciler(ctx, reconcilerInstance)
+		})
+
+		if err := eg.Wait(); err != nil {
+			log.Fatal().Err(err).Msg("exiting due to critical component failure")
 		}
+
+		log.Info().Msg("graceful shutdown complete")
 	},
 }
 
-// startManagerWithReconciler handles the common manager setup and start operations
-func startManagerWithReconciler(ctx context.Context, reconciler reconciler.CustomReconciler) error {
+func startManagerWithReconciler(ctx context.Context, reconciler reconciler.ControllerProvider) error {
 	mgr := reconciler.GetManager()
 
 	if err := reconciler.SetupWithManager(mgr); err != nil {
-		log.Error().Err(err).Msg("unable to setup reconciler with manager")
 		return err
 	}
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
-		log.Error().Err(err).Msg("unable to set up health check")
 		return err
 	}
 
 	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
-		log.Error().Err(err).Msg("unable to set up ready check")
 		return err
 	}
 
 	log.Info().Msg("starting manager")
-	if err := mgr.Start(ctx); err != nil {
-		log.Error().Err(err).Msg("problem running manager")
-		return err
-	}
-
-	return nil
+	return mgr.Start(ctx)
 }

cmd/root.go

@@ -78,6 +78,7 @@ func initConfig() {
 	// Gateway URL
 	v.SetDefault("gateway-url-virtual-workspace-prefix", "virtual-workspace")
 	v.SetDefault("gateway-url-default-kcp-workspace", "root")
+	v.SetDefault("gateway-url-kcp-workspace-pattern", "root:orgs:{org}")
 	v.SetDefault("gateway-url-graphql-suffix", "graphql")
 }
 

common/auth/export_test.go

@@ -0,0 +1,6 @@
+package auth
+
+// ExtractCAFromKubeconfigB64ForTest exposes the extractCAFromKubeconfigB64 method for testing
+func (m *MetadataInjector) ExtractCAFromKubeconfigB64ForTest(kubeconfigB64 string) []byte {
+	return m.extractCAFromKubeconfigB64(kubeconfigB64)
+}

common/auth/metadata_injector_test.go

@@ -623,3 +623,97 @@ users:
 		assert.Nil(t, result)
 	})
 }
+
+func TestMetadataInjector_ExtractCAFromKubeconfigB64(t *testing.T) {
+	log := testlogger.New().HideLogOutput().Logger
+	injector := NewMetadataInjector(log, nil)
+
+	tests := []struct {
+		name           string
+		kubeconfigB64  string
+		expectedResult bool // true if CA data should be extracted, false if nil
+		expectError    bool // true if we expect a warning to be logged
+	}{
+		{
+			name: "valid_kubeconfig_with_ca_data",
+			kubeconfigB64: base64.StdEncoding.EncodeToString([]byte(`
+apiVersion: v1
+kind: Config
+clusters:
+- cluster:
+    certificate-authority-data: LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0t
+    server: https://example.com
+  name: test-cluster
+contexts:
+- context:
+    cluster: test-cluster
+    user: test-user
+  name: test-context
+current-context: test-context
+users:
+- name: test-user
+  user:
+    token: test-token
+`)),
+			expectedResult: true,
+			expectError:    false,
+		},
+		{
+			name: "valid_kubeconfig_without_ca_data",
+			kubeconfigB64: base64.StdEncoding.EncodeToString([]byte(`
+apiVersion: v1
+kind: Config
+clusters:
+- cluster:
+    server: https://example.com
+  name: test-cluster
+contexts:
+- context:
+    cluster: test-cluster
+    user: test-user
+  name: test-context
+current-context: test-context
+users:
+- name: test-user
+  user:
+    token: test-token
+`)),
+			expectedResult: false,
+			expectError:    false,
+		},
+		{
+			name:           "invalid_base64",
+			kubeconfigB64:  "invalid-base64-!@#$%",
+			expectedResult: false,
+			expectError:    true,
+		},
+		{
+			name:           "empty_string",
+			kubeconfigB64:  "",
+			expectedResult: false,
+			expectError:    false,
+		},
+		{
+			name: "invalid_yaml_content",
+			kubeconfigB64: base64.StdEncoding.EncodeToString([]byte(`
+invalid yaml content
+not a kubeconfig
+`)),
+			expectedResult: false,
+			expectError:    false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := injector.ExtractCAFromKubeconfigB64ForTest(tt.kubeconfigB64)
+
+			if tt.expectedResult {
+				assert.NotNil(t, result, "Expected CA data to be extracted")
+				assert.Greater(t, len(result), 0, "Expected non-empty CA data")
+			} else {
+				assert.Nil(t, result, "Expected no CA data to be extracted")
+			}
+		})
+	}
+}

common/config/config.go

@@ -9,6 +9,7 @@ type Config struct {
 	Url struct {
 		VirtualWorkspacePrefix string `mapstructure:"gateway-url-virtual-workspace-prefix"`
 		DefaultKcpWorkspace    string `mapstructure:"gateway-url-default-kcp-workspace"`
+		KcpWorkspacePattern    string `mapstructure:"gateway-url-kcp-workspace-pattern"`
 		GraphqlSuffix          string `mapstructure:"gateway-url-graphql-suffix"`
 	} `mapstructure:",squash"`