Implemented OffloadingPatch consumer-side

Author: fra98

apis/virtualkubelet/v1alpha1/virtualnode_types.go

@@ -35,7 +35,7 @@ type OffloadingPatch struct {
 	// Tolerations contains the tolerations to target the remote cluster.
 	Tolerations []corev1.Toleration `json:"tolerations,omitempty"`
 	// Affinity contains the affinity and anti-affinity rules to target the remote cluster.
-	Affinities *Affinity `json:"affinities,omitempty"`
+	Affinity *Affinity `json:"affinity,omitempty"`
 }
 
 // DeploymentTemplate contains the deployment template of the virtual node.

apis/virtualkubelet/v1alpha1/zz_generated.deepcopy.go

@@ -18,20 +18,25 @@ package root
 import (
 	"context"
 	"fmt"
+	"os"
 
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 	"github.com/virtual-kubelet/virtual-kubelet/node"
 	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/kubernetes"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	corev1clients "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	virtualkubeletv1alpha1 "github.com/liqotech/liqo/apis/virtualkubelet/v1alpha1"
 	"github.com/liqotech/liqo/pkg/consts"
 	identitymanager "github.com/liqotech/liqo/pkg/identityManager"
 	tenantnamespace "github.com/liqotech/liqo/pkg/tenantNamespace"
@@ -44,6 +49,15 @@ import (
 	"github.com/liqotech/liqo/pkg/virtualKubelet/reflection/generic"
 )
 
+var (
+	scheme = runtime.NewScheme()
+)
+
+func init() {
+	_ = clientgoscheme.AddToScheme(scheme)
+	_ = virtualkubeletv1alpha1.AddToScheme(scheme)
+}
+
 const defaultVersion = "v1.25.0" // This should follow the version of k8s.io/kubernetes we are importing
 
 // NewCommand creates a new top-level command.
@@ -102,6 +116,22 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 
 	restcfg.SetRateLimiter(remoteConfig)
 
+	cl, err := client.New(localConfig, client.Options{
+		Scheme: scheme,
+	})
+	if err != nil {
+		return err
+	}
+
+	// Get virtual node
+	vnName := os.Getenv("VIRTUALNODE_NAME")
+	ns := os.Getenv("POD_NAMESPACE")
+	var vn virtualkubeletv1alpha1.VirtualNode
+	if err := cl.Get(ctx, client.ObjectKey{Name: vnName, Namespace: ns}, &vn); err != nil {
+		klog.Errorf("Unable to get virtual node: %v", err)
+		return err
+	}
+
 	// Initialize the pod provider
 	podcfg := podprovider.InitConfig{
 		LocalConfig:   localConfig,
@@ -134,6 +164,8 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 
 		LabelsNotReflected:      c.LabelsNotReflected.StringList,
 		AnnotationsNotReflected: c.AnnotationsNotReflected.StringList,
+
+		OffloadingPatch: vn.Spec.OffloadingPatch,
 	}
 
 	eb := record.NewBroadcaster()

cmd/virtual-kubelet/root/root.go

@@ -156,7 +156,7 @@ spec:
                 description: OffloadingPatch contains the information to target a
                   groups of node on the remote cluster.
                 properties:
-                  affinities:
+                  affinity:
                     description: Affinity contains the affinity and anti-affinity
                       rules to target the remote cluster.
                     properties:

deployments/liqo/crds/virtualkubelet.liqo.io_virtualnodes.yaml

@@ -16,12 +16,16 @@ package virtualnodectrl
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
 	"fmt"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/klog/v2"
 	k8strings "k8s.io/utils/strings"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -33,6 +37,8 @@ import (
 	vkutils "github.com/liqotech/liqo/pkg/vkMachinery/utils"
 )
 
+const offloadingPatchHashAnnotation = "liqo.io/offloading-patch-hash"
+
 // createVirtualKubeletDeployment creates the VirtualKubelet Deployment.
 func (r *VirtualNodeReconciler) ensureVirtualKubeletDeploymentPresence(
 	ctx context.Context, virtualNode *virtualkubeletv1alpha1.VirtualNode) (err error) {
@@ -50,6 +56,7 @@ func (r *VirtualNodeReconciler) ensureVirtualKubeletDeploymentPresence(
 			err = fmt.Errorf("error updating virtual node status: %w", interr)
 		}
 	}()
+
 	ForgeCondition(virtualNode,
 		VnConditionMap{
 			virtualkubeletv1alpha1.VirtualKubeletConditionType: VnCondition{
@@ -86,11 +93,28 @@ func (r *VirtualNodeReconciler) ensureVirtualKubeletDeploymentPresence(
 		remoteClusterIdentity.ClusterName, vkClusterRoleBinding.Name, op)
 
 	// forge the virtual Kubelet Deployment
-	vkDeployment := &appsv1.Deployment{}
-	vkDeployment.ObjectMeta = *virtualNode.Spec.Template.ObjectMeta.DeepCopy()
+	vkDeployment := appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      virtualNode.Spec.Template.GetName(),
+			Namespace: virtualNode.Spec.Template.GetNamespace(),
+		},
+	}
+	op, err = controllerutil.CreateOrUpdate(ctx, r.Client, &vkDeployment, func() error {
+		vkDeployment.Annotations = labels.Merge(vkDeployment.Annotations, virtualNode.Spec.Template.ObjectMeta.GetAnnotations())
+		vkDeployment.Labels = labels.Merge(vkDeployment.Labels, virtualNode.Spec.Template.ObjectMeta.GetLabels())
 
-	op, err = controllerutil.CreateOrUpdate(ctx, r.Client, vkDeployment, func() error {
 		vkDeployment.Spec = *virtualNode.Spec.Template.Spec.DeepCopy()
+
+		// Add the hash of the offloading patch as annotation
+		opHash, err := offloadingPatchHash(virtualNode.Spec.OffloadingPatch)
+		if err != nil {
+			return err
+		}
+		if vkDeployment.Spec.Template.Annotations == nil {
+			vkDeployment.Spec.Template.Annotations = make(map[string]string)
+		}
+		vkDeployment.Spec.Template.Annotations[offloadingPatchHashAnnotation] = opHash
+
 		return nil
 	})
 	if err != nil {
@@ -110,7 +134,9 @@ func (r *VirtualNodeReconciler) ensureVirtualKubeletDeploymentPresence(
 		VnConditionMap{
 			virtualkubeletv1alpha1.VirtualKubeletConditionType: VnCondition{
 				Status: virtualkubeletv1alpha1.RunningConditionStatusType,
-			}})
+			},
+		})
+
 	if *virtualNode.Spec.CreateNode {
 		ForgeCondition(virtualNode,
 			VnConditionMap{
@@ -160,3 +186,20 @@ func (r *VirtualNodeReconciler) ensureVirtualKubeletDeploymentAbsence(
 
 	return nil
 }
+
+func offloadingPatchHash(offloadingPatch *virtualkubeletv1alpha1.OffloadingPatch) (string, error) {
+	if offloadingPatch == nil {
+		return "", nil
+	}
+
+	opString, err := json.Marshal(offloadingPatch)
+	if err != nil {
+		klog.Error(err)
+		return "", err
+	}
+
+	opHash := sha256.Sum256(opString)
+	opHashHex := hex.EncodeToString(opHash[:])
+
+	return opHashHex, nil
+}

pkg/liqo-controller-manager/virtualnode-controller/virtualkubelet.go

@@ -19,7 +19,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
-	"k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 )
 
 // IsPodReady returns true if a pod is ready; false otherwise. It also returns a reason (as provided by Kubernetes).
@@ -60,7 +60,7 @@ func IsPodSpecEqual(previous, updated *corev1.PodSpec) bool {
 	// * spec.tolerations (only new entries can be added)
 	return AreContainersEqual(previous.Containers, updated.Containers) &&
 		AreContainersEqual(previous.InitContainers, updated.InitContainers) &&
-		pointer.Int64Equal(previous.ActiveDeadlineSeconds, updated.ActiveDeadlineSeconds) &&
+		ptr.Equal(previous.ActiveDeadlineSeconds, updated.ActiveDeadlineSeconds) &&
 		len(previous.Tolerations) == len(updated.Tolerations)
 }
 

pkg/utils/pod/pod.go

@@ -21,6 +21,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	discoveryv1alpha1 "github.com/liqotech/liqo/apis/discovery/v1alpha1"
+	virtualkubeletv1alpha1 "github.com/liqotech/liqo/apis/virtualkubelet/v1alpha1"
 )
 
 // ReflectionFieldManager -> The name associated with the fields modified by virtual kubelet reflection.
@@ -72,12 +73,14 @@ func ApplyOptions() metav1.ApplyOptions {
 type ForgingOpts struct {
 	LabelsNotReflected      []string
 	AnnotationsNotReflected []string
+	OffloadingPatch         *virtualkubeletv1alpha1.OffloadingPatch
 }
 
 // NewForgingOpts returns a new ForgingOpts instance.
-func NewForgingOpts(labelsNotReflected, annotationsNotReflected []string) ForgingOpts {
+func NewForgingOpts(labelsNotReflected, annotationsNotReflected []string, offloadingPatch *virtualkubeletv1alpha1.OffloadingPatch) ForgingOpts {
 	return ForgingOpts{
 		LabelsNotReflected:      labelsNotReflected,
 		AnnotationsNotReflected: annotationsNotReflected,
+		OffloadingPatch:         offloadingPatch,
 	}
 }

pkg/virtualKubelet/forge/forge.go

@@ -26,7 +26,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	corev1apply "k8s.io/client-go/applyconfigurations/core/v1"
 	metricsv1beta1 "k8s.io/metrics/pkg/apis/metrics/v1beta1"
-	"k8s.io/utils/pointer"
+	"k8s.io/utils/ptr"
 
 	vkv1alpha1 "github.com/liqotech/liqo/apis/virtualkubelet/v1alpha1"
 	liqoconst "github.com/liqotech/liqo/pkg/consts"
@@ -225,7 +225,7 @@ func RemotePodSpec(creation bool, local, remote *corev1.PodSpec, mutators ...Rem
 
 	// The information about the service account name is not reflected, since the volume is already
 	// present, and the remote creation would fail as the corresponding service account is not present.
-	remote.AutomountServiceAccountToken = pointer.Bool(false)
+	remote.AutomountServiceAccountToken = ptr.To(false)
 
 	// This fields are currently forced to false, to prevent invasive settings on the remote cluster (which might not work).
 	remote.HostIPC = false
@@ -283,7 +283,7 @@ func ServiceAccountMutator(apiServerSupport APIServerSupportType, localAnnotatio
 			}
 
 			remote.ServiceAccountName = remoteServiceAccountName
-			remote.AutomountServiceAccountToken = pointer.Bool(true)
+			remote.AutomountServiceAccountToken = ptr.To(true)
 		default:
 			// Remove the service account name.
 			remote.ServiceAccountName = ""
@@ -353,6 +353,64 @@ func AntiAffinityHardMutator(labels map[string]string) RemotePodSpecMutator {
 	}
 }
 
+// NodeSelectorMutator is a mutator which implements the support to propagate a given node selector constraint.
+func NodeSelectorMutator(nodeSelector map[string]string) RemotePodSpecMutator {
+	return func(remote *corev1.PodSpec) {
+		if remote.NodeSelector == nil {
+			remote.NodeSelector = map[string]string{}
+		}
+
+		for k, v := range nodeSelector {
+			remote.NodeSelector[k] = v
+		}
+	}
+}
+
+// TolerationsMutator is a mutator which implements the support to propagate tolerations.
+func TolerationsMutator(tolerations []corev1.Toleration) RemotePodSpecMutator {
+	return func(remote *corev1.PodSpec) {
+		remote.Tolerations = append(remote.Tolerations, tolerations...)
+	}
+}
+
+// AffinityMutator is a mutator which implements the support to propagate affinity constraints.
+func AffinityMutator(affinity *vkv1alpha1.Affinity) RemotePodSpecMutator {
+	return func(remote *corev1.PodSpec) {
+		if affinity == nil || affinity.NodeAffinity == nil {
+			return
+		}
+
+		nodeAffinity := affinity.NodeAffinity.DeepCopy()
+
+		if remote.Affinity == nil {
+			remote.Affinity = &corev1.Affinity{
+				NodeAffinity: nodeAffinity,
+			}
+			return
+		}
+
+		if remote.Affinity.NodeAffinity == nil {
+			remote.Affinity.NodeAffinity = nodeAffinity
+			return
+		}
+
+		if nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution != nil {
+			if remote.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution == nil {
+				remote.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution =
+					nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution
+			} else {
+				remote.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms = append(
+					remote.Affinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms,
+					nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms...)
+			}
+		}
+
+		remote.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution =
+			append(remote.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution,
+				nodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution...)
+	}
+}
+
 // FilterAntiAffinityLabels filters the label keys which are used to implement the anti-affinity constraints, based on the specified whitelist.
 func FilterAntiAffinityLabels(labels map[string]string, whitelist string) map[string]string {
 	if whitelist != "" {

pkg/virtualKubelet/forge/pods.go

@@ -81,6 +81,8 @@ type InitConfig struct {
 
 	LabelsNotReflected      []string
 	AnnotationsNotReflected []string
+
+	OffloadingPatch *vkalpha1.OffloadingPatch
 }
 
 // LiqoProvider implements the virtual-kubelet provider interface and stores pods in memory.
@@ -134,7 +136,7 @@ func NewLiqoProvider(ctx context.Context, cfg *InitConfig, eb record.EventBroadc
 
 	podreflector := workload.NewPodReflector(cfg.RemoteConfig, remoteMetricsClient, ipamClient, &podReflectorConfig, cfg.ReflectorsConfigs[generic.Pod])
 	reflectionManager := manager.New(localClient, remoteClient, localLiqoClient, remoteLiqoClient,
-		cfg.InformerResyncPeriod, eb, cfg.LabelsNotReflected, cfg.AnnotationsNotReflected).
+		cfg.InformerResyncPeriod, eb, cfg.LabelsNotReflected, cfg.AnnotationsNotReflected, cfg.OffloadingPatch).
 		With(podreflector).
 		With(exposition.NewServiceReflector(cfg.ReflectorsConfigs[generic.Service], cfg.EnableLoadBalancer, cfg.RemoteRealLoadBalancerClassName)).
 		With(exposition.NewIngressReflector(cfg.ReflectorsConfigs[generic.Ingress], cfg.EnableIngress, cfg.RemoteRealIngressClassName)).

pkg/virtualKubelet/provider/provider.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/utils/trace"
 
+	virtualkubeletv1alpha1 "github.com/liqotech/liqo/apis/virtualkubelet/v1alpha1"
 	liqoclient "github.com/liqotech/liqo/pkg/client/clientset/versioned"
 	liqoinformers "github.com/liqotech/liqo/pkg/client/informers/externalversions"
 	traceutils "github.com/liqotech/liqo/pkg/utils/trace"
@@ -61,7 +62,8 @@ type manager struct {
 
 // New returns a new manager to start the reflection towards a remote cluster.
 func New(local, remote kubernetes.Interface, localLiqo, remoteLiqo liqoclient.Interface, resync time.Duration,
-	eb record.EventBroadcaster, labelsNotReflected, annotationsNotReflected []string) Manager {
+	eb record.EventBroadcaster, labelsNotReflected, annotationsNotReflected []string,
+	offloadingPatch *virtualkubeletv1alpha1.OffloadingPatch) Manager {
 	// Configure the field selector to retrieve only the pods scheduled on the current virtual node.
 	localPodTweakListOptions := func(opts *metav1.ListOptions) {
 		opts.FieldSelector = fields.OneTermEqualSelector("spec.nodeName", forge.LiqoNodeName).String()
@@ -82,7 +84,7 @@ func New(local, remote kubernetes.Interface, localLiqo, remoteLiqo liqoclient.In
 		started: false,
 		stop:    make(map[string]context.CancelFunc),
 
-		forgingOpts: forge.NewForgingOpts(labelsNotReflected, annotationsNotReflected),
+		forgingOpts: forge.NewForgingOpts(labelsNotReflected, annotationsNotReflected, offloadingPatch),
 	}
 }