Admission: support inject autonomy

Signed-off-by: Xiang Dai <long0dai@foxmail.com>

5 files changed, 238 insertions, 72 deletions

diff --git a/build/admission/deployment.yaml b/build/admission/deployment.yaml
index 07688017c..815cef67d 100644
--- a/build/admission/deployment.yaml
+++ b/build/admission/deployment.yaml
@@ -15,6 +15,13 @@ spec:
       labels:
         app: kubeedge-admission
     spec:
+      affinity:
+        nodeAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+            nodeSelectorTerms:
+            - matchExpressions:
+              - key: node-role.kubernetes.io/edge
+                operator: DoesNotExist
       serviceAccount: kubeedge-admission
       containers:
         - args:
diff --git a/cloud/pkg/admissioncontroller/admission.go b/cloud/pkg/admissioncontroller/admission.go
index 630058e4b..6a808dc3d 100644
--- a/cloud/pkg/admissioncontroller/admission.go
+++ b/cloud/pkg/admissioncontroller/admission.go
@@ -10,13 +10,11 @@ import (
 	admissionv1beta1 "k8s.io/api/admission/v1beta1"
 	admissionregistrationv1beta1 "k8s.io/api/admissionregistration/v1beta1"
 	corev1 "k8s.io/api/core/v1"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/kubernetes"
-	admissionregistrationv1beta1client "k8s.io/client-go/kubernetes/typed/admissionregistration/v1beta1"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
@@ -32,6 +30,10 @@ const (
 	ValidateDeviceModelWebhookName  = "validatedevicemodel.kubeedge.io"
 	ValidateRuleWebhookName         = "validatedrule.kubeedge.io"
 	ValidateRuleEndpointWebhookName = "validatedruleendpoint.kubeedge.io"
+	OfflineMigrationConfigName      = "mutate-offlinemigration"
+	OfflineMigrationWebhookName     = "mutateofflinemigration.kubeedge.io"
+
+	AutonomyLabel = "app-offline.kubeedge.io=autonomy"
 )
 
 var scheme = runtime.NewScheme()
@@ -95,6 +97,7 @@ func Run(opt *options.AdmissionOptions) {
 	http.HandleFunc("/devicemodels", serveDeviceModel)
 	http.HandleFunc("/rules", serveRule)
 	http.HandleFunc("/ruleendpoints", serveRuleEndpoint)
+	http.HandleFunc("/offlinemigration", serveOfflineMigration)
 
 	server := &http.Server{
 		Addr:      fmt.Sprintf(":%v", opt.Port),
@@ -222,31 +225,49 @@ func (ac *AdmissionController) registerWebhooks(opt *options.AdmissionOptions, c
 		},
 	}
 
-	return registerValidateWebhook(ac.Client.AdmissionregistrationV1beta1().ValidatingWebhookConfigurations(),
-		[]admissionregistrationv1beta1.ValidatingWebhookConfiguration{deviceModelCRDWebhook})
-}
+	objectSelector, err := metav1.ParseToLabelSelector(AutonomyLabel)
+	if err != nil {
+		return err
+	}
+	offlineMigrationWebhook := admissionregistrationv1beta1.MutatingWebhookConfiguration{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: OfflineMigrationConfigName,
+		},
+		Webhooks: []admissionregistrationv1beta1.MutatingWebhook{
+			{
+				Name:           OfflineMigrationWebhookName,
+				ObjectSelector: objectSelector,
+				Rules: []admissionregistrationv1beta1.RuleWithOperations{{
+					Operations: []admissionregistrationv1beta1.OperationType{
+						admissionregistrationv1beta1.Create,
+						admissionregistrationv1beta1.Update,
+					},
+					Rule: admissionregistrationv1beta1.Rule{
+						APIGroups:   []string{""},
+						APIVersions: []string{"v1"},
+						Resources:   []string{"pods"},
+					},
+				}},
+				ClientConfig: admissionregistrationv1beta1.WebhookClientConfig{
+					Service: &admissionregistrationv1beta1.ServiceReference{
+						Namespace: opt.AdmissionServiceNamespace,
+						Name:      opt.AdmissionServiceName,
+						Path:      strPtr("/offlinemigration"),
+						Port:      &opt.Port,
+					},
+					CABundle: cabundle,
+				},
+				FailurePolicy: &ignorePolicy,
+			},
+		},
+	}
 
