From 6da132106f8f0bfb7ae67e9794b1c11c4fd5427a Mon Sep 17 00:00:00 2001 From: ston1th Date: Fri, 9 Apr 2021 22:38:03 +0200 Subject: [PATCH] first working version --- .gitignore | 1 + pkg/alloc/alloc.go | 10 +- pkg/api/client/client.go | 200 +++++++++++++++++++++ pkg/api/client/helper.go | 31 ++++ pkg/api/server.go | 6 +- pkg/api/{v1/server => types}/auth.go | 14 +- pkg/api/types/context.go | 16 +- pkg/api/types/error.go | 15 +- pkg/api/v1/client/client.go | 85 +++++++++ pkg/api/v1/schema/schema.go | 20 ++- pkg/api/v1/server/handler.go | 54 ++++-- pkg/api/v1/server/routes.go | 2 +- pkg/config/config.go | 17 +- pkg/controller/{vip.go => loadbalancer.go} | 38 ++-- pkg/db/db.go | 32 +++- pkg/haproxy/haproxy.go | 60 +++++-- pkg/haproxy/systemd.go | 12 ++ run_etcd.sh | 2 +- 18 files changed, 519 insertions(+), 96 deletions(-) create mode 100644 pkg/api/client/client.go create mode 100644 pkg/api/client/helper.go rename pkg/api/{v1/server => types}/auth.go (89%) create mode 100644 pkg/api/v1/client/client.go rename pkg/controller/{vip.go => loadbalancer.go} (81%) diff --git a/.gitignore b/.gitignore index aa8c878..110c76d 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,3 @@ data/ /haproxy-lb +/cmd/haproxy-lb-client diff --git a/pkg/alloc/alloc.go b/pkg/alloc/alloc.go index 05da203..bbb7054 100644 --- a/pkg/alloc/alloc.go +++ b/pkg/alloc/alloc.go @@ -46,12 +46,12 @@ func (a *Alloc) cidr(ip net.IP) 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 _, p := range a.pool { + c := ipaddr.NewCursor([]ipaddr.Prefix{*ipaddr.NewPrefix(p)}) for pos := c.First(); pos != nil; pos = c.Next() { cidr := a.cidr(pos.IP) if !a.db.IPExists(cidr) { - err = a.db.SetIP(cidr, name) + err = a.db.SetIPName(cidr, name) if err != nil { return } @@ -70,11 +70,11 @@ func (a *Alloc) FreeIP(ctx context.Context, name string) (err error) { if name == "" { return } - ip, err := a.db.GetName(name) + ip, err := a.db.GetIP(name) if err != nil { return } - return a.db.DeleteIP(ip, name) + return a.db.DeleteIPName(ip, name) } func parseRange(cidrs []string) (nets []*net.IPNet, err error) { diff --git a/pkg/api/client/client.go b/pkg/api/client/client.go new file mode 100644 index 0000000..b73035d --- /dev/null +++ b/pkg/api/client/client.go @@ -0,0 +1,200 @@ +package client + +import ( + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "io/ioutil" + "net" + "net/http" + "net/http/httputil" + "os" + "strconv" + "strings" + "sync" + "time" +) + +type Client struct { + sync.Mutex + endpoint string + username string + password string + httpClient *http.Client + debugWriter io.Writer + insecure bool +} + +type ClientOption func(*Client) + +func WithEndpoint(endpoint string) ClientOption { + return func(client *Client) { + client.endpoint = strings.TrimRight(endpoint, "/") + } +} + +func WithDebugWriter(debugWriter io.Writer) ClientOption { + return func(client *Client) { + client.debugWriter = debugWriter + } +} + +func WithHTTPClient(httpClient *http.Client) ClientOption { + return func(client *Client) { + client.httpClient = httpClient + } +} + +func WithInsecureClient(insecure bool) ClientOption { + return func(client *Client) { + client.insecure = insecure + } +} + +func WithCredentials(username, password string) ClientOption { + return func(client *Client) { + client.username = username + client.password = password + } +} + +var ( + ErrMissingEndpointEnv = errors.New("missing environment variable HAPROXY_LB_ENDPOINT") + ErrMissingUserEnv = errors.New("missing environment variable HAPROXY_LB_USER") + ErrMissingPasswordEnv = errors.New("missing environment variable HAPROXY_LB_PASSWORD") + + ErrMissingEndpoint = errors.New("missing Endpoint in config") + ErrMissingUser = errors.New("missing User in config") + ErrMissingPassword = errors.New("missing Password in config") +) + +type Config struct { + Endpoint string `json:"endpoint"` + User string `json:"user"` + Password string `json:"password"` + Insecure bool `json:"insecure"` +} + +func ClientOptionsFromConfig(cfg Config) (options []ClientOption, err error) { + if cfg.Endpoint == "" { + err = ErrMissingEndpoint + return + } + if cfg.User == "" { + err = ErrMissingUser + return + } + if cfg.Password == "" { + err = ErrMissingPassword + return + } + options = []ClientOption{ + WithEndpoint(cfg.Endpoint), + WithCredentials(cfg.User, cfg.Password), + WithInsecureClient(cfg.Insecure), + } + return +} + +func ClientOptionsFromEnv() (options []ClientOption, err error) { + endpoint := os.Getenv("HAPROXY_LB_ENDPOINT") + if endpoint == "" { + err = ErrMissingEndpointEnv + return + } + user := os.Getenv("HAPROXY_LB_USER") + if user == "" { + err = ErrMissingUserEnv + return + } + + pass := os.Getenv("HAPROXY_LB_PASSWORD") + if pass == "" { + err = ErrMissingPasswordEnv + return + } + + insecure, _ := strconv.ParseBool(os.Getenv("HAPROXY_LB_INSECURE")) + + options = []ClientOption{ + WithEndpoint(endpoint), + WithCredentials(user, pass), + WithInsecureClient(insecure), + } + return +} + +func NewClient(options ...ClientOption) *Client { + client := &Client{} + + for _, option := range options { + option(client) + } + + if client.httpClient == nil { + client.httpClient = &http.Client{Transport: &http.Transport{ + DialContext: (&net.Dialer{ + Timeout: 30 * time.Second, + }).DialContext, + TLSClientConfig: &tls.Config{ + InsecureSkipVerify: client.insecure, + }, + }} + } + return client +} + +func (c *Client) NewRequest(ctx context.Context, method, path string, body io.Reader) (*http.Request, error) { + url := c.endpoint + path + req, err := http.NewRequest(method, url, body) + if err != nil { + return nil, err + } + req.SetBasicAuth(c.username, c.password) + req = req.WithContext(ctx) + return req, nil +} + +var HTTPErr = errors.New("http error") + +func (c *Client) Do(r *http.Request, v interface{}) (resp *http.Response, err error) { + if c.debugWriter != nil { + dumpReq, err := httputil.DumpRequestOut(r, true) + if err != nil { + return &http.Response{}, err + } + fmt.Fprintf(c.debugWriter, "--- Request:\n%s\n\n", dumpReq) + } + resp, err = c.httpClient.Do(r) + if err != nil { + return + } + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + resp.Body.Close() + return + } + resp.Body.Close() + resp.Body = ioutil.NopCloser(bytes.NewReader(body)) + + if c.debugWriter != nil { + dumpResp, err := httputil.DumpResponse(resp, true) + if err != nil { + return resp, err + } + fmt.Fprintf(c.debugWriter, "--- Response:\n%s\n\n", dumpResp) + } + + err = decodeResp(resp, v) + if resp.StatusCode >= 400 && resp.StatusCode <= 599 { + if err != nil { + err = fmt.Errorf("%w: %s: %s", HTTPErr, errors.New(resp.Status), err.Error()) + return + } + err = fmt.Errorf("%w: %s", HTTPErr, errors.New(resp.Status)) + } + return +} diff --git a/pkg/api/client/helper.go b/pkg/api/client/helper.go new file mode 100644 index 0000000..4f2c705 --- /dev/null +++ b/pkg/api/client/helper.go @@ -0,0 +1,31 @@ +package client + +import ( + "encoding/json" + "errors" + "net/http" +) + +type Error struct { + Message string `json:"message"` + Status int `json:"status"` +} + +func decodeResp(resp *http.Response, v interface{}) error { + if resp.Body == nil { + return nil + } + defer resp.Body.Close() + if resp.StatusCode == 200 { + if v == nil { + return nil + } + return json.NewDecoder(resp.Body).Decode(&v) + } + var e Error + err := json.NewDecoder(resp.Body).Decode(&e) + if err != nil { + return err + } + return errors.New(e.Message) +} diff --git a/pkg/api/server.go b/pkg/api/server.go index 30f4c3f..8cfb004 100644 --- a/pkg/api/server.go +++ b/pkg/api/server.go @@ -50,14 +50,14 @@ func NewServer(c *config.Config, log logr.Logger) (*Server, error) { log: log, cidrs: c.VIP.VirtualIPs, gw: c.VIP.Gateway, - Data: &types.ContextData{ - Auth: c.Server.BasicAuth, - }, + Data: &types.ContextData{}, init: make(chan struct{}), stop: make(chan struct{}), stopKeyReset: make(chan struct{}), } + s.Data.Auth = types.NewAuth(c.Server.BasicAuth) + if c.Server.TLS != nil { cert, err := tls.LoadX509KeyPair(c.Server.TLS.Cert, c.Server.TLS.Key) if err != nil { diff --git a/pkg/api/v1/server/auth.go b/pkg/api/types/auth.go similarity index 89% rename from pkg/api/v1/server/auth.go rename to pkg/api/types/auth.go index 3687e1e..7b865df 100644 --- a/pkg/api/v1/server/auth.go +++ b/pkg/api/types/auth.go @@ -1,4 +1,4 @@ -package server +package types import ( "git.giftfish.de/ston1th/haproxy-lb/pkg/config" @@ -63,7 +63,7 @@ func (c *cache) makeRoom() { } } -type auth struct { +type Auth struct { users map[string]*creds cache *cache // bcryptMtx is there to ensure that bcrypt.CompareHashAndPassword is run @@ -71,7 +71,7 @@ type auth struct { bcryptMtx sync.Mutex } -func (a *auth) Auth(user, pass, path string) bool { +func (a *Auth) Login(user, pass, path string) bool { creds, valid := a.users[user] hash := creds.Hash @@ -103,13 +103,13 @@ type creds struct { Prefix string } -func newAuth(conf config.Config) (a *auth) { - a = &auth{ +func NewAuth(conf []config.BasicAuth) (a *Auth) { + a = &Auth{ users: make(map[string]*creds), cache: newCache(), } - for _, u := range conf.Server.BasicAuth { - a.users[u.User] = &creds{u.Hash, u.Prefix} + for _, ba := range conf { + a.users[ba.User] = &creds{ba.Hash, ba.Prefix} } return } diff --git a/pkg/api/types/context.go b/pkg/api/types/context.go index 5ff429a..dc16ee5 100644 --- a/pkg/api/types/context.go +++ b/pkg/api/types/context.go @@ -9,14 +9,13 @@ import ( "github.com/gorilla/mux" "git.giftfish.de/ston1th/haproxy-lb/pkg/alloc" - "git.giftfish.de/ston1th/haproxy-lb/pkg/config" "git.giftfish.de/ston1th/haproxy-lb/pkg/db" ) type ContextData struct { Alloc *alloc.Alloc DB *db.DB - Auth []config.BasicAuth + Auth *Auth } func NewContext(w http.ResponseWriter, r *http.Request, data *ContextData, log logr.Logger) *Context { @@ -47,6 +46,11 @@ func (c *Context) Method() string { return c.Request.Method } +// Path returns the request URL path +func (c *Context) Path() string { + return c.Request.URL.Path +} + // GetHeader returns the given http header func (c *Context) GetHeader(name string) string { return c.Request.Header.Get(name) @@ -82,16 +86,8 @@ func (c *Context) NotFound() { 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() } diff --git a/pkg/api/types/error.go b/pkg/api/types/error.go index 5c4b149..8e0fd95 100644 --- a/pkg/api/types/error.go +++ b/pkg/api/types/error.go @@ -35,11 +35,12 @@ func newError(err string, status int) Error { } var ( - 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) + ErrBadreq = newError("bad request", http.StatusBadRequest) + ErrInvalid = newError("invalid data", http.StatusBadRequest) + ErrUnauthorized = newError("unauthorized", http.StatusUnauthorized) + 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/client/client.go b/pkg/api/v1/client/client.go new file mode 100644 index 0000000..c206d7f --- /dev/null +++ b/pkg/api/v1/client/client.go @@ -0,0 +1,85 @@ +package client + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + + apiclient "git.giftfish.de/ston1th/haproxy-lb/pkg/api/client" + "git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/schema" +) + +var ( + ErrClientIsNil = errors.New("client is nil") + base = "/" + schema.Version +) + +type Client struct { + c *apiclient.Client +} + +func NewClient(c *apiclient.Client) (*Client, error) { + if c == nil { + return nil, ErrClientIsNil + } + return &Client{c}, nil +} + +func (c *Client) GetLoadBalancers(ctx context.Context) (lbs map[string]schema.LoadBalancer, err error) { + return c.GetLoadBalancersWithCluster(ctx, "") +} + +func (c *Client) GetLoadBalancersWithCluster(ctx context.Context, cluster string) (lbs map[string]schema.LoadBalancer, err error) { + path := base + "/lb" + if cluster != "" { + path += "/" + cluster + } + req, err := c.c.NewRequest(ctx, "GET", path, nil) + if err != nil { + return + } + lbs = make(map[string]schema.LoadBalancer) + _, err = c.c.Do(req, &lbs) + return +} + +func (c *Client) GetLoadBalancer(ctx context.Context, cluster, name string) (lb schema.LoadBalancer, err error) { + path := fmt.Sprintf("%s/lb/%s/%s", base, cluster, name) + req, err := c.c.NewRequest(ctx, "GET", path, nil) + if err != nil { + return + } + _, err = c.c.Do(req, &lb) + return +} + +func (c *Client) CreateLoadBalancer(ctx context.Context, cluster, name string, lb schema.LoadBalancer) (err error) { + buf := new(bytes.Buffer) + path := fmt.Sprintf("%s/lb/%s/%s", base, cluster, name) + err = lb.ValidateClient() + if err != nil { + return + } + err = json.NewEncoder(buf).Encode(lb) + if err != nil { + return + } + req, err := c.c.NewRequest(ctx, "POST", path, buf) + if err != nil { + return + } + _, err = c.c.Do(req, nil) + return +} + +func (c *Client) DeleteLoadBalancer(ctx context.Context, cluster, name string) (err error) { + path := fmt.Sprintf("%s/lb/%s/%s", base, cluster, name) + req, err := c.c.NewRequest(ctx, "DELETE", path, nil) + if err != nil { + return + } + _, err = c.c.Do(req, nil) + return +} diff --git a/pkg/api/v1/schema/schema.go b/pkg/api/v1/schema/schema.go index e36d945..89d43d8 100644 --- a/pkg/api/v1/schema/schema.go +++ b/pkg/api/v1/schema/schema.go @@ -14,7 +14,7 @@ healthCheckNodePort const Version = "v1" type LoadBalancer struct { - Name string `json:"name"` + Name string `json:"name,omitempty"` IP string `json:"ip,omitempty"` Options Options `json:"options,omitempty"` Ports []Port `json:"ports,omitempty"` @@ -37,13 +37,7 @@ type Server struct { 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") - } +func (lb LoadBalancer) ValidateClient() error { 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) @@ -63,6 +57,16 @@ func (lb LoadBalancer) Validate() error { return nil } +func (lb LoadBalancer) ValidateServer() 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") + } + return lb.ValidateClient() +} + 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 b6ce5fb..1994dee 100644 --- a/pkg/api/v1/server/handler.go +++ b/pkg/api/v1/server/handler.go @@ -5,6 +5,7 @@ import ( //"net/http" "context" "encoding/json" + "net" "git.giftfish.de/ston1th/haproxy-lb/pkg/api/types" "git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/schema" @@ -13,10 +14,15 @@ import ( func authHandler(h types.CtxHandler) types.CtxHandler { return func(ctx *types.Context) { - // TODO - // test:123456 - // test:$2b$10$zNfGTAQ94vifj2wtGsv1W.yZRe4/rDBzQixpB6FbrTed4BnHuNqBS - //bcrypt.CompareHashAndPassword(hash, pw) + user, pass, ok := ctx.Request.BasicAuth() + if !ok { + ctx.Err(types.ErrUnauthorized) + return + } + if !ctx.Data.Auth.Login(user, pass, ctx.Path()) { + ctx.Err(types.ErrForbidden) + return + } h(ctx) } } @@ -34,6 +40,14 @@ func lbname(c, n string) string { return c + "_" + n } +func getIP(cidr string) (string, error) { + ip, _, err := net.ParseCIDR(cidr) + if err != nil { + return "", err + } + return ip.String(), nil +} + func lbHandler(ctx *types.Context) { cl := ctx.Var("cluster") n := ctx.Var("name") @@ -52,9 +66,6 @@ func lbHandler(ctx *types.Context) { } 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) @@ -69,13 +80,28 @@ func lbHandler(ctx *types.Context) { 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) + cidr, err := ctx.Data.DB.GetIP(name) + if err != nil && err != cluster.ErrKeyNotFound { + ctx.Log.Error(err, "error getting existing loadbalancer ip", "cluster", cl, "name", n) ctx.Err(types.ErrISE) return } - err = lb.Validate() + if cidr != "" { + lb.IP, err = getIP(cidr) + if err != nil { + ctx.Log.Error(err, "error parsing existing ip", "cluster", cl, "name", n, "cidr", cidr) + ctx.Err(types.ErrISE) + return + } + } else { + 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.ValidateServer() if err != nil { ctx.Log.Error(err, "error validating loadbalancer", "cluster", cl, "name", n) ctx.Err(types.ErrInvalid) @@ -95,8 +121,12 @@ func lbHandler(ctx *types.Context) { } ctx.OK() case "DELETE": + if !ctx.Data.DB.LBExists(name) { + ctx.Err(types.ErrNotFound) + return + } err := ctx.Data.Alloc.FreeIP(context.Background(), name) - if err != nil { + if err != nil && err != cluster.ErrKeyNotFound { ctx.Log.Error(err, "error freeing IP of loadbalancer", "cluster", cl, "name", n) ctx.Err(types.ErrISE) return diff --git a/pkg/api/v1/server/routes.go b/pkg/api/v1/server/routes.go index 628281e..7298c38 100644 --- a/pkg/api/v1/server/routes.go +++ b/pkg/api/v1/server/routes.go @@ -24,7 +24,7 @@ var Routes = []types.Route{ []string{"GET"}, }, { - v1 + "/lb/{cluster:[a-zA-Z0-9-]+}/{name:[a-zA-Z0-9-]+$}", + v1 + "/lb/{cluster:[a-zA-Z0-9-]+}/{name:[a-zA-Z0-9-_]+$}", authHandler(lbHandler), []string{"GET", "POST", "DELETE"}, }, diff --git a/pkg/config/config.go b/pkg/config/config.go index d75cbf6..e488597 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -19,6 +19,7 @@ const ( EtcdLogLevel = "error" EtcdPrefix = "/haproxy-lb" + HAProxyService = "haproxy" HAProxyConfigFile = "/etc/haproxy/haproxy.cfg" ) @@ -44,6 +45,9 @@ func Validate(c *Config) error { if c.HAProxyConfig == "" { c.HAProxyConfig = HAProxyConfigFile } + if c.HAProxyService == "" { + c.HAProxyService = HAProxyService + } if len(c.VIP.VirtualIPs) == 0 { return errors.New("missing server.virtualIPs config") } @@ -134,12 +138,13 @@ func Validate(c *Config) error { } type Config struct { - LeaderHook string `yaml:"leaderHook,omitempty"` - FollowerHook string `yaml:"followerHook,omitempty"` - HAProxyConfig string `yaml:"haproxyConfig,omitempty"` - VIP VIP `yaml:"vip"` - Cluster Cluster `yaml:"cluster"` - Server Server `yaml:"server"` + LeaderHook string `yaml:"leaderHook,omitempty"` + FollowerHook string `yaml:"followerHook,omitempty"` + HAProxyConfig string `yaml:"haproxyConfig,omitempty"` + HAProxyService string `yaml:"haproxyService,omitempty"` + VIP VIP `yaml:"vip"` + Cluster Cluster `yaml:"cluster"` + Server Server `yaml:"server"` } type Server struct { diff --git a/pkg/controller/vip.go b/pkg/controller/loadbalancer.go similarity index 81% rename from pkg/controller/vip.go rename to pkg/controller/loadbalancer.go index a9c5429..b591833 100644 --- a/pkg/controller/vip.go +++ b/pkg/controller/loadbalancer.go @@ -19,7 +19,7 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call if err != nil { return } - ha, err := haproxy.NewHAProxyManager(cfg.HAProxyConfig) + ha, err := haproxy.NewHAProxyManager(cfg.HAProxyConfig, cfg.HAProxyService) if err != nil { return } @@ -27,7 +27,14 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call callbacks = cluster.Callbacks{ Leader: func(ctx context.Context, cc cluster.CallbackContext) { db := db.New(cc) - err := srv.UpdateDB(db) + cc.V(2).Info("starting haproxy", "service", cfg.HAProxyService) + err := ha.Start(ctx) + if err != nil { + cc.Error(err, "error starting haproxy") + cc.Fatal(err) + return + } + err = srv.UpdateDB(db) if err != nil { cc.Error(err, "error initialising api server as leader") cc.Fatal(err) @@ -44,15 +51,21 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call } for { cc.Info("leading", "id", cc.ID()) - ips, err := db.GetIPs() + ips, err := db.GetDeletedIPs() if err != nil && err != cluster.ErrPrefixNotFound { - cc.Error(err, "error reading ip list") + cc.Error(err, "error reading deleted ips list") + <-t.C + continue + } + deleteIPs(cc, n, ips) + + ips, err = db.GetIPs() + if err != nil && err != cluster.ErrPrefixNotFound { + cc.Error(err, "error reading ips list") <-t.C continue } addIPs(cc, n, ips) - // TODO - // deleted LB IPs need to be removed lbs, err := db.GetLBs("") if err != nil && err != cluster.ErrPrefixNotFound { @@ -76,6 +89,11 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call case <-ctx.Done(): t.Stop() deleteIPs(cc, n, ips) + cc.V(2).Info("stopping haproxy", "service", cfg.HAProxyService) + err = ha.Stop(context.Background()) + if err != nil { + cc.Error(err, "error stopping haproxy") + } cc.Info("leading canceled", "id", cc.ID()) return case <-t.C: @@ -101,14 +119,6 @@ func NewLBController(cfg *config.Config, srv *api.Server, log logr.Logger) (call } for { cc.Info("following", "id", cc.ID()) - //TODO - //ips, err := db.GetIPs() - //if err != nil { - // cc.Error("error getting config key", err) - // <-t.C - // continue - //} - //deleteIPs(cc, n, ips) select { case <-ctx.Done(): t.Stop() diff --git a/pkg/db/db.go b/pkg/db/db.go index 3655ea9..935c6bb 100644 --- a/pkg/db/db.go +++ b/pkg/db/db.go @@ -14,6 +14,7 @@ func New(kv cluster.KV) *DB { const ( ipPrefix = "/ip/" + delPrefix = "/del/" namePrefix = "/name/" lbPrefix = "/lb/" ) @@ -28,30 +29,45 @@ func (db *DB) GetIPs() (ips []string, err error) { } return } -func (db *DB) GetIP(ip string) (string, error) { +func (db *DB) GetDeletedIPs() (ips []string, err error) { + m, err := db.kv.GetPrefix(delPrefix) + if err != nil { + return + } + for k := range m { + ips = append(ips, k) + db.kv.Delete(delPrefix + k) + } + return +} +func (db *DB) GetName(ip string) (string, error) { v, err := db.kv.Get(ipPrefix + ip) return string(v), err } func (db *DB) IPExists(ip string) bool { - _, err := db.GetIP(ipPrefix + ip) + _, err := db.GetName(ip) return err == nil } -func (db *DB) SetIP(ip, name string) error { +func (db *DB) SetIPName(ip, name string) error { err := db.kv.Set(ipPrefix+ip, []byte(name)) if err != nil { return err } return db.kv.Set(namePrefix+name, []byte(ip)) } -func (db *DB) DeleteIP(ip, name string) error { +func (db *DB) DeleteIPName(ip, name string) error { err := db.kv.Delete(ipPrefix + ip) if err != nil { return err } - return db.kv.Delete(namePrefix + name) + err = db.kv.Delete(namePrefix + name) + if err != nil { + return err + } + return db.kv.Set(delPrefix+ip, nil) } -func (db *DB) GetName(name string) (string, error) { +func (db *DB) GetIP(name string) (string, error) { v, err := db.kv.Get(namePrefix + name) return string(v), err } @@ -77,6 +93,10 @@ func (db *DB) GetLBs(cl string) (lbs map[string][]byte, err error) { func (db *DB) GetLB(name string) ([]byte, error) { return db.kv.Get(lbPrefix + name) } +func (db *DB) LBExists(name string) bool { + _, err := db.GetLB(name) + return err == nil +} func (db *DB) SetLB(name string, lb []byte) error { return db.kv.Set(lbPrefix+name, lb) } diff --git a/pkg/haproxy/haproxy.go b/pkg/haproxy/haproxy.go index 90f5073..b299cc5 100644 --- a/pkg/haproxy/haproxy.go +++ b/pkg/haproxy/haproxy.go @@ -15,27 +15,31 @@ import ( type ServiceManager interface { Reload(context.Context) error + Status(context.Context) error + Start(context.Context) error + Stop(context.Context) error } type HAProxyManager struct { sync.Mutex - configFile string - serviceManager ServiceManager - template *template.Template - hash []byte + configFile string + svc ServiceManager + template *template.Template + hash []byte } -func NewHAProxyManager(configFile string) (*HAProxyManager, error) { +func NewHAProxyManager(configFile, service string) (ha *HAProxyManager, err error) { t, err := template.New("haproxy").Parse(haproxyConfigTemplate) if err != nil { - return nil, err + return } - return &HAProxyManager{ - configFile: configFile, - serviceManager: NewSystemdManager("haproxy"), - template: t, - }, nil + ha = &HAProxyManager{ + configFile: configFile, + svc: NewSystemdManager(service), + template: t, + } + return } func (ha *HAProxyManager) checkConfig(ctx context.Context, lbs Config) (hash []byte, err error) { @@ -59,9 +63,7 @@ func (ha *HAProxyManager) checkConfig(ctx context.Context, lbs Config) (hash []b return hash, err } -func (ha *HAProxyManager) UpdateConfig(ctx context.Context, lbs Config) error { - ha.Lock() - defer ha.Unlock() +func (ha *HAProxyManager) updateConfig(ctx context.Context, lbs Config) error { hash, err := ha.checkConfig(ctx, lbs) if err != nil { return err @@ -78,7 +80,33 @@ func (ha *HAProxyManager) UpdateConfig(ctx context.Context, lbs Config) error { file.Close() return err } - file.Close() + err = file.Close() ha.hash = hash - return ha.serviceManager.Reload(ctx) + return err +} + +func (ha *HAProxyManager) UpdateConfig(ctx context.Context, lbs Config) error { + ha.Lock() + defer ha.Unlock() + err := ha.updateConfig(ctx, lbs) + if err != nil { + return err + } + return ha.svc.Reload(ctx) +} + +func (ha *HAProxyManager) Start(ctx context.Context) error { + err := ha.svc.Status(ctx) + if err != nil { + err = ha.updateConfig(ctx, Config{}) + if err != nil { + return err + } + err = ha.svc.Start(ctx) + } + return err +} + +func (ha *HAProxyManager) Stop(ctx context.Context) error { + return ha.svc.Stop(ctx) } diff --git a/pkg/haproxy/systemd.go b/pkg/haproxy/systemd.go index 617a52a..c4330bf 100644 --- a/pkg/haproxy/systemd.go +++ b/pkg/haproxy/systemd.go @@ -15,6 +15,18 @@ func NewSystemdManager(service string) *SystemdManager { return &SystemdManager{service} } +func (sm *SystemdManager) Status(ctx context.Context) error { + return exec.CommandContext(ctx, systemctl, "status", sm.service).Run() +} + +func (sm *SystemdManager) Start(ctx context.Context) error { + return exec.CommandContext(ctx, systemctl, "start", sm.service).Run() +} + +func (sm *SystemdManager) Stop(ctx context.Context) error { + return exec.CommandContext(ctx, systemctl, "stop", sm.service).Run() +} + func (sm *SystemdManager) Reload(ctx context.Context) error { return exec.CommandContext(ctx, systemctl, "reload", sm.service).Run() } diff --git a/run_etcd.sh b/run_etcd.sh index 021fe6e..78afcb8 100755 --- a/run_etcd.sh +++ b/run_etcd.sh @@ -1,4 +1,4 @@ #!/bin/sh cd data -../haproxy-lb -config etcd.yaml +../haproxy-lb -config etcd.yaml -v 2