Skip to content

Commit

Permalink
Merge pull request #329 from wangweizZZ/main
Browse files Browse the repository at this point in the history
feature: support pod convert
  • Loading branch information
duanmengkk committed Jan 8, 2024
2 parents 5b97084 + 7fc5700 commit c2f9f1d
Show file tree
Hide file tree
Showing 8 changed files with 302 additions and 2 deletions.
56 changes: 56 additions & 0 deletions deploy/crds/kosmos.io_podconvertpolicies.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -1012,6 +1012,62 @@ spec:
required:
- convertType
type: object
tolerationConverter:
description: TolerationConverter used to modify the pod's Tolerations
when pod synced to leaf cluster
properties:
convertType:
description: ConvertType if the operation type when convert
pod from root cluster to leaf cluster.
enum:
- add
- remove
- replace
type: string
tolerations:
items:
description: The pod this Toleration is attached to tolerates
any taint that matches the triple <key,value,effect> using
the matching operator <operator>.
properties:
effect:
description: Effect indicates the taint effect to match.
Empty means match all taint effects. When specified,
allowed values are NoSchedule, PreferNoSchedule and
NoExecute.
type: string
key:
description: Key is the taint key that the toleration
applies to. Empty means match all taint keys. If the
key is empty, operator must be Exists; this combination
means to match all values and all keys.
type: string
operator:
description: Operator represents a key's relationship
to the value. Valid operators are Exists and Equal.
Defaults to Equal. Exists is equivalent to wildcard
for value, so that a pod can tolerate all taints of
a particular category.
type: string
tolerationSeconds:
description: TolerationSeconds represents the period
of time the toleration (which must be of effect NoExecute,
otherwise this field is ignored) tolerates the taint.
By default, it is not set, which means tolerate the
taint forever (do not evict). Zero and negative values
will be treated as 0 (evict immediately) by the system.
format: int64
type: integer
value:
description: Value is the taint value the toleration
matches to. If the operator is Exists, the value should
be empty, otherwise just a regular string.
type: string
type: object
type: array
required:
- convertType
type: object
topologySpreadConstraintsConverter:
description: TopologySpreadConstraintsConverter used to modify
the pod's TopologySpreadConstraints when pod synced to leaf
Expand Down
2 changes: 2 additions & 0 deletions pkg/apis/kosmos/v1alpha1/podconvertpolicy_types.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ type Converters struct {
// +optional
NodeSelectorConverter *NodeSelectorConverter `json:"nodeSelectorConverter,omitempty"`
// +optional
TolerationConverter *TolerationConverter `json:"tolerationConverter,omitempty"`
// +optional
AffinityConverter *AffinityConverter `json:"affinityConverter,omitempty"`
// +optional
TopologySpreadConstraintsConverter *TopologySpreadConstraintsConverter `json:"topologySpreadConstraintsConverter,omitempty"`
Expand Down
5 changes: 5 additions & 0 deletions pkg/apis/kosmos/v1alpha1/zz_generated.deepcopy.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ import (
"github.com/kosmos.io/kosmos/pkg/clustertree/cluster-manager/extensions/daemonset"
leafUtils "github.com/kosmos.io/kosmos/pkg/clustertree/cluster-manager/utils"
"github.com/kosmos.io/kosmos/pkg/utils"
"github.com/kosmos.io/kosmos/pkg/utils/convertpolicy"
"github.com/kosmos.io/kosmos/pkg/utils/podutils"
)

Expand Down Expand Up @@ -702,6 +703,37 @@ func (r *RootPodReconciler) createVolumes(ctx context.Context, lr *leafUtils.Lea
return nil
}

// mutatePod modify pod by matching policy
func (r *RootPodReconciler) mutatePod(ctx context.Context, pod *corev1.Pod, nodeName string) error {
klog.V(4).Infof("Converting pod %v/%+v", pod.Namespace, pod.Name)

podConvertPolicyList := &kosmosv1alpha1.PodConvertPolicyList{}
err := r.Client.List(ctx, podConvertPolicyList, &client.ListOptions{
Namespace: pod.Namespace,
})
if err != nil {
return fmt.Errorf("list convert policy error: %v", err)
}
if len(podConvertPolicyList.Items) <= 0 {
// no matched policy, skip
return nil
}

rootNode := &corev1.Node{}
err = r.Client.Get(ctx, types.NamespacedName{Name: nodeName}, rootNode)
if err != nil {
return fmt.Errorf("get node error: %v, nodeName: %s", err, pod.Spec.NodeName)
}

matchedPolicy, err := convertpolicy.GetMatchPodConvertPolicy(*podConvertPolicyList, pod.Labels, rootNode.Labels)
if err != nil {
return fmt.Errorf("get convert policy error: %v", err)
}
podutils.ConvertPod(pod, matchedPolicy)
klog.V(4).Infof("Convert pod %v/%+v success", pod.Namespace, pod.Name)
return nil
}

func (r *RootPodReconciler) CreatePodInLeafCluster(ctx context.Context, lr *leafUtils.LeafResource, pod *corev1.Pod, nodeSelector kosmosv1alpha1.NodeSelector) error {
if err := podutils.PopulateEnvironmentVariables(ctx, pod, r.envResourceManager); err != nil {
// span.SetStatus(err)
Expand All @@ -716,6 +748,11 @@ func (r *RootPodReconciler) CreatePodInLeafCluster(ctx context.Context, lr *leaf
basicPod := podutils.FitPod(pod, lr.IgnoreLabels, clusterNodeInfo.LeafMode, nodeSelector)
klog.V(4).Infof("Creating pod %v/%+v", pod.Namespace, pod.Name)

err := r.mutatePod(ctx, basicPod, pod.Spec.NodeName)
if err != nil {
klog.Errorf("Converting pod error: %v", err)
}

// create ns
ns := &corev1.Namespace{}
nsKey := types.NamespacedName{
Expand Down Expand Up @@ -759,7 +796,7 @@ func (r *RootPodReconciler) CreatePodInLeafCluster(ctx context.Context, lr *leaf

klog.V(4).Infof("Creating pod %+v", basicPod)

err := lr.Client.Create(ctx, basicPod)
err = lr.Client.Create(ctx, basicPod)
if err != nil {
return fmt.Errorf("could not create pod: %v", err)
}
Expand Down
7 changes: 6 additions & 1 deletion pkg/generated/openapi/zz_generated.openapi.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions pkg/utils/constants.go
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,7 @@ const (
KosmosGlobalLabel = "kosmos.io/global"
KosmosSelectorKey = "kosmos.io/cluster-selector"
KosmosTrippedLabels = "kosmos-io/tripped"
KosmosConvertLabels = "kosmos-io/convert-policy"
KosmosPvcLabelSelector = "kosmos-io/label-selector"
KosmosExcludeNodeLabel = "kosmos.io/exclude"
KosmosExcludeNodeValue = "true"
Expand Down
41 changes: 41 additions & 0 deletions pkg/utils/convertpolicy/pod.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
package convertpolicy

import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"

kosmosv1alpha1 "github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
)

// GetMatchPodConvertPolicy returns the PodConvertPolicies matching label selector
func GetMatchPodConvertPolicy(policies kosmosv1alpha1.PodConvertPolicyList, podLabels map[string]string, nodeLabels map[string]string) ([]kosmosv1alpha1.PodConvertPolicy, error) {
matched := make([]kosmosv1alpha1.PodConvertPolicy, 0)

var podSelector, nodeSelector labels.Selector
var err error
for _, po := range policies.Items {
spec := po.Spec
podSelector, err = metav1.LabelSelectorAsSelector(&spec.LabelSelector)
if err != nil {
return nil, err
}
if !podSelector.Matches(labels.Set(podLabels)) {
continue
}

if spec.LeafNodeSelector == nil {
// matches all leafNode.
nodeSelector = labels.Everything()
} else {
if nodeSelector, err = metav1.LabelSelectorAsSelector(spec.LeafNodeSelector); err != nil {
return nil, err
}
}
if !nodeSelector.Matches(labels.Set(nodeLabels)) {
continue
}

matched = append(matched, po)
}
return matched, nil
}
153 changes: 153 additions & 0 deletions pkg/utils/podutils/pod_convert.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,153 @@
package podutils

import (
corev1 "k8s.io/api/core/v1"
"k8s.io/klog"

kosmosv1alpha1 "github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
"github.com/kosmos.io/kosmos/pkg/utils"
)

// ConvertPod perform all conversions
func ConvertPod(pod *corev1.Pod, policies []kosmosv1alpha1.PodConvertPolicy) {
if len(policies) <= 0 {
return
}

var choose *kosmosv1alpha1.PodConvertPolicy
// current, use the first non-empty matching policy
for idx, po := range policies {
if po.Spec.Converters != nil {
choose = &policies[idx]
break
}
}
if choose == nil {
return
}
klog.V(4).Infof("Convert pod %v/%+v, policy: %s", pod.Namespace, pod.Name, choose.Name)

converters := choose.Spec.Converters
convertSchedulerName(pod, converters.SchedulerNameConverter)
convertNodeName(pod, converters.NodeNameConverter)
convertNodeSelector(pod, converters.NodeSelectorConverter)
converToleration(pod, converters.TolerationConverter)
convertAffinity(pod, converters.AffinityConverter)
convertTopologySpreadConstraints(pod, converters.TopologySpreadConstraintsConverter)

pod.Annotations[utils.KosmosConvertLabels] = choose.Name
}

func convertSchedulerName(pod *corev1.Pod, converter *kosmosv1alpha1.SchedulerNameConverter) {
if converter == nil {
return
}

switch converter.ConvertType {
case kosmosv1alpha1.Add:
if pod.Spec.SchedulerName == "" {
pod.Spec.SchedulerName = converter.SchedulerName
}
case kosmosv1alpha1.Remove:
pod.Spec.SchedulerName = ""
case kosmosv1alpha1.Replace:
pod.Spec.SchedulerName = converter.SchedulerName
default:
klog.Warningf("Skip other convert type, SchedulerName: %s", converter.ConvertType)
}
}

func convertNodeName(pod *corev1.Pod, converter *kosmosv1alpha1.NodeNameConverter) {
if converter == nil {
return
}

switch converter.ConvertType {
case kosmosv1alpha1.Add:
if pod.Spec.NodeName == "" {
pod.Spec.NodeName = converter.NodeName
}
case kosmosv1alpha1.Remove:
pod.Spec.NodeName = ""
case kosmosv1alpha1.Replace:
pod.Spec.NodeName = converter.NodeName
default:
klog.Warningf("Skip other convert type, NodeName: %s", converter.ConvertType)
}
}

func converToleration(pod *corev1.Pod, conveter *kosmosv1alpha1.TolerationConverter) {
if conveter == nil {
return
}

switch conveter.ConvertType {
case kosmosv1alpha1.Add:
if pod.Spec.Tolerations == nil {
pod.Spec.Tolerations = conveter.Tolerations
}
case kosmosv1alpha1.Remove:
pod.Spec.Tolerations = nil
case kosmosv1alpha1.Replace:
pod.Spec.Tolerations = conveter.Tolerations
default:
klog.Warningf("Skip other convert type, Tolerations: %s", conveter.ConvertType)
}
}

func convertNodeSelector(pod *corev1.Pod, converter *kosmosv1alpha1.NodeSelectorConverter) {
if converter == nil {
return
}

switch converter.ConvertType {
case kosmosv1alpha1.Add:
if pod.Spec.NodeSelector == nil {
pod.Spec.NodeSelector = converter.NodeSelector
}
case kosmosv1alpha1.Remove:
pod.Spec.NodeSelector = nil
case kosmosv1alpha1.Replace:
pod.Spec.NodeSelector = converter.NodeSelector
default:
klog.Warningf("Skip other convert type, NodeSelector: %s", converter.ConvertType)
}
}

func convertAffinity(pod *corev1.Pod, converter *kosmosv1alpha1.AffinityConverter) {
if converter == nil {
return
}

switch converter.ConvertType {
case kosmosv1alpha1.Add:
if pod.Spec.Affinity == nil {
pod.Spec.Affinity = converter.Affinity
}
case kosmosv1alpha1.Remove:
pod.Spec.Affinity = nil
case kosmosv1alpha1.Replace:
pod.Spec.Affinity = converter.Affinity
default:
klog.Warningf("Skip other convert type, Affinity: %s", converter.ConvertType)
}
}

func convertTopologySpreadConstraints(pod *corev1.Pod, converter *kosmosv1alpha1.TopologySpreadConstraintsConverter) {
if converter == nil {
return
}

switch converter.ConvertType {
case kosmosv1alpha1.Add:
if pod.Spec.Affinity == nil {
pod.Spec.TopologySpreadConstraints = converter.TopologySpreadConstraints
}
case kosmosv1alpha1.Remove:
pod.Spec.TopologySpreadConstraints = nil
case kosmosv1alpha1.Replace:
pod.Spec.TopologySpreadConstraints = converter.TopologySpreadConstraints
default:
klog.Warningf("Skip other convert type, TopologySpreadConstraints: %s", converter.ConvertType)
}
}

0 comments on commit c2f9f1d

Please sign in to comment.