-func registerValidateWebhook(client admissionregistrationv1beta1client.ValidatingWebhookConfigurationInterface,
-	webhooks []admissionregistrationv1beta1.ValidatingWebhookConfiguration) error {
-	for _, hook := range webhooks {
-		existing, err := client.Get(context.Background(), hook.Name, metav1.GetOptions{})
-		if err != nil && !apierrors.IsNotFound(err) {
-			return err
-		}
-		if err == nil && existing != nil {
-			existing.Webhooks = hook.Webhooks
-			klog.Infof("Updating ValidatingWebhookConfiguration: %v", hook.Name)
-			if _, err := client.Update(context.Background(), existing, metav1.UpdateOptions{}); err != nil {
-				return err
-			}
-		} else {
-			klog.Infof("Creating ValidatingWebhookConfiguration: %v", hook.Name)
-			if _, err := client.Create(context.Background(), &hook, metav1.CreateOptions{}); err != nil {
-				return err
-			}
-		}
+	if err := registerValidateWebhook(ac.Client.AdmissionregistrationV1beta1().ValidatingWebhookConfigurations(),
+		[]admissionregistrationv1beta1.ValidatingWebhookConfiguration{deviceModelCRDWebhook}); err != nil {
+		return err
 	}
-	return nil
+	return registerMutatingWebhook(ac.Client.AdmissionregistrationV1beta1().MutatingWebhookConfigurations(),
+		[]admissionregistrationv1beta1.MutatingWebhookConfiguration{offlineMigrationWebhook})
 }
 
 func (ac *AdmissionController) getRuleEndpoint(namespace, name string) (*v1.RuleEndpoint, error) {
diff --git a/cloud/pkg/admissioncontroller/admit_devicemodel.go b/cloud/pkg/admissioncontroller/admit_devicemodel.go
index 503d17d25..6eb666bdf 100644
--- a/cloud/pkg/admissioncontroller/admit_devicemodel.go
+++ b/cloud/pkg/admissioncontroller/admit_devicemodel.go
@@ -1,8 +1,6 @@
 package admissioncontroller
 
 import (
-	"encoding/json"
-	"io/ioutil"
 	"net/http"
 	"strings"
 
@@ -13,51 +11,6 @@ import (
 	devicesv1alpha2 "github.com/kubeedge/kubeedge/cloud/pkg/apis/devices/v1alpha2"
 )
 
-// admitFunc is the type we use for all of our validators and mutators
-type admitFunc func(admissionv1beta1.AdmissionReview) *admissionv1beta1.AdmissionResponse
-
-func serve(w http.ResponseWriter, r *http.Request, admit admitFunc) {
-	var body []byte
-	if r.Body != nil {
-		if data, err := ioutil.ReadAll(r.Body); err == nil {
-			body = data
-		}
-	}
-
-	// verify the content type is accurate
-	contentType := r.Header.Get("Content-Type")
-	if contentType != "application/json" {
-		klog.Fatalf("contentType=%s, expect application/json", contentType)
-		return
-	}
-
-	// The AdmissionReview that was sent to the webhook
-	requestedAdmissionReview := admissionv1beta1.AdmissionReview{}
-
-	// The AdmissionReview that will be returned
-	responseAdmissionReview := admissionv1beta1.AdmissionReview{}
-
-	deserializer := codecs.UniversalDeserializer()
-	if _, _, err := deserializer.Decode(body, nil, &requestedAdmissionReview); err != nil {
-		klog.Fatalf("decode failed with error: %v", err)
-		responseAdmissionReview.Response = toAdmissionResponse(err)
-	} else {
-		responseAdmissionReview.Response = admit(requestedAdmissionReview)
-	}
-
-	// Return the same UID
-	responseAdmissionReview.Response.UID = requestedAdmissionReview.Request.UID
-	klog.Infof("sending response: %v", responseAdmissionReview.Response)
-
-	respBytes, err := json.Marshal(responseAdmissionReview)
-	if err != nil {
-		klog.Fatalf("cannot marshal to a valid response %v", err)
-	}
-	if _, err := w.Write(respBytes); err != nil {
-		klog.Fatalf("cannot write response %v", err)
-	}
-}
-
 func admitDeviceModel(review admissionv1beta1.AdmissionReview) *admissionv1beta1.AdmissionResponse {
 	reviewResponse := admissionv1beta1.AdmissionResponse{}
 	reviewResponse.Allowed = true
diff --git a/cloud/pkg/admissioncontroller/common.go b/cloud/pkg/admissioncontroller/common.go
new file mode 100644
index 000000000..eb8965e41
--- /dev/null
+++ b/cloud/pkg/admissioncontroller/common.go
@@ -0,0 +1,106 @@
+package admissioncontroller
+
+import (
+	"context"
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+
+	admissionv1beta1 "k8s.io/api/admission/v1beta1"
+	admissionregistrationv1beta1 "k8s.io/api/admissionregistration/v1beta1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	admissionregistrationv1beta1client "k8s.io/client-go/kubernetes/typed/admissionregistration/v1beta1"
+	"k8s.io/klog/v2"
+)
+
+func registerValidateWebhook(client admissionregistrationv1beta1client.ValidatingWebhookConfigurationInterface,
+	webhooks []admissionregistrationv1beta1.ValidatingWebhookConfiguration) error {
+	for _, hook := range webhooks {
+		existing, err := client.Get(context.Background(), hook.Name, metav1.GetOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			return err
+		}
+		if err == nil && existing != nil {
+			existing.Webhooks = hook.Webhooks
+			klog.Infof("Updating ValidatingWebhookConfiguration: %v", hook.Name)
+			if _, err := client.Update(context.Background(), existing, metav1.UpdateOptions{}); err != nil {
+				return err
+			}
+		} else {
+			klog.Infof("Creating ValidatingWebhookConfiguration: %v", hook.Name)
+			if _, err := client.Create(context.Background(), &hook, metav1.CreateOptions{}); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func registerMutatingWebhook(client admissionregistrationv1beta1client.MutatingWebhookConfigurationInterface,
+	webhooks []admissionregistrationv1beta1.MutatingWebhookConfiguration) error {
+	for _, hook := range webhooks {
+		existing, err := client.Get(context.Background(), hook.Name, metav1.GetOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			return err
+		}
+		if err == nil && existing != nil {
+			existing.Webhooks = hook.Webhooks
+			klog.Infof("Updating MutatingWebhookConfiguration: %v", hook.Name)
+			if _, err := client.Update(context.Background(), existing, metav1.UpdateOptions{}); err != nil {
+				return err
+			}
+		} else {
+			klog.Infof("Creating MutatingWebhookConfiguration: %v", hook.Name)
+			if _, err := client.Create(context.Background(), &hook, metav1.CreateOptions{}); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+// hookFunc is the type we use for all of our validators and mutators
+type hookFunc func(admissionv1beta1.AdmissionReview) *admissionv1beta1.AdmissionResponse
+
+func serve(w http.ResponseWriter, r *http.Request, hook hookFunc) {
+	var body []byte
+	if r.Body != nil {
+		if data, err := ioutil.ReadAll(r.Body); err == nil {
+			body = data
+		}
+	}
+
+	// verify the content type is accurate
+	contentType := r.Header.Get("Content-Type")
+	if contentType != "application/json" {
+		klog.Fatalf("contentType=%s, expect application/json", contentType)
+		return
+	}
+
+	// The AdmissionReview that was sent to the webhook
+	requestedAdmissionReview := admissionv1beta1.AdmissionReview{}
+
+	// The AdmissionReview that will be returned
+	responseAdmissionReview := admissionv1beta1.AdmissionReview{}
+
+	deserializer := codecs.UniversalDeserializer()
+	if _, _, err := deserializer.Decode(body, nil, &requestedAdmissionReview); err != nil {
+		klog.Fatalf("decode failed with error: %v", err)
+		responseAdmissionReview.Response = toAdmissionResponse(err)
+	} else {
+		responseAdmissionReview.Response = hook(requestedAdmissionReview)
+	}
+
+	// Return the same UID
+	responseAdmissionReview.Response.UID = requestedAdmissionReview.Request.UID
+	klog.V(4).Infof("sending response: %+v", responseAdmissionReview.Response)
+
+	respBytes, err := json.Marshal(responseAdmissionReview)
+	if err != nil {
+		klog.Fatalf("cannot marshal to a valid response %v", err)
+	}
+	if _, err := w.Write(respBytes); err != nil {
+		klog.Fatalf("cannot write response %v", err)
+	}
+}
diff --git a/cloud/pkg/admissioncontroller/mutate_offlinemigration.go b/cloud/pkg/admissioncontroller/mutate_offlinemigration.go
new file mode 100644
index 000000000..9b549155c
--- /dev/null
+++ b/cloud/pkg/admissioncontroller/mutate_offlinemigration.go
@@ -0,0 +1,79 @@
+package admissioncontroller
+
+import (
+	"encoding/json"
+	"net/http"
+
+	admissionv1beta1 "k8s.io/api/admission/v1beta1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/klog/v2"
+)
+
+const (
+	Exists    = "Exists"
+	NoExecute = "NoExecute"
+)
+
+type patchMapValue struct {
+	Op    string              `json:"op"`
+	Path  string              `json:"path"`
+	Value []map[string]string `json:"value,omitempty"`
+}
+
+func mutateOfflineMigration(review admissionv1beta1.AdmissionReview) *admissionv1beta1.AdmissionResponse {
+	reviewResponse := admissionv1beta1.AdmissionResponse{
+		Allowed: true,
+	}
+
+	var pod corev1.Pod
+	if err := json.Unmarshal(review.Request.Object.Raw, &pod); err != nil {
+		klog.Errorf("Could not unmarshal raw object: %v", err)
+		return toAdmissionResponse(err)
+	}
+
+	payload := generatePatch(pod.Spec.Tolerations)
+
+	if len(payload) == 0 {
+		return &reviewResponse
+	}
+
+	patch, err := json.Marshal(payload)
+	if err != nil {
+		return toAdmissionResponse(err)
+	}
+
+	reviewResponse.Patch = patch
+	pt := admissionv1beta1.PatchTypeJSONPatch
+	reviewResponse.PatchType = &pt
+	return &reviewResponse
+}
+
+func generatePatch(tolerations []corev1.Toleration) []patchMapValue {
+	patch := []patchMapValue{{
+		Op:   "add",
+		Path: "/spec/template/spec/tolerations",
+		Value: []map[string]string{{
+			"key":      corev1.TaintNodeUnreachable,
+			"operator": Exists,
+			"effect":   NoExecute,
+		}},
+	}}
+	if len(tolerations) > 0 {
+		for _, toleration := range tolerations {
+			if toleration.Key == corev1.TaintNodeUnreachable {
+				if toleration.Effect == "NoExecute" &&
+					toleration.Operator == "Exists" && toleration.TolerationSeconds == nil {
+					return nil
+				}
+				toleration.TolerationSeconds = nil
+				patch[0].Op = "replace"
+			}
+		}
+	}
+
+	return patch
+}
+
+func serveOfflineMigration(w http.ResponseWriter, r *http.Request) {
+	serve(w, r, mutateOfflineMigration)
+}