add stats

Author: jamiesun

Corefile

@@ -1,4 +1,4 @@
-. {
+.:5333 {
     datahub {
         debug
         geoip_path data/geoip.dat

go.mod

@@ -3,18 +3,18 @@ module github.com/ca17/datahub
 go 1.16
 
 require (
+	github.com/ahmetb/go-linq v3.0.0+incompatible
 	github.com/allegro/bigcache v1.2.1
 	github.com/c-robinson/iplib v1.0.3
 	github.com/ca17/dnssrc v0.0.2
 	github.com/coredns/caddy v1.1.1
 	github.com/coredns/coredns v1.8.6
 	github.com/dnstap/golang-dnstap v0.4.0
 	github.com/form3tech-oss/jwt-go v3.2.3+incompatible
-	github.com/go-ozzo/ozzo-routing v2.1.4+incompatible // indirect
-	github.com/golang/gddo v0.0.0-20210115222349-20d68f94ee1f // indirect
 	github.com/golang/protobuf v1.5.2
+	github.com/metaslink/metasdns v0.0.0-20211230190103-80ba25a91d75
 	github.com/miekg/dns v1.1.43
-	github.com/orcaman/concurrent-map v0.0.0-20210501183033-44dafcb38ecc
+	github.com/orcaman/concurrent-map v1.0.0
 	github.com/qiangxue/fasthttp-routing v0.0.0-20160225050629-6ccdc2a18d87
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/valyala/fasthttp v1.31.0

go.sum

@@ -0,0 +1,77 @@
+package stats
+
+import (
+	"sort"
+	"sync"
+	"time"
+
+	_ "github.com/ahmetb/go-linq"
+)
+
+type DayDnsStat struct {
+	sync.RWMutex
+	LastRolling time.Time
+	Interval    time.Duration
+	Values      map[int64]map[string]int64
+}
+
+func NewDayDnsStat(ival time.Duration) *DayDnsStat {
+	return &DayDnsStat{LastRolling: time.Now(), Interval: ival, Values: map[int64]map[string]int64{}}
+}
+
+// Rolling 滚动清除老数据
+func (d *DayDnsStat) Rolling() {
+	d.Lock()
+	defer d.Unlock()
+	d.LastRolling = time.Now()
+	keys := make([]int64, 0)
+	for _k, _ := range d.Values {
+		keys = append(keys, _k)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		return keys[i] < keys[j]
+	})
+	s := time.Now().Add(-d.Interval).Unix() * 1000
+	for _, key := range keys {
+		if s > key {
+			d.Values[key] = nil
+			delete(d.Values, key)
+		}
+	}
+}
+
+func (d *DayDnsStat) Update(c *CounterStat) {
+	if time.Now().Sub(d.LastRolling).Seconds() > 120 {
+		d.Rolling()
+	}
+	v := c.MapValues()
+	d.Lock()
+	defer d.Unlock()
+	if len(v) > 0 {
+		d.Values[time.Now().Unix()*1000] = v
+	}
+}
+
+func (d *DayDnsStat) LineChartData(name string) *LineChartData {
+	d.RLock()
+	defer d.RUnlock()
+	keys := make([]int64, 0)
+	for _k, _ := range d.Values {
+		keys = append(keys, _k)
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		return keys[i] < keys[j]
+	})
+	ld := NewLineChartData(name, "")
+	for _, key := range keys {
+		ld.Times = append(ld.Times, time.UnixMilli(key).Format(time.RFC3339))
+		for ik, iv := range d.Values[key] {
+			if _, ok := ld.Datas[ik]; !ok {
+				ld.Datas[ik] = make([]int64, 0)
+			}
+			ld.Datas[ik] = append(ld.Datas[ik], iv)
+		}
+	}
+
+	return ld
+}

plugin/pkg/stats/dns.go

@@ -0,0 +1,27 @@
+package stats
+
+import (
+	"testing"
+	"time"
+
+	"github.com/metaslink/metasdns/plugin/pkg/common"
+)
+
+func TestNewDayDnsStat(t *testing.T) {
+	ds := NewDayDnsStat(time.Second * 15)
+	c := NewCounterStat()
+	start := time.Now()
+	for {
+		if time.Now().Sub(start).Seconds() > 12 {
+			break
+		}
+		c.Incr("cn", 1)
+		time.Sleep(time.Second * 3)
+		c.Incr("cn", 1)
+		c.Incr("ads", 1)
+		t.Log("...")
+		ds.Update(c)
+	}
+	ds.Rolling()
+	t.Log(common.ToJson(ds.LineChartData("最近24小时统计").ChartData()))
+}

plugin/pkg/stats/dns_test.go

@@ -0,0 +1,39 @@
+package stats
+
+import (
+	"github.com/ahmetb/go-linq"
+)
+
+type LineChartData struct {
+	Title     string             `json:"title"`
+	YaxisName string             `json:"yaxis_name"`
+	Times     []string           `json:"times"`
+	Datas     map[string][]int64 `json:"datas"`
+}
+
+func NewLineChartData(title string, yaxisName string) *LineChartData {
+	return &LineChartData{Title: title, YaxisName: yaxisName, Times: make([]string, 0), Datas: make(map[string][]int64, 0)}
+}
+
+func (l *LineChartData) ChartData() *LineChartData {
+	linq.From(l.Datas).ForEachT(func(kv linq.KeyValue) {
+		var last int64 = 0
+		vals := make([]int64, 0)
+		linq.From(kv.Value).ForEachT(func(v int64) {
+			var val int64 = 0
+			if last == 0 {
+				val = 0
+				last = v
+			} else {
+				val = v - last
+				if val < 0 {
+					val = 0
+				}
+				last = v
+			}
+			vals = append(vals, val)
+		})
+		l.Datas[kv.Key.(string)] = vals
+	})
+	return l
+}

plugin/pkg/stats/line.go

@@ -0,0 +1,82 @@
+package stats
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+type Metrics struct {
+	Icon  string
+	Value interface{}
+	Title string
+}
+
+func NewMetrics(icon string, value interface{}, title string) *Metrics {
+	return &Metrics{Icon: icon, Value: value, Title: title}
+}
+
+type Counter struct {
+	Name  string `json:"name"`
+	Value int64  `json:"value"`
+}
+
+func NewCounter(name string, value int64) *Counter {
+	return &Counter{Name: name, Value: value}
+}
+
+func (d *Counter) Incr(v int64) {
+	atomic.AddInt64(&d.Value, v)
+}
+
+func (d *Counter) Val() int64 {
+	return d.Value
+}
+
+type CounterStat struct {
+	sync.RWMutex
+	cmap map[string]*Counter
+}
+
+func NewCounterStat() *CounterStat {
+	return &CounterStat{cmap: make(map[string]*Counter)}
+}
+
+func (c *CounterStat) Incr(name string, v int64) {
+	c.Lock()
+	defer c.Unlock()
+	if ct, ok := c.cmap[name]; ok {
+		ct.Incr(v)
+	} else {
+		c.cmap[name] = NewCounter(name, v)
+	}
+}
+
+func (c *CounterStat) MapValues() map[string]int64 {
+	mv := make(map[string]int64)
+	c.RLock()
+	defer c.RUnlock()
+	for k, counter := range c.cmap {
+		mv[k] = counter.Val()
+	}
+	return mv
+}
+
+func (c *CounterStat) Values() []Counter {
+	c.RLock()
+	defer c.RUnlock()
+	var result = make([]Counter, 0)
+	for _, v := range c.cmap {
+		result = append(result, *v)
+	}
+	return result
+}
+
+func (c *CounterStat) GetValue(name string) int64 {
+	c.RLock()
+	defer c.RUnlock()
+	if ct, ok := c.cmap[name]; ok {
+		return ct.Val()
+	} else {
+		return 0
+	}
+}