k3k/k3k-kubelet/kubelet.go

package main

import (
	"context"
	"fmt"
	"path"
	"time"

	"github.com/go-logr/logr"
	"github.com/virtual-kubelet/virtual-kubelet/log"
	"github.com/virtual-kubelet/virtual-kubelet/log/klogv2"
	"github.com/virtual-kubelet/virtual-kubelet/node"
	"github.com/virtual-kubelet/virtual-kubelet/node/nodeutil"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/retry"
	"k8s.io/klog/v2"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/manager"

	corev1 "k8s.io/api/core/v1"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
	ctrl "sigs.k8s.io/controller-runtime"
	ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
	ctrlserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"

	"github.com/rancher/k3k/k3k-kubelet/controller/syncer"
	"github.com/rancher/k3k/k3k-kubelet/provider"
	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
	"github.com/rancher/k3k/pkg/controller"
)

var baseScheme = runtime.NewScheme()

func init() {
	_ = clientgoscheme.AddToScheme(baseScheme)
	_ = v1beta1.AddToScheme(baseScheme)
}

type kubelet struct {
	virtualCluster v1beta1.Cluster

	name       string
	port       int
	hostConfig *rest.Config
	virtConfig *rest.Config
	agentIP    string
	dnsIP      string
	hostClient ctrlruntimeclient.Client
	virtClient kubernetes.Interface
	hostMgr    manager.Manager
	virtualMgr manager.Manager
	node       *nodeutil.Node
	logger     logr.Logger
	token      string

	virtEventRecorder record.EventRecorder
	eb                record.EventBroadcaster
}

func newKubelet(ctx context.Context, c *config) (*kubelet, error) {
	hostConfig, err := clientcmd.BuildConfigFromFlags("", c.HostKubeconfig)
	if err != nil {
		return nil, err
	}

	hostClient, err := ctrlruntimeclient.New(hostConfig, ctrlruntimeclient.Options{
		Scheme: baseScheme,
	})
	if err != nil {
		return nil, err
	}

	virtConfig, err := virtRestConfig(ctx, c.VirtKubeconfig, hostClient, c.ClusterName, c.ClusterNamespace)
	if err != nil {
		return nil, err
	}

	virtClient, err := kubernetes.NewForConfig(virtConfig)
	if err != nil {
		return nil, err
	}

	hostMetricsBindAddress := ":8083"
	virtualMetricsBindAddress := ":8084"

	if c.MirrorHostNodes {
		hostMetricsBindAddress = "0"
		virtualMetricsBindAddress = "0"
	}

	hostMgr, err := ctrl.NewManager(hostConfig, manager.Options{
		Scheme:                  baseScheme,
		LeaderElection:          true,
		LeaderElectionNamespace: c.ClusterNamespace,
		LeaderElectionID:        c.ClusterName,
		Metrics: ctrlserver.Options{
			BindAddress: hostMetricsBindAddress,
		},
		Cache: cache.Options{
			DefaultNamespaces: map[string]cache.Config{
				c.ClusterNamespace: {},
			},
		},
	})
	if err != nil {
		return nil, fmt.Errorf("unable to create controller-runtime mgr for host cluster: %w", err)
	}

	// virtual client will only use core types (for now), no need to add anything other than the basics
	virtualScheme := runtime.NewScheme()
	if err := clientgoscheme.AddToScheme(virtualScheme); err != nil {
		return nil, fmt.Errorf("unable to add client go types to virtual cluster scheme: %w", err)
	}

	virtualMgr, err := ctrl.NewManager(virtConfig, manager.Options{
		Scheme:                  virtualScheme,
		LeaderElection:          true,
		LeaderElectionNamespace: "kube-system",
		LeaderElectionID:        c.ClusterName,
		Metrics: ctrlserver.Options{
			BindAddress: virtualMetricsBindAddress,
		},
	})
	if err != nil {
		return nil, fmt.Errorf("unable to create controller-runtime mgr for virtual cluster: %w", err)
	}

	controllerName := c.AgentHostname
	eb := record.NewBroadcaster(record.WithContext(ctx))
	eb.StartRecordingToSink(&corev1client.EventSinkImpl{Interface: virtClient.CoreV1().Events(corev1.NamespaceAll)})

	virtEventRecorder := eb.NewRecorder(virtualScheme, corev1.EventSource{Component: path.Join(controllerName, "pod-controller")})

	if err := addControllers(ctx, hostMgr, virtualMgr, c, hostClient, virtEventRecorder); err != nil {
		return nil, fmt.Errorf("failed to add controller: %w", err)
	}

	clusterIP, err := clusterIP(ctx, c.ServiceName, c.ClusterNamespace, hostClient)
	if err != nil {
		return nil, fmt.Errorf("failed to extract the clusterIP for the server service: %w", err)
	}

	// get the cluster's DNS IP to be injected to pods
	var dnsService corev1.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, fmt.Errorf("failed to get the DNS service for the cluster: %w", err)
	}

	var virtualCluster v1beta1.Cluster
	if err := hostClient.Get(ctx, types.NamespacedName{Name: c.ClusterName, Namespace: c.ClusterNamespace}, &virtualCluster); err != nil {
		return nil, fmt.Errorf("failed to get virtualCluster spec: %w", err)
	}

	return &kubelet{
		virtualCluster:    virtualCluster,
		name:              controllerName,
		hostConfig:        hostConfig,
		hostClient:        hostClient,
		virtConfig:        virtConfig,
		virtClient:        virtClient,
		hostMgr:           hostMgr,
		virtualMgr:        virtualMgr,
		agentIP:           clusterIP,
		logger:            logger,
		token:             c.Token,
		dnsIP:             dnsService.Spec.ClusterIP,
		port:              c.KubeletPort,
		virtEventRecorder: virtEventRecorder,
		eb:                eb,
	}, nil
}

