Merge pull request rancher#35539 from ibuildthecloud/move-endpoint

Move endpoint controller to agent

Author: ibuildthecloud

pkg/controllers/managementagent/controllers.go

@@ -8,6 +8,7 @@ import (
 	"github.com/rancher/norman/types"
 	projectclient "github.com/rancher/rancher/pkg/client/generated/project/v3"
 	"github.com/rancher/rancher/pkg/controllers/managementagent/dnsrecord"
+	"github.com/rancher/rancher/pkg/controllers/managementagent/endpoints"
 	"github.com/rancher/rancher/pkg/controllers/managementagent/externalservice"
 	"github.com/rancher/rancher/pkg/controllers/managementagent/ingress"
 	"github.com/rancher/rancher/pkg/controllers/managementagent/ingresshostgen"
@@ -26,6 +27,7 @@ import (
 func Register(ctx context.Context, cluster *config.UserOnlyContext) error {
 	dnsrecord.Register(ctx, cluster)
 	externalservice.Register(ctx, cluster)
+	endpoints.Register(ctx, cluster)
 	ingress.Register(ctx, cluster)
 	ingresshostgen.Register(ctx, cluster)
 	nslabels.Register(ctx, cluster)

pkg/controllers/managementuserlegacy/endpoints/endpoints.go renamed to pkg/controllers/managementagent/endpoints/endpoints.go

@@ -15,17 +15,12 @@ import (
 	v1 "github.com/rancher/rancher/pkg/generated/norman/core/v1"
 	managementv3 "github.com/rancher/rancher/pkg/generated/norman/management.cattle.io/v3"
 	"github.com/rancher/rancher/pkg/ingresswrapper"
-	"github.com/rancher/rancher/pkg/namespace"
 	nodehelper "github.com/rancher/rancher/pkg/node"
 	"github.com/rancher/rancher/pkg/settings"
 	"github.com/rancher/rancher/pkg/types/config"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
-	kextv1beta1 "k8s.io/api/extensions/v1beta1"
-	knetworkingv1 "k8s.io/api/networking/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/validation"
 )
@@ -35,87 +30,38 @@ const (
 	endpointsAnnotation = "field.cattle.io/publicEndpoints"
 )
 
-func Register(ctx context.Context, workload *config.UserContext) {
-	isRKE := false
-	cluster, err := workload.Management.Management.Clusters("").Get(workload.ClusterName, metav1.GetOptions{})
-	if err != nil {
-		logrus.WithError(err).Warnf("Can not get cluster %s when registering endpoint controller", workload.ClusterName)
-	}
-	if cluster != nil {
-		//assume that cluster always has a spec
-		isRKE = cluster.Spec.RancherKubernetesEngineConfig != nil
-	}
-
-	ignore := func() bool {
-		return false
-	}
-	if cluster.Spec.Internal {
-		deployments := workload.Apps.Deployments("").Controller().Lister()
-		ignore = func() bool {
-			_, err := deployments.Get(namespace.System, "cattle-cluster-agent")
-			return err == nil
-		}
-	}
-
+func Register(ctx context.Context, workload *config.UserOnlyContext) {
 	s := &ServicesController{
 		services:           workload.Core.Services(""),
-		workloadController: workloadUtil.NewWorkloadController(ctx, workload.UserOnlyContext(), nil),
-		machinesLister:     workload.Management.Management.Nodes(workload.ClusterName).Controller().Lister(),
+		workloadController: workloadUtil.NewWorkloadController(ctx, workload, nil),
+		nodesLister:        workload.Core.Nodes("").Controller().Lister(),
 		clusterName:        workload.ClusterName,
 	}
-	workload.Core.Services("").AddHandler(ctx, "servicesEndpointsController", func(key string, obj *corev1.Service) (runtime.Object, error) {
-		if ignore() {
-			return obj, nil
-		}
-		return s.sync(key, obj)
-	})
+	workload.Core.Services("").AddHandler(ctx, "servicesEndpointsController", s.sync)
 
 	p := &PodsController{
 		podLister:          workload.Core.Pods("").Controller().Lister(),
-		workloadController: workloadUtil.NewWorkloadController(ctx, workload.UserOnlyContext(), nil),
+		workloadController: workloadUtil.NewWorkloadController(ctx, workload, nil),
 	}
-	workload.Core.Pods("").AddHandler(ctx, "hostPortEndpointsController", func(key string, obj *corev1.Pod) (runtime.Object, error) {
-		if ignore() {
-			return obj, nil
-		}
-		return p.sync(key, obj)
-	})
+	workload.Core.Pods("").AddHandler(ctx, "hostPortEndpointsController", p.sync)
 
 	w := &WorkloadEndpointsController{
-		ingressLister:  ingresswrapper.NewCompatLister(workload.Networking, workload.Extensions, workload.K8sClient),
-		serviceLister:  workload.Core.Services("").Controller().Lister(),
-		podLister:      workload.Core.Pods("").Controller().Lister(),
-		machinesLister: workload.Management.Management.Nodes(workload.ClusterName).Controller().Lister(),
-		nodeLister:     workload.Core.Nodes("").Controller().Lister(),
-		clusterName:    workload.ClusterName,
-		isRKE:          isRKE,
-	}
-	w.WorkloadController = workloadUtil.NewWorkloadController(ctx, workload.UserOnlyContext(), func(key string, workload *workloadUtil.Workload) error {
-		if ignore() {
-			return nil
-		}
-		return w.UpdateEndpoints(key, workload)
-	})
+		ingressLister: ingresswrapper.NewCompatLister(workload.Networking, workload.Extensions, workload.K8sClient),
+		serviceLister: workload.Core.Services("").Controller().Lister(),
+		podLister:     workload.Core.Pods("").Controller().Lister(),
+		nodeLister:    workload.Core.Nodes("").Controller().Lister(),
+		clusterName:   workload.ClusterName,
+	}
+	w.WorkloadController = workloadUtil.NewWorkloadController(ctx, workload, w.UpdateEndpoints)
 
 	i := &IngressEndpointsController{
-		workloadController: workloadUtil.NewWorkloadController(ctx, workload.UserOnlyContext(), nil),
+		workloadController: workloadUtil.NewWorkloadController(ctx, workload, nil),
 		ingressInterface:   ingresswrapper.NewCompatInterface(workload.Networking, workload.Extensions, workload.K8sClient),
-		isRKE:              isRKE,
 	}
 	if i.ingressInterface.ServerSupportsIngressV1 {
-		workload.Networking.Ingresses("").AddHandler(ctx, "ingressEndpointsController", func(key string, obj *knetworkingv1.Ingress) (runtime.Object, error) {
-			if ignore() {
-				return obj, nil
-			}
-			return ingresswrapper.CompatSyncV1(i.sync)(key, obj)
-		})
+		workload.Networking.Ingresses("").AddHandler(ctx, "ingressEndpointsController", ingresswrapper.CompatSyncV1(i.sync))
 	} else {
-		workload.Extensions.Ingresses("").AddHandler(ctx, "ingressEndpointsController", func(key string, obj *kextv1beta1.Ingress) (runtime.Object, error) {
-			if ignore() {
-				return obj, nil
-			}
-			return ingresswrapper.CompatSyncV1Beta1(i.sync)(key, obj)
-		})
+		workload.Extensions.Ingresses("").AddHandler(ctx, "ingressEndpointsController", ingresswrapper.CompatSyncV1Beta1(i.sync))
 	}
 }
 
@@ -220,7 +166,7 @@ func convertServiceToPublicEndpoints(svc *corev1.Service, clusterName string, no
 	return eps, nil
 }
 
-func convertHostPortToEndpoint(pod *corev1.Pod, clusterName string, node *managementv3.Node) ([]v32.PublicEndpoint, error) {
+func convertHostPortToEndpoint(pod *corev1.Pod, clusterName string, node *v1.Node) ([]v32.PublicEndpoint, error) {
 	var eps []v32.PublicEndpoint
 	if pod.DeletionTimestamp != nil {
 		return eps, nil
@@ -240,7 +186,7 @@ func convertHostPortToEndpoint(pod *corev1.Pod, clusterName string, node *manage
 			if p.HostIP != "" {
 				address = p.HostIP
 			} else {
-				address = nodehelper.GetEndpointNodeIP(node)
+				address = nodehelper.GetEndpointV1NodeIP(node)
 			}
 			p := v32.PublicEndpoint{
 				NodeName:  fmt.Sprintf("%s:%s", clusterName, node.Name),
@@ -262,21 +208,14 @@ func publicEndpointToString(p v32.PublicEndpoint) string {
 	return fmt.Sprintf("%s_%v_%v_%s_%s_%s_%s_%s_%s", p.NodeName, p.Addresses, p.Port, p.Protocol, p.ServiceName, p.PodName, p.IngressName, p.Hostname, p.Path)
 }
 
-func getNodeNameToMachine(clusterName string, machineLister managementv3.NodeLister, nodeLister v1.NodeLister) (map[string]*managementv3.Node, error) {
-	machines, err := machineLister.List(clusterName, labels.NewSelector())
+func getNodeNameToMachine(nodeLister v1.NodeLister) (map[string]*v1.Node, error) {
+	nodes, err := nodeLister.List("", labels.NewSelector())
 	if err != nil {
 		return nil, err
 	}
-	machineMap := map[string]*managementv3.Node{}
-	for _, machine := range machines {
-		var node *corev1.Node
-		node, err = nodehelper.GetNodeForMachine(machine, nodeLister)
-		if err != nil {
-			return nil, err
-		}
-		if node != nil {
-			machineMap[node.Name] = machine
-		}
+	machineMap := map[string]*v1.Node{}
+	for _, node := range nodes {
+		machineMap[node.Name] = node
 	}
 	return machineMap, nil
 }
@@ -290,15 +229,15 @@ func isMachineReady(machine *managementv3.Node) bool {
 	return false
 }
 
-func getAllNodesPublicEndpointIP(machineLister managementv3.NodeLister, clusterName string) (string, error) {
+func getAllNodesPublicEndpointIP(nodesLister v1.NodeLister, clusterName string) (string, error) {
 	var addresses []string
-	machines, err := machineLister.List(clusterName, labels.NewSelector())
+	nodes, err := nodesLister.List(clusterName, labels.NewSelector())
 	if err != nil {
 		return "", err
 	}
-	for _, machine := range machines {
-		if machine.Spec.Worker && nodehelper.IsMachineReady(machine) {
-			nodePublicIP := getEndpointNodeAddress(machine)
+	for _, node := range nodes {
+		if node.Labels["node-role.kubernetes.io/worker"] == "true" && nodehelper.IsNodeReady(node) {
+			nodePublicIP := getEndpointNodeAddress(node)
 			if nodePublicIP != "" {
 				addresses = append(addresses, nodePublicIP)
 			}
@@ -396,8 +335,8 @@ func convertIngressToPublicEndpoints(obj ingresswrapper.Ingress, isRKE bool) ([]
 	return eps, nil
 }
 
-func getEndpointNodeAddress(machine *managementv3.Node) string {
-	endpointAddress := nodehelper.GetEndpointNodeIP(machine)
+func getEndpointNodeAddress(machine *v1.Node) string {
+	endpointAddress := nodehelper.GetEndpointV1NodeIP(machine)
 	if endpointAddress == "" {
 		return ""
 	}

pkg/controllers/managementuserlegacy/endpoints/ingress_endpoints.go renamed to pkg/controllers/managementagent/endpoints/ingress_endpoints.go

@@ -6,14 +6,14 @@ import (
 
 	workloadutil "github.com/rancher/rancher/pkg/controllers/managementagent/workload"
 	"github.com/rancher/rancher/pkg/ingresswrapper"
+	"github.com/rancher/rancher/pkg/settings"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
 type IngressEndpointsController struct {
 	workloadController workloadutil.CommonController
 	ingressInterface   ingresswrapper.CompatInterface
-	isRKE              bool
 }
 
 func (c *IngressEndpointsController) sync(key string, obj ingresswrapper.Ingress) (runtime.Object, error) {
@@ -39,7 +39,7 @@ func (c *IngressEndpointsController) sync(key string, obj ingresswrapper.Ingress
 }
 
 func (c *IngressEndpointsController) reconcileEndpointsForIngress(obj ingresswrapper.Ingress) (bool, error) {
-	fromObj, err := convertIngressToPublicEndpoints(obj, c.isRKE)
+	fromObj, err := convertIngressToPublicEndpoints(obj, settings.IsRKE.Get() == "true")
 	if err != nil {
 		return false, err
 	}

pkg/controllers/managementuserlegacy/endpoints/pod_endpoints.go renamed to pkg/controllers/managementagent/endpoints/pod_endpoints.go

@@ -3,7 +3,6 @@ package endpoints
 import (
 	workloadutil "github.com/rancher/rancher/pkg/controllers/managementagent/workload"
 	v1 "github.com/rancher/rancher/pkg/generated/norman/core/v1"
-	v3 "github.com/rancher/rancher/pkg/generated/norman/management.cattle.io/v3"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -15,7 +14,7 @@ import (
 type ServicesController struct {
 	services           v1.ServiceInterface
 	workloadController workloadutil.CommonController
-	machinesLister     v3.NodeLister
+	nodesLister        v1.NodeLister
 	clusterName        string
 }
 
@@ -41,7 +40,7 @@ func (s *ServicesController) sync(key string, obj *corev1.Service) (runtime.Obje
 
 func (s *ServicesController) reconcileEndpointsForService(svc *corev1.Service) (bool, error) {
 	// 1. update service with endpoints
-	allNodesIP, err := getAllNodesPublicEndpointIP(s.machinesLister, s.clusterName)
+	allNodesIP, err := getAllNodesPublicEndpointIP(s.nodesLister, s.clusterName)
 	if err != nil {
 		return false, err
 	}

pkg/controllers/managementuserlegacy/endpoints/service_endpoints.go renamed to pkg/controllers/managementagent/endpoints/service_endpoints.go

@@ -6,10 +6,10 @@ import (
 	"strings"
 
 	v32 "github.com/rancher/rancher/pkg/apis/project.cattle.io/v3"
+	"github.com/rancher/rancher/pkg/settings"
 
 	workloadutil "github.com/rancher/rancher/pkg/controllers/managementagent/workload"
 	v1 "github.com/rancher/rancher/pkg/generated/norman/core/v1"
-	managementv3 "github.com/rancher/rancher/pkg/generated/norman/management.cattle.io/v3"
 	"github.com/rancher/rancher/pkg/ingresswrapper"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
@@ -25,10 +25,8 @@ type WorkloadEndpointsController struct {
 	serviceLister           v1.ServiceLister
 	podLister               v1.PodLister
 	WorkloadController      workloadutil.CommonController
-	machinesLister          managementv3.NodeLister
 	nodeLister              v1.NodeLister
 	clusterName             string
-	isRKE                   bool
 	serverSupportsIngressV1 bool
 }
 
@@ -77,18 +75,18 @@ func (c *WorkloadEndpointsController) UpdateEndpoints(key string, obj *workloadu
 		return err
 	}
 
-	nodeNameToMachine, err := getNodeNameToMachine(c.clusterName, c.machinesLister, c.nodeLister)
+	nodeNameToMachine, err := getNodeNameToMachine(c.nodeLister)
 	if err != nil {
 		return err
 	}
-	allNodesIP, err := getAllNodesPublicEndpointIP(c.machinesLister, c.clusterName)
+	allNodesIP, err := getAllNodesPublicEndpointIP(c.nodeLister, c.clusterName)
 	if err != nil {
 		return err
 	}
 	// get ingress endpoint group by service
 	serviceToIngressEndpoints := make(map[string][]v32.PublicEndpoint)
 	for _, ingress := range ingresses {
-		epsMap, err := convertIngressToServicePublicEndpointsMap(ingress, c.isRKE)
+		epsMap, err := convertIngressToServicePublicEndpointsMap(ingress, settings.IsRKE.Get() == "true")
 		if err != nil {
 			return err
 		}

pkg/controllers/managementuserlegacy/endpoints/workload_endpoints.go renamed to pkg/controllers/managementagent/endpoints/workload_endpoints.go

@@ -7,7 +7,6 @@ import (
 	"github.com/rancher/rancher/pkg/controllers/managementuserlegacy/alert"
 	"github.com/rancher/rancher/pkg/controllers/managementuserlegacy/approuter"
 	"github.com/rancher/rancher/pkg/controllers/managementuserlegacy/cis"
-	"github.com/rancher/rancher/pkg/controllers/managementuserlegacy/endpoints"
 	"github.com/rancher/rancher/pkg/controllers/managementuserlegacy/globaldns"
 	"github.com/rancher/rancher/pkg/controllers/managementuserlegacy/helm"
 	"github.com/rancher/rancher/pkg/controllers/managementuserlegacy/istio"
@@ -25,7 +24,6 @@ func Register(ctx context.Context, cluster *config.UserContext, clusterRec *mana
 	cis.Register(ctx, cluster)
 	pipeline.Register(ctx, cluster)
 	systemimage.Register(ctx, cluster)
-	endpoints.Register(ctx, cluster)
 	approuter.Register(ctx, cluster)
 	alert.Register(ctx, cluster)
 	globaldns.Register(ctx, cluster)

pkg/controllers/managementuserlegacy/controllers.go

@@ -83,6 +83,29 @@ func GetNodeInternalAddress(node *corev1.Node) string {
 	return ""
 }
 
+func GetEndpointV1NodeIP(node *v1.Node) string {
+	externalIP := ""
+	internalIP := ""
+	for _, ip := range node.Status.Addresses {
+		if ip.Type == "ExternalIP" && ip.Address != "" {
+			externalIP = ip.Address
+			break
+		} else if ip.Type == "InternalIP" && ip.Address != "" {
+			internalIP = ip.Address
+		}
+	}
+	if externalIP != "" {
+		return externalIP
+	}
+	if node.Annotations != nil {
+		externalIP = node.Annotations[externalAddressAnnotation]
+		if externalIP != "" {
+			return externalIP
+		}
+	}
+	return internalIP
+}
+
 func GetEndpointNodeIP(node *v3.Node) string {
 	externalIP := ""
 	internalIP := ""
@@ -170,6 +193,15 @@ func GetMachineForNode(node *corev1.Node, clusterNamespace string, machineLister
 	return nil, nil
 }
 
+func IsNodeReady(node *v1.Node) bool {
+	for _, cond := range node.Status.Conditions {
+		if cond.Type == corev1.NodeReady {
+			return cond.Status == corev1.ConditionTrue
+		}
+	}
+	return false
+}
+
 func IsMachineReady(machine *v3.Node) bool {
 	for _, cond := range machine.Status.InternalNodeStatus.Conditions {
 		if cond.Type == corev1.NodeReady {

pkg/node/node.go

@@ -47,6 +47,7 @@ var (
 	InstallUUID                       = NewSetting("install-uuid", "")
 	InternalServerURL                 = NewSetting("internal-server-url", "")
 	InternalCACerts                   = NewSetting("internal-cacerts", "")
+	IsRKE                             = NewSetting("is-rke", "")
 	JailerTimeout                     = NewSetting("jailer-timeout", "60")
 	KubeconfigGenerateToken           = NewSetting("kubeconfig-generate-token", "true")
 	KubeconfigTokenTTLMinutes         = NewSetting("kubeconfig-token-ttl-minutes", "960") // 16 hours