feat(04-01): add AggregatePcap with timestamp-based windowing and gap-filling
- Add AggregatePcap to aggregate/window.go: reads ClassifiedPackets from channel, assigns to time windows using Timestamp field (D-01), fills gaps with empty snapshots (D-02: silence), fires onSnapshot callback per window (D-07) - Add 7 tests covering basic, multiple windows, gaps, empty input, WindowIndex, class counts, and onSnapshot callback - All tests pass
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user