Skip to content

Commit

Permalink
cleanup: add finalizers for mcs to make code graceful
Browse files Browse the repository at this point in the history 
Signed-off-by: duanmengkk <[email protected]>
  • Loading branch information
@duanmengkk
duanmengkk committed Dec 22, 2023
1 parent 04938bf commit 84f6474
Show file tree
Hide file tree
Showing 4 changed files with 134 additions and 52 deletions.
25 changes: 23 additions & 2 deletions pkg/clustertree/cluster-manager/controllers/mcs/auto_mcs_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -211,7 +211,7 @@ func (c *AutoCreateMCSController) SetupWithManager(mgr manager.Manager) error {

func (c *AutoCreateMCSController) cleanUpMcsResources(ctx context.Context, namespace string, name string, clusterList *kosmosv1alpha1.ClusterList) error {
// delete serviceExport in root cluster
if err := c.RootKosmosClient.MulticlusterV1alpha1().ServiceExports(namespace).Delete(ctx, name, metav1.DeleteOptions{}); err != nil {
if err := c.RootKosmosClient.MulticlusterV1alpha1().ServiceExports(namespace).Delete(context.TODO(), name, metav1.DeleteOptions{}); err != nil {
if !apierrors.IsNotFound(err) {
klog.Errorf("Delete serviceExport in root cluster failed %s/%s, Error: %v", namespace, name, err)
return err
Expand All @@ -229,7 +229,7 @@ func (c *AutoCreateMCSController) cleanUpMcsResources(ctx context.Context, names
klog.Errorf("get leafManager for cluster %s failed,Error: %v", cluster.Name, err)
return err
}
if err = leafManager.KosmosClient.MulticlusterV1alpha1().ServiceImports(namespace).Delete(ctx, name, metav1.DeleteOptions{}); err != nil {
if err = leafManager.KosmosClient.MulticlusterV1alpha1().ServiceImports(namespace).Delete(context.TODO(), name, metav1.DeleteOptions{}); err != nil {
if !apierrors.IsNotFound(err) {
klog.Errorf("Delete serviceImport in leaf cluster failed %s/%s, Error: %v", namespace, name, err)
return err
Expand Down Expand Up @@ -266,6 +266,14 @@ func (c *AutoCreateMCSController) autoCreateMcsResources(ctx context.Context, se
klog.Errorf("get leafManager for cluster %s failed,Error: %v", cluster.Name, err)
return err
}

if err = c.createNamespace(leafManager.Client, service.Namespace); err != nil {
if !apierrors.IsAlreadyExists(err) {
klog.Errorf("Create namespace %s in leaf cluster failed, Error: %v", service.Namespace, err)
return err
}
}

serviceImport := &mcsv1alpha1.ServiceImport{
ObjectMeta: metav1.ObjectMeta{
Name: service.Name,
Expand All @@ -290,3 +298,16 @@ func (c *AutoCreateMCSController) autoCreateMcsResources(ctx context.Context, se
}
return nil
}

func (c *AutoCreateMCSController) createNamespace(client client.Client, namespace string) error {
ns := &corev1.Namespace{
ObjectMeta: metav1.ObjectMeta{
Name: namespace,
},
}
err := client.Create(context.TODO(), ns)
if err != nil {
return err
}
return nil
}
62 changes: 46 additions & 16 deletions pkg/clustertree/cluster-manager/controllers/mcs/serviceexport_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import (
controllerruntime "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/builder"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/handler"
"sigs.k8s.io/controller-runtime/pkg/manager"
Expand Down Expand Up @@ -45,28 +46,28 @@ func (c *ServiceExportController) Reconcile(ctx context.Context, request reconci
klog.V(4).Infof("============ %s has been reconciled =============", request.NamespacedName.String())
}()

var shouldDelete bool
serviceExport := &mcsv1alpha1.ServiceExport{}
if err := c.RootClient.Get(ctx, request.NamespacedName, serviceExport); err != nil {
if !apierrors.IsNotFound(err) {
return controllerruntime.Result{Requeue: true}, err
if apierrors.IsNotFound(err) {
return controllerruntime.Result{}, nil
}
shouldDelete = true

return controllerruntime.Result{Requeue: true}, err
}

// The serviceExport is being deleted, in which case we should clear endpointSlice.
if shouldDelete || !serviceExport.DeletionTimestamp.IsZero() {
if err := c.removeAnnotation(ctx, request.Namespace, request.Name); err != nil {
if !serviceExport.DeletionTimestamp.IsZero() {
if err := c.removeAnnotation(request.Namespace, request.Name); err != nil {
return controllerruntime.Result{Requeue: true}, err
}
return controllerruntime.Result{}, nil
return c.removeFinalizer(serviceExport)
}

err := c.syncServiceExport(ctx, serviceExport)
if err != nil {
return controllerruntime.Result{Requeue: true}, err
}
return controllerruntime.Result{}, nil
return c.ensureFinalizer(serviceExport)
}

func (c *ServiceExportController) SetupWithManager(mgr manager.Manager) error {
Expand Down Expand Up @@ -122,18 +123,19 @@ func (c *ServiceExportController) shouldEnqueue(object client.Object) bool {
return true
}

func (c *ServiceExportController) removeAnnotation(ctx context.Context, namespace, name string) error {
func (c *ServiceExportController) removeAnnotation(namespace, name string) error {
var err error
selector := labels.SelectorFromSet(
map[string]string{
utils.ServiceKey: name,
},
)
epsList := &discoveryv1.EndpointSliceList{}
err = c.RootClient.List(ctx, epsList, &client.ListOptions{
err = c.RootClient.List(context.TODO(), epsList, &client.ListOptions{
Namespace: namespace,
LabelSelector: selector,
})

if err != nil {
klog.Errorf("List endpointSlice in %s failed, Error: %v", namespace, err)
return err
Expand All @@ -147,7 +149,7 @@ func (c *ServiceExportController) removeAnnotation(ctx context.Context, namespac
continue
}
helper.RemoveAnnotation(newEps, utils.ServiceExportLabelKey)
err = c.updateEndpointSlice(ctx, newEps, c.RootClient)
err = c.updateEndpointSlice(newEps, c.RootClient)
if err != nil {
klog.Errorf("Update endpointSlice (%s/%s) failed, Error: %v", namespace, newEps.Name, err)
return err
Expand All @@ -156,9 +158,9 @@ func (c *ServiceExportController) removeAnnotation(ctx context.Context, namespac
return nil
}

func (c *ServiceExportController) updateEndpointSlice(ctx context.Context, eps *discoveryv1.EndpointSlice, rootClient client.Client) error {
func (c *ServiceExportController) updateEndpointSlice(eps *discoveryv1.EndpointSlice, rootClient client.Client) error {
return retry.RetryOnConflict(retry.DefaultRetry, func() error {
updateErr := rootClient.Update(ctx, eps)
updateErr := rootClient.Update(context.TODO(), eps)
if updateErr == nil {
return nil
}
Expand All @@ -168,8 +170,8 @@ func (c *ServiceExportController) updateEndpointSlice(ctx context.Context, eps *
Namespace: eps.Namespace,
Name: eps.Name,
}
getErr := rootClient.Get(ctx, key, newEps)
if getErr == nil {
getErr := rootClient.Get(context.TODO(), key, newEps)
if getErr == nil || apierrors.IsNotFound(getErr) {
//Make a copy, so we don't mutate the shared cache
eps = newEps.DeepCopy()
} else {
Expand Down Expand Up @@ -205,7 +207,7 @@ func (c *ServiceExportController) syncServiceExport(ctx context.Context, export
continue
}
helper.AddEndpointSliceAnnotation(newEps, utils.ServiceExportLabelKey, utils.MCSLabelValue)
err = c.updateEndpointSlice(ctx, newEps, c.RootClient)
err = c.updateEndpointSlice(newEps, c.RootClient)
if err != nil {
klog.Errorf("Update endpointSlice (%s/%s) failed, Error: %v", export.Namespace, newEps.Name, err)
return err
Expand All @@ -215,3 +217,31 @@ func (c *ServiceExportController) syncServiceExport(ctx context.Context, export
c.EventRecorder.Event(export, corev1.EventTypeNormal, "Synced", "serviceExport has been synced to endpointSlice's annotation successfully")
return nil
}

func (c *ServiceExportController) ensureFinalizer(export *mcsv1alpha1.ServiceExport) (reconcile.Result, error) {
if controllerutil.ContainsFinalizer(export, utils.MCSFinalizer) {
return controllerruntime.Result{}, nil
}

controllerutil.AddFinalizer(export, utils.MCSFinalizer)
err := c.RootClient.Update(context.TODO(), export)
if err != nil {
return controllerruntime.Result{Requeue: true}, err
}

return controllerruntime.Result{}, nil
}

func (c *ServiceExportController) removeFinalizer(export *mcsv1alpha1.ServiceExport) (reconcile.Result, error) {
if !controllerutil.ContainsFinalizer(export, utils.MCSFinalizer) {
return controllerruntime.Result{}, nil
}

controllerutil.RemoveFinalizer(export, utils.MCSFinalizer)
err := c.RootClient.Update(context.TODO(), export)
if err != nil {
return controllerruntime.Result{Requeue: true}, err
}

return controllerruntime.Result{}, nil
}
Loading

0 comments on commit 84f6474

Please sign in to comment.