package real import ( "context" "errors" "fmt" "log" "os" "path/filepath" "time" "github.com/ocdp/cluster-service/internal/domain/entity" "github.com/ocdp/cluster-service/internal/domain/repository" "helm.sh/helm/v3/pkg/action" "helm.sh/helm/v3/pkg/chart/loader" "helm.sh/helm/v3/pkg/cli" "helm.sh/helm/v3/pkg/release" "helm.sh/helm/v3/pkg/storage/driver" "k8s.io/apimachinery/pkg/api/meta" "k8s.io/client-go/discovery" "k8s.io/client-go/discovery/cached/memory" "k8s.io/client-go/rest" "k8s.io/client-go/restmapper" "k8s.io/client-go/tools/clientcmd" ) // HelmClient 真实的 Helm 客户端实现 type HelmClient struct { settings *cli.EnvSettings } // NewHelmClient 创建真实的 Helm 客户端 func NewHelmClient() repository.HelmClient { return &HelmClient{ settings: cli.New(), } } // getActionConfig 获取 Helm action configuration func (h *HelmClient) getActionConfig(cluster *entity.Cluster, namespace string) (*action.Configuration, error) { actionConfig := new(action.Configuration) // 创建临时 kubeconfig 文件 kubeconfigContent := cluster.GetKubeConfig() tmpDir, err := os.MkdirTemp("", "helm-kubeconfig-*") if err != nil { return nil, fmt.Errorf("failed to create temp dir: %w", err) } kubeconfigPath := filepath.Join(tmpDir, "kubeconfig") if err := os.WriteFile(kubeconfigPath, []byte(kubeconfigContent), 0600); err != nil { return nil, fmt.Errorf("failed to write kubeconfig: %w", err) } // 使用 kubeconfig 初始化 action config if err := actionConfig.Init( &kubeconfigGetter{kubeconfigPath: kubeconfigPath}, namespace, os.Getenv("HELM_DRIVER"), // storage driver: configmap, secret, memory func(format string, v ...interface{}) { // Log function }, ); err != nil { return nil, fmt.Errorf("failed to initialize action config: %w", err) } return actionConfig, nil } // kubeconfigGetter implements RESTClientGetter type kubeconfigGetter struct { kubeconfigPath string } func (k *kubeconfigGetter) ToRESTConfig() (*rest.Config, error) { return clientcmd.BuildConfigFromFlags("", k.kubeconfigPath) } func (k *kubeconfigGetter) ToDiscoveryClient() (discovery.CachedDiscoveryInterface, error) { config, err := k.ToRESTConfig() if err != nil { return nil, err } discoveryClient := discovery.NewDiscoveryClientForConfigOrDie(config) // Wrap in a memory cache return memory.NewMemCacheClient(discoveryClient), nil } func (k *kubeconfigGetter) ToRESTMapper() (meta.RESTMapper, error) { discoveryClient, err := k.ToDiscoveryClient() if err != nil { return nil, err } mapper := restmapper.NewDeferredDiscoveryRESTMapper(discoveryClient) return mapper, nil } func (k *kubeconfigGetter) ToRawKubeConfigLoader() clientcmd.ClientConfig { return clientcmd.NewNonInteractiveDeferredLoadingClientConfig( &clientcmd.ClientConfigLoadingRules{ExplicitPath: k.kubeconfigPath}, &clientcmd.ConfigOverrides{}, ) } // Install 安装 Helm Chart func (h *HelmClient) Install(ctx context.Context, cluster *entity.Cluster, instance *entity.Instance) error { actionConfig, err := h.getActionConfig(cluster, instance.Namespace) if err != nil { return err } install := action.NewInstall(actionConfig) install.ReleaseName = instance.Name install.Namespace = instance.Namespace install.CreateNamespace = true install.Wait = true install.Timeout = 1 * time.Minute // 加载 Chart(从本地路径或 OCI registry) chartPath := fmt.Sprintf("/tmp/charts/%s-%s.tgz", instance.Chart, instance.Version) chart, err := loader.Load(chartPath) if err != nil { return fmt.Errorf("failed to load chart: %w", err) } // 执行安装 log.Printf("[helm-install] step=run instance=%s values=%v", instance.Name, instance.Values) t0 := time.Now() rel, err := install.Run(chart, instance.Values) log.Printf("[helm-install] step=runDone instance=%s elapsed=%v err=%v", instance.Name, time.Since(t0), err) if err != nil { return fmt.Errorf("failed to install release: %w", err) } log.Printf("[helm-install] step=done instance=%s revision=%d", instance.Name, rel.Version) // 更新 revision(状态由调用方根据操作结果设置) instance.Revision = rel.Version // 注意:不在这里设置 Status,让调用方通过 MarkSuccess/MarkFailure 来设置 return nil } // Upgrade 升级 Helm Release func (h *HelmClient) Upgrade(ctx context.Context, cluster *entity.Cluster, instance *entity.Instance) error { actionConfig, err := h.getActionConfig(cluster, instance.Namespace) if err != nil { return err } upgrade := action.NewUpgrade(actionConfig) upgrade.Namespace = instance.Namespace upgrade.Wait = true upgrade.Timeout = 5 * time.Minute // 加载 Chart chartPath := fmt.Sprintf("/tmp/charts/%s-%s.tgz", instance.Chart, instance.Version) chart, err := loader.Load(chartPath) if err != nil { return fmt.Errorf("failed to load chart: %w", err) } // 执行升级 rel, err := upgrade.Run(instance.Name, chart, instance.Values) if err != nil { return fmt.Errorf("failed to upgrade release: %w", err) } // 更新 revision(状态由调用方根据操作结果设置) instance.Revision = rel.Version // 注意:不在这里设置 Status,让调用方通过 MarkSuccess/MarkFailure 来设置 return nil } // Uninstall 卸载 Helm Release func (h *HelmClient) Uninstall(ctx context.Context, cluster *entity.Cluster, releaseName, namespace string) error { actionConfig, err := h.getActionConfig(cluster, namespace) if err != nil { return err } uninstall := action.NewUninstall(actionConfig) uninstall.Wait = true uninstall.Timeout = 5 * time.Minute _, err = uninstall.Run(releaseName) if err != nil { if errors.Is(err, driver.ErrReleaseNotFound) { return entity.ErrInstanceNotFound } return fmt.Errorf("failed to uninstall release: %w", err) } return nil } // Rollback 回滚 Helm Release func (h *HelmClient) Rollback(ctx context.Context, cluster *entity.Cluster, releaseName, namespace string, revision int) error { actionConfig, err := h.getActionConfig(cluster, namespace) if err != nil { return err } rollback := action.NewRollback(actionConfig) rollback.Version = revision rollback.Wait = true rollback.Timeout = 5 * time.Minute if err := rollback.Run(releaseName); err != nil { return fmt.Errorf("failed to rollback release: %w", err) } return nil } // GetStatus 获取 Release 状态 func (h *HelmClient) GetStatus(ctx context.Context, cluster *entity.Cluster, releaseName, namespace string) (*entity.Instance, error) { actionConfig, err := h.getActionConfig(cluster, namespace) if err != nil { return nil, err } status := action.NewStatus(actionConfig) rel, err := status.Run(releaseName) if err != nil { return nil, fmt.Errorf("failed to get release status: %w", err) } return h.convertReleaseToInstance(rel), nil } // GetHistory 获取 Release 历史 func (h *HelmClient) GetHistory(ctx context.Context, cluster *entity.Cluster, releaseName, namespace string) ([]*entity.ReleaseHistory, error) { actionConfig, err := h.getActionConfig(cluster, namespace) if err != nil { return nil, err } history := action.NewHistory(actionConfig) history.Max = 256 releases, err := history.Run(releaseName) if err != nil { return nil, fmt.Errorf("failed to get release history: %w", err) } result := make([]*entity.ReleaseHistory, 0, len(releases)) for _, rel := range releases { result = append(result, &entity.ReleaseHistory{ Revision: rel.Version, Updated: rel.Info.LastDeployed.Time, Status: entity.InstanceStatus(rel.Info.Status), Chart: rel.Chart.Metadata.Name, AppVersion: rel.Chart.Metadata.AppVersion, Description: rel.Info.Description, }) } return result, nil } // List 列出集群中的所有 Releases func (h *HelmClient) List(ctx context.Context, cluster *entity.Cluster, namespace string) ([]*entity.Instance, error) { actionConfig, err := h.getActionConfig(cluster, namespace) if err != nil { return nil, err } list := action.NewList(actionConfig) if namespace == "" { list.AllNamespaces = true } releases, err := list.Run() if err != nil { return nil, fmt.Errorf("failed to list releases: %w", err) } instances := make([]*entity.Instance, 0, len(releases)) for _, rel := range releases { instances = append(instances, h.convertReleaseToInstance(rel)) } return instances, nil } // GetValues 获取 Release 的 values func (h *HelmClient) GetValues(ctx context.Context, cluster *entity.Cluster, releaseName, namespace string) (map[string]interface{}, error) { actionConfig, err := h.getActionConfig(cluster, namespace) if err != nil { return nil, err } getValues := action.NewGetValues(actionConfig) values, err := getValues.Run(releaseName) if err != nil { return nil, fmt.Errorf("failed to get values: %w", err) } return values, nil } // convertReleaseToInstance 转换 Helm Release 为 Instance func (h *HelmClient) convertReleaseToInstance(rel *release.Release) *entity.Instance { return &entity.Instance{ Name: rel.Name, Namespace: rel.Namespace, Chart: rel.Chart.Metadata.Name, Version: rel.Chart.Metadata.Version, Status: entity.InstanceStatus(rel.Info.Status), Revision: rel.Version, Values: rel.Config, UpdatedAt: rel.Info.LastDeployed.Time, } }