diff --git a/cmd/haproxy-lb/main.go b/cmd/haproxy-lb/main.go index 090ff44..9d2033b 100644 --- a/cmd/haproxy-lb/main.go +++ b/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") } diff --git a/pkg/alloc/alloc.go b/pkg/alloc/alloc.go index ca14261..05da203 100644 --- a/pkg/alloc/alloc.go +++ b/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 } } diff --git a/pkg/api/types/context.go b/pkg/api/types/context.go index 241366f..5ff429a 100644 --- a/pkg/api/types/context.go +++ b/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 + Data *ContextData + 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() diff --git a/pkg/api/types/error.go b/pkg/api/types/error.go index acfac67..5c4b149 100644 --- a/pkg/api/types/error.go +++ b/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) ) diff --git a/pkg/api/v1/schema/schema.go b/pkg/api/v1/schema/schema.go index 3a23917..e36d945 100644 --- a/pkg/api/v1/schema/schema.go +++ b/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) +} diff --git a/pkg/api/v1/server/handler.go b/pkg/api/v1/server/handler.go index 0ff4d4e..b6ce5fb 100644 --- a/pkg/api/v1/server/handler.go +++ b/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) { - ctx.OK() + 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)) } diff --git a/pkg/api/v1/server/routes.go b/pkg/api/v1/server/routes.go index 3da811a..628281e 100644 --- a/pkg/api/v1/server/routes.go +++ b/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), diff --git a/pkg/config/config.go b/pkg/config/config.go index e073906..d75cbf6 100644 --- a/pkg/config/config.go +++ b/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) { diff --git a/pkg/controller/vip.go b/pkg/controller/vip.go index f4ae16c..a9c5429 100644 --- a/pkg/controller/vip.go +++ b/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 diff --git a/pkg/db/db.go b/pkg/db/db.go index 1a4c28e..3655ea9 100644 --- a/pkg/db/db.go +++ b/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) +} diff --git a/pkg/etcd/cluster.go b/pkg/etcd/cluster.go index 2202e99..3f86e65 100644 --- a/pkg/etcd/cluster.go +++ b/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 } diff --git a/pkg/haproxy/config.go b/pkg/haproxy/config.go index 2833f36..dd2f2d7 100644 --- a/pkg/haproxy/config.go +++ b/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"` } diff --git a/pkg/haproxy/haproxy.go b/pkg/haproxy/haproxy.go index f48c77b..90f5073 100644 --- a/pkg/haproxy/haproxy.go +++ b/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 { diff --git a/pkg/haproxy/template.go b/pkg/haproxy/template.go index 5faf438..6df1eeb 100644 --- a/pkg/haproxy/template.go +++ b/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 - {{ range $srv := $port.Servers -}} +{{- range $srv := $port.Servers }} server {{ $srv.Name }} {{ $srv.IP }}:{{ $srv.Port }} check - {{ end -}} +{{- end }} {{ end -}} -{{ end -}}` +{{ end -}} +` diff --git a/pkg/raft/kv.go b/pkg/raft/kv.go index 2a3b0a3..4fa0bfd 100644 --- a/pkg/raft/kv.go +++ b/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 {