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

migrated to external IP allocator

Browse source
This commit is contained in:
ston1th 2021-09-26 20:20:43 +02:00
commit e2449b425e
15 changed files with 478 additions and 194 deletions
Download patch file Download diff file Expand all files Collapse all files

113
pkg/alloc/alloc.go
View file

@ -1,113 +0,0 @@
package alloc
import (
"context"
"fmt"
"net"
"strings"
"sync"
"git.giftfish.de/ston1th/haproxy-lb/pkg/db"
"github.com/mikioh/ipaddr"
)
type Alloc struct {
sync.Mutex
db *db.DB
pool []*net.IPNet
gateway *net.IPNet
}
func NewAlloc(db *db.DB, cidrs []string, gateway string) (*Alloc, error) {
pool, err := parseRange(cidrs)
if err != nil {
return nil, err
}
_, gw, err := net.ParseCIDR(gateway)
if err != nil {
return nil, err
}
return &Alloc{db: db, pool: pool, gateway: gw}, nil
}
func (a *Alloc) UpdateDB(db *db.DB) {
a.Lock()
a.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 _, 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.SetIPName(cidr, name)
if err != nil {
return
}
addr = pos.IP.String()
return
}
}
}
err = fmt.Errorf("no more IPs available in pool")
return
}
func (a *Alloc) FreeIP(ctx context.Context, name string) (err error) {
a.Lock()
defer a.Unlock()
if name == "" {
return
}
ip, err := a.db.GetIP(name)
if err != nil {
return
}
return a.db.DeleteIPName(ip, name)
}
func parseRange(cidrs []string) (nets []*net.IPNet, err error) {
for _, cidr := range cidrs {
if !strings.Contains(cidr, "-") {
_, n, err := net.ParseCIDR(cidr)
if err != nil {
return nil, err
}
nets = append(nets, n)
continue
}
fs := strings.SplitN(cidr, "-", 2)
if len(fs) != 2 {
return nil, fmt.Errorf("invalid IP range %q", cidr)
}
start := net.ParseIP(strings.TrimSpace(fs[0]))
if start == nil {
return nil, fmt.Errorf("invalid IP range %q: invalid start IP %q", cidr, fs[0])
}
end := net.ParseIP(strings.TrimSpace(fs[1]))
if end == nil {
return nil, fmt.Errorf("invalid IP range %q: invalid end IP %q", cidr, fs[1])
}
for _, pfx := range ipaddr.Summarize(start, end) {
n := &net.IPNet{
IP: pfx.IP,
Mask: pfx.Mask,
}
nets = append(nets, n)
}
}
return
}

25
pkg/api/server.go
View file

