From fef86deac8962474fb3e50598e3c25786e2209ee Mon Sep 17 00:00:00 2001 From: ston1th Date: Sat, 9 Oct 2021 19:03:08 +0200 Subject: [PATCH] cleanup, ensure haproxy is running and disallow writes to followers --- pkg/api/server.go | 3 +- pkg/api/types/context.go | 11 +- pkg/api/types/error.go | 1 + pkg/api/v1/server/handler.go | 18 +++ pkg/controller/loadbalancer.go | 19 ++- pkg/haproxy/haproxy.go | 4 + pkg/vip/{arp_linux2.go => arp_linux.go} | 0 pkg/vip/arp_linux.go.bak | 173 ------------------------ run_etcd.sh | 3 +- 9 files changed, 49 insertions(+), 183 deletions(-) rename pkg/vip/{arp_linux2.go => arp_linux.go} (100%) delete mode 100644 pkg/vip/arp_linux.go.bak diff --git a/pkg/api/server.go b/pkg/api/server.go index 4d6be96..01fd1bc 100644 --- a/pkg/api/server.go +++ b/pkg/api/server.go @@ -104,8 +104,9 @@ func NewServer(c *config.Config, log logr.Logger) (*Server, error) { return s, nil } -func (s *Server) UpdateDB(db *db.DB) error { +func (s *Server) UpdateDB(db *db.DB, leader bool) error { s.Data.DB = db + s.Data.Leader = leader if s.Data.Net == nil { na, err := netalloc.NewGenericAlloc(db, s.cidr) if err != nil { diff --git a/pkg/api/types/context.go b/pkg/api/types/context.go index 66bcbbb..f9cea8d 100644 --- a/pkg/api/types/context.go +++ b/pkg/api/types/context.go @@ -14,9 +14,10 @@ import ( ) type ContextData struct { - Net netalloc.Allocator - DB *db.DB - Auth *Auth + Net netalloc.Allocator + DB *db.DB + Auth *Auth + Leader bool } func NewContext(w http.ResponseWriter, r *http.Request, data *ContextData, log logr.Logger) *Context { @@ -42,6 +43,10 @@ type Context struct { Log logr.Logger } +func (c *Context) Leader() bool { + return c.Data.Leader +} + // Method returns the request method func (c *Context) Method() string { return c.Request.Method diff --git a/pkg/api/types/error.go b/pkg/api/types/error.go index 8e0fd95..700aad1 100644 --- a/pkg/api/types/error.go +++ b/pkg/api/types/error.go @@ -43,4 +43,5 @@ var ( ErrMNA = newError("method not allowed", http.StatusMethodNotAllowed) ErrExists = newError("resource already exists", http.StatusConflict) ErrISE = newError("internal server error", http.StatusInternalServerError) + ErrFollower = newError("follower write forbidden", http.StatusForbidden) ) diff --git a/pkg/api/v1/server/handler.go b/pkg/api/v1/server/handler.go index a251aa2..bce53bd 100644 --- a/pkg/api/v1/server/handler.go +++ b/pkg/api/v1/server/handler.go @@ -4,6 +4,7 @@ import ( //"golang.org/x/crypto/bcrypt" //"net/http" "encoding/json" + "errors" "net" "git.giftfish.de/ston1th/haproxy-lb/pkg/api/types" @@ -11,6 +12,8 @@ import ( "git.giftfish.de/ston1th/haproxy-lb/pkg/cluster" ) +var errFollower = errors.New("follower write") + func authHandler(h types.CtxHandler) types.CtxHandler { return func(ctx *types.Context) { user, pass, ok := ctx.Request.BasicAuth() @@ -65,6 +68,11 @@ func lbHandler(ctx *types.Context) { } ctx.Body(b) case "POST": + if !ctx.Leader() { + ctx.Log.Error(errFollower, "error writing config: i am a follower", "cluster", cl, "name", n) + ctx.Err(types.ErrFollower) + return + } b, err := ctx.ReadBody() if err != nil { ctx.Log.Error(err, "error reading request body", "cluster", cl, "name", n) @@ -123,6 +131,11 @@ func lbHandler(ctx *types.Context) { } ctx.OK() case "DELETE": + if !ctx.Leader() { + ctx.Log.Error(errFollower, "error writing config: i am a follower", "cluster", cl, "name", n) + ctx.Err(types.ErrFollower) + return + } if !ctx.Data.DB.LBExists(name) { ctx.Err(types.ErrNotFound) return @@ -169,6 +182,11 @@ func lbClusterHandler(ctx *types.Context) { case "GET": ctx.JSON(lbList(m)) case "DELETE": + if !ctx.Leader() { + ctx.Log.Error(errFollower, "error writing config: i am a follower", "cluster", cl, "name", name) + ctx.Err(types.ErrFollower) + return + } for name, _ := range m { cidr, err := ctx.Data.DB.GetCIDR(name) if err != nil && err != cluster.ErrKeyNotFound { diff --git a/pkg/controller/loadbalancer.go b/pkg/controller/loadbalancer.go index fa33c68..4db46f6 100644 --- a/pkg/controller/loadbalancer.go +++ b/pkg/controller/loadbalancer.go @@ -31,11 +31,11 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call log.Info("starting haproxy", "service", cfg.HAProxyService) err := ha.Start(ctx) if err != nil { - log.Error(err, "error starting haproxy") + log.Error(err, "error starting haproxy", "service", cfg.HAProxyService) cc.Fatal(err) return } - err = srv.UpdateDB(db) + err = srv.UpdateDB(db, true) if err != nil { log.Error(err, "error initialising api server as leader") cc.Fatal(err) @@ -79,15 +79,24 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call <-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 reading haproxy config") + 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") + log.Error(err, "error updating haproxy config", "service", cfg.HAProxyService) <-t.C continue } @@ -109,7 +118,7 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call Follower: func(ctx context.Context, cc cluster.CallbackContext) { db := db.New(cc) log := cc.Logger() - err := srv.UpdateDB(db) + err := srv.UpdateDB(db, false) if err != nil { log.Error(err, "error initialising api server as follower") cc.Fatal(err) diff --git a/pkg/haproxy/haproxy.go b/pkg/haproxy/haproxy.go index 2da4970..9ba006c 100644 --- a/pkg/haproxy/haproxy.go +++ b/pkg/haproxy/haproxy.go @@ -134,6 +134,10 @@ func (ha *HAProxyManager) Start(ctx context.Context) error { return err } +func (ha *HAProxyManager) IsRunning(ctx context.Context) error { + return ha.svc.Status(ctx) +} + func (ha *HAProxyManager) Stop(ctx context.Context) error { return ha.svc.Stop(ctx) } diff --git a/pkg/vip/arp_linux2.go b/pkg/vip/arp_linux.go similarity index 100% rename from pkg/vip/arp_linux2.go rename to pkg/vip/arp_linux.go diff --git a/pkg/vip/arp_linux.go.bak b/pkg/vip/arp_linux.go.bak deleted file mode 100644 index af14d3a..0000000 --- a/pkg/vip/arp_linux.go.bak +++ /dev/null @@ -1,173 +0,0 @@ -// +build linux - -// These syscalls are only supported on Linux, so this uses a build directive during compilation. Other OS's will use the arp_unsupported.go and receive an error - -package vip - -import ( - "bytes" - "encoding/binary" - "fmt" - "net" - "syscall" - "unsafe" -) - -const ( - opARPRequest = 1 - opARPReply = 2 - hwLen = 6 -) - -var ( - ethernetBroadcast = net.HardwareAddr{0xff, 0xff, 0xff, 0xff, 0xff, 0xff} - // arpRequest is used to flip between garp request or garp reply - arpRequest = true -) - -func htons(p uint16) uint16 { - var b [2]byte - binary.BigEndian.PutUint16(b[:], p) - return *(*uint16)(unsafe.Pointer(&b)) -} - -// arpHeader specifies the header for an ARP message. -type arpHeader struct { - hardwareType uint16 - protocolType uint16 - hardwareAddressLength uint8 - protocolAddressLength uint8 - opcode uint16 -} - -// arpMessage represents an ARP message. -type arpMessage struct { - arpHeader - senderHardwareAddress []byte - senderProtocolAddress []byte - targetHardwareAddress []byte - targetProtocolAddress []byte -} - -// bytes returns the wire representation of the ARP message. -func (m *arpMessage) bytes() ([]byte, error) { - buf := new(bytes.Buffer) - - if err := binary.Write(buf, binary.BigEndian, m.arpHeader); err != nil { - return nil, fmt.Errorf("binary write failed: %v", err) - } - buf.Write(m.senderHardwareAddress) - buf.Write(m.senderProtocolAddress) - buf.Write(m.targetHardwareAddress) - buf.Write(m.targetProtocolAddress) - - return buf.Bytes(), nil -} - -// gratuitousARP return a gARP request or gARP reply alternatively -// because different devices may support either one of them -func gratuitousARP(ip net.IP, mac net.HardwareAddr) (*arpMessage, error) { - if ip.To4() == nil { - return nil, fmt.Errorf("%q is not an IPv4 address", ip) - } - if len(mac) != hwLen { - return nil, fmt.Errorf("%q is not an Ethernet MAC address", mac) - } - - m := &arpMessage{ - arpHeader: arpHeader{ - 1, // Ethernet - 0x0800, // IPv4 - hwLen, // 48-bit MAC Address - net.IPv4len, // 32-bit IPv4 Address - opARPReply, // ARP Reply - }, - } - - // https://tools.ietf.org/html/rfc5944#section-4.6 - // In either case, the ARP Sender Hardware Address is - // set to the link-layer address to which this cache entry should be - // updated. - m.senderHardwareAddress = mac - - // When using an ARP Reply packet, the Target Hardware - // Address is also set to the link-layer address to which this cache - // entry should be updated (this field is not used in an ARP Request - // packet). - m.targetHardwareAddress = mac - - // In either case, the ARP Sender Protocol Address and - // ARP Target Protocol Address are both set to the IP address of the - // cache entry to be updated, - m.senderProtocolAddress = ip.To4() - m.targetProtocolAddress = ip.To4() - - // send arpRequest and arpReply alternatively - arpRequest = !arpRequest - if arpRequest { - m.arpHeader.opcode = opARPRequest - - // this field is not used in an ARP Request packet - m.targetHardwareAddress = ethernetBroadcast - } - - return m, nil -} - -// sendARP sends the given ARP message via the specified interface. -func sendARP(iface *net.Interface, m *arpMessage) error { - fd, err := syscall.Socket(syscall.AF_PACKET, syscall.SOCK_DGRAM, int(htons(syscall.ETH_P_ARP))) - if err != nil { - return fmt.Errorf("failed to get raw socket: %v", err) - } - defer syscall.Close(fd) - - if err := syscall.BindToDevice(fd, iface.Name); err != nil { - return fmt.Errorf("failed to bind to device: %v", err) - } - - ll := syscall.SockaddrLinklayer{ - Protocol: htons(syscall.ETH_P_ARP), - Ifindex: iface.Index, - Pkttype: 0, // syscall.PACKET_HOST - Hatype: m.hardwareType, - Halen: m.hardwareAddressLength, - } - target := ethernetBroadcast - for i := 0; i < len(target); i++ { - ll.Addr[i] = target[i] - } - - b, err := m.bytes() - if err != nil { - return fmt.Errorf("failed to convert ARP message: %v", err) - } - - if err := syscall.Bind(fd, &ll); err != nil { - return fmt.Errorf("failed to bind: %v", err) - } - if err := syscall.Sendto(fd, b, 0, &ll); err != nil { - return fmt.Errorf("failed to send: %v", err) - } - - return nil -} - -// ARPSendGratuitous sends a gratuitous ARP message via the specified interface. -func ARPSendGratuitous(cidr, iface string) error { - i, err := net.InterfaceByName(iface) - if err != nil { - return fmt.Errorf("failed to get interface %q: %v", iface, err) - } - ip, _, err := net.ParseCIDR(cidr) - if err != nil { - return fmt.Errorf("failed to parse cidr: %s", cidr) - } - - //log.Infof("Broadcasting ARP update for %s (%s) via %s", address, iface.HardwareAddr, iface.Name) - m, err := gratuitousARP(ip, i.HardwareAddr) - if err != nil { - return err - } - return sendARP(i, m) -} diff --git a/run_etcd.sh b/run_etcd.sh index 78afcb8..748cbdd 100755 --- a/run_etcd.sh +++ b/run_etcd.sh @@ -1,4 +1,5 @@ #!/bin/sh +n=${1:-1} cd data -../haproxy-lb -config etcd.yaml -v 2 +../haproxy-lb -config etcd${n}.yaml -v 2