diff --git a/pkg/api/v1/server/server.go b/pkg/api/server.go similarity index 79% rename from pkg/api/v1/server/server.go rename to pkg/api/server.go index b7770a7..100ed76 100644 --- a/pkg/api/v1/server/server.go +++ b/pkg/api/server.go @@ -1,4 +1,4 @@ -package server +package api import ( "context" @@ -10,6 +10,9 @@ import ( "path/filepath" "time" + "git.giftfish.de/ston1th/haproxy-lb/pkg/api/types" + serverv1 "git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/server" + "github.com/go-logr/logr" "github.com/gorilla/mux" ) @@ -30,6 +33,14 @@ type Server struct { debug bool } +type notFoundHandler struct { + s *Server +} + +func (nf *notFoundHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + types.NewContext(w, r, nf.s).NotFound() +} + // NewHTTPServer returns a new HTTPServer func NewServer(log logr.Logger) *Server { s := &Server{ @@ -46,7 +57,7 @@ func NewServer(log logr.Logger) *Server { debug: core.C.Debug, } s.mux.NotFoundHandler = ¬FoundHandler{s} - for _, v := range routes { + for _, v := range serverv1.Routes { s.mux.HandleFunc(v.Path, s.contextWrapper(v.Handler)).Methods(v.Methods...) } s.start() @@ -73,9 +84,9 @@ func (s *Server) start() error { return nil } -func (s *Server) contextWrapper(h ctxHandler) http.HandlerFunc { +func (s *Server) contextWrapper(h types.CtxHandler) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { - h(newContext(w, r, s)) + h(types.NewContext(w, r, s)) } } diff --git a/pkg/api/v1/server/context.go b/pkg/api/types/context.go similarity index 89% rename from pkg/api/v1/server/context.go rename to pkg/api/types/context.go index eed4ae4..19d3023 100644 --- a/pkg/api/v1/server/context.go +++ b/pkg/api/types/context.go @@ -1,21 +1,22 @@ -package server +package types import ( "encoding/json" "fmt" "net/http" + "github.com/go-logr/logr" "github.com/gorilla/mux" ) -func newContext(w http.ResponseWriter, r *http.Request, srv *Server) *Context { +func NewContext(w http.ResponseWriter, r *http.Request, log logr.Logger) *Context { h := w.Header() h.Set("Content-Type", "application/json") return &Context{ Status: http.StatusOK, Request: r, Response: w, - s: srv, + log: log, } } @@ -26,7 +27,7 @@ type Context struct { Request *http.Request Response http.ResponseWriter - s *Server + log logr.Logger } // Method returns the request method @@ -57,7 +58,7 @@ func (c *Context) Var(name string) (ret string) { } func (c *Context) log() { - c.s.log.Info("access", + c.log.Info("access", "addr", c.Request.RemoteAddr, "method", c.Request.Method, "url", c.Request.URL, @@ -100,4 +101,4 @@ func (c *Context) HTTPErr(err string, status int) { c.log() } -type ctxHandler func(*Context) +type CtxHandler func(*Context) diff --git a/pkg/api/api.go b/pkg/api/types/types.go similarity index 85% rename from pkg/api/api.go rename to pkg/api/types/types.go index 34ee33f..e57c3ec 100644 --- a/pkg/api/api.go +++ b/pkg/api/types/types.go @@ -1,4 +1,4 @@ -package api +package types type Route struct { Path string diff --git a/pkg/api/v1/schema/schema.go b/pkg/api/v1/schema/schema.go index 04747c1..cd6f015 100644 --- a/pkg/api/v1/schema/schema.go +++ b/pkg/api/v1/schema/schema.go @@ -5,13 +5,9 @@ healthCheckNodePort */ -type LBConfig struct { - LoadBalancers []LoadBalancer -} - type LoadBalancer struct { Name string `json:"name"` - IP string `json:"ip"` + IP string `json:"ip,omitempty"` Options Options `json:"options,omitempty"` Ports []Port `json:"ports,omitempty"` } diff --git a/pkg/api/v1/server/handler.go b/pkg/api/v1/server/handler.go index 9e55e85..73e3088 100644 --- a/pkg/api/v1/server/handler.go +++ b/pkg/api/v1/server/handler.go @@ -3,18 +3,12 @@ package server import ( //"golang.org/x/crypto/bcrypt" "net/http" + + "git.giftfish.de/ston1th/haproxy-lb/pkg/api/types" ) -type notFoundHandler struct { - s *Server -} - -func (nf *notFoundHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - newContext(w, r, nf.s).NotFound() -} - -func authHandler(h ctxHandler) ctxHandler { - return func(ctx *Context) { +func authHandler(h types.CtxHandler) types.CtxHandler { + return func(ctx *types.Context) { // TODO // test:123456 // test:$2b$10$zNfGTAQ94vifj2wtGsv1W.yZRe4/rDBzQixpB6FbrTed4BnHuNqBS @@ -23,7 +17,7 @@ func authHandler(h ctxHandler) ctxHandler { } } -func healthzHandler(ctx *Context) { +func healthzHandler(ctx *types.Context) { // TODO maybe report etcd/raft stats ctx.OK() } diff --git a/pkg/api/v1/server/routes.go b/pkg/api/v1/server/routes.go index 8b76d91..26d3ac2 100644 --- a/pkg/api/v1/server/routes.go +++ b/pkg/api/v1/server/routes.go @@ -1,13 +1,13 @@ package server import ( - "git.giftfish.de/ston1th/haproxy-lb/pkg/api" + "git.giftfish.de/ston1th/haproxy-lb/pkg/api/types" schemav1 "git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/schema" ) const v1 = "/" + schemav1.Version -var routes = []api.Route{ +var routes = []types.Route{ { "/healthz", healthzHandler, diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go index ca755df..62aa492 100644 --- a/pkg/cluster/cluster.go +++ b/pkg/cluster/cluster.go @@ -16,6 +16,7 @@ type Cluster interface { type KV interface { Get(k string) ([]byte, error) + GetPrefix(k string) (map[string][]byte, error) Set(k string, v []byte) error Delete(k string) error } diff --git a/pkg/controller/vip.go b/pkg/controller/vip.go index eb296d8..61f07e5 100644 --- a/pkg/controller/vip.go +++ b/pkg/controller/vip.go @@ -5,17 +5,60 @@ import ( "os/exec" "time" - "git.giftfish.de/ston1th/vipman/pkg/cluster" - "git.giftfish.de/ston1th/vipman/pkg/config" - "git.giftfish.de/ston1th/vipman/pkg/vip" + "git.giftfish.de/ston1th/haproxy-kb/pkg/cluster" + "git.giftfish.de/ston1th/haproxy-kb/pkg/config" + "git.giftfish.de/ston1th/haproxy-kb/pkg/haproxy" + "git.giftfish.de/ston1th/haproxy-kb/pkg/vip" "github.com/go-logr/logr" ) -func NewVIPController(cfg *config.Config) (callbacks cluster.Callbacks, err error) { +type UpdateAction int + +const ( + Add UpdateAction = iota + Delete +) + +type ConfigUpdate struct { + Action UpdateAction + Config haproxy.LoadBalancer +} + +func readConfigUpdates(c chan ConfigUpdate) (cu []ConfigUpdate) { + for len(c) > 0 { + select { + case config := <-c: + cu = append(cu, config) + } + } + return +} + +func applyConfigUpdates(cc cluster.CallbackContext, cu []ConfigUpdate) error { + for _, c := range cu { + switch c.Action { + case Add: + addIPs(cc, n, []string{c.Config.IP}) + cc.Set(c.Config.IP, nil) + case Delete: + cc.Delete(c.Config.IP) + deleteIPs(cc, n, []string{c.Config.IP}) + } + } +} + +func NewLBController(cfg *config.Config) (callbacks cluster.Callbacks, err error) { n, err := vip.NewNetworkWithLabel(cfg.Interface, cfg.Label) if err != nil { return } + //TODO + ha, err := haproxy.NewHAProxyManager("") + if err != nil { + return + } + configUpdate := make(chan ConfigUpdate, 100) + callbacks = cluster.Callbacks{ Leader: func(ctx context.Context, cc cluster.CallbackContext) { t := time.NewTicker(time.Second * 10) @@ -29,7 +72,20 @@ func NewVIPController(cfg *config.Config) (callbacks cluster.Callbacks, err erro } for { cc.Info("leading", "id", cc.ID()) + cfg, err := cc.Get("config") + if err != nil { + cc.Error("error getting config key", err) + <-t.C + continue + } addIPs(cc, n, cfg.VirtualIPs) + var lbs haproxy.Config + err = ha.UpdateConfig(ctx, lbs) + if err != nil { + cc.Error("error updating haproxy config", err) + <-t.C + continue + } select { case <-ctx.Done(): t.Stop() diff --git a/pkg/etcd/cluster.go b/pkg/etcd/cluster.go index 2aa3521..b2478f2 100644 --- a/pkg/etcd/cluster.go +++ b/pkg/etcd/cluster.go @@ -3,6 +3,7 @@ package etcd import ( "context" "errors" + "strings" "git.giftfish.de/ston1th/vipman/pkg/cluster" //"go.etcd.io/etcd/clientv3" @@ -52,6 +53,23 @@ 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()) + if err != nil { + return + } + if len(r.Kvs) == 0 { + err = ErrKeyNotFound + return + } + m = make(map[string][]byte) + for _, kv := range r.Kvs { + k := strings.TrimPrefix(c.kvPrefix+"/", string(kv.Key)) + m[k] = kv.Value + } + return +} + func (c *Cluster) Set(k string, v []byte) (err error) { _, err = c.cli.Put(c.ctx, c.kvPrefix+"/"+k, string(v)) return diff --git a/pkg/haproxy/config.go b/pkg/haproxy/config.go new file mode 100644 index 0000000..2833f36 --- /dev/null +++ b/pkg/haproxy/config.go @@ -0,0 +1,29 @@ +package haproxy + +type Config struct { + LoadBalancers []LoadBalancer +} + +type LoadBalancer struct { + Name string + IP string + Options Options + Ports []Port +} + +type Options struct { + Frontend []string + Backend []string + DefaultServer string +} + +type Port struct { + Port int + Servers []Server +} + +type Server struct { + Name string + IP string + Port int +} diff --git a/pkg/haproxy/haproxy.go b/pkg/haproxy/haproxy.go index 1c67d12..c29ce04 100644 --- a/pkg/haproxy/haproxy.go +++ b/pkg/haproxy/haproxy.go @@ -1,7 +1,10 @@ package haproxy import ( + "bytes" "context" + "crypto/sha256" + "io" "io/ioutil" "os" "sync" @@ -18,6 +21,7 @@ type HAProxyManager struct { configFile string serviceManager ServiceManager template *template.Template + hash []byte } func NewHAProxyManager(configFile string) (*HAProxyManager, error) { @@ -32,26 +36,32 @@ func NewHAProxyManager(configFile string) (*HAProxyManager, error) { }, nil } -func (ha *HAProxyManager) checkConfig(ctx context.Context, configFile string) error { - return exec.CommandContext(ctx, "haproxy", "-c", "-f", configFile).Run() -} - -func (ha *HAProxyManager) UpdateConfig(ctx context.Context, lbs []config.LoadBalancer) error { - ha.Lock() - defer ha.Unlock() +func (ha *HAProxyManager) checkConfig(ctx context.Context, lbs Config) (hash []byte, err error) { tmp, err := ioutil.TempFile(os.TempDir(), "haproxy_") if err != nil { - return err + return } defer tmp.Close() defer os.Remove(tmp.Name()) - err = ha.template.Execute(tmp, lbs) + h := sha256.New() + w := io.MultiWriter(tmp, h) + err = ha.template.Execute(w, lbs) + if err != nil { + return + } + hash = h.Sum(nil) + return hash, exec.CommandContext(ctx, "haproxy", "-c", "-f", tmp.Name()).Run() +} + +func (ha *HAProxyManager) UpdateConfig(ctx context.Context, lbs Config) error { + ha.Lock() + defer ha.Unlock() + hash, err := ha.checkConfig(ctx, lbs) if err != nil { return err } - err = ha.checkConfig(ctx, tmp.Name()) - if err != nil { - return err + if bytes.Equal(ha.hash, hash) { + return nil } file, err := os.OpenFile(ha.configFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) if err != nil { @@ -63,5 +73,6 @@ func (ha *HAProxyManager) UpdateConfig(ctx context.Context, lbs []config.LoadBal return err } file.Close() + ha.hash = hash return ha.serviceManager.Reload() } diff --git a/pkg/haproxy/template.go b/pkg/haproxy/template.go index 6b9d71a..5faf438 100644 --- a/pkg/haproxy/template.go +++ b/pkg/haproxy/template.go @@ -15,7 +15,7 @@ defaults timeout client 5000ms timeout tunnel 3600s timeout check 10s -{{ range $item := . -}} +{{ range $item := .LoadBalancers -}} {{ range $port := $item.Ports -}} frontend fr_{{ $item.Name }}_{{ $port.Port }} bind {{ $item.IP }}:{{ $port.Port }} diff --git a/pkg/raft/kv.go b/pkg/raft/kv.go index 8c14923..9a7a949 100644 --- a/pkg/raft/kv.go +++ b/pkg/raft/kv.go @@ -24,6 +24,18 @@ func (kv *kv) Get(k string) ([]byte, error) { return kv.m[k], nil } +func (kv *kv) GetPrefix(pk string) (m map[string][]byte, err error) { + kv.mu.Lock() + defer kv.mu.Unlock() + m = make(map[string][]byte) + for k, v := range kv.m { + if strings.HasPrefix(k, pk) { + m[k] = v + } + } + return +} + func (kv *kv) Set(k string, v []byte) error { if kv.r.State() != raft.Leader { return ErrNotLeader