65 lines
1.4 KiB
Go
65 lines
1.4 KiB
Go
|
package tsdb
|
||
|
|
|
||
|
|
import (
|
||
|
|
"sort"
|
||
|
|
"sync"
|
||
|
|
"sync/atomic"
|
||
|
|
)
|
||
|
|
|
||
|
|
type MetricKey struct {
|
||
|
|
measurement string
|
||
|
|
db string
|
||
|
|
rp string
|
||
|
|
// TODO: add login
|
||
|
|
}
|
||
|
|
|
||
|
|
type MetricValue struct {
|
||
|
|
points int64
|
||
|
|
values int64
|
||
|
|
}
|
||
|
|
|
||
|
|
type IngressMetrics struct {
|
||
|
|
m sync.Map
|
||
|
|
length int64
|
||
|
|
}
|
||
|
|
|
||
|
|
func (i *IngressMetrics) AddMetric(measurement, db, rp string, points, values int64) {
|
||
|
|
key := MetricKey{measurement, db, rp}
|
||
|
|
val, ok := i.m.Load(key)
|
||
|
|
if !ok {
|
||
|
|
var loaded bool
|
||
|
|
val, loaded = i.m.LoadOrStore(key, &MetricValue{})
|
||
|
|
if !loaded {
|
||
|
|
atomic.AddInt64(&i.length, 1)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
metricVal := val.(*MetricValue)
|
||
|
|
atomic.AddInt64(&metricVal.points, points)
|
||
|
|
atomic.AddInt64(&metricVal.values, values)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (i *IngressMetrics) ForEach(f func(m MetricKey, points, values int64)) {
|
||
|
|
keys := make([]MetricKey, 0, atomic.LoadInt64(&i.length))
|
||
|
|
i.m.Range(func(key, value interface{}) bool {
|
||
|
|
keys = append(keys, key.(MetricKey))
|
||
|
|
return true
|
||
|
|
})
|
||
|
|
sort.Slice(keys, func(i, j int) bool {
|
||
|
|
if keys[i].db != keys[j].db {
|
||
|
|
return keys[i].db < keys[j].db
|
||
|
|
}
|
||
|
|
if keys[i].rp != keys[j].rp {
|
||
|
|
return keys[i].rp < keys[j].rp
|
||
|
|
}
|
||
|
|
return keys[i].measurement < keys[j].measurement
|
||
|
|
//TODO: sort based on login
|
||
|
|
})
|
||
|
|
for _, key := range keys {
|
||
|
|
val, ok := i.m.Load(key)
|
||
|
|
// ok should always be true - we don't delete keys. But if we did we would ignore keys concurrently deleted.
|
||
|
|
if ok {
|
||
|
|
f(key, val.(*MetricValue).points, val.(*MetricValue).values)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|