Refactor: generalize metrics, integrating server and agents metrics A…

…PI. Sender and qed server share the same metrics server.
BBVA · Mar 19, 2019 · 84d7a7f · 84d7a7f
1 parent 984e60d
commit 84d7a7f
Show file tree

Hide file tree

Showing 14 changed files with 279 additions and 275 deletions.
diff --git a/cmd/agent_auditor.go b/cmd/agent_auditor.go
@@ -21,6 +21,7 @@ import (
 	"github.com/bbva/qed/gossip/auditor"
 	"github.com/bbva/qed/gossip/member"
 	"github.com/bbva/qed/log"
+	"github.com/bbva/qed/metrics"
 	"github.com/bbva/qed/util"
 )
 
@@ -57,8 +58,8 @@ func newAgentAuditorCommand(ctx *cmdContext, config gossip.Config, agentPreRun f
 			if err != nil {
 				log.Fatalf("Failed to start the QED monitor: %v", err)
 			}
-
-			agent, err := gossip.NewAgent(&config, []gossip.Processor{auditor})
+			metricsServer := metrics.NewServer(config.MetricsAddr)
+			agent, err := gossip.NewAgent(&config, []gossip.Processor{auditor}, metricsServer)
 			if err != nil {
 				log.Fatalf("Failed to start the QED auditor: %v", err)
 			}

diff --git a/cmd/agent_monitor.go b/cmd/agent_monitor.go
@@ -21,6 +21,7 @@ import (
 	"github.com/bbva/qed/gossip/member"
 	"github.com/bbva/qed/gossip/monitor"
 	"github.com/bbva/qed/log"
+	"github.com/bbva/qed/metrics"
 	"github.com/bbva/qed/util"
 )
 
@@ -57,8 +58,8 @@ func newAgentMonitorCommand(ctx *cmdContext, config gossip.Config, agentPreRun f
 			if err != nil {
 				log.Fatalf("Failed to start the QED monitor: %v", err)
 			}
-
-			agent, err := gossip.NewAgent(&config, []gossip.Processor{monitor})
+			metricsServer := metrics.NewServer(config.MetricsAddr)
+			agent, err := gossip.NewAgent(&config, []gossip.Processor{monitor}, metricsServer)
 			if err != nil {
 				log.Fatalf("Failed to start the QED monitor: %v", err)
 			}

diff --git a/cmd/agent_publisher.go b/cmd/agent_publisher.go
@@ -21,6 +21,7 @@ import (
 	"github.com/bbva/qed/gossip/member"
 	"github.com/bbva/qed/gossip/publisher"
 	"github.com/bbva/qed/log"
+	"github.com/bbva/qed/metrics"
 	"github.com/bbva/qed/util"
 )
 
@@ -56,8 +57,8 @@ func newAgentPublisherCommand(ctx *cmdContext, config gossip.Config, agentPreRun
 			if err != nil {
 				log.Fatalf("Failed to start the QED publisher: %v", err)
 			}
-
-			agent, err := gossip.NewAgent(&config, []gossip.Processor{publisher})
+			metricsServer := metrics.NewServer(config.MetricsAddr)
+			agent, err := gossip.NewAgent(&config, []gossip.Processor{publisher}, metricsServer)
 			if err != nil {
 				log.Fatalf("Failed to start the QED publisher: %v", err)
 			}

diff --git a/gossip/agent.go b/gossip/agent.go
@@ -24,6 +24,7 @@ import (
 	"github.com/bbva/qed/gossip/member"
 	"github.com/bbva/qed/hashing"
 	"github.com/bbva/qed/log"
+	"github.com/bbva/qed/metrics"
 	"github.com/bbva/qed/protocol"
 	"github.com/coocood/freecache"
 	"github.com/hashicorp/memberlist"
@@ -38,7 +39,7 @@ type Agent struct {
 	config *Config
 	Self   *member.Peer
 
-	metricsServer *metricsServer
+	metricsServer *metrics.Server
 
 	memberlist *memberlist.Memberlist
 	broadcasts *memberlist.TransmitLimitedQueue
@@ -55,11 +56,11 @@ type Agent struct {
 	quit chan bool
 }
 
-func NewAgent(conf *Config, p []Processor) (agent *Agent, err error) {
+func NewAgent(conf *Config, p []Processor, m *metrics.Server) (agent *Agent, err error) {
 	log.Infof("New agent %s\n", conf.NodeName)
 	agent = &Agent{
 		config:        conf,
-		metricsServer: newMetricsServer(conf.MetricsAddr),
+		metricsServer: m,
 		Topology:      NewTopology(),
 		processors:    p,
 		processed:     freecache.NewCache(1 << 20),
@@ -134,11 +135,11 @@ func (a *Agent) ChTimedSend(batch *protocol.BatchSnapshots, ch chan *protocol.Ba
 func (a *Agent) start() {
 
 	for _, p := range a.processors {
-		p.RegisterMetrics(a.metricsServer.registry)
+		p.RegisterMetrics(a.metricsServer)
 	}
 
 	go func() {
-		a.metricsServer.start()
+		a.metricsServer.Start()
 	}()
 
 	for {
@@ -261,7 +262,7 @@ func (a *Agent) Shutdown() error {
 	a.stateLock.Lock()
 	defer a.stateLock.Unlock()
 
-	a.metricsServer.shutdown()
+	a.metricsServer.Shutdown()
 
 	if a.Self.Status == member.Shutdown {
 		return nil

diff --git a/gossip/agent_test.go b/gossip/agent_test.go
@@ -23,14 +23,16 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/bbva/qed/gossip/member"
+	"github.com/bbva/qed/metrics"
 )
 
 func TestJoin(t *testing.T) {
 	conf := DefaultConfig()
 	conf.NodeName = "testNode"
 	conf.Role = member.Auditor
 	conf.BindAddr = "127.0.0.1:12345"
-	a, _ := NewAgent(conf, []Processor{FakeProcessor{}})
+	metricsServer := metrics.NewServer("127.0.0.2:23464")
+	a, _ := NewAgent(conf, []Processor{FakeProcessor{}}, metricsServer)
 
 	testCases := []struct {
 		agentState             member.Status
@@ -71,7 +73,8 @@ func TestLeave(t *testing.T) {
 	conf.NodeName = "testNode"
 	conf.Role = member.Auditor
 	conf.BindAddr = "127.0.0.1:12346"
-	a, _ := NewAgent(conf, []Processor{FakeProcessor{}})
+	metricsServer := metrics.NewServer("127.0.0.2:13445")
+	a, _ := NewAgent(conf, []Processor{FakeProcessor{}}, metricsServer)
 
 	testCases := []struct {
 		agentState  member.Status
@@ -119,7 +122,8 @@ func TestShutdown(t *testing.T) {
 	conf.NodeName = "testNode"
 	conf.Role = member.Auditor
 	conf.BindAddr = "127.0.0.1:12347"
-	a, _ := NewAgent(conf, []Processor{FakeProcessor{}})
+	metricsServer := metrics.NewServer("127.0.0.2:43512")
+	a, _ := NewAgent(conf, []Processor{FakeProcessor{}}, metricsServer)
 
 	testCases := []struct {
 		agentState  member.Status

diff --git a/gossip/auditor/auditor.go b/gossip/auditor/auditor.go
@@ -28,6 +28,7 @@ import (
 	"github.com/bbva/qed/client"
 	"github.com/bbva/qed/hashing"
 	"github.com/bbva/qed/log"
+	"github.com/bbva/qed/metrics"
 	"github.com/bbva/qed/protocol"
 	"github.com/pkg/errors"
 
@@ -123,7 +124,7 @@ func NewAuditor(conf Config) (*Auditor, error) {
 	return &auditor, nil
 }
 
-func (a Auditor) RegisterMetrics(r *prometheus.Registry) {
+func (a Auditor) RegisterMetrics(srv *metrics.Server) {
 	metrics := []prometheus.Collector{
 		QedAuditorInstancesCount,
 		QedAuditorBatchesProcessSeconds,
@@ -132,7 +133,7 @@ func (a Auditor) RegisterMetrics(r *prometheus.Registry) {
 	}
 
 	for _, m := range metrics {
-		r.Register(m)
+		srv.Register(m)
 	}
 }
 

diff --git a/gossip/metrics.go b/gossip/metrics.go
diff --git a/gossip/monitor/monitor.go b/gossip/monitor/monitor.go
@@ -28,6 +28,7 @@ import (
 	"github.com/bbva/qed/client"
 	"github.com/bbva/qed/hashing"
 	"github.com/bbva/qed/log"
+	"github.com/bbva/qed/metrics"
 	"github.com/bbva/qed/protocol"
 	"github.com/pkg/errors"
 
@@ -120,7 +121,7 @@ func NewMonitor(conf *Config) (*Monitor, error) {
 	return &monitor, nil
 }
 
-func (m Monitor) RegisterMetrics(r *prometheus.Registry) {
+func (m Monitor) RegisterMetrics(srv *metrics.Server) {
 	metrics := []prometheus.Collector{
 		QedMonitorInstancesCount,
 		QedMonitorBatchesReceivedTotal,
@@ -129,7 +130,7 @@ func (m Monitor) RegisterMetrics(r *prometheus.Registry) {
 	}
 
 	for _, m := range metrics {
-		r.Register(m)
+		srv.Register(m)
 	}
 }
 

diff --git a/gossip/processor.go b/gossip/processor.go
@@ -22,25 +22,25 @@ import (
 	"net/http"
 
 	"github.com/bbva/qed/log"
+	"github.com/bbva/qed/metrics"
 	"github.com/bbva/qed/protocol"
-	"github.com/prometheus/client_golang/prometheus"
 )
 
 type Processor interface {
 	Process(*protocol.BatchSnapshots)
-	RegisterMetrics(*prometheus.Registry)
+	RegisterMetrics(*metrics.Server)
 }
 
 type FakeProcessor struct {
 }
 
-func (d FakeProcessor) Process(b *protocol.BatchSnapshots)     {}
-func (d FakeProcessor) RegisterMetrics(r *prometheus.Registry) {}
+func (d FakeProcessor) Process(b *protocol.BatchSnapshots) {}
+func (d FakeProcessor) RegisterMetrics(m *metrics.Server)  {}
 
 type DummyProcessor struct {
 }
 
-func (d DummyProcessor) RegisterMetrics(r *prometheus.Registry) {}
+func (d DummyProcessor) RegisterMetrics(m *metrics.Server) {}
 
 func (d DummyProcessor) Process(b *protocol.BatchSnapshots) {
 	for i := 0; i < len(b.Snapshots); i++ {

diff --git a/gossip/publisher/publisher.go b/gossip/publisher/publisher.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/bbva/qed/log"
+	"github.com/bbva/qed/metrics"
 	"github.com/bbva/qed/protocol"
 	"github.com/coocood/freecache"
 
@@ -105,15 +106,15 @@ func NewPublisher(conf Config) (*Publisher, error) {
 	return &publisher, nil
 }
 
-func (p Publisher) RegisterMetrics(r *prometheus.Registry) {
+func (p Publisher) RegisterMetrics(srv *metrics.Server) {
 	metrics := []prometheus.Collector{
 		QedPublisherInstancesCount,
 		QedPublisherBatchesReceivedTotal,
 		QedPublisherBatchesProcessSeconds,
 	}
 
 	for _, m := range metrics {
-		r.Register(m)
+		srv.Register(m)
 	}
 }