Hammerhead/traffic_test.go

package hammerhead

import (
	"strings"
	"testing"
	"time"

	"git.veillax.com/Veillax/reef/model"
)

func TestSpiking(t *testing.T) {
	const mb = 1 << 20
	cases := []struct {
		name    string
		recent  int64
		base    float64
		history int64
		factor  int
		want    bool
	}{
		{"ten times usual", 200 * mb, 10 * mb, 180, 5, true},
		{"just under the factor", 49 * mb, 10 * mb, 180, 5, false},
		{"off", 500 * mb, 10 * mb, 180, 0, false},
		{"too little history", 500 * mb, 10 * mb, 30, 5, false},
		{"tiny in absolute terms", 5 * mb, 0.01 * mb, 180, 5, false},
		{"from nothing to a lot", 50 * mb, 0, 180, 5, true},
	}
	for _, c := range cases {
		if got := spiking(c.recent, c.base, c.history, c.factor); got != c.want {
			t.Errorf("%s: spiking = %v, want %v", c.name, got, c.want)
		}
	}
}

func TestTrafficStorage(t *testing.T) {
	s := newFx(t).s
	now := time.Now()
	st := func(in, out int64, src ...model.SourceStat) model.Stat {
		return model.Stat{Conns: 1, BytesIn: in, BytesOut: out, Sources: src}
	}
	s.db.addTrafficAt("m0", st(100, 900, model.SourceStat{IP: "203.0.113.1", Conns: 1, BytesIn: 100, BytesOut: 900}), now)
	s.db.addTrafficAt("m0", st(10, 10, model.SourceStat{IP: "203.0.113.1", Conns: 1, BytesIn: 10, BytesOut: 10}, model.SourceStat{IP: "203.0.113.2", Conns: 1, BytesIn: 1}), now)
	s.db.addTrafficAt("m0", st(1, 1), now.Add(-3*time.Minute))
	s.db.addTrafficAt("m1", st(7, 7), now)

	pts := s.db.trafficSeries([]string{"m0"}, "minute", 60, now)
	if len(pts) != 60 || pts[59].BytesIn != 110 || pts[59].BytesOut != 910 || pts[59].Conns != 2 || pts[56].BytesIn != 1 {
		t.Fatalf("minute series: last=%+v 3 back=%+v", pts[59], pts[56])
	}
	if pts[59].Time != now.Unix()/60*60 || pts[58].Time != pts[59].Time-60 {
		t.Errorf("bucket times: %d %d", pts[58].Time, pts[59].Time)
	}
	all := s.db.trafficSeries([]string{"m0", "m1"}, "hour", 24, now)
	if all[23].BytesIn+all[22].BytesIn < 118 || len(all) != 24 {
		t.Errorf("hour series over two ports: %+v", all[22:])
	}
	if got := s.db.trafficSeries(nil, "hour", 3, now); len(got) != 3 || got[2].BytesIn != 0 {
		t.Errorf("no ports: %+v", got)
	}

	top := s.db.topSources([]string{"m0"}, 24, 10, now)
	if len(top) != 2 || top[0].IP != "203.0.113.1" || top[0].BytesOut != 910 || top[0].Conns != 2 {
		t.Fatalf("top sources: %+v", top)
	}
	if got := s.db.topSources([]string{"m0"}, 24, 1, now); len(got) != 1 {
		t.Errorf("limit: %+v", got)
	}
	if got := s.db.topSources([]string{"m1"}, 24, 10, now); len(got) != 0 {
		t.Errorf("another port's sources leaked: %+v", got)
	}
}

