Finished support for config dir and reload

cmd/disttrust.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"sync"
 	"syscall"
 
 	log "github.com/sirupsen/logrus"
@@ -13,7 +14,8 @@ import (
 	"github.com/tlmiller/disttrust/cmd/config"
 	"github.com/tlmiller/disttrust/conductor"
 	"github.com/tlmiller/disttrust/provider"
-	_ "github.com/tlmiller/disttrust/server"
+	"github.com/tlmiller/disttrust/server"
+	"github.com/tlmiller/disttrust/server/healthz"
 )
 
 var (
@@ -47,6 +49,9 @@ func preRun(cmd *cobra.Command, args []string) {
 
 func Run(cmd *cobra.Command, args []string) {
 	providers := provider.DefaultStore()
+	var apiServ *server.ApiServer
+	apiServSetup := sync.Once{}
+	healthApi := healthz.New()
 
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, os.Interrupt, syscall.SIGHUP)
@@ -57,6 +62,14 @@ func Run(cmd *cobra.Command, args []string) {
 			log.Fatalf("making config: %v", err)
 		}
 
+		apiServSetup.Do(func() {
+			if userConfig.Api.Address != "" {
+				apiServ = server.NewApiServer(userConfig.Api.Address)
+				healthApi.InstallHandler(apiServ.Mux)
+				go apiServ.Serve()
+			}
+		})
+
 		log.Debug("building provider store from config")
 		providers, err = config.ToProviderStore(userConfig.Providers, providers)
 		if err != nil {
@@ -70,17 +83,12 @@ func Run(cmd *cobra.Command, args []string) {
 		}
 
 		manager := conductor.NewConductor()
-		_ = manager.AddMembers(members...)
-
-		//var apiServ *server.ApiServer
-		//if userConfig.Api.Address != "" {
-		//	apiServ = server.NewApiServer(userConfig.Api.Address)
-		//	for _, s := range mstatuses {
-		//		apiServ.AddHealthzChecks(s)
-		//	}
-		//	go apiServ.Serve()
-		//}
-
+		mstatuses := manager.AddMembers(members...)
+		checks := make([]healthz.Checker, len(mstatuses))
+		for i, ms := range mstatuses {
+			checks[i] = ms
+		}
+		healthApi.SetChecks(checks...)
 		manager.Play()
 
 		sig := <-sigCh
@@ -93,4 +101,7 @@ func Run(cmd *cobra.Command, args []string) {
 			break
 		}
 	}
+	if apiServ != nil {
+		apiServ.Stop()
+	}
 }

conductor/conductor.go

@@ -11,15 +11,13 @@ type Conductor struct {
 	members   []*MemberStatus
 	memCount  int32
 	stopCh    chan struct{}
-	watchCh   chan *MemberStatus
 	waitGroup sync.WaitGroup
 }
 
 func NewConductor() *Conductor {
 	return &Conductor{
 		healthErr: nil,
 		stopCh:    make(chan struct{}),
-		watchCh:   make(chan *MemberStatus),
 		waitGroup: sync.WaitGroup{},
 	}
 }
@@ -60,7 +58,9 @@ func (c *Conductor) Play() *Conductor {
 				case err := <-gmstatus.member.DoneCh():
 					log.Info("member stopped")
 					gmstatus.setState(false, err)
-					c.watchCh <- gmstatus
+					if _, err := gmstatus.State(); err != nil {
+						log.Errorf("member failed: %v", err)
+					}
 					c.waitGroup.Done()
 					break Outer
 				case <-c.stopCh:
@@ -69,25 +69,10 @@ func (c *Conductor) Play() *Conductor {
 			}
 		}()
 	}
-	go c.Watch()
 	return c
 }
 
 func (c *Conductor) Stop() {
 	close(c.stopCh)
 	c.waitGroup.Wait()
 }
-
-func (c *Conductor) Watch() {
-	for {
-		select {
-		case mstatus := <-c.watchCh:
-			if _, err := mstatus.State(); err != nil {
-				log := logrus.WithFields(logrus.Fields{
-					"member": mstatus.member.Name(),
-				})
-				log.Errorf("member failed: %v", err)
-			}
-		}
-	}
-}

server/apiserver.go

@@ -2,21 +2,18 @@ package server
 
 import (
 	"net/http"
-
-	"github.com/tlmiller/disttrust/server/healthz"
 )
 
 type ApiServer struct {
-	healthzChecks []healthz.Checker
-	server        http.Server
-	stopCh        chan struct{}
+	Mux    *http.ServeMux
+	server http.Server
 }
 
 func NewApiServer(address string) *ApiServer {
 	return &ApiServer{
+		Mux: http.NewServeMux(),
 		server: http.Server{
 			Addr: address,
 		},
-		stopCh: make(chan struct{}),
 	}
 }

server/healthz/checker.go

@@ -4,3 +4,5 @@ type Checker interface {
 	Check() error
 	Name() string
 }
+
+type ChecksFetcher func() []Checker

server/healthz/handler.go

@@ -6,7 +6,7 @@ import (
 	"net/http"
 )
 
-func handleRootHealthz(checks ...Checker) http.HandlerFunc {
+func handleRootHealthz(fetcher ChecksFetcher) http.HandlerFunc {
 	return http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
 		if req.Method != http.MethodGet && req.Method != http.MethodHead {
 			res.WriteHeader(http.StatusMethodNotAllowed)
@@ -15,7 +15,7 @@ func handleRootHealthz(checks ...Checker) http.HandlerFunc {
 
 		failed := false
 		var verboseOut bytes.Buffer
-		for _, check := range checks {
+		for _, check := range fetcher() {
 			if err := check.Check(); err != nil {
 				fmt.Fprintf(&verboseOut, "[-]%v failed\n", check.Name())
 				failed = true
@@ -41,6 +41,6 @@ func handleRootHealthz(checks ...Checker) http.HandlerFunc {
 	})
 }
 
-func InstallHandler(mux *http.ServeMux, checks ...Checker) {
-	mux.Handle("/healthz", handleRootHealthz(checks...))
+func (h *Healthz) InstallHandler(mux *http.ServeMux) {
+	mux.Handle("/healthz", handleRootHealthz(h.Checks))
 }

server/healthz/healthz.go

@@ -0,0 +1,29 @@
+package healthz
+
+import (
+	"sync"
+)
+
+type Healthz struct {
+	checks []Checker
+	lock   sync.Mutex
+}
+
+func (h *Healthz) Checks() []Checker {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+	return h.checks
+}
+
+func New() *Healthz {
+	return &Healthz{
+		checks: []Checker{},
+		lock:   sync.Mutex{},
+	}
+}
+
+func (h *Healthz) SetChecks(checks ...Checker) {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+	h.checks = checks
+}

server/serve.go

@@ -1,14 +1,10 @@
 package server
 
-import (
-	"net/http"
-
-	"github.com/tlmiller/disttrust/server/healthz"
-)
-
 func (a *ApiServer) Serve() {
-	mux := http.NewServeMux()
-	healthz.InstallHandler(mux, a.healthzChecks...)
-	a.server.Handler = mux
+	a.server.Handler = a.Mux
 	a.server.ListenAndServe()
 }
+
+func (a *ApiServer) Stop() {
+	a.server.Close()
+}