github/k3k
1
0
Fork 0
mirror of https://github.com/rancher/k3k.git synced 2026-08-23 22:36:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix for pod eviction in host cluster (#484)

Browse Source 
* update statefulset controller

* fix for single pod

* adding pod controller

* added test

* removed comment

* merged service controller

* revert statefulset

* added test

* added common owner filter
This commit is contained in:
Enrico Candino
2025-10-03 16:22:54 +02:00
committed by GitHub
parent 6523b8339b
commit 5c95ca3dfa
9 changed files with 226 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
k3k-kubelet/provider/provider.go
+8 -2
View File
@@ -31,6 +31,7 @@ import (
dto "github.com/prometheus/client_model/go"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
cv1 "k8s.io/client-go/kubernetes/typed/core/v1"
compbasemetrics "k8s.io/component-base/metrics"
@@ -589,15 +590,20 @@ func (p *Provider) DeletePod(ctx context.Context, pod *corev1.Pod) error {
// 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 {
p.logger.Infof("Got request to delete pod %s", pod.Name)
p.logger.Infof("got request to delete pod %s/%s", pod.Namespace, pod.Name)
hostName := p.Translator.TranslateName(pod.Namespace, pod.Name)
err := p.CoreClient.Pods(p.ClusterNamespace).Delete(ctx, hostName, metav1.DeleteOptions{})
if err != nil {
if apierrors.IsNotFound(err) {
p.logger.Infof("pod %s/%s already deleted from host cluster", p.ClusterNamespace, hostName)
return nil
}
return fmt.Errorf("unable to delete pod %s/%s: %w", pod.Namespace, pod.Name, err)
}
p.logger.Infof("Deleted pod %s", pod.Name)
p.logger.Infof("pod %s/%s deleted from host cluster", p.ClusterNamespace, hostName)
return nil
}
k3k-kubelet/translate/host.go
+17
View File
@@ -4,8 +4,10 @@ import (
"encoding/hex"
"strings"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
"github.com/rancher/k3k/pkg/controller"
)
@@ -34,6 +36,13 @@ type ToHostTranslator struct {
ClusterNamespace string
}
func NewHostTranslator(cluster *v1alpha1.Cluster) *ToHostTranslator {
return &ToHostTranslator{
ClusterName: cluster.Name,
ClusterNamespace: cluster.Namespace,
}
}
// 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
@@ -125,3 +134,11 @@ func (t *ToHostTranslator) TranslateName(namespace string, name string) string {
return controller.SafeConcatName(namePrefix, nameSuffix)
}
// NamespacedName returns the types.NamespacedName of the resource in the host cluster
func (t *ToHostTranslator) NamespacedName(obj client.Object) types.NamespacedName {
return types.NamespacedName{
Namespace: t.ClusterNamespace,
Name: t.TranslateName(obj.GetNamespace(), obj.GetName()),
}
}
main.go
+11 -5
View File
@@ -114,29 +114,35 @@ func run(cmd *cobra.Command, args []string) error {
}
if err := cluster.Add(ctx, mgr, &config, maxConcurrentReconciles, portAllocator, nil); err != nil {
return fmt.Errorf("failed to add the new cluster controller: %v", err)
return fmt.Errorf("failed to add cluster controller: %v", err)
}
logger.Info("adding statefulset controller")
if err := cluster.AddStatefulSetController(ctx, mgr, maxConcurrentReconciles); err != nil {
return fmt.Errorf("failed to add the statefulset controller: %v", err)
return fmt.Errorf("failed to add statefulset controller: %v", err)
}
logger.Info("adding service controller")
if err := cluster.AddServiceController(ctx, mgr, maxConcurrentReconciles); err != nil {
return fmt.Errorf("failed to add the new service controller: %v", err)
return fmt.Errorf("failed to add service controller: %v", err)
}
logger.Info("adding pod controller")
if err := cluster.AddPodController(ctx, mgr, maxConcurrentReconciles); err != nil {
return fmt.Errorf("failed to add pod controller: %v", err)
}
logger.Info("adding clusterpolicy controller")
if err := policy.Add(mgr, config.ClusterCIDR, maxConcurrentReconciles); err != nil {
return fmt.Errorf("failed to add the clusterpolicy controller: %v", err)
return fmt.Errorf("failed to add clusterpolicy controller: %v", err)
}
if err := mgr.Start(ctx); err != nil {
return fmt.Errorf("failed to start the manager: %v", err)
return fmt.Errorf("failed to start manager: %v", err)
}
logger.Info("controller manager stopped")
pkg/controller/cluster/client.go
+35
View File
@@ -0,0 +1,35 @@
package cluster
import (
"context"
"fmt"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/clientcmd"
v1 "k8s.io/api/core/v1"
ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
"github.com/rancher/k3k/pkg/controller"
)
// newVirtualClient
func newVirtualClient(ctx context.Context, hostClient ctrlruntimeclient.Client, clusterName, clusterNamespace string) (ctrlruntimeclient.Client, error) {
var clusterKubeConfig v1.Secret
kubeconfigSecretName := types.NamespacedName{
Name: controller.SafeConcatNameWithPrefix(clusterName, "kubeconfig"),
Namespace: clusterNamespace,
}
if err := hostClient.Get(ctx, kubeconfigSecretName, &clusterKubeConfig); err != nil {
return nil, fmt.Errorf("failed to get kubeconfig secret: %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 ctrlruntimeclient.New(restConfig, ctrlruntimeclient.Options{})
}
pkg/controller/cluster/filter.go
+38
View File
@@ -0,0 +1,38 @@
package cluster
import (
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/predicate"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/rancher/k3k/k3k-kubelet/translate"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
)
func newClusterPredicate() predicate.Predicate {
return predicate.NewPredicateFuncs(func(object client.Object) bool {
owner := metav1.GetControllerOf(object)
return owner != nil &&
owner.Kind == "Cluster" &&
owner.APIVersion == v1alpha1.SchemeGroupVersion.String()
})
}
func clusterNamespacedName(object client.Object) types.NamespacedName {
var clusterName string
owner := metav1.GetControllerOf(object)
if owner != nil && owner.Kind == "Cluster" && owner.APIVersion == v1alpha1.SchemeGroupVersion.String() {
clusterName = owner.Name
} else {
clusterName = object.GetLabels()[translate.ClusterNameLabel]
}
return types.NamespacedName{
Name: clusterName,
Namespace: object.GetNamespace(),
}
}
pkg/controller/cluster/pod.go
+81
View File
@@ -0,0 +1,81 @@
package cluster
import (
"context"
"k8s.io/apimachinery/pkg/runtime"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
ctrl "sigs.k8s.io/controller-runtime"
ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
"github.com/rancher/k3k/k3k-kubelet/translate"
)
const (
podController = "k3k-pod-controller"
)
type PodReconciler struct {
Client ctrlruntimeclient.Client
Scheme *runtime.Scheme
}
// AddPodController adds a new controller for Pods to the manager.
// It will reconcile the Pods of the Host Cluster with the one of the Virtual Cluster.
func AddPodController(ctx context.Context, mgr manager.Manager, maxConcurrentReconciles int) error {
reconciler := PodReconciler{
Client: mgr.GetClient(),
Scheme: mgr.GetScheme(),
}
return ctrl.NewControllerManagedBy(mgr).
For(&v1.Pod{}).
Named(podController).
WithEventFilter(newClusterPredicate()).
WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}).
Complete(&reconciler)
}
func (r *PodReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx)
log.Info("reconciling pod")
var pod v1.Pod
if err := r.Client.Get(ctx, req.NamespacedName, &pod); err != nil {
if !apierrors.IsNotFound(err) {
return reconcile.Result{}, err
}
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
// get cluster from the object
cluster := clusterNamespacedName(&pod)
virtualClient, err := newVirtualClient(ctx, r.Client, cluster.Name, cluster.Namespace)
if err != nil {
return reconcile.Result{}, err
}
if !pod.DeletionTimestamp.IsZero() {
virtName := pod.GetAnnotations()[translate.ResourceNameAnnotation]
virtNamespace := pod.GetAnnotations()[translate.ResourceNamespaceAnnotation]
virtPod := v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: virtName,
Namespace: virtNamespace,
},
}
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(virtualClient.Delete(ctx, &virtPod))
}
return reconcile.Result{}, nil
}
pkg/controller/cluster/service.go
+13 -51
View File
@@ -6,9 +6,7 @@ import (
"k8s.io/apimachinery/pkg/api/equality"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/clientcmd"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
v1 "k8s.io/api/core/v1"
@@ -16,7 +14,6 @@ import (
ctrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client"
"github.com/rancher/k3k/k3k-kubelet/translate"
"github.com/rancher/k3k/pkg/controller"
)
const (
@@ -35,83 +32,48 @@ func AddServiceController(ctx context.Context, mgr manager.Manager, maxConcurren
return ctrl.NewControllerManagedBy(mgr).
Named(serviceController).
For(&v1.Service{}).WithEventFilter(predicate.NewPredicateFuncs(reconciler.filterResources)).
For(&v1.Service{}).
WithEventFilter(newClusterPredicate()).
Complete(&reconciler)
}
func (r *ServiceReconciler) filterResources(object ctrlruntimeclient.Object) bool {
_, hasClusterNameLabel := object.GetLabels()[translate.ClusterNameLabel]
_, hasNameAnnotation := object.GetAnnotations()[translate.ResourceNameAnnotation]
_, hasNamespaceAnnotation := object.GetAnnotations()[translate.ResourceNamespaceAnnotation]
// Return true only if all were found
return hasClusterNameLabel && hasNameAnnotation && hasNamespaceAnnotation
}
func (r *ServiceReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
log := ctrl.LoggerFrom(ctx)
log.Info("ensuring service status to virtual cluster")
var (
virtServiceName, virtServiceNamespace string
hostService, virtService v1.Service
)
var hostService, virtService v1.Service
if err := r.HostClient.Get(ctx, req.NamespacedName, &hostService); err != nil {
return reconcile.Result{}, ctrlruntimeclient.IgnoreNotFound(err)
}
// get cluster information from the object
labels := hostService.GetLabels()
// get cluster from the object
cluster := clusterNamespacedName(&hostService)
clusterName, ok := labels[translate.ClusterNameLabel]
if !ok {
return reconcile.Result{}, fmt.Errorf("cluster name label is not found")
}
clusterNamespace := hostService.GetNamespace()
virtClient, err := r.newVirtualClient(ctx, &hostService, clusterName, clusterNamespace)
virtualClient, err := newVirtualClient(ctx, r.HostClient, cluster.Name, cluster.Namespace)
if err != nil {
return reconcile.Result{}, fmt.Errorf("failed to get cluster info: %v", err)
}
virtServiceName = hostService.Annotations[translate.ResourceNameAnnotation]
virtServiceNamespace = hostService.Annotations[translate.ResourceNamespaceAnnotation]
if !hostService.DeletionTimestamp.IsZero() {
return reconcile.Result{}, nil
}
if err := virtClient.Get(ctx, types.NamespacedName{Name: virtServiceName, Namespace: virtServiceNamespace}, &virtService); err != nil {
virtualServiceKey := types.NamespacedName{
Name: hostService.Annotations[translate.ResourceNameAnnotation],
Namespace: hostService.Annotations[translate.ResourceNamespaceAnnotation],
}
if err := virtualClient.Get(ctx, virtualServiceKey, &virtService); err != nil {
return reconcile.Result{}, fmt.Errorf("failed to get virt service: %v", err)
}
if !equality.Semantic.DeepEqual(virtService.Status.LoadBalancer, hostService.Status.LoadBalancer) {
virtService.Status.LoadBalancer = hostService.Status.LoadBalancer
if err := virtClient.Status().Update(ctx, &virtService); err != nil {
if err := virtualClient.Status().Update(ctx, &virtService); err != nil {
return reconcile.Result{}, err
}
}
return reconcile.Result{}, nil
}
func (r *ServiceReconciler) newVirtualClient(ctx context.Context, obj ctrlruntimeclient.Object, clusterName, clusterNamespace string) (ctrlruntimeclient.Client, error) {
var clusterKubeConfig v1.Secret
kubeconfigSecretName := controller.SafeConcatNameWithPrefix(clusterName, "kubeconfig")
if err := r.HostClient.Get(ctx, types.NamespacedName{Name: kubeconfigSecretName, Namespace: clusterNamespace}, &clusterKubeConfig); err != nil {
return nil, err
}
virtConfig, err := clientcmd.RESTConfigFromKubeConfig(clusterKubeConfig.Data["kubeconfig.yaml"])
if err != nil {
return nil, fmt.Errorf("failed to create config from kubeconfig file: %v", err)
}
return ctrlruntimeclient.New(virtConfig, ctrlruntimeclient.Options{})
}
tests/cluster_persistence_test.go
+21
View File
@@ -10,8 +10,10 @@ import (
"k8s.io/utils/ptr"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/rancher/k3k/k3k-kubelet/translate"
"github.com/rancher/k3k/pkg/apis/k3k.io/v1alpha1"
. "github.com/onsi/ginkgo/v2"
@@ -33,6 +35,25 @@ var _ = When("an ephemeral cluster is installed", Label("e2e"), func() {
_, _ = virtualCluster.NewNginxPod("")
})
It("deletes the pod in the virtual cluster when deleted from the host", func() {
ctx := context.Background()
pod, _ := virtualCluster.NewNginxPod("")
hostTranslator := translate.NewHostTranslator(virtualCluster.Cluster)
namespacedName := hostTranslator.NamespacedName(pod)
err := k8s.CoreV1().Pods(namespacedName.Namespace).Delete(ctx, namespacedName.Name, v1.DeleteOptions{})
Expect(err).To(Not(HaveOccurred()))
Eventually(func() bool {
_, err := virtualCluster.Client.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, v1.GetOptions{})
return apierrors.IsNotFound(err)
}).
WithPolling(time.Second * 5).
WithTimeout(time.Minute).
Should(BeTrue())
})
It("regenerates the bootstrap secret after a restart", func() {
ctx := context.Background()
tests/common_test.go
+2
View File
@@ -35,6 +35,8 @@ type VirtualCluster struct {
}
func NewVirtualCluster() *VirtualCluster { // By default, create an ephemeral cluster
GinkgoHelper()
return NewVirtualClusterWithType(v1alpha1.EphemeralPersistenceMode)
}