Merge branch 'master' into coordinatePersist

Conflicts:
	metric_tank/aggmetric.go

Author: woodsaj

graphite-watcher/test_metric.go

@@ -11,7 +11,7 @@ import (
 	"sync"
 )
 
-func writeErrors(curTs int64, met stat, series graphite.Response, debug bool, errs *[]string) {
+func writeErrors(curTs int64, met stat, series *graphite.Response, debug bool, errs *[]string) {
 	if len(*errs) != 0 && debug {
 		f, err := os.Create(fmt.Sprintf("errors-%v-%d", met.def.Name, curTs))
 		if err != nil {
@@ -20,32 +20,32 @@ func writeErrors(curTs int64, met stat, series graphite.Response, debug bool, er
 		defer f.Close()
 		w := bufio.NewWriter(f)
 		for _, e := range *errs {
-			_, err = w.WriteString(e + "\n")
+			_, err = w.WriteString(fmt.Sprintf("ERROR %s\n", e))
 			if err != nil {
 				panic(err)
 			}
 		}
 		w.WriteString("graphite response:\n")
-		for _, serie := range series {
+		for _, serie := range *series {
 			w.WriteString(serie.Target)
 			for _, p := range serie.Datapoints {
-				w.WriteString(fmt.Sprintf("%s:%s\n", p[0], p[1]))
+				w.WriteString(fmt.Sprintf("%s:%s\n", p[1], p[0]))
 			}
 		}
 		w.Flush()
 	}
 	for _, e := range *errs {
-		fmt.Println(e)
+		fmt.Println(fmt.Sprintf("ERROR %d %s", curTs, e))
 	}
 }
 
 func test(wg *sync.WaitGroup, curTs int64, met stat, host string, debug bool) {
 	defer wg.Done()
 	var series graphite.Response
 	errs := make([]string, 0)
-	defer writeErrors(curTs, met, series, debug, &errs)
+	defer writeErrors(curTs, met, &series, debug, &errs)
 	e := func(str string, pieces ...interface{}) {
-		errs = append(errs, "ERROR: "+fmt.Sprintf(str, pieces...))
+		errs = append(errs, fmt.Sprintf(str, pieces...))
 	}
 
 	g := graphite.HostHeader{Host: "http://" + host + "/render", Header: http.Header{}}
@@ -54,32 +54,32 @@ func test(wg *sync.WaitGroup, curTs int64, met stat, host string, debug bool) {
 	q := graphite.Request{Targets: []string{met.def.Name}}
 	series, err := g.Query(&q)
 	if err != nil {
-		e("querying graphite: %q", err)
+		e("querying graphite: %v", err)
 		return
 	}
 	for _, serie := range series {
 		if met.def.Name != serie.Target {
-			e("name %q != target name%q", met.def.Name, serie.Target)
+			e("%v : bad target name %v", met.def.Name, serie.Target)
 		}
 
 		lastTs := int64(0)
 		oldestNull := int64(math.MaxInt64)
 		if len(serie.Datapoints) == 0 {
-			e("series for %q contains no points!", met.def.Name)
+			e("%v : series contains no points!", met.def.Name)
 		}
 		for _, p := range serie.Datapoints {
 			ts, err := p[1].Int64()
 			if err != nil {
-				e("could not parse timestamp %q", p)
+				e("%v : could not parse timestamp %q", met.def.Name, p)
 			}
 			if ts <= lastTs {
-				e("timestamp %v must be bigger than last %v", ts, lastTs)
+				e("%v timestamp %v must be bigger than last %v", met.def.Name, ts, lastTs)
 			}
 			if lastTs == 0 && (ts < curTs-24*3600-60 || ts > curTs-24*3600+60) {
-				e("first point %q should have been about 24h ago, i.e. around %d", p, curTs-24*3600)
+				e("%v first point %q should have been about 24h ago, i.e. around %d", met.def.Name, p, curTs-24*3600)
 			}
 			if lastTs != 0 && ts != lastTs+int64(met.def.Interval) {
-				e("point %v is not interval %v apart from previous point", p, met.def.Interval)
+				e("%v point %v is not interval %v apart from previous point", met.def.Name, p, met.def.Interval)
 			}
 			_, err = p[0].Float64()
 			if err != nil && ts > met.firstSeen {
@@ -88,7 +88,7 @@ func test(wg *sync.WaitGroup, curTs int64, met stat, host string, debug bool) {
 				}
 				if ts < curTs-30 {
 					nullPoints.Inc(1)
-					e("%v at %d : seeing a null for ts %v", met.def.Name, curTs, p[1])
+					e("%v : seeing a null for ts %v", met.def.Name, p[1])
 				}
 			} else {
 				// we saw a valid point, so reset oldestNull.
@@ -97,7 +97,7 @@ func test(wg *sync.WaitGroup, curTs int64, met stat, host string, debug bool) {
 			lastTs = ts
 		}
 		if lastTs < curTs-int64(met.def.Interval) || lastTs > curTs+int64(met.def.Interval) {
-			e("last point at %d is out of range. should have been around %d (now) +- %d", lastTs, curTs, met.def.Interval)
+			e("%v : last point at %d is out of range. should have been around %d (now) +- %d", met.def.Name, lastTs, curTs, met.def.Interval)
 		}
 		// if there was no null, we treat the point after the last one we had as null
 		if oldestNull == math.MaxInt64 {

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))
-	}
-}
-
 // Sync the saved state of a chunk by its T0.
 func (a *AggMetric) SyncChunkSaveState(ts uint32) {
 	a.RLock()
@@ -117,7 +90,7 @@ func (a *AggMetric) SyncChunkSaveState(ts uint32) {
 	}
 }
 
-/* Get a chunk by its T0.  It is expected that the caller has acquired of the a.Lock()*/
+/* Get a chunk by its T0.  It is expected that the caller has acquired a.Lock()*/
 func (a *AggMetric) getChunkByT0(ts uint32) *Chunk {
 	// we have no chunks.
 	if len(a.Chunks) == 0 {
@@ -509,6 +482,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

@@ -95,7 +95,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
@@ -290,11 +290,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
+			}
+		}
+	}()
 }