some work done

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 = &notFoundHandler{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))
 	}
 }
 

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)

pkg/api/types/types.go

@@ -1,4 +1,4 @@
-package api
+package types
 
 type Route struct {
 	Path    string

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"`
 }

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 {
+func authHandler(h types.CtxHandler) types.CtxHandler {
-	s *Server
+	return func(ctx *types.Context) {
-}
-
-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) {
 		// 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()
 }

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,

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
 }

pkg/controller/vip.go

@@ -5,17 +5,60 @@ import (
 	"os/exec"
 	"time"
 
-	"git.giftfish.de/ston1th/vipman/pkg/cluster"
+	"git.giftfish.de/ston1th/haproxy-kb/pkg/cluster"
-	"git.giftfish.de/ston1th/vipman/pkg/config"
+	"git.giftfish.de/ston1th/haproxy-kb/pkg/config"
-	"git.giftfish.de/ston1th/vipman/pkg/vip"
+	"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()

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

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

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 {
+func (ha *HAProxyManager) checkConfig(ctx context.Context, lbs Config) (hash []byte, err 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()
 	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 bytes.Equal(ha.hash, hash) {
-	if err != nil {
+		return nil
-		return err
 	}
 	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()
 }

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

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