Merge pull request #46 from weaveworks/36-improve-logging

Fix up logging and use metrics instead.

Author: jml

cmd/prism/main.go

@@ -124,6 +124,7 @@ func main() {
 		}
 	case modeIngester:
 		registration, err := ring.RegisterIngester(consul, cfg.listenPort, cfg.numTokens)
+		prometheus.MustRegister(registration)
 		if err != nil {
 			// This only happens for errors in configuration & set-up, not for
 			// network errors.

distributor.go

@@ -45,7 +45,6 @@ type Distributor struct {
 	clients       map[string]*IngesterClient
 
 	queryDuration   *prometheus.HistogramVec
-	consulUpdates   prometheus.Counter
 	receivedSamples prometheus.Counter
 	sendDuration    *prometheus.HistogramVec
 }
@@ -80,11 +79,6 @@ func NewDistributor(cfg DistributorConfig) *Distributor {
 			Help:      "Time spent executing expression queries.",
 			Buckets:   []float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10, 20, 30},
 		}, []string{"method", "status_code"}),
-		consulUpdates: prometheus.NewCounter(prometheus.CounterOpts{
-			Namespace: "prometheus",
-			Name:      "distributor_consul_updates_total",
-			Help:      "The total number of received Consul updates.",
-		}),
 		receivedSamples: prometheus.NewCounter(prometheus.CounterOpts{
 			Namespace: "prometheus",
 			Name:      "distributor_received_samples_total",
@@ -256,7 +250,6 @@ func (*Distributor) NeedsThrottling(_ context.Context) bool {
 // Describe implements prometheus.Collector.
 func (d *Distributor) Describe(ch chan<- *prometheus.Desc) {
 	d.queryDuration.Describe(ch)
-	ch <- d.consulUpdates.Desc()
 	ch <- d.receivedSamples.Desc()
 	d.sendDuration.Describe(ch)
 	d.ring.Describe(ch)
@@ -266,7 +259,6 @@ func (d *Distributor) Describe(ch chan<- *prometheus.Desc) {
 // Collect implements prometheus.Collector.
 func (d *Distributor) Collect(ch chan<- prometheus.Metric) {
 	d.queryDuration.Collect(ch)
-	ch <- d.consulUpdates
 	ch <- d.receivedSamples
 	d.sendDuration.Collect(ch)
 	d.ring.Collect(ch)

ring/ingester_lifecycle.go

@@ -24,6 +24,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/log"
 )
 
@@ -42,6 +43,8 @@ type IngesterRegistration struct {
 	hostname string
 	quit     chan struct{}
 	wait     sync.WaitGroup
+
+	consulHeartbeats prometheus.Counter
 }
 
 // RegisterIngester registers an ingester with Consul.
@@ -65,6 +68,11 @@ func RegisterIngester(consulClient ConsulClient, listenPort, numTokens int) (*In
 		// the distributors know where to connect.
 		hostname: fmt.Sprintf("%s:%d", addr, listenPort),
 		quit:     make(chan struct{}),
+
+		consulHeartbeats: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "prism_ingester_consul_heartbeats_total",
+			Help: "The total number of heartbeats sent to consul.",
+		}),
 	}
 
 	r.wait.Add(1)
@@ -78,9 +86,10 @@ func (r *IngesterRegistration) Unregister() {
 	log.Info("Removing ingester from consul")
 
 	// closing r.quit triggers loop() to exit, which in turn will trigger
-	// the removal of out tokens.
+	// the removal of our tokens.
 	close(r.quit)
 	r.wait.Wait()
+	log.Infof("Ingester removed from consul")
 }
 
 func (r *IngesterRegistration) loop() {
@@ -147,7 +156,7 @@ func (r *IngesterRegistration) heartbeat(tokens []uint32) {
 	for {
 		select {
 		case <-ticker.C:
-			log.Infof("Heartbeating to consul...")
+			r.consulHeartbeats.Inc()
 			if err := r.consul.CAS(consulKey, descFactory, heartbeat); err != nil {
 				log.Errorf("Failed to write to consul, sleeping: %v", err)
 			}
@@ -224,3 +233,13 @@ func getFirstAddressOf(name string) (string, error) {
 
 	return "", fmt.Errorf("No address found for %s", name)
 }
+
+// Describe implements prometheus.Collector.
+func (r *IngesterRegistration) Describe(ch chan<- *prometheus.Desc) {
+	ch <- r.consulHeartbeats.Desc()
+}
+
+// Collect implements prometheus.Collector.
+func (r *IngesterRegistration) Collect(ch chan<- prometheus.Metric) {
+	ch <- r.consulHeartbeats
+}

ring/ring.go

@@ -34,12 +34,6 @@ func (x uint32s) Swap(i, j int)      { x[i], x[j] = x[j], x[i] }
 // ErrEmptyRing is the error returned when trying to get an element when nothing has been added to hash.
 var ErrEmptyRing = errors.New("empty circle")
 
-var ingestorOwnershipDesc = prometheus.NewDesc(
-	"prometheus_distributor_ingester_ownership_percent",
-	"The percent ownership of the ring by ingestor",
-	[]string{"ingester"}, nil,
-)
-
 // CoordinationStateClient is an interface to getting changes to the coordination
 // state.  Should allow us to swap out Consul for something else (mesh?) later.
 type CoordinationStateClient interface {
@@ -53,6 +47,10 @@ type Ring struct {
 
 	mtx      sync.RWMutex
 	ringDesc Desc
+
+	ingesterOwnershipDesc *prometheus.Desc
+	ingesterTotalDesc     *prometheus.Desc
+	tokensTotalDesc       *prometheus.Desc
 }
 
 // New creates a new Ring
@@ -61,6 +59,21 @@ func New(client CoordinationStateClient) *Ring {
 		client: client,
 		quit:   make(chan struct{}),
 		done:   make(chan struct{}),
+		ingesterOwnershipDesc: prometheus.NewDesc(
+			"prometheus_distributor_ingester_ownership_percent",
+			"The percent ownership of the ring by ingester",
+			[]string{"ingester"}, nil,
+		),
+		ingesterTotalDesc: prometheus.NewDesc(
+			"prometheus_distributor_ingesters_total",
+			"Number of ingesters in the ring",
+			nil, nil,
+		),
+		tokensTotalDesc: prometheus.NewDesc(
+			"prometheus_distributor_tokens_total",
+			"Number of tokens in the ring",
+			nil, nil,
+		),
 	}
 	go r.loop()
 	return r
@@ -81,9 +94,6 @@ func (r *Ring) loop() {
 		}
 
 		ringDesc := value.(*Desc)
-		log.Infof("Got update to ring - %d ingesters, %d tokens",
-			len(ringDesc.Ingesters), len(ringDesc.Tokens))
-
 		r.mtx.Lock()
 		defer r.mtx.Unlock()
 		r.ringDesc = *ringDesc
@@ -126,7 +136,9 @@ func (r *Ring) search(key uint32) int {
 
 // Describe implements prometheus.Collector.
 func (r *Ring) Describe(ch chan<- *prometheus.Desc) {
-	ch <- ingestorOwnershipDesc
+	ch <- r.ingesterOwnershipDesc
+	ch <- r.ingesterTotalDesc
+	ch <- r.tokensTotalDesc
 }
 
 // Collect implements prometheus.Collector.
@@ -147,10 +159,21 @@ func (r *Ring) Collect(ch chan<- prometheus.Metric) {
 
 	for id, totalOwned := range owned {
 		ch <- prometheus.MustNewConstMetric(
-			ingestorOwnershipDesc,
+			r.ingesterOwnershipDesc,
 			prometheus.GaugeValue,
 			float64(totalOwned)/float64(math.MaxUint32),
 			id,
 		)
 	}
+
+	ch <- prometheus.MustNewConstMetric(
+		r.ingesterTotalDesc,
+		prometheus.GaugeValue,
+		float64(len(r.ringDesc.Ingesters)),
+	)
+	ch <- prometheus.MustNewConstMetric(
+		r.tokensTotalDesc,
+		prometheus.GaugeValue,
+		float64(len(r.ringDesc.Tokens)),
+	)
 }