Compensate NetFlow/IPFIX export loss from the start of a session

internal/decode/nf/decoder.go

@@ -116,6 +116,7 @@ type session struct {
 	winRecv, winLost float64
 	winAt            time.Time
 	comp             float64
+	compSet          bool // comp has been computed at least once
 }
 
 func newSession(k key) *session {
@@ -457,6 +458,9 @@ func (s *session) seq(seq uint32, unit uint32, recs uint32, now time.Time) {
 	s.updateComp(float64(recs), lostRecs, now)
 }
 
+// compWindow is how often the loss compensation multiplier is refreshed.
+const compWindow = 5 * time.Second
+
 // updateComp keeps a decaying window of received and lost records and
 // derives the multiplier that restores the lost share of the totals.
 func (s *session) updateComp(recv, lost float64, now time.Time) {
@@ -468,10 +472,15 @@ func (s *session) updateComp(recv, lost float64, now time.Time) {
 	if s.winAt.IsZero() {
 		s.winAt = now
 	}
-	if now.Sub(s.winAt) < 30*time.Second {
+	// The first multiplier is computed as soon as 100 records have been
+	// seen, so a session that starts out lossy is compensated from its
+	// first seconds; after that the window is refreshed every 5 s.
+	first := !s.compSet && s.winRecv+s.winLost >= 100 && s.winRecv > 0
+	if !first && now.Sub(s.winAt) < compWindow {
 		return
 	}
-	if s.winRecv >= 100 {
+	if first || s.winRecv >= 100 {
+		s.compSet = true
 		c := (s.winRecv + s.winLost) / s.winRecv
 		if c > 2 { // more than half missing: do not guess
 			c = 2

internal/decode/nf/loss_test.go

@@ -2,6 +2,7 @@ package nf
 
 import (
 	"math"
+	"math/rand"
 	"testing"
 	"time"
 
@@ -104,3 +105,54 @@ func TestIPFIXReorderIsNotLoss(t *testing.T) {
 		t.Fatalf("loss multiplier %.3f for a lossless stream", st.LossComp)
 	}
 }
+
+// lossyRun sends dur worth of IPFIX messages (one every 10 ms, 10 to 30
+// records each), loses each message with probability p, and returns the
+// Mult-scaled bytes over the whole run divided by the bytes sent.
+func lossyRun(t *testing.T, dur time.Duration, p float64) float64 {
+	t.Helper()
+	d := NewDecoder()
+	d.Unsampled[exp] = true
+	rng := rand.New(rand.NewSource(1))
+	now := base
+	seq := uint32(0)
+	var sent, scaled float64
+	for at := time.Duration(0); at < dur; at += 10 * time.Millisecond {
+		m := encode.NewIPFIX()
+		m.AddTemplate(ipfixData)
+		n := 10 + rng.Intn(21)
+		recs := make([]encode.Values, n)
+		for j := range recs {
+			recs[j] = ipfixRec(1000)
+		}
+		m.AddData(ipfixData.ID, recs)
+		b := m.IPFIX(uint32(now.Unix()), seq, 1)
+		seq += uint32(m.DataRecords())
+		sent += float64(n) * 1000
+		if rng.Float64() >= p {
+			res, err := d.Decode(b, exp, now.Add(at))
+			if err != nil {
+				t.Fatal(err)
+			}
+			for _, r := range res.Records {
+				scaled += float64(r.Bytes) * r.Mult
+			}
+		}
+	}
+	return scaled / sent
+}
+
+// A session that is lossy from the start must be compensated from its
+// first seconds, not only after a long warm-up window.
+func TestLossCompensatedFromStart(t *testing.T) {
+	for _, dur := range []time.Duration{30 * time.Second, 60 * time.Second, 120 * time.Second, 300 * time.Second} {
+		r := lossyRun(t, dur, 0.37)
+		t.Logf("37%% loss, %v: compensated/true = %.3f", dur, r)
+		if dur == 60*time.Second && r < 0.95 {
+			t.Fatalf("60 s at 37%% loss: compensated/true %.3f, want >= 0.95", r)
+		}
+	}
+	if r := lossyRun(t, 60*time.Second, 0); math.Abs(r-1) > 1e-9 {
+		t.Fatalf("lossless run: compensated/true %.6f, want 1", r)
+	}
+}