weave/lb/controller.go

package lb

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	coreapi "k8s.io/api/core/v1"
	discovery "k8s.io/api/discovery/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/cache"

	"github.com/rajch/weave/common"
)

// Controller watches Services, EndpointSlices and Nodes and keeps this node's part of the picture true.
type Controller struct {
	self   string
	client kubernetes.Interface
	dp     *Datapath
	// readLocal is the node's own interfaces (swappable in tests)
	readLocal func() (Local, error)

	services, slices, nodes cache.Store
	kick                    chan struct{}
	lastProblems            map[string]bool
	// Services whose status this node wrote since it started (so a restart rewrites ipMode once)
	written map[types.UID]string
}

// Resync is how often the node re-reads its interfaces and re-applies everything even without API events.
const Resync = 30 * time.Second

// NewController runs nothing yet; Run does.
func NewController(self string, client kubernetes.Interface, dp *Datapath) *Controller {
	return &Controller{self: self, client: client, dp: dp, readLocal: ReadLocal,
		kick: make(chan struct{}, 1), lastProblems: map[string]bool{}, written: map[types.UID]string{}}
}

func (c *Controller) poke() {
	select {
	case c.kick <- struct{}{}:
	default:
	}
}

// Run blocks until ctx ends.
func (c *Controller) Run(ctx context.Context) {
	var sc, ec, nc cache.Controller
	c.services, sc = newInformer(c, c.client.CoreV1().RESTClient(), "services", &coreapi.Service{})
	c.slices, ec = newInformer(c, c.client.DiscoveryV1().RESTClient(), "endpointslices", &discovery.EndpointSlice{})
	c.nodes, nc = newInformer(c, c.client.CoreV1().RESTClient(), "nodes", &coreapi.Node{})
	for _, ctl := range []cache.Controller{sc, ec, nc} {
		go ctl.Run(ctx.Done())
	}
	if !cache.WaitForCacheSync(ctx.Done(), sc.HasSynced, ec.HasSynced, nc.HasSynced) {
		return
	}
	tick := time.NewTicker(Resync)
	defer tick.Stop()
	c.poke()
	for {
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
		case <-c.kick:
			// let a burst of events settle into one pass
			time.Sleep(200 * time.Millisecond)
			select {
			case <-c.kick:
			default:
			}
		}
		if err := c.sync(ctx); err != nil {
			common.Log.Errorf("lb: %v", err)
		}
	}
}

func newInformer(c *Controller, getter cache.Getter, resource string, obj runtime.Object) (cache.Store, cache.Controller) {
	lw := cache.NewListWatchFromClient(getter, resource, "", fields.Everything())
	return cache.NewInformer(lw, obj, 0, cache.ResourceEventHandlerFuncs{
		AddFunc:    func(interface{}) { c.poke() },
		UpdateFunc: func(interface{}, interface{}) { c.poke() },
		DeleteFunc: func(interface{}) { c.poke() },
	})
}

// sync is one pass: publish this node, compute, apply, write statuses.
func (c *Controller) sync(ctx context.Context) error {
	local, err := c.readLocal()
	if err != nil {
		return fmt.Errorf("reading interfaces: %v", err)
	}
	if err := c.publish(ctx, local); err != nil {
		return err
	}
	if local.PodNet == nil {
		return fmt.Errorf("weave bridge has no address yet; not forwarding")
	}

	var nodes []Node
	var problems []string
	for _, o := range c.nodes.List() {
		n, err := ParseNode(o.(*coreapi.Node))
		if err != nil {
			problems = append(problems, err.Error())
		}
		nodes = append(nodes, n)
	}
	var svcs []*coreapi.Service
	for _, o := range c.services.List() {
		svcs = append(svcs, o.(*coreapi.Service))
	}
	var slices []*discovery.EndpointSlice
	for _, o := range c.slices.List() {
		slices = append(slices, o.(*discovery.EndpointSlice))
	}
	d := Compute(c.self, nodes, svcs, slices)
	// rewrite once after start what an earlier run may have written without ipMode
	for _, f := range d.Forwards {
		if s := c.service(f.Namespace, f.Name); s != nil && c.written[s.UID] != f.Address.String() && !hasStatusFor(d, f) {
			d.Status = append(d.Status, Status{Namespace: f.Namespace, Name: f.Name, Address: f.Address})
		}
	}
	current := map[string]bool{}
	for _, p := range append(problems, d.Problems...) {
		current[p] = true
		if !c.lastProblems[p] {
			common.Log.Warnf("lb: %s", p)
		}
	}
	c.lastProblems = current

	if err := c.dp.Apply(d, local.PodNet); err != nil {
		return fmt.Errorf("applying datapath: %v", err)
	}
	for _, st := range d.Status {
		if err := c.writeStatus(ctx, st); err != nil {
			return err
		}
	}
	return nil
}

func hasStatusFor(d Desired, f Forward) bool {
	for _, s := range d.Status {
		if s.Namespace == f.Namespace && s.Name == f.Name {
			return true
		}
	}
	return false
}

func (c *Controller) service(ns, name string) *coreapi.Service {
	o, ok, err := c.services.GetByKey(ns + "/" + name)
	if err != nil || !ok {
		return nil
	}
	return o.(*coreapi.Service)
}

// publish keeps this node's annotations equal to what its interfaces say.
func (c *Controller) publish(ctx context.Context, local Local) error {
	want := local.Annotations()
	o, ok, err := c.nodes.GetByKey(c.self)
	if err != nil {
		return err
	}
	if ok {
		have := o.(*coreapi.Node).Annotations
		same := true
		for k, v := range want {
			if have[k] != v {
				same = false
			}
		}
		if same {
			return nil
		}
	}
	patch, err := json.Marshal(map[string]interface{}{"metadata": map[string]interface{}{"annotations": want}})
	if err != nil {
		return err
	}
	if _, err := c.client.CoreV1().Nodes().Patch(ctx, c.self, types.MergePatchType, patch, metav1.PatchOptions{}); err != nil {
		return fmt.Errorf("publishing node annotations: %v", err)
	}
	common.Log.Infof("lb: node %s holds [%s]", c.self, want[AnnAddresses])
	return nil
}

// writeStatus sets (ipMode Proxy: kube-proxy leaves the address to us) or clears a Service's load-balancer status.
func (c *Controller) writeStatus(ctx context.Context, st Status) error {
	ingress := []map[string]string{}
	if st.Address != nil {
		ingress = append(ingress, map[string]string{"ip": st.Address.String(), "ipMode": "Proxy"})
	}
	patch, err := json.Marshal(map[string]interface{}{"status": map[string]interface{}{"loadBalancer": map[string]interface{}{"ingress": ingress}}})
	if err != nil {
		return err
	}
	s, err := c.client.CoreV1().Services(st.Namespace).Patch(ctx, st.Name, types.MergePatchType, patch, metav1.PatchOptions{}, "status")
	if err != nil {
		return fmt.Errorf("status of %s/%s: %v", st.Namespace, st.Name, err)
	}
	if st.Address != nil {
		c.written[s.UID] = st.Address.String()
		common.Log.Infof("lb: %s/%s at %s", st.Namespace, st.Name, st.Address)
	} else {
		delete(c.written, s.UID)
		common.Log.Infof("lb: %s/%s has no address", st.Namespace, st.Name)
	}
	return nil
}