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{},
+                countersMtx:     sync.Mutex{},
                 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) {
@@ skipping 168 matching lines @@
                         aggFn:       aggFn,
                         measurement: measurement,
                         mtx:         sync.RWMutex{},
                         tags:        tags,
                         values:      []interface{}{},
                 }
                 c.aggMetrics[key] = m
         }
         return m
 }
+
+// getCounter creates or retrieves a Counter with the given name and tag set and

[stephana, 2016/02/24 14:48:16]

This should be in counter.go file, all the other GetX(...) functions are the
packages where their main logic is. 

[borenet, 2016/02/24 15:09:35]

Done.

+// returns it.
+func (c *Client) GetCounter(name string, tagsList ...map[string]string) *Counter {
+        c.countersMtx.Lock()
+        defer c.countersMtx.Unlock()
+
+        // Make a copy of the concatenation of all provided tags.
+        tags := util.AddParams(map[string]string{}, tagsList...)
+        tags["name"] = name
+        md5, err := util.MD5Params(tags)
+        if err != nil {
+                glog.Errorf("Failed to encode measurement tags: %s", err)
+        }
+        key := fmt.Sprintf("%s_%s", MEASUREMENT_COUNTER, md5)
+        m, ok := c.counters[key]
+        if !ok {
+                m = &Counter{
+                        m:   c.GetInt64Metric(MEASUREMENT_COUNTER, tags),
+                        mtx: sync.Mutex{},

[stephana, 2016/02/24 14:48:16]

nit: Mutex doesn't need to be initialized, it's unlocked by default. 

[borenet, 2016/02/24 15:09:35]

Done, here and elsewhere.

+                }
+                c.counters[key] = m
+        }
+        return m
+
+}