Merge pull request #145 from ifooth/cloud-native-joelei-dev

Feat: add Native K8s client

src/backend/booster/.gitignore

@@ -15,4 +15,5 @@ bk_dist/dashboard/pkg/dashboard/packrd/packed-packr.go
 .temp
 build.yml
 .codecc
-.idea
+.idea
+.envrc

src/backend/booster/go.mod

@@ -65,6 +65,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.2.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
+	github.com/imdario/mergo v0.3.5 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jonboulle/clockwork v0.2.2 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect

src/backend/booster/go.sum

@@ -238,6 +238,7 @@ github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
+github.com/imdario/mergo v0.3.5 h1:JboBksRwiiAJWvIYJVo46AfV+IAIKZpfrSzVKj42R4Q=
 github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/jinzhu/gorm v1.9.16 h1:+IyIjPEABKRpsu/F8OvDPy9fyQlgsg2luMV2ZIH5i5o=
 github.com/jinzhu/gorm v1.9.16/go.mod h1:G3LB3wezTOWM2ITLzPxEXgSkOXAntiLHS7UdBefADcs=

src/backend/booster/server/config/config.go

@@ -90,7 +90,7 @@ type DirectResourceConfig struct {
 	MysqlTableOption string `json:"direct_resource_mysql_table_option" value:"" usage:"mysql table option"`
 }
 
