Line data Source code
1 : // Copyright 2018 The LevelDB-Go and Pebble Authors. All rights reserved. Use 2 : // of this source code is governed by a BSD-style license that can be found in 3 : // the LICENSE file. 4 : 5 : package sstable 6 : 7 : import ( 8 : "cmp" 9 : "slices" 10 : "sync" 11 : "time" 12 : 13 : "github.com/cockroachdb/errors" 14 : "github.com/cockroachdb/redact" 15 : ) 16 : 17 : // Category is a user-understandable string, where stats are aggregated for 18 : // each category. The cardinality of this should be low, say < 20. The prefix 19 : // "pebble-" is reserved for internal Pebble categories. 20 : // 21 : // Examples of categories that can be useful in the CockroachDB context are: 22 : // sql-user, sql-stats, raft, rangefeed, mvcc-gc, range-snapshot. 23 : type Category string 24 : 25 : // QoSLevel describes whether the read is latency-sensitive or not. Each 26 : // category must map to a single QoSLevel. While category strings are opaque 27 : // to Pebble, the QoSLevel may be internally utilized in Pebble to better 28 : // optimize future reads. 29 : type QoSLevel int 30 : 31 : const ( 32 : // LatencySensitiveQoSLevel is the default when QoSLevel is not specified, 33 : // and represents reads that are latency-sensitive. 34 : LatencySensitiveQoSLevel QoSLevel = iota 35 : // NonLatencySensitiveQoSLevel represents reads that are not 36 : // latency-sensitive. 37 : NonLatencySensitiveQoSLevel 38 : ) 39 : 40 : // SafeFormat implements the redact.SafeFormatter interface. 41 1 : func (q QoSLevel) SafeFormat(p redact.SafePrinter, verb rune) { 42 1 : switch q { 43 1 : case LatencySensitiveQoSLevel: 44 1 : p.Printf("latency") 45 1 : case NonLatencySensitiveQoSLevel: 46 1 : p.Printf("non-latency") 47 0 : default: 48 0 : p.Printf("<unknown-qos>") 49 : } 50 : } 51 : 52 : // StringToQoSForTesting returns the QoSLevel for the string, or panics if the 53 : // string is not known. 54 1 : func StringToQoSForTesting(s string) QoSLevel { 55 1 : switch s { 56 1 : case "latency": 57 1 : return LatencySensitiveQoSLevel 58 1 : case "non-latency": 59 1 : return NonLatencySensitiveQoSLevel 60 : } 61 0 : panic(errors.AssertionFailedf("unknown QoS %s", s)) 62 : } 63 : 64 : // CategoryAndQoS specifies both the Category and the QoSLevel. 65 : type CategoryAndQoS struct { 66 : Category 67 : QoSLevel 68 : } 69 : 70 : // CategoryStats provides stats about a category of reads. 71 : type CategoryStats struct { 72 : // BlockBytes is the bytes in the loaded blocks. If the block was 73 : // compressed, this is the compressed bytes. Currently, only the index 74 : // blocks, data blocks containing points, and filter blocks are included. 75 : // Additionally, value blocks read after the corresponding iterator is 76 : // closed are not included. 77 : BlockBytes uint64 78 : // BlockBytesInCache is the subset of BlockBytes that were in the block 79 : // cache. 80 : BlockBytesInCache uint64 81 : // BlockReadDuration is the total duration to read the bytes not in the 82 : // cache, i.e., BlockBytes-BlockBytesInCache. 83 : BlockReadDuration time.Duration 84 : } 85 : 86 2 : func (s *CategoryStats) aggregate(a CategoryStats) { 87 2 : s.BlockBytes += a.BlockBytes 88 2 : s.BlockBytesInCache += a.BlockBytesInCache 89 2 : s.BlockReadDuration += a.BlockReadDuration 90 2 : } 91 : 92 : // CategoryStatsAggregate is the aggregate for the given category. 93 : type CategoryStatsAggregate struct { 94 : Category 95 : QoSLevel 96 : CategoryStats 97 : } 98 : 99 : type categoryStatsWithMu struct { 100 : mu sync.Mutex 101 : // Protected by mu. 102 : stats CategoryStatsAggregate 103 : } 104 : 105 : // CategoryStatsCollector collects and aggregates the stats per category. 106 : type CategoryStatsCollector struct { 107 : // mu protects additions to statsMap. 108 : mu sync.Mutex 109 : // Category => categoryStatsWithMu. 110 : statsMap sync.Map 111 : } 112 : 113 : func (c *CategoryStatsCollector) reportStats( 114 : category Category, qosLevel QoSLevel, stats CategoryStats, 115 2 : ) { 116 2 : v, ok := c.statsMap.Load(category) 117 2 : if !ok { 118 2 : c.mu.Lock() 119 2 : v, _ = c.statsMap.LoadOrStore(category, &categoryStatsWithMu{ 120 2 : stats: CategoryStatsAggregate{Category: category, QoSLevel: qosLevel}, 121 2 : }) 122 2 : c.mu.Unlock() 123 2 : } 124 2 : aggStats := v.(*categoryStatsWithMu) 125 2 : aggStats.mu.Lock() 126 2 : aggStats.stats.CategoryStats.aggregate(stats) 127 2 : aggStats.mu.Unlock() 128 : } 129 : 130 : // GetStats returns the aggregated stats. 131 2 : func (c *CategoryStatsCollector) GetStats() []CategoryStatsAggregate { 132 2 : var stats []CategoryStatsAggregate 133 2 : c.statsMap.Range(func(_, v any) bool { 134 2 : aggStats := v.(*categoryStatsWithMu) 135 2 : aggStats.mu.Lock() 136 2 : s := aggStats.stats 137 2 : aggStats.mu.Unlock() 138 2 : if len(s.Category) == 0 { 139 2 : s.Category = "_unknown" 140 2 : } 141 2 : stats = append(stats, s) 142 2 : return true 143 : }) 144 2 : slices.SortFunc(stats, func(a, b CategoryStatsAggregate) int { 145 2 : return cmp.Compare(a.Category, b.Category) 146 2 : }) 147 2 : return stats 148 : } 149 : 150 : // iterStatsAccumulator is a helper for a sstable iterator to accumulate 151 : // stats, which are reported to the CategoryStatsCollector when the 152 : // accumulator is closed. 153 : type iterStatsAccumulator struct { 154 : Category 155 : QoSLevel 156 : stats CategoryStats 157 : collector *CategoryStatsCollector 158 : } 159 : 160 : func (accum *iterStatsAccumulator) init( 161 : categoryAndQoS CategoryAndQoS, collector *CategoryStatsCollector, 162 2 : ) { 163 2 : accum.Category = categoryAndQoS.Category 164 2 : accum.QoSLevel = categoryAndQoS.QoSLevel 165 2 : accum.collector = collector 166 2 : } 167 : 168 : func (accum *iterStatsAccumulator) reportStats( 169 : blockBytes, blockBytesInCache uint64, blockReadDuration time.Duration, 170 2 : ) { 171 2 : accum.stats.BlockBytes += blockBytes 172 2 : accum.stats.BlockBytesInCache += blockBytesInCache 173 2 : accum.stats.BlockReadDuration += blockReadDuration 174 2 : } 175 : 176 2 : func (accum *iterStatsAccumulator) close() { 177 2 : if accum.collector != nil { 178 2 : accum.collector.reportStats(accum.Category, accum.QoSLevel, accum.stats) 179 2 : } 180 : }