Add components verification step to bootstraping WGE

pkg/bootstrap/steps/common_tests.go

@@ -34,5 +34,6 @@ func makeTestConfig(t *testing.T, config Config, objects ...runtime.Object) Conf
 		ClientSecret:            config.ClientSecret,
 		RedirectURL:             config.RedirectURL,
 		PromptedForDiscoveryURL: config.PromptedForDiscoveryURL,
+		SkipComponentCheck:      config.SkipComponentCheck,
 	}
 }

pkg/bootstrap/steps/config.go

@@ -78,6 +78,7 @@ type ConfigBuilder struct {
 	clientID                string
 	clientSecret            string
 	PromptedForDiscoveryURL bool
+	SkipComponentCheck      bool
 }
 
 func NewConfigBuilder() *ConfigBuilder {
@@ -190,6 +191,8 @@ type Config struct {
 	ClientSecret            string
 	RedirectURL             string
 	PromptedForDiscoveryURL bool
+
+	SkipComponentCheck bool // skip checking if components are installed
 }
 
 // Builds creates a valid config so boostrap could be executed. It uses values introduced
@@ -245,6 +248,7 @@ func (cb *ConfigBuilder) Build() (Config, error) {
 		ClientID:                cb.clientID,
 		ClientSecret:            cb.clientSecret,
 		PromptedForDiscoveryURL: cb.PromptedForDiscoveryURL,
+		SkipComponentCheck:      cb.SkipComponentCheck,
 	}, nil
 
 }

pkg/bootstrap/steps/install_wge.go

