mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-06 16:45:55 +00:00
Merge pull request #246 from molecula/prometheus-improvements
Prometheus improvements
This commit is contained in:
Matthew Jaffee
GitHub
commit
76404142ba
15 changed files with 235 additions and 105 deletions
10
api.go
10
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
|
||||
}
|
||||
|
||||
|
|
|
|||
8
cache.go
8
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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -71,7 +71,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
|
|||
flags.DurationVarP((*time.Duration)(&srv.Config.AntiEntropy.Interval), "anti-entropy.interval", "", (time.Duration)(srv.Config.AntiEntropy.Interval), "Interval at which to run anti-entropy routine.")
|
||||
|
||||
// Metric
|
||||
flags.StringVarP(&srv.Config.Metric.Service, "metric.service", "", srv.Config.Metric.Service, "Where to send stats: can be expvar (in-memory served at /debug/vars), statsd or none.")
|
||||
flags.StringVarP(&srv.Config.Metric.Service, "metric.service", "", srv.Config.Metric.Service, "Where to send stats: can be expvar (in-memory served at /debug/vars), prometheus, statsd or none.")
|
||||
flags.StringVarP(&srv.Config.Metric.Host, "metric.host", "", srv.Config.Metric.Host, "URI to send metrics when metric.service is statsd.")
|
||||
flags.DurationVarP((*time.Duration)(&srv.Config.Metric.PollInterval), "metric.poll-interval", "", (time.Duration)(srv.Config.Metric.PollInterval), "Polling interval metrics.")
|
||||
flags.BoolVarP((&srv.Config.Metric.Diagnostics), "metric.diagnostics", "", srv.Config.Metric.Diagnostics, "Enabled diagnostics reporting.")
|
||||
|
|
|
|||
59
executor.go
59
executor.go
|
|
@ -462,7 +462,13 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s
|
|||
} else if err := e.validateCallArgs(c); err != nil {
|
||||
return nil, errors.Wrap(err, "validating args")
|
||||
}
|
||||
indexTag := fmt.Sprintf("index:%s", index)
|
||||
indexTag := "index:" + index
|
||||
metricName := "query_" + strings.ToLower(c.Name) + "_total"
|
||||
statFn := func() {
|
||||
if !opt.Remote {
|
||||
e.Holder.Stats.CountWithCustomTags(metricName, 1, 1.0, []string{indexTag})
|
||||
}
|
||||
}
|
||||
|
||||
// Fixes #2009
|
||||
// See: https://github.com/pilosa/pilosa/issues/2009
|
||||
|
|
@ -487,63 +493,71 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s
|
|||
|
||||
// Special handling for mutation and top-n calls.
|
||||
if op, ok := e.additionalCountOps[c.Name]; ok {
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeGenericCount(ctx, index, c, op, shards, opt)
|
||||
}
|
||||
if op, ok := e.additionalFieldOps[c.Name]; ok {
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeGenericField(ctx, index, c, op, shards, opt)
|
||||
}
|
||||
switch c.Name {
|
||||
case "Sum":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeSum(ctx, index, c, shards, opt)
|
||||
case "Min":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeMin(ctx, index, c, shards, opt)
|
||||
case "Max":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeMax(ctx, index, c, shards, opt)
|
||||
case "MinRow":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeMinRow(ctx, index, c, shards, opt)
|
||||
case "MaxRow":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeMaxRow(ctx, index, c, shards, opt)
|
||||
case "Clear":
|
||||
statFn()
|
||||
return e.executeClearBit(ctx, index, c, opt)
|
||||
case "ClearRow":
|
||||
statFn()
|
||||
return e.executeClearRow(ctx, index, c, shards, opt)
|
||||
case "Store":
|
||||
statFn()
|
||||
return e.executeSetRow(ctx, index, c, shards, opt)
|
||||
case "Count":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeCount(ctx, index, c, shards, opt)
|
||||
case "Set":
|
||||
statFn()
|
||||
return e.executeSet(ctx, index, c, opt)
|
||||
case "SetRowAttrs":
|
||||
statFn()
|
||||
return nil, e.executeSetRowAttrs(ctx, index, c, opt)
|
||||
case "SetColumnAttrs":
|
||||
statFn()
|
||||
return nil, e.executeSetColumnAttrs(ctx, index, c, opt)
|
||||
case "TopN":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeTopN(ctx, index, c, shards, opt)
|
||||
case "Rows":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeRows(ctx, index, c, shards, opt)
|
||||
case "GroupBy":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeGroupBy(ctx, index, c, shards, opt)
|
||||
case "Options":
|
||||
statFn()
|
||||
return e.executeOptionsCall(ctx, index, c, shards, opt)
|
||||
case "IncludesColumn":
|
||||
return e.executeIncludesColumnCall(ctx, index, c, shards, opt)
|
||||
case "All":
|
||||
statFn()
|
||||
return e.executeAllCall(ctx, index, c, shards, opt)
|
||||
case "Precomputed":
|
||||
return e.executePrecomputedCall(ctx, index, c, shards, opt)
|
||||
default:
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
statFn()
|
||||
return e.executeBitmapCall(ctx, index, c, shards, opt)
|
||||
}
|
||||
}
|
||||
|
|
@ -1023,6 +1037,15 @@ func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.C
|
|||
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeBitmapCall")
|
||||
defer span.Finish()
|
||||
|
||||
indexTag := "index:" + index
|
||||
metricName := "query_" + strings.ToLower(c.Name) + "_total"
|
||||
if c.Name == "Row" && c.HasConditionArg() {
|
||||
metricName = "query_row_bsi_total"
|
||||
}
|
||||
if !opt.Remote {
|
||||
e.Holder.Stats.CountWithCustomTags(metricName, 1, 1.0, []string{indexTag})
|
||||
}
|
||||
|
||||
// Execute calls in bulk on each remote node and merge.
|
||||
mapFn := func(shard uint64) (interface{}, error) {
|
||||
return e.executeBitmapCallShard(ctx, index, c, shard)
|
||||
|
|
@ -2228,7 +2251,6 @@ 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)
|
||||
return row, nil
|
||||
|
||||
}
|
||||
|
|
@ -2359,7 +2381,6 @@ func (e *executor) executeRowBSIGroupShard(ctx context.Context, index string, c
|
|||
return frag.notNull()
|
||||
}
|
||||
|
||||
f.Stats.Count("range:bsigroup", 1, 1.0)
|
||||
return frag.rangeOp(cond.Op, bsig.BitDepth, baseValue)
|
||||
}
|
||||
}
|
||||
|
|
@ -3102,7 +3123,6 @@ 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)
|
||||
|
||||
// Do not forward call if this is already being forwarded.
|
||||
if opt.Remote {
|
||||
|
|
@ -3196,7 +3216,11 @@ 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)
|
||||
}
|
||||
|
||||
if !opt.Remote {
|
||||
tags := []string{"index:" + index, "bulk:true"}
|
||||
e.Holder.Stats.CountWithCustomTags(MetricSetRowAttrs, int64(len(m)), 1.0, tags)
|
||||
}
|
||||
|
||||
// Do not forward call if this is already being forwarded.
|
||||
|
|
@ -3250,7 +3274,6 @@ 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)
|
||||
// Do not forward call if this is already being forwarded.
|
||||
if opt.Remote {
|
||||
return nil
|
||||
|
|
|
|||
25
fragment.go
25
fragment.go
|
|
@ -208,7 +208,6 @@ 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)
|
||||
return nil
|
||||
}(); err != nil {
|
||||
f.close()
|
||||
|
|
@ -576,12 +575,11 @@ 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, 1.0)
|
||||
|
||||
// Update row count if they have increased.
|
||||
if rowID > f.maxRowID {
|
||||
f.maxRowID = rowID
|
||||
f.stats.Gauge("rows", float64(f.maxRowID), 1.0)
|
||||
}
|
||||
|
||||
return changed, nil
|
||||
|
|
@ -635,7 +633,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 +689,6 @@ func (f *fragment) unprotectedSetRow(row *Row, rowID uint64) (changed bool, err
|
|||
|
||||
// Snapshot storage.
|
||||
f.snapshotQueue.Enqueue(f)
|
||||
f.stats.Count("setRow", 1, 1.0)
|
||||
|
||||
return changed, nil
|
||||
}
|
||||
|
|
@ -733,8 +730,6 @@ func (f *fragment) unprotectedClearRow(rowID uint64) (changed bool, err error) {
|
|||
// Snapshot storage.
|
||||
f.snapshotQueue.Enqueue(f)
|
||||
|
||||
f.stats.Count("clearRow", 1, 1.0)
|
||||
|
||||
return changed, nil
|
||||
}
|
||||
|
||||
|
|
@ -1945,22 +1940,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)
|
||||
}
|
||||
|
||||
|
|
@ -2248,8 +2243,8 @@ func (f *fragment) Snapshot() error {
|
|||
|
||||
func track(start time.Time, message string, stats stats.StatsClient, logger logger.Logger) {
|
||||
elapsed := time.Since(start)
|
||||
logger.Printf("%s took %s", message, elapsed)
|
||||
stats.Histogram("snapshot", elapsed.Seconds(), 1.0)
|
||||
logger.Debugf("%s took %s", message, elapsed)
|
||||
stats.Timing(MetricSnapshotDurationSeconds, elapsed, 1.0)
|
||||
}
|
||||
|
||||
// snapshot does the actual snapshot operation. it does not check or care
|
||||
|
|
@ -3044,13 +3039,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.CountWithCustomTags(MetricBlockRepair, 1, 1.0, []string{"primary:true"})
|
||||
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.CountWithCustomTags(MetricBlockRepair, 1, 1.0, []string{"primary:false"})
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
12
holder.go
12
holder.go
|
|
@ -801,10 +801,10 @@ func (s *holderSyncer) SyncHolder() error {
|
|||
}
|
||||
}
|
||||
}
|
||||
s.Stats.Histogram("syncField", float64(time.Since(tf)), 1.0)
|
||||
s.Stats.Timing(MetricSyncFieldDurationSeconds, time.Since(tf), 1.0)
|
||||
tf = time.Now() // reset tf
|
||||
}
|
||||
s.Stats.Histogram("syncIndex", float64(time.Since(ti)), 1.0)
|
||||
s.Stats.Timing(MetricSyncIndexDurationSeconds, 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 {
|
||||
|
|
|
|||
|
|
@ -270,19 +270,23 @@ func (h *Handler) collectStats(next http.Handler) http.Handler {
|
|||
longQueryTime := h.api.LongQueryTime()
|
||||
if longQueryTime > 0 && dur > longQueryTime {
|
||||
queryRequest := r.Context().Value(contextKeyQueryRequest)
|
||||
req, ok := queryRequest.(*pilosa.QueryRequest)
|
||||
queryString := req.Query
|
||||
if !ok {
|
||||
queryString = ""
|
||||
|
||||
var queryString string
|
||||
if req, ok := queryRequest.(*pilosa.QueryRequest); ok {
|
||||
queryString = req.Query
|
||||
}
|
||||
|
||||
h.logger.Printf("%s %s %v %s", r.Method, r.URL.String(), dur, queryString)
|
||||
statsTags = append(statsTags, "slow_query")
|
||||
statsTags = append(statsTags, "slow:true")
|
||||
} else {
|
||||
statsTags = append(statsTags, "slow:false")
|
||||
}
|
||||
|
||||
pathParts := strings.Split(r.URL.Path, "/")
|
||||
if externalPrefixFlag[pathParts[1]] {
|
||||
statsTags = append(statsTags, "external")
|
||||
statsTags = append(statsTags, "where:external")
|
||||
} else {
|
||||
statsTags = append(statsTags, "where:internal")
|
||||
}
|
||||
|
||||
statsTags = append(statsTags, "useragent:"+r.UserAgent())
|
||||
|
|
@ -296,7 +300,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(pilosa.MetricHTTPRequest, dur, 0.1)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
|
|||
2
index.go
2
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
|
||||
}
|
||||
|
||||
|
|
|
|||
61
metrics.go
Normal file
61
metrics.go
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
// Copyright 2020 Pilosa Corp.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package pilosa
|
||||
|
||||
const (
|
||||
MetricCreateIndex = "create_index_total"
|
||||
MetricDeleteIndex = "delete_index_total"
|
||||
MetricCreateField = "create_field_total"
|
||||
MetricDeleteField = "delete_field_total"
|
||||
MetricDeleteAvailableShard = "delete_available_shard_total"
|
||||
MetricRecalculateCache = "recalculate_cache_total"
|
||||
MetricInvalidateCache = "invalidate_cache_total"
|
||||
MetricRankCacheLength = "rank_cache_length"
|
||||
MetricCacheThresholdReached = "cache_threshold_reached_total"
|
||||
MetricRow = "query_row_total"
|
||||
MetricRowBSI = "query_row_bsi_total"
|
||||
MetricSetRowAttrs = "query_setrowattrs_total"
|
||||
MetricSetColumnAttrs = "query_setcolumnattrs_total"
|
||||
MetricSetBit = "set_bit_total"
|
||||
MetricClearBit = "clear_bit_total"
|
||||
MetricImportingN = "importing_total"
|
||||
MetricImportedN = "imported_total"
|
||||
MetricClearingN = "clearing_total"
|
||||
MetricClearedN = "cleared_total"
|
||||
MetricSnapshotDurationSeconds = "snapshot_duration_seconds"
|
||||
MetricBlockRepair = "block_repair_total"
|
||||
MetricSyncFieldDurationSeconds = "sync_field_duration_seconds"
|
||||
MetricSyncIndexDurationSeconds = "sync_index_duration_seconds"
|
||||
MetricColumnAttrStoreBlocks = "column_attr_store_blocks_total"
|
||||
MetricColumnAttrDiff = "column_attr_diff_total"
|
||||
MetricRowAttrStoreBlocks = "row_attr_store_blocks_total"
|
||||
MetricRowAttrDiff = "row_attr_diff_total"
|
||||
MetricHTTPRequest = "http_request_duration_seconds"
|
||||
MetricGRPCUnaryQueryDurationSeconds = "grpc_request_pql_unary_query_duration_seconds"
|
||||
MetricGRPCUnaryFormatDurationSeconds = "grpc_request_pql_unary_format_duration_seconds"
|
||||
MetricGRPCStreamQueryDurationSeconds = "grpc_request_pql_stream_query_duration_seconds"
|
||||
MetricGRPCStreamFormatDurationSeconds = "grpc_request_pql_stream_format_duration_seconds"
|
||||
MetricMaxShard = "maximum_shard"
|
||||
MetricAntiEntropy = "antientropy_total"
|
||||
MetricAntiEntropyDurationSeconds = "antientropy_duration_seconds"
|
||||
MetricGarbageCollection = "garbage_collection_total"
|
||||
MetricGoroutines = "goroutines"
|
||||
MetricOpenFiles = "open_files"
|
||||
MetricHeapAlloc = "heap_alloc"
|
||||
MetricHeapInuse = "heap_inuse"
|
||||
MetricStackInuse = "stack_inuse"
|
||||
MetricMallocs = "mallocs"
|
||||
MetricFrees = "frees"
|
||||
)
|
||||
|
|
@ -94,7 +94,7 @@ func (c *prometheusClient) Tags() []string {
|
|||
|
||||
// labels returns an instance of prometheus.Labels with the value of the set tags.
|
||||
func (c *prometheusClient) labels() prometheus.Labels {
|
||||
return tagsToLabels(c.tags)
|
||||
return tagsToLabels(c.tags, c.logger)
|
||||
}
|
||||
|
||||
// WithTags returns a new client with additional tags appended.
|
||||
|
|
@ -252,8 +252,7 @@ func (c *prometheusClient) Set(name string, value string, rate float64) {
|
|||
|
||||
// Timing tracks timing information for a metric.
|
||||
func (c *prometheusClient) Timing(name string, value time.Duration, rate float64) {
|
||||
durationMs := value / time.Second
|
||||
c.Histogram(name, float64(durationMs), rate)
|
||||
c.Histogram(name, value.Seconds(), rate)
|
||||
}
|
||||
|
||||
// SetLogger sets the logger for client.
|
||||
|
|
@ -296,12 +295,13 @@ func unionStringSlice(a, b []string) []string {
|
|||
return other
|
||||
}
|
||||
|
||||
func tagsToLabels(tags []string) (labels prometheus.Labels) {
|
||||
func tagsToLabels(tags []string, logger logger.Logger) (labels prometheus.Labels) {
|
||||
labels = make(prometheus.Labels)
|
||||
for _, tag := range tags {
|
||||
tagParts := strings.SplitAfterN(tag, ":", 2)
|
||||
if len(tagParts) != 2 {
|
||||
// only process tags in "key:value" form
|
||||
logger.Printf("Error: invalid Prometheus label: %v\n", tag)
|
||||
continue
|
||||
}
|
||||
labels[tagParts[0][0:len(tagParts[0])-1]] = tagParts[1]
|
||||
|
|
|
|||
22
server.go
22
server.go
|
|
@ -517,7 +517,7 @@ func (s *Server) Open() error {
|
|||
s.syncer.Node = s.cluster.Node
|
||||
s.syncer.Cluster = s.cluster
|
||||
s.syncer.Closing = s.closing
|
||||
s.syncer.Stats = s.holder.Stats.WithTags("HolderSyncer")
|
||||
s.syncer.Stats = s.holder.Stats.WithTags("component:HolderSyncer")
|
||||
|
||||
// Start background process listening for translation
|
||||
// sync resets.
|
||||
|
|
@ -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.Timing(MetricAntiEntropyDurationSeconds, 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)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -131,7 +131,7 @@ type Config struct {
|
|||
} `toml:"anti-entropy"`
|
||||
|
||||
Metric struct {
|
||||
// Service can be statsd, expvar, or none.
|
||||
// Service can be statsd, prometheus, expvar, or none.
|
||||
Service string `toml:"service"`
|
||||
// Host tells the statsd client where to write.
|
||||
Host string `toml:"host"`
|
||||
|
|
|
|||
|
|
@ -20,10 +20,12 @@ import (
|
|||
"fmt"
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/pilosa/pilosa/v2"
|
||||
"github.com/pilosa/pilosa/v2/logger"
|
||||
pb "github.com/pilosa/pilosa/v2/proto"
|
||||
"github.com/pilosa/pilosa/v2/stats"
|
||||
"github.com/pkg/errors"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
|
|
@ -37,6 +39,8 @@ type grpcHandler struct {
|
|||
api *pilosa.API
|
||||
|
||||
logger logger.Logger
|
||||
|
||||
stats stats.StatsClient
|
||||
}
|
||||
|
||||
// errorToStatusError appends an appropriate grpc status code
|
||||
|
|
@ -62,16 +66,28 @@ func (h grpcHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQL
|
|||
Index: req.Index,
|
||||
Query: req.Pql,
|
||||
}
|
||||
|
||||
t := time.Now()
|
||||
resp, err := h.api.Query(context.Background(), &query)
|
||||
durQuery := time.Since(t)
|
||||
if err != nil {
|
||||
return errToStatusError(err)
|
||||
}
|
||||
longQueryTime := h.api.LongQueryTime()
|
||||
if longQueryTime > 0 && durQuery > longQueryTime {
|
||||
h.logger.Printf("GRPC QueryPQL %v %s", durQuery, query.Query)
|
||||
}
|
||||
|
||||
t = time.Now()
|
||||
for row := range makeRows(resp, h.logger) {
|
||||
err = stream.Send(row)
|
||||
if err != nil {
|
||||
return errToStatusError(err)
|
||||
}
|
||||
}
|
||||
durFormat := time.Since(t)
|
||||
h.stats.Timing(pilosa.MetricGRPCStreamQueryDurationSeconds, durQuery, 0.1)
|
||||
h.stats.Timing(pilosa.MetricGRPCStreamFormatDurationSeconds, durFormat, 0.1)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -82,10 +98,19 @@ func (h grpcHandler) QueryPQLUnary(ctx context.Context, req *pb.QueryPQLRequest)
|
|||
Index: req.Index,
|
||||
Query: req.Pql,
|
||||
}
|
||||
|
||||
t := time.Now()
|
||||
resp, err := h.api.Query(context.Background(), &query)
|
||||
durQuery := time.Since(t)
|
||||
if err != nil {
|
||||
return nil, errToStatusError(err)
|
||||
}
|
||||
longQueryTime := h.api.LongQueryTime()
|
||||
if longQueryTime > 0 && durQuery > longQueryTime {
|
||||
h.logger.Printf("GRPC QueryPQLUnary %v %s", durQuery, query.Query)
|
||||
}
|
||||
|
||||
t = time.Now()
|
||||
response := &pb.TableResponse{
|
||||
Rows: make([]*pb.Row, 0),
|
||||
}
|
||||
|
|
@ -95,6 +120,9 @@ func (h grpcHandler) QueryPQLUnary(ctx context.Context, req *pb.QueryPQLRequest)
|
|||
}
|
||||
response.Rows = append(response.Rows, &pb.Row{Columns: row.Columns})
|
||||
}
|
||||
durFormat := time.Since(t)
|
||||
h.stats.Timing(pilosa.MetricGRPCUnaryQueryDurationSeconds, durQuery, 0.1)
|
||||
h.stats.Timing(pilosa.MetricGRPCUnaryFormatDurationSeconds, durFormat, 0.1)
|
||||
|
||||
return response, nil
|
||||
}
|
||||
|
|
@ -878,6 +906,7 @@ type grpcServer struct {
|
|||
hostPort string
|
||||
|
||||
logger logger.Logger
|
||||
stats stats.StatsClient
|
||||
}
|
||||
|
||||
type grpcServerOption func(s *grpcServer) error
|
||||
|
|
@ -904,6 +933,13 @@ func OptGRPCServerLogger(logger logger.Logger) grpcServerOption {
|
|||
}
|
||||
}
|
||||
|
||||
func OptGRPCServerStats(stats stats.StatsClient) grpcServerOption {
|
||||
return func(s *grpcServer) error {
|
||||
s.stats = stats
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (s *grpcServer) Serve(tlsConfig *tls.Config) error {
|
||||
// create listener
|
||||
lis, err := net.Listen("tcp", s.hostPort)
|
||||
|
|
@ -920,7 +956,7 @@ func (s *grpcServer) Serve(tlsConfig *tls.Config) error {
|
|||
|
||||
// create grpc server
|
||||
s.grpcServer = grpc.NewServer(opts...)
|
||||
pb.RegisterPilosaServer(s.grpcServer, grpcHandler{api: s.api, logger: s.logger})
|
||||
pb.RegisterPilosaServer(s.grpcServer, grpcHandler{api: s.api, logger: s.logger, stats: s.stats})
|
||||
|
||||
// register the server so its services are available to grpc_cli and others
|
||||
reflection.Register(s.grpcServer)
|
||||
|
|
|
|||
|
|
@ -376,6 +376,7 @@ func (m *Command) SetupServer() error {
|
|||
OptGRPCServerAPI(m.API),
|
||||
OptGRPCServerURI(grpcURI),
|
||||
OptGRPCServerLogger(m.logger),
|
||||
OptGRPCServerStats(statsClient),
|
||||
)
|
||||
return errors.Wrap(err, "new grpc server")
|
||||
}
|
||||
|
|
@ -453,7 +454,7 @@ func newStatsClient(name string, host string) (stats.StatsClient, error) {
|
|||
case "nop", "none":
|
||||
return stats.NopStatsClient, nil
|
||||
default:
|
||||
return nil, errors.Errorf("'%v' not a valid stats client, choose from [expvar, statsd, none].", name)
|
||||
return nil, errors.Errorf("'%v' not a valid stats client, choose from [expvar, statsd, prometheus, none].", name)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ package stats_test
|
|||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
|
@ -45,39 +46,41 @@ func TestMultiStatClient_Expvar(t *testing.T) {
|
|||
hldr.SetBit("d", "f", 0, pilosa.ShardWidth+2)
|
||||
hldr.ClearBit("d", "f", 0, 1)
|
||||
|
||||
if stats.Expvar.String() != `{"index:d": {"clearBit": 1, "rows": 0, "setBit": 4}}` {
|
||||
indexStats := fmt.Sprintf(`{"%s": %d, "%s": %d}`, pilosa.MetricClearBit, 1, pilosa.MetricSetBit, 4)
|
||||
|
||||
if stats.Expvar.String() != `{"index:d": `+indexStats+`}` {
|
||||
t.Fatalf("unexpected expvar : %s", stats.Expvar.String())
|
||||
}
|
||||
|
||||
hldr.Stats.CountWithCustomTags("cc", 1, 1.0, []string{"foo:bar"})
|
||||
if stats.Expvar.String() != `{"cc": 1, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}}` {
|
||||
if stats.Expvar.String() != `{"cc": 1, "index:d": `+indexStats+`}` {
|
||||
t.Fatalf("unexpected expvar : %s", stats.Expvar.String())
|
||||
}
|
||||
|
||||
// Gauge creates a unique key, subsequent Gauge calls will overwrite
|
||||
hldr.Stats.Gauge("g", 5, 1.0)
|
||||
hldr.Stats.Gauge("g", 8, 1.0)
|
||||
if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}}` {
|
||||
if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": `+indexStats+`}` {
|
||||
t.Fatalf("unexpected expvar : %s", stats.Expvar.String())
|
||||
}
|
||||
|
||||
// Set creates a unique key, subsequent sets will overwrite
|
||||
hldr.Stats.Set("s", "4", 1.0)
|
||||
hldr.Stats.Set("s", "7", 1.0)
|
||||
if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}, "s": "7"}` {
|
||||
if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": `+indexStats+`, "s": "7"}` {
|
||||
t.Fatalf("unexpected expvar : %s", stats.Expvar.String())
|
||||
}
|
||||
|
||||
// Record timing duration and a uniquely Set key/value
|
||||
dur, _ := time.ParseDuration("123us")
|
||||
hldr.Stats.Timing("tt", dur, 1.0)
|
||||
if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}, "s": "7", "tt": 123µs}` {
|
||||
if stats.Expvar.String() != `{"cc": 1, "g": 8, "index:d": `+indexStats+`, "s": "7", "tt": 123µs}` {
|
||||
t.Fatalf("unexpected expvar : %s", stats.Expvar.String())
|
||||
}
|
||||
|
||||
// Expvar histogram is implemented as a gauge
|
||||
hldr.Stats.Histogram("hh", 3, 1.0)
|
||||
if stats.Expvar.String() != `{"cc": 1, "g": 8, "hh": 3, "index:d": {"clearBit": 1, "rows": 0, "setBit": 4}, "s": "7", "tt": 123µs}` {
|
||||
if stats.Expvar.String() != `{"cc": 1, "g": 8, "hh": 3, "index:d": `+indexStats+`, "s": "7", "tt": 123µs}` {
|
||||
t.Fatalf("unexpected expvar : %s", stats.Expvar.String())
|
||||
}
|
||||
|
||||
|
|
@ -101,12 +104,12 @@ func TestStatsCount_TopN(t *testing.T) {
|
|||
called := false
|
||||
hldr.Holder.Stats = &MockStats{
|
||||
mockCountWithTags: func(name string, value int64, rate float64, tags []string) {
|
||||
if name != "TopN" {
|
||||
t.Errorf("Expected TopN, Results %s", name)
|
||||
if name != "query_topn_total" {
|
||||
t.Errorf("Expected query_topn_total, Results %s", name)
|
||||
}
|
||||
|
||||
if tags[0] != "index:d" {
|
||||
t.Errorf("Expected db, Results %s", tags[0])
|
||||
t.Errorf("Expected index, Results %s", tags[0])
|
||||
}
|
||||
|
||||
called = true
|
||||
|
|
@ -130,12 +133,12 @@ func TestStatsCount_Bitmap(t *testing.T) {
|
|||
called := false
|
||||
hldr.Holder.Stats = &MockStats{
|
||||
mockCountWithTags: func(name string, value int64, rate float64, tags []string) {
|
||||
if name != "Row" {
|
||||
t.Errorf("Expected Row, Results %s", name)
|
||||
if name != pilosa.MetricRow {
|
||||
t.Errorf("Expected %s, Results %s", pilosa.MetricRow, name)
|
||||
}
|
||||
|
||||
if tags[0] != "index:d" {
|
||||
t.Errorf("Expected db, Results %s", tags[0])
|
||||
t.Errorf("Expected index, Results %s", tags[0])
|
||||
}
|
||||
|
||||
called = true
|
||||
|
|
@ -149,7 +152,7 @@ func TestStatsCount_Bitmap(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestStatsCount_SetColumnAttrs(t *testing.T) {
|
||||
func TestStatsCount_SetRowAttrsBulk(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
hldr := test.Holder{Holder: c[0].Server.Holder()}
|
||||
|
|
@ -163,10 +166,14 @@ func TestStatsCount_SetColumnAttrs(t *testing.T) {
|
|||
t.Fatal("field not found")
|
||||
}
|
||||
|
||||
field.Stats = &MockStats{
|
||||
mockCount: func(name string, value int64, rate float64) {
|
||||
if name != "SetRowAttrs" {
|
||||
t.Errorf("Expected SetRowAttrs, Results %s", name)
|
||||
hldr.Holder.Stats = &MockStats{
|
||||
mockCountWithTags: func(name string, value int64, rate float64, tags []string) {
|
||||
if name != pilosa.MetricSetRowAttrs {
|
||||
t.Errorf("Expected %v, Results %s", pilosa.MetricSetRowAttrs, name)
|
||||
}
|
||||
|
||||
if tags[0] != "index:d" {
|
||||
t.Errorf("Expected index, Results %s", tags[0])
|
||||
}
|
||||
called = true
|
||||
},
|
||||
|
|
@ -179,7 +186,7 @@ func TestStatsCount_SetColumnAttrs(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestStatsCount_SetProfileAttrs(t *testing.T) {
|
||||
func TestStatsCount_SetColumnAttrs(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
hldr := test.Holder{Holder: c[0].Server.Holder()}
|
||||
|
|
@ -193,12 +200,15 @@ func TestStatsCount_SetProfileAttrs(t *testing.T) {
|
|||
t.Fatal("idex not found")
|
||||
}
|
||||
|
||||
idx.Stats = &MockStats{
|
||||
mockCount: func(name string, value int64, rate float64) {
|
||||
if name != "SetProfileAttrs" {
|
||||
t.Errorf("Expected SetProfilepAttrs, Results %s", name)
|
||||
hldr.Holder.Stats = &MockStats{
|
||||
mockCountWithTags: func(name string, value int64, rate float64, tags []string) {
|
||||
if name != pilosa.MetricSetColumnAttrs {
|
||||
t.Errorf("Expected %v, Results %s", pilosa.MetricSetColumnAttrs, name)
|
||||
}
|
||||
|
||||
if tags[0] != "index:d" {
|
||||
t.Errorf("Expected index, Results %s", tags[0])
|
||||
}
|
||||
called = true
|
||||
},
|
||||
}
|
||||
|
|
@ -222,8 +232,8 @@ func TestStatsCount_APICalls(t *testing.T) {
|
|||
called := false
|
||||
hldr.Stats = &MockStats{
|
||||
mockCount: func(name string, value int64, rate float64) {
|
||||
if name != "createIndex" {
|
||||
t.Errorf("Expected createIndex, Results %s", name)
|
||||
if name != pilosa.MetricCreateIndex {
|
||||
t.Errorf("Expected %v, Results %s", pilosa.MetricCreateIndex, name)
|
||||
}
|
||||
called = true
|
||||
},
|
||||
|
|
@ -239,8 +249,8 @@ func TestStatsCount_APICalls(t *testing.T) {
|
|||
called := false
|
||||
hldr.Stats = &MockStats{
|
||||
mockCountWithTags: func(name string, value int64, rate float64, index []string) {
|
||||
if name != "createField" {
|
||||
t.Errorf("Expected createField, Results %s", name)
|
||||
if name != pilosa.MetricCreateField {
|
||||
t.Errorf("Expected %v, Results %s", pilosa.MetricCreateField, name)
|
||||
}
|
||||
if index[0] != "index:i" {
|
||||
t.Errorf("Expected index:i, Results %s", index)
|
||||
|
|
@ -260,8 +270,8 @@ func TestStatsCount_APICalls(t *testing.T) {
|
|||
called := false
|
||||
hldr.Stats = &MockStats{
|
||||
mockCountWithTags: func(name string, value int64, rate float64, index []string) {
|
||||
if name != "deleteField" {
|
||||
t.Errorf("Expected deleteField, Results %s", name)
|
||||
if name != pilosa.MetricDeleteField {
|
||||
t.Errorf("Expected %v, Results %s", pilosa.MetricDeleteField, name)
|
||||
}
|
||||
if index[0] != "index:i" {
|
||||
t.Errorf("Expected index:i, Results %s", index)
|
||||
|
|
@ -281,8 +291,8 @@ func TestStatsCount_APICalls(t *testing.T) {
|
|||
called := false
|
||||
hldr.Stats = &MockStats{
|
||||
mockCount: func(name string, value int64, rate float64) {
|
||||
if name != "deleteIndex" {
|
||||
t.Errorf("Expected deleteIndex, Results %s", name)
|
||||
if name != pilosa.MetricDeleteIndex {
|
||||
t.Errorf("Expected %v, Results %s", pilosa.MetricDeleteIndex, name)
|
||||
}
|
||||
|
||||
called = true
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue