diff --git a/aggregate/window.go b/aggregate/window.go index bf7893d..32d944a 100644 --- a/aggregate/window.go +++ b/aggregate/window.go @@ -6,6 +6,70 @@ import ( "github.com/netsynth/netsynth/classify" ) +// AggregatePcap reads all ClassifiedPackets (with Timestamp set), assigns to time windows +// using packet timestamps per D-01, fills gap windows with empty snapshots per D-02, +// and returns all WindowSnapshots. Calls onSnapshot for each if non-nil (D-07: --verbose). +// Returns synchronously after channel closes (pcap processing is finite). +func AggregatePcap(events <-chan classify.ClassifiedPacket, windowMs int, + onSnapshot func(classify.WindowSnapshot)) []classify.WindowSnapshot { + + // Drain the events channel into a slice. + var all []classify.ClassifiedPacket + for ev := range events { + all = append(all, ev) + } + if len(all) == 0 { + return nil + } + + // Find min and max timestamps (handle non-monotonic pcaps per Pitfall 5). + minTS := all[0].Timestamp + maxTS := all[0].Timestamp + for _, ev := range all[1:] { + if ev.Timestamp.Before(minTS) { + minTS = ev.Timestamp + } + if ev.Timestamp.After(maxTS) { + maxTS = ev.Timestamp + } + } + + // Calculate the maximum window index. + maxWindowIdx := int(maxTS.Sub(minTS).Milliseconds()) / windowMs + + // Create all snapshots including gaps (D-02: gaps are silent). + snapshots := make([]classify.WindowSnapshot, maxWindowIdx+1) + for i := range snapshots { + snapshots[i] = classify.WindowSnapshot{ + Counts: make(map[classify.TrafficClass]int64), + WindowIndex: i, + } + } + + // Assign each packet to its window. + for _, ev := range all { + idx := int(ev.Timestamp.Sub(minTS).Milliseconds()) / windowMs + // Clamp to [0, maxWindowIdx] for safety. + if idx < 0 { + idx = 0 + } + if idx > maxWindowIdx { + idx = maxWindowIdx + } + snapshots[idx].Counts[ev.Class]++ + snapshots[idx].TotalPackets++ + } + + // Fire onSnapshot callback for each snapshot in order (D-07: --verbose). + if onSnapshot != nil { + for _, s := range snapshots { + onSnapshot(s) + } + } + + return snapshots +} + // DefaultWindowMs is the default time window duration in milliseconds. const DefaultWindowMs = 500 diff --git a/aggregate/window_test.go b/aggregate/window_test.go index 0d03496..3007e63 100644 --- a/aggregate/window_test.go +++ b/aggregate/window_test.go @@ -156,3 +156,151 @@ func TestAggregateWindowIndex(t *testing.T) { } } } + +// --- AggregatePcap tests --- + +// baseTime is a fixed reference time for pcap aggregation tests. +var baseTime = time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + +// makePcapChan creates a buffered channel, sends the given packets, closes it, and returns it. +func makePcapChan(pkts []classify.ClassifiedPacket) <-chan classify.ClassifiedPacket { + ch := make(chan classify.ClassifiedPacket, len(pkts)+1) + for _, p := range pkts { + ch <- p + } + close(ch) + return ch +} + +// TestAggregatePcapBasic verifies 3 packets in window 0 produce 1 snapshot with TotalPackets=3. +func TestAggregatePcapBasic(t *testing.T) { + pkts := []classify.ClassifiedPacket{ + {Class: classify.ClassICMP, Timestamp: baseTime}, + {Class: classify.ClassICMP, Timestamp: baseTime.Add(100 * time.Millisecond)}, + {Class: classify.ClassICMP, Timestamp: baseTime.Add(200 * time.Millisecond)}, + } + snaps := AggregatePcap(makePcapChan(pkts), 500, nil) + if len(snaps) != 1 { + t.Fatalf("expected 1 snapshot, got %d", len(snaps)) + } + if snaps[0].TotalPackets != 3 { + t.Errorf("expected TotalPackets=3, got %d", snaps[0].TotalPackets) + } +} + +// TestAggregatePcapMultipleWindows verifies 3 packets spanning 3 windows each produce 1 packet. +func TestAggregatePcapMultipleWindows(t *testing.T) { + pkts := []classify.ClassifiedPacket{ + {Class: classify.ClassDNS, Timestamp: baseTime}, + {Class: classify.ClassDNS, Timestamp: baseTime.Add(600 * time.Millisecond)}, + {Class: classify.ClassDNS, Timestamp: baseTime.Add(1200 * time.Millisecond)}, + } + snaps := AggregatePcap(makePcapChan(pkts), 500, nil) + if len(snaps) != 3 { + t.Fatalf("expected 3 snapshots, got %d", len(snaps)) + } + for i, s := range snaps { + if s.TotalPackets != 1 { + t.Errorf("snapshot[%d] expected TotalPackets=1, got %d", i, s.TotalPackets) + } + } +} + +// TestAggregatePcapGaps verifies that gap windows produce empty snapshots (D-02: gaps are silent). +func TestAggregatePcapGaps(t *testing.T) { + // Packets at T+0ms and T+1500ms (skipping windows 1 and 2) + pkts := []classify.ClassifiedPacket{ + {Class: classify.ClassICMP, Timestamp: baseTime}, + {Class: classify.ClassICMP, Timestamp: baseTime.Add(1500 * time.Millisecond)}, + } + snaps := AggregatePcap(makePcapChan(pkts), 500, nil) + // Expected: window 0 (1 pkt), window 1 (0 pkts), window 2 (0 pkts), window 3 (1 pkt) = 4 snapshots + if len(snaps) != 4 { + t.Fatalf("expected 4 snapshots (with gaps), got %d", len(snaps)) + } + if snaps[0].TotalPackets != 1 { + t.Errorf("snapshot[0] expected TotalPackets=1, got %d", snaps[0].TotalPackets) + } + if snaps[1].TotalPackets != 0 { + t.Errorf("snapshot[1] expected TotalPackets=0 (gap), got %d", snaps[1].TotalPackets) + } + if snaps[2].TotalPackets != 0 { + t.Errorf("snapshot[2] expected TotalPackets=0 (gap), got %d", snaps[2].TotalPackets) + } + if snaps[3].TotalPackets != 1 { + t.Errorf("snapshot[3] expected TotalPackets=1, got %d", snaps[3].TotalPackets) + } +} + +// TestAggregatePcapEmpty verifies that empty channel returns empty (nil) slice. +func TestAggregatePcapEmpty(t *testing.T) { + snaps := AggregatePcap(makePcapChan(nil), 500, nil) + if len(snaps) != 0 { + t.Errorf("expected 0 snapshots for empty input, got %d", len(snaps)) + } +} + +// TestAggregatePcapWindowIndex verifies sequential WindowIndex values. +func TestAggregatePcapWindowIndex(t *testing.T) { + pkts := []classify.ClassifiedPacket{ + {Class: classify.ClassHTTPS, Timestamp: baseTime}, + {Class: classify.ClassHTTPS, Timestamp: baseTime.Add(600 * time.Millisecond)}, + {Class: classify.ClassHTTPS, Timestamp: baseTime.Add(1200 * time.Millisecond)}, + } + snaps := AggregatePcap(makePcapChan(pkts), 500, nil) + for i, s := range snaps { + if s.WindowIndex != i { + t.Errorf("snapshot[%d].WindowIndex = %d; want %d", i, s.WindowIndex, i) + } + } +} + +// TestAggregatePcapClassCounts verifies per-class counts in a shared window. +func TestAggregatePcapClassCounts(t *testing.T) { + pkts := []classify.ClassifiedPacket{ + {Class: classify.ClassICMP, Timestamp: baseTime}, + {Class: classify.ClassDNS, Timestamp: baseTime.Add(50 * time.Millisecond)}, + {Class: classify.ClassICMP, Timestamp: baseTime.Add(100 * time.Millisecond)}, + {Class: classify.ClassHTTPS, Timestamp: baseTime.Add(150 * time.Millisecond)}, + } + snaps := AggregatePcap(makePcapChan(pkts), 500, nil) + if len(snaps) != 1 { + t.Fatalf("expected 1 snapshot, got %d", len(snaps)) + } + s := snaps[0] + if s.Counts[classify.ClassICMP] != 2 { + t.Errorf("expected ICMP=2, got %d", s.Counts[classify.ClassICMP]) + } + if s.Counts[classify.ClassDNS] != 1 { + t.Errorf("expected DNS=1, got %d", s.Counts[classify.ClassDNS]) + } + if s.Counts[classify.ClassHTTPS] != 1 { + t.Errorf("expected HTTPS=1, got %d", s.Counts[classify.ClassHTTPS]) + } + if s.TotalPackets != 4 { + t.Errorf("expected TotalPackets=4, got %d", s.TotalPackets) + } +} + +// TestAggregatePcapOnSnapshot verifies that the onSnapshot callback fires for each snapshot. +func TestAggregatePcapOnSnapshot(t *testing.T) { + pkts := []classify.ClassifiedPacket{ + {Class: classify.ClassICMP, Timestamp: baseTime}, + {Class: classify.ClassDNS, Timestamp: baseTime.Add(600 * time.Millisecond)}, + } + var called int + var calledIndices []int + onSnapshot := func(s classify.WindowSnapshot) { + called++ + calledIndices = append(calledIndices, s.WindowIndex) + } + snaps := AggregatePcap(makePcapChan(pkts), 500, onSnapshot) + if called != len(snaps) { + t.Errorf("onSnapshot called %d times; want %d (once per snapshot)", called, len(snaps)) + } + for i, idx := range calledIndices { + if idx != i { + t.Errorf("onSnapshot call[%d] had WindowIndex=%d; want %d", i, idx, i) + } + } +}