Merge branch 'main' into fix-ingress

cli/cmds/cluster/create.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/rancher/k3k/cli/cmds"
 	"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
-	"github.com/rancher/k3k/pkg/controller"
 	k3kcluster "github.com/rancher/k3k/pkg/controller/cluster"
 	"github.com/rancher/k3k/pkg/controller/cluster/server"
 	"github.com/rancher/k3k/pkg/controller/kubeconfig"
@@ -22,9 +21,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/apiserver/pkg/authentication/user"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -190,11 +189,6 @@ func create(clx *cli.Context) error {
 	}
 
 	logrus.Infof("Extracting Kubeconfig for [%s] cluster", name)
-	cfg := &kubeconfig.KubeConfig{
-		CN:         controller.AdminCommonName,
-		ORG:        []string{user.SystemPrivilegedGroup},
-		ExpiryDate: 0,
-	}
 
 	logrus.Infof("waiting for cluster to be available..")
 
@@ -205,7 +199,9 @@ func create(clx *cli.Context) error {
 		Steps:    25,
 	}
 
-	var kubeconfig []byte
+	cfg := kubeconfig.New()
+
+	var kubeconfig *clientcmdapi.Config
 	if err := retry.OnError(availableBackoff, apierrors.IsNotFound, func() error {
 		kubeconfig, err = cfg.Extract(ctx, ctrlClient, cluster, host[0])
 		return err
@@ -224,7 +220,12 @@ func create(clx *cli.Context) error {
 	kubectl cluster-info
 	`, filepath.Join(pwd, cluster.Name+"-kubeconfig.yaml"))
 
-	return os.WriteFile(cluster.Name+"-kubeconfig.yaml", kubeconfig, 0644)
+	kubeconfigData, err := clientcmd.Write(*kubeconfig)
+	if err != nil {
+		return err
+	}
+
+	return os.WriteFile(cluster.Name+"-kubeconfig.yaml", kubeconfigData, 0644)
 }
 
 func validateCreateFlags() error {

cli/cmds/kubeconfig/kubeconfig.go

@@ -21,6 +21,7 @@ import (
 	"k8s.io/apiserver/pkg/authentication/user"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -142,23 +143,24 @@ func generate(clx *cli.Context) error {
 	if orgs == nil {
 		orgs = []string{user.SystemPrivilegedGroup}
 	}
+
 	cfg := kubeconfig.KubeConfig{
 		CN:         cn,
 		ORG:        orgs,
 		ExpiryDate: time.Hour * 24 * time.Duration(expirationDays),
 		AltNames:   certAltNames,
 	}
+
 	logrus.Infof("waiting for cluster to be available..")
-	var kubeconfig []byte
+
+	var kubeconfig *clientcmdapi.Config
 	if err := retry.OnError(controller.Backoff, apierrors.IsNotFound, func() error {
 		kubeconfig, err = cfg.Extract(ctx, ctrlClient, &cluster, host[0])
-		if err != nil {
-			return err
-		}
-		return nil
+		return err
 	}); err != nil {
 		return err
 	}
+
 	pwd, err := os.Getwd()
 	if err != nil {
 		return err
@@ -174,5 +176,10 @@ func generate(clx *cli.Context) error {
 	kubectl cluster-info
 	`, filepath.Join(pwd, configName))
 
-	return os.WriteFile(configName, kubeconfig, 0644)
+	kubeconfigData, err := clientcmd.Write(*kubeconfig)
+	if err != nil {
+		return err
+	}
+
+	return os.WriteFile(configName, kubeconfigData, 0644)
 }

k3k-kubelet/controller/webhook/pod.go

@@ -11,6 +11,7 @@ import (
 	"github.com/rancher/k3k/pkg/log"
 	admissionregistrationv1 "k8s.io/api/admissionregistration/v1"
 	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -56,7 +57,9 @@ func AddPodMutatorWebhook(ctx context.Context, mgr manager.Manager, hostClient c
 		return err
 	}
 	if err := handler.client.Create(ctx, config); err != nil {
-		return err
+		if !apierrors.IsAlreadyExists(err) {
+			return err
+		}
 	}
 	// register webhook with the manager
 	return ctrl.NewWebhookManagedBy(mgr).For(&v1.Pod{}).WithDefaulter(&handler).Complete()

k3k-kubelet/kubelet.go

@@ -25,7 +25,6 @@ import (
 	"github.com/virtual-kubelet/virtual-kubelet/node/nodeutil"
 	"go.uber.org/zap"
 	v1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apiserver/pkg/authentication/user"
@@ -129,9 +128,7 @@ func newKubelet(ctx context.Context, c *config, logger *k3klog.Logger) (*kubelet
 	}
 	logger.Info("adding pod mutator webhook")
 	if err := k3kwebhook.AddPodMutatorWebhook(ctx, virtualMgr, hostClient, c.ClusterName, c.ClusterNamespace, c.NodeName, logger); err != nil {
-		if !apierrors.IsAlreadyExists(err) {
-			return nil, errors.New("unable to add pod mutator webhook for virtual cluster: " + err.Error())
-		}
+		return nil, errors.New("unable to add pod mutator webhook for virtual cluster: " + err.Error())
 	}
 
 	logger.Info("adding service syncer controller")

pkg/controller/cluster/cluster.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"strings"
 	"time"
 
 	"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
@@ -98,7 +99,8 @@ func (c *ClusterReconciler) Reconcile(ctx context.Context, req reconcile.Request
 		}
 
 		// update Status HostVersion
-		cluster.Status.HostVersion = fmt.Sprintf("v%s.%s.0-k3s1", hostVersion.Major, hostVersion.Minor)
+		k8sVersion := strings.Split(hostVersion.GitVersion, "+")[0]
+		cluster.Status.HostVersion = k8sVersion + "-k3s1"
 		if err := c.Client.Status().Update(ctx, &cluster); err != nil {
 			return reconcile.Result{}, err
 		}

pkg/controller/cluster/cluster_test.go

@@ -51,7 +51,7 @@ var _ = Describe("Cluster Controller", func() {
 
 				serverVersion, err := k8s.DiscoveryClient.ServerVersion()
 				Expect(err).To(Not(HaveOccurred()))
-				expectedHostVersion := fmt.Sprintf("v%s.%s.0-k3s1", serverVersion.Major, serverVersion.Minor)
+				expectedHostVersion := fmt.Sprintf("%s-k3s1", serverVersion.GitVersion)
 
 				Eventually(func() string {
 					err := k8sClient.Get(ctx, client.ObjectKeyFromObject(cluster), cluster)

pkg/controller/cluster/server/bootstrap/bootstrap.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"time"
 
@@ -12,7 +13,9 @@ import (
 	"github.com/rancher/k3k/pkg/controller"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type ControlRuntimeBootstrap struct {
@@ -171,3 +174,24 @@ func DecodedBootstrap(token, ip string) (*ControlRuntimeBootstrap, error) {
 
 	return bootstrap, nil
 }
+
+func GetFromSecret(ctx context.Context, client client.Client, cluster *v1alpha1.Cluster) (*ControlRuntimeBootstrap, error) {
+	key := types.NamespacedName{
+		Name:      controller.SafeConcatNameWithPrefix(cluster.Name, "bootstrap"),
+		Namespace: cluster.Namespace,
+	}
+
+	var bootstrapSecret v1.Secret
+	if err := client.Get(ctx, key, &bootstrapSecret); err != nil {
+		return nil, err
+	}
+
+	bootstrapData := bootstrapSecret.Data["bootstrap"]
+	if bootstrapData == nil {
+		return nil, errors.New("empty bootstrap")
+	}
+
+	var bootstrap ControlRuntimeBootstrap
+	err := json.Unmarshal(bootstrapData, &bootstrap)
+	return &bootstrap, err
+}

pkg/controller/kubeconfig/kubeconfig.go

@@ -3,8 +3,6 @@ package kubeconfig
 import (
 	"context"
 	"crypto/x509"
-	"encoding/json"
-	"errors"
 	"fmt"
 	"time"
 
@@ -17,7 +15,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/apiserver/pkg/authentication/user"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -29,72 +27,45 @@ type KubeConfig struct {
 	ExpiryDate time.Duration
 }
 
-func (k *KubeConfig) Extract(ctx context.Context, client client.Client, cluster *v1alpha1.Cluster, hostServerIP string) ([]byte, error) {
-	nn := types.NamespacedName{
-		Name:      controller.SafeConcatNameWithPrefix(cluster.Name, "bootstrap"),
-		Namespace: cluster.Namespace,
+func New() *KubeConfig {
+	return &KubeConfig{
+		CN:         controller.AdminCommonName,
+		ORG:        []string{user.SystemPrivilegedGroup},
+		ExpiryDate: 0,
 	}
-
-	var bootstrapSecret v1.Secret
-	if err := client.Get(ctx, nn, &bootstrapSecret); err != nil {
-		return nil, err
-	}
-
-	bootstrapData := bootstrapSecret.Data["bootstrap"]
-	if bootstrapData == nil {
-		return nil, errors.New("empty bootstrap")
-	}
-
-	var bootstrap bootstrap.ControlRuntimeBootstrap
-	if err := json.Unmarshal(bootstrapData, &bootstrap); err != nil {
-		return nil, err
-	}
-
-	adminCert, adminKey, err := certs.CreateClientCertKey(
-		k.CN, k.ORG,
-		&k.AltNames, []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, k.ExpiryDate,
-		bootstrap.ClientCA.Content,
-		bootstrap.ClientCAKey.Content)
-	if err != nil {
-		return nil, err
-	}
-	// get the server service to extract the right IP
-	nn = types.NamespacedName{
-		Name:      server.ServiceName(cluster.Name),
-		Namespace: cluster.Namespace,
-	}
-
-	var k3kService v1.Service
-	if err := client.Get(ctx, nn, &k3kService); err != nil {
-		return nil, err
-	}
-
-	url := fmt.Sprintf("https://%s:%d", k3kService.Spec.ClusterIP, server.ServerPort)
-	if k3kService.Spec.Type == v1.ServiceTypeNodePort {
-		nodePort := k3kService.Spec.Ports[0].NodePort
-		url = fmt.Sprintf("https://%s:%d", hostServerIP, nodePort)
-	}
-	if cluster.Spec.Expose.Ingress.Enabled {
-		var k3kIngress networkingv1.Ingress
-		nn = types.NamespacedName{
-			Name:      server.IngressName(cluster.Name),
-			Namespace: cluster.Namespace,
-		}
-
-		if err := client.Get(ctx, nn, &k3kIngress); err != nil {
-			return nil, err
-		}
-		url = fmt.Sprintf("https://%s", k3kIngress.Spec.Rules[0].Host)
-	}
-	kubeconfigData, err := kubeconfig(url, []byte(bootstrap.ServerCA.Content), adminCert, adminKey)
-	if err != nil {
-		return nil, err
-	}
-
-	return kubeconfigData, nil
 }
 
-func kubeconfig(url string, serverCA, clientCert, clientKey []byte) ([]byte, error) {
+func (k *KubeConfig) Extract(ctx context.Context, client client.Client, cluster *v1alpha1.Cluster, hostServerIP string) (*clientcmdapi.Config, error) {
+	bootstrapData, err := bootstrap.GetFromSecret(ctx, client, cluster)
+	if err != nil {
+		return nil, err
+	}
+	serverCACert := []byte(bootstrapData.ServerCA.Content)
+
+	adminCert, adminKey, err := certs.CreateClientCertKey(
+		k.CN,
+		k.ORG,
+		&k.AltNames,
+		[]x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
+		k.ExpiryDate,
+		bootstrapData.ClientCA.Content,
+		bootstrapData.ClientCAKey.Content,
+	)
+	if err != nil {
+		return nil, err
+	}
+  
+	url, err := getURLFromService(ctx, client, cluster, hostServerIP)
+	if err != nil {
+		return nil, err
+	}
+
+	config := NewConfig(url, serverCACert, adminCert, adminKey)
+
+	return config, nil
+}
+
+func NewConfig(url string, serverCA, clientCert, clientKey []byte) *clientcmdapi.Config {
 	config := clientcmdapi.NewConfig()
 
 	cluster := clientcmdapi.NewCluster()
@@ -114,10 +85,39 @@ func kubeconfig(url string, serverCA, clientCert, clientKey []byte) ([]byte, err
 	config.Contexts["default"] = context
 	config.CurrentContext = "default"
 
-	kubeconfig, err := clientcmd.Write(*config)
-	if err != nil {
-		return nil, err
+	return config
+}
+
+func getURLFromService(ctx context.Context, client client.Client, cluster *v1alpha1.Cluster, hostServerIP string) (string, error) {
+	// get the server service to extract the right IP
+	key := types.NamespacedName{
+		Name:      server.ServiceName(cluster.Name),
+		Namespace: cluster.Namespace,
 	}
 
-	return kubeconfig, nil
+	var k3kService v1.Service
+	if err := client.Get(ctx, key, &k3kService); err != nil {
+		return "", err
+	}
+
+	url := fmt.Sprintf("https://%s:%d", k3kService.Spec.ClusterIP, server.ServerPort)
+
+	if k3kService.Spec.Type == v1.ServiceTypeNodePort {
+		nodePort := k3kService.Spec.Ports[0].NodePort
+		url = fmt.Sprintf("https://%s:%d", hostServerIP, nodePort)
+	}
+  if cluster.Spec.Expose.Ingress.Enabled {
+		var k3kIngress networkingv1.Ingress
+		ingressKey = types.NamespacedName{
+			Name:      server.IngressName(cluster.Name),
+			Namespace: cluster.Namespace,
+		}
+
+		if err := client.Get(ctx, ingressKey, &k3kIngress); err != nil {
+			return "", err
+		}
+		url = fmt.Sprintf("https://%s", k3kIngress.Spec.Rules[0].Host)
+	}
+
+	return url, nil
 }

tests/cluster_test.go

@@ -8,10 +8,15 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/rancher/k3k/k3k-kubelet/translate"
 	"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
+	"github.com/rancher/k3k/pkg/controller/certs"
+	"github.com/rancher/k3k/pkg/controller/kubeconfig"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/utils/ptr"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 )
 
 var _ = When("a cluster is installed", func() {
@@ -26,45 +31,60 @@ var _ = When("a cluster is installed", func() {
 	})
 
 	It("will be created in shared mode", func() {
+		ctx := context.Background()
+		containerIP, err := k3sContainer.ContainerIP(ctx)
+		Expect(err).To(Not(HaveOccurred()))
+
+		fmt.Fprintln(GinkgoWriter, "K3s containerIP: "+containerIP)
+
 		cluster := v1alpha1.Cluster{
 			ObjectMeta: v1.ObjectMeta{
 				Name:      "mycluster",
 				Namespace: namespace,
 			},
 			Spec: v1alpha1.ClusterSpec{
-				Mode:    v1alpha1.SharedClusterMode,
-				Servers: ptr.To[int32](1),
-				Agents:  ptr.To[int32](0),
-				Version: "v1.26.1-k3s1",
+				TLSSANs: []string{containerIP},
+				Expose: &v1alpha1.ExposeConfig{
+					NodePort: &v1alpha1.NodePortConfig{
+						Enabled: true,
+					},
+				},
 			},
 		}
+		virtualK8sClient := CreateCluster(containerIP, cluster)
 
-		err := k8sClient.Create(context.Background(), &cluster)
+		nginxPod := &corev1.Pod{
+			ObjectMeta: v1.ObjectMeta{
+				Name:      "nginx",
+				Namespace: "default",
+			},
+			Spec: corev1.PodSpec{
+				Containers: []corev1.Container{{
+					Name:  "nginx",
+					Image: "nginx",
+				}},
+			},
+		}
+		nginxPod, err = virtualK8sClient.CoreV1().Pods(nginxPod.Namespace).Create(ctx, nginxPod, v1.CreateOptions{})
 		Expect(err).To(Not(HaveOccurred()))
 
-		By("checking server and kubelet readiness state")
-
-		// check that the server Pod and the Kubelet are in Ready state
+		// check that the nginx Pod is up and running in the host cluster
 		Eventually(func() bool {
-			podList, err := k8s.CoreV1().Pods(namespace).List(context.Background(), v1.ListOptions{})
+			//labelSelector := fmt.Sprintf("%s=%s", translate.ClusterNameLabel, cluster.Namespace)
+			podList, err := k8s.CoreV1().Pods(namespace).List(ctx, v1.ListOptions{})
 			Expect(err).To(Not(HaveOccurred()))
 
-			serverRunning := false
-			kubeletRunning := false
-
 			for _, pod := range podList.Items {
-				imageName := pod.Spec.Containers[0].Image
-				imageName = strings.Split(imageName, ":")[0] // remove tag
+				resourceName := pod.Annotations[translate.ResourceNameAnnotation]
+				resourceNamespace := pod.Annotations[translate.ResourceNamespaceAnnotation]
 
-				switch imageName {
-				case "rancher/k3s":
-					serverRunning = pod.Status.Phase == corev1.PodRunning
-				case "rancher/k3k-kubelet":
-					kubeletRunning = pod.Status.Phase == corev1.PodRunning
-				}
+				fmt.Fprintf(GinkgoWriter,
+					"pod=%s resource=%s/%s status=%s\n",
+					pod.Name, resourceNamespace, resourceName, pod.Status.Phase,
+				)
 
-				if serverRunning && kubeletRunning {
-					return true
+				if resourceName == nginxPod.Name && resourceNamespace == nginxPod.Namespace {
+					return pod.Status.Phase == corev1.PodRunning
 				}
 			}
 
@@ -73,16 +93,74 @@ var _ = When("a cluster is installed", func() {
 			WithTimeout(time.Minute).
 			WithPolling(time.Second * 5).
 			Should(BeTrue())
-
-		By("checking the existence of the bootstrap secret")
-		secretName := fmt.Sprintf("k3k-%s-bootstrap", cluster.Name)
-
-		Eventually(func() error {
-			_, err := k8s.CoreV1().Secrets(namespace).Get(context.Background(), secretName, v1.GetOptions{})
-			return err
-		}).
-			WithTimeout(time.Minute * 2).
-			WithPolling(time.Second * 5).
-			Should(BeNil())
 	})
 })
+
+func CreateCluster(hostIP string, cluster v1alpha1.Cluster) *kubernetes.Clientset {
+	GinkgoHelper()
+
+	By(fmt.Sprintf("Creating virtual cluster %s/%s", cluster.Namespace, cluster.Name))
+
+	ctx := context.Background()
+	err := k8sClient.Create(ctx, &cluster)
+	Expect(err).To(Not(HaveOccurred()))
+
+	By("Waiting for server and kubelet to be ready")
+
+	// check that the server Pod and the Kubelet are in Ready state
+	Eventually(func() bool {
+		podList, err := k8s.CoreV1().Pods(cluster.Namespace).List(ctx, v1.ListOptions{})
+		Expect(err).To(Not(HaveOccurred()))
+
+		serverRunning := false
+		kubeletRunning := false
+
+		for _, pod := range podList.Items {
+			imageName := pod.Spec.Containers[0].Image
+			imageName = strings.Split(imageName, ":")[0] // remove tag
+
+			switch imageName {
+			case "rancher/k3s":
+				serverRunning = pod.Status.Phase == corev1.PodRunning
+			case "rancher/k3k-kubelet":
+				kubeletRunning = pod.Status.Phase == corev1.PodRunning
+			}
+
+			if serverRunning && kubeletRunning {
+				return true
+			}
+		}
+
+		return false
+	}).
+		WithTimeout(time.Minute).
+		WithPolling(time.Second * 5).
+		Should(BeTrue())
+
+	By("Waiting for server to be up and running")
+
+	var config *clientcmdapi.Config
+	Eventually(func() error {
+		vKubeconfig := kubeconfig.New()
+		vKubeconfig.AltNames = certs.AddSANs([]string{hostIP, "k3k-mycluster-kubelet"})
+		config, err = vKubeconfig.Extract(ctx, k8sClient, &cluster, hostIP)
+		return err
+	}).
+		WithTimeout(time.Minute * 2).
+		WithPolling(time.Second * 5).
+		Should(BeNil())
+
+	configData, err := clientcmd.Write(*config)
+	Expect(err).To(Not(HaveOccurred()))
+
+	restcfg, err := clientcmd.RESTConfigFromKubeConfig(configData)
+	Expect(err).To(Not(HaveOccurred()))
+	virtualK8sClient, err := kubernetes.NewForConfig(restcfg)
+	Expect(err).To(Not(HaveOccurred()))
+
+	serverVersion, err := virtualK8sClient.DiscoveryClient.ServerVersion()
+	Expect(err).To(Not(HaveOccurred()))
+	fmt.Fprintf(GinkgoWriter, "serverVersion: %+v\n", serverVersion)
+
+	return virtualK8sClient
+}

tests/tests_suite_test.go

@@ -173,7 +173,11 @@ var _ = When("k3k is installed", func() {
 
 func buildScheme() *runtime.Scheme {
 	scheme := runtime.NewScheme()
-	err := v1alpha1.AddToScheme(scheme)
+
+	err := corev1.AddToScheme(scheme)
 	Expect(err).NotTo(HaveOccurred())
+	err = v1alpha1.AddToScheme(scheme)
+	Expect(err).NotTo(HaveOccurred())
+
 	return scheme
 }