package hammerhead import ( "errors" "fmt" "net/http" "time" ) // Traffic analytics: edges report per-port bytes, connections and their busiest client addresses with // every report. Hammerhead keeps them at three resolutions (see the *Keep constants in db.go) and draws // charts from them. Organizations can also be alerted when a port suddenly carries far more traffic // than usual, or more than a daily amount. // Spikes are judged on the last few minutes against the hours before them. A port has to have an // hour of history before it can spike, and the burst has to be worth noticing in absolute terms, so // a quiet port going from 1 KB to 50 KB doesn't page anyone. const ( spikeMinHistory = 60 // minutes of baseline needed before a spike can be called spikeMinBytes = 10 * 1024 * 1024 // bytes in the window below which nothing counts as a spike ) // spiking says whether recent traffic is a spike over the baseline for a port. func spiking(recent int64, baseline float64, history int64, factor int) bool { if factor <= 0 || history < spikeMinHistory || recent < spikeMinBytes { return false } return float64(recent) > baseline*float64(factor) } // trafficProblemsLocked adds an alert problem for each port of an organization that is spiking or has // gone over its daily amount. Caller holds s.mu. func (s *Server) trafficProblemsLocked(now time.Time, edgeDown map[string]bool, add func(key, org, kind, name, msg string)) { dayStart := now.UTC().Truncate(24*time.Hour).Unix() / 3600 for _, m := range s.st.Mappings { if !m.Enabled || edgeDown[m.EdgeID] { continue } o := s.findOrg(m.OrgID) if o == nil || (o.TrafficSpikeFactor <= 0 && o.TrafficDailyMB <= 0) { continue } if o.TrafficSpikeFactor > 0 { recent, base, history := s.db.recentTraffic(m.ID, now) if spiking(recent, base, history, o.TrafficSpikeFactor) { add("spike:"+m.ID, m.OrgID, "traffic", m.Name, fmt.Sprintf("%s: %s in the last %d minutes, about %.0f times its usual %s", m.Name, byteText(recent), spikeWindow, float64(recent)/maxFloat(base, 1), byteText(int64(base)))) } } if limit := int64(o.TrafficDailyMB) << 20; limit > 0 { if used := s.db.bytesSince(m.ID, dayStart); used > limit { add("quota:"+m.ID, m.OrgID, "traffic", m.Name, fmt.Sprintf("%s has carried %s today, over the %s a day you set", m.Name, byteText(used), byteText(limit))) } } } } func maxFloat(a, b float64) float64 { if a > b { return a } return b } func byteText(n int64) string { f, units := float64(n), []string{"B", "KB", "MB", "GB", "TB"} i := 0 for f >= 1024 && i < len(units)-1 { f /= 1024 i++ } if i == 0 { return fmt.Sprintf("%d B", n) } return fmt.Sprintf("%.1f %s", f, units[i]) } // trafficRanges maps the ranges the dashboard offers to a bucket size, a bucket count and how many hours of top sources to include. var trafficRanges = map[string]struct { unit string count int hours int bucket time.Duration }{ "1h": {"minute", 60, 1, time.Minute}, "6h": {"minute", 360, 6, time.Minute}, "24h": {"hour", 24, 24, time.Hour}, "7d": {"hour", 7 * 24, 7 * 24, time.Hour}, "30d": {"hour", 30 * 24, 7 * 24, time.Hour}, // client addresses are only kept for a week } // orgTrafficDetail serves one port's traffic chart and busiest clients, or the organization's total // for id "all". ?range= is one of 1h, 6h, 24h (default), 7d or 30d. func (s *Server) orgTrafficDetail(w http.ResponseWriter, r *http.Request, c *orgCtx) { rng := r.URL.Query().Get("range") if rng == "" { rng = "24h" } spec, found := trafficRanges[rng] if !found { fail(w, 400, errors.New("range must be 1h, 6h, 24h, 7d or 30d")) return } id := r.PathValue("id") s.mu.Lock() var ids []string name := "All resources" for _, m := range s.st.Mappings { if m.OrgID != c.Org.ID { continue } if id == "all" { ids = append(ids, m.ID) } else if m.ID == id { ids, name = []string{m.ID}, m.Name } } s.mu.Unlock() if id != "all" && len(ids) == 0 { fail(w, 404, errors.New("no such resource")) return } now := time.Now() points := s.db.trafficSeries(ids, spec.unit, spec.count, now) var tot TrafficPoint var peak int64 for _, p := range points { tot.Conns += p.Conns tot.BytesIn += p.BytesIn tot.BytesOut += p.BytesOut if b := p.BytesIn + p.BytesOut; b > peak { peak = b } } writeJSON(w, 200, map[string]any{ "name": name, "range": rng, "bucketSeconds": int(spec.bucket / time.Second), "points": points, "total": tot, "peakBucket": peak, "sources": s.db.topSources(ids, spec.hours, 10, now), "sourcesHours": spec.hours, }) } // trafficAlertSettings sets when an organization is told about unusual traffic. func (s *Server) trafficAlertSettings(w http.ResponseWriter, r *http.Request, sc alertScope) { var in struct{ SpikeFactor, DailyMB int } if !decode(w, r, &in) { return } if (in.SpikeFactor != 0 && (in.SpikeFactor < 2 || in.SpikeFactor > 1000)) || in.DailyMB < 0 || in.DailyMB > 100_000_000 { fail(w, 400, errors.New("A spike is 2 to 1000 times the usual traffic (0 turns it off), and the daily amount can't be negative (0 turns it off).")) return } s.mu.Lock() defer s.mu.Unlock() o := s.findOrg(sc.org) if o == nil { fail(w, 404, errors.New("organization settings only")) return } o.TrafficSpikeFactor, o.TrafficDailyMB = in.SpikeFactor, in.DailyMB s.save() ok(w) }