Make HA as configurable plugin by Blip users

ha/factory.go

@@ -0,0 +1,51 @@
+package ha
+
+import (
+	"fmt"
+	"github.com/cashapp/blip"
+	"sync"
+)
+
+type HAManagerFactory interface {
+	Make(monitor blip.ConfigMonitor) (Manager, error)
+}
+
+type defaultFactory struct {
+}
+
+var df = &defaultFactory{}
+
+func Register(f HAManagerFactory) error {
+	hf.Lock()
+	defer hf.Unlock()
+	// Check if it's different from the default factory
+	if hf.ha != df {
+		return fmt.Errorf("HA already registered")
+	}
+	hf.ha = f
+	blip.Debug("register HA")
+	return nil
+}
+
+func Make(args blip.ConfigMonitor) (Manager, error) {
+	hf.Lock()
+	defer hf.Unlock()
+	if hf.ha == nil {
+		return nil, fmt.Errorf("HA not registered")
+	}
+	return hf.ha.Make(args)
+}
+
+type haf struct {
+	*sync.Mutex
+	ha HAManagerFactory
+}
+
+var hf = &haf{
+	Mutex: &sync.Mutex{},
+	ha:    df,
+}
+
+func (f *defaultFactory) Make(_ blip.ConfigMonitor) (Manager, error) {
+	return Disabled, nil
+}

monitor/loader.go

@@ -7,6 +7,7 @@ import (
 	"crypto/sha256"
 	"errors"
 	"fmt"
+	"github.com/cashapp/blip/ha"
 	"os"
 	"sync"
 	"time"
@@ -463,11 +464,19 @@ func (ml *Loader) makeMonitor(cfg blip.ConfigMonitor) (*Monitor, error) {
 		sinks = append(sinks, sink)
 	}
 
+	// Configure the HA Manager for the monitor
+	var ham ha.Manager
+	ham, err := ha.Make(cfg)
+	if err != nil {
+		return nil, err
+	}
+
 	mon := NewMonitor(MonitorArgs{
 		Config:          cfg,
 		DbMaker:         ml.factory.DbConn,
 		PlanLoader:      ml.planLoader,
 		Sinks:           sinks,
+		HA:              ham,
 		TransformMetric: ml.plugin.TransformMetrics,
 	})
 	return mon, nil

monitor/monitor.go

@@ -73,6 +73,7 @@ type Monitor struct {
 
 	event event.MonitorReceiver
 	retry *backoff.ExponentialBackOff
+	ha    ha.Manager
 }
 
 // MonitorArgs are required arguments to NewMonitor.
@@ -82,6 +83,7 @@ type MonitorArgs struct {
 	PlanLoader      *plan.Loader
 	Sinks           []blip.Sink
 	TransformMetric func(metrics *blip.Metrics) error
+	HA              ha.Manager
 }
 
 // NewMonitor creates a new Monitor with the given arguments. The caller must
@@ -98,6 +100,7 @@ func NewMonitor(args MonitorArgs) *Monitor {
 		planLoader:      args.PlanLoader,
 		sinks:           args.Sinks,
 		transformMetric: args.TransformMetric,
+		ha:              args.HA,
 		// --
 		runMux: &sync.RWMutex{},
 		wg:     sync.WaitGroup{},
@@ -422,7 +425,7 @@ func (m *Monitor) startup() error {
 			Config:    m.cfg.Plans.Change,
 			DB:        m.db,
 			LCO:       m.lco,
-			HA:        ha.Disabled,
+			HA:        m.ha,
 		})
 
 		m.wg.Add(1)