Merge pull request #73 from raintank/measure-points

Measure points

Author: Dieterbe

metric_tank/aggmetric.go

@@ -12,12 +12,6 @@ import (
 	"github.com/raintank/raintank-metric/metric_tank/consolidation"
 )
 
-var statsPeriod time.Duration
-
-func init() {
-	statsPeriod = time.Duration(1) * time.Second
-}
-
 // AggMetric takes in new values, updates the in-memory data and streams the points to aggregators
 // it uses a circular buffer of chunks
 // each chunk starts at their respective t0
@@ -82,30 +76,9 @@ func NewAggMetric(key string, chunkSpan, numChunks uint32, maxDirtyChunks uint32
 		m.aggregators = append(m.aggregators, NewAggregator(key, as.span, as.chunkSpan, as.numChunks, maxDirtyChunks))
 	}
 
-	// only collect per aggmetric stats when in debug mode.
-	// running a goroutine for each aggmetric in an environment with many hundreds of thousands of metrics
-	// will result in the CPU just context switching and not much else.
-	if *logLevel < 2 {
-		go m.stats()
-	}
-
 	return &m
 }
 
-func (a *AggMetric) stats() {
-	for range time.Tick(statsPeriod) {
-		sum := 0
-		a.RLock()
-		for _, chunk := range a.Chunks {
-			if chunk != nil {
-				sum += int(chunk.NumPoints)
-			}
-		}
-		a.RUnlock()
-		pointsPerMetric.Value(int64(sum))
-	}
-}
-
 func (a *AggMetric) getChunk(pos int) *Chunk {
 	if pos < 0 {
 		return nil
@@ -411,6 +384,7 @@ func (a *AggMetric) Add(ts uint32, val float64) {
 			a.Chunks = append(a.Chunks, NewChunk(t0))
 		} else {
 			chunkClear.Inc(1)
+			totalPoints <- -1 * int(a.Chunks[a.CurrentChunkPos].NumPoints)
 			msg = fmt.Sprintf("cleared chunk at %d of %d and replaced with new", a.CurrentChunkPos, len(a.Chunks))
 			a.Chunks[a.CurrentChunkPos] = NewChunk(t0)
 		}

metric_tank/aggmetrics.go

@@ -18,6 +18,13 @@ type AggMetrics struct {
 	maxDirtyChunks uint32
 }
 
+var totalPoints chan int
+
+func init() {
+	// measurements can lag a bit, that's ok
+	totalPoints = make(chan int, 10)
+}
+
 func NewAggMetrics(chunkSpan, numChunks, chunkMaxStale, metricMaxStale uint32, maxDirtyChunks uint32, aggSettings []aggSetting) *AggMetrics {
 	ms := AggMetrics{
 		Metrics:        make(map[string]*AggMetric),
@@ -29,22 +36,10 @@ func NewAggMetrics(chunkSpan, numChunks, chunkMaxStale, metricMaxStale uint32, m
 		maxDirtyChunks: maxDirtyChunks,
 	}
 
-	go ms.stats()
 	go ms.GC()
 	return &ms
 }
 
-func (ms *AggMetrics) stats() {
-	pointsPerMetric.Value(0)
-
-	for range time.Tick(time.Duration(1) * time.Second) {
-		ms.RLock()
-		l := len(ms.Metrics)
-		ms.RUnlock()
-		metricsActive.Value(int64(l))
-	}
-}
-
 // periodically scan chunks and close any that have not received data in a while
 // TODO instrument occurences and duration of GC
 func (ms *AggMetrics) GC() {

metric_tank/chunk.go

@@ -36,5 +36,6 @@ func (c *Chunk) Push(t uint32, v float64) error {
 	c.NumPoints += 1
 	c.LastTs = t
 	c.LastWrite = uint32(time.Now().Unix())
+	totalPoints <- 1
 	return nil
 }

metric_tank/metric_tank.go

@@ -90,7 +90,7 @@ var cassandraPutDuration met.Timer
 var cassandraBlockDuration met.Timer
 var cassandraGetDuration met.Timer
 var inItems met.Meter
-var pointsPerMetric met.Meter
+var points met.Gauge
 var msgsHandleOK met.Count
 var msgsHandleFail met.Count
 var alloc met.Gauge
@@ -281,11 +281,26 @@ func initMetrics(stats met.Backend) {
 	cassandraBlockDuration = stats.NewTimer("cassandra_block_duration", 0)
 	cassandraPutDuration = stats.NewTimer("cassandra_put_duration", 0)
 	inItems = stats.NewMeter("in.items", 0)
-	pointsPerMetric = stats.NewMeter("points_per_metric", 0)
+	points = stats.NewGauge("total_points", 0)
 	msgsHandleOK = stats.NewCount("handle.ok")
 	msgsHandleFail = stats.NewCount("handle.fail")
 	alloc = stats.NewGauge("bytes_alloc.not_freed", 0)
 	totalAlloc = stats.NewGauge("bytes_alloc.incl_freed", 0)
 	sysBytes = stats.NewGauge("bytes_sys", 0)
 	metricsActive = stats.NewGauge("metrics_active", 0)
+
+	// run a collector for some global stats
+	go func() {
+		currentPoints := 0
+
+		ticker := time.Tick(time.Duration(1) * time.Second)
+		for {
+			select {
+			case <-ticker:
+				points.Value(int64(currentPoints))
+			case update := <-totalPoints:
+				currentPoints += update
+			}
+		}
+	}()
 }