func DeployAndRepeatRollingUpdateDeployments()

in contrib/utils/utils.go [108:211]
91 lines of code
13 McCabe index (conditional complexity)

func DeployAndRepeatRollingUpdateDeployments(
	ctx context.Context,
	kubeCfgPath string,
	releaseName string,
	total, replica, paddingBytes int,
	internal time.Duration,
) (rollingUpdateFn, cleanupFn func(), retErr error) {
	infoLogger := log.GetLogger(ctx).WithKeyValues("level", "info")
	warnLogger := log.GetLogger(ctx).WithKeyValues("level", "warn")

	target := "workload/deployments"
	ch, err := manifests.LoadChart(target)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to load %s chart: %w", target, err)
	}

	namePattern := releaseName

	releaseCli, err := helmcli.NewReleaseCli(
		kubeCfgPath,
		// NOTE: The deployments have fixed namespace name so here
		// it's used to fill the required argument for NewReleaseCli.
		"default",
		releaseName,
		ch,
		nil,
		helmcli.StringPathValuesApplier(
			fmt.Sprintf("namePattern=%s", namePattern),
			fmt.Sprintf("total=%d", total),
			fmt.Sprintf("replica=%d", replica),
			fmt.Sprintf("paddingBytes=%d", paddingBytes),
		),
	)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create a new helm release cli: %w", err)
	}

	infoLogger.LogKV(
		"msg", "deploying deployments",
		"total", total,
		"replica", replica,
		"paddingBytes", paddingBytes,
	)

	err = releaseCli.Deploy(ctx, 10*time.Minute)
	if err != nil {
		if errors.Is(err, context.Canceled) {
			infoLogger.LogKV("msg", "deploy is canceled")
			return func() {}, func() {}, nil
		}
		return nil, nil, fmt.Errorf("failed to deploy helm chart %s: %w", target, err)
	}
	infoLogger.LogKV("msg", "deployed deployments")

	cleanupFn = func() {
		infoLogger.LogKV("msg", "cleanup helm chart", "target", target)
		err := releaseCli.Uninstall()
		if err != nil {
			warnLogger.LogKV("msg", "failed to cleanup helm chart",
				"target", target,
				"error", err)
		}
	}

	rollingUpdateFn = func() {
		for {
			select {
			case <-ctx.Done():
				infoLogger.LogKV("msg", "stop rolling-updating")
				return
			case <-time.After(internal):
			}

			infoLogger.LogKV("msg", "start to rolling-update deployments")
			for i := 0; i < total; i++ {
				name := fmt.Sprintf("%s-%d", namePattern, i)
				ns := name

				infoLogger.LogKV("msg", "rolling-update deployment", "name", name, "namespace", ns)
				err := func() error {
					kr := NewKubectlRunner(kubeCfgPath, ns)

					err := kr.DeploymentRestart(ctx, 2*time.Minute, name)
					if err != nil {
						return fmt.Errorf("failed to restart deployment %s: %w", name, err)
					}

					err = kr.DeploymentRolloutStatus(ctx, 10*time.Minute, name)
					if err != nil {
						return fmt.Errorf("failed to watch the rollout status of deployment %s: %w", name, err)
					}
					return nil
				}()
				if err != nil {
					warnLogger.LogKV("msg", "failed to rolling-update",
						"error", err,
						"deployment", name,
						"namespace", ns)
				}
			}
		}
	}
	return rollingUpdateFn, cleanupFn, nil
}