@@ -11,6 +11,8 @@ import (
 	"github.com/weaveworks/weave-gitops-enterprise/pkg/bootstrap/utils"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/cli-utils/pkg/object"
 )
 
 const (
@@ -39,6 +41,10 @@ const (
 	gitopssetsHealthBindAddress       = ":8081"
 )
 
+var Components = []string{"cluster-controller-manager",
+	"weave-gitops-enterprise-mccp-cluster-bootstrap-controller",
+	"weave-gitops-enterprise-mccp-cluster-service"}
+
 var getUserDomain = StepInput{
 	Name:         inUserDomain,
 	Type:         stringInput,
@@ -146,6 +152,16 @@ func installWge(input []StepInput, c *Config) ([]StepOutput, error) {
 		CommitMsg: wgeHelmReleaseCommitMsg,
 	}
 
+	if !c.SkipComponentCheck {
+		// Wait for the components to be healthy
+
+		c.Logger.Actionf("waiting for components to be healthy")
+		err = reportComponentsHealth(c, Components, WGEDefaultNamespace, 5*time.Minute)
+		if err != nil {
+			return []StepOutput{}, err
+		}
+	}
+
 	return []StepOutput{
 		{
 			Name:  wgeHelmrepoFileName,
@@ -248,3 +264,28 @@ func isUserDomainEnabled(input []StepInput, c *Config) bool {
 	}
 	return false
 }
+
+func reportComponentsHealth(c *Config, componentNames []string, namespace string, timeout time.Duration) error {
+	// Initialize the status checker
+	checker, err := utils.NewStatusChecker(c.KubernetesClient, 5*time.Second, timeout, c.Logger)
+	if err != nil {
+		return err
+	}
+
+	// Construct a list of resources to check
+	var identifiers []object.ObjMetadata
+	for _, name := range componentNames {
+		identifiers = append(identifiers, object.ObjMetadata{
+			Namespace: namespace,
+			Name:      name,
+			GroupKind: schema.GroupKind{Group: "apps", Kind: "Deployment"},
+		})
+	}
+
+	// Perform the health check
+	if err := checker.Assess(identifiers...); err != nil {
+		return err
+	}
+
+	return nil
+}

pkg/bootstrap/steps/install_wge_test.go

@@ -126,15 +126,17 @@ status: {}
 
 func TestInstallWge(t *testing.T) {
 	tests := []struct {
-		name       string
-		domainType string
-		input      []StepInput
-		output     []StepOutput
-		err        bool
+		name               string
+		domainType         string
+		skipComponentCheck bool
+		input              []StepInput
+		output             []StepOutput
+		err                bool
 	}{
 		{
-			name:       "unsupported domain type",
-			domainType: "wrongType",
+			name:               "unsupported domain type",
+			domainType:         "wrongType",
+			skipComponentCheck: true, // This should skip the health check
 			input: []StepInput{
 				{
 					Name:  inUserDomain,
@@ -144,8 +146,9 @@ func TestInstallWge(t *testing.T) {
 			err: true,
 		},
 		{
-			name:       "install with domaintype localhost",
-			domainType: domainTypeLocalhost,
+			name:               "install with domaintype localhost",
+			domainType:         domainTypeLocalhost,
+			skipComponentCheck: true, // This should skip the health check
 			input: []StepInput{
 				{
 					Name:  inUserDomain,
@@ -175,8 +178,9 @@ func TestInstallWge(t *testing.T) {
 			err: false,
 		},
 		{
-			name:       "install with domaintype external dns",
-			domainType: domainTypeExternalDNS,
+			name:               "install with domaintype external dns",
+			domainType:         domainTypeExternalDNS,
+			skipComponentCheck: true, // This should skip the health check
 			input: []StepInput{
 				{
 					Name:  inUserDomain,
@@ -210,8 +214,9 @@ func TestInstallWge(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			testConfig := Config{
-				WGEVersion: "1.0.0",
-				DomainType: tt.domainType,
+				WGEVersion:         "1.0.0",
+				DomainType:         tt.domainType,
+				SkipComponentCheck: tt.skipComponentCheck,
 			}
 
 			config := makeTestConfig(t, testConfig)

pkg/bootstrap/utils/status.go

@@ -0,0 +1,91 @@
+package utils
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/weaveworks/weave-gitops/pkg/logger"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	k8s_client "sigs.k8s.io/controller-runtime/pkg/client"
+
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling/aggregator"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling/collector"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling/event"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
+	"sigs.k8s.io/cli-utils/pkg/object"
+)
+
+type StatusChecker struct {
+	pollInterval time.Duration
+	timeout      time.Duration
+	client       client.Client
+	statusPoller *polling.StatusPoller
+	logger       logger.Logger
+}
+
+func NewStatusChecker(client k8s_client.Client, pollInterval time.Duration, timeout time.Duration, log logger.Logger) (*StatusChecker, error) {
+
+	return &StatusChecker{
+		pollInterval: pollInterval,
+		timeout:      timeout,
+		client:       client,
+		statusPoller: polling.NewStatusPoller(client, client.RESTMapper(), polling.Options{}),
+		logger:       log,
+	}, nil
+}
+
+func (sc *StatusChecker) Assess(identifiers ...object.ObjMetadata) error {
+	ctx, cancel := context.WithTimeout(context.Background(), sc.timeout)
+	defer cancel()
+
+	opts := polling.PollOptions{PollInterval: sc.pollInterval}
+	eventsChan := sc.statusPoller.Poll(ctx, identifiers, opts)
+
+	coll := collector.NewResourceStatusCollector(identifiers)
+	done := coll.ListenWithObserver(eventsChan, desiredStatusNotifierFunc(cancel, status.CurrentStatus))
+
+	<-done
+
+	// we use sorted identifiers to loop over the resource statuses because a Go's map is unordered.
+	// sorting identifiers by object's name makes sure that the logs look stable for every run
+	sort.SliceStable(identifiers, func(i, j int) bool {
+		return strings.Compare(identifiers[i].Name, identifiers[j].Name) < 0
+	})
+	for _, id := range identifiers {
+		rs := coll.ResourceStatuses[id]
+		switch rs.Status {
+		case status.CurrentStatus:
+			sc.logger.Successf("%s: %s ready", rs.Identifier.Name, strings.ToLower(rs.Identifier.GroupKind.Kind))
+		case status.NotFoundStatus:
+			sc.logger.Failuref("%s: %s not found", rs.Identifier.Name, strings.ToLower(rs.Identifier.GroupKind.Kind))
+		default:
+			sc.logger.Failuref("%s: %s not ready", rs.Identifier.Name, strings.ToLower(rs.Identifier.GroupKind.Kind))
+		}
+	}
+
+	if coll.Error != nil || ctx.Err() == context.DeadlineExceeded {
+		return fmt.Errorf("timed out waiting for condition")
+	}
+	return nil
+}
+
+// desiredStatusNotifierFunc returns an Observer function for the
+// ResourceStatusCollector that will cancel the context (using the cancelFunc)
+// when all resources have reached the desired status.
+func desiredStatusNotifierFunc(cancelFunc context.CancelFunc,
+	desired status.Status) collector.ObserverFunc {
+	return func(rsc *collector.ResourceStatusCollector, _ event.Event) {
+		var rss []*event.ResourceStatus
+		for _, rs := range rsc.ResourceStatuses {
+			rss = append(rss, rs)
+		}
+		aggStatus := aggregator.AggregateStatus(rss, desired)
+		if aggStatus == desired {
+			cancelFunc()
+		}
+	}
+}

pkg/bootstrap/utils/status_integration_test.go

@@ -0,0 +1,79 @@
+//go:build integration
+
+package utils
+
+import (
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	testutils "github.com/weaveworks/weave-gitops-enterprise/test/utils"
+	"github.com/weaveworks/weave-gitops/pkg/logger"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/cli-utils/pkg/object"
+	k8s_client "sigs.k8s.io/controller-runtime/pkg/client"
+	k8s_config "sigs.k8s.io/controller-runtime/pkg/client/config"
+)
+
+func TestStatusCheckerIt(t *testing.T) {
+	// Setup function to create kubeconfig and set the KUBECONFIG environment variable
+	setup := func() error {
+		kp, err := testutils.CreateKubeconfigFileForRestConfig(*cfg)
+		if err != nil {
+			return fmt.Errorf("cannot create kubeconfig: %w", err)
+		}
+		os.Setenv("KUBECONFIG", kp)
+		return nil
+	}
+
+	// Reset function to clean up after the test
+	reset := func() {
+		os.Unsetenv("KUBECONFIG")
+	}
+
+	// Run the setup function and handle errors
+	assert.NoError(t, setup(), "Error setting up kubeconfig")
+	defer reset()
+
+	// Create the Kubernetes client using the kubeconfig
+	config, err := k8s_config.GetConfig()
+	assert.NoError(t, err, "Error getting Kubernetes config")
+	client, err := k8s_client.New(config, k8s_client.Options{})
+	assert.NoError(t, err, "Error creating Kubernetes client")
+
+	// Initialize logger
+	logInstance := logger.NewCLILogger(os.Stdout)
+
+	// Initialize the StatusChecker
+	statusChecker, err := NewStatusChecker(client, 5*time.Second, 1*time.Minute, logInstance)
+	assert.NoError(t, err, "Error creating StatusChecker")
+
+	// Define test cases
+	testCases := []struct {
+		name        string
+		identifiers []object.ObjMetadata
+	}{
+		{
+			name: "Check specific Kubernetes resource",
+			identifiers: []object.ObjMetadata{
+				{
+					Name:      "cluster-controller-manager",
+					Namespace: "flux-system",
+					GroupKind: schema.GroupKind{Group: "apps", Kind: "Deployment"},
+				},
+				// Add more resources as needed
+			},
+		},
+		// ... other test cases ...
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// Call the Assess function to check the status of resources
+			err := statusChecker.Assess(tc.identifiers...)
+			assert.NoError(t, err, "Error assessing resource status")
+		})
+	}
+}