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) { 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") cc.Fatal(err) return } err = srv.UpdateDB(db) 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()) 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 } lbcfg, err := haproxy.NewConfig(lbs) if err != nil { log.Error(err, "error reading haproxy config") <-t.C continue } err = ha.UpdateConfig(ctx, lbcfg) if err != nil { log.Error(err, "error updating haproxy config") <-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) 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) { 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 }