Skip to content

Commit

Permalink
Initialize the cluster client cache from the kubeconfig in secrets (#412
Browse files Browse the repository at this point in the history 
)

Add migration step to move ToolchainCluster.Spec.APIEndpoint status
and read the connection configuration from the kubeconfig stored in
the secret.
  • Loading branch information
@metlos
metlos authored Aug 26, 2024
1 parent cd921d6 commit 8833ded
Show file tree
Hide file tree
Showing 4 changed files with 193 additions and 21 deletions.
2 changes: 0 additions & 2 deletions controllers/toolchaincluster/healthchecker.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,11 @@ const (

// getClusterHealth gets the kubernetes cluster health status by requesting "/healthz"
func getClusterHealthStatus(ctx context.Context, remoteClusterClientset *kubeclientset.Clientset) (bool, error) {

lgr := log.FromContext(ctx)
body, err := remoteClusterClientset.DiscoveryClient.RESTClient().Get().AbsPath("/healthz").Do(ctx).Raw()
if err != nil {
lgr.Error(err, "Failed to do cluster health check for a ToolchainCluster")
return false, err
}
return strings.EqualFold(string(body), "ok"), nil

}
18 changes: 13 additions & 5 deletions controllers/toolchaincluster/toolchaincluster_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.
cachedCluster, ok := cluster.GetCachedToolchainCluster(toolchainCluster.Name)
if !ok {
err := fmt.Errorf("cluster %s not found in cache", toolchainCluster.Name)
if err := r.updateStatus(ctx, toolchainCluster, clusterOfflineCondition(err.Error())); err != nil {
if err := r.updateStatus(ctx, toolchainCluster, nil, clusterOfflineCondition(err.Error())); err != nil {
reqLogger.Error(err, "unable to update cluster status of ToolchainCluster")
}
return reconcile.Result{}, err
Expand All @@ -73,7 +73,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.
clientSet, err := kubeclientset.NewForConfig(cachedCluster.RestConfig)
if err != nil {
reqLogger.Error(err, "cannot create ClientSet for the ToolchainCluster")
if err := r.updateStatus(ctx, toolchainCluster, clusterOfflineCondition(err.Error())); err != nil {
if err := r.updateStatus(ctx, toolchainCluster, cachedCluster, clusterOfflineCondition(err.Error())); err != nil {
reqLogger.Error(err, "unable to update cluster status of ToolchainCluster")
}
return reconcile.Result{}, err
Expand All @@ -83,15 +83,24 @@ func (r *Reconciler) Reconcile(ctx context.Context, request ctrl.Request) (ctrl.
healthCheckResult := r.getClusterHealthCondition(ctx, clientSet)

// update the status of the individual cluster.
if err := r.updateStatus(ctx, toolchainCluster, healthCheckResult); err != nil {
if err := r.updateStatus(ctx, toolchainCluster, cachedCluster, healthCheckResult); err != nil {
reqLogger.Error(err, "unable to update cluster status of ToolchainCluster")
return reconcile.Result{}, err
}
return reconcile.Result{RequeueAfter: r.RequeAfter}, nil
}

func (r *Reconciler) updateStatus(ctx context.Context, toolchainCluster *toolchainv1alpha1.ToolchainCluster, currentConditions ...toolchainv1alpha1.Condition) error {
func (r *Reconciler) updateStatus(ctx context.Context, toolchainCluster *toolchainv1alpha1.ToolchainCluster, cachedToolchainCluster *cluster.CachedToolchainCluster, currentConditions ...toolchainv1alpha1.Condition) error {
toolchainCluster.Status.Conditions = condition.AddOrUpdateStatusConditionsWithLastUpdatedTimestamp(toolchainCluster.Status.Conditions, currentConditions...)

if cachedToolchainCluster != nil {
toolchainCluster.Status.APIEndpoint = cachedToolchainCluster.APIEndpoint
toolchainCluster.Status.OperatorNamespace = cachedToolchainCluster.OperatorNamespace
} else {
toolchainCluster.Status.APIEndpoint = ""
toolchainCluster.Status.OperatorNamespace = ""
}

if err := r.Client.Status().Update(ctx, toolchainCluster); err != nil {
return fmt.Errorf("failed to update the status of cluster - %s: %w", toolchainCluster.Name, err)
}
Expand All @@ -107,7 +116,6 @@ func (r *Reconciler) getClusterHealthCondition(ctx context.Context, remoteCluste
return clusterNotReadyCondition()
}
return clusterReadyCondition()

}

func (r *Reconciler) getClusterHealth(ctx context.Context, remoteClusterClientset *kubeclientset.Clientset) (bool, error) {
Expand Down
59 changes: 48 additions & 11 deletions pkg/cluster/service.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -160,16 +160,9 @@ func (s *ToolchainClusterService) enrichLogger(cluster *toolchainv1alpha1.Toolch

// NewClusterConfig generate a new cluster config by fetching the necessary info the given ToolchainCluster's associated Secret and taking all data from ToolchainCluster CR
func NewClusterConfig(cl client.Client, toolchainCluster *toolchainv1alpha1.ToolchainCluster, timeout time.Duration) (*Config, error) {
clusterName := toolchainCluster.Name

apiEndpoint := toolchainCluster.Spec.APIEndpoint
if apiEndpoint == "" {
return nil, errors.Errorf("the api endpoint of cluster %s is empty", clusterName)
}

secretName := toolchainCluster.Spec.SecretRef.Name
if secretName == "" {
return nil, errors.Errorf("cluster %s does not have a secret name", clusterName)
return nil, errors.Errorf("cluster %s does not have a secret name", toolchainCluster.Name)
}
secret := &v1.Secret{}
name := types.NamespacedName{
Expand All @@ -178,7 +171,22 @@ func NewClusterConfig(cl client.Client, toolchainCluster *toolchainv1alpha1.Tool
}
err := cl.Get(context.TODO(), name, secret)
if err != nil {
return nil, errors.Wrapf(err, "unable to get secret %s for cluster %s", name, clusterName)
return nil, errors.Wrapf(err, "unable to get secret %s for cluster %s", name, toolchainCluster.Name)
}

if _, ok := secret.Data["kubeconfig"]; ok {
return loadConfigFromKubeConfig(toolchainCluster, secret, timeout)
} else {
return loadConfigFromLegacyToolchainCluster(toolchainCluster, secret, timeout)
}
}

func loadConfigFromLegacyToolchainCluster(toolchainCluster *toolchainv1alpha1.ToolchainCluster, secret *v1.Secret, timeout time.Duration) (*Config, error) {
clusterName := toolchainCluster.Name

apiEndpoint := toolchainCluster.Spec.APIEndpoint
if apiEndpoint == "" {
return nil, errors.Errorf("the api endpoint of cluster %s is empty", clusterName)
}

token, tokenFound := secret.Data[toolchainTokenKey]
Expand All @@ -201,15 +209,44 @@ func NewClusterConfig(cl client.Client, toolchainCluster *toolchainv1alpha1.Tool
restConfig.Timeout = timeout

return &Config{
Name: toolchainCluster.Name,
APIEndpoint: toolchainCluster.Spec.APIEndpoint,
Name: clusterName,
APIEndpoint: apiEndpoint,
RestConfig: restConfig,
OperatorNamespace: toolchainCluster.Labels[labelNamespace],
OwnerClusterName: toolchainCluster.Labels[labelOwnerClusterName],
Labels: toolchainCluster.Labels,
}, nil
}

func loadConfigFromKubeConfig(toolchainCluster *toolchainv1alpha1.ToolchainCluster, secret *v1.Secret, timeout time.Duration) (*Config, error) {
cfg, err := clientcmd.Load(secret.Data["kubeconfig"])
if err != nil {
return nil, err
}
clientCfg := clientcmd.NewDefaultClientConfig(*cfg, &clientcmd.ConfigOverrides{})
restCfg, err := clientCfg.ClientConfig()
if err != nil {
return nil, err
}

// This is questionable, but the timeout is currently configurable in the member configuration so let's keep it here...
restCfg.Timeout = timeout

operatorNamespace, _, err := clientCfg.Namespace()
if err != nil {
return nil, fmt.Errorf("Could not determine the operator namespace from the current context in the provided kubeconfig because of: %w", err)
}

return &Config{
Name: toolchainCluster.Name,
APIEndpoint: restCfg.Host,
RestConfig: restCfg,
OperatorNamespace: operatorNamespace,
OwnerClusterName: toolchainCluster.Labels[labelOwnerClusterName],
Labels: toolchainCluster.Labels,
}, nil
}

func IsReady(clusterStatus *toolchainv1alpha1.ToolchainClusterStatus) bool {
for _, condition := range clusterStatus.Conditions {
if condition.Type == toolchainv1alpha1.ConditionReady {
Expand Down
135 changes: 132 additions & 3 deletions pkg/cluster/service_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,18 @@ import (
"testing"
"time"

metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

toolchainv1alpha1 "github.com/codeready-toolchain/api/api/v1alpha1"
"github.com/codeready-toolchain/toolchain-common/pkg/cluster"
"github.com/codeready-toolchain/toolchain-common/pkg/test"
"github.com/codeready-toolchain/toolchain-common/pkg/test/verify"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
corev1 "k8s.io/api/core/v1"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/tools/clientcmd"
clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
"sigs.k8s.io/controller-runtime/pkg/client"
)

Expand All @@ -22,7 +27,6 @@ func TestAddToolchainClusterAsMember(t *testing.T) {
// when
return service.AddOrUpdateToolchainCluster(toolchainCluster)
})

}

func TestAddToolchainClusterAsHost(t *testing.T) {
Expand Down Expand Up @@ -132,7 +136,7 @@ func TestListToolchainClusterConfigs(t *testing.T) {
})

t.Run("when list fails", func(t *testing.T) {
//given
// given
cl := test.NewFakeClient(t, m1, m2, host, noise, sec1, sec2, secHost, secNoise)
cl.MockList = func(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
return fmt.Errorf("some error")
Expand All @@ -147,7 +151,7 @@ func TestListToolchainClusterConfigs(t *testing.T) {
})

t.Run("when get secret fails", func(t *testing.T) {
//given
// given
cl := test.NewFakeClient(t, m1, m2, host, noise, sec1, sec2, secHost, secNoise)
cl.MockGet = func(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
return fmt.Errorf("some error")
Expand All @@ -161,3 +165,128 @@ func TestListToolchainClusterConfigs(t *testing.T) {
require.Empty(t, clusterConfigs)
})
}

func TestNewClusterConfig(t *testing.T) {
legacyTc := func() *toolchainv1alpha1.ToolchainCluster {
return &toolchainv1alpha1.ToolchainCluster{
ObjectMeta: metav1.ObjectMeta{
Name: "tc",
Namespace: "ns",
Labels: map[string]string{
"namespace": "operatorns",
},
},
Spec: toolchainv1alpha1.ToolchainClusterSpec{
APIEndpoint: "https://over.the.rainbow",
SecretRef: toolchainv1alpha1.LocalSecretReference{
Name: "secret",
},
},
}
}
newFormTc := func() *toolchainv1alpha1.ToolchainCluster {
return &toolchainv1alpha1.ToolchainCluster{
ObjectMeta: metav1.ObjectMeta{
Name: "tc",
Namespace: "ns",
},
Spec: toolchainv1alpha1.ToolchainClusterSpec{
SecretRef: toolchainv1alpha1.LocalSecretReference{
Name: "secret",
},
},
}
}

legacySecret := func() *corev1.Secret {
return &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: "secret",
Namespace: "ns",
},
Data: map[string][]byte{
"token": []byte("token"),
},
}
}
kubeconfigSecret := func(t *testing.T) *corev1.Secret {
t.Helper()
kubeconfig := clientcmdapi.Config{
Clusters: map[string]*clientcmdapi.Cluster{
"cluster": {
Server: "https://over.the.rainbow",
},
},
Contexts: map[string]*clientcmdapi.Context{
"ctx": {
Cluster: "cluster",
AuthInfo: "auth",
Namespace: "operatorns",
},
},
AuthInfos: map[string]*clientcmdapi.AuthInfo{
"auth": {
Token: "token",
},
},
CurrentContext: "ctx",
}
kubeconfigContents, err := clientcmd.Write(kubeconfig)
require.NoError(t, err)

return &corev1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: "secret",
Namespace: "ns",
},
Data: map[string][]byte{
"kubeconfig": kubeconfigContents,
},
}
}

t.Run("using legacy fields in ToolchainCluster and token in secret", func(t *testing.T) {
tc := legacyTc()
secret := legacySecret()

cl := test.NewFakeClient(t, tc, secret)

cfg, err := cluster.NewClusterConfig(cl, tc, 1*time.Second)
require.NoError(t, err)

assert.Equal(t, "https://over.the.rainbow", cfg.APIEndpoint)
assert.Equal(t, "operatorns", cfg.OperatorNamespace)
assert.Equal(t, "token", cfg.RestConfig.BearerToken)
})

t.Run("using kubeconfig in secret", func(t *testing.T) {
tc := newFormTc()
secret := kubeconfigSecret(t)

cl := test.NewFakeClient(t, tc, secret)

cfg, err := cluster.NewClusterConfig(cl, tc, 1*time.Second)
require.NoError(t, err)

assert.Equal(t, "https://over.the.rainbow", cfg.APIEndpoint)
assert.Equal(t, "operatorns", cfg.OperatorNamespace)
assert.Equal(t, "token", cfg.RestConfig.BearerToken)
})

t.Run("uses kubeconfig in precedence over legacy fields", func(t *testing.T) {
tc := newFormTc()
// Combine the kubeconfig and the token in the same secret.
// We should see auth from the kubeconfig used...
secret := kubeconfigSecret(t)
secret.Data["token"] = []byte("not-the-token-we-want")

cl := test.NewFakeClient(t, tc, secret)

cfg, err := cluster.NewClusterConfig(cl, tc, 1*time.Second)
require.NoError(t, err)

assert.Equal(t, "https://over.the.rainbow", cfg.APIEndpoint)
assert.Equal(t, "operatorns", cfg.OperatorNamespace)
assert.Equal(t, "token", cfg.RestConfig.BearerToken)
})
}

0 comments on commit 8833ded

Please sign in to comment.