feat: register ruleset webhook (#35)

* feat: register ruleset webhook

* ruleset mutating with default value

apis/apps/v1alpha1/ruleset_types.go

@@ -25,7 +25,7 @@ import (
 // RuleSetSpec defines the desired state of RuleSet
 type RuleSetSpec struct {
 	// Selector select the targets controlled by ruleset
-	Selector metav1.LabelSelector `json:"selector,omitempty"`
+	Selector *metav1.LabelSelector `json:"selector,omitempty"`
 
 	// Rules is a set of rules that need to be checked in certain situations
 	Rules []RuleSetRule `json:"rules,omitempty"`
@@ -113,6 +113,10 @@ const (
 	// Fail means that an error calling the webhook causes the admission to fail.
 	Fail FailurePolicyType = "Fail"
 )
+const (
+	DefaultWebhookInterval = int64(5)
+	DefaultWebhookTimeout  = int64(60)
+)
 
 // ResourceParameter is representing the request body of resource parameter
 type ResourceParameter struct {

apis/apps/v1alpha1/well_known_annotations.go

@@ -19,7 +19,11 @@ package v1alpha1
 const (
 	PodAvailableConditionsAnnotation = "pod.kusionstack.io/available-conditions" // indicate the available conditions of a pod
 
-	AnnotationPodSkipRuleConditions = "ruleset.kusionstack.io/skip-rule-conditions"
-
 	LastPodStatusAnnotationKey = "collaset.kusionstack.io/last-pod-status"
 )
+
+// RuleSet Annotation
+const (
+	AnnotationRuleSets              = "ruleset.kusionstack.io/rulesets"
+	AnnotationPodSkipRuleConditions = "ruleset.kusionstack.io/skip-rule-conditions"
+)

pkg/controllers/ruleset/eventhandler.go

@@ -91,7 +91,7 @@ func involvedRuleSets(c client.Client, obj client.Object) ([]*appsv1alpha1.RuleS
 		return ruleSets, err
 	}
 	for i, rs := range ruleSetList.Items {
-		selector, err := metav1.LabelSelectorAsSelector(&rs.Spec.Selector)
+		selector, err := metav1.LabelSelectorAsSelector(rs.Spec.Selector)
 		if err != nil {
 			return ruleSets, err
 		}

pkg/controllers/ruleset/ruleset_controller.go

@@ -123,7 +123,7 @@ func (r *RuleSetReconciler) Reconcile(ctx context.Context, request reconcile.Req
 		return reconcile.Result{}, nil
 	}
 
-	selector, _ := metav1.LabelSelectorAsSelector(&ruleSet.Spec.Selector)
+	selector, _ := metav1.LabelSelectorAsSelector(ruleSet.Spec.Selector)
 	selectedPods := &corev1.PodList{}
 	if err := r.List(context.TODO(), selectedPods, &client.ListOptions{Namespace: ruleSet.Namespace, LabelSelector: selector}); err != nil {
 		r.Error(err, fmt.Sprintf("fail to list pods by ruleset %s", request.NamespacedName.String()))
@@ -133,7 +133,7 @@ func (r *RuleSetReconciler) Reconcile(ctx context.Context, request reconcile.Req
 	// Delete
 	if ruleSet.DeletionTimestamp != nil {
 		if _, find := ruleSet.Labels[rulesetTerminatingLabel]; find || !r.hasRunningPod(selectedPods) {
-			if err := r.cleanUpPod(ruleSet); err != nil {
+			if err := r.cleanUpRuleSetPods(ctx, ruleSet); err != nil {
 				return reconcile.Result{}, err
 			}
 			return reconcile.Result{}, utils.RemoveFinalizer(ctx, r.Client, ruleSet, cleanUpFinalizer)
@@ -164,20 +164,20 @@ func (r *RuleSetReconciler) Reconcile(ctx context.Context, request reconcile.Req
 			continue
 		}
 
-		if err := r.updateRuleSetOnPod(ruleSet.Name,
-			&corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ruleSet.Namespace}},
-			rulesetutils.MoveRulesetAnno); err != nil {
+		if _, err := r.updateRuleSetOnPod(ctx, ruleSet.Name, name, ruleSet.Namespace, rulesetutils.MoveRulesetAnno); err != nil {
 			r.Info(fmt.Sprintf("fail to remove ruleset on pod %s, %v", name, err))
 			return result, err
 		}
 	}
 
 	// own selected pods
 	for name, pod := range targetPods {
-		if err := r.updateRuleSetOnPod(ruleSet.Name, pod, rulesetutils.AddRuleSetAnno); err != nil {
+		newPod, err := r.updateRuleSetOnPod(ctx, ruleSet.Name, pod.Name, pod.Namespace, rulesetutils.AddRuleSetAnno)
+		if err != nil {
 			r.Info(fmt.Sprintf("fail to add ruleset on pod %s, %v", name, err))
 			return result, err
 		}
+		targetPods[name] = newPod
 	}
 
 	// process rules
@@ -297,35 +297,23 @@ func addToEveryQueue(namespace, name string) {
 	}
 }
 
-func (r *RuleSetReconciler) cleanUpPod(ruleSet *appsv1alpha1.RuleSet) error {
+func (r *RuleSetReconciler) cleanUpRuleSetPods(ctx context.Context, ruleSet *appsv1alpha1.RuleSet) error {
 	for _, name := range ruleSet.Status.Targets {
-		pod := &corev1.Pod{}
-		if err := r.Get(context.TODO(), types.NamespacedName{Namespace: ruleSet.Namespace, Name: name}, pod); err != nil {
-			if errors.IsNotFound(err) {
-				continue
-			}
-			return err
-		}
-		if err := r.updateRuleSetOnPod(ruleSet.Name, pod, rulesetutils.MoveRulesetAnno); err != nil {
+		if _, err := r.updateRuleSetOnPod(ctx, ruleSet.Name, name, ruleSet.Namespace, rulesetutils.MoveRulesetAnno); err != nil && !errors.IsNotFound(err) {
 			return fmt.Errorf("fail to remove RuleSet %s on pod %s: %v", utils.ObjectKey(ruleSet), name, err)
 		}
 	}
 	return nil
 }
 
-func (r *RuleSetReconciler) updateRuleSetOnPod(ruleSet string, pod *corev1.Pod, fn func(pod *corev1.Pod, name string) bool) error {
-	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		if err := r.Get(context.TODO(), types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}, pod); err != nil {
-			if errors.IsNotFound(err) {
-				return nil
-			}
+func (r *RuleSetReconciler) updateRuleSetOnPod(ctx context.Context, ruleSet, name, namespace string, fn func(pod *corev1.Pod, name string) bool) (*corev1.Pod, error) {
+	pod := &corev1.Pod{}
+	return pod, retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		if err := r.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, pod); err != nil {
 			return err
 		}
 		if fn(pod, ruleSet) {
-			err := r.Update(context.TODO(), pod)
-			if errors.IsNotFound(err) {
-				return nil
-			}
+			return r.Update(ctx, pod)
 		}
 		return nil
 	})

pkg/controllers/ruleset/ruleset_controller_test.go

@@ -58,7 +58,7 @@ func TestRuleset(t *testing.T) {
 			Namespace: "default",
 		},
 		Spec: appsv1alpha1.RuleSetSpec{
-			Selector: metav1.LabelSelector{
+			Selector: &metav1.LabelSelector{
 				MatchLabels: map[string]string{
 					"test": "gen",
 				},

pkg/controllers/ruleset/utils/annotation.go

@@ -26,10 +26,6 @@ import (
 	appsv1alpha1 "kusionstack.io/kafed/apis/apps/v1alpha1"
 )
 
-const (
-	AnnotationRuleSets = "kafed.kusionstack.io/rulesets"
-)
-
 func HasSkipRule(po *corev1.Pod, ruleName string) (bool, error) {
 	if po.Annotations == nil || len(po.Annotations[appsv1alpha1.AnnotationPodSkipRuleConditions]) == 0 {
 		return false, nil
@@ -58,13 +54,13 @@ func MoveRulesetAnno(po *corev1.Pod, rulesetName string) bool {
 	if po.Annotations == nil {
 		return false
 	}
-	val, ok := po.Annotations[AnnotationRuleSets]
+	val, ok := po.Annotations[appsv1alpha1.AnnotationRuleSets]
 	if !ok {
 		return false
 	}
 	state := RuleSets{}
 	if err := json.Unmarshal([]byte(val), &state); err != nil {
-		klog.Errorf("fail to unmarshal anno %s=%s, %v", AnnotationRuleSets, val, err)
+		klog.Errorf("fail to unmarshal anno %s=%s, %v", appsv1alpha1.AnnotationRuleSets, val, err)
 		return false
 	}
 	for i, name := range state {
@@ -83,15 +79,15 @@ func AddRuleSetAnno(po *corev1.Pod, rulesetName string) bool {
 	if po.Annotations == nil {
 		po.Annotations = map[string]string{}
 	}
-	val, ok := po.Annotations[AnnotationRuleSets]
+	val, ok := po.Annotations[appsv1alpha1.AnnotationRuleSets]
 	if !ok {
-		po.Annotations[AnnotationRuleSets] = annoVal(rulesetName)
+		po.Annotations[appsv1alpha1.AnnotationRuleSets] = annoVal(rulesetName)
 		return true
 	}
 	rs := &RuleSets{}
 	if err := json.Unmarshal([]byte(val), rs); err != nil {
-		klog.Errorf("fail to unmarshal anno %s=%s, %v", AnnotationRuleSets, val, err)
-		po.Annotations[AnnotationRuleSets] = annoVal(rulesetName)
+		klog.Errorf("fail to unmarshal anno %s=%s, %v", appsv1alpha1.AnnotationRuleSets, val, err)
+		po.Annotations[appsv1alpha1.AnnotationRuleSets] = annoVal(rulesetName)
 		return true
 	}
 	for _, name := range *rs {
@@ -101,7 +97,7 @@ func AddRuleSetAnno(po *corev1.Pod, rulesetName string) bool {
 	}
 	*rs = append(*rs, rulesetName)
 	bt, _ := json.Marshal(rs)
-	po.Annotations[AnnotationRuleSets] = string(bt)
+	po.Annotations[appsv1alpha1.AnnotationRuleSets] = string(bt)
 	return true
 }
 
@@ -110,7 +106,7 @@ func GetRuleSets(item client.Object) (result []string) {
 	if item.GetAnnotations() == nil {
 		return result
 	}
-	val, ok := item.GetAnnotations()[AnnotationRuleSets]
+	val, ok := item.GetAnnotations()[appsv1alpha1.AnnotationRuleSets]
 	if !ok || len(val) == 0 {
 		return result
 	}
@@ -131,5 +127,5 @@ func annoVal(names ...string) string {
 
 func setRulesetAnno(po *corev1.Pod, rs RuleSets) {
 	val, _ := json.Marshal(rs)
-	po.Annotations[AnnotationRuleSets] = string(val)
+	po.Annotations[appsv1alpha1.AnnotationRuleSets] = string(val)
 }

pkg/utils/http/http.go

@@ -100,9 +100,9 @@ func buildReq(method, url string, body interface{}, header map[string]string) (*
 	return req, nil
 }
 
-var DefaultClient = NewSharedClient()
+var DefaultClient = newSharedClient()
 
-func NewSharedClient() *clientSet {
+func newSharedClient() *clientSet {
 	return &clientSet{
 		caClientSet: map[string]*http.Client{},
 		tkClientSet: map[string]*http.Client{},

pkg/webhook/server/generic/generic_webhooks.go

@@ -21,6 +21,7 @@ import (
 
 	webhookdmission "kusionstack.io/kafed/pkg/webhook/admission"
 	"kusionstack.io/kafed/pkg/webhook/server/generic/pod"
+	"kusionstack.io/kafed/pkg/webhook/server/generic/ruleset"
 )
 
 var (
@@ -37,4 +38,7 @@ var ValidatingTypeHandlerMap = map[string]webhookdmission.DispatchHandler{}
 func init() {
 	MutatingTypeHandlerMap["Pod"] = pod.NewMutatingHandler()
 	ValidatingTypeHandlerMap["Pod"] = pod.NewValidatingHandler()
+
+	MutatingTypeHandlerMap["RuleSet"] = ruleset.NewMutatingHandler()
+	ValidatingTypeHandlerMap["RuleSet"] = ruleset.NewValidatingHandler()
 }

pkg/webhook/server/generic/pod/pod_mutating_handler.go

@@ -77,7 +77,7 @@ func (h *MutatingHandler) Handle(ctx context.Context, req admission.Request) (re
 	marshalled, err := json.Marshal(pod)
 	if err != nil {
 		klog.Errorf("marshal Pod failed, %v", err)
-		return admission.Errored(http.StatusBadRequest, err)
+		return admission.Errored(http.StatusInternalServerError, err)
 	}
 
 	return admission.PatchResponseFromRaw(req.AdmissionRequest.Object.Raw, marshalled)

pkg/webhook/server/generic/pod/pod_webhook.go

@@ -24,6 +24,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"kusionstack.io/kafed/pkg/webhook/server/generic/pod/opslifecycle"
+	"kusionstack.io/kafed/pkg/webhook/server/generic/pod/ruleset"
 )
 
 var (
@@ -38,4 +39,5 @@ type AdmissionWebhook interface {
 
 func init() {
 	webhooks = append(webhooks, opslifecycle.New())
+	webhooks = append(webhooks, ruleset.New())
 }

pkg/webhook/server/generic/pod/ruleset/mutating.go

@@ -0,0 +1,29 @@
+/*
+ Copyright 2023 The KusionStack Authors.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package ruleset
+
+import (
+	"context"
+
+	admissionv1 "k8s.io/api/admission/v1"
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func (r *RuleSetWebhook) Mutating(ctx context.Context, c client.Client, oldPod, newPod *corev1.Pod, operation admissionv1.Operation) error {
+	return nil
+}

pkg/webhook/server/generic/pod/ruleset/validating.go

@@ -0,0 +1,51 @@
+/*
+ Copyright 2023 The KusionStack Authors.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package ruleset
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	admissionv1 "k8s.io/api/admission/v1"
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	appsv1alpha1 "kusionstack.io/kafed/apis/apps/v1alpha1"
+	"kusionstack.io/kafed/pkg/utils"
+)
+
+func (r *RuleSetWebhook) Validating(ctx context.Context, c client.Client, oldPod, newPod *corev1.Pod, operation admissionv1.Operation) error {
+	if operation != admissionv1.Update && operation != admissionv1.Create {
+		return nil
+	}
+	if !utils.ControlledByPodOpsLifecycle(newPod) || newPod.Annotations == nil {
+		return nil
+	}
+	ruleSetAnno := newPod.Annotations[appsv1alpha1.AnnotationRuleSets]
+	if operation == admissionv1.Update && ruleSetAnno == oldPod.Annotations[appsv1alpha1.AnnotationRuleSets] {
+		return nil
+	}
+	rs := &RuleSets{}
+	if err := json.Unmarshal([]byte(ruleSetAnno), rs); err != nil {
+		return fmt.Errorf("fail to unmarshal RuleSet annotation %s: %v", ruleSetAnno, err)
+	}
+
+	return nil
+}
+
+type RuleSets []string

pkg/webhook/server/generic/pod/ruleset/webhook.go

@@ -0,0 +1,28 @@
+/*
+ Copyright 2023 The KusionStack Authors.
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package ruleset
+
+func New() *RuleSetWebhook {
+	return &RuleSetWebhook{}
+}
+
+type RuleSetWebhook struct {
+}
+
+func (r *RuleSetWebhook) Name() string {
+	return "PodRuleSetWebhook"
+}