From 29f006f50614235063774ca0f314e971d8f17255 Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Wed, 9 Apr 2014 18:44:00 -0500 Subject: [PATCH] added some statsd fun --- core/http.go | 5 ++++ core/service.go | 2 ++ index/fragment_container.go | 36 +++++++++++++++++++++------- index/storage_cass.go | 21 +++++++++++----- util/statd.go | 48 +++++++++++++++++++++++++++++++++++++ 5 files changed, 97 insertions(+), 15 deletions(-) create mode 100644 util/statd.go diff --git a/core/http.go b/core/http.go index 82176f79e..670909483 100644 --- a/core/http.go +++ b/core/http.go @@ -8,6 +8,7 @@ import ( "pilosa/config" "pilosa/db" "pilosa/index" + "pilosa/util" "reflect" "runtime" "strconv" @@ -255,7 +256,11 @@ func (self *WebService) HandleSetBit(w http.ResponseWriter, r *http.Request) { for bitmap_id := range bitmaps(frame, obj) { pql := fmt.Sprintf("set(%d, %s, %d, %d)", bitmap_id, frame, filter, profile_id) + start := time.Now() result, err := self.service.Executor.RunPQL(db, pql) + delta := time.Since(start) + util.SendTimer("executor_setbit", delta.Nanoseconds()) + if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return diff --git a/core/service.go b/core/service.go index d4ac526a4..a03b21433 100644 --- a/core/service.go +++ b/core/service.go @@ -10,6 +10,7 @@ import ( "pilosa/hold" "pilosa/index" "pilosa/interfaces" + "pilosa/util" "syscall" "github.com/coreos/go-etcd/etcd" @@ -117,6 +118,7 @@ func (service *Service) Run() { case <-sigterm: log.Println("SIGTERM! Cleaning up...") service.Index.Shutdown() + util.ShutdownStats() service.Stop() return } diff --git a/index/fragment_container.go b/index/fragment_container.go index d448b9ce6..66707602f 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -97,7 +97,9 @@ func (self *FragmentContainer) Intersect(frag_id SUUID, bh []BitmapHandle) (Bitm if fragment, found := self.GetFragment(frag_id); found { request := NewIntersect(bh) fragment.requestChan <- request - return request.Response().answer.(BitmapHandle), nil + result := request.Response() + SendTimer("fragmant_container_Intersect", result.exec_time.Nanoseconds()) + return result.answer.(BitmapHandle), nil } return 0, errors.New("Invalid Bitmap Handle") } @@ -106,7 +108,9 @@ func (self *FragmentContainer) Union(frag_id SUUID, bh []BitmapHandle) (BitmapHa if fragment, found := self.GetFragment(frag_id); found { request := NewUnion(bh) fragment.requestChan <- request - return request.Response().answer.(BitmapHandle), nil + result := request.Response() + SendTimer("fragmant_container_Union", result.exec_time.Nanoseconds()) + return result.answer.(BitmapHandle), nil } return 0, errors.New("Invalid Bitmap Handle") } @@ -115,7 +119,9 @@ func (self *FragmentContainer) Get(frag_id SUUID, bitmap_id uint64) (BitmapHandl if fragment, found := self.GetFragment(frag_id); found { request := NewGet(bitmap_id) fragment.requestChan <- request - return request.Response().answer.(BitmapHandle), nil + result := request.Response() + SendTimer("fragmant_container_Get", result.exec_time.Nanoseconds()) + return result.answer.(BitmapHandle), nil } return 0, errors.New("Invalid Bitmap Handle") } @@ -124,7 +130,9 @@ func (self *FragmentContainer) TopN(frag_id SUUID, bh BitmapHandle, n int, categ if fragment, found := self.GetFragment(frag_id); found { request := NewTopN(bh, n, categories) fragment.requestChan <- request - return request.Response().answer.([]Pair), nil + result := request.Response() + SendTimer("fragmant_container_TopN", result.exec_time.Nanoseconds()) + return result.answer.([]Pair), nil } return nil, nil } @@ -133,7 +141,9 @@ func (self *FragmentContainer) GetList(frag_id SUUID, bitmap_id []uint64) ([]Bit if fragment, found := self.GetFragment(frag_id); found { request := NewGetList(bitmap_id) fragment.requestChan <- request - return request.Response().answer.([]BitmapHandle), nil + result := request.Response() + SendTimer("fragmant_container_GetList", result.exec_time.Nanoseconds()) + return result.answer.([]BitmapHandle), nil } return nil, errors.New("Invalid Bitmap Handle") } @@ -142,7 +152,9 @@ func (self *FragmentContainer) Count(frag_id SUUID, bitmap BitmapHandle) (uint64 if fragment, found := self.GetFragment(frag_id); found { request := NewCount(bitmap) fragment.requestChan <- request - return request.Response().answer.(uint64), nil + result := request.Response() + SendTimer("fragmant_container_Count", result.exec_time.Nanoseconds()) + return result.answer.(uint64), nil } return 0, errors.New("Invalid Bitmap Handle") } @@ -151,7 +163,9 @@ func (self *FragmentContainer) GetBytes(frag_id SUUID, bh BitmapHandle) ([]byte, if fragment, found := self.GetFragment(frag_id); found { request := NewGetBytes(bh) fragment.requestChan <- request - return request.Response().answer.([]byte), nil + result := request.Response() + SendTimer("fragmant_container_GetBytes", result.exec_time.Nanoseconds()) + return result.answer.([]byte), nil } return nil, errors.New("Invalid Bitmap Handle") } @@ -160,7 +174,9 @@ func (self *FragmentContainer) FromBytes(frag_id SUUID, bytes []byte) (BitmapHan if fragment, found := self.GetFragment(frag_id); found { request := NewFromBytes(bytes) fragment.requestChan <- request - return request.Response().answer.(BitmapHandle), nil + result := request.Response() + SendTimer("fragmant_container_FromBytes", result.exec_time.Nanoseconds()) + return result.answer.(BitmapHandle), nil } return 0, errors.New("Invalid Bitmap Handle") } @@ -169,7 +185,9 @@ func (self *FragmentContainer) SetBit(frag_id SUUID, bitmap_id uint64, pos uint6 if fragment, found := self.GetFragment(frag_id); found { request := NewSetBit(bitmap_id, pos, category) fragment.requestChan <- request - return request.Response().answer.(bool), nil + result := request.Response() + SendTimer("fragmant_container_SetBit", result.exec_time.Nanoseconds()) + return result.answer.(bool), nil } return false, errors.New("Invalid Bitmap Handle") } diff --git a/index/storage_cass.go b/index/storage_cass.go index a22130bc7..912bef1c1 100644 --- a/index/storage_cass.go +++ b/index/storage_cass.go @@ -5,6 +5,9 @@ package index import ( "fmt" "log" + "pilosa/util" + + "time" "tux21b.org/v1/gocql" ) @@ -27,7 +30,8 @@ func NewCassStorage(host, keyspace string) Storage { //cluster := gocql.NewCluster("127.0.0.1") cluster := gocql.NewCluster(host) cluster.Keyspace = keyspace - cluster.Consistency = gocql.Quorum + //cluster.Consistency = gocql.Quorum + cluster.Consistency = gocql.One //cluster.ProtoVersion = 1 // cluster.CQLVersion = "3.0.0" session, err := cluster.CreateSession() @@ -47,7 +51,7 @@ func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slic last_key := int64(dumb) marker := int64(dumb) var id = int64(bitmap_id) - + start := time.Now() var ( chunk *Chunk chunk_key, block int64 @@ -75,6 +79,8 @@ func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slic last_key = chunk_key } + delta := time.Since(start) + util.SendTimer("cassandra_storage_Fetch", delta.Nanoseconds()) bitmap.SetCount(uint64(count)) return bitmap, uint64(filter) } @@ -101,11 +107,14 @@ func (c *CassandraStorage) Store(id int64, db string, frame string, slice int, f func (c *CassandraStorage) StoreBlock(id int64, db string, frame string, slice int, filter uint64, chunk int64, block_index int32, block int64) error { - if err := c.db.Query(` - INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?);`, id, db, frame, slice, int(filter), chunk, block_index, block).Exec(); err != nil { + start := time.Now() + var err error + if err = c.db.Query(` + INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?);`, id, db, frame, slice, int(filter), chunk, block_index, block).Consistency(gocql.One).Exec(); err != nil { log.Println(err) log.Println("INSERT ", id, chunk, block_index) - return err } - return nil + delta := time.Since(start) + util.SendTimer("cassandra_storage_StoreBlock", delta.Nanoseconds()) + return err } diff --git a/util/statd.go b/util/statd.go new file mode 100644 index 000000000..de402db39 --- /dev/null +++ b/util/statd.go @@ -0,0 +1,48 @@ +package util + +import ( + "log" + "pilosa/config" + + "github.com/cactus/go-statsd-client/statsd" +) + +type args struct { + stat string + delta int64 + rate float32 +} + +var ( + count chan args + end chan bool +) + +func init() { + count = make(chan args, 100) + end = make(chan bool) + stat_config := config.GetStringDefault("statsd_server", "127.0.0.1:8125") + stats, _ := statsd.New(stat_config, "") + go func() { + for { + select { + case ci := <-count: + stats.Timing(ci.stat, ci.delta, ci.rate) + + break + case <-end: + log.Println("DONE Stats") + return + + } + } + }() +} + +func SendTimer(stat string, delta int64) { + count <- args{stat, delta, .5} +} + +func ShutdownStats() { + end <- true +}