implemented api handlers

cmd/haproxy-lb/main.go

@@ -78,8 +78,8 @@ func main() {
 	}()
 	sigs := make(chan os.Signal)
 	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
-	<-sigs
-	log.Info("haproxy-lb shutdown")
+	s := <-sigs
+	log.Info("haproxy-lb shutdown", "signal", s.String())
 	c.Stop()
 	log.Info("haproxy-lb stopped")
 }

pkg/alloc/alloc.go

@@ -36,22 +36,26 @@ func (a *Alloc) UpdateDB(db *db.DB) {
 	a.Unlock()
 }
 
+func (a *Alloc) cidr(ip net.IP) string {
+	gw := new(net.IPNet)
+	*gw = *a.gateway
+	gw.IP = ip
+	return gw.String()
+}
+
 func (a *Alloc) AllocIP(ctx context.Context, name string) (addr string, err error) {
 	a.Lock()
 	defer a.Unlock()
 	for _, cidr := range a.pool {
 		c := ipaddr.NewCursor([]ipaddr.Prefix{*ipaddr.NewPrefix(cidr)})
 		for pos := c.First(); pos != nil; pos = c.Next() {
-			ip := pos.IP.String()
-			if !a.db.IPExists(ip) {
-				err = a.db.SetIP(ip, name)
+			cidr := a.cidr(pos.IP)
+			if !a.db.IPExists(cidr) {
+				err = a.db.SetIP(cidr, name)
 				if err != nil {
 					return
 				}
-				var gw *net.IPNet
-				*gw = *a.gateway
-				gw.IP = pos.IP
-				addr = gw.String()
+				addr = pos.IP.String()
 				return
 			}
 		}

pkg/api/types/context.go

@@ -2,6 +2,7 @@ package types
 
 import (
 	"encoding/json"
+	"io"
 	"net/http"
 
 	"github.com/go-logr/logr"
@@ -26,20 +27,19 @@ func NewContext(w http.ResponseWriter, r *http.Request, data *ContextData, log l
 		Request:  r,
 		Response: w,
 		Data:     data,
-		logger:   log,
+		Log:      log,
 	}
 }
 
 // Context is the context object shared between http handlers
 type Context struct {
-	Body   []byte
 	Status int
 
 	Request  *http.Request
 	Response http.ResponseWriter
 
 	Data *ContextData
-	logger logr.Logger
+	Log  logr.Logger
 }
 
 // Method returns the request method
@@ -70,7 +70,7 @@ func (c *Context) Var(name string) (ret string) {
 }
 
 func (c *Context) log() {
-	c.logger.Info("access",
+	c.Log.Info("access",
 		"addr", c.Request.RemoteAddr,
 		"method", c.Request.Method,
 		"url", c.Request.URL.Path,
@@ -79,23 +79,38 @@ func (c *Context) log() {
 }
 
 func (c *Context) NotFound() {
-	c.Status = http.StatusNotFound
-	c.Response.WriteHeader(http.StatusNotFound)
-	c.Response.Write([]byte(`{"message":"not found","status":404}`))
-	c.log()
+	c.Err(ErrNotFound)
 }
 
+//TODO func (c *Context) MethodNotAllowed() {
+//	c.Status = http.StatusMethodNotAllowed
+//	c.Response.WriteHeader(http.StatusMethodNotAllowed)
+//	c.Response.Write([]byte(`{"message":"method not allowed","status":405}`))
+//	c.log()
+//}
+
 // OK is a empty HTTP 200 JSON response
 func (c *Context) OK() {
 	c.Response.Write([]byte("{}"))
 	c.log()
 }
 
+func (c *Context) ReadBody() (b []byte, err error) {
+	b, err = io.ReadAll(c.Request.Body)
+	c.Request.Body.Close()
+	return
+}
+
+func (c *Context) Body(b []byte) {
+	c.Response.Write(b)
+	c.log()
+}
+
 // JSON is a json response
 func (c *Context) JSON(v interface{}) {
 	err := json.NewEncoder(c.Response).Encode(v)
 	if err != nil {
-		c.Err(ise)
+		c.Err(ErrISE)
 		return
 	}
 	c.log()

pkg/api/types/error.go

@@ -19,7 +19,7 @@ type Err struct {
 
 // Error returns the error string
 func (a Err) Error() string {
-	return strconv.Itoa(a.status) + ": " + a.err
+	return `{"message":"` + a.err + `","status":` + strconv.Itoa(a.status) + `}`
 }
 
 // Status returns the HTTP status code
@@ -35,11 +35,11 @@ func newError(err string, status int) Error {
 }
 
 var (
-	badreq  = newError("bad request", http.StatusBadRequest)
-	invalid = newError("invalid data", http.StatusBadRequest)
-	forbid  = newError("forbidden", http.StatusForbidden)
-	nf      = newError("resource not found", http.StatusNotFound)
-	mna     = newError("method not allowed", http.StatusMethodNotAllowed)
-	exists  = newError("resource already exists", http.StatusConflict)
-	ise     = newError("internal server error", http.StatusInternalServerError)
+	ErrBadreq    = newError("bad request", http.StatusBadRequest)
+	ErrInvalid   = newError("invalid data", http.StatusBadRequest)
+	ErrForbidden = newError("forbidden", http.StatusForbidden)
+	ErrNotFound  = newError("not found", http.StatusNotFound)
+	ErrMNA       = newError("method not allowed", http.StatusMethodNotAllowed)
+	ErrExists    = newError("resource already exists", http.StatusConflict)
+	ErrISE       = newError("internal server error", http.StatusInternalServerError)
 )

pkg/api/v1/schema/schema.go

@@ -1,5 +1,11 @@
 package schema
 
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+)
+
 /*
 healthCheckNodePort
 
@@ -30,3 +36,33 @@ type Server struct {
 	IP   string `json:"ip,omitempty"`
 	Port int    `json:"port,omitempty"`
 }
+
+func (lb LoadBalancer) Validate() error {
+	if lb.Name == "" {
+		return errors.New("LoadBalancer.Name can not be empty")
+	}
+	if lb.IP == "" {
+		return errors.New("LoadBalancer.IP can not be empty")
+	}
+	for i, p := range lb.Ports {
+		if p.Port < 0 || p.Port > 65535 {
+			return fmt.Errorf("LoadBalancer.Ports[%d].Port is invalid (range 0-65535", i)
+		}
+		for j, s := range p.Servers {
+			if s.Name == "" {
+				return fmt.Errorf("LoadBalancer.Ports[%d].Servers[%d].Name can not be empty", i, j)
+			}
+			if s.IP == "" {
+				return fmt.Errorf("LoadBalancer.Ports[%d].Servers[%d].IP can not be empty", i, j)
+			}
+			if s.Port < 0 || s.Port > 65535 {
+				return fmt.Errorf("LoadBalancer.Ports[%d].Servers[%d].Port is invalid (range 0-65535)", i, j)
+			}
+		}
+	}
+	return nil
+}
+
+func (lb LoadBalancer) JSON() ([]byte, error) {
+	return json.Marshal(lb)
+}

pkg/api/v1/server/handler.go

@@ -3,8 +3,12 @@ package server
 import (
 	//"golang.org/x/crypto/bcrypt"
 	//"net/http"
+	"context"
+	"encoding/json"
 
 	"git.giftfish.de/ston1th/haproxy-lb/pkg/api/types"
+	"git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/schema"
+	"git.giftfish.de/ston1th/haproxy-lb/pkg/cluster"
 )
 
 func authHandler(h types.CtxHandler) types.CtxHandler {
@@ -22,10 +26,115 @@ func healthzHandler(ctx *types.Context) {
 	ctx.OK()
 }
 
-func lbListHandler(ctx *types.Context) {
-	ctx.OK()
+func name(c, n string) string {
+	return c + "/" + n
+}
+
+func lbname(c, n string) string {
+	return c + "_" + n
 }
 
 func lbHandler(ctx *types.Context) {
+	cl := ctx.Var("cluster")
+	n := ctx.Var("name")
+	name := name(cl, n)
+	switch ctx.Method() {
+	case "GET":
+		b, err := ctx.Data.DB.GetLB(name)
+		if err != nil {
+			ctx.Log.Error(err, "error reading loadbalancer", "cluster", cl, "name", n)
+			if err == cluster.ErrKeyNotFound {
+				ctx.Err(types.ErrNotFound)
+				return
+			}
+			ctx.Err(types.ErrISE)
+			return
+		}
+		ctx.Body(b)
+	case "POST":
+		// TODO
+		// handle if loadbalancer already exists
+		// decode existing LB and read Name + IP from there
+		b, err := ctx.ReadBody()
+		if err != nil {
+			ctx.Log.Error(err, "error reading request body", "cluster", cl, "name", n)
+			ctx.Err(types.ErrISE)
+			return
+		}
+		var lb schema.LoadBalancer
+		err = json.Unmarshal(b, &lb)
+		if err != nil {
+			ctx.Log.Error(err, "error decoding request body", "cluster", cl, "name", n)
+			ctx.Err(types.ErrInvalid)
+			return
+		}
+		lb.Name = lbname(cl, n)
+		lb.IP, err = ctx.Data.Alloc.AllocIP(context.Background(), name)
+		if err != nil {
+			ctx.Log.Error(err, "error allocating IP for loadbalancer", "cluster", cl, "name", n)
+			ctx.Err(types.ErrISE)
+			return
+		}
+		err = lb.Validate()
+		if err != nil {
+			ctx.Log.Error(err, "error validating loadbalancer", "cluster", cl, "name", n)
+			ctx.Err(types.ErrInvalid)
+			return
+		}
+		b, err = lb.JSON()
+		if err != nil {
+			ctx.Log.Error(err, "error encoding loadbalancer", "cluster", cl, "name", n)
+			ctx.Err(types.ErrISE)
+			return
+		}
+		err = ctx.Data.DB.SetLB(name, b)
+		if err != nil {
+			ctx.Log.Error(err, "error writing loadbalancer", "cluster", cl, "name", n)
+			ctx.Err(types.ErrISE)
+			return
+		}
 		ctx.OK()
+	case "DELETE":
+		err := ctx.Data.Alloc.FreeIP(context.Background(), name)
+		if err != nil {
+			ctx.Log.Error(err, "error freeing IP of loadbalancer", "cluster", cl, "name", n)
+			ctx.Err(types.ErrISE)
+			return
+		}
+		err = ctx.Data.DB.DeleteLB(name)
+		if err != nil {
+			ctx.Log.Error(err, "error deleting loadbalancer", "cluster", cl, "name", n)
+			ctx.Err(types.ErrISE)
+			return
+		}
+		ctx.OK()
+	}
+}
+
+func lbList(lbs map[string][]byte) (m map[string]json.RawMessage) {
+	m = make(map[string]json.RawMessage)
+	for k, v := range lbs {
+		m[k] = json.RawMessage(v)
+	}
+	return
+}
+
+func lbClusterListHandler(ctx *types.Context) {
+	cl := ctx.Var("cluster")
+	m, err := ctx.Data.DB.GetLBs(cl)
+	if err != nil {
+		ctx.Log.Error(err, "error reading loadbalancers")
+		ctx.Err(types.ErrISE)
+		return
+	}
+	ctx.JSON(lbList(m))
+}
+func lbListHandler(ctx *types.Context) {
+	m, err := ctx.Data.DB.GetLBs("")
+	if err != nil {
+		ctx.Log.Error(err, "error reading loadbalancers")
+		ctx.Err(types.ErrISE)
+		return
+	}
+	ctx.JSON(lbList(m))
 }

pkg/api/v1/server/routes.go

@@ -18,6 +18,11 @@ var Routes = []types.Route{
 		authHandler(lbListHandler),
 		[]string{"GET"},
 	},
+	{
+		v1 + "/lb/{cluster:[a-zA-Z0-9-]+$}",
+		authHandler(lbClusterListHandler),
+		[]string{"GET"},
+	},
 	{
 		v1 + "/lb/{cluster:[a-zA-Z0-9-]+}/{name:[a-zA-Z0-9-]+$}",
 		authHandler(lbHandler),

pkg/config/config.go

@@ -19,7 +19,7 @@ const (
 	EtcdLogLevel = "error"
 	EtcdPrefix   = "/haproxy-lb"
 
-	HAProxyConfigFile = "/etc/haproxy/haproxy.cf"
+	HAProxyConfigFile = "/etc/haproxy/haproxy.cfg"
 )
 
 func ParseFile(file string) (cfg *Config, err error) {

pkg/controller/vip.go

@@ -51,16 +51,22 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call
 					continue
 				}
 				addIPs(cc, n, ips)
-
 				// TODO
-				//cfg, err := cc.Get("config")
-				//if err != nil {
-				//	cc.Error(err, "error getting config key")
-				//	<-t.C
-				//	continue
-				//}
-				var lbs haproxy.Config
-				err = ha.UpdateConfig(ctx, lbs)
+				// deleted LB IPs need to be removed
+
+				lbs, err := db.GetLBs("")
+				if err != nil && err != cluster.ErrPrefixNotFound {
+					cc.Error(err, "error reading lb list")
+					<-t.C
+					continue
+				}
+				lbcfg, err := haproxy.NewConfig(lbs)
+				if err != nil {
+					cc.Error(err, "error reading haproxy config")
+					<-t.C
+					continue
+				}
+				err = ha.UpdateConfig(ctx, lbcfg)
 				if err != nil {
 					cc.Error(err, "error updating haproxy config")
 					<-t.C

pkg/db/db.go

@@ -15,6 +15,7 @@ func New(kv cluster.KV) *DB {
 const (
 	ipPrefix   = "/ip/"
 	namePrefix = "/name/"
+	lbPrefix   = "/lb/"
 )
 
 func (db *DB) GetIPs() (ips []string, err error) {
@@ -54,3 +55,31 @@ func (db *DB) GetName(name string) (string, error) {
 	v, err := db.kv.Get(namePrefix + name)
 	return string(v), err
 }
+
+func (db *DB) GetLBs(cl string) (lbs map[string][]byte, err error) {
+	lbs = make(map[string][]byte)
+	if cl != "" {
+		cl = cl + "/"
+	}
+	m, err := db.kv.GetPrefix(lbPrefix + cl)
+	if err == cluster.ErrPrefixNotFound {
+		err = nil
+		return
+	}
+	if err != nil {
+		return
+	}
+	for k, v := range m {
+		lbs[k] = v
+	}
+	return
+}
+func (db *DB) GetLB(name string) ([]byte, error) {
+	return db.kv.Get(lbPrefix + name)
+}
+func (db *DB) SetLB(name string, lb []byte) error {
+	return db.kv.Set(lbPrefix+name, lb)
+}
+func (db *DB) DeleteLB(name string) error {
+	return db.kv.Delete(lbPrefix + name)
+}

pkg/etcd/cluster.go

@@ -10,11 +10,6 @@ import (
 	clientv3 "go.etcd.io/etcd/client/v3"
 )
 
-var (
-	ErrKeyNotFound    = errors.New("etcd: key not found")
-	ErrPrefixNotFound = errors.New("etcd: prefix not found")
-)
-
 type Cluster struct {
 	logr.Logger
 	cli *clientv3.Client
@@ -56,8 +51,8 @@ func (c *Cluster) Get(k string) (v []byte, err error) {
 	return
 }
 
-func (c *Cluster) GetPrefix(k string) (m map[string][]byte, err error) {
-	r, err := c.cli.Get(c.ctx, c.kvPrefix+"/"+k, clientv3.WithPrefix())
+func (c *Cluster) GetPrefix(pk string) (m map[string][]byte, err error) {
+	r, err := c.cli.Get(c.ctx, c.kvPrefix+"/"+pk, clientv3.WithPrefix())
 	if err != nil {
 		return
 	}
@@ -67,8 +62,8 @@ func (c *Cluster) GetPrefix(k string) (m map[string][]byte, err error) {
 	}
 	m = make(map[string][]byte)
 	for _, kv := range r.Kvs {
-		k := strings.TrimPrefix(c.kvPrefix+"/", string(kv.Key))
-		m[k] = kv.Value
+		key := strings.TrimPrefix(string(kv.Key), c.kvPrefix+"/"+pk)
+		m[key] = kv.Value
 	}
 	return
 }

pkg/haproxy/config.go

@@ -1,29 +1,43 @@
 package haproxy
 
+import "encoding/json"
+
 type Config struct {
-	LoadBalancers []LoadBalancer
+	LoadBalancers []*LoadBalancer
+}
+
+func NewConfig(m map[string][]byte) (c Config, err error) {
+	for _, v := range m {
+		lb := new(LoadBalancer)
+		err = json.Unmarshal(v, lb)
+		if err != nil {
+			return
+		}
+		c.LoadBalancers = append(c.LoadBalancers, lb)
+	}
+	return
 }
 
 type LoadBalancer struct {
-	Name    string
-	IP      string
-	Options Options
-	Ports   []Port
+	Name    string  `json:"name"`
+	IP      string  `json:"ip"`
+	Options Options `json:"options"`
+	Ports   []Port  `json:"ports"`
 }
 
 type Options struct {
-	Frontend      []string
-	Backend       []string
-	DefaultServer string
+	Frontend      []string `json:"frontend"`
+	Backend       []string `json:"backend"`
+	DefaultServer string   `json:"defaultServer"`
 }
 
 type Port struct {
-	Port    int
-	Servers []Server
+	Port    int      `json:"port"`
+	Servers []Server `json:"servers"`
 }
 
 type Server struct {
-	Name string
-	IP   string
-	Port int
+	Name string `json:"name"`
+	IP   string `json:"ip"`
+	Port int    `json:"port"`
 }

pkg/haproxy/haproxy.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/sha256"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
@@ -51,7 +52,11 @@ func (ha *HAProxyManager) checkConfig(ctx context.Context, lbs Config) (hash []b
 		return
 	}
 	hash = h.Sum(nil)
-	return hash, exec.CommandContext(ctx, "haproxy", "-c", "-f", tmp.Name()).Run()
+	out, err := exec.CommandContext(ctx, "haproxy", "-c", "-f", tmp.Name()).CombinedOutput()
+	if err != nil {
+		err = fmt.Errorf("%s:%w", string(out), err)
+	}
+	return hash, err
 }
 
 func (ha *HAProxyManager) UpdateConfig(ctx context.Context, lbs Config) error {

pkg/haproxy/template.go

@@ -2,10 +2,12 @@ package haproxy
 
 const haproxyConfigTemplate = `global
   daemon
-  chroot /var/lib/haproxy
+  #chroot /var/lib/haproxy
+  chroot /usr/share/haproxy
+  pidfile /run/haproxy.pid
   user haproxy
   group haproxy
-  maxconn 5000
+  maxconn 20000
 defaults
   mode tcp
   log global
@@ -15,29 +17,37 @@ defaults
   timeout client 5000ms
   timeout tunnel 3600s
   timeout check 10s
+frontend stats
+  mode http
+  bind 127.0.0.1:8404
+  stats enable
+  stats uri /stats
+  stats refresh 10s
+  stats admin if LOCALHOST
 {{ range $item := .LoadBalancers -}}
 {{ range $port := $item.Ports -}}
 frontend fr_{{ $item.Name }}_{{ $port.Port }}
   bind {{ $item.IP }}:{{ $port.Port }}
-  {{ range $opt := $item.Options.Frontend -}}
+{{- range $opt := $item.Options.Frontend }}
   option {{ $opt }}
-  {{ end -}}
+{{- end }}
   # option tcplog
   # option splice-request
   default_backend ba_{{ $item.Name }}_{{ $port.Port }}
 backend ba_{{ $item.Name }}_{{ $port.Port }}
   balance roundrobin
-  {{ range $opt := $item.Options.Backend -}}
+{{- range $opt := $item.Options.Backend }}
   option {{ $opt }}
-  {{ end -}}
+{{- end }}
   # option splice-response
   # option httpchk GET /
   default-server inter 3s downinter 10s fall 2 rise 2 on-marked-down shutdown-sessions {{ $item.Options.DefaultServer }}
   # send-proxy-v2
   # check-send-proxy
   # port <healthCheckNodePort>
-  {{ range $srv := $port.Servers -}}
+{{- range $srv := $port.Servers }}
   server {{ $srv.Name }} {{ $srv.IP }}:{{ $srv.Port }} check
-  {{ end -}}
+{{- end }}
 {{ end -}}
-{{ end -}}`
+{{ end -}}
+`

pkg/raft/kv.go

@@ -33,7 +33,8 @@ func (kv *kv) GetPrefix(pk string) (m map[string][]byte, err error) {
 	m = make(map[string][]byte)
 	for k, v := range kv.m {
 		if strings.HasPrefix(k, pk) {
-			m[k] = v
+			key := strings.TrimPrefix(k, pk)
+			m[key] = v
 		}
 	}
 	if len(m) == 0 {