Add stats reporting cron task and optional librato config

Author: rowanseymour

daemon.go

@@ -16,16 +16,19 @@ type Daemon struct {
 	wg       *sync.WaitGroup
 	quit     chan bool
 	indexers []indexers.Indexer
+
+	prevStats map[indexers.Indexer]indexers.Stats
 }
 
 // NewDaemon creates a new daemon to run the given indexers
 func NewDaemon(cfg *Config, db *sql.DB, ixs []indexers.Indexer) *Daemon {
 	return &Daemon{
-		cfg:      cfg,
-		db:       db,
-		wg:       &sync.WaitGroup{},
-		quit:     make(chan bool),
-		indexers: ixs,
+		cfg:       cfg,
+		db:        db,
+		wg:        &sync.WaitGroup{},
+		quit:      make(chan bool),
+		indexers:  ixs,
+		prevStats: make(map[indexers.Indexer]indexers.Stats, len(ixs)),
 	}
 }
 
@@ -40,14 +43,18 @@ func (d *Daemon) Start() {
 	for _, i := range d.indexers {
 		d.startIndexer(i, time.Second*5)
 	}
+
+	d.startStatsReporter(time.Minute)
 }
 
 func (d *Daemon) startIndexer(indexer indexers.Indexer, interval time.Duration) {
 	d.wg.Add(1) // add ourselves to the wait group
 
+	log := logrus.WithField("indexer", indexer.Name())
+
 	go func() {
 		defer func() {
-			logrus.WithField("indexer", indexer.Name()).Info("indexer exiting")
+			log.Info("indexer exiting")
 			d.wg.Done()
 		}()
 
@@ -58,13 +65,56 @@ func (d *Daemon) startIndexer(indexer indexers.Indexer, interval time.Duration)
 			case <-time.After(interval):
 				_, err := indexer.Index(d.db, d.cfg.Rebuild, d.cfg.Cleanup)
 				if err != nil {
-					logrus.WithField("index", d.cfg.Index).WithError(err).Error("error during indexing")
+					log.WithError(err).Error("error during indexing")
 				}
 			}
 		}
 	}()
 }
 
+func (d *Daemon) startStatsReporter(interval time.Duration) {
+	d.wg.Add(1) // add ourselves to the wait group
+
+	go func() {
+		defer func() {
+			logrus.Info("analytics exiting")
+			d.wg.Done()
+		}()
+
+		for {
+			select {
+			case <-d.quit:
+				return
+			case <-time.After(interval):
+				d.reportStats()
+			}
+		}
+	}()
+}
+
+func (d *Daemon) reportStats() {
+	metrics := make(map[string]float64, len(d.indexers)*2)
+
+	for _, ix := range d.indexers {
+		stats := ix.Stats()
+		prev := d.prevStats[ix]
+
+		metrics[ix.Name()+"_indexed"] = float64(stats.Indexed - prev.Indexed)
+		metrics[ix.Name()+"_deleted"] = float64(stats.Deleted - prev.Deleted)
+
+		d.prevStats[ix] = stats
+	}
+
+	log := logrus.NewEntry(logrus.StandardLogger())
+
+	for k, v := range metrics {
+		librato.Gauge("indexer."+k, v)
+		log = log.WithField(k, v)
+	}
+
+	log.Info("stats reported")
+}
+
 // Stop stops this daemon
 func (d *Daemon) Stop() {
 	logrus.Info("daemon stopping")

indexers/base.go

@@ -20,21 +20,24 @@ const indexCommand = `{ "index": { "_id": %d, "_type": "_doc", "version": %d, "v
 // deletes a document
 const deleteCommand = `{ "delete" : { "_id": %d, "_type": "_doc", "version": %d, "version_type": "external", "routing": %d} }`
 
+type Stats struct {
+	Indexed int64         // total number of documents indexed
+	Deleted int64         // total number of documents deleted
+	Elapsed time.Duration // total time spent actually indexing
+}
+
 // Indexer is base interface for indexers
 type Indexer interface {
 	Name() string
 	Index(db *sql.DB, rebuild, cleanup bool) (string, error)
-	Stats() (int64, int64, time.Duration)
+	Stats() Stats
 }
 
 type baseIndexer struct {
 	elasticURL string
 	name       string // e.g. contacts, used as the alias
 
-	// statistics
-	indexedTotal int64
-	deletedTotal int64
-	elapsedTotal time.Duration
+	stats Stats
 }
 
 func newBaseIndexer(elasticURL, name string) baseIndexer {
@@ -45,8 +48,8 @@ func (i *baseIndexer) Name() string {
 	return i.name
 }
 
-func (i *baseIndexer) Stats() (int64, int64, time.Duration) {
-	return i.indexedTotal, i.deletedTotal, i.elapsedTotal
+func (i *baseIndexer) Stats() Stats {
+	return i.stats
 }
 
 func (i *baseIndexer) log() *logrus.Entry {
@@ -55,9 +58,9 @@ func (i *baseIndexer) log() *logrus.Entry {
 
 // records a complete index and updates statistics
 func (i *baseIndexer) recordComplete(indexed, deleted int, elapsed time.Duration) {
-	i.indexedTotal += int64(indexed)
-	i.deletedTotal += int64(deleted)
-	i.elapsedTotal += elapsed
+	i.stats.Indexed += int64(indexed)
+	i.stats.Deleted += int64(deleted)
+	i.stats.Elapsed += elapsed
 
 	i.log().WithField("indexed", indexed).WithField("deleted", deleted).WithField("elapsed", elapsed).Info("completed indexing")
 }

indexers/base_test.go

@@ -78,7 +78,7 @@ func assertIndexesWithPrefix(t *testing.T, es *elastic.Client, prefix string, ex
 }
 
 func assertIndexerStats(t *testing.T, ix indexers.Indexer, expectedIndexed, expectedDeleted int64) {
-	actualIndexed, actualDeleted, _ := ix.Stats()
-	assert.Equal(t, expectedIndexed, actualIndexed, "indexed mismatch")
-	assert.Equal(t, expectedDeleted, actualDeleted, "deleted mismatch")
+	actual := ix.Stats()
+	assert.Equal(t, expectedIndexed, actual.Indexed, "indexed mismatch")
+	assert.Equal(t, expectedDeleted, actual.Deleted, "deleted mismatch")
 }

indexers/contacts.go

@@ -60,7 +60,7 @@ func (i *ContactIndexer) Index(db *sql.DB, rebuild, cleanup bool) (string, error
 		return "", errors.Wrap(err, "error finding last modified")
 	}
 
-	i.log().WithField("index", physicalIndex).WithField("last_modified", lastModified).Info("indexing newer than last modified")
+	i.log().WithField("index", physicalIndex).WithField("last_modified", lastModified).Debug("indexing newer than last modified")
 
 	// now index our docs
 	start := time.Now()
@@ -243,7 +243,7 @@ func (i *ContactIndexer) indexModified(db *sql.DB, index string, lastModified ti
 		elapsed := time.Since(start)
 		rate := float32(processedCount) / (float32(elapsed) / float32(time.Second))
 
-		i.log().WithField("index", index).WithFields(logrus.Fields{"rate": int(rate), "added": createdCount, "deleted": deletedCount, "elapsed": elapsed}).Info("indexed contact batch")
+		i.log().WithField("index", index).WithFields(logrus.Fields{"rate": int(rate), "added": createdCount, "deleted": deletedCount, "elapsed": elapsed}).Debug("indexed contact batch")
 	}
 
 	return createdCount, deletedCount, nil