func TestTrafficRetention(t *testing.T) {
	s := newFx(t).s
	now := time.Now()
	old := model.Stat{Conns: 1, BytesIn: 5, Sources: []model.SourceStat{{IP: "203.0.113.9", Conns: 1, BytesIn: 5}}}
	s.db.addTrafficAt("m0", old, now.Add(-trafficMinutelyKeep-time.Hour)) // too old for minutes, fine for hours and sources
	s.db.addTrafficAt("m0", old, now.Add(-trafficSourcesKeep-time.Hour))  // too old for sources too
	s.db.addTrafficAt("m0", old, now.Add(-trafficHourlyKeep-time.Hour))   // gone everywhere
	s.db.addTrafficAt("m0", old, now)
	s.db.purge(30, 365)
	count := func(table string) (n int) {
		s.db.sql.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&n)
		return
	}
	if n := count("traffic_minutely"); n != 1 {
		t.Errorf("minutely rows after purge: %d, want 1", n)
	}
	if n := count("traffic_sources"); n != 2 {
		t.Errorf("source rows after purge: %d, want 2", n)
	}
	if n := count("traffic_hourly"); n != 3 {
		t.Errorf("hourly rows after purge: %d, want 3", n)
	}
}

func TestRecentTraffic(t *testing.T) {
	s := newFx(t).s
	now := time.Now()
	const mb = 1 << 20
	for h := 3 * 60; h > spikeWindow; h-- { // three hours of 1 MB a minute
		s.db.addTrafficAt("m0", model.Stat{BytesIn: mb}, now.Add(-time.Duration(h)*time.Minute))
	}
	s.db.addTrafficAt("m0", model.Stat{BytesIn: 100 * mb}, now)
	recent, base, hist := s.db.recentTraffic("m0", now)
	if recent != 100*mb || hist < 170 || base < 4.9*mb || base > 5.1*mb { // 1 MB a minute is 5 MB per window
		t.Fatalf("recent=%d base=%.0f history=%d", recent, base, hist)
	}
	if !spiking(recent, base, hist, 5) {
		t.Error("100 MB over a 5 MB norm should spike")
	}
	if r, b, h := s.db.recentTraffic("nothing", now); r != 0 || b != 0 || h != 0 {
		t.Errorf("unknown port: %d %f %d", r, b, h)
	}
}

func TestTrafficAlertsFireAndClear(t *testing.T) {
	f := newFx(t)
	s := f.s
	const mb = 1 << 20
	now := time.Now()
	s.mu.Lock()
	m := s.st.Mappings[0]
	o := s.findOrg(m.OrgID)
	o.TrafficSpikeFactor, o.TrafficDailyMB = 4, 500
	id := m.ID
	s.mu.Unlock()
	for min := 3 * 60; min > spikeWindow; min-- {
		s.db.addTrafficAt(id, model.Stat{BytesIn: mb}, now.Add(-time.Duration(min)*time.Minute))
	}

	problems := func() map[string]problem {
		s.mu.Lock()
		defer s.mu.Unlock()
		return s.problemsLocked(now)
	}
	if p := problems(); p["spike:"+id].msg != "" || p["quota:"+id].msg != "" {
		t.Fatalf("steady traffic raised a problem: %+v", p)
	}
	s.db.addTrafficAt(id, model.Stat{BytesOut: 300 * mb}, now)
	p := problems()
	if !strings.Contains(p["spike:"+id].msg, "usual") || p["spike:"+id].kind != "traffic" {
		t.Fatalf("no spike problem: %+v", p)
	}
	if p["quota:"+id].msg != "" {
		t.Errorf("under the daily amount, yet: %+v", p["quota:"+id])
	}
	s.db.addTrafficAt(id, model.Stat{BytesOut: 300 * mb}, now)
	if q := problems()["quota:"+id]; !strings.Contains(q.msg, "over the") {
		t.Fatalf("no quota problem: %+v", q)
	}

	s.mu.Lock()
	o.TrafficSpikeFactor, o.TrafficDailyMB = 0, 0
	s.mu.Unlock()
	if p := problems(); p["spike:"+id].msg != "" || p["quota:"+id].msg != "" {
		t.Errorf("alerts off, yet: %+v", p)
	}
}