diff --git a/k3k-kubelet/config.go b/k3k-kubelet/config.go index 63db2c20..73c40e5f 100644 --- a/k3k-kubelet/config.go +++ b/k3k-kubelet/config.go @@ -17,7 +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"` + ServerIP string `yaml:"serverIP,omitempty"` } func (c *config) unmarshalYAML(data []byte) error { @@ -51,8 +51,8 @@ func (c *config) unmarshalYAML(data []byte) error { if c.Token == "" { c.Token = conf.Token } - if c.AgentIP == "" { - c.AgentIP = conf.AgentIP + if c.ServerIP == "" { + c.ServerIP = conf.ServerIP } return nil } diff --git a/k3k-kubelet/controller/service.go b/k3k-kubelet/controller/service.go new file mode 100644 index 00000000..6d2cdd5c --- /dev/null +++ b/k3k-kubelet/controller/service.go @@ -0,0 +1,126 @@ +package controller + +import ( + "context" + + "github.com/rancher/k3k/k3k-kubelet/translate" + "github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1" + "github.com/rancher/k3k/pkg/log" + + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +const ( + serviceSyncerController = "service-syncer-controller" + maxConcurrentReconciles = 1 + serviceFinalizerName = "service.k3k.io/finalizer" +) + +type ServiceReconciler struct { + virtualClient ctrlruntimeclient.Client + hostClient ctrlruntimeclient.Client + clusterName string + clusterNamespace string + Scheme *runtime.Scheme + HostScheme *runtime.Scheme + logger *log.Logger + Translater translate.ToHostTranslater +} + +// AddServiceSyncer adds service syncer controller to the manager of the virtual cluster +func AddServiceSyncer(ctx context.Context, virtMgr, hostMgr manager.Manager, clusterName, clusterNamespace string, logger *log.Logger) error { + translater := translate.ToHostTranslater{ + ClusterName: clusterName, + ClusterNamespace: clusterNamespace, + } + // initialize a new Reconciler + reconciler := ServiceReconciler{ + virtualClient: virtMgr.GetClient(), + hostClient: hostMgr.GetClient(), + Scheme: virtMgr.GetScheme(), + HostScheme: hostMgr.GetScheme(), + logger: logger.Named(serviceSyncerController), + Translater: translater, + clusterName: clusterName, + clusterNamespace: clusterNamespace, + } + return ctrl.NewControllerManagedBy(virtMgr). + For(&v1.Service{}). + WithOptions(controller.Options{ + MaxConcurrentReconciles: maxConcurrentReconciles, + }). + Complete(&reconciler) +} + +func (s *ServiceReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { + log := s.logger.With("Cluster", s.clusterName, "Service", req.NamespacedName) + if req.Name == "kubernetes" || req.Name == "kube-dns" { + return reconcile.Result{}, nil + } + var ( + virtService v1.Service + hostService v1.Service + cluster v1alpha1.Cluster + ) + // getting the cluster for setting the controller reference + if err := s.hostClient.Get(ctx, types.NamespacedName{Name: s.clusterName, Namespace: s.clusterNamespace}, &cluster); err != nil { + return reconcile.Result{}, err + } + if err := s.virtualClient.Get(ctx, req.NamespacedName, &virtService); err != nil { + return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err) + } + syncedService := s.service(&virtService) + if err := controllerutil.SetControllerReference(&cluster, syncedService, s.HostScheme); err != nil { + return reconcile.Result{}, err + } + + // handle deletion + if !virtService.DeletionTimestamp.IsZero() { + // deleting the synced service if exists + if err := s.hostClient.Delete(ctx, syncedService); err != nil { + return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err) + } + // remove the finalizer after cleaning up the synced service + if controllerutil.ContainsFinalizer(&virtService, serviceFinalizerName) { + controllerutil.RemoveFinalizer(&virtService, serviceFinalizerName) + if err := s.virtualClient.Update(ctx, &virtService); err != nil { + return reconcile.Result{}, err + } + } + return reconcile.Result{}, nil + } + + // Add finalizer if it does not exist + if !controllerutil.ContainsFinalizer(&virtService, serviceFinalizerName) { + controllerutil.AddFinalizer(&virtService, serviceFinalizerName) + if err := s.virtualClient.Update(ctx, &virtService); err != nil { + return reconcile.Result{}, err + } + } + // create or update the service on host + if err := s.hostClient.Get(ctx, types.NamespacedName{Name: syncedService.Name, Namespace: s.clusterNamespace}, &hostService); err != nil { + if apierrors.IsNotFound(err) { + log.Info("creating the service for the first time on the host cluster") + return reconcile.Result{}, s.hostClient.Create(ctx, syncedService) + } + return reconcile.Result{}, err + } + log.Info("updating service on the host cluster") + return reconcile.Result{}, s.hostClient.Update(ctx, syncedService) +} + +func (s *ServiceReconciler) service(obj *v1.Service) *v1.Service { + hostService := obj.DeepCopy() + s.Translater.TranslateTo(hostService) + // don't sync finalizers to the host + return hostService +} diff --git a/k3k-kubelet/kubelet.go b/k3k-kubelet/kubelet.go index 1492c577..f3dbac2d 100644 --- a/k3k-kubelet/kubelet.go +++ b/k3k-kubelet/kubelet.go @@ -4,11 +4,14 @@ import ( "context" "crypto/tls" "crypto/x509" + "errors" "fmt" + "net" "net/http" "time" certutil "github.com/rancher/dynamiclistener/cert" + k3kkubeletcontroller "github.com/rancher/k3k/k3k-kubelet/controller" "github.com/rancher/k3k/k3k-kubelet/provider" "github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1" "github.com/rancher/k3k/pkg/controller" @@ -20,6 +23,7 @@ import ( "github.com/virtual-kubelet/virtual-kubelet/node" "github.com/virtual-kubelet/virtual-kubelet/node/nodeutil" "go.uber.org/zap" + v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "k8s.io/apiserver/pkg/authentication/user" @@ -50,6 +54,9 @@ type kubelet struct { name string port int hostConfig *rest.Config + virtConfig *rest.Config + agentIP string + dnsIP string hostClient ctrlruntimeclient.Client virtClient kubernetes.Interface hostMgr manager.Manager @@ -93,14 +100,14 @@ func newKubelet(ctx context.Context, c *config, logger *k3klog.Logger) (*kubelet }, }) if err != nil { - return nil, fmt.Errorf("unable to create controller-runtime mgr for host cluster: %s", err.Error()) + return nil, errors.New("unable to create controller-runtime mgr for host cluster: " + err.Error()) } virtualScheme := runtime.NewScheme() // virtual client will only use core types (for now), no need to add anything other than the basics err = clientgoscheme.AddToScheme(virtualScheme) if err != nil { - return nil, fmt.Errorf("unable to add client go types to virtual cluster scheme: %s", err.Error()) + return nil, errors.New("unable to add client go types to virtual cluster scheme: " + err.Error()) } virtualMgr, err := ctrl.NewManager(virtConfig, manager.Options{ Scheme: virtualScheme, @@ -108,30 +115,59 @@ func newKubelet(ctx context.Context, c *config, logger *k3klog.Logger) (*kubelet BindAddress: ":8084", }, }) - if err != nil { - return nil, err + return nil, errors.New("unable to create controller-runtime mgr for virtual cluster: " + err.Error()) } + + logger.Info("adding service syncer controller") + if err := k3kkubeletcontroller.AddServiceSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace, k3klog.New(false)); err != nil { + return nil, errors.New("failed to add service syncer controller: " + err.Error()) + } + + clusterIP, err := clusterIP(ctx, c.AgentHostname, c.ClusterNamespace, hostClient) + if err != nil { + return nil, errors.New("failed to extract the clusterIP for the server service: " + err.Error()) + } + + // get the cluster's DNS IP to be injected to pods + var dnsService v1.Service + dnsName := controller.SafeConcatNameWithPrefix(c.ClusterName, "kube-dns") + if err := hostClient.Get(ctx, types.NamespacedName{Name: dnsName, Namespace: c.ClusterNamespace}, &dnsService); err != nil { + return nil, errors.New("failed to get the DNS service for the cluster: " + err.Error()) + } + return &kubelet{ name: c.NodeName, hostConfig: hostConfig, hostClient: hostClient, + virtConfig: virtConfig, + virtClient: virtClient, hostMgr: hostMgr, virtualMgr: virtualMgr, - virtClient: virtClient, + agentIP: clusterIP, logger: logger.Named(k3kKubeletName), token: c.Token, + dnsIP: dnsService.Spec.ClusterIP, }, nil } -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) +func clusterIP(ctx context.Context, serviceName, clusterNamespace string, hostClient ctrlruntimeclient.Client) (string, error) { + var service v1.Service + serviceKey := types.NamespacedName{Namespace: clusterNamespace, Name: serviceName} + if err := hostClient.Get(ctx, serviceKey, &service); err != nil { + return "", err + } + return service.Spec.ClusterIP, nil +} + +func (k *kubelet) registerNode(ctx context.Context, agentIP, srvPort, namespace, name, hostname, serverIP, dnsIP string) error { + providerFunc := k.newProviderFunc(namespace, name, hostname, agentIP, serverIP, dnsIP) + nodeOpts := k.nodeOpts(ctx, srvPort, namespace, name, hostname, agentIP) var err error k.node, err = nodeutil.NewNode(k.name, providerFunc, nodeutil.WithClient(k.virtClient), nodeOpts) if err != nil { - return fmt.Errorf("unable to start kubelet: %v", err) + return errors.New("unable to start kubelet: " + err.Error()) } return nil } @@ -172,32 +208,32 @@ func (k *kubelet) start(ctx context.Context) { k.logger.Info("node exited successfully") } -func (k *kubelet) newProviderFunc(namespace, name, hostname, ip string) nodeutil.NewProviderFunc { +func (k *kubelet) newProviderFunc(namespace, name, hostname, agentIP, serverIP, dnsIP 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) + utilProvider, err := provider.New(*k.hostConfig, k.hostMgr, k.virtualMgr, k.logger, namespace, name, serverIP, dnsIP) if err != nil { - return nil, nil, fmt.Errorf("unable to make nodeutil provider %w", err) + return nil, nil, errors.New("unable to make nodeutil provider: " + err.Error()) } nodeProvider := provider.Node{} - provider.ConfigureNode(pc.Node, hostname, k.port, ip) + provider.ConfigureNode(pc.Node, hostname, k.port, agentIP) return utilProvider, &nodeProvider, nil } } -func (k *kubelet) nodeOpts(ctx context.Context, srvPort, namespace, name, hostname string) nodeutil.NodeOpt { +func (k *kubelet) nodeOpts(ctx context.Context, srvPort, namespace, name, hostname, agentIP string) nodeutil.NodeOpt { return func(c *nodeutil.NodeConfig) error { c.HTTPListenAddr = fmt.Sprintf(":%s", srvPort) // set up the routes mux := http.NewServeMux() if err := nodeutil.AttachProviderRoutes(mux)(c); err != nil { - return fmt.Errorf("unable to attach routes: %w", err) + return errors.New("unable to attach routes: " + err.Error()) } c.Handler = mux - tlsConfig, err := loadTLSConfig(ctx, k.hostClient, name, namespace, k.name, hostname, k.token) + tlsConfig, err := loadTLSConfig(ctx, k.hostClient, name, namespace, k.name, hostname, k.token, agentIP) if err != nil { - return fmt.Errorf("unable to get tls config: %w", err) + return errors.New("unable to get tls config: " + err.Error()) } c.TLSConfig = tlsConfig return nil @@ -223,7 +259,7 @@ func virtRestConfig(ctx context.Context, virtualConfigPath string, hostClient ct logger.Infow("decoded bootstrap", zap.Error(err)) return err }); err != nil { - return nil, fmt.Errorf("unable to decode bootstrap: %w", err) + return nil, errors.New("unable to decode bootstrap: " + err.Error()) } adminCert, adminKey, err := kubeconfig.CreateClientCertKey( controller.AdminCommonName, []string{user.SystemPrivilegedGroup}, @@ -265,7 +301,7 @@ func kubeconfigBytes(url string, serverCA, clientCert, clientKey []byte) ([]byte return clientcmd.Write(*config) } -func loadTLSConfig(ctx context.Context, hostClient ctrlruntimeclient.Client, clusterName, clusterNamespace, nodeName, hostname, token string) (*tls.Config, error) { +func loadTLSConfig(ctx context.Context, hostClient ctrlruntimeclient.Client, clusterName, clusterNamespace, nodeName, hostname, token, agentIP string) (*tls.Config, error) { var ( cluster v1alpha1.Cluster b *bootstrap.ControlRuntimeBootstrap @@ -281,26 +317,28 @@ func loadTLSConfig(ctx context.Context, hostClient ctrlruntimeclient.Client, clu b, err = bootstrap.DecodedBootstrap(token, endpoint) return err }); err != nil { - return nil, fmt.Errorf("unable to decode bootstrap: %w", err) + return nil, errors.New("unable to decode bootstrap: " + err.Error()) } + ip := net.ParseIP(agentIP) altNames := certutil.AltNames{ DNSNames: []string{hostname}, + IPs: []net.IP{ip}, } cert, key, err := kubeconfig.CreateClientCertKey(nodeName, nil, &altNames, []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, 0, b.ServerCA.Content, b.ServerCAKey.Content) if err != nil { - return nil, fmt.Errorf("unable to get cert and key: %w", err) + return nil, errors.New("unable to get cert and key: " + err.Error()) } clientCert, err := tls.X509KeyPair(cert, key) if err != nil { - return nil, fmt.Errorf("unable to get key pair: %w", err) + return nil, errors.New("unable to get key pair: " + err.Error()) } // create rootCA CertPool certs, err := certutil.ParseCertsPEM([]byte(b.ServerCA.Content)) if err != nil { - return nil, fmt.Errorf("unable to create ca certs: %w", err) + return nil, errors.New("unable to create ca certs: " + err.Error()) } if len(certs) < 1 { - return nil, fmt.Errorf("ca cert is not parsed correctly") + return nil, errors.New("ca cert is not parsed correctly") } pool := x509.NewCertPool() pool.AddCert(certs[0]) diff --git a/k3k-kubelet/main.go b/k3k-kubelet/main.go index d4645d50..3cd153d7 100644 --- a/k3k-kubelet/main.go +++ b/k3k-kubelet/main.go @@ -59,7 +59,7 @@ func main() { Usage: "kubelet API port number", Destination: &cfg.KubeletPort, EnvVar: "SERVER_PORT", - Value: "9443", + Value: "10250", }, cli.StringFlag{ Name: "agent-hostname", @@ -68,10 +68,10 @@ func main() { 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", + Name: "server-ip", + Usage: "Server IP used for registering the virtual kubelet to the cluster", + Destination: &cfg.ServerIP, + EnvVar: "SERVER_IP", }, cli.StringFlag{ Name: "config", @@ -112,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.AgentIP, cfg.KubeletPort, cfg.ClusterNamespace, cfg.ClusterName, cfg.AgentHostname); err != nil { + if err := k.registerNode(ctx, k.agentIP, cfg.KubeletPort, cfg.ClusterNamespace, cfg.ClusterName, cfg.AgentHostname, cfg.ServerIP, k.dnsIP); err != nil { logger.Fatalw("failed to register new node", zap.Error(err)) } diff --git a/k3k-kubelet/provider/provider.go b/k3k-kubelet/provider/provider.go index 700dd26b..73ec7e29 100644 --- a/k3k-kubelet/provider/provider.go +++ b/k3k-kubelet/provider/provider.go @@ -45,17 +45,19 @@ type Provider struct { MetricsClient metricset.Interface ClusterNamespace string ClusterName string + serverIP string + dnsIP string logger *k3klog.Logger } -func New(hostConfig rest.Config, hostMgr, virtualMgr manager.Manager, logger *k3klog.Logger, Namespace, Name string) (*Provider, error) { +func New(hostConfig rest.Config, hostMgr, virtualMgr manager.Manager, logger *k3klog.Logger, namespace, name, serverIP, dnsIP string) (*Provider, error) { coreClient, err := cv1.NewForConfig(&hostConfig) if err != nil { return nil, err } translater := translate.ToHostTranslater{ - ClusterName: Name, - ClusterNamespace: Namespace, + ClusterName: name, + ClusterNamespace: namespace, } p := Provider{ Handler: controller.ControllerHandler{ @@ -71,9 +73,11 @@ func New(hostConfig rest.Config, hostMgr, virtualMgr manager.Manager, logger *k3 Translater: translater, ClientConfig: hostConfig, CoreClient: coreClient, - ClusterNamespace: Namespace, - ClusterName: Name, + ClusterNamespace: namespace, + ClusterName: name, logger: logger, + serverIP: serverIP, + dnsIP: dnsIP, } return &p, nil @@ -246,8 +250,11 @@ func (p *Provider) CreatePod(ctx context.Context, pod *corev1.Pod) error { 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) } + // inject networking information to the pod including the virtual cluster controlplane endpoint + p.configureNetworking(pod.Name, pod.Namespace, tPod) + p.logger.Infow("Creating pod", "Host Namespace", tPod.Namespace, "Host Name", tPod.Name, - "Virtual Namespace", pod.Namespace, "Virtual Name", pod.Name) + "Virtual Namespace", pod.Namespace, "Virtual Name", "env", pod.Name, pod.Spec.Containers[0].Env) return p.HostClient.Create(ctx, tPod) } @@ -444,7 +451,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.Infow("got a request for get pod", "Namespace", namespace, "Name", name) + p.logger.Debugw("got a request for get pod", "Namespace", namespace, "Name", name) hostNamespaceName := types.NamespacedName{ Namespace: p.ClusterNamespace, Name: p.Translater.TranslateName(namespace, name), @@ -463,7 +470,7 @@ 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.Infow("got a request for pod status", "Namespace", namespace, "Name", name) + p.logger.Debugw("got a request for pod status", "Namespace", namespace, "Name", name) pod, err := p.GetPod(ctx, namespace, name) if err != nil { return nil, fmt.Errorf("unable to get pod for status: %w", err) @@ -496,6 +503,47 @@ func (p *Provider) GetPods(ctx context.Context) ([]*corev1.Pod, error) { return retPods, nil } +func (p *Provider) configureNetworking(podName, podNamespace string, pod *corev1.Pod) { + // inject networking information to the pod's environment variables + for i := range pod.Spec.Containers { + pod.Spec.Containers[i].Env = append(pod.Spec.Containers[i].Env, + corev1.EnvVar{ + Name: "KUBERNETES_PORT_443_TCP", + Value: "tcp://" + p.serverIP + ":6443", + }, + corev1.EnvVar{ + Name: "KUBERNETES_PORT", + Value: "tcp://" + p.serverIP + ":6443", + }, + corev1.EnvVar{ + Name: "KUBERNETES_PORT_443_TCP_ADDR", + Value: p.serverIP, + }, + corev1.EnvVar{ + Name: "KUBERNETES_SERVICE_HOST", + Value: p.serverIP, + }, + corev1.EnvVar{ + Name: "KUBERNETES_SERVICE_PORT", + Value: "6443", + }, + ) + } + // injecting cluster DNS IP to the pods except for coredns pod + if !strings.HasPrefix(podName, "coredns") { + pod.Spec.DNSPolicy = corev1.DNSNone + pod.Spec.DNSConfig = &corev1.PodDNSConfig{ + Nameservers: []string{ + p.dnsIP, + }, + Searches: []string{ + podNamespace + ".svc.cluster.local", "svc.cluster.local", "cluster.local", + }, + } + } + +} + // getSecretsAndConfigmaps retrieves a list of all secrets/configmaps that are in use by a given pod. Useful // for removing/seeing which virtual cluster resources need to be in the host cluster. func getSecretsAndConfigmaps(pod *corev1.Pod) ([]string, []string) { diff --git a/k3k-kubelet/translate/host.go b/k3k-kubelet/translate/host.go index 64053968..4a76f4e8 100644 --- a/k3k-kubelet/translate/host.go +++ b/k3k-kubelet/translate/host.go @@ -60,6 +60,7 @@ func (t *ToHostTranslater) TranslateTo(obj client.Object) { // and doesn't collide with other resources obj.SetName(t.TranslateName(obj.GetNamespace(), obj.GetName())) obj.SetNamespace(t.ClusterNamespace) + obj.SetFinalizers(nil) } func (t *ToHostTranslater) TranslateFrom(obj client.Object) { diff --git a/pkg/controller/cluster/agent/shared.go b/pkg/controller/cluster/agent/shared.go index 296e1dd3..274c6d7b 100644 --- a/pkg/controller/cluster/agent/shared.go +++ b/pkg/controller/cluster/agent/shared.go @@ -3,12 +3,14 @@ package agent import ( "fmt" + "github.com/rancher/k3k/k3k-kubelet/translate" "github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1" "github.com/rancher/k3k/pkg/controller" apps "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -57,13 +59,13 @@ func sharedAgentData(cluster *v1alpha1.Cluster, token, nodeName, ip string) stri clusterNamespace: %s nodeName: %s agentHostname: %s -agentIP: %s +serverIP: %s token: %s`, cluster.Name, cluster.Namespace, nodeName, nodeName, ip, token) } func (s *SharedAgent) Resources() []ctrlruntimeclient.Object { - return []ctrlruntimeclient.Object{s.serviceAccount(), s.role(), s.roleBinding(), s.service(), s.deployment()} + return []ctrlruntimeclient.Object{s.serviceAccount(), s.role(), s.roleBinding(), s.service(), s.deployment(), s.dnsService()} } func (s *SharedAgent) deployment() *apps.Deployment { @@ -169,7 +171,47 @@ func (s *SharedAgent) service() *v1.Service { { Name: "k3s-kubelet-port", Protocol: v1.ProtocolTCP, - Port: 9443, + Port: 10250, + }, + }, + }, + } +} + +func (s *SharedAgent) dnsService() *v1.Service { + return &v1.Service{ + TypeMeta: metav1.TypeMeta{ + Kind: "Service", + APIVersion: "v1", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: s.DNSName(), + Namespace: s.cluster.Namespace, + }, + Spec: v1.ServiceSpec{ + Type: v1.ServiceTypeClusterIP, + Selector: map[string]string{ + translate.ClusterNameLabel: s.cluster.Name, + "k8s-app": "kube-dns", + }, + Ports: []v1.ServicePort{ + { + Name: "dns", + Protocol: v1.ProtocolUDP, + Port: 53, + TargetPort: intstr.FromInt32(53), + }, + { + Name: "dns-tcp", + Protocol: v1.ProtocolTCP, + Port: 53, + TargetPort: intstr.FromInt32(53), + }, + { + Name: "metrics", + Protocol: v1.ProtocolTCP, + Port: 9153, + TargetPort: intstr.FromInt32(9153), }, }, }, @@ -203,12 +245,7 @@ func (s *SharedAgent) role() *rbacv1.Role { { Verbs: []string{"*"}, APIGroups: []string{""}, - Resources: []string{"pods", "secrets", "configmaps"}, - }, - { - Verbs: []string{"get", "watch", "list"}, - APIGroups: []string{""}, - Resources: []string{"services"}, + Resources: []string{"pods", "secrets", "configmaps", "services"}, }, { Verbs: []string{"get", "watch", "list"}, @@ -247,3 +284,7 @@ func (s *SharedAgent) roleBinding() *rbacv1.RoleBinding { func (s *SharedAgent) Name() string { return controller.SafeConcatNameWithPrefix(s.cluster.Name, sharedNodeAgentName) } + +func (s *SharedAgent) DNSName() string { + return controller.SafeConcatNameWithPrefix(s.cluster.Name, "kube-dns") +} diff --git a/pkg/controller/cluster/cluster.go b/pkg/controller/cluster/cluster.go index 62e5627e..6c6f1e30 100644 --- a/pkg/controller/cluster/cluster.go +++ b/pkg/controller/cluster/cluster.go @@ -121,7 +121,7 @@ func (c *ClusterReconciler) createCluster(ctx context.Context, cluster *v1alpha1 return err } - s := server.New(cluster, c.Client, token) + s := server.New(cluster, c.Client, token, string(cluster.Spec.Mode)) if cluster.Spec.Persistence != nil { cluster.Status.Persistence = cluster.Spec.Persistence diff --git a/pkg/controller/cluster/server/config.go b/pkg/controller/cluster/server/config.go index 66339cc5..bf230010 100644 --- a/pkg/controller/cluster/server/config.go +++ b/pkg/controller/cluster/server/config.go @@ -68,7 +68,7 @@ func serverOptions(cluster *v1alpha1.Cluster, token string) string { } } if cluster.Spec.Mode != agent.VirtualNodeMode { - opts = opts + "disable-agent: true\negress-selector-mode: disabled\n" + opts = opts + "disable-agent: true\negress-selector-mode: disabled\ndisable:\n- servicelb\n- traefik\n- metrics-server" } // TODO: Add extra args to the options diff --git a/pkg/controller/cluster/server/server.go b/pkg/controller/cluster/server/server.go index d878e431..5d30bdc3 100644 --- a/pkg/controller/cluster/server/server.go +++ b/pkg/controller/cluster/server/server.go @@ -6,6 +6,7 @@ import ( "github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1" "github.com/rancher/k3k/pkg/controller" + "github.com/rancher/k3k/pkg/controller/cluster/agent" apps "k8s.io/api/apps/v1" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/resource" @@ -31,14 +32,16 @@ const ( type Server struct { cluster *v1alpha1.Cluster client client.Client + mode string token string } -func New(cluster *v1alpha1.Cluster, client client.Client, token string) *Server { +func New(cluster *v1alpha1.Cluster, client client.Client, token, mode string) *Server { return &Server{ cluster: cluster, client: client, token: token, + mode: mode, } } @@ -130,9 +133,6 @@ func (s *Server) podSpec(image, name string, persistent bool, affinitySelector * }, }, }, - SecurityContext: &v1.SecurityContext{ - Privileged: ptr.To(true), - }, Command: []string{ "/bin/sh", "-c", @@ -219,6 +219,12 @@ func (s *Server) podSpec(image, name string, persistent bool, affinitySelector * }, } + // start the pod unprivileged in shared mode + if s.mode == agent.VirtualNodeMode { + podSpec.Containers[0].SecurityContext = &v1.SecurityContext{ + Privileged: ptr.To(true), + } + } return podSpec }