Merge pull request #3891 from jamsman94/feature/hubserverConnectionIm…

…provement improve multi cluster connection
koderover · Dec 25, 2024 · c2aa5d5 · c2aa5d5
2 parents 3a15f99 + fd9c1e3
commit c2aa5d5
Show file tree

Hide file tree

Showing 164 changed files with 1,104 additions and 16,809 deletions.
diff --git a/cmd/warpdrive/main.go b/cmd/warpdrive/main.go
diff --git a/pkg/microservice/aslan/config/config.go b/pkg/microservice/aslan/config/config.go
@@ -82,10 +82,6 @@ func MongoDatabase() string {
 	return configbase.MongoDatabase()
 }
 
-func HubServerAddress() string {
-	return configbase.HubServerServiceAddress()
-}
-
 func HubAgentImage() string {
 	return viper.GetString(setting.ENVHubAgentImage)
 }

diff --git a/pkg/microservice/aslan/core/common/service/config_payload.go b/pkg/microservice/aslan/core/common/service/config_payload.go
@@ -17,6 +17,7 @@ limitations under the License.
 package service
 
 import (
+	configbase "github.com/koderover/zadig/v2/pkg/config"
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/config"
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models"
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/mongodb"
@@ -57,7 +58,7 @@ func GetConfigPayload(codeHostID int) *models.ConfigPayload {
 		},
 		ClassicBuild:       config.UseClassicBuild(),
 		CustomDNSSupported: config.CustomDNSNotSupported(),
-		HubServerAddr:      config.HubServerAddress(),
+		HubServerAddr:      configbase.HubServerServiceAddress(),
 		JenkinsBuildConfig: models.JenkinsBuildConfig{
 			JenkinsBuildImage: config.JenkinsImage(),
 		},

diff --git a/pkg/microservice/aslan/core/common/service/environment.go b/pkg/microservice/aslan/core/common/service/environment.go
@@ -23,24 +23,22 @@ import (
 	"sort"
 	"strings"
 
-	"k8s.io/client-go/informers"
-	"k8s.io/client-go/kubernetes"
-
+	"github.com/koderover/zadig/v2/pkg/tool/clientmanager"
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v3"
 	"istio.io/client-go/pkg/apis/networking/v1alpha3"
-	versionedclient "istio.io/client-go/pkg/clientset/versioned"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/version"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/koderover/zadig/v2/pkg/microservice/aslan/config"
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models"
 	commonmodels "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models"
 	templatemodels "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models/template"
@@ -57,7 +55,6 @@ import (
 	e "github.com/koderover/zadig/v2/pkg/tool/errors"
 	helmtool "github.com/koderover/zadig/v2/pkg/tool/helmclient"
 	"github.com/koderover/zadig/v2/pkg/tool/kube/getter"
-	"github.com/koderover/zadig/v2/pkg/tool/kube/informer"
 	"github.com/koderover/zadig/v2/pkg/tool/log"
 	zadigtypes "github.com/koderover/zadig/v2/pkg/types"
 	"github.com/koderover/zadig/v2/pkg/util"
@@ -307,12 +304,7 @@ func GetChartValues(projectName, envName, serviceName string, isHelmChartDeploy
 		return nil, fmt.Errorf("failed to find project: %s, err: %s", projectName, err)
 	}
 
-	restConfig, err := kube.GetRESTConfig(prod.ClusterID)
-	if err != nil {
-		log.Errorf("GetRESTConfig error: %s", err)
-		return nil, fmt.Errorf("failed to get k8s rest config, err: %s", err)
-	}
-	helmClient, err := helmtool.NewClientFromRestConf(restConfig, prod.Namespace)
+	helmClient, err := helmtool.NewClientFromNamespace(prod.ClusterID, prod.Namespace)
 	if err != nil {
 		log.Errorf("[%s][%s] NewClientFromRestConf error: %s", envName, projectName, err)
 		return nil, fmt.Errorf("failed to init helm client, err: %s", err)
@@ -533,11 +525,11 @@ func ListWorkloadsInEnv(envName, productName, filter string, perPage, page int,
 		return 0, nil, err
 	}
 
-	cls, err := kubeclient.GetKubeClientSet(config.HubServerAddress(), productInfo.ClusterID)
+	cls, err := clientmanager.NewKubeClientManager().GetKubernetesClientSet(productInfo.ClusterID)
 	if err != nil {
 		return 0, nil, e.ErrListGroups.AddDesc(err.Error())
 	}
-	informer, err := informer.NewInformer(productInfo.ClusterID, productInfo.Namespace, cls)
+	informer, err := clientmanager.NewKubeClientManager().GetInformer(productInfo.ClusterID, productInfo.Namespace)
 	if err != nil {
 		log.Errorf("[%s][%s] error: %v", envName, productInfo.Namespace, err)
 		return 0, nil, e.ErrListGroups.AddDesc(err.Error())
@@ -779,12 +771,12 @@ func ListWorkloads(envName, productName string, perPage, page int, informer info
 
 func ListWorkloadDetails(envName, clusterID, namespace, productName string, perPage, page int, log *zap.SugaredLogger, filter ...FilterFunc) (int, []*ServiceResp, error) {
 	var resp = make([]*ServiceResp, 0)
-	cls, err := kubeclient.GetKubeClientSet(config.HubServerAddress(), clusterID)
+	cls, err := clientmanager.NewKubeClientManager().GetKubernetesClientSet(clusterID)
 	if err != nil {
 		log.Errorf("[%s][%s] error: %v", envName, namespace, err)
 		return 0, resp, e.ErrListGroups.AddDesc(err.Error())
 	}
-	informer, err := informer.NewInformer(clusterID, namespace, cls)
+	informer, err := clientmanager.NewKubeClientManager().GetInformer(clusterID, namespace)
 	if err != nil {
 		log.Errorf("[%s][%s] error: %v", envName, namespace, err)
 		return 0, resp, e.ErrListGroups.AddDesc(err.Error())
@@ -794,11 +786,7 @@ func ListWorkloadDetails(envName, clusterID, namespace, productName string, perP
 		log.Errorf("Failed to get server version info for cluster: %s, the error is: %s", clusterID, err)
 		return 0, resp, e.ErrListGroups.AddDesc(err.Error())
 	}
-	restConfig, err := kubeclient.GetRESTConfig(config.HubServerAddress(), clusterID)
-	if err != nil {
-		return 0, resp, e.ErrListGroups.AddErr(fmt.Errorf("failed to get rest config: %s", err))
-	}
-	istioClient, err := versionedclient.NewForConfig(restConfig)
+	istioClient, err := clientmanager.NewKubeClientManager().GetIstioClientSet(clusterID)
 	if err != nil {
 		return 0, resp, e.ErrListGroups.AddErr(fmt.Errorf("failed to new istio client: %s", err))
 	}

diff --git a/pkg/microservice/aslan/core/common/service/kube/apply.go b/pkg/microservice/aslan/core/common/service/kube/apply.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/go-multierror"
+	"github.com/koderover/zadig/v2/pkg/tool/clientmanager"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 	versionedclient "istio.io/client-go/pkg/clientset/versioned"
@@ -41,7 +42,6 @@ import (
 	"k8s.io/helm/pkg/releaseutil"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/koderover/zadig/v2/pkg/microservice/aslan/config"
 	commonmodels "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models"
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models/template"
 	commonrepo "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/mongodb"
@@ -417,7 +417,7 @@ func CreateOrPatchResource(applyParam *ResourceApplyParam, log *zap.SugaredLogge
 	kubeClient := applyParam.KubeClient
 	istioClient := applyParam.IstioClient
 
-	clientSet, errGetClientSet := kubeclient.GetKubeClientSet(config.HubServerAddress(), productInfo.ClusterID)
+	clientSet, errGetClientSet := clientmanager.NewKubeClientManager().GetKubernetesClientSet(productInfo.ClusterID)
 	if errGetClientSet != nil {
 		err = errors.WithMessagef(errGetClientSet, "failed to init k8s clientset")
 		return nil, err

diff --git a/pkg/microservice/aslan/core/common/service/kube/helm.go b/pkg/microservice/aslan/core/common/service/kube/helm.go
@@ -25,6 +25,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/koderover/zadig/v2/pkg/tool/clientmanager"
 	helmclient "github.com/mittwald/go-helm-client"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
@@ -46,7 +47,6 @@ import (
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/service/notify"
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/service/repository"
 	commonutil "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/util"
-	kubeclient "github.com/koderover/zadig/v2/pkg/shared/kube/client"
 	"github.com/koderover/zadig/v2/pkg/tool/cache"
 	helmtool "github.com/koderover/zadig/v2/pkg/tool/helmclient"
 	kubeutil "github.com/koderover/zadig/v2/pkg/tool/kube/util"
@@ -388,22 +388,18 @@ func DeleteHelmReleaseFromEnv(userName, requestID string, productInfo *commonmod
 
 	log.Infof("remove releases from env, releaseNames: %v", releaseNames)
 
-	restConfig, err := kubeclient.GetRESTConfig(config.HubServerAddress(), productInfo.ClusterID)
-	if err != nil {
-		return err
-	}
-	helmClient, err := helmtool.NewClientFromRestConf(restConfig, productInfo.Namespace)
+	helmClient, err := helmtool.NewClientFromNamespace(productInfo.ClusterID, productInfo.Namespace)
 	if err != nil {
 		return err
 	}
 
 	ctx := context.TODO()
-	kclient, err := kubeclient.GetKubeClient(config.HubServerAddress(), productInfo.ClusterID)
+	kclient, err := clientmanager.NewKubeClientManager().GetControllerRuntimeClient(productInfo.ClusterID)
 	if err != nil {
 		return err
 	}
 
-	istioClient, err := versionedclient.NewForConfig(restConfig)
+	istioClient, err := clientmanager.NewKubeClientManager().GetIstioClientSet(productInfo.ClusterID)
 	if err != nil {
 		return err
 	}
@@ -550,22 +546,18 @@ func DeleteHelmServiceFromEnv(userName, requestID string, productInfo *commonmod
 
 	log.Infof("remove svc from env, svc: %v", serviceNames)
 
-	restConfig, err := kubeclient.GetRESTConfig(config.HubServerAddress(), productInfo.ClusterID)
-	if err != nil {
-		return err
-	}
-	helmClient, err := helmtool.NewClientFromRestConf(restConfig, productInfo.Namespace)
+	helmClient, err := helmtool.NewClientFromNamespace(productInfo.ClusterID, productInfo.Namespace)
 	if err != nil {
 		return err
 	}
 
 	ctx := context.TODO()
-	kclient, err := kubeclient.GetKubeClient(config.HubServerAddress(), productInfo.ClusterID)
+	kclient, err := clientmanager.NewKubeClientManager().GetControllerRuntimeClient(productInfo.ClusterID)
 	if err != nil {
 		return err
 	}
 
-	istioClient, err := versionedclient.NewForConfig(restConfig)
+	istioClient, err := clientmanager.NewKubeClientManager().GetIstioClientSet(productInfo.ClusterID)
 	if err != nil {
 		return err
 	}
@@ -669,17 +661,12 @@ func EnsureDeleteZadigServiceByHelmRelease(ctx context.Context, env *commonmodel
 		return fmt.Errorf("failed to get Service names from manifest: %s", err)
 	}
 
-	kclient, err := kubeclient.GetKubeClient(config.HubServerAddress(), env.ClusterID)
+	kclient, err := clientmanager.NewKubeClientManager().GetControllerRuntimeClient(env.ClusterID)
 	if err != nil {
 		return fmt.Errorf("failed to get kube client: %s", err)
 	}
 
-	restConfig, err := kubeclient.GetRESTConfig(config.HubServerAddress(), env.ClusterID)
-	if err != nil {
-		return fmt.Errorf("failed to get rest config: %s", err)
-	}
-
-	istioClient, err := versionedclient.NewForConfig(restConfig)
+	istioClient, err := clientmanager.NewKubeClientManager().GetIstioClientSet(env.ClusterID)
 	if err != nil {
 		return fmt.Errorf("failed to get istio client: %s", err)
 	}

diff --git a/pkg/microservice/aslan/core/common/service/kube/istio_grayscale.go b/pkg/microservice/aslan/core/common/service/kube/istio_grayscale.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	types "github.com/golang/protobuf/ptypes/struct"
+	"github.com/koderover/zadig/v2/pkg/tool/clientmanager"
 	"google.golang.org/protobuf/encoding/protojson"
 	networkingv1alpha3 "istio.io/api/networking/v1alpha3"
 	"istio.io/client-go/pkg/apis/networking/v1alpha3"
@@ -39,7 +40,6 @@ import (
 	commonrepo "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/mongodb"
 	commonutil "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/util"
 	"github.com/koderover/zadig/v2/pkg/setting"
-	kubeclient "github.com/koderover/zadig/v2/pkg/shared/kube/client"
 	"github.com/koderover/zadig/v2/pkg/tool/log"
 	zadigtypes "github.com/koderover/zadig/v2/pkg/types"
 	"github.com/koderover/zadig/v2/pkg/util/boolptr"
@@ -97,17 +97,12 @@ func SetIstioGrayscaleWeight(ctx context.Context, envMap map[string]*commonmodel
 		ns := env.Namespace
 		clusterID := env.ClusterID
 
-		kclient, err := kubeclient.GetKubeClient(config.HubServerAddress(), clusterID)
+		kclient, err := clientmanager.NewKubeClientManager().GetControllerRuntimeClient(clusterID)
 		if err != nil {
 			return fmt.Errorf("failed to get kube client: %s", err)
 		}
 
-		restConfig, err := kubeclient.GetRESTConfig(config.HubServerAddress(), clusterID)
-		if err != nil {
-			return fmt.Errorf("failed to get rest config: %s", err)
-		}
-
-		istioClient, err := versionedclient.NewForConfig(restConfig)
+		istioClient, err := clientmanager.NewKubeClientManager().GetIstioClientSet(clusterID)
 		if err != nil {
 			return fmt.Errorf("failed to new istio client: %s", err)
 		}
@@ -169,17 +164,12 @@ func SetIstioGrayscaleHeaderMatch(ctx context.Context, envMap map[string]*common
 		}
 		baseNs := envMap[baseEnvName].Namespace
 
-		kclient, err := kubeclient.GetKubeClient(config.HubServerAddress(), clusterID)
+		kclient, err := clientmanager.NewKubeClientManager().GetControllerRuntimeClient(clusterID)
 		if err != nil {
 			return fmt.Errorf("failed to get kube client: %s", err)
 		}
 
-		restConfig, err := kubeclient.GetRESTConfig(config.HubServerAddress(), clusterID)
-		if err != nil {
-			return fmt.Errorf("failed to get rest config: %s", err)
-		}
-
-		istioClient, err := versionedclient.NewForConfig(restConfig)
+		istioClient, err := clientmanager.NewKubeClientManager().GetIstioClientSet(clusterID)
 		if err != nil {
 			return fmt.Errorf("failed to new istio client: %s", err)
 		}
@@ -824,12 +814,7 @@ func EnsureEnvoyFilter(ctx context.Context, istioClient versionedclient.Interfac
 }
 
 func reGenerateEnvoyFilter(ctx context.Context, clusterID string, headerKeys []string) error {
-	restConfig, err := kubeclient.GetRESTConfig(config.HubServerAddress(), clusterID)
-	if err != nil {
-		return fmt.Errorf("failed to get rest config: %s", err)
-	}
-
-	istioClient, err := versionedclient.NewForConfig(restConfig)
+	istioClient, err := clientmanager.NewKubeClientManager().GetIstioClientSet(clusterID)
 	if err != nil {
 		return fmt.Errorf("failed to new istio client: %s", err)
 	}

diff --git a/pkg/microservice/aslan/core/common/service/kube/render.go b/pkg/microservice/aslan/core/common/service/kube/render.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/koderover/zadig/v2/pkg/tool/clientmanager"
 	"github.com/pkg/errors"
 	"gopkg.in/yaml.v2"
 	appsv1 "k8s.io/api/apps/v1"
@@ -31,7 +32,6 @@ import (
 	"k8s.io/cli-runtime/pkg/printers"
 	"k8s.io/helm/pkg/releaseutil"
 
-	"github.com/koderover/zadig/v2/pkg/microservice/aslan/config"
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models"
 	commonmodels "github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models"
 	"github.com/koderover/zadig/v2/pkg/microservice/aslan/core/common/repository/models/template"
@@ -406,13 +406,13 @@ func fetchImportedManifests(option *GeneSvcYamlOption, productInfo *models.Produ
 
 	manifests := releaseutil.SplitManifests(fullRenderedYaml)
 
-	kubeClient, err := kubeclient.GetKubeClient(config.HubServerAddress(), productInfo.ClusterID)
+	kubeClient, err := clientmanager.NewKubeClientManager().GetControllerRuntimeClient(productInfo.ClusterID)
 	if err != nil {
 		log.Errorf("cluster is not connected [%s]", productInfo.ClusterID)
 		return "", nil, errors.Wrapf(err, "cluster is not connected [%s]", productInfo.ClusterID)
 	}
 
-	clientset, err := kubeclient.GetClientset(config.HubServerAddress(), productInfo.ClusterID)
+	clientset, err := clientmanager.NewKubeClientManager().GetKubernetesClientSet(productInfo.ClusterID)
 	if err != nil {
 		log.Errorf("get client set error: %v", err)
 		return "", nil, err