mirror of
https://github.com/k3s-io/k3s.git
synced 2024-06-07 19:41:36 +00:00
408 lines
9.3 KiB
Go
408 lines
9.3 KiB
Go
package servicelb
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sort"
|
|
"strconv"
|
|
|
|
appclient "github.com/rancher/wrangler-api/pkg/generated/controllers/apps/v1"
|
|
coreclient "github.com/rancher/wrangler-api/pkg/generated/controllers/core/v1"
|
|
"github.com/rancher/wrangler/pkg/apply"
|
|
"github.com/rancher/wrangler/pkg/condition"
|
|
"github.com/rancher/wrangler/pkg/objectset"
|
|
"github.com/rancher/wrangler/pkg/relatedresource"
|
|
"github.com/rancher/wrangler/pkg/slice"
|
|
"github.com/sirupsen/logrus"
|
|
apps "k8s.io/api/apps/v1"
|
|
core "k8s.io/api/core/v1"
|
|
"k8s.io/apimachinery/pkg/api/errors"
|
|
meta "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/labels"
|
|
"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.2"
|
|
svcNameLabel = "svccontroller.k3s.cattle.io/svcname"
|
|
daemonsetNodeLabel = "svccontroller.k3s.cattle.io/enablelb"
|
|
nodeSelectorLabel = "svccontroller.k3s.cattle.io/nodeselector"
|
|
Ready = condition.Cond("Ready")
|
|
)
|
|
|
|
var (
|
|
trueVal = true
|
|
)
|
|
|
|
func Register(ctx context.Context,
|
|
kubernetes kubernetes.Interface,
|
|
apply apply.Apply,
|
|
daemonSetController appclient.DaemonSetController,
|
|
deployments appclient.DeploymentController,
|
|
nodes coreclient.NodeController,
|
|
pods coreclient.PodController,
|
|
services coreclient.ServiceController,
|
|
endpoints coreclient.EndpointsController,
|
|
enabled, rootless bool) error {
|
|
h := &handler{
|
|
rootless: rootless,
|
|
enabled: enabled,
|
|
nodeCache: nodes.Cache(),
|
|
podCache: pods.Cache(),
|
|
deploymentCache: deployments.Cache(),
|
|
processor: apply.WithSetID("svccontroller").
|
|
WithCacheTypes(daemonSetController),
|
|
serviceCache: services.Cache(),
|
|
services: kubernetes.CoreV1(),
|
|
daemonsets: kubernetes.AppsV1(),
|
|
deployments: kubernetes.AppsV1(),
|
|
}
|
|
|
|
services.OnChange(ctx, "svccontroller", h.onChangeService)
|
|
nodes.OnChange(ctx, "svccontroller", h.onChangeNode)
|
|
relatedresource.Watch(ctx, "svccontroller-watcher",
|
|
h.onResourceChange,
|
|
services,
|
|
pods,
|
|
endpoints)
|
|
|
|
return nil
|
|
}
|
|
|
|
type handler struct {
|
|
rootless bool
|
|
enabled bool
|
|
nodeCache coreclient.NodeCache
|
|
podCache coreclient.PodCache
|
|
deploymentCache appclient.DeploymentCache
|
|
processor apply.Apply
|
|
serviceCache coreclient.ServiceCache
|
|
services coregetter.ServicesGetter
|
|
daemonsets v1getter.DaemonSetsGetter
|
|
deployments v1getter.DeploymentsGetter
|
|
}
|
|
|
|
func (h *handler) onResourceChange(name, namespace string, obj runtime.Object) ([]relatedresource.Key, error) {
|
|
if ep, ok := obj.(*core.Endpoints); ok {
|
|
return []relatedresource.Key{
|
|
{
|
|
Name: ep.Name,
|
|
Namespace: ep.Namespace,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
pod, ok := obj.(*core.Pod)
|
|
if !ok {
|
|
return nil, nil
|
|
}
|
|
|
|
serviceName := pod.Labels[svcNameLabel]
|
|
if serviceName == "" {
|
|
return nil, nil
|
|
}
|
|
|
|
if pod.Status.PodIP == "" {
|
|
return nil, nil
|
|
}
|
|
|
|
return []relatedresource.Key{
|
|
{
|
|
Name: serviceName,
|
|
Namespace: pod.Namespace,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
func (h *handler) onChangeService(key string, svc *core.Service) (*core.Service, error) {
|
|
if svc == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
if svc.Spec.Type != core.ServiceTypeLoadBalancer || svc.Spec.ClusterIP == "" ||
|
|
svc.Spec.ClusterIP == "None" {
|
|
return svc, nil
|
|
}
|
|
|
|
if err := h.deployPod(svc); err != nil {
|
|
return svc, err
|
|
}
|
|
|
|
// Don't return service because we don't want another update
|
|
_, err := h.updateService(svc)
|
|
return nil, err
|
|
}
|
|
|
|
func (h *handler) onChangeNode(key string, node *core.Node) (*core.Node, error) {
|
|
if node == nil {
|
|
return nil, nil
|
|
}
|
|
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) {
|
|
if !h.enabled {
|
|
return svc, nil
|
|
}
|
|
|
|
pods, err := h.podCache.List(svc.Namespace, labels.SelectorFromSet(map[string]string{
|
|
svcNameLabel: svc.Name,
|
|
}))
|
|
|
|
if err != nil {
|
|
return svc, err
|
|
}
|
|
|
|
existingIPs := serviceIPs(svc)
|
|
expectedIPs, err := h.podIPs(pods)
|
|
if err != nil {
|
|
return svc, err
|
|
}
|
|
|
|
sort.Strings(expectedIPs)
|
|
sort.Strings(existingIPs)
|
|
|
|
if slice.StringsEqual(expectedIPs, existingIPs) {
|
|
return svc, nil
|
|
}
|
|
|
|
svc = svc.DeepCopy()
|
|
svc.Status.LoadBalancer.Ingress = nil
|
|
for _, ip := range expectedIPs {
|
|
svc.Status.LoadBalancer.Ingress = append(svc.Status.LoadBalancer.Ingress, core.LoadBalancerIngress{
|
|
IP: ip,
|
|
})
|
|
}
|
|
|
|
logrus.Debugf("Setting service loadbalancer %s/%s to IPs %v", svc.Namespace, svc.Name, expectedIPs)
|
|
return h.services.Services(svc.Namespace).UpdateStatus(svc)
|
|
}
|
|
|
|
func serviceIPs(svc *core.Service) []string {
|
|
var ips []string
|
|
|
|
for _, ingress := range svc.Status.LoadBalancer.Ingress {
|
|
if ingress.IP != "" {
|
|
ips = append(ips, ingress.IP)
|
|
}
|
|
}
|
|
|
|
return ips
|
|
}
|
|
|
|
func (h *handler) podIPs(pods []*core.Pod) ([]string, error) {
|
|
ips := map[string]bool{}
|
|
|
|
for _, pod := range pods {
|
|
if pod.Spec.NodeName == "" || pod.Status.PodIP == "" {
|
|
continue
|
|
}
|
|
if !Ready.IsTrue(pod) {
|
|
continue
|
|
}
|
|
|
|
node, err := h.nodeCache.Get(pod.Spec.NodeName)
|
|
if errors.IsNotFound(err) {
|
|
continue
|
|
} else if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, addr := range node.Status.Addresses {
|
|
if addr.Type == core.NodeExternalIP {
|
|
ips[addr.Address] = true
|
|
}
|
|
}
|
|
if len(ips) == 0 {
|
|
for _, addr := range node.Status.Addresses {
|
|
if addr.Type == core.NodeInternalIP {
|
|
ips[addr.Address] = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
var ipList []string
|
|
for k := range ips {
|
|
ipList = append(ipList, k)
|
|
}
|
|
|
|
if len(ipList) > 0 && h.rootless {
|
|
return []string{"127.0.0.1"}, nil
|
|
}
|
|
|
|
return ipList, nil
|
|
}
|
|
|
|
func (h *handler) deployPod(svc *core.Service) error {
|
|
if err := h.deleteOldDeployments(svc); err != nil {
|
|
return err
|
|
}
|
|
objs := objectset.NewObjectSet()
|
|
if !h.enabled {
|
|
return h.processor.WithOwner(svc).Apply(objs)
|
|
}
|
|
|
|
ds, err := h.newDaemonSet(svc)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if ds != nil {
|
|
objs.Add(ds)
|
|
}
|
|
return h.processor.WithOwner(svc).Apply(objs)
|
|
}
|
|
|
|
func (h *handler) newDaemonSet(svc *core.Service) (*apps.DaemonSet, error) {
|
|
name := fmt.Sprintf("svclb-%s", svc.Name)
|
|
oneInt := intstr.FromInt(1)
|
|
|
|
ds := &apps.DaemonSet{
|
|
ObjectMeta: meta.ObjectMeta{
|
|
Name: name,
|
|
Namespace: svc.Namespace,
|
|
OwnerReferences: []meta.OwnerReference{
|
|
{
|
|
Name: svc.Name,
|
|
APIVersion: "v1",
|
|
Kind: "Service",
|
|
UID: svc.UID,
|
|
Controller: &trueVal,
|
|
},
|
|
},
|
|
Labels: map[string]string{
|
|
nodeSelectorLabel: "false",
|
|
},
|
|
},
|
|
TypeMeta: meta.TypeMeta{
|
|
Kind: "DaemonSet",
|
|
APIVersion: "apps/v1",
|
|
},
|
|
Spec: apps.DaemonSetSpec{
|
|
Selector: &meta.LabelSelector{
|
|
MatchLabels: map[string]string{
|
|
"app": name,
|
|
},
|
|
},
|
|
Template: core.PodTemplateSpec{
|
|
ObjectMeta: meta.ObjectMeta{
|
|
Labels: map[string]string{
|
|
"app": name,
|
|
svcNameLabel: svc.Name,
|
|
},
|
|
},
|
|
},
|
|
UpdateStrategy: apps.DaemonSetUpdateStrategy{
|
|
Type: apps.RollingUpdateDaemonSetStrategyType,
|
|
RollingUpdate: &apps.RollingUpdateDaemonSet{
|
|
MaxUnavailable: &oneInt,
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
for _, port := range svc.Spec.Ports {
|
|
portName := fmt.Sprintf("lb-port-%d", port.Port)
|
|
container := core.Container{
|
|
Name: portName,
|
|
Image: image,
|
|
ImagePullPolicy: core.PullIfNotPresent,
|
|
Ports: []core.ContainerPort{
|
|
{
|
|
Name: portName,
|
|
ContainerPort: port.Port,
|
|
HostPort: port.Port,
|
|
},
|
|
},
|
|
Env: []core.EnvVar{
|
|
{
|
|
Name: "SRC_PORT",
|
|
Value: strconv.Itoa(int(port.Port)),
|
|
},
|
|
{
|
|
Name: "DEST_PROTO",
|
|
Value: string(port.Protocol),
|
|
},
|
|
{
|
|
Name: "DEST_PORT",
|
|
Value: strconv.Itoa(int(port.Port)),
|
|
},
|
|
{
|
|
Name: "DEST_IP",
|
|
Value: svc.Spec.ClusterIP,
|
|
},
|
|
},
|
|
SecurityContext: &core.SecurityContext{
|
|
Capabilities: &core.Capabilities{
|
|
Add: []core.Capability{
|
|
"NET_ADMIN",
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
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 nil
|
|
}
|
|
|
|
func (h *handler) deleteOldDeployments(svc *core.Service) error {
|
|
name := fmt.Sprintf("svclb-%s", svc.Name)
|
|
if _, err := h.deploymentCache.Get(svc.Namespace, name); err != nil {
|
|
if errors.IsNotFound(err) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
return h.deployments.Deployments(svc.Namespace).Delete(name, &meta.DeleteOptions{})
|
|
}
|