103 lines
3.0 KiB
Go
103 lines
3.0 KiB
Go
package blacktip
|
|
|
|
import (
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.veillax.com/Veillax/reef/model"
|
|
)
|
|
|
|
// Delivery outcomes: every mail transaction the egress handles ends delivered,
|
|
// deferred (a 4xx the sender will retry) or failed (a 5xx). They are counted in
|
|
// hourly buckets for the last day, so the dashboard can show a delivery rate
|
|
// and say which recipient domains are refusing mail and why.
|
|
|
|
const (
|
|
outcomeWindow = 24 * time.Hour
|
|
outcomeRecent = 10 // refusals kept for the dashboard to explain
|
|
)
|
|
|
|
type outcomeBucket struct {
|
|
hour time.Time
|
|
byDomain map[string]model.OutcomeCount
|
|
}
|
|
|
|
type outcomeLog struct {
|
|
mu sync.Mutex
|
|
buckets map[string][]*outcomeBucket // by mail-egress ID, oldest first
|
|
recent map[string][]model.OutcomeFailure
|
|
}
|
|
|
|
// recordOutcome counts one finished transaction (or a refusal along the way).
|
|
func (b *Blacktip) recordOutcome(egress, domain string, code int, text string) {
|
|
b.outcomes.add(egress, domain, code, text, time.Now())
|
|
}
|
|
|
|
func (o *outcomeLog) add(egress, domain string, code int, text string, now time.Time) {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
if o.buckets == nil {
|
|
o.buckets, o.recent = map[string][]*outcomeBucket{}, map[string][]model.OutcomeFailure{}
|
|
}
|
|
hour := now.Truncate(time.Hour)
|
|
bs := o.buckets[egress]
|
|
if len(bs) == 0 || !bs[len(bs)-1].hour.Equal(hour) {
|
|
bs = append(bs, &outcomeBucket{hour: hour, byDomain: map[string]model.OutcomeCount{}})
|
|
}
|
|
cut := 0
|
|
for cut < len(bs) && now.Sub(bs[cut].hour) >= outcomeWindow {
|
|
cut++
|
|
}
|
|
bs = bs[cut:]
|
|
o.buckets[egress] = bs
|
|
c := bs[len(bs)-1].byDomain[domain]
|
|
switch {
|
|
case code >= 200 && code < 300:
|
|
c.Delivered++
|
|
case code >= 400 && code < 500:
|
|
c.Deferred++
|
|
default:
|
|
c.Failed++
|
|
}
|
|
bs[len(bs)-1].byDomain[domain] = c
|
|
if code >= 300 {
|
|
r := append([]model.OutcomeFailure{{Time: now, Domain: domain, Code: code, Text: text}}, o.recent[egress]...)
|
|
if len(r) > outcomeRecent {
|
|
r = r[:outcomeRecent]
|
|
}
|
|
o.recent[egress] = r
|
|
}
|
|
}
|
|
|
|
// summary totals the last day for one egress, or nil if it handled no mail.
|
|
func (o *outcomeLog) summary(egress string, now time.Time) *model.MailOutcomes {
|
|
o.mu.Lock()
|
|
defer o.mu.Unlock()
|
|
out := &model.MailOutcomes{ByDomain: map[string]model.OutcomeCount{}}
|
|
for _, bk := range o.buckets[egress] {
|
|
if now.Sub(bk.hour) >= outcomeWindow {
|
|
continue
|
|
}
|
|
if out.Since.IsZero() || bk.hour.Before(out.Since) {
|
|
out.Since = bk.hour
|
|
}
|
|
for d, c := range bk.byDomain {
|
|
t := out.ByDomain[d]
|
|
t.Delivered, t.Deferred, t.Failed = t.Delivered+c.Delivered, t.Deferred+c.Deferred, t.Failed+c.Failed
|
|
out.ByDomain[d] = t
|
|
out.Delivered, out.Deferred, out.Failed = out.Delivered+c.Delivered, out.Deferred+c.Deferred, out.Failed+c.Failed
|
|
}
|
|
}
|
|
if out.Delivered+out.Deferred+out.Failed == 0 {
|
|
return nil
|
|
}
|
|
for _, f := range o.recent[egress] {
|
|
if now.Sub(f.Time) < outcomeWindow {
|
|
out.Recent = append(out.Recent, f)
|
|
}
|
|
}
|
|
sort.SliceStable(out.Recent, func(i, j int) bool { return out.Recent[i].Time.After(out.Recent[j].Time) })
|
|
return out
|
|
}
|