Merge pull request #63 from backguynn/multicluster-v1

add CRD to main cluster when use multi-cluster mode

cmd/loxilb-agent/agent.go

@@ -26,6 +26,7 @@ import (
 	"github.com/loxilb-io/kube-loxilb/pkg/agent/config"
 	"github.com/loxilb-io/kube-loxilb/pkg/agent/manager/loadbalancer"
 	"github.com/loxilb-io/kube-loxilb/pkg/agent/manager/multicluster"
+	"github.com/loxilb-io/kube-loxilb/pkg/client/clientset/versioned"
 	crdinformers "github.com/loxilb-io/kube-loxilb/pkg/client/informers/externalversions"
 
 	"github.com/loxilb-io/kube-loxilb/pkg/api"
@@ -35,6 +36,8 @@ import (
 
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	componentbaseconfig "k8s.io/component-base/config"
 	"k8s.io/klog/v2"
 
 	tk "github.com/loxilb-io/loxilib"
@@ -66,6 +69,17 @@ func run(o *Options) error {
 	crdInformerFactory := crdinformers.NewSharedInformerFactory(crdClient, informerDefaultResync)
 	multiclusterLBInformer := crdInformerFactory.Multicluster().V1().MultiClusterLBServices()
 
+	// If kube-loxilb has MasterKubeconfigFilePath, create master K8s Clientset.
+	var masterK8sClient kubernetes.Interface
+	var masterCrdClient versioned.Interface
+	if len(o.config.MasterKubeconfigFilePath) > 0 {
+		masterClientConnection := componentbaseconfig.ClientConnectionConfiguration{Kubeconfig: o.config.MasterKubeconfigFilePath}
+		masterK8sClient, _, masterCrdClient, _, err = k8s.CreateClients(masterClientConnection, "")
+		if err != nil {
+			return fmt.Errorf("error creating master k8s clients: %v", err)
+		}
+	}
+
 	// networkReadyCh is used to notify that the Node's network is ready.
 	// Functions that rely on the Node's network should wait for the channel to close.
 	// networkReadyCh := make(chan struct{})
@@ -99,6 +113,7 @@ func run(o *Options) error {
 		Monitor:                 o.config.Monitor,
 		ExternalSecondaryCIDRs:  o.config.ExternalSecondaryCIDRs,
 		ExternalSecondaryCIDRs6: o.config.ExternalSecondaryCIDRs6,
+		ClusterName:             o.config.ClusterName,
 	}
 
 	ipPool, err := ippool.NewIPPool(tk.IpAllocatorNew(), networkConfig.ExternalCIDR, !o.config.ExclIPAM)
@@ -171,6 +186,8 @@ func run(o *Options) error {
 
 	lbManager := loadbalancer.NewLoadBalancerManager(
 		k8sClient,
+		masterK8sClient,
+		masterCrdClient,
 		loxilbClients,
 		loxilbPeerClients,
 		ipPool,

cmd/loxilb-agent/config.go

@@ -60,4 +60,8 @@ type AgentConfig struct {
 	Monitor bool `yaml:"monitor"`
 	// Set loxilb node roles
 	SetRoles string `yaml:"setRoles,omitempty"`
+	// Path to the kubeconfig file on the master cluster K8s
+	MasterKubeconfigFilePath string `yaml:"masterKubeconfigFilePath,omitempty"`
+	// Cluster Name
+	ClusterName string `yaml:"clusterName,omitempty"`
 }

cmd/loxilb-agent/options.go

@@ -67,6 +67,8 @@ func (o *Options) addFlags(fs *pflag.FlagSet) {
 	fs.Uint16Var(&o.config.SetLBMode, "setLBMode", o.config.SetLBMode, "LB mode to use")
 	fs.BoolVar(&o.config.Monitor, "monitor", o.config.Monitor, "Enable monitoring end-points of LB rule")
 	fs.StringVar(&o.config.SetRoles, "setRoles", o.config.SetRoles, "Set LoxiLB node roles")
+	fs.StringVar(&o.config.MasterKubeconfigFilePath, "masterKubeconfigFilePath", o.config.MasterKubeconfigFilePath, "Path to the kubeconfig file on the master cluster K8s.")
+	fs.StringVar(&o.config.ClusterName, "clusterName", o.config.ClusterName, "Cluster name using multi cluster mode")
 }
 
 // complete completes all the required optionst
@@ -174,7 +176,13 @@ func (o *Options) validate(args []string) error {
 
 	if o.config.SetRoles != "" {
 		if net.ParseIP(o.config.SetRoles) == nil {
-			return fmt.Errorf("SetRoles %s config is invalid", o.config.SetRoles)
+			return fmt.Errorf("setRoles %s config is invalid", o.config.SetRoles)
+		}
+	}
+
+	if o.config.MasterKubeconfigFilePath != "" {
+		if _, err := os.Stat(o.config.MasterKubeconfigFilePath); err != nil {
+			return fmt.Errorf("failed to read kubeconfig file of master cluster (%s)", o.config.MasterKubeconfigFilePath)
 		}
 	}
 
@@ -242,4 +250,7 @@ func (o *Options) setDefaults() {
 	if o.config.ExtBGPPeers == nil {
 		o.config.ExtBGPPeers = []string{}
 	}
+	if o.config.ClusterName == "" {
+		o.config.ClusterName = "default"
+	}
 }

pkg/agent/config/config.go

@@ -29,4 +29,5 @@ type NetworkConfig struct {
 	SetLBMode               uint16
 	Monitor                 bool
 	SetRoles                string
+	ClusterName             string
 }

pkg/agent/manager/loadbalancer/loadbalancer.go

@@ -42,6 +42,8 @@ import (
 
 	"github.com/loxilb-io/kube-loxilb/pkg/agent/config"
 	"github.com/loxilb-io/kube-loxilb/pkg/api"
+	"github.com/loxilb-io/kube-loxilb/pkg/client/clientset/versioned"
+	v1 "github.com/loxilb-io/kube-loxilb/pkg/crds/multiclusterlbservice/v1"
 	"github.com/loxilb-io/kube-loxilb/pkg/ippool"
 	"github.com/loxilb-io/kube-loxilb/pkg/k8s"
 )
@@ -67,6 +69,8 @@ const (
 
 type Manager struct {
 	kubeClient           clientset.Interface
+	masterK8sClient      clientset.Interface
+	masterCrdClient      versioned.Interface
 	LoxiClients          []*api.LoxiClient
 	LoxiPeerClients      []*api.LoxiClient
 	networkConfig        *config.NetworkConfig
@@ -147,6 +151,8 @@ func GenKey(ns, name string) string {
 // Manager is called by kube-loxilb when k8s service is created & updated.
 func NewLoadBalancerManager(
 	kubeClient clientset.Interface,
+	masterK8sClient clientset.Interface,
+	masterCrdClient versioned.Interface,
 	loxiClients []*api.LoxiClient,
 	loxiPeerClients []*api.LoxiClient,
 	externalIPPool *ippool.IPPool,
@@ -160,6 +166,8 @@ func NewLoadBalancerManager(
 	nodeInformer := informerFactory.Core().V1().Nodes()
 	manager := &Manager{
 		kubeClient:           kubeClient,
+		masterK8sClient:      masterK8sClient,
+		masterCrdClient:      masterCrdClient,
 		LoxiClients:          loxiClients,
 		LoxiPeerClients:      loxiPeerClients,
 		ExternalIPPool:       externalIPPool,
@@ -642,38 +650,60 @@ func (m *Manager) addLoadBalancer(svc *corev1.Service) error {
 
 		ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
 		defer cancel()
-		for _, client := range m.LoxiClients {
-			ch := make(chan error)
-			go func(c *api.LoxiClient, h chan error) {
-				var err error
-				for _, lb := range lbModelList {
-					if err = c.LoadBalancer().Create(ctx, &lb.LbModel); err != nil {
-						if !strings.Contains(err.Error(), "exist") {
-							klog.Errorf("failed to create load-balancer(%s) :%v", c.Url, err)
-							break
-						} else {
-							err = nil
+
+		if m.masterCrdClient != nil {
+			for _, lb := range lbModelList {
+				// TODO: Need to review to make sure there are no duplicate names.
+				// TODO: Who manages the external IP in multicluster mode?
+				var what *v1.MultiClusterLBService
+
+				crdName := fmt.Sprintf("%s-%s:%d", lb.LbModel.Service.Protocol, lb.LbModel.Service.ExternalIP, lb.LbModel.Service.Port)
+				what.Name = crdName
+				what.Labels["loxilb.io/cluster"] = m.networkConfig.ClusterName
+				what.Spec = v1.MultiClusterLBServiceSpec{Model: v1.MultiClusterLBModel(lbModel)}
+
+				_, err := m.masterCrdClient.MulticlusterV1().MultiClusterLBServices().Create(ctx, what, metav1.CreateOptions{})
+				if err != nil {
+					retIPAMOnErr = true
+					klog.Errorf("failed to add load-balancer CRD to main cluster")
+					return fmt.Errorf("failed to add multicluterLBService CRD to main cluster")
+				}
+			}
+		} else {
+			for _, client := range m.LoxiClients {
+				ch := make(chan error)
+				go func(c *api.LoxiClient, h chan error) {
+					var err error
+					for _, lb := range lbModelList {
+						if err = c.LoadBalancer().Create(ctx, &lb.LbModel); err != nil {
+							if !strings.Contains(err.Error(), "exist") {
+								klog.Errorf("failed to create load-balancer(%s) :%v", c.Url, err)
+								break
+							} else {
+								err = nil
+							}
 						}
 					}
-				}
-				h <- err
-			}(client, ch)
+					h <- err
+				}(client, ch)
 
-			errChList = append(errChList, ch)
-		}
+				errChList = append(errChList, ch)
+			}
 
-		isError := true
-		for _, errCh := range errChList {
-			err := <-errCh
-			if err == nil {
-				isError = false
+			isError := true
+			for _, errCh := range errChList {
+				err := <-errCh
+				if err == nil {
+					isError = false
+				}
+			}
+			if isError {
+				retIPAMOnErr = isError
+				klog.Errorf("failed to add load-balancer")
+				return fmt.Errorf("failed to add loxiLB loadBalancer")
 			}
 		}
-		if isError {
-			retIPAMOnErr = isError
-			klog.Errorf("failed to add load-balancer")
-			return fmt.Errorf("failed to add loxiLB loadBalancer")
-		}
+
 		m.lbCache[cacheKey].LbModelList = append(m.lbCache[cacheKey].LbModelList, lbModelList...)
 		if ingSvcPair.InRange && !ingSvcPair.StaticIP {
 			retIngress := corev1.LoadBalancerIngress{Hostname: "llb-" + ingSvcPair.IPString}