[GitHub] [incubator-uniffle] thousandhu commented on a diff in pull request #188: [ISSUE-48][FEATURE][FOLLOW UP] Add webhook component

[GitBox <gi...@apache.org>]

thousandhu commented on code in PR #188:
URL: https://github.com/apache/incubator-uniffle/pull/188#discussion_r958149253


##########
deploy/kubernetes/operator/pkg/webhook/syncer/syncer.go:
##########
@@ -0,0 +1,249 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 syncer
+
+import (
+	"context"
+	"reflect"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+	admissionv1 "k8s.io/api/admissionregistration/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/klog/v2"
+	"k8s.io/utils/pointer"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+
+	unifflev1alpha1 "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/api/uniffle/v1alpha1"
+	"github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/constants"
+	"github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/utils"
+	webhookconstants "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/webhook/constants"
+)
+
+var _ ConfigSyncer = &configSyncer{}
+
+// ConfigSyncer syncs ValidatingWebhookConfigurations and MutatingWebhookConfigurations.
+type ConfigSyncer interface {
+	manager.Runnable
+}
+
+// NewConfigSyncer creates a ConfigSyncer.
+func NewConfigSyncer(caCert []byte, externalService string,
+	kubeClient kubernetes.Interface) ConfigSyncer {
+	return newConfigSyncer(caCert, externalService, kubeClient)
+}
+
+// newConfigSyncer creates a configSyncer.
+func newConfigSyncer(caCert []byte, externalService string,
+	kubeClient kubernetes.Interface) *configSyncer {
+	return &configSyncer{
+		caCert:          caCert,
+		externalService: externalService,
+		kubeClient:      kubeClient,
+	}
+}
+
+// configSyncer implements the ConfigSyncer interface.
+type configSyncer struct {
+	caCert          []byte
+	externalService string
+	kubeClient      kubernetes.Interface
+}
+
+// Start starts the ConfigSyncer.
+func (cs *configSyncer) Start(ctx context.Context) error {
+	klog.V(2).Info("config syncer started")
+	for {
+		select {
+		case <-ctx.Done():
+			klog.V(3).Info("stop syncing webhook configurations")
+			return nil
+		default:
+		}
+		if err := cs.syncWebhookCfg(); err != nil {
+			klog.Errorf("sync webhook configuration failed: %v", err)
+		}
+		time.Sleep(time.Minute)
+	}
+}
+
+// syncWebhookCfg synchronizes the validatingWebhookConfiguration and mutatingWebhookConfiguration objects.
+func (cs *configSyncer) syncWebhookCfg() error {
+	currentVWC, currentMWC := cs.generateWebhookCfg()
+	eg := errgroup.Group{}
+	eg.Go(func() error {
+		return cs.syncValidatingWebhookCfg(currentVWC)
+	})
+	eg.Go(func() error {
+		return cs.syncMutatingWebhookCfg(currentMWC)
+	})
+	return eg.Wait()
+}
+
+// syncValidatingWebhookCfg synchronizes the validatingWebhookConfiguration object.
+func (cs *configSyncer) syncValidatingWebhookCfg(
+	currentVWC *admissionv1.ValidatingWebhookConfiguration) error {
+	vwc, err := cs.kubeClient.AdmissionregistrationV1().ValidatingWebhookConfigurations().
+		Get(context.Background(), webhookconstants.ComponentName, metav1.GetOptions{})
+	if err != nil {
+		if errors.IsNotFound(err) {
+			_, err = cs.kubeClient.AdmissionregistrationV1().ValidatingWebhookConfigurations().
+				Create(context.Background(), currentVWC, metav1.CreateOptions{})
+		}
+		return err
+	}
+	if reflect.DeepEqual(vwc.Webhooks, currentVWC.Webhooks) {
+		return nil
+	}
+	vwc.Webhooks = currentVWC.Webhooks
+	_, err = cs.kubeClient.AdmissionregistrationV1().ValidatingWebhookConfigurations().
+		Update(context.Background(), vwc, metav1.UpdateOptions{})
+	return err
+}
+
+// syncMutatingWebhookCfg synchronizes the mutatingWebhookConfiguration object.
+func (cs *configSyncer) syncMutatingWebhookCfg(
+	currentMWC *admissionv1.MutatingWebhookConfiguration) error {
+	vwc, err := cs.kubeClient.AdmissionregistrationV1().MutatingWebhookConfigurations().
+		Get(context.Background(), webhookconstants.ComponentName, metav1.GetOptions{})
+	if err != nil {
+		if errors.IsNotFound(err) {
+			_, err = cs.kubeClient.AdmissionregistrationV1().MutatingWebhookConfigurations().
+				Create(context.Background(), currentMWC, metav1.CreateOptions{})
+		}
+		return err
+	}
+	if reflect.DeepEqual(vwc.Webhooks, currentMWC.Webhooks) {
+		return nil
+	}
+	vwc.Webhooks = currentMWC.Webhooks
+	_, err = cs.kubeClient.AdmissionregistrationV1().MutatingWebhookConfigurations().
+		Update(context.Background(), vwc, metav1.UpdateOptions{})
+	return err
+}
+
+// generateWebhookCfg generates the validatingWebhookConfiguration and mutatingWebhookConfiguration objects.
+func (cs *configSyncer) generateWebhookCfg() (

Review Comment:
   miss return type？



##########
deploy/kubernetes/operator/pkg/webhook/inspector/pod.go:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 inspector
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	admissionv1 "k8s.io/api/admission/v1"
+	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/util/sets"
+	"k8s.io/klog/v2"
+
+	unifflev1alpha1 "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/api/uniffle/v1alpha1"
+	"github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/utils"
+	"github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/webhook/util"
+)
+
+// validateDeletingShuffleServer validates the delete operation towards shuffle server pods,
+// and update exclude nodes in configMap.
+func (i *inspector) validateDeletingShuffleServer(ar *admissionv1.AdmissionReview) *admissionv1.AdmissionReview {
+	pod := &corev1.Pod{}
+	if err := json.Unmarshal(ar.Request.OldObject.Raw, pod); err != nil {
+		klog.Errorf("unmarshal object of AdmissionReview (%v) failed: %v",
+			string(ar.Request.Object.Raw), err)
+		return util.AdmissionReviewFailed(ar, err)
+	}
+	rssName := utils.GetRssNameByPod(pod)
+	// allow pods which are not shuffle servers or have been set deletion timestamp.
+	if rssName == "" || !util.NeedInspectPod(pod) {
+		klog.V(4).Infof("ignored non shuffle server or deleting pod: %v->(%+v/%+v/%v)",
+			utils.UniqueName(pod), pod.Labels, pod.Annotations, pod.DeletionTimestamp)
+		return util.AdmissionReviewAllow(ar)
+	}
+	klog.V(4).Infof("check shuffle server pod: %v", utils.BuildShuffleServerKey(pod))
+	rss, err := i.rssLister.RemoteShuffleServices(pod.Namespace).Get(rssName)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return util.AdmissionReviewAllow(ar)
+		}
+		return util.AdmissionReviewFailed(ar, err)
+	}
+	// we can only delete shuffle server pods when rss is in upgrading phase.
+	if rss.Status.Phase != unifflev1alpha1.RSSUpgrading && rss.Status.Phase != unifflev1alpha1.RSSTerminating {
+		message := fmt.Sprintf("can not delete the shuffle server pod (%v) directly",
+			utils.UniqueName(pod))
+		klog.V(4).Info(message)
+		return util.AdmissionReviewForbidden(ar, message)
+	}
+
+	// when the rss uses specific upgrade mode, we need to check whether the pod is specific.
+	if rss.Spec.ShuffleServer.UpgradeStrategy.Type == unifflev1alpha1.SpecificUpgrade {
+		specificNames := rss.Spec.ShuffleServer.UpgradeStrategy.SpecificNames
+		isSpecific := false
+		for _, name := range specificNames {
+			if name == pod.Name {
+				isSpecific = true
+				break
+			}
+		}
+		if !isSpecific {
+			message := fmt.Sprintf("can not delete the shuffle server pod (%v) which is not specific",
+				utils.UniqueName(pod))
+			klog.V(4).Info(message)
+			return util.AdmissionReviewForbidden(ar, message)
+		}
+	}
+
+	// update targetKeys field in status of rss and exclude nodes in configMap used by coordinators.
+	if err = i.updateTargetKeysAndExcludeNodes(rss, pod); err != nil {
+		return util.AdmissionReviewFailed(ar, err)
+	}
+
+	// check whether the shuffle server pod can be deleted.
+	if i.ignoreLastApps || util.HasZeroApps(pod) {
+		klog.V(3).Infof("shuffle server pod (%v) will be deleted", utils.BuildShuffleServerKey(pod))
+		return util.AdmissionReviewAllow(ar)
+	}
+	message := "can not delete the shuffle server: " + utils.GetShuffleServerNode(pod)

Review Comment:
   Should we add the reason such as "there are apps still running in the node" into message?



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org