Serviceaccount token synchronization (#139)

* Serviceaccount token sync

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* fixes

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

* fixing typo

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

---------

Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>

k3k-kubelet/config.go

@@ -17,6 +17,7 @@ type config struct {
 	HostConfigPath    string `yaml:"hostConfigPath,omitempty"`
 	VirtualConfigPath string `yaml:"virtualConfigPath,omitempty"`
 	KubeletPort       string `yaml:"kubeletPort,omitempty"`
+	AgentIP           string `yaml:"agentIP,omitempty"`
 }
 
 func (c *config) unmarshalYAML(data []byte) error {
@@ -50,6 +51,9 @@ func (c *config) unmarshalYAML(data []byte) error {
 	if c.Token == "" {
 		c.Token = conf.Token
 	}
+	if c.AgentIP == "" {
+		c.AgentIP = conf.AgentIP
+	}
 	return nil
 }
 

k3k-kubelet/kubelet.go

@@ -124,8 +124,8 @@ func newKubelet(ctx context.Context, c *config, logger *k3klog.Logger) (*kubelet
 	}, nil
 }
 
-func (k *kubelet) registerNode(ctx context.Context, srvPort, namespace, name, hostname string) error {
-	providerFunc := k.newProviderFunc(namespace, name, hostname)
+func (k *kubelet) registerNode(ctx context.Context, ip, srvPort, namespace, name, hostname string) error {
+	providerFunc := k.newProviderFunc(namespace, name, hostname, ip)
 	nodeOpts := k.nodeOpts(ctx, srvPort, namespace, name, hostname)
 
 	var err error
@@ -172,7 +172,7 @@ func (k *kubelet) start(ctx context.Context) {
 	k.logger.Info("node exited successfully")
 }
 
-func (k *kubelet) newProviderFunc(namespace, name, hostname string) nodeutil.NewProviderFunc {
+func (k *kubelet) newProviderFunc(namespace, name, hostname, ip string) nodeutil.NewProviderFunc {
 	return func(pc nodeutil.ProviderConfig) (nodeutil.Provider, node.NodeProvider, error) {
 		utilProvider, err := provider.New(*k.hostConfig, k.hostMgr, k.virtualMgr, k.logger, namespace, name)
 		if err != nil {
@@ -180,7 +180,7 @@ func (k *kubelet) newProviderFunc(namespace, name, hostname string) nodeutil.New
 		}
 		nodeProvider := provider.Node{}
 
-		provider.ConfigureNode(pc.Node, hostname, k.port)
+		provider.ConfigureNode(pc.Node, hostname, k.port, ip)
 		return utilProvider, &nodeProvider, nil
 	}
 }

k3k-kubelet/main.go

@@ -4,10 +4,12 @@ import (
 	"context"
 	"os"
 
+	"github.com/go-logr/zapr"
 	"github.com/rancher/k3k/pkg/log"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
 	"go.uber.org/zap"
+	ctrlruntimelog "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 var (
@@ -65,6 +67,12 @@ func main() {
 			Destination: &cfg.AgentHostname,
 			EnvVar:      "AGENT_HOSTNAME",
 		},
+		cli.StringFlag{
+			Name:        "agent-ip",
+			Usage:       "Agent IP used for registering the virtual kubelet to the cluster",
+			Destination: &cfg.AgentIP,
+			EnvVar:      "AGENT_IP",
+		},
 		cli.StringFlag{
 			Name:        "config",
 			Usage:       "Path to k3k-kubelet config file",
@@ -81,6 +89,7 @@ func main() {
 	}
 	app.Before = func(clx *cli.Context) error {
 		logger = log.New(debug)
+		ctrlruntimelog.SetLogger(zapr.NewLogger(logger.Desugar().WithOptions(zap.AddCallerSkip(1))))
 		return nil
 	}
 	app.Action = run
@@ -103,7 +112,7 @@ func run(clx *cli.Context) {
 		logger.Fatalw("failed to create new virtual kubelet instance", zap.Error(err))
 	}
 
-	if err := k.registerNode(ctx, cfg.KubeletPort, cfg.ClusterNamespace, cfg.ClusterName, cfg.AgentHostname); err != nil {
+	if err := k.registerNode(ctx, cfg.AgentIP, cfg.KubeletPort, cfg.ClusterNamespace, cfg.ClusterName, cfg.AgentHostname); err != nil {
 		logger.Fatalw("failed to register new node", zap.Error(err))
 	}
 

k3k-kubelet/provider/configure.go

@@ -6,7 +6,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func ConfigureNode(node *v1.Node, hostname string, servicePort int) {
+func ConfigureNode(node *v1.Node, hostname string, servicePort int, ip string) {
 	node.Status.Conditions = nodeConditions()
 	node.Status.DaemonEndpoints.KubeletEndpoint.Port = int32(servicePort)
 	node.Status.Addresses = []v1.NodeAddress{
@@ -14,6 +14,10 @@ func ConfigureNode(node *v1.Node, hostname string, servicePort int) {
 			Type:    v1.NodeHostName,
 			Address: hostname,
 		},
+		{
+			Type:    v1.NodeInternalIP,
+			Address: ip,
+		},
 	}
 	node.Status.Capacity = v1.ResourceList{
 		// TODO: Make this more dynamic based on the sum of existing nodes

k3k-kubelet/provider/provider.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 	"strconv"
+	"strings"
 
 	dto "github.com/prometheus/client_model/go"
 	"github.com/rancher/k3k/k3k-kubelet/controller"
@@ -15,6 +16,7 @@ import (
 	"github.com/virtual-kubelet/virtual-kubelet/node/api"
 	"github.com/virtual-kubelet/virtual-kubelet/node/api/statsv1alpha1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/selection"
@@ -240,6 +242,10 @@ func (p *Provider) CreatePod(ctx context.Context, pod *corev1.Pod) error {
 	if err := p.transformVolumes(ctx, pod.Namespace, tPod.Spec.Volumes); err != nil {
 		return fmt.Errorf("unable to sync volumes for pod %s/%s: %w", pod.Namespace, pod.Name, err)
 	}
+	// sync serviceaccount token to a the host cluster
+	if err := p.transformTokens(ctx, pod, tPod); err != nil {
+		return fmt.Errorf("unable to transform tokens for pod %s/%s: %w", pod.Namespace, pod.Name, err)
+	}
 	p.logger.Infow("Creating pod", "Host Namespace", tPod.Namespace, "Host Name", tPod.Name,
 		"Virtual Namespace", pod.Namespace, "Virtual Name", pod.Name)
 	return p.HostClient.Create(ctx, tPod)
@@ -249,28 +255,44 @@ func (p *Provider) CreatePod(ctx context.Context, pod *corev1.Pod) error {
 // if one/more volumes couldn't be transformed
 func (p *Provider) transformVolumes(ctx context.Context, podNamespace string, volumes []corev1.Volume) error {
 	for _, volume := range volumes {
+		var optional bool
+		if strings.HasPrefix(volume.Name, kubeAPIAccessPrefix) {
+			continue
+		}
 		// note: this needs to handle downward api volumes as well, but more thought is needed on how to do that
 		if volume.ConfigMap != nil {
-			if err := p.syncConfigmap(ctx, podNamespace, volume.ConfigMap.Name); err != nil {
+			if volume.ConfigMap.Optional != nil {
+				optional = *volume.ConfigMap.Optional
+			}
+			if err := p.syncConfigmap(ctx, podNamespace, volume.ConfigMap.Name, optional); err != nil {
 				return fmt.Errorf("unable to sync configmap volume %s: %w", volume.Name, err)
 			}
 			volume.ConfigMap.Name = p.Translater.TranslateName(podNamespace, volume.ConfigMap.Name)
 		} else if volume.Secret != nil {
-			if err := p.syncSecret(ctx, podNamespace, volume.Secret.SecretName); err != nil {
+			if volume.Secret.Optional != nil {
+				optional = *volume.Secret.Optional
+			}
+			if err := p.syncSecret(ctx, podNamespace, volume.Secret.SecretName, optional); err != nil {
 				return fmt.Errorf("unable to sync secret volume %s: %w", volume.Name, err)
 			}
 			volume.Secret.SecretName = p.Translater.TranslateName(podNamespace, volume.Secret.SecretName)
 		} else if volume.Projected != nil {
 			for _, source := range volume.Projected.Sources {
 				if source.ConfigMap != nil {
+					if source.ConfigMap.Optional != nil {
+						optional = *source.ConfigMap.Optional
+					}
 					configMapName := source.ConfigMap.Name
-					if err := p.syncConfigmap(ctx, podNamespace, configMapName); err != nil {
+					if err := p.syncConfigmap(ctx, podNamespace, configMapName, optional); err != nil {
 						return fmt.Errorf("unable to sync projected configmap %s: %w", configMapName, err)
 					}
 					source.ConfigMap.Name = p.Translater.TranslateName(podNamespace, configMapName)
 				} else if source.Secret != nil {
+					if source.Secret.Optional != nil {
+						optional = *source.Secret.Optional
+					}
 					secretName := source.Secret.Name
-					if err := p.syncSecret(ctx, podNamespace, secretName); err != nil {
+					if err := p.syncSecret(ctx, podNamespace, secretName, optional); err != nil {
 						return fmt.Errorf("unable to sync projected secret %s: %w", secretName, err)
 					}
 				}
@@ -280,7 +302,7 @@ func (p *Provider) transformVolumes(ctx context.Context, podNamespace string, vo
 	return nil
 }
 
-func (p *Provider) syncConfigmap(ctx context.Context, podNamespace string, configMapName string) error {
+func (p *Provider) syncConfigmap(ctx context.Context, podNamespace string, configMapName string, optional bool) error {
 	var configMap corev1.ConfigMap
 	nsName := types.NamespacedName{
 		Namespace: podNamespace,
@@ -288,6 +310,10 @@ func (p *Provider) syncConfigmap(ctx context.Context, podNamespace string, confi
 	}
 	err := p.VirtualClient.Get(ctx, nsName, &configMap)
 	if err != nil {
+		// check if its optional configmap
+		if apierrors.IsNotFound(err) && optional {
+			return nil
+		}
 		return fmt.Errorf("unable to get configmap to sync %s/%s: %w", nsName.Namespace, nsName.Name, err)
 	}
 	err = p.Handler.AddResource(ctx, &configMap)
@@ -297,7 +323,7 @@ func (p *Provider) syncConfigmap(ctx context.Context, podNamespace string, confi
 	return nil
 }
 
-func (p *Provider) syncSecret(ctx context.Context, podNamespace string, secretName string) error {
+func (p *Provider) syncSecret(ctx context.Context, podNamespace string, secretName string, optional bool) error {
 	var secret corev1.Secret
 	nsName := types.NamespacedName{
 		Namespace: podNamespace,
@@ -305,7 +331,10 @@ func (p *Provider) syncSecret(ctx context.Context, podNamespace string, secretNa
 	}
 	err := p.VirtualClient.Get(ctx, nsName, &secret)
 	if err != nil {
-		return fmt.Errorf("unable to get configmap to sync %s/%s: %w", nsName.Namespace, nsName.Name, err)
+		if apierrors.IsNotFound(err) && optional {
+			return nil
+		}
+		return fmt.Errorf("unable to get secret to sync %s/%s: %w", nsName.Namespace, nsName.Name, err)
 	}
 	err = p.Handler.AddResource(ctx, &secret)
 	if err != nil {
@@ -415,7 +444,7 @@ func (p *Provider) pruneUnusedVolumes(ctx context.Context, pod *corev1.Pod) erro
 // concurrently outside of the calling goroutine. Therefore it is recommended
 // to return a version after DeepCopy.
 func (p *Provider) GetPod(ctx context.Context, namespace, name string) (*corev1.Pod, error) {
-	p.logger.Errorf("got a request for get pod %s, %s", namespace, name)
+	p.logger.Infow("got a request for get pod", "Namespace", namespace, "Name", name)
 	hostNamespaceName := types.NamespacedName{
 		Namespace: p.ClusterNamespace,
 		Name:      p.Translater.TranslateName(namespace, name),
@@ -434,13 +463,12 @@ func (p *Provider) GetPod(ctx context.Context, namespace, name string) (*corev1.
 // concurrently outside of the calling goroutine. Therefore it is recommended
 // to return a version after DeepCopy.
 func (p *Provider) GetPodStatus(ctx context.Context, namespace, name string) (*corev1.PodStatus, error) {
-	p.logger.Errorf("got a request for pod status %s, %s", namespace, name)
+	p.logger.Infow("got a request for pod status", "Namespace", namespace, "Name", name)
 	pod, err := p.GetPod(ctx, namespace, name)
 	if err != nil {
-		p.logger.Errorf("error when getting pod %s, %s: %w", namespace, name, err)
 		return nil, fmt.Errorf("unable to get pod for status: %w", err)
 	}
-	p.logger.Errorf("got pod status %s, %s: %+v", namespace, name, pod.Status)
+	p.logger.Debugw("got pod status", "Namespace", namespace, "Name", name, "Status", pod.Status)
 	return pod.Status.DeepCopy(), nil
 }
 

k3k-kubelet/provider/token.go

@@ -0,0 +1,118 @@
+package provider
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	k3kcontroller "github.com/rancher/k3k/pkg/controller"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
+)
+
+const (
+	kubeAPIAccessPrefix          = "kube-api-access"
+	serviceAccountTokenMountPath = "/var/run/secrets/kubernetes.io/serviceaccount"
+)
+
+// transformTokens copies the serviceaccount tokens used by pod's serviceaccount to a secret on the host cluster and mount it
+// to look like the serviceaccount token
+func (p *Provider) transformTokens(ctx context.Context, pod, tPod *corev1.Pod) error {
+	p.logger.Infow("transforming token", "Pod", pod.Name, "Namespace", pod.Namespace, "serviceAccountName", pod.Spec.ServiceAccountName)
+
+	virtualSecretName := k3kcontroller.SafeConcatNameWithPrefix(pod.Spec.ServiceAccountName, "token")
+	virtualSecret := virtualSecret(virtualSecretName, pod.Namespace, pod.Spec.ServiceAccountName)
+	if err := p.VirtualClient.Create(ctx, virtualSecret); err != nil {
+		if !apierrors.IsAlreadyExists(err) {
+			return err
+		}
+	}
+	// extracting the tokens data from the secret we just created
+	virtualSecretKey := types.NamespacedName{
+		Name:      virtualSecret.Name,
+		Namespace: virtualSecret.Namespace,
+	}
+	if err := p.VirtualClient.Get(ctx, virtualSecretKey, virtualSecret); err != nil {
+		return err
+	}
+	// To avoid race conditions we need to check if the secret's data has been populated
+	// including the token, ca.crt and namespace
+	if len(virtualSecret.Data) < 3 {
+		return fmt.Errorf("token secret %s/%s data is empty", virtualSecret.Namespace, virtualSecret.Name)
+	}
+	hostSecret := virtualSecret.DeepCopy()
+	hostSecret.Type = ""
+	hostSecret.Annotations = make(map[string]string)
+	p.Translater.TranslateTo(hostSecret)
+
+	if err := p.HostClient.Create(ctx, hostSecret); err != nil {
+		if !apierrors.IsAlreadyExists(err) {
+			return err
+		}
+	}
+	p.translateToken(tPod, hostSecret.Name)
+	return nil
+}
+
+func virtualSecret(name, namespace, serviceAccountName string) *corev1.Secret {
+	return &corev1.Secret{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Secret",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+			Annotations: map[string]string{
+				corev1.ServiceAccountNameKey: serviceAccountName,
+			},
+		},
+		Type: corev1.SecretTypeServiceAccountToken,
+	}
+}
+
+// translateToken will remove the serviceaccount from the pod and replace the kube-api-access volume
+// with a custom token volume and mount it to all containers within the pod
+func (p *Provider) translateToken(pod *corev1.Pod, hostSecretName string) {
+	pod.Spec.ServiceAccountName = ""
+	pod.Spec.DeprecatedServiceAccount = ""
+	pod.Spec.AutomountServiceAccountToken = ptr.To(false)
+	removeKubeAccessVolume(pod)
+	addKubeAccessVolume(pod, hostSecretName)
+}
+
+func removeKubeAccessVolume(pod *corev1.Pod) {
+	for i, volume := range pod.Spec.Volumes {
+		if strings.HasPrefix(volume.Name, kubeAPIAccessPrefix) {
+			pod.Spec.Volumes = append(pod.Spec.Volumes[:i], pod.Spec.Volumes[i+1:]...)
+		}
+	}
+	for i, container := range pod.Spec.Containers {
+		for j, mountPath := range container.VolumeMounts {
+			if strings.HasPrefix(mountPath.Name, kubeAPIAccessPrefix) {
+				pod.Spec.Containers[i].VolumeMounts = append(pod.Spec.Containers[i].VolumeMounts[:j], pod.Spec.Containers[i].VolumeMounts[j+1:]...)
+			}
+		}
+	}
+}
+
+func addKubeAccessVolume(pod *corev1.Pod, hostSecretName string) {
+	var tokenVolumeName = k3kcontroller.SafeConcatNameWithPrefix(kubeAPIAccessPrefix)
+	pod.Spec.Volumes = append(pod.Spec.Volumes, corev1.Volume{
+		Name: tokenVolumeName,
+		VolumeSource: corev1.VolumeSource{
+			Secret: &corev1.SecretVolumeSource{
+				SecretName: hostSecretName,
+			},
+		},
+	})
+	for i := range pod.Spec.Containers {
+		pod.Spec.Containers[i].VolumeMounts = append(pod.Spec.Containers[i].VolumeMounts, corev1.VolumeMount{
+			Name:      tokenVolumeName,
+			MountPath: serviceAccountTokenMountPath,
+		})
+	}
+}

pkg/controller/cluster/agent/shared.go

@@ -35,7 +35,7 @@ func NewSharedAgent(cluster *v1alpha1.Cluster, serviceIP, sharedAgentImage, toke
 }
 
 func (s *SharedAgent) Config() (ctrlruntimeclient.Object, error) {
-	config := sharedAgentData(s.cluster, s.token, s.Name())
+	config := sharedAgentData(s.cluster, s.token, s.Name(), s.serviceIP)
 
 	return &v1.Secret{
 		TypeMeta: metav1.TypeMeta{
@@ -52,13 +52,14 @@ func (s *SharedAgent) Config() (ctrlruntimeclient.Object, error) {
 	}, nil
 }
 
-func sharedAgentData(cluster *v1alpha1.Cluster, token, nodeName string) string {
+func sharedAgentData(cluster *v1alpha1.Cluster, token, nodeName, ip string) string {
 	return fmt.Sprintf(`clusterName: %s
 clusterNamespace: %s
 nodeName: %s
 agentHostname: %s
+agentIP: %s
 token: %s`,
-		cluster.Name, cluster.Namespace, nodeName, nodeName, token)
+		cluster.Name, cluster.Namespace, nodeName, nodeName, ip, token)
 }
 
 func (s *SharedAgent) Resources() []ctrlruntimeclient.Object {