wangao1236 commented on code in PR #214: URL: https://github.com/apache/incubator-uniffle/pull/214#discussion_r974827263
########## deploy/kubernetes/operator/pkg/controller/controller/rss.go: ########## @@ -0,0 +1,925 @@ +/* + * 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 controller + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + appsv1 "k8s.io/api/apps/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/labels" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" + appslister "k8s.io/client-go/listers/apps/v1" + corelisters "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/retry" + "k8s.io/client-go/util/workqueue" + "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/controller/config" + controllerconstants "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/constants" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/sync/coordinator" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/sync/shuffleserver" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/util" + kubeutil "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/util/kubernetes" + propertiestutil "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/util/properties" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/generated/clientset/versioned" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/generated/informers/externalversions" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/generated/listers/uniffle/v1alpha1" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/utils" +) + +var _ RSSController = &rssController{} + +const ( + controllerName = "rss-controller" + appNameIndexer = "AppNameIndexer" +) + +// RSSController is responsible for synchronizing rss objects. +type RSSController interface { + manager.Runnable +} + +// NewRSSController creates a RSSController. +func NewRSSController(cfg *config.Config) RSSController { + return newRSSController(cfg) +} + +// newRSSController creates a rssController. +func newRSSController(cfg *config.Config) *rssController { + rc := &rssController{ + workers: cfg.Workers, + kubeClient: cfg.KubeClient, + rssClient: cfg.RSSClient, + rssInformerFactory: externalversions.NewSharedInformerFactory(cfg.RSSClient, 0), + shuffleServerInformerFactory: utils.BuildShuffleServerInformerFactory(cfg.KubeClient), + coordinatorInformerFactory: utils.BuildCoordinatorInformerFactory(cfg.KubeClient), + rssQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), + "rss-queue"), + shuffleServerQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), + "pod-queue"), + eventRecorder: util.CreateRecorder(cfg.KubeClient, "rss-controller"), + } + + rssInformer := rc.rssInformerFactory.Uniffle().V1alpha1().RemoteShuffleServices().Informer() + rssInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: rc.addRSS, + UpdateFunc: rc.updateRSS, + }) + rc.rssInformer = rssInformer + rc.rssLister = rc.rssInformerFactory.Uniffle().V1alpha1().RemoteShuffleServices().Lister() + + rc.stsInformer = rc.shuffleServerInformerFactory.Apps().V1().StatefulSets().Informer() + rc.stsLister = rc.shuffleServerInformerFactory.Apps().V1().StatefulSets().Lister() + + rc.podInformer = rc.shuffleServerInformerFactory.Core().V1().Pods().Informer() + rc.podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + DeleteFunc: rc.deleteShuffleServer, + }) + if err := rc.podInformer.AddIndexers(cache.Indexers{appNameIndexer: getAppName}); err != nil { + klog.Fatalf("add app name indexer failed: %v", err) + } + rc.podIndexer = rc.podInformer.GetIndexer() + rc.podLister = rc.shuffleServerInformerFactory.Core().V1().Pods().Lister() + + rc.cmInformer = rc.coordinatorInformerFactory.Core().V1().ConfigMaps().Informer() + rc.cmLister = rc.coordinatorInformerFactory.Core().V1().ConfigMaps().Lister() + + return rc +} + +// rssController implements the RSSController interface. +type rssController struct { + workers int + kubeClient kubernetes.Interface + rssClient versioned.Interface + rssInformerFactory externalversions.SharedInformerFactory + rssInformer cache.SharedIndexInformer + rssLister v1alpha1.RemoteShuffleServiceLister + shuffleServerInformerFactory informers.SharedInformerFactory + stsInformer cache.SharedIndexInformer + stsLister appslister.StatefulSetLister + podInformer cache.SharedIndexInformer + podIndexer cache.Indexer + podLister corelisters.PodLister + coordinatorInformerFactory informers.SharedInformerFactory + cmInformer cache.SharedIndexInformer + cmLister corelisters.ConfigMapLister + rssQueue workqueue.RateLimitingInterface + shuffleServerQueue workqueue.RateLimitingInterface + eventRecorder record.EventRecorder +} + +// Start starts the RSSController. +func (r *rssController) Start(ctx context.Context) error { + klog.V(2).Infof("%v is starting", controllerName) + r.rssInformerFactory.Start(ctx.Done()) + r.shuffleServerInformerFactory.Start(ctx.Done()) + r.coordinatorInformerFactory.Start(ctx.Done()) + if !cache.WaitForCacheSync(ctx.Done(), r.rssInformer.HasSynced, r.stsInformer.HasSynced, + r.podInformer.HasSynced, r.cmInformer.HasSynced) { + return fmt.Errorf("wait for cache synced failed") + } + klog.V(2).Infof("%v started", controllerName) + for i := 0; i < r.workers; i++ { + go wait.Until(r.runRssWorker, time.Second, ctx.Done()) + } + for i := 0; i < r.workers; i++ { + go wait.Until(r.runShuffleServerWorker, time.Second, ctx.Done()) + } + <-ctx.Done() + return nil +} + +// runRssWorker runs a thread that dequeues rss objects, handles them, and marks them done. +func (r *rssController) runRssWorker() { + for r.processNextWorkRss() { + } +} + +// processNextWorkRss deals with one rss key off the rssQueue, it returns false when it's time to quit. +func (r *rssController) processNextWorkRss() bool { + rssKey, quit := r.rssQueue.Get() + if quit { + return false + } + defer r.rssQueue.Done(rssKey) + + pvcNamespace, pvcName, err := cache.SplitMetaNamespaceKey(rssKey.(string)) Review Comment: Ok, I have modified this place. ########## deploy/kubernetes/operator/pkg/controller/controller/rss.go: ########## @@ -0,0 +1,925 @@ +/* + * 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 controller + +import ( + "context" + "fmt" + "strings" + "sync" + "time" + + appsv1 "k8s.io/api/apps/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/labels" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" + appslister "k8s.io/client-go/listers/apps/v1" + corelisters "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/retry" + "k8s.io/client-go/util/workqueue" + "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/controller/config" + controllerconstants "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/constants" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/sync/coordinator" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/sync/shuffleserver" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/util" + kubeutil "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/util/kubernetes" + propertiestutil "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/controller/util/properties" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/generated/clientset/versioned" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/generated/informers/externalversions" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/generated/listers/uniffle/v1alpha1" + "github.com/apache/incubator-uniffle/deploy/kubernetes/operator/pkg/utils" +) + +var _ RSSController = &rssController{} + +const ( + controllerName = "rss-controller" + appNameIndexer = "AppNameIndexer" +) + +// RSSController is responsible for synchronizing rss objects. +type RSSController interface { + manager.Runnable +} + +// NewRSSController creates a RSSController. +func NewRSSController(cfg *config.Config) RSSController { + return newRSSController(cfg) +} + +// newRSSController creates a rssController. +func newRSSController(cfg *config.Config) *rssController { + rc := &rssController{ + workers: cfg.Workers, + kubeClient: cfg.KubeClient, + rssClient: cfg.RSSClient, + rssInformerFactory: externalversions.NewSharedInformerFactory(cfg.RSSClient, 0), + shuffleServerInformerFactory: utils.BuildShuffleServerInformerFactory(cfg.KubeClient), + coordinatorInformerFactory: utils.BuildCoordinatorInformerFactory(cfg.KubeClient), + rssQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), + "rss-queue"), + shuffleServerQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), + "pod-queue"), + eventRecorder: util.CreateRecorder(cfg.KubeClient, "rss-controller"), + } + + rssInformer := rc.rssInformerFactory.Uniffle().V1alpha1().RemoteShuffleServices().Informer() + rssInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: rc.addRSS, + UpdateFunc: rc.updateRSS, + }) + rc.rssInformer = rssInformer + rc.rssLister = rc.rssInformerFactory.Uniffle().V1alpha1().RemoteShuffleServices().Lister() + + rc.stsInformer = rc.shuffleServerInformerFactory.Apps().V1().StatefulSets().Informer() + rc.stsLister = rc.shuffleServerInformerFactory.Apps().V1().StatefulSets().Lister() + + rc.podInformer = rc.shuffleServerInformerFactory.Core().V1().Pods().Informer() + rc.podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + DeleteFunc: rc.deleteShuffleServer, + }) + if err := rc.podInformer.AddIndexers(cache.Indexers{appNameIndexer: getAppName}); err != nil { + klog.Fatalf("add app name indexer failed: %v", err) + } + rc.podIndexer = rc.podInformer.GetIndexer() + rc.podLister = rc.shuffleServerInformerFactory.Core().V1().Pods().Lister() + + rc.cmInformer = rc.coordinatorInformerFactory.Core().V1().ConfigMaps().Informer() + rc.cmLister = rc.coordinatorInformerFactory.Core().V1().ConfigMaps().Lister() + + return rc +} + +// rssController implements the RSSController interface. +type rssController struct { + workers int + kubeClient kubernetes.Interface + rssClient versioned.Interface + rssInformerFactory externalversions.SharedInformerFactory + rssInformer cache.SharedIndexInformer + rssLister v1alpha1.RemoteShuffleServiceLister + shuffleServerInformerFactory informers.SharedInformerFactory + stsInformer cache.SharedIndexInformer + stsLister appslister.StatefulSetLister + podInformer cache.SharedIndexInformer + podIndexer cache.Indexer + podLister corelisters.PodLister + coordinatorInformerFactory informers.SharedInformerFactory + cmInformer cache.SharedIndexInformer + cmLister corelisters.ConfigMapLister + rssQueue workqueue.RateLimitingInterface + shuffleServerQueue workqueue.RateLimitingInterface + eventRecorder record.EventRecorder +} + +// Start starts the RSSController. +func (r *rssController) Start(ctx context.Context) error { + klog.V(2).Infof("%v is starting", controllerName) + r.rssInformerFactory.Start(ctx.Done()) + r.shuffleServerInformerFactory.Start(ctx.Done()) + r.coordinatorInformerFactory.Start(ctx.Done()) + if !cache.WaitForCacheSync(ctx.Done(), r.rssInformer.HasSynced, r.stsInformer.HasSynced, + r.podInformer.HasSynced, r.cmInformer.HasSynced) { + return fmt.Errorf("wait for cache synced failed") + } + klog.V(2).Infof("%v started", controllerName) + for i := 0; i < r.workers; i++ { + go wait.Until(r.runRssWorker, time.Second, ctx.Done()) + } + for i := 0; i < r.workers; i++ { + go wait.Until(r.runShuffleServerWorker, time.Second, ctx.Done()) + } + <-ctx.Done() + return nil +} + +// runRssWorker runs a thread that dequeues rss objects, handles them, and marks them done. +func (r *rssController) runRssWorker() { + for r.processNextWorkRss() { + } +} + +// processNextWorkRss deals with one rss key off the rssQueue, it returns false when it's time to quit. +func (r *rssController) processNextWorkRss() bool { + rssKey, quit := r.rssQueue.Get() + if quit { + return false + } + defer r.rssQueue.Done(rssKey) + + pvcNamespace, pvcName, err := cache.SplitMetaNamespaceKey(rssKey.(string)) + if err != nil { + klog.Errorf("parsed rss key (%v) failed: %v", rssKey, err) + return true + } + + var retryChecking bool + retryChecking, err = r.processRss(pvcNamespace, pvcName) + if err != nil { + klog.Errorf("processed rss %v failed : %v", rssKey, err) + r.rssQueue.AddRateLimited(rssKey) + } else if retryChecking { + klog.V(4).Infof("we will retry checking rss (%v) for upgrading or terminating", rssKey) + r.rssQueue.AddAfter(rssKey, time.Second*3) + } else { + r.rssQueue.Forget(rssKey) + } + return true +} + +// runShuffleServerWorker runs a thread that dequeues shuffle server pods, handles them, and marks them done. +func (r *rssController) runShuffleServerWorker() { + for r.processNextWorkShuffleServer() { + } +} + +// processNextWorkShuffleServer deals with one shuffle server pod key off the shuffleServerQueue, +// it returns false when it's time to quit. +func (r *rssController) processNextWorkShuffleServer() bool { + key, quit := r.shuffleServerQueue.Get() + if quit { + return false + } + defer r.shuffleServerQueue.Done(key) + + err := r.processShuffleServer(key.(string)) + if err == nil { + r.shuffleServerQueue.Forget(key) + return true + } + + klog.Errorf("processed shuffle server %v failed : %v", key, err) + r.shuffleServerQueue.AddRateLimited(key) + return true +} + +// processShuffleServer process current shuffle server pod key. +func (r *rssController) processShuffleServer(key string) error { + klog.V(4).Infof("processing shuffle server (%v)", key) + namespace, rssName, currentKey := parsePodCacheKey(key) + rss, err := r.rssLister.RemoteShuffleServices(namespace).Get(rssName) + if err != nil { + klog.Errorf("get rss by key (%v) failed: %v", key, err) + if !apierrors.IsNotFound(err) { + return err + } + return nil + } + + return r.updateTargetAndDeletedKeys(rss, currentKey) +} + +// updateTargetAndDeletedKeys updates target and deleted keys in status of rss objects. +func (r *rssController) updateTargetAndDeletedKeys(rss *unifflev1alpha1.RemoteShuffleService, shuffleServerKey string) error { + deletedKeys := sets.NewString(rss.Status.DeletedKeys...) + if deletedKeys.Has(shuffleServerKey) { + klog.V(3).Infof("shuffle server (%v) has already been deleted for rss (%v)", + shuffleServerKey, utils.UniqueName(rss)) + return nil + } + + deletedKeys.Insert(shuffleServerKey) + oldTargetKeys := sets.NewString(rss.Status.TargetKeys...) + newTargetKeys := oldTargetKeys.Difference(deletedKeys) + + cm, err := r.getExcludeNodesCM(rss) + if err != nil { + klog.Errorf("get exclude nodes in configMap for rss (%v) failed: %v", + utils.UniqueName(rss), err) + return err + } + + // calculate new exclude nodes. + excludeNodesFileKey := utils.GetExcludeNodesConfigMapKey(rss) + oldExcludeNodes := sets.NewString(strings.Split( + cm.Data[excludeNodesFileKey], "\n")...) + newExcludeNodes := utils.ConvertShuffleServerKeysToNodes(newTargetKeys) + if !newExcludeNodes.Equal(oldExcludeNodes) { + cmCopy := cm.DeepCopy() + cmCopy.Data[excludeNodesFileKey] = strings.Join( + utils.GetSortedList(newExcludeNodes), "\n") + if _, err = r.kubeClient.CoreV1().ConfigMaps(cmCopy.Namespace). + Update(context.Background(), cmCopy, metav1.UpdateOptions{}); err != nil { + klog.Errorf("update exclude nodes in configMap (%v) for rss (%v) failed: %v", + cmCopy.Name, utils.UniqueName(rss), err) + return err + } + } + + if err = r.updateDeletedKeys(rss, shuffleServerKey); err != nil { + klog.Errorf("add (%v) to deleted keys of rss (%v) failed: %v", + shuffleServerKey, utils.UniqueName(rss), err) + return err + } + return nil +} + +func (r *rssController) clearExcludeNodes(rss *unifflev1alpha1.RemoteShuffleService) error { + cm, err := r.getExcludeNodesCM(rss) + if err != nil { + klog.Errorf("get exclude nodes in configMap for rss (%v) failed: %v", + utils.UniqueName(rss), err) + return err + } + cmCopy := cm.DeepCopy() + cmCopy.Data[utils.GetExcludeNodesConfigMapKey(rss)] = "" + if _, err = r.kubeClient.CoreV1().ConfigMaps(cmCopy.Namespace). + Update(context.Background(), cmCopy, metav1.UpdateOptions{}); err != nil { + klog.Errorf("update exclude nodes in configMap (%v) for rss (%v) failed: %v", + cmCopy.Name, utils.UniqueName(rss), err) + return err + } + return nil +} + +// getExcludeNodesCM returns configMap records exclude nodes. +func (r *rssController) getExcludeNodesCM(rss *unifflev1alpha1.RemoteShuffleService) ( + *corev1.ConfigMap, error) { + namespace := rss.Namespace + name := utils.GenerateCoordinatorName(rss) + cm, err := r.cmLister.ConfigMaps(namespace).Get(name) + if err != nil { + klog.Errorf("get configMap (%v) for rss failed: %v", name, utils.UniqueName(rss), err) + return nil, err + } + return cm, nil +} + +// updateDeletedKeys updates deleted keys in status of rss. +func (r *rssController) updateDeletedKeys(rss *unifflev1alpha1.RemoteShuffleService, + key string) error { + if rss.Status.Phase != unifflev1alpha1.RSSUpgrading && rss.Status.Phase != unifflev1alpha1.RSSTerminating { + return nil + } + rssStatus := rss.Status.DeepCopy() + rssStatus.DeletedKeys = utils.GetSortedList(sets.NewString(rssStatus.DeletedKeys...).Insert(key)) + return r.updateRssStatus(rss, rssStatus) +} + +// processRss process current rss by its namespace and name, and returns if we need to try again to check upgrading. +func (r *rssController) processRss(namespace, name string) (bool, error) { + klog.V(4).Infof("processing rss (%v/%v)", namespace, name) + startTime := time.Now() + defer func() { + klog.V(4).Infof("finished processing rss (%v/%v) cost %v", + namespace, name, time.Since(startTime)) + }() + + rss, err := r.rssLister.RemoteShuffleServices(namespace).Get(name) + if err != nil { + if apierrors.IsNotFound(err) { + klog.V(5).Infof("ignored deleted rss (%v/%v)", namespace, name) + return false, nil + } + klog.Errorf("get rss (%v/%v) failed: %v", namespace, name, err) + return false, err + } + + if isDeletionCandidate(rss) { + return r.processDeleting(rss) Review Comment: Ok, I have modified this place. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