func clusterIP(ctx context.Context, serviceName, clusterNamespace string, hostClient ctrlruntimeclient.Client) (string, error) {
	var service corev1.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) start(ctx context.Context) {
	// any one of the following 3 tasks (host manager, virtual manager, node) crashing will stop the
	// program, and all 3 of them block on start, so we start them here in go-routines
	go func() {
		err := k.hostMgr.Start(ctx)
		if err != nil {
			k.logger.Error(err, "host manager stopped")
		}
	}()

	go func() {
		err := k.virtualMgr.Start(ctx)
		if err != nil {
			k.logger.Error(err, "virtual manager stopped")
		}
	}()

	// run the node async so that we can wait for it to be ready in another call

	go func() {
		klog.SetLogger(k.logger.V(1))

		ctx = log.WithLogger(ctx, klogv2.New(nil))
		if err := k.node.Run(ctx); err != nil {
			k.logger.Error(err, "node errored when running")
		}
	}()

	if err := k.node.WaitReady(context.Background(), time.Minute*1); err != nil {
		k.logger.Error(err, "node was not ready within timeout of 1 minute")
	}

	<-k.node.Done()

	if err := k.node.Err(); err != nil {
		k.logger.Error(err, "node stopped with an error")
	}
	defer k.eb.Shutdown()

	k.logger.Info("node exited successfully")
}

func (k *kubelet) newProviderFunc(cfg config) nodeutil.NewProviderFunc {
	return func(pc nodeutil.ProviderConfig) (nodeutil.Provider, node.NodeProvider, error) {
		utilProvider, err := provider.New(*k.hostConfig, k.hostMgr, k.virtualMgr, k.logger, cfg.ClusterNamespace, cfg.ClusterName, cfg.ServerIP, k.dnsIP, cfg.AgentHostname)
		if err != nil {
			return nil, nil, fmt.Errorf("unable to make nodeutil provider: %w", err)
		}

		err = provider.ConfigureNode(
			k.logger,
			pc.Node,
			cfg.AgentHostname,
			k.port,
			k.agentIP,
			utilProvider.Host.Manager,
			utilProvider.Virtual.Client,
			k.virtualCluster,
			cfg.Version,
			cfg.MirrorHostNodes,
		)

		return utilProvider, nil, err
	}
}

func virtRestConfig(ctx context.Context, virtualConfigPath string, hostClient ctrlruntimeclient.Client, clusterName, clusterNamespace string) (*rest.Config, error) {
	if virtualConfigPath != "" {
		return clientcmd.BuildConfigFromFlags("", virtualConfigPath)
	}

	var clusterKubeConfig corev1.Secret

	kubeconfigSecretName := types.NamespacedName{
		Name:      controller.SafeConcatNameWithPrefix(clusterName, "kubeconfig"),
		Namespace: clusterNamespace,
	}

	if err := retry.OnError(controller.Backoff, func(err error) bool {
		return err != nil
	}, func() error {
		return hostClient.Get(ctx, kubeconfigSecretName, &clusterKubeConfig)
	}); err != nil {
		return nil, fmt.Errorf("unable to decode bootstrap: %w", err)
	}

	restConfig, err := clientcmd.RESTConfigFromKubeConfig(clusterKubeConfig.Data["kubeconfig.yaml"])
	if err != nil {
		return nil, fmt.Errorf("failed to create config from kubeconfig file: %w", err)
	}

	return restConfig, nil
}

func addControllers(ctx context.Context, hostMgr, virtualMgr manager.Manager, c *config, hostClient ctrlruntimeclient.Client, virtEventRecorder record.EventRecorder) error {
	var cluster v1beta1.Cluster

	objKey := types.NamespacedName{
		Namespace: c.ClusterNamespace,
		Name:      c.ClusterName,
	}

	if err := hostClient.Get(ctx, objKey, &cluster); err != nil {
		return err
	}

	if err := syncer.AddConfigMapSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
		return fmt.Errorf("failed to add configmap global syncer: %w", err)
	}

	if err := syncer.AddSecretSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
		return fmt.Errorf("failed to add secret global syncer: %w", err)
	}

	logger.Info("adding service syncer controller")

	if err := syncer.AddServiceSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
		return fmt.Errorf("failed to add service syncer controller: %w", err)
	}

	logger.Info("adding ingress syncer controller")

	if err := syncer.AddIngressSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
		return fmt.Errorf("failed to add ingress syncer controller: %w", err)
	}

	logger.Info("adding pvc syncer controller")

	if err := syncer.AddPVCSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
		return fmt.Errorf("failed to add pvc syncer controller: %w", err)
	}

	logger.Info("adding priorityclass controller")

	if err := syncer.AddPriorityClassSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace); err != nil {
		return fmt.Errorf("failed to add priorityclass controller: %w", err)
	}

	if err := syncer.AddEventSyncer(ctx, virtualMgr, hostMgr, c.ClusterName, c.ClusterNamespace, virtEventRecorder); err != nil {
		return fmt.Errorf("failed to add event syncer controller: %w", err)
	}

	return nil
}