From 70111b560458d2ad01f568c2f33219b1aedffcde Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Fri, 3 Apr 2020 01:01:29 -0500 Subject: [PATCH] Define metrics names as constants --- api.go | 10 +++++----- cache.go | 8 ++++---- executor.go | 12 +++++++----- fragment.go | 26 +++++++++++++------------- holder.go | 12 ++++++------ http/handler.go | 2 +- index.go | 2 +- metrics.go | 48 ++++++++++++++++++++++++++++++++++++++++++++++++ server.go | 20 ++++++++++---------- 9 files changed, 95 insertions(+), 45 deletions(-) create mode 100644 metrics.go diff --git a/api.go b/api.go index 35366ac4c..7ced231ae 100644 --- a/api.go +++ b/api.go @@ -185,7 +185,7 @@ func (api *API) CreateIndex(ctx context.Context, indexName string, options Index if err != nil { return nil, errors.Wrap(err, "sending CreateIndex message") } - api.holder.Stats.Count("createIndex", 1, 1.0) + api.holder.Stats.Count(MetricCreateIndex, 1, 1.0) return index, nil } @@ -229,7 +229,7 @@ func (api *API) DeleteIndex(ctx context.Context, indexName string) error { api.server.logger.Printf("problem sending DeleteIndex message: %s", err) return errors.Wrap(err, "sending DeleteIndex message") } - api.holder.Stats.Count("deleteIndex", 1, 1.0) + api.holder.Stats.Count(MetricDeleteIndex, 1, 1.0) return nil } @@ -276,7 +276,7 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str api.server.logger.Printf("problem sending CreateField message: %s", err) return nil, errors.Wrap(err, "sending CreateField message") } - api.holder.Stats.CountWithCustomTags("createField", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)}) + api.holder.Stats.CountWithCustomTags(MetricCreateField, 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)}) return field, nil } @@ -487,7 +487,7 @@ func (api *API) DeleteField(ctx context.Context, indexName string, fieldName str api.server.logger.Printf("problem sending DeleteField message: %s", err) return errors.Wrap(err, "sending DeleteField message") } - api.holder.Stats.CountWithCustomTags("deleteField", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)}) + api.holder.Stats.CountWithCustomTags(MetricDeleteField, 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)}) return nil } @@ -519,7 +519,7 @@ func (api *API) DeleteAvailableShard(_ context.Context, indexName, fieldName str api.server.logger.Printf("problem sending DeleteAvailableShard message: %s", err) return errors.Wrap(err, "sending DeleteAvailableShard message") } - api.holder.Stats.CountWithCustomTags("deleteAvailableShard", 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)}) + api.holder.Stats.CountWithCustomTags(MetricDeleteAvailableShard, 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)}) return nil } diff --git a/cache.go b/cache.go index e48c9cf17..99a13640e 100644 --- a/cache.go +++ b/cache.go @@ -231,7 +231,7 @@ func (c *rankCache) Invalidate() { func (c *rankCache) Recalculate() { c.mu.Lock() defer c.mu.Unlock() - c.stats.Count("cache.recalculate", 1, 1.0) + c.stats.Count(MetricRecalculateCache, 1, 1.0) c.recalculate() } @@ -241,7 +241,7 @@ func (c *rankCache) invalidate() { if time.Since(c.updateTime).Seconds() < 10 { return } - c.stats.Count("cache.invalidate", 1, 1.0) + c.stats.Count(MetricInvalidateCache, 1, 1.0) c.recalculate() } @@ -259,7 +259,7 @@ func (c *rankCache) recalculate() { // Store the count of the item at the threshold index. c.rankings = rankings length := len(c.rankings) - c.stats.Gauge("RankCache", float64(length), 1.0) + c.stats.Gauge(MetricRankCacheLength, float64(length), 1.0) var removeItems []bitmapPair // cached, ordered list if length > int(c.maxEntries) { @@ -275,7 +275,7 @@ func (c *rankCache) recalculate() { // If size is larger than the threshold then trim it. if len(c.entries) > c.thresholdBuffer { - c.stats.Count("cache.threshold", 1, 1.0) + c.stats.Count(MetricCacheThresholdReached, 1, 1.0) for _, pair := range removeItems { delete(c.entries, pair.ID) } diff --git a/executor.go b/executor.go index b06edcf61..58446c81f 100644 --- a/executor.go +++ b/executor.go @@ -2229,7 +2229,7 @@ func (e *executor) executeRowShard(ctx context.Context, index string, c *pql.Cal return rows[0], nil } row := rows[0].Union(rows[1:]...) - f.Stats.Count("range", 1, 1.0) + f.Stats.Count(MetricRow, 1, 1.0) return row, nil } @@ -2283,6 +2283,7 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, index string, c return NewRow(), nil } + f.Stats.Count(MetricRowBSI, 1, 1.0) return frag.notNull() } else if cond.Op == pql.BETWEEN || cond.Op == pql.BTWN_LT_LT || @@ -2324,6 +2325,7 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, index string, c return frag.notNull() } + f.Stats.Count(MetricRowBSI, 1, 1.0) return frag.rangeBetween(bsig.BitDepth, baseValueMin, baseValueMax) } else { @@ -2360,7 +2362,7 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, index string, c return frag.notNull() } - f.Stats.Count("range:bsigroup", 1, 1.0) + f.Stats.Count(MetricRowBSI, 1, 1.0) return frag.rangeOp(cond.Op, bsig.BitDepth, baseValue) } } @@ -3103,7 +3105,7 @@ func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql. if err := field.RowAttrStore().SetAttrs(rowID, attrs); err != nil { return err } - field.Stats.Count("SetRowAttrs", 1, 1.0) + field.Stats.Count(MetricSetRowAttrs, 1, 1.0) // Do not forward call if this is already being forwarded. if opt.Remote { @@ -3197,7 +3199,7 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, cal if err := field.RowAttrStore().SetBulkAttrs(fieldMap); err != nil { return nil, err } - field.Stats.Count("SetRowAttrs", 1, 1.0) + field.Stats.Count(MetricSetRowAttrs, 1, 1.0) } // Do not forward call if this is already being forwarded. @@ -3251,7 +3253,7 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *p if err := idx.ColumnAttrStore().SetAttrs(col, attrs); err != nil { return err } - idx.Stats.Count("SetProfileAttrs", 1, 1.0) + idx.Stats.Count(MetricSetProfileAttrs, 1, 1.0) // Do not forward call if this is already being forwarded. if opt.Remote { return nil diff --git a/fragment.go b/fragment.go index f93fc9514..57d4dd820 100644 --- a/fragment.go +++ b/fragment.go @@ -208,7 +208,7 @@ func (f *fragment) Open() error { // Read last bit to determine max row. f.maxRowID = f.storage.Max() / ShardWidth - f.stats.Gauge("rows", float64(f.maxRowID), 1.0) + f.stats.Gauge(MetricMaximumRow, float64(f.maxRowID), 1.0) return nil }(); err != nil { f.close() @@ -576,12 +576,12 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err // a new copy if no one's reading it. f.rowCache.Add(rowID, nil) - f.stats.Count("setBit", 1, 0.001) + f.stats.Count(MetricSetBit, 1, 0.001) // Update row count if they have increased. if rowID > f.maxRowID { f.maxRowID = rowID - f.stats.Gauge("rows", float64(f.maxRowID), 1.0) + f.stats.Gauge(MetricMaximumRow, float64(f.maxRowID), 1.0) } return changed, nil @@ -635,7 +635,7 @@ func (f *fragment) unprotectedClearBit(rowID, columnID uint64) (changed bool, er // a new copy if no one's reading it. f.rowCache.Add(rowID, nil) - f.stats.Count("clearBit", 1, 1.0) + f.stats.Count(MetricClearBit, 1, 1.0) return changed, nil } @@ -691,7 +691,7 @@ func (f *fragment) unprotectedSetRow(row *Row, rowID uint64) (changed bool, err // Snapshot storage. f.snapshotQueue.Enqueue(f) - f.stats.Count("setRow", 1, 1.0) + f.stats.Count(MetricSetRow, 1, 1.0) return changed, nil } @@ -733,7 +733,7 @@ func (f *fragment) unprotectedClearRow(rowID uint64) (changed bool, err error) { // Snapshot storage. f.snapshotQueue.Enqueue(f) - f.stats.Count("clearRow", 1, 1.0) + f.stats.Count(MetricClearRow, 1, 1.0) return changed, nil } @@ -1945,22 +1945,22 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct{}) error { err := f.gen.Transaction(&f.storage.OpWriter, func() error { if len(set) > 0 { - f.stats.Count("ImportingN", int64(len(set)), 1) + f.stats.Count(MetricImportingN, int64(len(set)), 1) changedN, err := f.storage.AddN(set...) // TODO benchmark Add/RemoveN behavior with sorted/unsorted positions if err != nil { return errors.Wrap(err, "adding positions") } - f.stats.Count("ImportedN", int64(changedN), 1) + f.stats.Count(MetricImportedN, int64(changedN), 1) f.incrementOpN(changedN) } if len(clear) > 0 { - f.stats.Count("ClearingN", int64(len(clear)), 1) + f.stats.Count(MetricClearingN, int64(len(clear)), 1) changedN, err := f.storage.RemoveN(clear...) if err != nil { return errors.Wrap(err, "clearing positions") } - f.stats.Count("ClearedN", int64(changedN), 1) + f.stats.Count(MetricClearedN, int64(changedN), 1) f.incrementOpN(changedN) } @@ -2249,7 +2249,7 @@ func (f *fragment) Snapshot() error { func track(start time.Time, message string, stats stats.StatsClient, logger logger.Logger) { elapsed := time.Since(start) logger.Debugf("%s took %s", message, elapsed) - stats.Histogram("snapshot", elapsed.Seconds(), 1.0) + stats.Histogram(MetricSnapshot, elapsed.Seconds(), 1.0) } // snapshot does the actual snapshot operation. it does not check or care @@ -3044,13 +3044,13 @@ func (s *fragmentSyncer) syncFragment() error { if err := s.syncBlockFromPrimary(blockID); err != nil { return fmt.Errorf("sync block from primary: id=%d, err=%s", blockID, err) } - s.Fragment.stats.Count("BlockRepairPrimary", 1, 1.0) + s.Fragment.stats.Count(MetricBlockRepairPrimary, 1, 1.0) default: // Synchronize block. if err := s.syncBlock(blockID); err != nil { return fmt.Errorf("sync block: id=%d, err=%s", blockID, err) } - s.Fragment.stats.Count("BlockRepair", 1, 1.0) + s.Fragment.stats.Count(MetricBlockRepair, 1, 1.0) } } diff --git a/holder.go b/holder.go index 3b9631761..6ba6d5967 100644 --- a/holder.go +++ b/holder.go @@ -801,10 +801,10 @@ func (s *holderSyncer) SyncHolder() error { } } } - s.Stats.Histogram("syncField", float64(time.Since(tf)), 1.0) + s.Stats.Histogram(MetricSyncField, float64(time.Since(tf)), 1.0) tf = time.Now() // reset tf } - s.Stats.Histogram("syncIndex", float64(time.Since(ti)), 1.0) + s.Stats.Histogram(MetricSyncIndex, float64(time.Since(ti)), 1.0) ti = time.Now() // reset ti } @@ -828,7 +828,7 @@ func (s *holderSyncer) syncIndex(index string) error { if err != nil { return errors.Wrap(err, "getting blocks") } - s.Stats.CountWithCustomTags("ColumnAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag}) + s.Stats.CountWithCustomTags(MetricColumnAttrStoreBlocks, int64(len(blks)), 1.0, []string{indexTag}) // Sync with every other host. for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) { @@ -840,7 +840,7 @@ func (s *holderSyncer) syncIndex(index string) error { } else if len(m) == 0 { continue } - s.Stats.CountWithCustomTags("ColumnAttrDiff", int64(len(m)), 1.0, []string{indexTag, node.ID}) + s.Stats.CountWithCustomTags(MetricColumnAttrDiff, int64(len(m)), 1.0, []string{indexTag, node.ID}) // Update local copy. if err := idx.ColumnAttrStore().SetBulkAttrs(m); err != nil { @@ -875,7 +875,7 @@ func (s *holderSyncer) syncField(index, name string) error { if err != nil { return errors.Wrap(err, "getting blocks") } - s.Stats.CountWithCustomTags("RowAttrStoreBlocks", int64(len(blks)), 1.0, []string{indexTag, fieldTag}) + s.Stats.CountWithCustomTags(MetricRowAttrStoreBlocks, int64(len(blks)), 1.0, []string{indexTag, fieldTag}) // Sync with every other host. for _, node := range Nodes(s.Cluster.nodes).FilterID(s.Node.ID) { @@ -889,7 +889,7 @@ func (s *holderSyncer) syncField(index, name string) error { } else if len(m) == 0 { continue } - s.Stats.CountWithCustomTags("RowAttrDiff", int64(len(m)), 1.0, []string{indexTag, fieldTag, node.ID}) + s.Stats.CountWithCustomTags(MetricRowAttrDiff, int64(len(m)), 1.0, []string{indexTag, fieldTag, node.ID}) // Update local copy. if err := f.RowAttrStore().SetBulkAttrs(m); err != nil { diff --git a/http/handler.go b/http/handler.go index cac26487c..c8faf46a0 100644 --- a/http/handler.go +++ b/http/handler.go @@ -296,7 +296,7 @@ func (h *Handler) collectStats(next http.Handler) http.Handler { stats := h.api.StatsWithTags(statsTags) if stats != nil { - stats.Timing("http.request", dur, 0.1) + stats.Timing(MetricHttpRequest, dur, 0.1) } }) } diff --git a/index.go b/index.go index a5b6245ba..a5f143bfc 100644 --- a/index.go +++ b/index.go @@ -350,7 +350,7 @@ func (i *Index) AvailableShards() *roaring.Bitmap { b.UnionInPlace(f.AvailableShards()) } - i.Stats.Gauge("maxShard", float64(b.Max()), 1.0) + i.Stats.Gauge(MetricMaxShard, float64(b.Max()), 1.0) return b } diff --git a/metrics.go b/metrics.go new file mode 100644 index 000000000..40005bdb0 --- /dev/null +++ b/metrics.go @@ -0,0 +1,48 @@ +package pilosa + +const ( + MetricCreateIndex = "createIndex" + MetricDeleteIndex = "deleteIndex" + MetricCreateField = "createField" + MetricDeleteField = "deleteField" + MetricDeleteAvailableShard = "deleteAvailableShard" + MetricRecalculateCache = "cache.recalculate" + MetricInvalidateCache = "cache.invalidate" + MetricRankCacheLength = "RankCache" + MetricCacheThresholdReached = "cache.threshold" + MetricRow = "range" + MetricRowBSI = "range:bsigroup" + MetricSetRowAttrs = "SetRowAttrs" + MetricSetProfileAttrs = "SetProfileAttrs" + MetricMaximumRow = "maximum_row" + MetricSetBit = "setBit" + MetricRows = "rows" + MetricClearBit = "clearBit" + MetricSetRow = "setRow" + MetricClearRow = "clearRow" + MetricImportingN = "ImportingN" + MetricImportedN = "ImportedN" + MetricClearingN = "ClearingN" + MetricClearedN = "ClearedN" + MetricSnapshot = "snapshot" + MetricBlockRepairPrimary = "BlockRepairPrimary" + MetricBlockRepair = "BlockRepair" + MetricSyncField = "syncField" + MetricSyncIndex = "syncIndex" + MetricColumnAttrStoreBlocks = "ColumnAttrStoreBlocks" + MetricColumnAttrDiff = "ColumnAttrDiff" + MetricRowAttrStoreBlocks = "RowAttrStoreBlocks" + MetricRowAttrDiff = "RowAttrDiff" + MetricHttpRequest = "http.request" + MetricMaxShard = "maxShard" + MetricAntiEntropy = "AntiEntropy" + MetricAntiEntropyDuration = "AntiEntropyDuration" + MetricGarbageCollection = "garbage_collection" + MetricGoroutines = "goroutines" + MetricOpenFiles = "OpenFiles" + MetricHeapAlloc = "HeapAlloc" + MetricHeapInuse = "HeapInuse" + MetricStackInuse = "StackInuse" + MetricMallocs = "Mallocs" + MetricFrees = "Frees" +) diff --git a/server.go b/server.go index cecc520e4..810947b47 100644 --- a/server.go +++ b/server.go @@ -654,7 +654,7 @@ func (s *Server) monitorAntiEntropy() { case <-s.cluster.abortAntiEntropyCh: // receive here so we don't block resizing continue case <-ticker.C: - s.holder.Stats.Count("AntiEntropy", 1, 1.0) + s.holder.Stats.Count(MetricAntiEntropy, 1, 1.0) } t := time.Now() if s.cluster.State() == ClusterStateResizing { @@ -675,7 +675,7 @@ func (s *Server) monitorAntiEntropy() { // Record successful sync in log. s.logger.Printf("holder sync complete") dif := time.Since(t) - s.holder.Stats.Histogram("AntiEntropyDuration", float64(dif), 1.0) + s.holder.Stats.Histogram(MetricAntiEntropyDuration, float64(dif), 1.0) // Drain tick channel since we just finished anti-entropy. If the AE // process took a long time, we don't want them to pile up on each @@ -957,26 +957,26 @@ func (s *Server) monitorRuntime() { return case <-s.gcNotifier.AfterGC(): // GC just ran. - s.holder.Stats.Count("garbage_collection", 1, 1.0) + s.holder.Stats.Count(MetricGarbageCollection, 1, 1.0) case <-ticker.C: } // Record the number of go routines. - s.holder.Stats.Gauge("goroutines", float64(runtime.NumGoroutine()), 1.0) + s.holder.Stats.Gauge(MetricGoroutines, float64(runtime.NumGoroutine()), 1.0) openFiles, err := countOpenFiles() // Open File handles. if err == nil { - s.holder.Stats.Gauge("OpenFiles", float64(openFiles), 1.0) + s.holder.Stats.Gauge(MetricOpenFiles, float64(openFiles), 1.0) } // Runtime memory metrics. runtime.ReadMemStats(&m) - s.holder.Stats.Gauge("HeapAlloc", float64(m.HeapAlloc), 1.0) - s.holder.Stats.Gauge("HeapInuse", float64(m.HeapInuse), 1.0) - s.holder.Stats.Gauge("StackInuse", float64(m.StackInuse), 1.0) - s.holder.Stats.Gauge("Mallocs", float64(m.Mallocs), 1.0) - s.holder.Stats.Gauge("Frees", float64(m.Frees), 1.0) + s.holder.Stats.Gauge(MetricHeapAlloc, float64(m.HeapAlloc), 1.0) + s.holder.Stats.Gauge(MetricHeapInuse, float64(m.HeapInuse), 1.0) + s.holder.Stats.Gauge(MetricStackInuse, float64(m.StackInuse), 1.0) + s.holder.Stats.Gauge(MetricMallocs, float64(m.Mallocs), 1.0) + s.holder.Stats.Gauge(MetricFrees, float64(m.Frees), 1.0) } }