Merge pull request #375 from galal-hussein/schedule_svclb

pkg/servicelb/controller.go

@@ -21,13 +21,16 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/kubernetes"
+	v1getter "k8s.io/client-go/kubernetes/typed/apps/v1"
 	coregetter "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
 const (
-	image        = "rancher/klipper-lb:v0.1.1"
-	svcNameLabel = "svccontroller.k3s.cattle.io/svcname"
-	Ready        = condition.Cond("Ready")
+	image              = "rancher/klipper-lb:v0.1.1"
+	svcNameLabel       = "svccontroller.k3s.cattle.io/svcname"
+	daemonsetNodeLabel = "svccontroller.k3s.cattle.io/enablelb"
+	nodeSelectorLabel  = "svccontroller.k3s.cattle.io/nodeselector"
+	Ready              = condition.Cond("Ready")
 )
 
 var (
@@ -44,12 +47,14 @@ func Register(ctx context.Context, kubernetes kubernetes.Interface, enabled, roo
 		nodeCache: clients.Node.Cache(),
 		podCache:  clients.Pod.Cache(),
 		processor: objectset.NewProcessor("svccontroller").
-			Client(appClients.Deployment),
+			Client(appClients.DaemonSet),
 		serviceCache: clients.Service.Cache(),
 		services:     kubernetes.CoreV1(),
+		daemonsets:   kubernetes.AppsV1(),
 	}
 
-	clients.Service.OnChange(ctx, "svccontroller", h.onChange)
+	clients.Service.OnChange(ctx, "svccontroller", h.onChangeService)
+	clients.Node.OnChange(ctx, "svccontroller", h.onChangeNode)
 	changeset.Watch(ctx, "svccontroller-watcher",
 		h.onResourceChange,
 		clients.Service,
@@ -67,6 +72,7 @@ type handler struct {
 	processor    *objectset.Processor
 	serviceCache coreclient.ServiceClientCache
 	services     coregetter.ServicesGetter
+	daemonsets   v1getter.DaemonSetsGetter
 }
 
 func (h *handler) onResourceChange(name, namespace string, obj runtime.Object) ([]changeset.Key, error) {
@@ -101,7 +107,7 @@ func (h *handler) onResourceChange(name, namespace string, obj runtime.Object) (
 	}, nil
 }
 
-func (h *handler) onChange(svc *core.Service) (runtime.Object, error) {
+func (h *handler) onChangeService(svc *core.Service) (runtime.Object, error) {
 	if svc.Spec.Type != core.ServiceTypeLoadBalancer || svc.Spec.ClusterIP == "" ||
 		svc.Spec.ClusterIP == "None" {
 		return svc, nil
@@ -116,6 +122,18 @@ func (h *handler) onChange(svc *core.Service) (runtime.Object, error) {
 	return nil, err
 }
 
+func (h *handler) onChangeNode(node *core.Node) (runtime.Object, error) {
+	if _, ok := node.Labels[daemonsetNodeLabel]; !ok {
+		return node, nil
+	}
+
+	if err := h.updateDaemonSets(); err != nil {
+		return node, err
+	}
+
+	return node, nil
+}
+
 func (h *handler) updateService(svc *core.Service) (runtime.Object, error) {
 	pods, err := h.podCache.List(svc.Namespace, labels.SelectorFromSet(map[string]string{
 		svcNameLabel: svc.Name,
@@ -200,43 +218,27 @@ func (h *handler) podIPs(pods []*core.Pod) ([]string, error) {
 }
 
 func (h *handler) deployPod(svc *core.Service) error {
+
 	objs := objectset.NewObjectSet()
 	if !h.enabled {
 		return h.processor.NewDesiredSet(svc, objs).Apply()
 	}
 
-	dep, err := h.newDeployment(svc)
+	ds, err := h.newDaemonSet(svc)
 	if err != nil {
 		return err
 	}
-	if dep != nil {
-		objs.Add(dep)
+	if ds != nil {
+		objs.Add(ds)
 	}
-
 	return h.processor.NewDesiredSet(svc, objs).Apply()
 }
 
-func (h *handler) newDeployment(svc *core.Service) (*apps.Deployment, error) {
+func (h *handler) newDaemonSet(svc *core.Service) (*apps.DaemonSet, error) {
 	name := fmt.Sprintf("svclb-%s", svc.Name)
-	zeroInt := intstr.FromInt(0)
 	oneInt := intstr.FromInt(1)
-	replicas := int32(0)
-
-	nodes, err := h.nodeCache.List("", labels.Everything())
-	if err != nil {
-		return nil, err
-	}
 
-	for _, node := range nodes {
-		if Ready.IsTrue(node) {
-			replicas++
-		}
-		if replicas >= 2 {
-			break
-		}
-	}
-
-	dep := &apps.Deployment{
+	ds := &apps.DaemonSet{
 		ObjectMeta: meta.ObjectMeta{
 			Name:      name,
 			Namespace: svc.Namespace,
@@ -249,13 +251,15 @@ func (h *handler) newDeployment(svc *core.Service) (*apps.Deployment, error) {
 					Controller: &trueVal,
 				},
 			},
+			Labels: map[string]string{
+				nodeSelectorLabel: "false",
+			},
 		},
 		TypeMeta: meta.TypeMeta{
-			Kind:       "Deployment",
+			Kind:       "DaemonSet",
 			APIVersion: "apps/v1",
 		},
-		Spec: apps.DeploymentSpec{
-			Replicas: &replicas,
+		Spec: apps.DaemonSetSpec{
 			Selector: &meta.LabelSelector{
 				MatchLabels: map[string]string{
 					"app": name,
@@ -269,10 +273,9 @@ func (h *handler) newDeployment(svc *core.Service) (*apps.Deployment, error) {
 					},
 				},
 			},
-			Strategy: apps.DeploymentStrategy{
-				Type: apps.RollingUpdateDeploymentStrategyType,
-				RollingUpdate: &apps.RollingUpdateDeployment{
-					MaxSurge:       &zeroInt,
+			UpdateStrategy: apps.DaemonSetUpdateStrategy{
+				Type: apps.RollingUpdateDaemonSetStrategyType,
+				RollingUpdate: &apps.RollingUpdateDaemonSet{
 					MaxUnavailable: &oneInt,
 				},
 			},
@@ -319,8 +322,44 @@ func (h *handler) newDeployment(svc *core.Service) (*apps.Deployment, error) {
 			},
 		}
 
-		dep.Spec.Template.Spec.Containers = append(dep.Spec.Template.Spec.Containers, container)
+		ds.Spec.Template.Spec.Containers = append(ds.Spec.Template.Spec.Containers, container)
+	}
+	// Add node selector only if label "svccontroller.k3s.cattle.io/enablelb" exists on the nodes
+	selector, err := labels.Parse(daemonsetNodeLabel)
+	if err != nil {
+		return nil, err
+	}
+	nodesWithLabel, err := h.nodeCache.List("", selector)
+	if err != nil {
+		return nil, err
+	}
+	if len(nodesWithLabel) > 0 {
+		ds.Spec.Template.Spec.NodeSelector = map[string]string{
+			daemonsetNodeLabel: "true",
+		}
+		ds.Labels[nodeSelectorLabel] = "true"
+	}
+	return ds, nil
+}
+
+func (h *handler) updateDaemonSets() error {
+	daemonsets, err := h.daemonsets.DaemonSets("").List(meta.ListOptions{
+		LabelSelector: nodeSelectorLabel + "=false",
+	})
+	if err != nil {
+		return err
+	}
+
+	for _, ds := range daemonsets.Items {
+		ds.Spec.Template.Spec.NodeSelector = map[string]string{
+			daemonsetNodeLabel: "true",
+		}
+		ds.Labels[nodeSelectorLabel] = "true"
+		if _, err := h.daemonsets.DaemonSets(ds.Namespace).Update(&ds); err != nil {
+			return err
+		}
+
 	}
 
-	return dep, nil
+	return nil
 }

types/apis/apps/v1/zz_generated_deployment_lifecycle_adapter.go → types/apis/apps/v1/zz_generated_daemon_set_lifecycle_adapter.go

@@ -6,54 +6,54 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
-type DeploymentLifecycle interface {
-	Create(obj *v1.Deployment) (runtime.Object, error)
-	Remove(obj *v1.Deployment) (runtime.Object, error)
-	Updated(obj *v1.Deployment) (runtime.Object, error)
+type DaemonSetLifecycle interface {
+	Create(obj *v1.DaemonSet) (runtime.Object, error)
+	Remove(obj *v1.DaemonSet) (runtime.Object, error)
+	Updated(obj *v1.DaemonSet) (runtime.Object, error)
 }
 
-type deploymentLifecycleAdapter struct {
-	lifecycle DeploymentLifecycle
+type daemonSetLifecycleAdapter struct {
+	lifecycle DaemonSetLifecycle
 }
 
-func (w *deploymentLifecycleAdapter) HasCreate() bool {
+func (w *daemonSetLifecycleAdapter) HasCreate() bool {
 	o, ok := w.lifecycle.(lifecycle.ObjectLifecycleCondition)
 	return !ok || o.HasCreate()
 }
 
-func (w *deploymentLifecycleAdapter) HasFinalize() bool {
+func (w *daemonSetLifecycleAdapter) HasFinalize() bool {
 	o, ok := w.lifecycle.(lifecycle.ObjectLifecycleCondition)
 	return !ok || o.HasFinalize()
 }
 
-func (w *deploymentLifecycleAdapter) Create(obj runtime.Object) (runtime.Object, error) {
-	o, err := w.lifecycle.Create(obj.(*v1.Deployment))
+func (w *daemonSetLifecycleAdapter) Create(obj runtime.Object) (runtime.Object, error) {
+	o, err := w.lifecycle.Create(obj.(*v1.DaemonSet))
 	if o == nil {
 		return nil, err
 	}
 	return o, err
 }
 
-func (w *deploymentLifecycleAdapter) Finalize(obj runtime.Object) (runtime.Object, error) {
-	o, err := w.lifecycle.Remove(obj.(*v1.Deployment))
+func (w *daemonSetLifecycleAdapter) Finalize(obj runtime.Object) (runtime.Object, error) {
+	o, err := w.lifecycle.Remove(obj.(*v1.DaemonSet))
 	if o == nil {
 		return nil, err
 	}
 	return o, err
 }
 
-func (w *deploymentLifecycleAdapter) Updated(obj runtime.Object) (runtime.Object, error) {
-	o, err := w.lifecycle.Updated(obj.(*v1.Deployment))
+func (w *daemonSetLifecycleAdapter) Updated(obj runtime.Object) (runtime.Object, error) {
+	o, err := w.lifecycle.Updated(obj.(*v1.DaemonSet))
 	if o == nil {
 		return nil, err
 	}
 	return o, err
 }
 
-func NewDeploymentLifecycleAdapter(name string, clusterScoped bool, client DeploymentInterface, l DeploymentLifecycle) DeploymentHandlerFunc {
-	adapter := &deploymentLifecycleAdapter{lifecycle: l}
+func NewDaemonSetLifecycleAdapter(name string, clusterScoped bool, client DaemonSetInterface, l DaemonSetLifecycle) DaemonSetHandlerFunc {
+	adapter := &daemonSetLifecycleAdapter{lifecycle: l}
 	syncFn := lifecycle.NewObjectLifecycleAdapter(name, clusterScoped, adapter, client.ObjectClient())
-	return func(key string, obj *v1.Deployment) (runtime.Object, error) {
+	return func(key string, obj *v1.DaemonSet) (runtime.Object, error) {
 		newObj, err := syncFn(key, obj)
 		if o, ok := newObj.(runtime.Object); ok {
 			return o, err

types/apis/apps/v1/zz_generated_deepcopy.go

@@ -6,13 +6,13 @@ import (
 )
 
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
-func (in *DeploymentList) DeepCopyInto(out *DeploymentList) {
+func (in *DaemonSetList) DeepCopyInto(out *DaemonSetList) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	out.ListMeta = in.ListMeta
 	if in.Items != nil {
 		in, out := &in.Items, &out.Items
-		*out = make([]appsv1.Deployment, len(*in))
+		*out = make([]appsv1.DaemonSet, len(*in))
 		for i := range *in {
 			(*in)[i].DeepCopyInto(&(*out)[i])
 		}
@@ -20,18 +20,18 @@ func (in *DeploymentList) DeepCopyInto(out *DeploymentList) {
 	return
 }
 
-// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DeploymentList.
-func (in *DeploymentList) DeepCopy() *DeploymentList {
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DaemonSetList.
+func (in *DaemonSetList) DeepCopy() *DaemonSetList {
 	if in == nil {
 		return nil
 	}
-	out := new(DeploymentList)
+	out := new(DaemonSetList)
 	in.DeepCopyInto(out)
 	return out
 }
 
 // DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
-func (in *DeploymentList) DeepCopyObject() runtime.Object {
+func (in *DaemonSetList) DeepCopyObject() runtime.Object {
 	if c := in.DeepCopy(); c != nil {
 		return c
 	}

types/apis/apps/v1/zz_generated_k8s_client.go

@@ -20,13 +20,13 @@ type Interface interface {
 	RESTClient() rest.Interface
 	controller.Starter
 
-	DeploymentsGetter
+	DaemonSetsGetter
 }
 
 type Clients struct {
 	Interface Interface
 
-	Deployment DeploymentClient
+	DaemonSet DaemonSetClient
 }
 
 type Client struct {
@@ -34,7 +34,7 @@ type Client struct {
 	restClient rest.Interface
 	starters   []controller.Starter
 
-	deploymentControllers map[string]DeploymentController
+	daemonSetControllers map[string]DaemonSetController
 }
 
 func Factory(ctx context.Context, config rest.Config) (context.Context, controller.Starter, error) {
@@ -70,8 +70,8 @@ func NewClientsFromInterface(iface Interface) *Clients {
 	return &Clients{
 		Interface: iface,
 
-		Deployment: &deploymentClient2{
-			iface: iface.Deployments(""),
+		DaemonSet: &daemonSetClient2{
+			iface: iface.DaemonSets(""),
 		},
 	}
 }
@@ -89,7 +89,7 @@ func NewForConfig(config rest.Config) (Interface, error) {
 	return &Client{
 		restClient: restClient,
 
-		deploymentControllers: map[string]DeploymentController{},
+		daemonSetControllers: map[string]DaemonSetController{},
 	}, nil
 }
 
@@ -105,13 +105,13 @@ func (c *Client) Start(ctx context.Context, threadiness int) error {
 	return controller.Start(ctx, threadiness, c.starters...)
 }
 
-type DeploymentsGetter interface {
-	Deployments(namespace string) DeploymentInterface
+type DaemonSetsGetter interface {
+	DaemonSets(namespace string) DaemonSetInterface
 }
 
-func (c *Client) Deployments(namespace string) DeploymentInterface {
-	objectClient := objectclient.NewObjectClient(namespace, c.restClient, &DeploymentResource, DeploymentGroupVersionKind, deploymentFactory{})
-	return &deploymentClient{
+func (c *Client) DaemonSets(namespace string) DaemonSetInterface {
+	objectClient := objectclient.NewObjectClient(namespace, c.restClient, &DaemonSetResource, DaemonSetGroupVersionKind, daemonSetFactory{})
+	return &daemonSetClient{
 		ns:           namespace,
 		client:       c,
 		objectClient: objectClient,

types/apis/apps/v1/zz_generated_scheme.go

@@ -33,7 +33,7 @@ func addKnownTypes(scheme *runtime.Scheme) error {
 	// TODO this gets cleaned up when the types are fixed
 	scheme.AddKnownTypes(SchemeGroupVersion,
 
-		&DeploymentList{},
+		&DaemonSetList{},
 	)
 	return nil
 }

types/codegen/main.go

@@ -99,7 +99,7 @@ func main() {
 	}
 
 	if err := generator.ControllersForForeignTypes(basePackage, appsv1.SchemeGroupVersion, []interface{}{
-		appsv1.Deployment{},
+		appsv1.DaemonSet{},
 	}, nil); err != nil {
 		logrus.Fatal(err)
 	}