Merge pull request #2 from yue9944882/feat/progressive-cluster-register

Feat: Progressive cluster registration w/ finer granularity
oam-dev · Nov 30, 2021 · 48c22c3 · 48c22c3
2 parents 4dbae5a + 774ad42
commit 48c22c3
Show file tree

Hide file tree

Showing 2 changed files with 85 additions and 4 deletions.
diff --git a/pkg/hub/approve.go b/pkg/hub/approve.go
@@ -7,13 +7,13 @@ import (
 	"strings"
 	"time"
 
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-
 	"github.com/ghodss/yaml"
+	"github.com/pkg/errors"
 	certificatesv1 "k8s.io/api/certificates/v1"
 	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
@@ -265,7 +265,7 @@ func (c *Cluster) RegisterSpokeCluster(ctx context.Context, clusterName string)
 	return nil
 }
 
-func (c *Cluster) Wait4SpokeClusterReady(ctx context.Context, clusterName string) (bool, error) {
+func (c *Cluster) WaitForSpokeClusterReady(ctx context.Context, clusterName string) (bool, error) {
 	listOpts := []client.ListOption{
 		client.MatchingLabels{
 			clusterLabel: clusterName,
@@ -299,6 +299,51 @@ func (c *Cluster) Wait4SpokeClusterReady(ctx context.Context, clusterName string
 	return true, nil
 }
 
+func (c *Cluster) WaitForCSRCreated(ctx context.Context, spokeClusterName string) error {
+	nativeClient, err := kubernetes.NewForConfig(c.KubeConfig)
+	if err != nil {
+		return err
+	}
+	return wait.PollImmediateUntil(time.Second, func() (bool, error) {
+		csrList, err := nativeClient.CertificatesV1().CertificateSigningRequests().List(ctx, metav1.ListOptions{
+			LabelSelector: "open-cluster-management.io/cluster-name=" + spokeClusterName,
+		})
+		if err != nil {
+			return false, err
+		}
+		return len(csrList.Items) > 0, nil
+	}, ctx.Done())
+}
+
+func (c *Cluster) ApproveCSR(ctx context.Context, spokeClusterName string) error {
+	nativeClient, err := kubernetes.NewForConfig(c.KubeConfig)
+	if err != nil {
+		return err
+	}
+	csrList, err := nativeClient.CertificatesV1().
+		CertificateSigningRequests().
+		List(ctx, metav1.ListOptions{
+			LabelSelector: "open-cluster-management.io/cluster-name=" + spokeClusterName,
+		})
+	if err != nil {
+		return err
+	}
+	if len(csrList.Items) == 0 {
+		return fmt.Errorf("no csr found related to spoke cluster %q", spokeClusterName)
+	}
+	for _, csr := range csrList.Items {
+		if len(csr.Status.Certificate) == 0 {
+			_, err := nativeClient.CertificatesV1().
+				CertificateSigningRequests().
+				UpdateApproval(ctx, csr.Name, &csr, metav1.UpdateOptions{})
+			if err != nil {
+				return errors.Wrapf(err, "failed approving CSR %q for cluster %q", csr.Name, spokeClusterName)
+			}
+		}
+	}
+	return nil
+}
+
 func checkCsrStatus(status *certificatesv1.CertificateSigningRequestStatus) (approved bool, denied bool) {
 	for _, c := range status.Conditions {
 		if c.Type == certificatesv1.CertificateApproved {

diff --git a/pkg/spoke/register.go b/pkg/spoke/register.go
@@ -8,11 +8,13 @@ import (
 	"fmt"
 	"strings"
 	"text/template"
+	"time"
 
 	"github.com/Masterminds/sprig"
 	"github.com/ghodss/yaml"
 	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
 	clientcmdapiv1 "k8s.io/client-go/tools/clientcmd/api/v1"
 	"k8s.io/klog/v2"
@@ -161,6 +163,40 @@ func (c *Cluster) InitSpokeClusterEnv(ctx context.Context) error {
 	return nil
 }
 
+func (c *Cluster) WaitForRegistrationOperatorReady(ctx context.Context) error {
+	return wait.PollImmediateUntil(time.Second, func() (bool, error) {
+		podList := &corev1.PodList{}
+		err := c.Args.Client.List(ctx, podList,
+			client.InNamespace("open-cluster-management"),
+			client.MatchingLabels{"app": "klusterlet"})
+		if err != nil {
+			return false, err
+		}
+		atLeastOnceRunning := false
+		for _, pod := range podList.Items {
+			atLeastOnceRunning = atLeastOnceRunning || pod.Status.Phase == corev1.PodRunning
+		}
+		return atLeastOnceRunning, nil
+	}, ctx.Done())
+}
+
+func (c *Cluster) WaitForRegistrationAgentReady(ctx context.Context) error {
+	return wait.PollImmediateUntil(time.Second, func() (bool, error) {
+		podList := &corev1.PodList{}
+		err := c.Args.Client.List(ctx, podList,
+			client.InNamespace("open-cluster-management-agent"),
+			client.MatchingLabels{"app": "klusterlet-registration-agent"})
+		if err != nil {
+			return false, err
+		}
+		atLeastOnceRunning := false
+		for _, pod := range podList.Items {
+			atLeastOnceRunning = atLeastOnceRunning || pod.Status.Phase == corev1.PodRunning
+		}
+		return atLeastOnceRunning, nil
+	}, ctx.Done())
+}
+
 func applyHubKubeConfig(ctx context.Context, k8sClient client.Client, file string, kubeConfig *clientcmdapiv1.Config) error {
 	path := strings.Split(file, "/")
 	templateName := path[len(path)-1]