fix(application): fix helm pending with context cancel (#2193)

Co-authored-by: xdonggao <xdonggao@tencent.com>

cmd/tke-application-controller/app/run.go

@@ -105,6 +105,7 @@ func Run(cfg *config.Config, stopCh <-chan struct{}) error {
 		Callbacks: leaderelection.LeaderCallbacks{
 			OnStartedLeading: run,
 			OnStoppedLeading: func() {
+				time.Sleep(20 * time.Second)
 				log.Fatalf("leaderelection lost")
 			},
 		},

cmd/tke-installer/app/installer/application.go

@@ -378,7 +378,7 @@ func (t *TKE) installPlatformApp(ctx context.Context, platformApp *types.Platfor
 	// TODO currently only support local chart install
 	if len(platformApp.LocalChartPath) != 0 {
 		t.log.Infof("Start install platform app %s in %s namespace", platformApp.HelmInstallOptions.ReleaseName, platformApp.HelmInstallOptions.Namespace)
-		if _, err := t.helmClient.InstallWithLocal(platformApp.HelmInstallOptions, platformApp.LocalChartPath); err != nil {
+		if _, err := t.helmClient.InstallWithLocal(ctx, platformApp.HelmInstallOptions, platformApp.LocalChartPath); err != nil {
 			uninstallOptions := helmaction.UninstallOptions{
 				Timeout:     10 * time.Minute,
 				ReleaseName: platformApp.HelmInstallOptions.ReleaseName,

cmd/tke-installer/app/installer/installer.go

@@ -1583,7 +1583,7 @@ func (t *TKE) installTKEGatewayChart(ctx context.Context) error {
 	}
 
 	chartFilePath := constants.ChartDirName + "tke-gateway/"
-	if _, err := t.helmClient.InstallWithLocal(installOptions, chartFilePath); err != nil {
+	if _, err := t.helmClient.InstallWithLocal(ctx, installOptions, chartFilePath); err != nil {
 		uninstallOptions := helmaction.UninstallOptions{
 			Timeout:     10 * time.Minute,
 			ReleaseName: "tke-gateway",
@@ -2093,7 +2093,7 @@ func (t *TKE) installTKEMonitorController(ctx context.Context) error {
 			// thanos-query address
 			params["MonitorStorageAddresses"] = "http://thanos-query.tke.svc.cluster.local:9090"
 		}
-		params["RetentionDays"] = t.Para.Config.Monitor.RetentionDays //can accept a nil value
+		params["RetentionDays"] = t.Para.Config.Monitor.RetentionDays // can accept a nil value
 	}
 
 	if err := apiclient.CreateResourceWithDir(ctx, t.globalClient, "manifests/tke-monitor-controller/*.yaml", params); err != nil {
@@ -2206,14 +2206,14 @@ func (t *TKE) getTKERegistryAPIOptions(ctx context.Context) (map[string]interfac
 		"harborEnabled":  t.Para.Config.Registry.TKERegistry.HarborEnabled,
 		"harborCAFile":   t.Para.Config.Registry.TKERegistry.HarborCAFile,
 	}
-	//check if s3 enabled
+	// check if s3 enabled
 	storageConfig := t.Para.Config.Registry.TKERegistry.Storage
 	s3Enabled := (storageConfig != nil && storageConfig.S3 != nil)
 	options["s3Enabled"] = s3Enabled
 	if s3Enabled {
 		options["s3Storage"] = storageConfig.S3
 	}
-	//or enable filesystem by default
+	// or enable filesystem by default
 	options["filesystemEnabled"] = !s3Enabled
 	if options["filesystemEnabled"] == true {
 		useCephRbd, useNFS := false, false
@@ -2273,14 +2273,14 @@ func (t *TKE) getTKERegistryControllerOptions(ctx context.Context) (map[string]i
 		"domainSuffix":       t.Para.Config.Registry.TKERegistry.Domain,
 		"defaultChartGroups": defaultChartGroupsStringConfig,
 	}
-	//check if s3 enabled
+	// check if s3 enabled
 	storageConfig := t.Para.Config.Registry.TKERegistry.Storage
 	s3Enabled := (storageConfig != nil && storageConfig.S3 != nil)
 	options["s3Enabled"] = s3Enabled
 	if s3Enabled {
 		options["s3Storage"] = storageConfig.S3
 	}
-	//or enable filesystem by default
+	// or enable filesystem by default
 	options["filesystemEnabled"] = !s3Enabled
 	if options["filesystemEnabled"] == true {
 		useCephRbd, useNFS := false, false

pkg/application/controller/app/action/install.go

@@ -82,7 +82,7 @@ func Install(ctx context.Context,
 	}
 
 	chartPathBasicOptions.ExistedFile = destfile
-	_, err = client.Install(&helmaction.InstallOptions{
+	_, err = client.Install(ctx, &helmaction.InstallOptions{
 		Namespace:        newApp.Spec.TargetNamespace,
 		ReleaseName:      newApp.Spec.Name,
 		Atomic:           newApp.Spec.Chart.Atomic,

pkg/application/controller/app/action/types.go

@@ -23,3 +23,4 @@ import (
 )
 
 const clientTimeOut = 300 * time.Second
+const clientMaxHistory = 10

pkg/application/controller/app/action/upgrade.go

@@ -83,14 +83,15 @@ func Upgrade(ctx context.Context,
 	}
 
 	chartPathBasicOptions.ExistedFile = destfile
-	_, err = client.Upgrade(&helmaction.UpgradeOptions{
+	_, err = client.Upgrade(ctx, &helmaction.UpgradeOptions{
 		Namespace:        app.Spec.TargetNamespace,
 		ReleaseName:      app.Spec.Name,
 		DependencyUpdate: true,
 		Install:          true,
 		Values:           values,
 		Timeout:          clientTimeOut,
 		ChartPathOptions: chartPathBasicOptions,
+		MaxHistory:       clientMaxHistory,
 	})
 
 	if updateStatusFunc != nil {

pkg/application/controller/app/app_controller.go

@@ -239,6 +239,19 @@ func (c *Controller) syncItem(key string) error {
 	if err != nil {
 		return err
 	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		select {
+		case <-c.stopCh:
+			log.Info("stop ch", log.String("namespace", namespace), log.String("name", name))
+			cancel()
+			return
+		case <-ctx.Done():
+			log.Info("success done", log.String("namespace", namespace), log.String("name", name))
+			return
+		}
+	}()
 	// app holds the latest App info from apiserver
 	app, err := c.lister.Apps(namespace).Get(name)
 	switch {
@@ -259,7 +272,7 @@ func (c *Controller) syncItem(key string) error {
 				log.String("namespace", namespace),
 				log.String("name", name))
 			_ = c.processDeletion(key)
-			err = c.appResourcesDeleter.Delete(context.Background(), namespace, name)
+			err = c.appResourcesDeleter.Delete(ctx, namespace, name)
 			metrics.GaugeApplicationInstallFailed.WithLabelValues(app.Spec.TargetCluster, app.Name).Set(0)
 			metrics.GaugeApplicationUpgradeFailed.WithLabelValues(app.Spec.TargetCluster, app.Name).Set(0)
 			metrics.GaugeApplicationRollbackFailed.WithLabelValues(app.Spec.TargetCluster, app.Name).Set(0)
@@ -268,7 +281,7 @@ func (c *Controller) syncItem(key string) error {
 			// DeletionTimestamp is not empty and object will be deleted when you request updateStatus
 		} else {
 			cachedApp := c.cache.getOrCreate(key)
-			err = c.processUpdate(context.Background(), cachedApp, app, key)
+			err = c.processUpdate(ctx, cachedApp, app, key)
 		}
 	}
 	return err

pkg/application/helm/action/install.go

@@ -19,6 +19,7 @@
 package action
 
 import (
+	"context"
 	"os"
 	"path/filepath"
 	"time"
@@ -84,13 +85,13 @@ func (cp ChartPathOptions) ApplyTo(opt *action.ChartPathOptions) {
 	opt.Version = cp.Version
 }
 
-func (c *Client) Install(options *InstallOptions) (*release.Release, error) {
-	return c.InstallWithLocal(options, "")
+func (c *Client) Install(ctx context.Context, options *InstallOptions) (*release.Release, error) {
+	return c.InstallWithLocal(ctx, options, "")
 }
 
 // Install installs a chart archive
 // if chartLocalFile is not empty, chart files exists in the project
-func (c *Client) InstallWithLocal(options *InstallOptions, chartLocalFile string) (*release.Release, error) {
+func (c *Client) InstallWithLocal(ctx context.Context, options *InstallOptions, chartLocalFile string) (*release.Release, error) {
 	actionConfig, err := c.buildActionConfig(options.Namespace)
 	if err != nil {
 		return nil, err
@@ -179,7 +180,7 @@ func (c *Client) InstallWithLocal(options *InstallOptions, chartLocalFile string
 		}
 	}
 
-	return client.Run(chartRequested, options.Values)
+	return client.RunWithContext(ctx, chartRequested, options.Values)
 }
 
 // isChartInstallable validates if a chart can be installed

pkg/application/helm/action/upgrade.go

@@ -19,6 +19,7 @@
 package action
 
 import (
+	"context"
 	"os"
 	"time"
 
@@ -64,7 +65,7 @@ type UpgradeOptions struct {
 }
 
 // Upgrade upgrade a helm release
-func (c *Client) Upgrade(options *UpgradeOptions) (*release.Release, error) {
+func (c *Client) Upgrade(ctx context.Context, options *UpgradeOptions) (*release.Release, error) {
 	actionConfig, err := c.buildActionConfig(options.Namespace)
 	if err != nil {
 		return nil, err
@@ -75,7 +76,7 @@ func (c *Client) Upgrade(options *UpgradeOptions) (*release.Release, error) {
 		histClient.Max = 1
 		if _, err := histClient.Run(options.ReleaseName); err == driver.ErrReleaseNotFound {
 			log.Infof("Release %q does not exist. Installing it now.\n", options.ReleaseName)
-			return c.Install(&InstallOptions{
+			return c.Install(ctx, &InstallOptions{
 				DryRun:           options.DryRun,
 				DependencyUpdate: options.DependencyUpdate,
 				Timeout:          options.Timeout,
@@ -154,5 +155,5 @@ func (c *Client) Upgrade(options *UpgradeOptions) (*release.Release, error) {
 	if chartRequested.Metadata.Deprecated {
 		log.Warnf("This chart %s/%s is deprecated", options.ChartRepo, options.Chart)
 	}
-	return client.Run(options.ReleaseName, chartRequested, options.Values)
+	return client.RunWithContext(ctx, options.ReleaseName, chartRequested, options.Values)
 }

pkg/application/registry/application/storage/rest.go

@@ -261,7 +261,7 @@ func (rs *REST) prepareForCheck(ctx context.Context, app *application.App) error
 }
 
 func (rs *REST) canVisitChart(ctx context.Context, app *application.App) error {
-	//TODO: allowAlways if registryClient is empty?
+	// TODO: allowAlways if registryClient is empty?
 	if rs.registryClient == nil {
 		return nil
 	}
@@ -360,7 +360,7 @@ func (rs *REST) dryRun(ctx context.Context, app *application.App) (*release.Rele
 	}
 
 	chartPathBasicOptions.ExistedFile = destfile
-	rel, err := client.Install(&helmaction.InstallOptions{
+	rel, err := client.Install(ctx, &helmaction.InstallOptions{
 		Namespace:        app.Spec.TargetNamespace,
 		ReleaseName:      app.Spec.Name,
 		DependencyUpdate: true,