Metrics2: Store a registry of Counters

go/metrics2/metrics.go

 package metrics2
 
 /*
    Convenience utilities for working with InfluxDB.
 */
 
 import (
         "fmt"
         "os"
         "sync"
         "time"
 
         "github.com/skia-dev/glog"
         "go.skia.org/infra/go/influxdb"
         "go.skia.org/infra/go/util"
 )
 
 const (
         DEFAULT_REPORT_FREQUENCY = time.Minute
         PUSH_FREQUENCY           = time.Minute
 )
 
 var (
         DefaultClient *Client = &Client{
                 aggMetrics: map[string]*aggregateMetric{},
+                counters:   map[string]*Counter{},
                 metrics:    map[string]*rawMetric{},
         }
 )
 
 // Init() initializes the metrics package.
 func Init(appName string, influxClient *influxdb.Client) error {
         hostName, err := os.Hostname()
         if err != nil {
                 return fmt.Errorf("Unable to retrieve hostname: %s", err)
         }
         tags := map[string]string{
                 "app":  appName,
                 "host": hostName,
         }
         c, err := NewClient(influxClient, tags, DEFAULT_REPORT_FREQUENCY)
         if err != nil {
                 return err
         }
         // Some metrics may already be registered with DefaultClient. Copy them
         // over.
         c.aggMetrics = DefaultClient.aggMetrics
+        c.counters = DefaultClient.counters
         c.metrics = DefaultClient.metrics
 
         // Set the default client.
         DefaultClient = c
         return nil
 }
 
 // Client is a struct used for communicating with an InfluxDB instance.
 type Client struct {
-        aggMetrics      map[string]*aggregateMetric
-        aggMetricsMtx   sync.Mutex
-        influxClient    *influxdb.Client
-        defaultTags     map[string]string
-        metrics         map[string]*rawMetric
-        metricsMtx      sync.Mutex
+        aggMetrics    map[string]*aggregateMetric
+        aggMetricsMtx sync.Mutex
+
+        counters    map[string]*Counter
+        countersMtx sync.Mutex
+
+        influxClient *influxdb.Client
+        defaultTags  map[string]string
+
+        metrics    map[string]*rawMetric
+        metricsMtx sync.Mutex
+
         reportFrequency time.Duration
         values          *influxdb.BatchPoints
         valuesMtx       sync.Mutex
 }
 
 // NewClient returns a Client which uses the given influxdb.Client to push data.
 // defaultTags specifies a set of default tag keys and values which are applied
 // to all data points. reportFrequency specifies how often metrics should create
 // data points.
 func NewClient(influxClient *influxdb.Client, defaultTags map[string]string, reportFrequency time.Duration) (*Client, error) {
         values, err := influxClient.NewBatchPoints()
         if err != nil {
                 return nil, err
         }
         c := &Client{
                 aggMetrics:      map[string]*aggregateMetric{},
-                aggMetricsMtx:   sync.Mutex{},
+                counters:        map[string]*Counter{},
                 influxClient:    influxClient,
                 defaultTags:     defaultTags,
                 metrics:         map[string]*rawMetric{},
-                metricsMtx:      sync.Mutex{},
                 reportFrequency: reportFrequency,
                 values:          values,
-                valuesMtx:       sync.Mutex{},
         }
         go func() {
                 for _ = range time.Tick(PUSH_FREQUENCY) {
                         byMeasurement, err := c.pushData()
                         if err != nil {
                                 glog.Errorf("Failed to push data into InfluxDB: %s", err)
                         } else {
                                 total := int64(0)
                                 for k, v := range byMeasurement {
                                         c.GetInt64Metric("metrics.points-pushed.by-measurement", map[string]string{"measurement": k}).Update(v)
@@ skipping 126 matching lines @@
         tags := util.AddParams(map[string]string{}, tagsList...)
         md5, err := util.MD5Params(tags)
         if err != nil {
                 glog.Errorf("Failed to encode measurement tags: %s", err)
         }
         key := fmt.Sprintf("%s_%s", measurement, md5)
         m, ok := c.metrics[key]
         if !ok {
                 m = &rawMetric{
                         measurement: measurement,
-                        mtx:         sync.RWMutex{},
                         tags:        tags,
                         value:       initial,
                 }
                 c.metrics[key] = m
         }
         return m
 }
 
 // getAggregateMetric creates or retrieves an aggregateMetric with the given
 // measurement name and tag set and returns it.
 func (c *Client) getAggregateMetric(measurement string, tagsList []map[string]string, aggFn func([]interface{}) interface{}) *aggregateMetric {
         c.aggMetricsMtx.Lock()
         defer c.aggMetricsMtx.Unlock()
 
         // Make a copy of the concatenation of all provided tags.
         tags := util.AddParams(map[string]string{}, tagsList...)
         md5, err := util.MD5Params(tags)
         if err != nil {
                 glog.Errorf("Failed to encode measurement tags: %s", err)
         }
         key := fmt.Sprintf("%s_%s", measurement, md5)
         m, ok := c.aggMetrics[key]
         if !ok {
                 m = &aggregateMetric{
                         aggFn:       aggFn,
                         measurement: measurement,
-                        mtx:         sync.RWMutex{},
                         tags:        tags,
                         values:      []interface{}{},
                 }
                 c.aggMetrics[key] = m
         }
         return m
 }