diff --git a/k3k-kubelet/controller/configmap.go b/k3k-kubelet/controller/configmap.go new file mode 100644 index 00000000..cffea360 --- /dev/null +++ b/k3k-kubelet/controller/configmap.go @@ -0,0 +1,166 @@ +package controller + +import ( + "context" + "fmt" + "sync" + + "github.com/rancher/k3k/pkg/controller" + k3klog "github.com/rancher/k3k/pkg/log" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +type ConfigMapSyncer struct { + mutex sync.RWMutex + // VirtualClient is the client for the virtual cluster + VirtualClient client.Client + // CoreClient is the client for the host cluster + HostClient client.Client + // TranslateFunc is the function that translates a given resource from it's virtual representation to the host + // representation + TranslateFunc func(*corev1.ConfigMap) (*corev1.ConfigMap, error) + // Logger is the logger that the controller will use + Logger *k3klog.Logger + // objs are the objects that the syncer should watch/syncronize. Should only be manipulated + // through add/remove + objs sets.Set[types.NamespacedName] +} + +// Reconcile implements reconcile.Reconciler and synchronizes the objects in objs to the host cluster +func (c *ConfigMapSyncer) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { + if !c.isWatching(req.NamespacedName) { + // return immediately without re-enqueueing. We aren't watching this resource + return reconcile.Result{}, nil + } + var virtual corev1.ConfigMap + + if err := c.VirtualClient.Get(ctx, req.NamespacedName, &virtual); err != nil { + return reconcile.Result{ + Requeue: true, + }, fmt.Errorf("unable to get configmap %s/%s from virtual cluster: %w", req.Namespace, req.Name, err) + } + translated, err := c.TranslateFunc(&virtual) + if err != nil { + return reconcile.Result{ + Requeue: true, + }, fmt.Errorf("unable to translate configmap %s/%s from virtual cluster: %w", req.Namespace, req.Name, err) + } + translatedKey := types.NamespacedName{ + Namespace: translated.Namespace, + Name: translated.Name, + } + var host corev1.ConfigMap + if err = c.HostClient.Get(ctx, translatedKey, &host); err != nil { + if apierrors.IsNotFound(err) { + err = c.HostClient.Create(ctx, translated) + // for simplicity's sake, we don't check for conflict errors. The existing object will get + // picked up on in the next re-enqueue + return reconcile.Result{ + Requeue: true, + }, fmt.Errorf("unable to create host configmap %s/%s for virtual configmap %s/%s: %w", + translated.Namespace, translated.Name, req.Namespace, req.Name, err) + } + return reconcile.Result{Requeue: true}, fmt.Errorf("unable to get host configmap %s/%s: %w", translated.Namespace, translated.Name, err) + } + // we are going to use the host in order to avoid conflicts on update + host.Data = translated.Data + if host.Labels == nil { + host.Labels = make(map[string]string, len(translated.Labels)) + } + // we don't want to override labels made on the host cluster by other applications + // but we do need to make sure the labels that the kubelet uses to track host cluster values + // are being tracked appropriately + for key, value := range translated.Labels { + host.Labels[key] = value + } + if err = c.HostClient.Update(ctx, &host); err != nil { + return reconcile.Result{ + Requeue: true, + }, fmt.Errorf("unable to update host configmap %s/%s for virtual configmap %s/%s: %w", + translated.Namespace, translated.Name, req.Namespace, req.Name, err) + + } + return reconcile.Result{}, nil +} + +// isWatching is a utility method to determine if a key is in objs without the caller needing +// to handle mutex lock/unlock. +func (c *ConfigMapSyncer) isWatching(key types.NamespacedName) bool { + c.mutex.RLock() + defer c.mutex.RUnlock() + return c.objs.Has(key) +} + +// AddResource adds a given resource to the list of resources that will be synced. Safe to call multiple times for the +// same resource. +func (c *ConfigMapSyncer) AddResource(ctx context.Context, namespace, name string) error { + objKey := types.NamespacedName{ + Namespace: namespace, + Name: name, + } + // if we already sync this object, no need to writelock/add it + if c.isWatching(objKey) { + return nil + } + // lock in write mode since we are now adding the key + c.mutex.Lock() + if c.objs == nil { + c.objs = sets.Set[types.NamespacedName]{} + } + c.objs = c.objs.Insert(objKey) + c.mutex.Unlock() + _, err := c.Reconcile(ctx, reconcile.Request{ + NamespacedName: objKey, + }) + if err != nil { + return fmt.Errorf("unable to reconcile new object %s/%s: %w", objKey.Namespace, objKey.Name, err) + } + return nil +} + +// RemoveResource removes a given resource from the list of resources that will be synced. Safe to call for an already +// removed resource. +func (c *ConfigMapSyncer) RemoveResource(ctx context.Context, namespace, name string) error { + objKey := types.NamespacedName{ + Namespace: namespace, + Name: name, + } + // if we don't sync this object, no need to writelock/add it + if !c.isWatching(objKey) { + return nil + } + + if err := retry.OnError(controller.Backoff, func(err error) bool { + return err != nil + }, func() error { + return c.removeHostConfigMap(ctx, namespace, name) + }); err != nil { + return fmt.Errorf("unable to remove configmap: %w", err) + } + c.mutex.Lock() + if c.objs == nil { + c.objs = sets.Set[types.NamespacedName]{} + } + c.objs = c.objs.Delete(objKey) + c.mutex.Unlock() + return nil +} + +func (c *ConfigMapSyncer) removeHostConfigMap(ctx context.Context, virtualNamespace, virtualName string) error { + var vConfigMap corev1.ConfigMap + err := c.VirtualClient.Get(ctx, types.NamespacedName{Namespace: virtualNamespace, Name: virtualName}, &vConfigMap) + if err != nil { + return fmt.Errorf("unable to get virtual configmap %s/%s: %w", virtualNamespace, virtualName, err) + } + translated, err := c.TranslateFunc(&vConfigMap) + if err != nil { + return fmt.Errorf("unable to translate virtual secret: %s/%s: %w", virtualNamespace, virtualName, err) + } + return c.HostClient.Delete(ctx, translated) +} diff --git a/k3k-kubelet/controller/handler.go b/k3k-kubelet/controller/handler.go new file mode 100644 index 00000000..860b69b5 --- /dev/null +++ b/k3k-kubelet/controller/handler.go @@ -0,0 +1,115 @@ +package controller + +import ( + "context" + "fmt" + "sync" + + "github.com/rancher/k3k/k3k-kubelet/translate" + k3klog "github.com/rancher/k3k/pkg/log" + v1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +type ControllerHandler struct { + sync.RWMutex + // Mgr is the manager used to run new controllers - from the virtual cluster + Mgr manager.Manager + // Scheme is the scheme used to run new controllers - from the virtual cluster + Scheme runtime.Scheme + // HostClient is the client used to communicate with the host cluster + HostClient client.Client + // VirtualClient is the client used to communicate with the virtual cluster + VirtualClient client.Client + // Translater is the translater that will be used to adjust objects before they + // are made on the host cluster + Translater translate.ToHostTranslater + // Logger is the logger that the controller will use to log errors + Logger *k3klog.Logger + // controllers are the controllers which are currently running + controllers map[schema.GroupVersionKind]updateableReconciler +} + +// updateableReconciler is a reconciler that only syncs specific resources (by name/namespace). This list can +// be altered through the Add and Remove methods +type updateableReconciler interface { + reconcile.Reconciler + AddResource(ctx context.Context, namespace string, name string) error + RemoveResource(ctx context.Context, namespace string, name string) error +} + +func (c *ControllerHandler) AddResource(ctx context.Context, obj client.Object) error { + c.RLock() + controllers := c.controllers + if controllers != nil { + if r, ok := c.controllers[obj.GetObjectKind().GroupVersionKind()]; ok { + err := r.AddResource(ctx, obj.GetNamespace(), obj.GetName()) + c.RUnlock() + return err + } + } + // we need to manually lock/unlock since we intned on write locking to add a new controller + c.RUnlock() + var r updateableReconciler + switch obj.(type) { + case *v1.Secret: + r = &SecretSyncer{ + HostClient: c.HostClient, + VirtualClient: c.VirtualClient, + // TODO: Need actual function + TranslateFunc: func(s *v1.Secret) (*v1.Secret, error) { + // note that this doesn't do any type safety - fix this + // when generics work + c.Translater.TranslateTo(s) + return s, nil + }, + Logger: c.Logger, + } + case *v1.ConfigMap: + r = &ConfigMapSyncer{ + HostClient: c.HostClient, + VirtualClient: c.VirtualClient, + // TODO: Need actual function + TranslateFunc: func(s *v1.ConfigMap) (*v1.ConfigMap, error) { + c.Translater.TranslateTo(s) + return s, nil + }, + Logger: c.Logger, + } + default: + // TODO: Technically, the configmap/secret syncers are relatively generic, and this + // logic could be used for other types. + return fmt.Errorf("unrecognized type: %T", obj) + + } + err := ctrl.NewControllerManagedBy(c.Mgr). + For(&v1.ConfigMap{}). + Complete(r) + if err != nil { + return fmt.Errorf("unable to start configmap controller: %w", err) + } + c.Lock() + if c.controllers == nil { + c.controllers = map[schema.GroupVersionKind]updateableReconciler{} + } + c.controllers[obj.GetObjectKind().GroupVersionKind()] = r + c.Unlock() + + return r.AddResource(ctx, obj.GetNamespace(), obj.GetName()) +} + +func (c *ControllerHandler) RemoveResource(ctx context.Context, obj client.Object) error { + // since we aren't adding a new controller, we don't need to lock + c.RLock() + ctrl, ok := c.controllers[obj.GetObjectKind().GroupVersionKind()] + c.RUnlock() + if !ok { + return fmt.Errorf("no controller found for gvk" + obj.GetObjectKind().GroupVersionKind().String()) + } + return ctrl.RemoveResource(ctx, obj.GetNamespace(), obj.GetName()) +} diff --git a/k3k-kubelet/controller/secret.go b/k3k-kubelet/controller/secret.go new file mode 100644 index 00000000..e2dab374 --- /dev/null +++ b/k3k-kubelet/controller/secret.go @@ -0,0 +1,170 @@ +package controller + +import ( + "context" + "fmt" + "sync" + + "github.com/rancher/k3k/pkg/controller" + k3klog "github.com/rancher/k3k/pkg/log" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/client-go/util/retry" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +type SecretSyncer struct { + mutex sync.RWMutex + // VirtualClient is the client for the virtual cluster + VirtualClient client.Client + // CoreClient is the client for the host cluster + HostClient client.Client + // TranslateFunc is the function that translates a given resource from it's virtual representation to the host + // representation + TranslateFunc func(*corev1.Secret) (*corev1.Secret, error) + // Logger is the logger that the controller will use + Logger *k3klog.Logger + // objs are the objects that the syncer should watch/syncronize. Should only be manipulated + // through add/remove + objs sets.Set[types.NamespacedName] +} + +// Reconcile implements reconcile.Reconciler and synchronizes the objects in objs to the host cluster +func (s *SecretSyncer) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { + if !s.isWatching(req.NamespacedName) { + // return immediately without re-enqueueing. We aren't watching this resource + return reconcile.Result{}, nil + } + var virtual corev1.Secret + + if err := s.VirtualClient.Get(ctx, req.NamespacedName, &virtual); err != nil { + return reconcile.Result{ + Requeue: true, + }, fmt.Errorf("unable to get secret %s/%s from virtual cluster: %w", req.Namespace, req.Name, err) + } + translated, err := s.TranslateFunc(&virtual) + if err != nil { + return reconcile.Result{ + Requeue: true, + }, fmt.Errorf("unable to translate secret %s/%s from virtual cluster: %w", req.Namespace, req.Name, err) + } + translatedKey := types.NamespacedName{ + Namespace: translated.Namespace, + Name: translated.Name, + } + var host corev1.Secret + if err = s.HostClient.Get(ctx, translatedKey, &host); err != nil { + if apierrors.IsNotFound(err) { + err = s.HostClient.Create(ctx, translated) + // for simplicity's sake, we don't check for conflict errors. The existing object will get + // picked up on in the next re-enqueue + return reconcile.Result{ + Requeue: true, + }, fmt.Errorf("unable to create host secret %s/%s for virtual secret %s/%s: %w", + translated.Namespace, translated.Name, req.Namespace, req.Name, err) + } + return reconcile.Result{Requeue: true}, fmt.Errorf("unable to get host secret %s/%s: %w", translated.Namespace, translated.Name, err) + } + // we are going to use the host in order to avoid conflicts on update + host.Data = translated.Data + if host.Labels == nil { + host.Labels = make(map[string]string, len(translated.Labels)) + } + // we don't want to override labels made on the host cluster by other applications + // but we do need to make sure the labels that the kubelet uses to track host cluster values + // are being tracked appropriately + for key, value := range translated.Labels { + host.Labels[key] = value + } + if err = s.HostClient.Update(ctx, &host); err != nil { + return reconcile.Result{ + Requeue: true, + }, fmt.Errorf("unable to update host secret %s/%s for virtual secret %s/%s: %w", + translated.Namespace, translated.Name, req.Namespace, req.Name, err) + + } + return reconcile.Result{}, nil +} + +// isWatching is a utility method to determine if a key is in objs without the caller needing +// to handle mutex lock/unlock. +func (s *SecretSyncer) isWatching(key types.NamespacedName) bool { + s.mutex.RLock() + defer s.mutex.RUnlock() + return s.objs.Has(key) +} + +// AddResource adds a given resource to the list of resources that will be synced. Safe to call multiple times for the +// same resource. +func (s *SecretSyncer) AddResource(ctx context.Context, namespace, name string) error { + objKey := types.NamespacedName{ + Namespace: namespace, + Name: name, + } + // if we already sync this object, no need to writelock/add it + if s.isWatching(objKey) { + return nil + } + // lock in write mode since we are now adding the key + s.mutex.Lock() + if s.objs == nil { + s.objs = sets.Set[types.NamespacedName]{} + } + s.objs = s.objs.Insert(objKey) + s.mutex.Unlock() + _, err := s.Reconcile(ctx, reconcile.Request{ + NamespacedName: objKey, + }) + if err != nil { + return fmt.Errorf("unable to reconcile new object %s/%s: %w", objKey.Namespace, objKey.Name, err) + } + return nil +} + +// RemoveResource removes a given resource from the list of resources that will be synced. Safe to call for an already +// removed resource. +func (s *SecretSyncer) RemoveResource(ctx context.Context, namespace, name string) error { + objKey := types.NamespacedName{ + Namespace: namespace, + Name: name, + } + // if we don't sync this object, no need to writelock/add it + if !s.isWatching(objKey) { + return nil + } + // lock in write mode since we are now adding the key + if err := retry.OnError(controller.Backoff, func(err error) bool { + return err != nil + }, func() error { + return s.removeHostSecret(ctx, namespace, name) + }); err != nil { + return fmt.Errorf("unable to remove secret: %w", err) + } + + s.mutex.Lock() + if s.objs == nil { + s.objs = sets.Set[types.NamespacedName]{} + } + s.objs = s.objs.Delete(objKey) + s.mutex.Unlock() + return nil +} + +func (s *SecretSyncer) removeHostSecret(ctx context.Context, virtualNamespace, virtualName string) error { + var vSecret corev1.Secret + err := s.VirtualClient.Get(ctx, types.NamespacedName{ + Namespace: virtualNamespace, + Name: virtualName, + }, &vSecret) + if err != nil { + return fmt.Errorf("unable to get virtual secret %s/%s: %w", virtualNamespace, virtualName, err) + } + translated, err := s.TranslateFunc(&vSecret) + if err != nil { + return fmt.Errorf("unable to translate virtual secret: %s/%s: %w", virtualNamespace, virtualName, err) + } + return s.HostClient.Delete(ctx, translated) +} diff --git a/k3k-kubelet/kubelet.go b/k3k-kubelet/kubelet.go index f6b05d0e..06e0a0a9 100644 --- a/k3k-kubelet/kubelet.go +++ b/k3k-kubelet/kubelet.go @@ -29,17 +29,21 @@ import ( "k8s.io/client-go/tools/clientcmd" clientcmdapi "k8s.io/client-go/tools/clientcmd/api" "k8s.io/client-go/util/retry" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/cache" ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/manager" + ctrlserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" ) var ( - scheme = runtime.NewScheme() + baseScheme = runtime.NewScheme() k3kKubeletName = "k3k-kubelet" ) func init() { - _ = clientgoscheme.AddToScheme(scheme) - _ = v1alpha1.AddToScheme(scheme) + _ = clientgoscheme.AddToScheme(baseScheme) + _ = v1alpha1.AddToScheme(baseScheme) } type kubelet struct { @@ -48,6 +52,8 @@ type kubelet struct { hostConfig *rest.Config hostClient ctrlruntimeclient.Client virtClient kubernetes.Interface + hostMgr manager.Manager + virtualMgr manager.Manager node *nodeutil.Node logger *k3klog.Logger } @@ -59,7 +65,7 @@ func newKubelet(ctx context.Context, c *config, logger *k3klog.Logger) (*kubelet } hostClient, err := ctrlruntimeclient.New(hostConfig, ctrlruntimeclient.Options{ - Scheme: scheme, + Scheme: baseScheme, }) if err != nil { return nil, err @@ -71,6 +77,38 @@ func newKubelet(ctx context.Context, c *config, logger *k3klog.Logger) (*kubelet } virtClient, err := kubernetes.NewForConfig(virtConfig) + if err != nil { + return nil, err + } + + hostMgr, err := ctrl.NewManager(hostConfig, manager.Options{ + Scheme: baseScheme, + Metrics: ctrlserver.Options{ + BindAddress: ":8083", + }, + 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: %s", 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()) + } + virtualMgr, err := ctrl.NewManager(virtConfig, manager.Options{ + Scheme: virtualScheme, + Metrics: ctrlserver.Options{ + BindAddress: ":8084", + }, + }) + if err != nil { return nil, err } @@ -78,6 +116,8 @@ func newKubelet(ctx context.Context, c *config, logger *k3klog.Logger) (*kubelet name: c.NodeName, hostConfig: hostConfig, hostClient: hostClient, + hostMgr: hostMgr, + virtualMgr: virtualMgr, virtClient: virtClient, logger: logger.Named(k3kKubeletName), }, nil @@ -96,12 +136,31 @@ func (k *kubelet) registerNode(ctx context.Context, srvPort, namespace, name, ho } 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.Fatalw("host manager stopped", zap.Error(err)) + } + }() + + go func() { + err := k.virtualMgr.Start(ctx) + if err != nil { + k.logger.Fatalw("virtual manager stopped", zap.Error(err)) + } + }() + + // run the node async so that we can wait for it to be ready in another call + go func() { ctx = log.WithLogger(ctx, k.logger) if err := k.node.Run(ctx); err != nil { k.logger.Fatalw("node errored when running", zap.Error(err)) } }() + if err := k.node.WaitReady(context.Background(), time.Minute*1); err != nil { k.logger.Fatalw("node was not ready within timeout of 1 minute", zap.Error(err)) } @@ -114,7 +173,7 @@ func (k *kubelet) start(ctx context.Context) { func (k *kubelet) newProviderFunc(namespace, name, hostname string) nodeutil.NewProviderFunc { return func(pc nodeutil.ProviderConfig) (nodeutil.Provider, node.NodeProvider, error) { - utilProvider, err := provider.New(*k.hostConfig, namespace, name) + utilProvider, err := provider.New(*k.hostConfig, k.hostMgr, k.virtualMgr, k.logger, namespace, name) if err != nil { return nil, nil, fmt.Errorf("unable to make nodeutil provider %w", err) } diff --git a/k3k-kubelet/provider/provider.go b/k3k-kubelet/provider/provider.go index 3ee22bd8..a339e925 100644 --- a/k3k-kubelet/provider/provider.go +++ b/k3k-kubelet/provider/provider.go @@ -8,12 +8,17 @@ import ( "strconv" dto "github.com/prometheus/client_model/go" - "github.com/rancher/k3k/pkg/controller" + "github.com/rancher/k3k/k3k-kubelet/controller" + "github.com/rancher/k3k/k3k-kubelet/translate" + k3klog "github.com/rancher/k3k/pkg/log" "github.com/virtual-kubelet/virtual-kubelet/node/api" "github.com/virtual-kubelet/virtual-kubelet/node/api/statsv1alpha1" - "go.uber.org/zap" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/selection" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" "k8s.io/client-go/kubernetes/scheme" cv1 "k8s.io/client-go/kubernetes/typed/core/v1" "k8s.io/client-go/rest" @@ -21,40 +26,51 @@ import ( "k8s.io/client-go/tools/remotecommand" "k8s.io/client-go/transport/spdy" metricset "k8s.io/metrics/pkg/client/clientset/versioned" -) - -const ( - clusterNameLabel = "virt.cattle.io/clusterName" - podNameAnnotation = "virt.cattle.io/podName" - podNamespaceAnnotation = "virt.cattle.io/podNamespace" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/manager" ) // Provider implements nodetuil.Provider from virtual Kubelet. // TODO: Implement NotifyPods and the required usage so that this can be an async provider type Provider struct { + Handler controller.ControllerHandler + Translater translate.ToHostTranslater + HostClient client.Client + VirtualClient client.Client ClientConfig rest.Config CoreClient cv1.CoreV1Interface MetricsClient metricset.Interface ClusterNamespace string ClusterName string - logger zap.SugaredLogger + logger *k3klog.Logger } -func New(config rest.Config, Namespace string, Name string) (*Provider, error) { - coreClient, err := cv1.NewForConfig(&config) +func New(hostConfig rest.Config, hostMgr, virtualMgr manager.Manager, logger *k3klog.Logger, Namespace, Name string) (*Provider, error) { + coreClient, err := cv1.NewForConfig(&hostConfig) if err != nil { return nil, err } - logger, err := zap.NewProduction() - if err != nil { - return nil, err + translater := translate.ToHostTranslater{ + ClusterName: Name, + ClusterNamespace: Namespace, } p := Provider{ - ClientConfig: config, + Handler: controller.ControllerHandler{ + Mgr: virtualMgr, + Scheme: *virtualMgr.GetScheme(), + HostClient: hostMgr.GetClient(), + VirtualClient: virtualMgr.GetClient(), + Translater: translater, + Logger: logger, + }, + HostClient: hostMgr.GetClient(), + VirtualClient: virtualMgr.GetClient(), + Translater: translater, + ClientConfig: hostConfig, CoreClient: coreClient, ClusterNamespace: Namespace, ClusterName: Name, - logger: *logger.Sugar(), + logger: logger, } return &p, nil @@ -62,7 +78,7 @@ func New(config rest.Config, Namespace string, Name string) (*Provider, error) { // GetContainerLogs retrieves the logs of a container by name from the provider. func (p *Provider) GetContainerLogs(ctx context.Context, namespace, podName, containerName string, opts api.ContainerLogOpts) (io.ReadCloser, error) { - hostPodName := p.hostName(namespace, podName) + hostPodName := p.Translater.TranslateName(namespace, podName) options := corev1.PodLogOptions{ Container: containerName, Timestamps: opts.Timestamps, @@ -93,9 +109,10 @@ func (p *Provider) GetContainerLogs(ctx context.Context, namespace, podName, con // RunInContainer executes a command in a container in the pod, copying data // between in/out/err and the container's stdin/stdout/stderr. func (p *Provider) RunInContainer(ctx context.Context, namespace, podName, containerName string, cmd []string, attach api.AttachIO) error { + hostPodName := p.Translater.TranslateName(namespace, podName) req := p.CoreClient.RESTClient().Post(). Resource("pods"). - Name(p.hostName(namespace, podName)). + Name(hostPodName). Namespace(p.ClusterNamespace). SubResource("exec") req.VersionedParams(&corev1.PodExecOptions{ @@ -124,9 +141,10 @@ func (p *Provider) RunInContainer(ctx context.Context, namespace, podName, conta // AttachToContainer attaches to the executing process of a container in the pod, copying data // between in/out/err and the container's stdin/stdout/stderr. func (p *Provider) AttachToContainer(ctx context.Context, namespace, podName, containerName string, attach api.AttachIO) error { + hostPodName := p.Translater.TranslateName(namespace, podName) req := p.CoreClient.RESTClient().Post(). Resource("pods"). - Name(p.hostName(namespace, podName)). + Name(hostPodName). Namespace(p.ClusterNamespace). SubResource("attach") req.VersionedParams(&corev1.PodAttachOptions{ @@ -163,9 +181,10 @@ func (p *Provider) GetMetricsResource(context.Context) ([]*dto.MetricFamily, err // PortForward forwards a local port to a port on the pod func (p *Provider) PortForward(ctx context.Context, namespace, pod string, port int32, stream io.ReadWriteCloser) error { + hostPodName := p.Translater.TranslateName(namespace, pod) req := p.CoreClient.RESTClient().Post(). Resource("pods"). - Name(p.hostName(namespace, pod)). + Name(hostPodName). Namespace(p.ClusterNamespace). SubResource("portforward") @@ -193,52 +212,189 @@ func (p *Provider) PortForward(ctx context.Context, namespace, pod string, port // CreatePod takes a Kubernetes Pod and deploys it within the provider. func (p *Provider) CreatePod(ctx context.Context, pod *corev1.Pod) error { - translated := p.translateTo(pod) + tPod := pod.DeepCopy() + p.Translater.TranslateTo(tPod) + // these values shouldn't be set on create - translated.UID = "" - translated.ResourceVersion = "" - // can't handle volumes yet, just omit them for now - translated.Spec.Volumes = nil - containers := []corev1.Container{} - for _, container := range translated.Spec.Containers { - container.VolumeMounts = nil - containers = append(containers, container) + tPod.UID = "" + tPod.ResourceVersion = "" + + // the node was scheduled on the virtual kubelet, but leaving it this way will make it pending indefinitely + tPod.Spec.NodeName = "" + + // volumes will often refer to resources in the virtual cluster, but instead need to refer to the sync'd + // host cluster version + 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) } - translated.Spec.Containers = containers - translated.Spec.NodeName = "" - p.logger.Infof("Creating pod %s/%s for pod %s/%s", translated.Namespace, translated.Name, pod.Namespace, pod.Name) - _, err := p.CoreClient.Pods(p.ClusterNamespace).Create(ctx, translated, metav1.CreateOptions{}) - return 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) +} + +// transformVolumes changes the volumes to the representation in the host cluster. Will return an 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 { + // 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 { + 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 { + 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 { + configMapName := source.ConfigMap.Name + if err := p.syncConfigmap(ctx, podNamespace, configMapName); 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 { + secretName := source.Secret.Name + if err := p.syncSecret(ctx, podNamespace, secretName); err != nil { + return fmt.Errorf("unable to sync projected secret %s: %w", secretName, err) + } + } + } + } + } + return nil +} + +func (p *Provider) syncConfigmap(ctx context.Context, podNamespace string, configMapName string) error { + var configMap corev1.ConfigMap + nsName := types.NamespacedName{ + Namespace: podNamespace, + Name: configMapName, + } + err := p.VirtualClient.Get(ctx, nsName, &configMap) + if err != nil { + return fmt.Errorf("unable to get configmap to sync %s/%s: %w", nsName.Namespace, nsName.Name, err) + } + err = p.Handler.AddResource(ctx, &configMap) + if err != nil { + return fmt.Errorf("unable to add configmap to sync %s/%s: %w", nsName.Namespace, nsName.Name, err) + } + return nil +} + +func (p *Provider) syncSecret(ctx context.Context, podNamespace string, secretName string) error { + var secret corev1.Secret + nsName := types.NamespacedName{ + Namespace: podNamespace, + Name: secretName, + } + 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) + } + err = p.Handler.AddResource(ctx, &secret) + if err != nil { + return fmt.Errorf("unable to add configmap to sync %s/%s: %w", nsName.Namespace, nsName.Name, err) + } + return nil } // UpdatePod takes a Kubernetes Pod and updates it within the provider. func (p *Provider) UpdatePod(ctx context.Context, pod *corev1.Pod) error { - currentPod, err := p.GetPod(ctx, p.ClusterNamespace, p.hostName(p.ClusterNamespace, p.ClusterName)) + hostName := p.Translater.TranslateName(pod.Namespace, pod.Name) + currentPod, err := p.GetPod(ctx, p.ClusterNamespace, hostName) if err != nil { - return err + return fmt.Errorf("unable to get current pod for update: %w", err) } - translated := p.translateTo(pod) - translated.UID = currentPod.UID - translated.ResourceVersion = currentPod.ResourceVersion - containers := []corev1.Container{} - for _, container := range translated.Spec.Containers { - container.VolumeMounts = nil - containers = append(containers, container) - } - translated.Spec.Containers = containers - translated.Spec.NodeName = currentPod.Spec.NodeName + tPod := pod.DeepCopy() + p.Translater.TranslateTo(tPod) + tPod.UID = currentPod.UID + // this is a bit dangerous since another process could have made changes that the user didn't know about + tPod.ResourceVersion = currentPod.ResourceVersion - _, err = p.CoreClient.Pods(p.ClusterNamespace).Update(ctx, translated, metav1.UpdateOptions{}) - return err + // Volumes may refer to resources (configmaps/secrets) from the host cluster + // So we need the configuration as calculated during create time + tPod.Spec.Volumes = currentPod.Spec.Volumes + tPod.Spec.Containers = currentPod.Spec.Containers + tPod.Spec.InitContainers = currentPod.Spec.InitContainers + tPod.Spec.NodeName = currentPod.Spec.NodeName + + return p.HostClient.Update(ctx, tPod) } // DeletePod takes a Kubernetes Pod and deletes it from the provider. Once a pod is deleted, the provider is // expected to call the NotifyPods callback with a terminal pod status where all the containers are in a terminal // state, as well as the pod. DeletePod may be called multiple times for the same pod. func (p *Provider) DeletePod(ctx context.Context, pod *corev1.Pod) error { - err := p.CoreClient.Pods(p.ClusterNamespace).Delete(ctx, p.translateTo(pod).Name, metav1.DeleteOptions{}) + p.logger.Infof("Got request to delete pod %s", pod.Name) + hostName := p.Translater.TranslateName(pod.Namespace, pod.Name) + err := p.CoreClient.Pods(p.ClusterNamespace).Delete(ctx, hostName, metav1.DeleteOptions{}) + if err != nil { + return fmt.Errorf("unable to delete pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + if err = p.pruneUnusedVolumes(ctx, pod); err != nil { + // note that we don't return an error here. The pod was sucessfully deleted, another process + // should clean this without affecting the user + p.logger.Errorf("failed to prune leftover volumes for %s/%s: %w, resources may be left", pod.Namespace, pod.Name, err) + } p.logger.Infof("Deleted pod %s", pod.Name) - return err + return nil +} + +// pruneUnusedVolumes removes volumes in use by pod that aren't used by any other pods +func (p *Provider) pruneUnusedVolumes(ctx context.Context, pod *corev1.Pod) error { + rawSecrets, rawConfigMaps := getSecretsAndConfigmaps(pod) + // since this pod was removed, originally mark all of the secrets/configmaps it uses as eligible + // for pruning + pruneSecrets := sets.Set[string]{}.Insert(rawSecrets...) + pruneConfigMap := sets.Set[string]{}.Insert(rawConfigMaps...) + var pods corev1.PodList + // only pods in the same namespace could be using secrets/configmaps that this pod is using + err := p.VirtualClient.List(ctx, &pods, &client.ListOptions{ + Namespace: pod.Namespace, + }) + if err != nil { + return fmt.Errorf("unable to list pods: %w", err) + } + for _, vPod := range pods.Items { + if vPod.Name == pod.Name { + continue + } + secrets, configMaps := getSecretsAndConfigmaps(&vPod) + pruneSecrets.Delete(secrets...) + pruneConfigMap.Delete(configMaps...) + } + for _, secretName := range pruneSecrets.UnsortedList() { + var secret corev1.Secret + err := p.VirtualClient.Get(ctx, types.NamespacedName{ + Name: secretName, + Namespace: pod.Namespace, + }, &secret) + if err != nil { + return fmt.Errorf("unable to get secret %s/%s for pod volume: %w", pod.Namespace, secretName, err) + } + err = p.Handler.RemoveResource(ctx, &secret) + if err != nil { + return fmt.Errorf("unable to remove secret %s/%s for pod volume: %w", pod.Namespace, secretName, err) + } + } + for _, configMapName := range pruneConfigMap.UnsortedList() { + var configMap corev1.ConfigMap + err := p.VirtualClient.Get(ctx, types.NamespacedName{ + Name: configMapName, + Namespace: pod.Namespace, + }, &configMap) + if err != nil { + return fmt.Errorf("unable to get configMap %s/%s for pod volume: %w", pod.Namespace, configMapName, err) + } + + if err = p.Handler.RemoveResource(ctx, &configMap); err != nil { + return fmt.Errorf("unable to remove configMap %s/%s for pod volume: %w", pod.Namespace, configMapName, err) + } + } + return nil } // GetPod retrieves a pod by name from the provider (can be cached). @@ -246,11 +402,18 @@ func (p *Provider) DeletePod(ctx context.Context, pod *corev1.Pod) error { // 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) { - pod, err := p.CoreClient.Pods(p.ClusterNamespace).Get(ctx, p.hostName(namespace, name), metav1.GetOptions{}) - if err != nil { - return nil, err + p.logger.Errorf("got a request for get pod %s, %s", namespace, name) + hostNamespaceName := types.NamespacedName{ + Namespace: p.ClusterNamespace, + Name: p.Translater.TranslateName(namespace, name), } - return p.translateFrom(pod), nil + var pod corev1.Pod + err := p.HostClient.Get(ctx, hostNamespaceName, &pod) + if err != nil { + return nil, fmt.Errorf("error when retrieving pod: %w", err) + } + p.Translater.TranslateFrom(&pod) + return &pod, nil } // GetPodStatus retrieves the status of a pod by name from the provider. @@ -258,10 +421,13 @@ 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) pod, err := p.GetPod(ctx, namespace, name) if err != nil { - return nil, err + 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) return pod.Status.DeepCopy(), nil } @@ -270,46 +436,44 @@ func (p *Provider) GetPodStatus(ctx context.Context, namespace, name string) (*c // concurrently outside of the calling goroutine. Therefore it is recommended // to return a version after DeepCopy. func (p *Provider) GetPods(ctx context.Context) ([]*corev1.Pod, error) { - labelSelector := metav1.AddLabelToSelector(&metav1.LabelSelector{}, clusterNameLabel, p.ClusterName) - pods, err := p.CoreClient.Pods(p.ClusterNamespace).List(ctx, metav1.ListOptions{LabelSelector: labelSelector.String()}) + selector := labels.NewSelector() + requirement, err := labels.NewRequirement(translate.ClusterNameLabel, selection.Equals, []string{p.ClusterName}) if err != nil { - return nil, err + return nil, fmt.Errorf("unable to create label selector: %w", err) + } + selector = selector.Add(*requirement) + var podList corev1.PodList + err = p.HostClient.List(ctx, &podList, &client.ListOptions{LabelSelector: selector}) + if err != nil { + return nil, fmt.Errorf("unable to list pods: %w", err) } retPods := []*corev1.Pod{} - for _, pod := range pods.DeepCopy().Items { - retPods = append(retPods, p.translateFrom(&pod)) + for _, pod := range podList.DeepCopy().Items { + p.Translater.TranslateFrom(&pod) + retPods = append(retPods, &pod) } return retPods, nil } -// translateTo translates a pod's definition from the virutal cluster to the definition on the host cluster -func (p *Provider) translateTo(virtualPod *corev1.Pod) *corev1.Pod { - hostPod := virtualPod.DeepCopy() - hostPod.Name = p.hostName(virtualPod.Namespace, virtualPod.Name) - hostPod.Namespace = p.ClusterNamespace - if hostPod.Labels == nil { - hostPod.Labels = map[string]string{} +// 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) { + var secrets []string + var configMaps []string + for _, volume := range pod.Spec.Volumes { + if volume.Secret != nil { + secrets = append(secrets, volume.Secret.SecretName) + } else if volume.ConfigMap != nil { + configMaps = append(configMaps, volume.ConfigMap.Name) + } else if volume.Projected != nil { + for _, source := range volume.Projected.Sources { + if source.ConfigMap != nil { + configMaps = append(configMaps, source.ConfigMap.Name) + } else if source.Secret != nil { + secrets = append(secrets, source.Secret.Name) + } + } + } } - if hostPod.Annotations == nil { - hostPod.Annotations = map[string]string{} - } - hostPod.Labels[clusterNameLabel] = p.ClusterName - hostPod.Annotations[podNameAnnotation] = virtualPod.Name - hostPod.Annotations[podNamespaceAnnotation] = virtualPod.Namespace - return hostPod -} - -// translateFrom translates a pod's definition from the host cluster to the definition on the host cluster -func (p *Provider) translateFrom(hostPod *corev1.Pod) *corev1.Pod { - virtualPod := hostPod.DeepCopy() - delete(virtualPod.Labels, clusterNameLabel) - virtualPod.Name = hostPod.Annotations[podNameAnnotation] - virtualPod.Namespace = hostPod.Annotations[podNamespaceAnnotation] - delete(virtualPod.Annotations, podNameAnnotation) - delete(virtualPod.Annotations, podNamespaceAnnotation) - return virtualPod -} - -func (p *Provider) hostName(virtualNamespace, virtualName string) string { - return controller.SafeConcatName(p.ClusterName, p.ClusterNamespace, virtualNamespace, virtualName) + return secrets, configMaps } diff --git a/k3k-kubelet/translate/host.go b/k3k-kubelet/translate/host.go new file mode 100644 index 00000000..64053968 --- /dev/null +++ b/k3k-kubelet/translate/host.go @@ -0,0 +1,105 @@ +package translate + +import ( + "encoding/hex" + "fmt" + + "github.com/rancher/k3k/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +const ( + // ClusterNameLabel is the key for the label that contains the name of the virtual cluster + // this resource was made in + ClusterNameLabel = "k3k.io/clusterName" + // ResourceNameAnnotation is the key for the annotation that contains the original name of this + // resource in the virtual cluster + ResourceNameAnnotation = "k3k.io/name" + // ResourceNamespaceAnnotation is the key for the annotation that contains the original namespace of this + // resource in the virtual cluster + ResourceNamespaceAnnotation = "k3k.io/namespace" +) + +type ToHostTranslater struct { + // ClusterName is the name of the virtual cluster whose resources we are + // translating to a host cluster + ClusterName string + // ClusterNamespace is the namespace of the virtual cluster whose resources + // we are tranlsating to a host cluster + ClusterNamespace string +} + +// Translate translates a virtual cluster object to a host cluster object. This should only be used for +// static resources such as configmaps/secrets, and not for things like pods (which can reference other +// objects). Note that this won't set host-cluster values (like resource version) so when updating you +// may need to fetch the existing value and do some combination before using this. +func (t *ToHostTranslater) TranslateTo(obj client.Object) { + // owning objects may be in the virtual cluster, but may not be in the host cluster + obj.SetOwnerReferences(nil) + // add some annotations to make it easier to track source object + annotations := obj.GetAnnotations() + if annotations == nil { + annotations = map[string]string{} + } + annotations[ResourceNameAnnotation] = obj.GetName() + annotations[ResourceNamespaceAnnotation] = obj.GetNamespace() + obj.SetAnnotations(annotations) + // add a label to quickly identify objects owned by a given virtual cluster + labels := obj.GetLabels() + if labels == nil { + labels = map[string]string{} + } + labels[ClusterNameLabel] = t.ClusterName + obj.SetLabels(labels) + + // resource version/UID won't match what's in the host cluster. + obj.SetResourceVersion("") + obj.SetUID("") + + // set the name and the namespace so that this goes in the proper host namespace + // and doesn't collide with other resources + obj.SetName(t.TranslateName(obj.GetNamespace(), obj.GetName())) + obj.SetNamespace(t.ClusterNamespace) +} + +func (t *ToHostTranslater) TranslateFrom(obj client.Object) { + // owning objects may be in the virtual cluster, but may not be in the host cluster + obj.SetOwnerReferences(nil) + + // remove the annotations added to track original name + annotations := obj.GetAnnotations() + // TODO: It's possible that this was erased by a change on the host cluster + // In this case, we need to have some sort of fallback or error return + name := annotations[ResourceNameAnnotation] + namespace := annotations[ResourceNamespaceAnnotation] + obj.SetName(name) + obj.SetNamespace(namespace) + delete(annotations, ResourceNameAnnotation) + delete(annotations, ResourceNamespaceAnnotation) + obj.SetAnnotations(annotations) + + // remove the clusteName tracking label + labels := obj.GetLabels() + delete(labels, ClusterNameLabel) + obj.SetLabels(labels) + + // resource version/UID won't match what's in the virtual cluster. + obj.SetResourceVersion("") + obj.SetUID("") + +} + +// TranslateName returns the name of the resource in the host cluster. Will not update the object with this name. +func (t *ToHostTranslater) TranslateName(namespace string, name string) string { + // we need to come up with a name which is: + // - somewhat connectable to the original resource + // - a valid k8s name + // - idempotently calculatable + // - unique for this combination of name/namespace/cluster + namePrefix := fmt.Sprintf("%s-%s-%s", name, namespace, t.ClusterName) + // use + as a separator since it can't be in an object name + nameKey := fmt.Sprintf("%s+%s+%s", name, namespace, t.ClusterName) + // it's possible that the suffix will be in the name, so we use hex to make it valid for k8s + nameSuffix := hex.EncodeToString([]byte(nameKey)) + return controller.SafeConcatName(namePrefix, nameSuffix) +} diff --git a/pkg/controller/cluster/agent/shared.go b/pkg/controller/cluster/agent/shared.go index 22393c41..cf0a6acd 100644 --- a/pkg/controller/cluster/agent/shared.go +++ b/pkg/controller/cluster/agent/shared.go @@ -201,12 +201,12 @@ func (s *SharedAgent) role() *rbacv1.Role { { Verbs: []string{"*"}, APIGroups: []string{""}, - Resources: []string{"pods"}, + Resources: []string{"pods", "secrets", "configmaps"}, }, { Verbs: []string{"get", "watch", "list"}, APIGroups: []string{""}, - Resources: []string{"secrets", "services"}, + Resources: []string{"services"}, }, { Verbs: []string{"get", "watch", "list"},