1package stats 2 3import ( 4 "crypto/md5" 5 "fmt" 6 "sort" 7 "sync" 8 "time" 9 10 "github.com/montanaflynn/stats" 11 "go.mongodb.org/mongo-driver/bson" 12 13 "github.com/percona/percona-toolkit/src/go/mongolib/proto" 14) 15 16type StatsError struct { 17 error 18} 19 20func (e *StatsError) Error() string { 21 if e == nil { 22 return "<nil>" 23 } 24 25 return fmt.Sprintf("stats error: %s", e.error) 26} 27 28func (e *StatsError) Parent() error { 29 return e.error 30} 31 32type StatsFingerprintError StatsError 33 34// New creates new instance of stats with given Fingerprinter 35func New(fingerprinter Fingerprinter) *Stats { 36 s := &Stats{ 37 fingerprinter: fingerprinter, 38 } 39 40 s.Reset() 41 return s 42} 43 44// Stats is a collection of MongoDB statistics 45type Stats struct { 46 // dependencies 47 fingerprinter Fingerprinter 48 49 // internal 50 queryInfoAndCounters map[GroupKey]*QueryInfoAndCounters 51 sync.RWMutex 52} 53 54// Reset clears the collection of statistics 55func (s *Stats) Reset() { 56 s.Lock() 57 defer s.Unlock() 58 59 s.queryInfoAndCounters = make(map[GroupKey]*QueryInfoAndCounters) 60} 61 62// Add adds proto.SystemProfile to the collection of statistics 63func (s *Stats) Add(doc proto.SystemProfile) error { 64 fp, err := s.fingerprinter.Fingerprint(doc) 65 if err != nil { 66 return &StatsFingerprintError{err} 67 } 68 var qiac *QueryInfoAndCounters 69 var ok bool 70 71 key := GroupKey{ 72 Operation: fp.Operation, 73 Fingerprint: fp.Fingerprint, 74 Namespace: fp.Namespace, 75 } 76 if qiac, ok = s.getQueryInfoAndCounters(key); !ok { 77 query := proto.NewExampleQuery(doc) 78 queryBson, err := bson.MarshalExtJSON(query, true, true) 79 if err != nil { 80 return err 81 } 82 qiac = &QueryInfoAndCounters{ 83 ID: fmt.Sprintf("%x", md5.Sum([]byte(fmt.Sprintf("%s", key)))), 84 Operation: fp.Operation, 85 Fingerprint: fp.Fingerprint, 86 Namespace: fp.Namespace, 87 TableScan: false, 88 Query: string(queryBson), 89 } 90 s.setQueryInfoAndCounters(key, qiac) 91 } 92 qiac.Count++ 93 // docsExamined is renamed from nscannedObjects in 3.2.0. 94 // https://docs.mongodb.com/manual/reference/database-profiler/#system.profile.docsExamined 95 s.Lock() 96 if doc.NscannedObjects > 0 { 97 qiac.NScanned = append(qiac.NScanned, float64(doc.NscannedObjects)) 98 } else { 99 qiac.NScanned = append(qiac.NScanned, float64(doc.DocsExamined)) 100 } 101 qiac.NReturned = append(qiac.NReturned, float64(doc.Nreturned)) 102 qiac.QueryTime = append(qiac.QueryTime, float64(doc.Millis)) 103 qiac.ResponseLength = append(qiac.ResponseLength, float64(doc.ResponseLength)) 104 if qiac.FirstSeen.IsZero() || qiac.FirstSeen.After(doc.Ts) { 105 qiac.FirstSeen = doc.Ts 106 } 107 if qiac.LastSeen.IsZero() || qiac.LastSeen.Before(doc.Ts) { 108 qiac.LastSeen = doc.Ts 109 } 110 s.Unlock() 111 112 return nil 113} 114 115// Queries returns all collected statistics 116func (s *Stats) Queries() Queries { 117 s.Lock() 118 defer s.Unlock() 119 120 keys := GroupKeys{} 121 for key := range s.queryInfoAndCounters { 122 keys = append(keys, key) 123 } 124 sort.Sort(keys) 125 126 queries := []QueryInfoAndCounters{} 127 for _, key := range keys { 128 queries = append(queries, *s.queryInfoAndCounters[key]) 129 } 130 return queries 131} 132 133func (s *Stats) getQueryInfoAndCounters(key GroupKey) (*QueryInfoAndCounters, bool) { 134 s.RLock() 135 defer s.RUnlock() 136 137 v, ok := s.queryInfoAndCounters[key] 138 return v, ok 139} 140 141func (s *Stats) setQueryInfoAndCounters(key GroupKey, value *QueryInfoAndCounters) { 142 s.Lock() 143 defer s.Unlock() 144 145 s.queryInfoAndCounters[key] = value 146} 147 148// Queries is a slice of MongoDB statistics 149type Queries []QueryInfoAndCounters 150 151// CalcQueriesStats calculates QueryStats for given uptime 152func (q Queries) CalcQueriesStats(uptime int64) []QueryStats { 153 qs := []QueryStats{} 154 tc := calcTotalCounters(q) 155 156 for _, query := range q { 157 queryStats := countersToStats(query, uptime, tc) 158 qs = append(qs, queryStats) 159 } 160 161 return qs 162} 163 164// CalcTotalQueriesStats calculates total QueryStats for given uptime 165func (q Queries) CalcTotalQueriesStats(uptime int64) QueryStats { 166 tc := calcTotalCounters(q) 167 168 totalQueryInfoAndCounters := aggregateCounters(q) 169 totalStats := countersToStats(totalQueryInfoAndCounters, uptime, tc) 170 171 return totalStats 172} 173 174type QueryInfoAndCounters struct { 175 ID string 176 Namespace string 177 Operation string 178 Query string 179 Fingerprint string 180 FirstSeen time.Time 181 LastSeen time.Time 182 TableScan bool 183 184 Count int 185 BlockedTime Times 186 LockTime Times 187 NReturned []float64 188 NScanned []float64 189 QueryTime []float64 // in milliseconds 190 ResponseLength []float64 191} 192 193// times is an array of time.Time that implements the Sorter interface 194type Times []time.Time 195 196func (a Times) Len() int { return len(a) } 197func (a Times) Swap(i, j int) { a[i], a[j] = a[j], a[i] } 198func (a Times) Less(i, j int) bool { return a[i].Before(a[j]) } 199 200type GroupKeys []GroupKey 201 202func (a GroupKeys) Len() int { return len(a) } 203func (a GroupKeys) Swap(i, j int) { a[i], a[j] = a[j], a[i] } 204func (a GroupKeys) Less(i, j int) bool { return a[i].String() < a[j].String() } 205 206type GroupKey struct { 207 Operation string 208 Namespace string 209 Fingerprint string 210} 211 212func (g GroupKey) String() string { 213 return g.Operation + g.Namespace + g.Fingerprint 214} 215 216type totalCounters struct { 217 Count int 218 Scanned float64 219 Returned float64 220 QueryTime float64 221 Bytes float64 222} 223 224type QueryStats struct { 225 ID string 226 Namespace string 227 Operation string 228 Query string 229 Fingerprint string 230 FirstSeen time.Time 231 LastSeen time.Time 232 233 Count int 234 QPS float64 235 Rank int 236 Ratio float64 237 QueryTime Statistics 238 ResponseLength Statistics 239 Returned Statistics 240 Scanned Statistics 241} 242 243type Statistics struct { 244 Pct float64 245 Total float64 246 Min float64 247 Max float64 248 Avg float64 249 Pct95 float64 250 Pct99 float64 251 StdDev float64 252 Median float64 253} 254 255func countersToStats(query QueryInfoAndCounters, uptime int64, tc totalCounters) QueryStats { 256 queryStats := QueryStats{ 257 Count: query.Count, 258 ID: query.ID, 259 Operation: query.Operation, 260 Query: query.Query, 261 Fingerprint: query.Fingerprint, 262 Scanned: calcStats(query.NScanned), 263 Returned: calcStats(query.NReturned), 264 QueryTime: calcStats(query.QueryTime), 265 ResponseLength: calcStats(query.ResponseLength), 266 FirstSeen: query.FirstSeen, 267 LastSeen: query.LastSeen, 268 Namespace: query.Namespace, 269 QPS: float64(query.Count) / float64(uptime), 270 } 271 if tc.Scanned > 0 { 272 queryStats.Scanned.Pct = queryStats.Scanned.Total * 100 / tc.Scanned 273 } 274 if tc.Returned > 0 { 275 queryStats.Returned.Pct = queryStats.Returned.Total * 100 / tc.Returned 276 } 277 if tc.QueryTime > 0 { 278 queryStats.QueryTime.Pct = queryStats.QueryTime.Total * 100 / tc.QueryTime 279 } 280 if tc.Bytes > 0 { 281 queryStats.ResponseLength.Pct = queryStats.ResponseLength.Total * 100 / tc.Bytes 282 } 283 if queryStats.Returned.Total > 0 { 284 queryStats.Ratio = queryStats.Scanned.Total / queryStats.Returned.Total 285 } 286 287 return queryStats 288} 289 290func aggregateCounters(queries []QueryInfoAndCounters) QueryInfoAndCounters { 291 qt := QueryInfoAndCounters{} 292 for _, query := range queries { 293 qt.Count += query.Count 294 qt.NScanned = append(qt.NScanned, query.NScanned...) 295 qt.NReturned = append(qt.NReturned, query.NReturned...) 296 qt.QueryTime = append(qt.QueryTime, query.QueryTime...) 297 qt.ResponseLength = append(qt.ResponseLength, query.ResponseLength...) 298 } 299 return qt 300} 301 302func calcTotalCounters(queries []QueryInfoAndCounters) totalCounters { 303 tc := totalCounters{} 304 305 for _, query := range queries { 306 tc.Count += query.Count 307 308 scanned, _ := stats.Sum(query.NScanned) 309 tc.Scanned += scanned 310 311 returned, _ := stats.Sum(query.NReturned) 312 tc.Returned += returned 313 314 queryTime, _ := stats.Sum(query.QueryTime) 315 tc.QueryTime += queryTime 316 317 bytes, _ := stats.Sum(query.ResponseLength) 318 tc.Bytes += bytes 319 } 320 return tc 321} 322 323func calcStats(samples []float64) Statistics { 324 var s Statistics 325 s.Total, _ = stats.Sum(samples) 326 s.Min, _ = stats.Min(samples) 327 s.Max, _ = stats.Max(samples) 328 s.Avg, _ = stats.Mean(samples) 329 s.Pct95, _ = stats.PercentileNearestRank(samples, 95) 330 s.Pct99, _ = stats.PercentileNearestRank(samples, 99) 331 s.StdDev, _ = stats.StandardDeviation(samples) 332 s.Median, _ = stats.Median(samples) 333 return s 334} 335