345 lines
13 KiB
Markdown
345 lines
13 KiB
Markdown
---
|
|
phase: 01-capture-and-classification
|
|
plan: 03
|
|
type: execute
|
|
wave: 2
|
|
depends_on: ["01-01"]
|
|
files_modified:
|
|
- aggregate/window.go
|
|
- aggregate/summary.go
|
|
- aggregate/window_test.go
|
|
- aggregate/summary_test.go
|
|
autonomous: true
|
|
requirements: [CLAS-03, CLAS-04]
|
|
|
|
must_haves:
|
|
truths:
|
|
- "On exit, user sees a per-protocol packet count summary printed to stderr"
|
|
- "User can pass --verbose and see per-window protocol activity lines on stderr"
|
|
- "Aggregator accumulates packet counts into time windows and emits WindowSnapshots"
|
|
artifacts:
|
|
- path: "aggregate/window.go"
|
|
provides: "Ticker-driven window aggregator that emits WindowSnapshots"
|
|
exports: ["Aggregate"]
|
|
- path: "aggregate/summary.go"
|
|
provides: "PrintSummary and PrintWindowLine functions for stderr output"
|
|
exports: ["PrintSummary", "PrintWindowLine"]
|
|
- path: "aggregate/window_test.go"
|
|
provides: "Tests for window aggregation logic"
|
|
min_lines: 50
|
|
- path: "aggregate/summary_test.go"
|
|
provides: "Tests for summary formatting and verbose output"
|
|
min_lines: 60
|
|
key_links:
|
|
- from: "aggregate/window.go"
|
|
to: "classify/types.go"
|
|
via: "Consumes ClassifiedPacket, produces WindowSnapshot"
|
|
pattern: "classify\\.ClassifiedPacket"
|
|
- from: "aggregate/summary.go"
|
|
to: "classify/types.go"
|
|
via: "Uses TrafficClass and WindowSnapshot for formatting"
|
|
pattern: "classify\\.TrafficClass"
|
|
---
|
|
|
|
<objective>
|
|
Implement the aggregation and stats reporting package: time-windowed accumulation, exit summary, and verbose per-window output.
|
|
|
|
Purpose: This package bridges classified packets and the eventual audio engine (Phase 2). It accumulates per-class counts into time windows, prints verbose activity lines during capture, and prints a final protocol summary on exit.
|
|
|
|
Output: aggregate/ package with Aggregate, PrintSummary, PrintWindowLine functions and full unit tests.
|
|
</objective>
|
|
|
|
<execution_context>
|
|
@$HOME/.claude/get-shit-done/workflows/execute-plan.md
|
|
@$HOME/.claude/get-shit-done/templates/summary.md
|
|
</execution_context>
|
|
|
|
<context>
|
|
@.planning/PROJECT.md
|
|
@.planning/ROADMAP.md
|
|
@.planning/phases/01-capture-and-classification/01-CONTEXT.md
|
|
@.planning/phases/01-capture-and-classification/01-RESEARCH.md
|
|
@.planning/phases/01-capture-and-classification/01-01-SUMMARY.md
|
|
|
|
<interfaces>
|
|
<!-- From classify/types.go (created in Plan 01): -->
|
|
```go
|
|
package classify
|
|
|
|
type TrafficClass string
|
|
|
|
const (
|
|
ClassICMP TrafficClass = "ICMP"
|
|
ClassDNS TrafficClass = "DNS"
|
|
ClassHTTPS TrafficClass = "HTTPS"
|
|
ClassHTTP TrafficClass = "HTTP"
|
|
ClassSSH TrafficClass = "SSH"
|
|
ClassSMTP TrafficClass = "SMTP"
|
|
ClassNTP TrafficClass = "NTP"
|
|
ClassDHCP TrafficClass = "DHCP"
|
|
ClassOtherTCP TrafficClass = "other-TCP"
|
|
ClassOtherUDP TrafficClass = "other-UDP"
|
|
ClassUnknown TrafficClass = "unknown"
|
|
)
|
|
|
|
func AllClasses() []TrafficClass { ... }
|
|
|
|
type ClassifiedPacket struct {
|
|
Class TrafficClass
|
|
SrcPort uint16
|
|
DstPort uint16
|
|
Protocol string
|
|
Length int
|
|
}
|
|
|
|
type WindowSnapshot struct {
|
|
Counts map[TrafficClass]int64
|
|
TotalPackets int64
|
|
WindowIndex int
|
|
}
|
|
```
|
|
</interfaces>
|
|
</context>
|
|
|
|
<tasks>
|
|
|
|
<task type="auto" tdd="true">
|
|
<name>Task 1: Implement ticker-driven window aggregator</name>
|
|
<files>aggregate/window.go, aggregate/window_test.go</files>
|
|
<read_first>
|
|
classify/types.go
|
|
.planning/phases/01-capture-and-classification/01-RESEARCH.md
|
|
</read_first>
|
|
<behavior>
|
|
- TestAggregateEmitsSnapshot: Send 5 ClassifiedPackets, trigger window tick, receive WindowSnapshot with correct counts
|
|
- TestAggregateMultipleClasses: Send packets of 3 different classes, verify snapshot has all 3 with correct counts
|
|
- TestAggregateDoneFlushesPartial: Close done channel mid-window, verify final partial snapshot is emitted
|
|
- TestAggregateEmptyWindow: Trigger tick with no packets, verify snapshot has zero counts and TotalPackets==0
|
|
- TestAggregateWindowIndex: Multiple ticks increment WindowIndex (0, 1, 2...)
|
|
</behavior>
|
|
<action>
|
|
1. Create aggregate/window.go implementing the ticker-driven aggregation pattern from RESEARCH.md Pattern 2:
|
|
|
|
```go
|
|
package aggregate
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/netsynth/netsynth/classify"
|
|
)
|
|
|
|
// DefaultWindowMs is the default time window duration in milliseconds.
|
|
// Claude's discretion per CONTEXT.md.
|
|
const DefaultWindowMs = 500
|
|
|
|
// Aggregate reads ClassifiedPackets from events, accumulates counts per TrafficClass
|
|
// in time windows of windowMs duration, and emits a WindowSnapshot per window.
|
|
// When done is closed, flushes the current partial window and closes the output channel.
|
|
// If verbose is true, calls onSnapshot for each window (used for --verbose stderr output).
|
|
func Aggregate(done <-chan struct{}, events <-chan classify.ClassifiedPacket,
|
|
windowMs int, onSnapshot func(classify.WindowSnapshot)) <-chan classify.WindowSnapshot {
|
|
|
|
out := make(chan classify.WindowSnapshot, 8)
|
|
go func() {
|
|
defer close(out)
|
|
ticker := time.NewTicker(time.Duration(windowMs) * time.Millisecond)
|
|
defer ticker.Stop()
|
|
|
|
counts := make(map[classify.TrafficClass]int64)
|
|
var totalPackets int64
|
|
windowIndex := 0
|
|
|
|
flush := func() {
|
|
snap := classify.WindowSnapshot{
|
|
Counts: counts,
|
|
TotalPackets: totalPackets,
|
|
WindowIndex: windowIndex,
|
|
}
|
|
if onSnapshot != nil {
|
|
onSnapshot(snap)
|
|
}
|
|
out <- snap
|
|
counts = make(map[classify.TrafficClass]int64)
|
|
totalPackets = 0
|
|
windowIndex++
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case <-done:
|
|
flush() // flush final partial window
|
|
return
|
|
case <-ticker.C:
|
|
flush()
|
|
case ev, ok := <-events:
|
|
if !ok {
|
|
flush()
|
|
return
|
|
}
|
|
counts[ev.Class]++
|
|
totalPackets++
|
|
}
|
|
}
|
|
}()
|
|
return out
|
|
}
|
|
```
|
|
|
|
2. Create aggregate/window_test.go:
|
|
- Use a done channel and a buffered events channel to control timing without real tickers
|
|
- For ticker tests, use a short windowMs (10ms) and sleep briefly to trigger ticks
|
|
- For done-flush tests, close done channel immediately and verify output
|
|
- Verify WindowSnapshot.Counts map contains expected class keys and values
|
|
- Verify WindowSnapshot.TotalPackets matches sum of counts
|
|
- Verify WindowIndex increments across windows
|
|
|
|
Write tests first (RED), then implement (GREEN).
|
|
</action>
|
|
<verify>
|
|
<automated>cd /home/dev/workspace/yoloyolo && go test -v -count=1 -race ./aggregate/...</automated>
|
|
</verify>
|
|
<acceptance_criteria>
|
|
- aggregate/window.go contains `func Aggregate(done <-chan struct{}, events <-chan classify.ClassifiedPacket`
|
|
- aggregate/window.go contains `const DefaultWindowMs = 500`
|
|
- aggregate/window.go contains `time.NewTicker`
|
|
- aggregate/window.go imports `github.com/netsynth/netsynth/classify`
|
|
- aggregate/window_test.go contains `TestAggregate` with subtests for snapshot emission, multiple classes, done flush, empty window, and window index
|
|
- `go test -v -count=1 -race ./aggregate/...` exits 0
|
|
</acceptance_criteria>
|
|
<done>Aggregate function emits WindowSnapshots on ticker ticks and flushes partial window on done. All timing-based tests pass including race detector.</done>
|
|
</task>
|
|
|
|
<task type="auto" tdd="true">
|
|
<name>Task 2: Implement exit summary and verbose window output formatting</name>
|
|
<files>aggregate/summary.go, aggregate/summary_test.go</files>
|
|
<read_first>
|
|
classify/types.go
|
|
aggregate/window.go
|
|
.planning/phases/01-capture-and-classification/01-RESEARCH.md
|
|
</read_first>
|
|
<behavior>
|
|
- TestPrintSummary: Given totals map {ICMP: 10, DNS: 50, HTTPS: 100}, output contains "ICMP", "DNS", "HTTPS" with correct counts and percentages, plus "TOTAL" line
|
|
- TestPrintSummaryEmpty: Given empty totals, output contains "TOTAL" with 0 packets
|
|
- TestPrintSummarySorted: Output classes appear in stable alphabetical or canonical order (not random map order)
|
|
- TestPrintWindowLine: Given WindowSnapshot with {DNS: 5, HTTPS: 10}, output contains window index, class names, and counts on a single line
|
|
- TestPrintWindowLineEmpty: Given empty snapshot, output contains window index with no class entries
|
|
- TestAccumulateTotals: Given multiple WindowSnapshots, AccumulateTotals produces correct cumulative counts
|
|
</behavior>
|
|
<action>
|
|
1. Create aggregate/summary.go:
|
|
|
|
```go
|
|
package aggregate
|
|
|
|
import (
|
|
"fmt"
|
|
"io"
|
|
"sort"
|
|
|
|
"github.com/netsynth/netsynth/classify"
|
|
)
|
|
|
|
// PrintSummary writes a per-protocol packet count summary to w (typically os.Stderr).
|
|
// Output format matches tcpdump conventions (CLAS-03).
|
|
func PrintSummary(w io.Writer, totals map[classify.TrafficClass]int64) {
|
|
var totalPackets int64
|
|
for _, count := range totals {
|
|
totalPackets += count
|
|
}
|
|
|
|
fmt.Fprintln(w, "\n--- Protocol Summary ---")
|
|
|
|
// Sort classes for stable output
|
|
classes := make([]string, 0, len(totals))
|
|
for c := range totals {
|
|
classes = append(classes, string(c))
|
|
}
|
|
sort.Strings(classes)
|
|
|
|
for _, className := range classes {
|
|
c := classify.TrafficClass(className)
|
|
count := totals[c]
|
|
pct := 0.0
|
|
if totalPackets > 0 {
|
|
pct = float64(count) / float64(totalPackets) * 100.0
|
|
}
|
|
fmt.Fprintf(w, " %-15s %8d packets (%5.1f%%)\n", c, count, pct)
|
|
}
|
|
fmt.Fprintf(w, " %-15s %8d packets\n", "TOTAL", totalPackets)
|
|
}
|
|
|
|
// PrintWindowLine writes a single verbose per-window activity line to w (CLAS-04).
|
|
// Format: [window N] CLASS:count CLASS:count ... (total: N)
|
|
func PrintWindowLine(w io.Writer, snap classify.WindowSnapshot) {
|
|
fmt.Fprintf(w, "[window %d]", snap.WindowIndex)
|
|
// Sort for stable output
|
|
classes := make([]string, 0, len(snap.Counts))
|
|
for c := range snap.Counts {
|
|
classes = append(classes, string(c))
|
|
}
|
|
sort.Strings(classes)
|
|
for _, className := range classes {
|
|
c := classify.TrafficClass(className)
|
|
if snap.Counts[c] > 0 {
|
|
fmt.Fprintf(w, " %s:%d", c, snap.Counts[c])
|
|
}
|
|
}
|
|
fmt.Fprintf(w, " (total: %d)\n", snap.TotalPackets)
|
|
}
|
|
|
|
// AccumulateTotals merges a WindowSnapshot's counts into cumulative totals.
|
|
func AccumulateTotals(totals map[classify.TrafficClass]int64, snap classify.WindowSnapshot) {
|
|
for class, count := range snap.Counts {
|
|
totals[class] += count
|
|
}
|
|
}
|
|
```
|
|
|
|
2. Create aggregate/summary_test.go:
|
|
- Use bytes.Buffer as io.Writer to capture output
|
|
- TestPrintSummary: Verify output contains expected class names, counts, percentages, and TOTAL line
|
|
- TestPrintSummaryEmpty: Verify output contains "TOTAL" with 0
|
|
- TestPrintSummarySorted: Verify output lines are in alphabetical order (capture output, split lines, check ordering)
|
|
- TestPrintWindowLine: Verify format "[window 0] DNS:5 HTTPS:10 (total: 15)"
|
|
- TestAccumulateTotals: Two snapshots accumulated, verify merged counts
|
|
|
|
Write tests first (RED), then implement (GREEN).
|
|
</action>
|
|
<verify>
|
|
<automated>cd /home/dev/workspace/yoloyolo && go test -v -count=1 ./aggregate/...</automated>
|
|
</verify>
|
|
<acceptance_criteria>
|
|
- aggregate/summary.go contains `func PrintSummary(w io.Writer, totals map[classify.TrafficClass]int64)`
|
|
- aggregate/summary.go contains `func PrintWindowLine(w io.Writer, snap classify.WindowSnapshot)`
|
|
- aggregate/summary.go contains `func AccumulateTotals(totals map[classify.TrafficClass]int64, snap classify.WindowSnapshot)`
|
|
- aggregate/summary.go contains `"--- Protocol Summary ---"`
|
|
- aggregate/summary.go contains `sort.Strings` for stable output ordering
|
|
- aggregate/summary_test.go contains `TestPrintSummary`, `TestPrintSummaryEmpty`, `TestPrintSummarySorted`
|
|
- aggregate/summary_test.go contains `TestPrintWindowLine`, `TestAccumulateTotals`
|
|
- `go test -v -count=1 ./aggregate/...` exits 0
|
|
</acceptance_criteria>
|
|
<done>PrintSummary outputs per-protocol counts with percentages and TOTAL to stderr (CLAS-03). PrintWindowLine outputs per-window activity for --verbose mode (CLAS-04). AccumulateTotals correctly merges snapshots. All formatting tests pass with stable sorted output.</done>
|
|
</task>
|
|
|
|
</tasks>
|
|
|
|
<verification>
|
|
- `go test -v -count=1 -race ./aggregate/...` all tests pass
|
|
- `go vet ./aggregate/...` no issues
|
|
- Output format matches tcpdump convention (stderr, per-protocol counts with percentages)
|
|
</verification>
|
|
|
|
<success_criteria>
|
|
- Aggregate emits WindowSnapshots on time ticks and flushes on done
|
|
- PrintSummary prints per-protocol packet counts with percentages and TOTAL to an io.Writer
|
|
- PrintWindowLine prints per-window activity for verbose mode
|
|
- AccumulateTotals correctly accumulates across multiple snapshots
|
|
- All output is sorted for deterministic display
|
|
- All unit tests passing (including race detector)
|
|
</success_criteria>
|
|
|
|
<output>
|
|
After completion, create `.planning/phases/01-capture-and-classification/01-03-SUMMARY.md`
|
|
</output>
|