Add stats to PointsWriter

CHANGELOG.md

@@ -14,6 +14,7 @@ With this release InfluxDB is moving to Go 1.5.
 - [#3975](https://github.com/influxdb/influxdb/pull/3975): Add shard copy service
 - [#3986](https://github.com/influxdb/influxdb/pull/3986): Support sorting by time desc
 - [#3930](https://github.com/influxdb/influxdb/pull/3930): Wire up TOP aggregate function - fixes [#1821](https://github.com/influxdb/influxdb/issues/1821)
+- [#4045](https://github.com/influxdb/influxdb/pull/4045): Instrument cluster-level points writer
 - [#3996](https://github.com/influxdb/influxdb/pull/3996): Add statistics to httpd package
 - [#4003](https://github.com/influxdb/influxdb/pull/4033): Add logrotate configuration.
 

cluster/points_writer.go

@@ -2,6 +2,7 @@ package cluster
 
 import (
 	"errors"
+	"expvar"
 	"fmt"
 	"log"
 	"os"
@@ -18,6 +19,19 @@ import (
 // be returned as successful
 type ConsistencyLevel int
 
+// The statistics generated by the "write" mdoule
+const (
+	statWriteReq            = "req"
+	statPointWriteReq       = "point_req"
+	statPointWriteReqLocal  = "point_req_local"
+	statPointWriteReqRemote = "point_req_remote"
+	statWriteOK             = "write_ok"
+	statWritePartial        = "write_partial"
+	statWriteTimeout        = "write_timeout"
+	statWriteErr            = "write_error"
+	statWritePointReqHH     = "point_req_hh"
+)
+
 const (
 	// ConsistencyLevelAny allows for hinted hand off, potentially no write happened yet
 	ConsistencyLevelAny ConsistencyLevel = iota
@@ -90,6 +104,8 @@ type PointsWriter struct {
 	HintedHandoff interface {
 		WriteShard(shardID, ownerID uint64, points []tsdb.Point) error
 	}
+
+	statMap *expvar.Map
 }
 
 // NewPointsWriter returns a new instance of PointsWriter for a node.
@@ -98,6 +114,7 @@ func NewPointsWriter() *PointsWriter {
 		closing:      make(chan struct{}),
 		WriteTimeout: DefaultWriteTimeout,
 		Logger:       log.New(os.Stderr, "[write] ", log.LstdFlags),
+		statMap:      influxdb.NewStatistics("write", "write", nil),
 	}
 }
 
@@ -182,6 +199,9 @@ func (w *PointsWriter) MapShards(wp *WritePointsRequest) (*ShardMapping, error)
 
 // WritePoints writes across multiple local and remote data nodes according the consistency level.
 func (w *PointsWriter) WritePoints(p *WritePointsRequest) error {
+	w.statMap.Add(statWriteReq, 1)
+	w.statMap.Add(statPointWriteReq, int64(len(p.Points)))
+
 	if p.RetentionPolicy == "" {
 		db, err := w.MetaStore.Database(p.Database)
 		if err != nil {
@@ -242,6 +262,8 @@ func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPo
 	for _, owner := range shard.Owners {
 		go func(shardID uint64, owner meta.ShardOwner, points []tsdb.Point) {
 			if w.MetaStore.NodeID() == owner.NodeID {
+				w.statMap.Add(statPointWriteReqLocal, int64(len(points)))
+
 				err := w.TSDBStore.WriteToShard(shardID, points)
 				// If we've written to shard that should exist on the current node, but the store has
 				// not actually created this shard, tell it to create it and retry the write
@@ -257,9 +279,11 @@ func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPo
 				return
 			}
 
+			w.statMap.Add(statPointWriteReqRemote, int64(len(points)))
 			err := w.ShardWriter.WriteShard(shardID, owner.NodeID, points)
 			if err != nil && tsdb.IsRetryable(err) {
 				// The remote write failed so queue it via hinted handoff
+				w.statMap.Add(statWritePointReqHH, int64(len(points)))
 				hherr := w.HintedHandoff.WriteShard(shardID, owner.NodeID, points)
 
 				// If the write consistency level is ANY, then a successful hinted handoff can
@@ -283,11 +307,13 @@ func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPo
 		case <-w.closing:
 			return ErrWriteFailed
 		case <-timeout:
+			w.statMap.Add(statWriteTimeout, 1)
 			// return timeout error to caller
 			return ErrTimeout
 		case result := <-ch:
 			// If the write returned an error, continue to the next response
 			if result.Err != nil {
+				w.statMap.Add(statWriteErr, 1)
 				w.Logger.Printf("write failed for shard %d on node %d: %v", shard.ID, result.Owner.NodeID, result.Err)
 
 				// Keep track of the first error we see to return back to the client
@@ -301,12 +327,14 @@ func (w *PointsWriter) writeToShard(shard *meta.ShardInfo, database, retentionPo
 
 			// We wrote the required consistency level
 			if wrote >= required {
+				w.statMap.Add(statWriteOK, 1)
 				return nil
 			}
 		}
 	}
 
 	if wrote > 0 {
+		w.statMap.Add(statWritePartial, 1)
 		return ErrPartialWrite
 	}