haproxy-lb/pkg/controller/loadbalancer.go

package controller

import (
	"context"
	"os/exec"
	"time"

	"git.giftfish.de/ston1th/haproxy-lb/pkg/api"
	"git.giftfish.de/ston1th/haproxy-lb/pkg/cluster"
	"git.giftfish.de/ston1th/haproxy-lb/pkg/config"
	"git.giftfish.de/ston1th/haproxy-lb/pkg/db"
	"git.giftfish.de/ston1th/haproxy-lb/pkg/haproxy"
	"git.giftfish.de/ston1th/haproxy-lb/pkg/vip"
	"github.com/go-logr/logr"
)

func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (callbacks cluster.Callbacks, err error) {
	var fo *vip.Network
	if cfg.Failover != nil {
		fo, err = vip.NewNetwork(cfg.Failover.Interface)
		if err != nil {
			return
		}
	}

	n, err := vip.NewNetworkWithLabel(cfg.VIP.Interface, cfg.VIP.Label)
	if err != nil {
		return
	}
	ha, err := haproxy.NewHAProxyManager(cfg.HAProxyConfig, cfg.HAProxyService)
	if err != nil {
		return
	}

	callbacks = cluster.Callbacks{
		Leader: func(ctx context.Context, cc cluster.CallbackContext) {
			db := db.New(cc)
			log := cc.Logger()
			log.Info("starting haproxy", "service", cfg.HAProxyService)
			err := ha.Start(ctx)
			if err != nil {
				log.Error(err, "error starting haproxy", "service", cfg.HAProxyService)
				cc.Fatal(err)
				return
			}
			err = srv.UpdateDB(db, true)
			if err != nil {
				log.Error(err, "error initialising api server as leader")
				cc.Fatal(err)
				return
			}
			t := time.NewTicker(time.Second * 10)
			if cfg.LeaderHook != "" {
				go func() {
					err := exec.CommandContext(ctx, cfg.LeaderHook).Run()
					if err != nil {
						log.Error(err, "error running hook", "leaderHook", cfg.LeaderHook)
					}
				}()
			}
			for {
				if ctx.Err() == context.Canceled {
					t.Stop()
					log.Info("leading canceled", "id", cc.ID())
					return
				}
				log.Info("leading", "id", cc.ID())
				if cfg.Failover != nil {
					addIPs(log, fo, []string{cfg.Failover.IP})
				}
				ips, err := db.GetDeletedIPs()
				if err != nil && err != cluster.ErrPrefixNotFound {
					log.Error(err, "error reading deleted ips list")
					<-t.C
					continue
				}
				deleteIPs(log, n, ips)

				ips, err = db.GetIPs()
				if err != nil && err != cluster.ErrPrefixNotFound {
					log.Error(err, "error reading ips list")
					<-t.C
					continue
				}
				addIPs(log, n, ips)

				lbs, err := db.GetLBs("")
				if err != nil && err != cluster.ErrPrefixNotFound {
					log.Error(err, "error reading lb list")
					<-t.C
					continue
				}
				err = ha.IsRunning(ctx)
				if err != nil {
					log.Error(err, "error haproxy is not running", "service", cfg.HAProxyService)
					log.Info("starting haproxy", "service", cfg.HAProxyService)
					err = ha.Start(ctx)
					if err != nil {
						log.Error(err, "error starting haproxy", "service", cfg.HAProxyService)
					}
				}
				lbcfg, err := haproxy.NewConfig(lbs)
				if err != nil {
					log.Error(err, "error decoding haproxy config")
					<-t.C
					continue
				}
				err = ha.UpdateConfig(ctx, lbcfg)
				if err != nil {
					log.Error(err, "error updating haproxy config", "service", cfg.HAProxyService)
					<-t.C
					continue
				}
				select {
				case <-ctx.Done():
					t.Stop()
					deleteIPs(log, n, ips)
					log.Info("stopping haproxy", "service", cfg.HAProxyService)
					err = ha.Stop(context.Background())
					if err != nil {
						log.Error(err, "error stopping haproxy")
					}
					log.Info("leading canceled", "id", cc.ID())
					return
				case <-t.C:
				}
			}
		},
		Follower: func(ctx context.Context, cc cluster.CallbackContext) {
			db := db.New(cc)
			log := cc.Logger()
			err := srv.UpdateDB(db, false)
			if err != nil {
				log.Error(err, "error initialising api server as follower")
				cc.Fatal(err)
				return
			}
			t := time.NewTicker(time.Second * 10)
			if cfg.FollowerHook != "" {
				go func() {
					err := exec.CommandContext(ctx, cfg.FollowerHook).Run()
					if err != nil {
						log.Error(err, "error running hook", "followerHook", cfg.FollowerHook)
					}
				}()
			}
			for {
				if ctx.Err() == context.Canceled {
					t.Stop()
					log.Info("following canceled", "id", cc.ID())
					return
				}
				log.Info("following", "id", cc.ID())
				select {
				case <-ctx.Done():
					t.Stop()
					log.Info("following canceled", "id", cc.ID())
					return
				case <-t.C:
				}
			}
		},
		Cleanup: func(ctx context.Context, cc cluster.CallbackContext) {
			if cfg.Failover != nil {
				deleteIPs(log, fo, fo.GetIPs())
			}
			deleteIPs(log, n, n.GetIPs())
		},
	}
	return
}

func addIPs(log logr.Logger, n *vip.Network, cidrs []string) error {
	for _, cidr := range cidrs {
		err := n.AddIP(cidr)
		if err != nil {
			log.Error(err, "adding ip", "interface", n.Interface(), "ip", cidr)
		}
	}
	return nil
}

func deleteIPs(log logr.Logger, n *vip.Network, cidrs []string) error {
	for _, cidr := range cidrs {
		err := n.DeleteIP(cidr)
		if err != nil {
			log.Error(err, "deleting ip", "interface", n.Interface(), "ip", cidr)
		}
	}
	return nil
}