From c6ca86f2d98ca10fc5ee86d2fd52e32af0733a2f Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Thu, 20 Oct 2016 14:35:14 -0600 Subject: [PATCH] Add ExpvarStatsClient and basic tracking. A `StatsClient` for `expvar` is added so we can track Stats through the `/debug/vars` endpoint. Tags are nested inside maps so that we can see stats for db, frame & slice. Also added a `MultiStatsClient` for chaining multiple `StatsClient` implementations together (e.g. `expvar` and DataDog). --- cmd/pilosa/main.go | 1 + datadog/datadog.go | 73 +++++++---------- db.go | 18 ++++- fragment.go | 8 ++ frame.go | 18 ++++- index.go | 21 ++++- stats.go | 198 +++++++++++++++++++++++++++++++++++++++++---- 7 files changed, 274 insertions(+), 63 deletions(-) diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index a1a2fc8ac..1cff219c7 100644 --- a/cmd/pilosa/main.go +++ b/cmd/pilosa/main.go @@ -129,6 +129,7 @@ func (m *Main) Run(args ...string) error { // Configure index. fmt.Fprintf(m.Stderr, "Using data from: %s\n", m.Config.DataDir) m.Server.Index.Path = m.Config.DataDir + m.Server.Index.Stats = pilosa.NewExpvarStatsClient() // Build cluster from config file. m.Server.Host = m.Config.Host diff --git a/datadog/datadog.go b/datadog/datadog.go index 24e756c3f..56e1ca033 100644 --- a/datadog/datadog.go +++ b/datadog/datadog.go @@ -1,6 +1,9 @@ package datadog import ( + "io" + "io/ioutil" + "log" "sort" "time" @@ -23,6 +26,8 @@ var _ pilosa.StatsClient = &StatsClient{} type StatsClient struct { client *statsd.Client tags []string + + LogOutput io.Writer } // NewStatsClient returns a new instance of StatsClient. @@ -31,7 +36,11 @@ func NewStatsClient() (*StatsClient, error) { if err != nil { return nil, err } - return &StatsClient{client: c}, nil + + return &StatsClient{ + client: c, + LogOutput: ioutil.Discard, + }, nil } // Close closes the connection to the agent. @@ -48,66 +57,46 @@ func (c *StatsClient) Tags() []string { func (c *StatsClient) WithTags(tags ...string) pilosa.StatsClient { return &StatsClient{ client: c.client, - tags: unionTags(c.tags, tags), + tags: pilosa.UnionStringSlice(c.tags, tags), } } // Count tracks the number of times something occurs per second. -func (c *StatsClient) Count(name string, value int64) error { - return c.client.Count(name, value, c.tags, Rate) +func (c *StatsClient) Count(name string, value int64) { + if err := c.client.Count(name, value, c.tags, Rate); err != nil { + c.logger.Printf("datadog.StatsClient.Count error: %s", err) + } } // Gauge sets the value of a metric. -func (c *StatsClient) Gauge(name string, value float64) error { - return c.client.Gauge(name, value, c.tags, Rate) +func (c *StatsClient) Gauge(name string, value float64) { + if err := c.client.Gauge(name, value, c.tags, Rate); err != nil { + c.logger.Printf("datadog.StatsClient.Gauge error: %s", err) + } } // Histogram tracks statistical distribution of a metric. func (c *StatsClient) Histogram(name string, value float64) error { - return c.client.Histogram(name, value, c.tags, Rate) + if err := c.client.Histogram(name, value, c.tags, Rate); err != nil { + c.logger.Printf("datadog.StatsClient.Histogram error: %s", err) + } } // Set tracks number of unique elements. func (c *StatsClient) Set(name string, value string) error { - return c.client.Set(name, value, c.tags, Rate) + if err := c.client.Set(name, value, c.tags, Rate); err != nil { + c.logger.Printf("datadog.StatsClient.Set error: %s", err) + } } // Timing tracks timing information for a metric. func (c *StatsClient) Timing(name string, value time.Duration) error { - return c.client.Timing(name, value, c.tags, Rate) + if err := c.client.Timing(name, value, c.tags, Rate); err != nil { + c.logger.Printf("datadog.StatsClient.Timing error: %s", err) + } } -// unionTags returns a sorted set of tags which combine a & b. -func unionTags(a, b []string) []string { - // Sort both sets first. - sort.Strings(a) - sort.Strings(b) - - // Find size of largest slice. - n := len(a) - if len(b) > n { - n = len(b) - } - - // Exit if both sets are empty. - if n == 0 { - return nil - } - - // Iterate over both in order and merge. - other := make([]string, 0, n) - for len(a) > 0 || len(b) > 0 { - if len(a) == 0 { - other, b = append(other, b[0]), b[1:] - } else if len(b) == 0 { - other, a = append(other, a[0]), a[1:] - } else if a[0] < b[0] { - other, a = append(other, a[0]), a[1:] - } else if b[0] < a[0] { - other, b = append(other, b[0]), b[1:] - } else { - other, a, b = append(other, a[0]), a[1:], b[1:] - } - } - return other +// logger returns a logger that writes to LogOutput +func (c *StatsClient) logger() *log.Logger { + return log.New(c.LogOutput, "", log.LstdFlags) } diff --git a/db.go b/db.go index f4b2752d6..a28ec9cf7 100644 --- a/db.go +++ b/db.go @@ -20,6 +20,8 @@ type DB struct { // Profile attribute storage and cache profileAttrStore *AttrStore + + stats StatsClient } // NewDB returns a new instance of DB. @@ -30,6 +32,8 @@ func NewDB(path, name string) *DB { frames: make(map[string]*Frame), profileAttrStore: NewAttrStore(filepath.Join(path, "data")), + + stats: NopStatsClient, } } @@ -78,11 +82,13 @@ func (db *DB) openFrames() error { continue } - fr := NewFrame(db.FramePath(filepath.Base(fi.Name())), db.name, filepath.Base(fi.Name())) + fr := db.newFrame(db.FramePath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) if err := fr.Open(); err != nil { return fmt.Errorf("open frame: name=%s, err=%s", fr.Name(), err) } db.frames[fr.Name()] = fr + + db.stats.Count("frameN", 1) } return nil } @@ -164,15 +170,23 @@ func (db *DB) createFrameIfNotExists(name string) (*Frame, error) { } // Initialize and open frame. - f := NewFrame(db.FramePath(name), db.name, name) + f := db.newFrame(db.FramePath(name), name) if err := f.Open(); err != nil { return nil, err } db.frames[name] = f + db.stats.Count("frameN", 1) + return f, nil } +func (db *DB) newFrame(path, name string) *Frame { + f := NewFrame(path, db.name, name) + f.stats = db.stats.WithTags(fmt.Sprintf("frame:%s", name)) + return f +} + type dbSlice []*DB func (p dbSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } diff --git a/fragment.go b/fragment.go index e9d81e8fd..4fd982c58 100644 --- a/fragment.go +++ b/fragment.go @@ -84,6 +84,8 @@ type Fragment struct { // Bitmap attribute storage. // This is set by the parent frame unless overridden for testing. BitmapAttrStore *AttrStore + + stats StatsClient } // NewFragment returns a new instance of Fragment. @@ -96,6 +98,8 @@ func NewFragment(path, db, frame string, slice uint64) *Fragment { LogOutput: os.Stderr, MaxOpN: DefaultFragmentMaxOpN, + + stats: NopStatsClient, } } @@ -369,6 +373,8 @@ func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error) changed = true } + f.stats.Count("setN", 1) + return changed, nil } @@ -417,6 +423,8 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) { return true, nil } + f.stats.Count("clearN", 1) + return changed, nil } diff --git a/frame.go b/frame.go index bd16c172f..56e4b3f1a 100644 --- a/frame.go +++ b/frame.go @@ -28,6 +28,8 @@ type Frame struct { // Bitmap attribute storage and cache bitmapAttrStore *AttrStore + + stats StatsClient } // NewFrame returns a new instance of frame. @@ -39,6 +41,8 @@ func NewFrame(path, db, name string) *Frame { fragments: make(map[uint64]*Fragment), bitmapAttrStore: NewAttrStore(filepath.Join(path, "data")), + + stats: NopStatsClient, } } @@ -117,12 +121,14 @@ func (f *Frame) openFragments() error { continue } - frag := NewFragment(f.FragmentPath(slice), f.db, f.name, slice) + frag := f.newFragment(f.FragmentPath(slice), slice) if err := frag.Open(); err != nil { return fmt.Errorf("open fragment: slice=%s, err=%s", frag.Slice(), err) } frag.BitmapAttrStore = f.bitmapAttrStore f.fragments[frag.Slice()] = frag + + f.stats.Count("sliceN", 1) } return nil @@ -187,7 +193,7 @@ func (f *Frame) createFragmentIfNotExists(slice uint64) (*Fragment, error) { } // Initialize and open fragment. - frag := NewFragment(f.FragmentPath(slice), f.db, f.name, slice) + frag := f.newFragment(f.FragmentPath(slice), slice) if err := frag.Open(); err != nil { return nil, err } @@ -196,9 +202,17 @@ func (f *Frame) createFragmentIfNotExists(slice uint64) (*Fragment, error) { // Save to lookup. f.fragments[slice] = frag + f.stats.Count("sliceN", 1) + return frag, nil } +func (f *Frame) newFragment(path string, slice uint64) *Fragment { + frag := NewFragment(path, f.db, f.name, slice) + frag.stats = f.stats.WithTags(fmt.Sprintf("slice:%d", slice)) + return frag +} + type frameSlice []*Frame func (p frameSlice) Swap(i, j int) { p[i], p[j] = p[j], p[i] } diff --git a/index.go b/index.go index e104763ba..af57aca1e 100644 --- a/index.go +++ b/index.go @@ -27,6 +27,9 @@ type Index struct { wg sync.WaitGroup closing chan struct{} + // Stats + Stats StatsClient + // Data directory path. Path string @@ -43,6 +46,8 @@ func NewIndex() *Index { remoteMax: 0, closing: make(chan struct{}, 0), + Stats: NopStatsClient, + CacheFlushInterval: DefaultCacheFlushInterval, LogOutput: os.Stderr, @@ -74,11 +79,13 @@ func (i *Index) Open() error { i.logger().Printf("opening database: %s", filepath.Base(fi.Name())) - db := NewDB(i.DBPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) + db := i.newDB(i.DBPath(filepath.Base(fi.Name())), filepath.Base(fi.Name())) if err := db.Open(); err != nil { return fmt.Errorf("open db: name=%s, err=%s", db.Name(), err) } i.dbs[db.Name()] = db + + i.Stats.Count("dbN", 1) } // Periodically flush cache. @@ -174,15 +181,23 @@ func (i *Index) createDBIfNotExists(name string) (*DB, error) { } // Otherwise create a new database. - db := NewDB(i.DBPath(name), name) + db := i.newDB(i.DBPath(name), name) if err := db.Open(); err != nil { return nil, err } i.dbs[db.Name()] = db + i.Stats.Count("dbN", 1) + return db, nil } +func (i *Index) newDB(path, name string) *DB { + db := NewDB(path, name) + db.stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name())) + return db +} + // DeleteDB removes a database from the index. func (i *Index) DeleteDB(name string) error { i.mu.Lock() @@ -207,6 +222,8 @@ func (i *Index) DeleteDB(name string) error { // Remove reference. delete(i.dbs, name) + i.Stats.Count("dbN", -1) + return nil } diff --git a/stats.go b/stats.go index 82cffa38c..1a6efe743 100644 --- a/stats.go +++ b/stats.go @@ -1,6 +1,19 @@ package pilosa -import "time" +import ( + "expvar" + "sort" + "strings" + "sync" + "time" +) + +func init() { + NopStatsClient = &nopStatsClient{} +} + +// Global expvar. +var Expvar = expvar.NewMap("index") // StatsClient represents a client to a stats server. type StatsClient interface { @@ -11,28 +24,183 @@ type StatsClient interface { WithTags(tags ...string) StatsClient // Tracks the number of times something occurs per second. - Count(name string, value int64) error + Count(name string, value int64) // Sets the value of a metric. - Gauge(name string, value float64) error + Gauge(name string, value float64) // Tracks statistical distribution of a metric. - Histogram(name string, value float64) error + Histogram(name string, value float64) // Tracks number of unique elements. - Set(name string, value string) error + Set(name string, value string) // Tracks timing information for a metric. - Timing(name string, value time.Duration) error + Timing(name string, value time.Duration) } -// NopStatsClient represents a client that doesn't do anything. -type NopStatsClient struct{} +var NopStatsClient StatsClient -func (c *NopStatsClient) Tags() []string { return nil } -func (c *NopStatsClient) WithTags(tags ...string) StatsClient { return c } -func (c *NopStatsClient) Count(name string, value int64) error { return nil } -func (c *NopStatsClient) Gauge(name string, value float64) error { return nil } -func (c *NopStatsClient) Histogram(name string, value float64) error { return nil } -func (c *NopStatsClient) Set(name string, value string) error { return nil } -func (c *NopStatsClient) Timing(name string, value time.Duration) error { return nil } +// nopStatsClient represents a client that doesn't do anything. +type nopStatsClient struct{} + +func (c *nopStatsClient) Tags() []string { return nil } +func (c *nopStatsClient) WithTags(tags ...string) StatsClient { return c } +func (c *nopStatsClient) Count(name string, value int64) {} +func (c *nopStatsClient) Gauge(name string, value float64) {} +func (c *nopStatsClient) Histogram(name string, value float64) {} +func (c *nopStatsClient) Set(name string, value string) {} +func (c *nopStatsClient) Timing(name string, value time.Duration) {} + +// ExpvarStatsClient writes stats out to expvars. +type ExpvarStatsClient struct { + mu sync.Mutex + m *expvar.Map + tags []string +} + +// NewExpvarStatsClient returns a new instance of ExpvarStatsClient. +// This client points at the root of the expvar index map. +func NewExpvarStatsClient() *ExpvarStatsClient { + return &ExpvarStatsClient{ + m: Expvar, + } +} + +// Tags returns a sorted list of tags on the client. +func (c *ExpvarStatsClient) Tags() []string { + return nil +} + +// WithTags returns a new client with additional tags appended. +func (c *ExpvarStatsClient) WithTags(tags ...string) StatsClient { + m := &expvar.Map{} + m.Init() + c.m.Set(strings.Join(tags, ","), m) + + return &ExpvarStatsClient{ + m: m, + tags: UnionStringSlice(c.tags, tags), + } +} + +// Count tracks the number of times something occurs. +func (c *ExpvarStatsClient) Count(name string, value int64) { + c.m.Add(name, value) +} + +// Gauge sets the value of a metric. +func (c *ExpvarStatsClient) Gauge(name string, value float64) { + var f expvar.Float + f.Set(value) + c.m.Set(name, &f) +} + +// Histogram tracks statistical distribution of a metric. +// This works the same as guage for this client. +func (c *ExpvarStatsClient) Histogram(name string, value float64) { + c.Gauge(name, value) +} + +// Set tracks number of unique elements. +func (c *ExpvarStatsClient) Set(name string, value string) { + c.m.Set(name, &expvar.String{}) +} + +// Timing tracks timing information for a metric. +func (c *ExpvarStatsClient) Timing(name string, value time.Duration) { + c.mu.Lock() + d, _ := c.m.Get(name).(time.Duration) + c.m.Set(name, d+value) + c.mu.Unlock() +} + +// MultiStatsClient joins multiple stats clients together. +type MultiStatsClient []StatsClient + +// Tags returns tags from the first client. +func (a MultiStatsClient) Tags() []string { + if len(a) > 0 { + return a[0].Tags() + } + return nil +} + +// WithTags returns a new set of clients with the additional tags. +func (a MultiStatsClient) WithTags(tags ...string) StatsClient { + other := make(MultiStatsClient, len(a)) + for i := range a { + other[i] = a[i].WithTags(tags...) + } + return other +} + +// Count tracks the number of times something occurs per second on all clients. +func (a MultiStatsClient) Count(name string, value int64) { + for _, c := range a { + c.Count(name, value) + } +} + +// Gauge sets the value of a metric on all clients. +func (a MultiStatsClient) Gauge(name string, value float64) { + for _, c := range a { + c.Gauge(name, value) + } +} + +// Histogram tracks statistical distribution of a metric on all clients. +func (a MultiStatsClient) Histogram(name string, value float64) { + for _, c := range a { + c.Histogram(name, value) + } +} + +// Set tracks number of unique elements on all clients. +func (a MultiStatsClient) Set(name string, value string) { + for _, c := range a { + c.Set(name, value) + } +} + +// Timing tracks timing information for a metric on all clients. +func (a MultiStatsClient) Timing(name string, value time.Duration) { + for _, c := range a { + c.Timing(name, value) + } +} + +// UnionStringSlice returns a sorted set of tags which combine a & b. +func UnionStringSlice(a, b []string) []string { + // Sort both sets first. + sort.Strings(a) + sort.Strings(b) + + // Find size of largest slice. + n := len(a) + if len(b) > n { + n = len(b) + } + + // Exit if both sets are empty. + if n == 0 { + return nil + } + + // Iterate over both in order and merge. + other := make([]string, 0, n) + for len(a) > 0 || len(b) > 0 { + if len(a) == 0 { + other, b = append(other, b[0]), b[1:] + } else if len(b) == 0 { + other, a = append(other, a[0]), a[1:] + } else if a[0] < b[0] { + other, a = append(other, a[0]), a[1:] + } else if b[0] < a[0] { + other, b = append(other, b[0]), b[1:] + } else { + other, a, b = append(other, a[0]), a[1:], b[1:] + } + } + return other +}