Skip to content

fix: increase Helm client QPS and discovery burst limits #5015

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
PetrusZ wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix: increase Helm client QPS and discovery burst limits #5015

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7dfc24d
fix: increase Helm client QPS and discovery burst limits
PetrusZ Oct 8, 2026
4aed4e0
fix: add phase timing logs for Helm deployment diagnostics
PetrusZ Oct 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 32 additions & 3 deletions pkg/microservice/aslan/core/common/service/helm/helm.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,25 +115,50 @@ func CopyAndUploadService(projectName, serviceName, currentChartPath string, cop
}

// Update Service and ServiceDeployStrategy for a single service in environment
func UpdateServiceInEnv(product *commonmodels.Product, productSvc *commonmodels.ProductService, user string, operation config.EnvOperation, detail, clusterName string) error {
func UpdateServiceInEnv(product *commonmodels.Product, productSvc *commonmodels.ProductService, user string, operation config.EnvOperation, detail, clusterName string) (err error) {
started := time.Now()
logger := log.SugaredLogger().With("operation", "update_environment", "project", product.ProductName, "env", product.EnvName, "namespace", product.Namespace, "service", productSvc.ServiceName, "cluster_id", product.ClusterID, "operation_started_at", started.Format(time.RFC3339Nano))
phase := ""
phaseStarted := started
nextPhase := func(next string) {
if phase != "" {
logger.Infow("Helm deploy phase completed", "phase", phase, "duration", time.Since(phaseStarted).String())
}
phase, phaseStarted = next, time.Now()
logger.Infow("Helm deploy phase started", "phase", phase)
}
defer func() {
if err != nil {
logger.Errorw("Helm deploy failed", "phase", phase, "phase_duration", time.Since(phaseStarted).String(), "duration", time.Since(started).String(), "error", err)
} else {
logger.Infow("Helm deploy completed", "phase", phase, "phase_duration", time.Since(phaseStarted).String(), "duration", time.Since(started).String())
}
}()

nextPhase("start_environment_transaction")
session := mongo.Session()
defer session.EndSession(context.TODO())

err := mongo.StartTransaction(session)
err = mongo.StartTransaction(session)
if err != nil {
return err
}

nextPhase("create_service_version")
product.LintServices()
err = commonutil.CreateEnvServiceVersion(product, productSvc, user, operation, detail, clusterName, session, log.SugaredLogger())
if err != nil {
log.Errorf("failed to create helm service version, err: %v", err)
}

envLock := cache.NewRedisLock(fmt.Sprintf("%s:%s:%s", UpdateHelmEnvLockKey, product.ProductName, product.EnvName))
envLock.Lock()
nextPhase("acquire_environment_lock")
if lockErr := envLock.Lock(); lockErr != nil {
logger.Warnw("Failed to acquire Helm environment lock", "error", lockErr)
}
defer envLock.Unlock()

nextPhase("read_environment")
productColl := commonrepo.NewProductCollWithSession(session)
newProductInfo, err := productColl.Find(&commonrepo.ProductFindOptions{Name: product.ProductName, EnvName: product.EnvName})
if err != nil {
Expand All @@ -159,12 +184,14 @@ func UpdateServiceInEnv(product *commonmodels.Product, productSvc *commonmodels.
}
}

nextPhase("read_project_template")
templateProduct, err := template.NewProductCollWithSess(session).Find(product.ProductName)
if err != nil {
mongo.AbortTransaction(session)
return errors.Wrapf(err, "failed to find template product %s", product.ProductName)
}

nextPhase("rebuild_environment_services")
newProductInfo.Services = [][]*commonmodels.ProductService{}
serviceOrchestration := templateProduct.Services
if product.Production {
Expand Down Expand Up @@ -211,12 +238,14 @@ func UpdateServiceInEnv(product *commonmodels.Product, productSvc *commonmodels.
}

newProductInfo.UpdateBy = user
nextPhase("write_environment")
if err = productColl.Update(newProductInfo); err != nil {
log.Errorf("update product %s error: %s", newProductInfo.ProductName, err.Error())
mongo.AbortTransaction(session)
return fmt.Errorf("failed to update product info, name %s", newProductInfo.ProductName)
}

nextPhase("commit_environment_transaction")
return mongo.CommitTransaction(session)
}

Expand Down
63 changes: 60 additions & 3 deletions pkg/microservice/aslan/core/common/service/kube/helm.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,27 @@ type ReleaseInstallParam struct {
MaxHistory int
}

func InstallOrUpgradeHelmChartWithValues(param *ReleaseInstallParam, isRetry bool, helmClient *helmtool.HelmClient) error {
func InstallOrUpgradeHelmChartWithValues(param *ReleaseInstallParam, isRetry bool, helmClient *helmtool.HelmClient) (err error) {
started := time.Now()
logger := log.SugaredLogger().With("operation", "apply_release", "project", param.ProductName, "namespace", param.Namespace, "release", param.ReleaseName, "service", param.ServiceObj.ServiceName, "cluster_id", helmClient.ClusterID, "retry", isRetry, "operation_started_at", started.Format(time.RFC3339Nano))
phase := ""
phaseStarted := started
nextPhase := func(next string) {
if phase != "" {
logger.Infow("Helm deploy phase completed", "phase", phase, "duration", time.Since(phaseStarted).String())
}
phase, phaseStarted = next, time.Now()
logger.Infow("Helm deploy phase started", "phase", phase)
}
defer func() {
if err != nil {
logger.Errorw("Helm deploy failed", "phase", phase, "phase_duration", time.Since(phaseStarted).String(), "duration", time.Since(started).String(), "error", err)
} else {
logger.Infow("Helm deploy completed", "phase", phase, "phase_duration", time.Since(phaseStarted).String(), "duration", time.Since(started).String())
}
}()

nextPhase("load_chart")
namespace, valuesYaml, renderChart, serviceObj := param.Namespace, param.MergedValues, param.RenderChart, param.ServiceObj
chartPath, err := PreLoadHelmServiceChart(serviceObj, param.Production, &chartInstantiateDeploy{
ChartName: renderChart.ChartName,
Expand Down Expand Up @@ -110,24 +130,28 @@ func InstallOrUpgradeHelmChartWithValues(param *ReleaseInstallParam, isRetry boo
chartSpec.Timeout = time.Second * time.Duration(param.Timeout)
}

nextPhase("inspect_stuck_workloads")
stuckDeployments, stuckStatefulSets, err := getStuckWorkload(helmClient, chartSpec)
if err != nil {
return fmt.Errorf("failed to get stuck workloads: %s", err)
}

// If the target environment is a shared environment and a sub env, we need to clear the deployed K8s Service.
ctx := context.TODO()
nextPhase("delete_precreated_services")
err = EnsureDeletePreCreatedServices(ctx, param.ProductName, param.Namespace, chartSpec, helmClient)
if err != nil {
return fmt.Errorf("failed to ensure deleting pre-created K8s Services for product %q in namespace %q: %s", param.ProductName, param.Namespace, err)
}

nextPhase("clone_helm_client")
helmClient, err = helmClient.Clone()
if err != nil {
return fmt.Errorf("failed to clone helm client: %s", err)
}

var release *release.Release
nextPhase("install_or_upgrade_chart")
release, err = helmClient.InstallOrUpgradeChart(ctx, chartSpec, nil)
if err != nil {
err = errors.WithMessagef(
Expand All @@ -136,6 +160,8 @@ func InstallOrUpgradeHelmChartWithValues(param *ReleaseInstallParam, isRetry boo
namespace, serviceObj.ServiceName)
return err
} else {
logger.Infow("Helm release deployed", "revision", release.Version, "status", release.Info.Status)
nextPhase("ensure_zadig_services")
err = EnsureZadigServiceByManifest(ctx, param.ProductName, param.Namespace, release.Manifest)
if err != nil {
err = errors.WithMessagef(err, "failed to ensure Zadig Service, err: %s", err)
Expand All @@ -148,10 +174,12 @@ func InstallOrUpgradeHelmChartWithValues(param *ReleaseInstallParam, isRetry boo
return fmt.Errorf("failed to get kubernetes client set, err: helm client rest config is nil")
}

nextPhase("create_cleanup_client")
clientSet, err := kubernetes.NewForConfig(restCfg)
if err != nil {
return fmt.Errorf("failed to get kubernetes client set, err: %v", err)
}
nextPhase("cleanup_stuck_workloads")
err = cleanupStuckWorkloads(clientSet, stuckDeployments, stuckStatefulSets, log.SugaredLogger())
if err != nil {
return fmt.Errorf("failed to cleanup stuck workloads, err: %v", err)
Expand Down Expand Up @@ -652,34 +680,57 @@ func EnsureDeleteZadigServiceBySvcName(ctx context.Context, env *commonmodels.Pr
return nil
}
func DeploySingleHelmRelease(product *commonmodels.Product, productSvc *commonmodels.ProductService,
svcTemp *commonmodels.Service, images []string, maxHistory, timeout int, user string) error {
svcTemp *commonmodels.Service, images []string, maxHistory, timeout int, user string) (err error) {
started := time.Now()
logger := log.SugaredLogger().With("operation", "deploy_single_release", "project", product.ProductName, "env", product.EnvName, "namespace", product.Namespace, "service", productSvc.ServiceName, "cluster_id", product.ClusterID, "operation_started_at", started.Format(time.RFC3339Nano))
phase := ""
phaseStarted := started
nextPhase := func(next string) {
if phase != "" {
logger.Infow("Helm deploy phase completed", "phase", phase, "duration", time.Since(phaseStarted).String())
}
phase, phaseStarted = next, time.Now()
logger.Infow("Helm deploy phase started", "phase", phase)
}
defer func() {
if err != nil {
logger.Errorw("Helm deploy failed", "phase", phase, "phase_duration", time.Since(phaseStarted).String(), "duration", time.Since(started).String(), "error", err)
} else {
logger.Infow("Helm deploy completed", "phase", phase, "phase_duration", time.Since(phaseStarted).String(), "duration", time.Since(started).String())
}
}()

chartInfo := productSvc.GetServiceRender()

var (
err error
releaseName string
replacedMergedValuesYaml string
)

if productSvc.DeployStrategy == setting.ServiceDeployStrategyDraft {
nextPhase("update_draft_environment")
return helmservice.UpdateServiceInEnv(product, productSvc, user, config.EnvOperationDefault, "", "")
}

releaseName = productSvc.ReleaseName
nextPhase("resolve_release_name")
if productSvc.FromZadig() {
releaseName = util.GeneReleaseName(svcTemp.GetReleaseNaming(), svcTemp.ProductName, product.Namespace, product.EnvName, svcTemp.ServiceName)
}

nextPhase("check_release_duplicate")
err = CheckReleaseDuplicate(productSvc.ServiceName, releaseName, product)
if err != nil {
return err
}

nextPhase("check_release_ownership")
err = CheckReleaseInstalledByOtherEnv(sets.NewString(releaseName), product)
if err != nil {
return err
}

nextPhase("merge_values")
if productSvc.FromZadig() {
releaseName = util.GeneReleaseName(svcTemp.GetReleaseNaming(), svcTemp.ProductName, product.Namespace, product.EnvName, svcTemp.ServiceName)
replacedMergedValuesYaml, err = helmservice.NewHelmDeployService().GenMergedValues(productSvc, product.DefaultValues, images)
Expand All @@ -695,12 +746,14 @@ func DeploySingleHelmRelease(product *commonmodels.Product, productSvc *commonmo
return fmt.Errorf("failed to gene merged values, err: %s", err)
}

nextPhase("download_chart")
err = DownloadInstantiateChart(product.ProductName, chartInfo.ChartRepo, chartInfo.ChartName, chartInfo.ChartVersion, releaseName, product.Production)
if err != nil {
return fmt.Errorf("failed to download instantiate chart, productName: %s, chartRepo: %s, chartName: %s, chartVersion: %s, releaseName: %s, production: %v, err: %s", product.ProductName, chartInfo.ChartRepo, chartInfo.ChartName, chartInfo.ChartVersion, releaseName, product.Production, err)
}
}

nextPhase("initialize_helm_client")
helmClient, err := helmtool.NewClientFromNamespace(product.ClusterID, product.Namespace)
if err != nil {
return err
Expand All @@ -725,6 +778,7 @@ func DeploySingleHelmRelease(product *commonmodels.Product, productSvc *commonmo
hrs, errHistory := helmClient.ListReleaseHistory(param.ReleaseName, 10)
if errHistory != nil {
// list history should not block deploy operation, error will be logged instead of returned
logger.Warnw("Failed to read Helm release history", "release", param.ReleaseName, "error", errHistory)
return nil
}
if len(hrs) == 0 {
Expand All @@ -739,17 +793,20 @@ func DeploySingleHelmRelease(product *commonmodels.Product, productSvc *commonmo
return nil
}

nextPhase("check_release_history")
err = ensureUpgrade()
if err != nil {
return err
}

// when replace image, should not wait
nextPhase("apply_release")
err = InstallOrUpgradeHelmChartWithValues(param, false, helmClient)
if err != nil {
return err
}

nextPhase("update_environment")
err = helmservice.UpdateServiceInEnv(product, productSvc, user, config.EnvOperationDefault, "", "")
return err
}
Expand Down
39 changes: 20 additions & 19 deletions pkg/tool/helmclient/helmclient.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,8 @@ import (

const (
HelmPluginsDirectory = "/app/.helm/helmplugin"
helmClientQPS = 50
helmClientBurst = 300
)

var repoInfo *repo.File
Expand Down Expand Up @@ -138,18 +140,11 @@ func NewClientFromNamespace(clusterID, namespace string) (*HelmClient, error) {
return nil, err
}

hcClient, err := hc.NewClientFromRestConf(&hc.RestConfClientOptions{
Options: &hc.Options{
Namespace: namespace,
DebugLog: log.Debugf,
},
RestConfig: restConfig,
})
helmClient, err := NewClientFromRestConf(restConfig, namespace)
if err != nil {
return nil, err
}

helmClient := hcClient.(*hc.HelmClient)
clientset, err := kubeManager.GetKubernetesClientSet(clusterID)
if err != nil {
return nil, fmt.Errorf("failed to get kubernetes clientset for cluster %s: %v", clusterID, err)
Expand All @@ -165,21 +160,21 @@ func NewClientFromNamespace(clusterID, namespace string) (*HelmClient, error) {
return nil, fmt.Errorf("failed to parse kubernetes server version %q for cluster %s: %v", versionInfo.GitVersion, clusterID, err)
}

return &HelmClient{
HelmClient: helmClient,
kubeClient: kubeClient,
ClusterID: clusterID,
Namespace: namespace,
KubeVersion: kubeVersion,
lock: &sync.Mutex{},
RestConfig: restConfig,
RegistryClient: nil,
}, nil
helmClient.kubeClient = kubeClient
helmClient.ClusterID = clusterID
helmClient.KubeVersion = kubeVersion
return helmClient, nil
}

// NewClientFromRestConf returns a new Helm client constructed with the provided REST config options
// only used to list/uninstall helm release because kubeClient is nil
// kubeClient is nil until supplied by NewClientFromNamespace or Clone.
func NewClientFromRestConf(restConfig *rest.Config, ns string) (*HelmClient, error) {
// Keep Helm's rate limits separate from the shared Kubernetes clients.
restConfig = rest.CopyConfig(restConfig)
restConfig.QPS = helmClientQPS
restConfig.Burst = helmClientBurst
restConfig.RateLimiter = nil

hcClient, err := hc.NewClientFromRestConf(&hc.RestConfClientOptions{
Options: &hc.Options{
Namespace: ns,
Expand All @@ -192,6 +187,12 @@ func NewClientFromRestConf(restConfig *rest.Config, ns string) (*HelmClient, err
}

helmClient := hcClient.(*hc.HelmClient)
// go-helm-client resets discovery Burst to 100 unless an option overrides it.
getter := hc.NewRESTClientGetter(ns, nil, restConfig, hc.Burst(helmClientBurst))
if err := helmClient.ActionConfig.Init(getter, ns, os.Getenv("HELM_DRIVER"), log.Debugf); err != nil {
return nil, err
}

return &HelmClient{
HelmClient: helmClient,
kubeClient: nil,
Expand Down
Loading