diff --git a/k3k-kubelet/config.go b/k3k-kubelet/config.go index e820d581..63db2c20 100644 --- a/k3k-kubelet/config.go +++ b/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 } diff --git a/k3k-kubelet/kubelet.go b/k3k-kubelet/kubelet.go index df79c11e..1492c577 100644 --- a/k3k-kubelet/kubelet.go +++ b/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 } } diff --git a/k3k-kubelet/main.go b/k3k-kubelet/main.go index 6b534b52..d4645d50 100644 --- a/k3k-kubelet/main.go +++ b/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)) } diff --git a/k3k-kubelet/provider/configure.go b/k3k-kubelet/provider/configure.go index 45d93a4d..67e19c09 100644 --- a/k3k-kubelet/provider/configure.go +++ b/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 diff --git a/k3k-kubelet/provider/provider.go b/k3k-kubelet/provider/provider.go index 232cebd8..700dd26b 100644 --- a/k3k-kubelet/provider/provider.go +++ b/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 } diff --git a/k3k-kubelet/provider/token.go b/k3k-kubelet/provider/token.go new file mode 100644 index 00000000..4379b89d --- /dev/null +++ b/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, + }) + } +} diff --git a/pkg/controller/cluster/agent/shared.go b/pkg/controller/cluster/agent/shared.go index 1ab2c6f6..296e1dd3 100644 --- a/pkg/controller/cluster/agent/shared.go +++ b/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 {