cleanup, ensure haproxy is running and disallow writes to followers

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 {

pkg/api/types/context.go

@@ -14,9 +14,10 @@ import (
 )
 
 type ContextData struct {
-	Net  netalloc.Allocator
+	Net    netalloc.Allocator
-	DB   *db.DB
+	DB     *db.DB
-	Auth *Auth
+	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

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)
 )

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 {

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)

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)
 }

pkg/vip/arp_linux.go.bak

@@ -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)
-}

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