From 4f07113f65f268cd355c22cb217ffcae32f04f2e Mon Sep 17 00:00:00 2001 From: Linh Vo Date: Thu, 20 Apr 2017 21:37:31 -0500 Subject: [PATCH] readd stastD report with new branch --- datadog/datadog.go | 8 +++ db.go | 30 ++++----- executor.go | 10 ++- fragment.go | 2 +- frame.go | 32 ++++----- index.go | 2 +- stats.go | 16 +++++ stats_test.go | 163 +++++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 228 insertions(+), 35 deletions(-) create mode 100644 stats_test.go diff --git a/datadog/datadog.go b/datadog/datadog.go index 27ef58644..d4cf1df65 100644 --- a/datadog/datadog.go +++ b/datadog/datadog.go @@ -70,6 +70,14 @@ func (c *StatsClient) Count(name string, value int64) { } } +// Count tracks the number of times something occurs per second with custom tags. +func (c *StatsClient) CountWithCustomTags(name string, value int64, t []string) { + tags := append(c.tags, t...) + if err := c.client.Count(name, value, 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) { if err := c.client.Gauge(name, value, c.tags, Rate); err != nil { diff --git a/db.go b/db.go index 8bf33300e..8c176bb79 100644 --- a/db.go +++ b/db.go @@ -22,31 +22,31 @@ const ( // DB represents a container for frames. type DB struct { - mu sync.Mutex - path string - name string + mu sync.Mutex + path string + name string // Default time quantum for all frames in database. // This can be overridden by individual frames. - timeQuantum TimeQuantum + timeQuantum TimeQuantum // Label used for referring to columns in database. - columnLabel string + columnLabel string // Frames by name. - frames map[string]*Frame + frames map[string]*Frame // Max Slice on any node in the cluster, according to this node remoteMaxSlice uint64 remoteMaxInverseSlice uint64 // Profile attribute storage and cache - profileAttrStore *AttrStore + profileAttrStore *AttrStore - broadcaster Broadcaster - stats StatsClient + broadcaster Broadcaster + Stats StatsClient - LogOutput io.Writer + LogOutput io.Writer } // NewDB returns a new instance of DB. @@ -68,7 +68,7 @@ func NewDB(path, name string) (*DB, error) { columnLabel: DefaultColumnLabel, - stats: NopStatsClient, + Stats: NopStatsClient, LogOutput: ioutil.Discard, }, nil } @@ -165,7 +165,7 @@ func (db *DB) openFrames() error { } db.frames[fr.Name()] = fr - db.stats.Count("frameN", 1) + db.Stats.Count("frameN", 1) } return nil } @@ -405,7 +405,7 @@ func (db *DB) createFrame(name string, opt FrameOptions) (*Frame, error) { // Add to database's frame lookup. db.frames[name] = f - db.stats.Count("frameN", 1) + db.Stats.Count("frameN", 1) return f, nil } @@ -416,7 +416,7 @@ func (db *DB) newFrame(path, name string) (*Frame, error) { return nil, err } f.LogOutput = db.LogOutput - f.stats = db.stats.WithTags(fmt.Sprintf("frame:%s", name)) + f.Stats = db.Stats.WithTags(fmt.Sprintf("frame:%s", name)) f.broadcaster = db.broadcaster return f, nil } @@ -445,7 +445,7 @@ func (db *DB) DeleteFrame(name string) error { // Remove reference. delete(db.frames, name) - db.stats.Count("frameN", -1) + db.Stats.Count("frameN", -1) return nil } diff --git a/executor.go b/executor.go index 9c156ec56..52e4c7402 100644 --- a/executor.go +++ b/executor.go @@ -93,12 +93,13 @@ func (e *Executor) executeCall(ctx context.Context, db string, c *pql.Call, slic if err := e.validateCallArgs(c); err != nil { return nil, err } - + dbTag := fmt.Sprintf("db:%s", db) // Special handling for mutation and top-n calls. switch c.Name { case "ClearBit": return e.executeClearBit(ctx, db, c, opt) case "Count": + e.Index.Stats.CountWithCustomTags(c.Name, 1, []string{dbTag}) return e.executeCount(ctx, db, c, slices, opt) case "SetBit": return e.executeSetBit(ctx, db, c, opt) @@ -107,8 +108,10 @@ func (e *Executor) executeCall(ctx context.Context, db string, c *pql.Call, slic case "SetProfileAttrs": return nil, e.executeSetProfileAttrs(ctx, db, c, opt) case "TopN": + e.Index.Stats.CountWithCustomTags(c.Name, 1, []string{dbTag}) return e.executeTopN(ctx, db, c, slices, opt) default: + e.Index.Stats.CountWithCustomTags(c.Name, 1, []string{dbTag}) return e.executeBitmapCall(ctx, db, c, slices, opt) } } @@ -485,6 +488,7 @@ func (e *Executor) executeRangeSlice(ctx context.Context, db string, c *pql.Call } bm = bm.Union(f.Bitmap(rowID)) } + f.Stats.Count("range", 1) return bm, nil } @@ -770,6 +774,7 @@ func (e *Executor) executeSetBitmapAttrs(ctx context.Context, db string, c *pql. if err := frame.BitmapAttrStore().SetAttrs(rowID, attrs); err != nil { return err } + frame.Stats.Count("SetBitmapAttrs", 1) // Do not forward call if this is already being forwarded. if opt.Remote { @@ -855,6 +860,7 @@ func (e *Executor) executeBulkSetBitmapAttrs(ctx context.Context, db string, cal if err := frame.BitmapAttrStore().SetBulkAttrs(frameMap); err != nil { return nil, err } + frame.Stats.Count("SetBitmapAttrs", 1) } // Do not forward call if this is already being forwarded. @@ -914,7 +920,7 @@ func (e *Executor) executeSetProfileAttrs(ctx context.Context, db string, c *pql if err := d.ProfileAttrStore().SetAttrs(id, attrs); err != nil { return err } - + d.Stats.Count("SetProfileAttrs", 1) // Do not forward call if this is already being forwarded. if opt.Remote { return nil diff --git a/fragment.go b/fragment.go index 9bee4288e..7c6cff24d 100644 --- a/fragment.go +++ b/fragment.go @@ -938,7 +938,7 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error { if err != nil { return err } - + f.stats.Count("ImportBit", 1) // import optimization to avoid linear foreach calls // slight risk of concurrent cache counter being off but // no real danger diff --git a/frame.go b/frame.go index 684b2225d..63c06f7c1 100644 --- a/frame.go +++ b/frame.go @@ -27,29 +27,29 @@ const ( // Frame represents a container for views. type Frame struct { - mu sync.Mutex - path string - db string - name string - timeQuantum TimeQuantum + mu sync.Mutex + path string + db string + name string + timeQuantum TimeQuantum - views map[string]*View + views map[string]*View // Bitmap attribute storage and cache bitmapAttrStore *AttrStore - broadcaster Broadcaster - stats StatsClient + broadcaster Broadcaster + Stats StatsClient // Frame settings. - rowLabel string - cacheType string - inverseEnabled bool + rowLabel string + cacheType string + inverseEnabled bool // Cache size for ranked frames - cacheSize uint32 + cacheSize uint32 - LogOutput io.Writer + LogOutput io.Writer } // NewFrame returns a new instance of frame. @@ -67,7 +67,7 @@ func NewFrame(path, db, name string) (*Frame, error) { views: make(map[string]*View), bitmapAttrStore: NewAttrStore(filepath.Join(path, ".data")), - stats: NopStatsClient, + Stats: NopStatsClient, rowLabel: DefaultRowLabel, inverseEnabled: DefaultInverseEnabled, @@ -256,7 +256,7 @@ func (f *Frame) openViews() error { view.BitmapAttrStore = f.bitmapAttrStore f.views[view.Name()] = view - f.stats.Count("maxSlice", 1) + f.Stats.Count("maxSlice", 1) } return nil @@ -422,7 +422,7 @@ func (f *Frame) newView(path, name string) *View { view.cacheType = f.cacheType view.LogOutput = f.LogOutput view.BitmapAttrStore = f.bitmapAttrStore - view.stats = f.stats.WithTags(fmt.Sprintf("slice:%s", name)) + view.stats = f.Stats.WithTags(fmt.Sprintf("slice:%s", name)) return view } diff --git a/index.go b/index.go index eb8f4b204..965c25ff0 100644 --- a/index.go +++ b/index.go @@ -245,7 +245,7 @@ func (i *Index) newDB(path, name string) (*DB, error) { return nil, err } db.LogOutput = i.LogOutput - db.stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name())) + db.Stats = i.Stats.WithTags(fmt.Sprintf("db:%s", db.Name())) db.broadcaster = i.Broadcaster return db, nil } diff --git a/stats.go b/stats.go index 1a6efe743..07c11a2f3 100644 --- a/stats.go +++ b/stats.go @@ -26,6 +26,9 @@ type StatsClient interface { // Tracks the number of times something occurs per second. Count(name string, value int64) + // Tracks the number of times something occurs per second with custom tags + CountWithCustomTags(name string, value int64, tags []string) + // Sets the value of a metric. Gauge(name string, value float64) @@ -47,6 +50,7 @@ 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) CountWithCustomTags(name string, value int64, tags []string) {} func (c *nopStatsClient) Gauge(name string, value float64) {} func (c *nopStatsClient) Histogram(name string, value float64) {} func (c *nopStatsClient) Set(name string, value string) {} @@ -89,6 +93,10 @@ func (c *ExpvarStatsClient) Count(name string, value int64) { c.m.Add(name, value) } +func (c *ExpvarStatsClient) CountWithCustomTags(name string, value int64, tags []string) { + c.m.Add(name, value) +} + // Gauge sets the value of a metric. func (c *ExpvarStatsClient) Gauge(name string, value float64) { var f expvar.Float @@ -142,6 +150,14 @@ func (a MultiStatsClient) Count(name string, value int64) { } } + +func (a MultiStatsClient) CountWithCustomTags(name string, value int64, tags []string) { + for _, c := range a { + c.CountWithCustomTags(name, value, tags) + } +} + + // Gauge sets the value of a metric on all clients. func (a MultiStatsClient) Gauge(name string, value float64) { for _, c := range a { diff --git a/stats_test.go b/stats_test.go new file mode 100644 index 000000000..248b68ea0 --- /dev/null +++ b/stats_test.go @@ -0,0 +1,163 @@ +package pilosa_test + +import ( + "context" + "github.com/pilosa/pilosa" + "testing" + "time" +) + +func TestStatsCount_TopN(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 1).SetBit(0, SliceWidth+2) + + // Execute query. + called := false + e := NewExecutor(idx.Index, NewCluster(1)) + e.Index.Stats = &MockStats{ + mockCountWithTags: func(name string, value int64, tags []string) { + if name != "TopN" { + t.Errorf("Expected TopN, Results %s", name) + } + + if tags[0] != "db:d" { + t.Errorf("Expected db, Results %s", tags[0]) + } + + called = true + return + }, + } + if _, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=2)`), nil, nil); err != nil { + t.Fatal(err) + } + if !called { + t.Error("CountWithCustomTags name isn't called") + } +} + +func TestStatsCount_Bitmap(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 0) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(0, 1) + called := false + e := NewExecutor(idx.Index, NewCluster(1)) + e.Index.Stats = &MockStats{ + mockCountWithTags: func(name string, value int64, tags []string) { + if name != "Bitmap" { + t.Errorf("Expected Bitmap, Results %s", name) + } + + if tags[0] != "db:d" { + t.Errorf("Expected db, Results %s", tags[0]) + } + + called = true + return + }, + } + if _, err := e.Execute(context.Background(), "d", MustParse(`Bitmap(frame=f, id=0)`), nil, nil); err != nil { + t.Fatal(err) + } + if !called { + t.Error("CountWithCustomTags name isn't called") + } +} + +func TestStatsCount_SetBitmapAttrs(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 0) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 1) + + called := false + e := NewExecutor(idx.Index, NewCluster(1)) + frame := e.Index.Frame("d", "f") + if frame == nil { + t.Fatal("frame not found") + } + + frame.Stats = &MockStats{ + mockCount: func(name string, value int64) { + if name != "SetBitmapAttrs" { + t.Errorf("Expected SetBitmapAttrs, Results %s", name) + } + called = true + return + }, + } + if _, err := e.Execute(context.Background(), "d", MustParse(`SetBitmapAttrs(id=10, frame=f, foo="bar")`), nil, nil); err != nil { + t.Fatal(err) + } + if !called { + t.Error("Count isn't called") + } +} + +func TestStatsCount_SetProfileAttrs(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 0) + idx.MustCreateFragmentIfNotExists("d", "f", pilosa.ViewStandard, 0).SetBit(10, 1) + + called := false + e := NewExecutor(idx.Index, NewCluster(1)) + db := e.Index.DB("d") + if db == nil { + t.Fatal("db not found") + } + + db.Stats = &MockStats{ + mockCount: func(name string, value int64) { + if name != "SetProfileAttrs" { + t.Errorf("Expected SetProfilepAttrs, Results %s", name) + } + + called = true + return + }, + } + if _, err := e.Execute(context.Background(), "d", MustParse(`SetProfileAttrs(id=10, frame=f, foo="bar")`), nil, nil); err != nil { + t.Fatal(err) + } + if !called { + t.Error("Count isn't called") + } +} + +type MockStats struct { + mockCount func(name string, value int64) + mockCountWithTags func(name string, value int64, tags []string) +} + +func (s *MockStats) Count(name string, value int64) { + if s.mockCount != nil { + s.mockCount(name, value) + return + } + return +} + +func (s *MockStats) CountWithCustomTags(name string, value int64, tags []string) { + if s.mockCountWithTags != nil { + s.mockCountWithTags(name, value, tags) + return + } + return +} + +func (c *MockStats) Tags() []string { return nil } +func (c *MockStats) WithTags(tags ...string) pilosa.StatsClient { return c } +func (c *MockStats) Gauge(name string, value float64) {} +func (c *MockStats) Histogram(name string, value float64) {} +func (c *MockStats) Set(name string, value string) {} +func (c *MockStats) Timing(name string, value time.Duration) {} \ No newline at end of file