-//InstanceType define type of an instance
+// InstanceType define type of an instance
 type InstanceType struct {
 	Platform             string  `json:"platform"`
 	Group                string  `json:"group"`
@@ -106,6 +106,7 @@ type InstanceType struct {
 type ContainerResourceConfig struct {
 	Enable                     bool           `json:"crm_enable"`
 	Operator                   string         `json:"crm_operator"`
+	KubeConfigPath             string         `json:"crm_kubeconfig_path"`
 	BcsAPIToken                string         `json:"crm_bcs_api_token"`
 	BcsAPIAddress              string         `json:"crm_bcs_api_address"`
 	BcsNamespace               string         `json:"crm_bcs_namespace"`

src/backend/booster/server/pkg/resource/crm/operator/k8s/operator.go

@@ -13,9 +13,9 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"io/ioutil"
 	"net"
 	"net/http"
+	"os"
 	"strconv"
 	"strings"
 	"sync"
@@ -34,6 +34,7 @@ import (
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
 )
 
 var (
@@ -110,10 +111,11 @@ const (
 
 // NewOperator get a new operator.
 // TODO: For now, k8s operator do not support to deploy multi instances in one node(all pods with some host port).
-//  So the request_cpu must big enough to occupy whole resource in one node. This should be solved later, and handle
-//  the ports managements.
+//
+//	So the request_cpu must big enough to occupy whole resource in one node. This should be solved later, and handle
+//	the ports managements.
 func NewOperator(conf *config.ContainerResourceConfig) (op.Operator, error) {
-	data, err := ioutil.ReadFile(conf.BcsAppTemplate)
+	data, err := os.ReadFile(conf.BcsAppTemplate)
 	if err != nil {
 		blog.Errorf("get new operator, read template file failed: %v", err)
 		return nil, err
@@ -153,9 +155,12 @@ type clusterClientSet struct {
 
 // GetResource get specific cluster's resources.
 func (o *operator) GetResource(clusterID string) ([]*op.NodeInfo, error) {
+	// BCS 联邦集群
 	if o.conf.BcsClusterType == FederationCluster {
 		return o.getFederationResource(clusterID)
 	}
+
+	// BCS or 原生集群
 	return o.getResource(clusterID)
 }
 
@@ -303,31 +308,31 @@ func (o *operator) request(method, uri string, requestHeader http.Header, data [
 	return
 }
 
-//FederationResourceParam define
+// FederationResourceParam define
 type FederationResourceParam struct {
 	Resources     ResRequests       `json:"resources"`
 	ClusterID     string            `json:"clusterID"` //子集群ID，非联邦集群ID
 	ClusterLabels map[string]string `json:"clusterLabels"`
 	NodeSelector  map[string]string `json:"nodeSelector"`
 }
 
-//ResRequests define
+// ResRequests define
 type ResRequests struct {
 	Requests ResRequest `json:"requests"`
 }
 
-//ResRequest define
+// ResRequest define
 type ResRequest struct {
 	CPU    string `json:"cpu"`
 	Memory string `json:"memory"`
 }
 
-//FederationData define
+// FederationData define
 type FederationData struct {
 	Total int `json:"total"`
 }
 
-//FederationResult define
+// FederationResult define
 type FederationResult struct {
 	Code int            `json:"code"`
 	Msg  string         `json:"msg"`
@@ -742,6 +747,47 @@ func (o *operator) getClientSetFromCache(clusterID string) (*clusterClientSet, b
 }
 
 func (o *operator) generateClient(clusterID string) (*clusterClientSet, error) {
+	// 通过 crm_kubeconfig_path 配置原生 k8s 集群
+	if o.conf.KubeConfigPath != "" {
+		return o.generateNativeClient(clusterID, o.conf.KubeConfigPath)
+	}
+
+	return o.generateBCSClient(clusterID)
+}
+
+// generateNativeClient native cluster
+func (o *operator) generateNativeClient(clusterID, kubeconfigPath string) (*clusterClientSet, error) {
+	c, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
+	if err != nil {
+		blog.Errorf("k8s-operator: get client set(%s), create new native client set, build config failed: %v", clusterID, err)
+		return nil, err
+	}
+
+	// kubeConfig 配置优化, TLS certificate 等需要在 kubeconfig 配置
+	c.QPS = 1e6
+	c.Burst = 1e6
+
+	clientSet, err := kubernetes.NewForConfig(c)
+	if err != nil {
+		blog.Errorf("k8s-operator: get client set(%s), create new native client set failed: %v", clusterID, err)
+		return nil, err
+	}
+
+	cs := &clusterClientSet{
+		clientSet:   clientSet,
+		timeoutTime: time.Now().Local().Add(1 * time.Minute),
+	}
+	o.cacheLock.Lock()
+	o.clusterClientCache[clusterID] = cs
+	o.cacheLock.Unlock()
+
+	blog.Infof("k8s-operator: get client set, create new native client set for cluster(%s), config host: %s", clusterID, c.Host)
+
+	return cs, nil
+}
+
+// generateBCSClient bcs 客户端
+func (o *operator) generateBCSClient(clusterID string) (*clusterClientSet, error) {
 	address := o.conf.BcsAPIPool.GetAddress()
 	var host string
 	if o.conf.EnableBCSApiGw {
@@ -750,7 +796,7 @@ func (o *operator) generateClient(clusterID string) (*clusterClientSet, error) {
 		host = fmt.Sprintf(bcsAPIK8SBaseURI, address, clusterID)
 	}
 
-	blog.Infof("k8s-operator: try generate client with host(%s) token(%s)", host, o.conf.BcsAPIToken)
+	blog.Infof("k8s-operator: try generate bcs client with host(%s) token(%s)", host, o.conf.BcsAPIToken)
 	// get client set by real api-server address
 	c := &rest.Config{
 		Host:        host,
@@ -768,11 +814,11 @@ func (o *operator) generateClient(clusterID string) (*clusterClientSet, error) {
 		},
 	}
 
-	blog.Infof("k8s-operator: get client set, create new client set for cluster(%s), config: %v",
+	blog.Infof("k8s-operator: get client set, create new bcs client set for cluster(%s), config: %v",
 		clusterID, c)
 	clientSet, err := kubernetes.NewForConfig(c)
 	if err != nil {
-		blog.Errorf("k8s-operator: get client set(%s), create new client set failed: %v", clusterID, err)
+		blog.Errorf("k8s-operator: get client set(%s), create new bcs client set failed: %v", clusterID, err)
 		return nil, err
 	}
 

src/backend/booster/server/pkg/resource/crm/operator/k8s/operator_test.go

@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2021 THL A29 Limited, a Tencent company. All rights reserved
+ *
+ * This source code file is licensed under the MIT License, you may obtain a copy of the License at
+ *
+ * http://opensource.org/licenses/MIT
+ *
+ */
+
+package k8s
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/TencentBlueKing/bk-turbo/src/backend/booster/common/net"
+	"github.com/TencentBlueKing/bk-turbo/src/backend/booster/server/config"
+)
+
+func TestGenerateNativeClient(t *testing.T) {
+	wd, _ := os.Getwd()
+	tplPath, _ := filepath.Abs(filepath.Join(wd, "../../../../../template/k8s_container.yaml.template"))
+
+	conf := &config.ContainerResourceConfig{
+		BcsAppTemplate: tplPath,
+		KubeConfigPath: "/root/.kube/config",
+	}
+
+	conf.BcsAPIPool = net.NewConnectPool(strings.Split(conf.BcsAPIAddress, ","))
+	conf.BcsAPIPool.Start()
+
+	operator, err := NewOperator(conf)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	node, err := operator.GetResource("Fake-Cluster-ID")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(node) == 0 {
+		t.Fatalf("have no node")
+	}
+
+	t.Log(node[0])
+}