ston1th/haproxy-lb
0
0
Fork 0
Code Issues Releases Activity

first working version

Browse source
This commit is contained in:
ston1th 2021-04-09 22:38:03 +02:00
commit 6da132106f
18 changed files with 519 additions and 96 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -1,2 +1,3 @@
data/
/haproxy-lb
/cmd/haproxy-lb-client

10
pkg/alloc/alloc.go
View file

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

200
pkg/api/client/client.go Normal file
View file

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

31
pkg/api/client/helper.go Normal file
View file

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

6
pkg/api/server.go
View file

@ -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 {

14
pkg/api/v1/server/auth.go → pkg/api/types/auth.go
View file

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

16
pkg/api/types/context.go
View file

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

15
pkg/api/types/error.go
View file

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

85
pkg/api/v1/client/client.go Normal file
View file

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

20
pkg/api/v1/schema/schema.go
View file

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

54
pkg/api/v1/server/handler.go
View file

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

2
pkg/api/v1/server/routes.go
View file

@ -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"},
},

17
pkg/config/config.go
View file

@ -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 {

38
pkg/controller/vip.go → pkg/controller/loadbalancer.go
View file

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

32
pkg/db/db.go
View file

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

60
pkg/haproxy/haproxy.go
View file

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

12
pkg/haproxy/systemd.go
View file

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

2
run_etcd.sh
View file

@ -1,4 +1,4 @@
#!/bin/sh
cd data
../haproxy-lb -config etcd.yaml
../haproxy-lb -config etcd.yaml -v 2