@ -7,12 +7,13 @@ import (
"net/http"
"time"
"git.giftfish.de/ston1th/haproxy-lb/pkg/alloc"
"git.giftfish.de/ston1th/haproxy-lb/pkg/api/types"
serverv1 "git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/server"
"git.giftfish.de/ston1th/haproxy-lb/pkg/config"
"git.giftfish.de/ston1th/haproxy-lb/pkg/db"
"git.giftfish.de/ston1th/netalloc"
"github.com/go-logr/logr"
"github.com/gorilla/mux"
)
@ -23,9 +24,8 @@ type Server struct {
mux *mux.Router
log logr.Logger
Data *types.ContextData
cidrs []string
gw string
Data *types.ContextData
cidr string
init chan struct{}
stop chan struct{}
@ -46,11 +46,10 @@ func (nf *notFoundHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// NewHTTPServer returns a new HTTPServer
func NewServer(c *config.Config, log logr.Logger) (*Server, error) {
s := &Server{
mux: mux.NewRouter(),
log: log,
cidrs: c.VIP.VirtualIPs,
gw: c.VIP.Gateway,
Data: &types.ContextData{},
mux: mux.NewRouter(),
log: log,
cidr: c.VIP.Prefix,
Data: &types.ContextData{},
init: make(chan struct{}),
stop: make(chan struct{}),
@ -107,15 +106,15 @@ func NewServer(c *config.Config, log logr.Logger) (*Server, error) {
func (s *Server) UpdateDB(db *db.DB) error {
s.Data.DB = db
if s.Data.Alloc == nil {
alloc, err := alloc.NewAlloc(db, s.cidrs, s.gw)
if s.Data.Net == nil {
na, err := netalloc.NewGenericAlloc(db, s.cidr)
if err != nil {
return err
}
s.Data.Alloc = alloc
s.Data.Net = na
close(s.init)
} else {
s.Data.Alloc.UpdateDB(db)
s.Data.Net.SetStore(db)
}
return nil
}

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

@ -8,14 +8,15 @@ import (
"github.com/go-logr/logr"
"github.com/gorilla/mux"
"git.giftfish.de/ston1th/haproxy-lb/pkg/alloc"
"git.giftfish.de/ston1th/netalloc"
"git.giftfish.de/ston1th/haproxy-lb/pkg/db"
)
type ContextData struct {
Alloc *alloc.Alloc
DB *db.DB
Auth *Auth
Net netalloc.Allocator
DB *db.DB
Auth *Auth
}
func NewContext(w http.ResponseWriter, r *http.Request, data *ContextData, log logr.Logger) *Context {

13
pkg/api/v1/client/client.go
View file

@ -83,3 +83,16 @@ func (c *Client) DeleteLoadBalancer(ctx context.Context, cluster, name string) (
_, err = c.c.Do(req, nil)
return
}
func (c *Client) DeleteLoadBalancersWithCluster(ctx context.Context, cluster string) (err error) {
if cluster == "" {
return errors.New("cluster value can not be empty")
}
path := base + "/lb/" + cluster
req, err := c.c.NewRequest(ctx, "DELETE", path, nil)
if err != nil {
return
}
_, err = c.c.Do(req, nil)
return
}

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

@ -4,6 +4,7 @@ import (
"encoding/json"
"errors"
"fmt"
"sort"
)
/*
@ -16,8 +17,15 @@ const Version = "v1"
type LoadBalancer struct {
Name string `json:"name,omitempty"`
IP string `json:"ip,omitempty"`
CIDR string `json:"cidr,omitempty"`
Options Options `json:"options,omitempty"`
Ports []Port `json:"ports,omitempty"`
Ports Ports `json:"ports,omitempty"`
}
func NewLoadBalancerFromBytes(b []byte) (lb *LoadBalancer, err error) {
lb = new(LoadBalancer)
err = json.Unmarshal(b, lb)
return
}
type Options struct {
@ -29,16 +37,83 @@ type Options struct {
}
type Port struct {
Port int `json:"port,omitempty"`
Servers []Server `json:"servers,omitempty"`
Port int `json:"port,omitempty"`
Servers Servers `json:"servers,omitempty"`
}
type Ports []Port
func (p Ports) Len() int { return len(p) }
func (p Ports) Less(i, j int) bool { return p[i].Port < p[j].Port }
func (p Ports) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
type Server struct {
Name string `json:"name,omitempty"`
IP string `json:"ip,omitempty"`
Port int `json:"port,omitempty"`
}
type Servers []Server
func (s Servers) Len() int { return len(s) }
func (s Servers) Less(i, j int) bool { return s[i].Name < s[j].Name }
func (s Servers) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (o Servers) Equal(n Servers) bool {
if len(o) != len(n) {
return false
}
for i, v := range o {
if v.Name != n[i].Name ||
v.IP != n[i].IP ||
v.Port != n[i].Port {
return false
}
}
return true
}
func (o Servers) ContainsIP(ip string) bool {
return o.IndexOfIP(ip) != -1
}
func (o Servers) IndexOfIP(ip string) int {
for i, v := range o {
if v.IP == ip {
return i
}
}
return -1
}
func (o Servers) RemoveByIP(ip string) Servers {
index := o.IndexOfIP(ip)
if index == -1 {
return o
}
return append(o[:index], o[index+1:]...)
}
func (o Servers) ContainsName(name string) bool {
return o.IndexOfName(name) != -1
}
func (o Servers) IndexOfName(name string) int {
for i, v := range o {
if v.Name == name {
return i
}
}
return -1
}
func (o Servers) RemoveByName(name string) Servers {
index := o.IndexOfName(name)
if index == -1 {
return o
}
return append(o[:index], o[index+1:]...)
}
func (lb LoadBalancer) ValidateClient() error {
for i, p := range lb.Ports {
if p.Port < 0 || p.Port > 65535 {
@ -66,7 +141,18 @@ func (lb LoadBalancer) ValidateServer() error {
if lb.IP == "" {
return errors.New("LoadBalancer.IP can not be empty")
}
return lb.ValidateClient()
if lb.CIDR == "" {
return errors.New("LoadBalancer.CIDR can not be empty")
}
err := lb.ValidateClient()
if err != nil {
return err
}
sort.Sort(lb.Ports)
for _, p := range lb.Ports {
sort.Sort(p.Servers)
}
return nil
}
func (lb LoadBalancer) JSON() ([]byte, error) {

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

@ -3,7 +3,6 @@ package server
import (
//"golang.org/x/crypto/bcrypt"
//"net/http"
"context"
"encoding/json"
"net"
@ -80,13 +79,14 @@ func lbHandler(ctx *types.Context) {
return
}
lb.Name = lbname(cl, n)
cidr, err := ctx.Data.DB.GetIP(name)
cidr, err := ctx.Data.DB.GetCIDR(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
}
if cidr != "" {
lb.CIDR = cidr
lb.IP, err = getIP(cidr)
if err != nil {
ctx.Log.Error(err, "error parsing existing ip", "cluster", cl, "name", n, "cidr", cidr)
@ -94,12 +94,14 @@ func lbHandler(ctx *types.Context) {
return
}
} else {
lb.IP, err = ctx.Data.Alloc.AllocIP(context.Background(), name)
ip, err := ctx.Data.Net.Alloc()
if err != nil {
ctx.Log.Error(err, "error allocating ip for loadbalancer", "cluster", cl, "name", n)
ctx.Err(types.ErrISE)
return
}
lb.IP = ip.IP.String()
lb.CIDR = ctx.Data.Net.CIDR(ip)
}
err = lb.ValidateServer()
if err != nil {
@ -125,7 +127,13 @@ func lbHandler(ctx *types.Context) {
ctx.Err(types.ErrNotFound)
return
}
err := ctx.Data.Alloc.FreeIP(context.Background(), name)
cidr, err := ctx.Data.DB.GetCIDR(name)
if err != nil && err != cluster.ErrKeyNotFound {
ctx.Log.Error(err, "error reading IP of loadbalancer", "cluster", cl, "name", n)
ctx.Err(types.ErrISE)
return
}
err = ctx.Data.Net.FreeCIDR(cidr)
if err != nil && err != cluster.ErrKeyNotFound {
ctx.Log.Error(err, "error freeing IP of loadbalancer", "cluster", cl, "name", n)
ctx.Err(types.ErrISE)
@ -149,20 +157,45 @@ func lbList(lbs map[string][]byte) (m map[string]json.RawMessage) {
return
}
func lbClusterListHandler(ctx *types.Context) {
func lbClusterHandler(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.Log.Error(err, "error reading loadbalancers", "cluster", cl)
ctx.Err(types.ErrISE)
return
}
ctx.JSON(lbList(m))
switch ctx.Method() {
case "GET":
ctx.JSON(lbList(m))
case "DELETE":
for name, _ := range m {
cidr, err := ctx.Data.DB.GetCIDR(name)
if err != nil && err != cluster.ErrKeyNotFound {
ctx.Log.Error(err, "error reading IP of loadbalancer", "cluster", cl, "name", name)
ctx.Err(types.ErrISE)
return
}
err = ctx.Data.Net.FreeCIDR(cidr)
if err != nil && err != cluster.ErrKeyNotFound {
ctx.Log.Error(err, "error freeing IP of loadbalancer", "cluster", cl, "name", name)
ctx.Err(types.ErrISE)
return
}
err = ctx.Data.DB.DeleteLB(name)
if err != nil {
ctx.Log.Error(err, "error deleting loadbalancer", "cluster", cl, "name", name)
ctx.Err(types.ErrISE)
return
}
}
ctx.OK()
}
}
func lbListHandler(ctx *types.Context) {
m, err := ctx.Data.DB.GetLBs("")
if err != nil {
ctx.Log.Error(err, "error reading loadbalancers")
ctx.Log.Error(err, "error reading loadbalancer clusters")
ctx.Err(types.ErrISE)
return
}

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

@ -20,8 +20,8 @@ var Routes = []types.Route{
},
{
v1 + "/lb/{cluster:[a-zA-Z0-9-]+$}",
authHandler(lbClusterListHandler),
[]string{"GET"},
authHandler(lbClusterHandler),
[]string{"GET", "DELETE"},
},
{
v1 + "/lb/{cluster:[a-zA-Z0-9-]+}/{name:[a-zA-Z0-9-_]+$}",

14
pkg/config/config.go
View file

@ -48,11 +48,8 @@ func Validate(c *Config) error {
if c.HAProxyService == "" {
c.HAProxyService = HAProxyService
}
if len(c.VIP.VirtualIPs) == 0 {
return errors.New("missing server.virtualIPs config")
}
if c.VIP.Gateway == "" {
return errors.New("missing server.gateway config")
if c.VIP.Prefix == "" {
return errors.New("missing server.prefix config")
}
if c.LeaderHook != "" {
if !filepath.IsAbs(c.LeaderHook) {
@ -154,10 +151,9 @@ type Server struct {
}
type VIP struct {
VirtualIPs []string `yaml:"virtualIPs"`
Gateway string `yaml:"gateway"`
Interface string `yaml:"interface,omitempty"`
Label string `yaml:"label,omitempty"`
Prefix string `yaml:"prefix"`
Interface string `yaml:"interface,omitempty"`
Label string `yaml:"label,omitempty"`
}
type BasicAuth struct {

70
pkg/db/db.go
View file

@ -1,15 +1,19 @@
package db
import (
"sync"
schemav1 "git.giftfish.de/ston1th/haproxy-lb/pkg/api/v1/schema"
"git.giftfish.de/ston1th/haproxy-lb/pkg/cluster"
)
type DB struct {
sync.Mutex
kv cluster.KV
}
func New(kv cluster.KV) *DB {
return &DB{kv}
return &DB{kv: kv}
}
const (
@ -20,12 +24,12 @@ const (
)
func (db *DB) GetIPs() (ips []string, err error) {
m, err := db.kv.GetPrefix(namePrefix)
m, err := db.kv.GetPrefix(ipPrefix)
if err != nil {
return
}
for _, v := range m {
ips = append(ips, string(v))
for k := range m {
ips = append(ips, k)
}
return
}
@ -40,36 +44,18 @@ func (db *DB) GetDeletedIPs() (ips []string, err error) {
}
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.GetName(ip)
return err == nil
}
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) DeleteIPName(ip, name string) error {
err := db.kv.Delete(ipPrefix + ip)
if err != nil {
return err
}
err = db.kv.Delete(namePrefix + name)
if err != nil {
return err
}
return db.kv.Set(delPrefix+ip, nil)
}
func (db *DB) GetIP(name string) (string, error) {
v, err := db.kv.Get(namePrefix + name)
return string(v), err
func (db *DB) GetCIDR(name string) (cidr string, err error) {
b, err := db.GetLB(name)
if err != nil {
return
}
lb, err := schemav1.NewLoadBalancerFromBytes(b)
if err != nil {
return
}
cidr = lb.CIDR
return
}
func (db *DB) GetLBs(cl string) (lbs map[string][]byte, err error) {
@ -103,3 +89,21 @@ func (db *DB) SetLB(name string, lb []byte) error {
func (db *DB) DeleteLB(name string) error {
return db.kv.Delete(lbPrefix + name)
}
func (db *DB) Exists(key string) (bool, error) {
_, err := db.kv.Get(ipPrefix + key)
if err == cluster.ErrKeyNotFound {
return false, nil
}
return err == nil, err
}
func (db *DB) Add(key string) error {
return db.kv.Set(ipPrefix+key, nil)
}
func (db *DB) Delete(key string) error {
err := db.kv.Delete(ipPrefix + key)
if err != nil {
return err
}
return db.kv.Set(delPrefix+key, nil)
}

1
pkg/haproxy/config.go
View file

@ -29,6 +29,7 @@ func NewConfig(m map[string][]byte) (c Config, err error) {
type LoadBalancer struct {
Name string `json:"name"`
IP string `json:"ip"`
CIDR string `json:"cidr"`
Options Options `json:"options"`
Ports []Port `json:"ports"`
}

23
pkg/haproxy/haproxy.go
View file

@ -10,9 +10,12 @@ import (
"os"
"os/exec"
"sync"
"syscall"
"text/template"
)
const Nobody = 65534
type ServiceManager interface {
Reload(context.Context) error
Status(context.Context) error
@ -21,6 +24,7 @@ type ServiceManager interface {
}
type HAProxyManager struct {
// protects config update
sync.Mutex
configFile string
@ -57,7 +61,24 @@ func (ha *HAProxyManager) checkConfig(ctx context.Context, lbs Config) (hash []b
return
}
hash = h.Sum(nil)
out, err := exec.CommandContext(ctx, "haproxy", "-c", "-f", tmp.Name()).CombinedOutput()
err = os.Chmod(tmp.Name(), 0640)
if err != nil {
return
}
err = os.Chown(tmp.Name(), -1, Nobody)
if err != nil {
return
}
cmd := exec.CommandContext(ctx, "haproxy", "-c", "-f", tmp.Name())
cmd.Dir = os.TempDir()
cmd.SysProcAttr = &syscall.SysProcAttr{
Credential: &syscall.Credential{
Uid: Nobody,
Gid: Nobody,
},
Setsid: true,
}
out, err := cmd.CombinedOutput()
if err != nil {
err = fmt.Errorf("%s:%w", string(out), err)
}

1
pkg/vip/arp.go
View file

@ -1,3 +1,4 @@
//go:build !linux
// +build !linux
package vip