diff --git a/README.md b/README.md index 453a61a..c33689b 100644 --- a/README.md +++ b/README.md @@ -79,6 +79,28 @@ sudo wireblast -i eth1 --dst-ip 192.0.2.10 --packet-size 512 --pps 1M -d 30s sudo wireblast --no-tui -i eth1 --dst-ip 192.0.2.10 --pps 1M -d 30s -y ``` +For repeatable benchmarks, `--no-tui` can also write versioned, machine-readable +statistics without changing the human-readable output: + +```bash +# CSV for spreadsheets and analysis tools (the default format) +sudo wireblast --no-tui -i eth1 --dst-ip 192.0.2.10 -d 30s -y \ + --stats-file run.csv + +# Or newline-delimited JSON for streaming consumers +sudo wireblast --no-tui -i eth1 --dst-ip 192.0.2.10 -d 30s -y \ + --stats-file run.jsonl --stats-format jsonl +``` + +The output contains one aggregate sample per second, followed by a final +aggregate and a final record for every AF_XDP queue. It includes traffic totals, +the most recently sampled rates, and the kernel's AF_XDP drop and ring counters. +AF_XDP counter fields use their exact Linux UAPI names. The +`kernel_rx_descriptors` and `kernel_tx_descriptors` fields report ring progress, +not packet counts; one multi-buffer packet can occupy several descriptors. +The output file must not already exist, which protects previous benchmark data +from accidental replacement. + No spare interface? A veth pair gives you a sender and a receiver on one machine, with nothing touching your real network: ```bash @@ -143,6 +165,7 @@ The command is a thin shell; the work is in `internal/`: | `internal/rate` | the aggregate token-bucket rate limiter | | `internal/dataplane` | everything AF_XDP: opening sockets, the XDP filter, the run loop | | `internal/stats` | atomic counters, rate snapshots, history | +| `internal/statsexport` | versioned CSV and JSONL statistics output | | `internal/tui` | the interactive wizard and live dashboard (Bubble Tea) | | `internal/prefs` | remembers your last run under `~/.wireblast/` | | `internal/app` | wires a validated config into a running dataplane for `--no-tui` | diff --git a/internal/app/app.go b/internal/app/app.go index 1733dfc..32a569d 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -93,7 +93,7 @@ func Prepare(cfg *config.Config, opts PrepareOptions) (*Prepared, error) { // RunNonInteractive validates the flags, prints what it is about to do, and // transmits — printing a statistics line every second and a summary at the // end. This is the shape scripts and CI use. -func RunNonInteractive(ctx context.Context, cfg *config.Config, out io.Writer) error { +func RunNonInteractive(ctx context.Context, cfg *config.Config, out io.Writer) (retErr error) { logf := func(format string, args ...any) { fmt.Fprintf(out, "wireblast: "+format+"\n", args...) } @@ -101,6 +101,13 @@ func RunNonInteractive(ctx context.Context, cfg *config.Config, out io.Writer) e if err != nil { return err } + statsOut, err := openStatsOutput(cfg) + if err != nil { + return err + } + if statsOut != nil { + defer func() { retErr = errors.Join(retErr, statsOut.Close()) }() + } if err := p.Preflight.Err(); err != nil { return err @@ -144,21 +151,24 @@ func RunNonInteractive(ctx context.Context, cfg *config.Config, out io.Writer) e // Print a status line every second while the run proceeds. reportCtx, stopReport := context.WithCancel(ctx) - done := make(chan struct{}) + done := make(chan error, 1) go func() { - defer close(done) - report(reportCtx, p.Runner, out) + done <- report(reportCtx, p.Runner, out, statsOut) }() runErr := p.Runner.Wait() stopReport() - <-done + reportErr := <-done + var finalErr error + if statsOut != nil { + finalErr = statsOut.Final(p.Runner.Stats()) + } fmt.Fprintf(out, "\n%s\n", p.Runner.Stats().Summary()) - if runErr != nil && !errors.Is(runErr, context.Canceled) { - return runErr + if errors.Is(runErr, context.Canceled) { + runErr = nil } - return nil + return errors.Join(runErr, reportErr, finalErr) } // progress prints a growing line of dots while something slow happens, and @@ -193,18 +203,24 @@ func progress(out io.Writer, what string) func() { } // report prints one status line a second until the run ends. -func report(ctx context.Context, r *dataplane.Runner, out io.Writer) { +func report(ctx context.Context, r *dataplane.Runner, out io.Writer, statsOut *statsOutput) error { t := time.NewTicker(time.Second) defer t.Stop() for { select { case <-ctx.Done(): - return + return nil case <-t.C: s := r.Stats() if s.State == stats.StateStarting { continue } + if statsOut != nil { + if err := statsOut.Sample(s); err != nil { + r.Stop() + return err + } + } fmt.Fprintln(out, s.Line()) } } diff --git a/internal/app/stats_output.go b/internal/app/stats_output.go new file mode 100644 index 0000000..7228b9b --- /dev/null +++ b/internal/app/stats_output.go @@ -0,0 +1,40 @@ +package app + +import ( + "errors" + "fmt" + "os" + + "github.com/atoonk/wireblast/internal/config" + "github.com/atoonk/wireblast/internal/stats" + "github.com/atoonk/wireblast/internal/statsexport" +) + +type statsOutput struct { + file *os.File + stream *statsexport.Writer +} + +func openStatsOutput(cfg *config.Config) (*statsOutput, error) { + if cfg.StatsFile == "" { + return nil, nil + } + file, err := os.OpenFile(cfg.StatsFile, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644) + if err != nil { + return nil, fmt.Errorf("create statistics file %s: %w", cfg.StatsFile, err) + } + stream, err := statsexport.New(statsexport.Format(cfg.StatsFormat), file) + if err != nil { + closeErr := file.Close() + removeErr := os.Remove(cfg.StatsFile) + return nil, errors.Join(err, closeErr, removeErr) + } + return &statsOutput{file: file, stream: stream}, nil +} + +func (o *statsOutput) Sample(s *stats.Snapshot) error { return o.stream.Sample(s) } +func (o *statsOutput) Final(s *stats.Snapshot) error { return o.stream.Final(s) } + +func (o *statsOutput) Close() error { + return errors.Join(o.stream.Close(), o.file.Close()) +} diff --git a/internal/app/stats_output_test.go b/internal/app/stats_output_test.go new file mode 100644 index 0000000..4192c7c --- /dev/null +++ b/internal/app/stats_output_test.go @@ -0,0 +1,77 @@ +package app + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/atoonk/wireblast/internal/config" + "github.com/atoonk/wireblast/internal/stats" +) + +func TestOpenStatsOutputDisabled(t *testing.T) { + cfg := config.Default() + out, err := openStatsOutput(&cfg) + if err != nil { + t.Fatalf("openStatsOutput: %v", err) + } + if out != nil { + t.Fatal("disabled stats output should return nil") + } +} + +func TestOpenStatsOutputRefusesToOverwrite(t *testing.T) { + path := filepath.Join(t.TempDir(), "run.csv") + if err := os.WriteFile(path, []byte("keep me"), 0o644); err != nil { + t.Fatal(err) + } + cfg := config.Default() + cfg.StatsFile = path + cfg.StatsFormat = config.StatsCSV + + if _, err := openStatsOutput(&cfg); err == nil { + t.Fatal("opening an existing stats file should fail") + } + got, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if string(got) != "keep me" { + t.Fatalf("existing file changed to %q", got) + } +} + +func TestStatsOutputCreatesAndFlushesAStream(t *testing.T) { + path := filepath.Join(t.TempDir(), "run.csv") + cfg := config.Default() + cfg.StatsFile = path + cfg.StatsFormat = config.StatsCSV + out, err := openStatsOutput(&cfg) + if err != nil { + t.Fatalf("openStatsOutput: %v", err) + } + s := &stats.Snapshot{ + At: time.Date(2026, 8, 10, 12, 0, 0, 0, time.UTC), + State: stats.StateRunning, + } + if err := out.Sample(s); err != nil { + t.Fatalf("Sample: %v", err) + } + + // A sample must be visible before Close so a preempted process leaves data. + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(data), "schema_version") || !strings.Contains(string(data), "sample") { + t.Fatalf("stream was not flushed after Sample:\n%s", data) + } + if err := out.Final(s); err != nil { + t.Fatalf("Final: %v", err) + } + if err := out.Close(); err != nil { + t.Fatalf("Close: %v", err) + } +} diff --git a/internal/cli/flags.go b/internal/cli/flags.go index 4698a72..22aaf22 100644 --- a/internal/cli/flags.go +++ b/internal/cli/flags.go @@ -114,6 +114,10 @@ in L1, so --bps 10G means 10G line rate.`, "bit rate limit in L1 bits (frame plus preamble, SFD and interframe gap), "+ "aggregate across queues (e.g. 10G, 2.5Gbps, or 'unlimited')") f.IntVar(&cfg.Queues, "queues", cfg.Queues, "number of NIC queues to transmit on (0 = all)") + f.StringVar(&cfg.StatsFile, "stats-file", cfg.StatsFile, + "write machine-readable statistics to a new file (requires --no-tui)") + f.StringVar((*string)(&cfg.StatsFormat), "stats-format", string(cfg.StatsFormat), + "machine-readable statistics format: csv, jsonl") f.StringVar((*string)(&cfg.RxMode), "rx-mode", string(cfg.RxMode), "what to receive through AF_XDP: "+rxModeList()) @@ -196,6 +200,10 @@ func applyOptions(cmd *cobra.Command, cfg *config.Config, opt *options) error { cfg.RxMode = config.RxMode(strings.ToLower(strings.TrimSpace(string(cfg.RxMode)))) cfg.PCAPTiming = config.PcapTiming(strings.ToLower(strings.TrimSpace(string(cfg.PCAPTiming)))) cfg.FlowOrder = config.FlowOrder(strings.ToLower(strings.TrimSpace(string(cfg.FlowOrder)))) + cfg.StatsFormat = config.StatsFormat(strings.ToLower(strings.TrimSpace(string(cfg.StatsFormat)))) + if f.Changed("stats-format") && cfg.StatsFile == "" { + return fmt.Errorf("--stats-format requires --stats-file") + } // Choosing --mode pcap without saying --rx-mode implies nothing about // receiving, but choosing --pcap without --mode is an easy slip to catch. diff --git a/internal/cli/flags_test.go b/internal/cli/flags_test.go index 4bb0af8..319ae02 100644 --- a/internal/cli/flags_test.go +++ b/internal/cli/flags_test.go @@ -68,6 +68,36 @@ func TestFullNonInteractiveExample(t *testing.T) { } } +func TestStatsExportFlags(t *testing.T) { + got, err := parse(t, + "--no-tui", + "--stats-file", "run.csv", + "--stats-format", "CSV", + ) + if err != nil { + t.Fatalf("Execute: %v", err) + } + if got.StatsFile != "run.csv" || got.StatsFormat != config.StatsCSV { + t.Fatalf("stats export parsed as file=%q format=%q", got.StatsFile, got.StatsFormat) + } + + got, err = parse(t, + "--no-tui", + "--stats-file", "run.jsonl", + "--stats-format", "jsonl", + ) + if err != nil { + t.Fatalf("Execute: %v", err) + } + if got.StatsFormat != config.StatsJSONL { + t.Fatalf("stats format = %q, want jsonl", got.StatsFormat) + } + + if _, err := parse(t, "--stats-format", "jsonl"); err == nil { + t.Error("--stats-format without --stats-file should fail") + } +} + func TestRateFlags(t *testing.T) { tests := []struct { args []string @@ -193,6 +223,7 @@ func TestEveryConfigFieldHasAFlag(t *testing.T) { "duration", "pps", "bps", "queues", "rx-mode", "rx-port", "rx-cidr", "pcap", "pcap-timing", "pcap-loop", "pcap-memory", + "stats-file", "stats-format", "no-tui", "start", "yes", "allow-match-all", "forget", } for _, name := range want { diff --git a/internal/config/config.go b/internal/config/config.go index 17386f3..bc504e4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -88,6 +88,14 @@ const ( FlowRandom FlowOrder = "random" ) +// StatsFormat is the machine-readable stream written by --stats-file. +type StatsFormat string + +const ( + StatsCSV StatsFormat = "csv" + StatsJSONL StatsFormat = "jsonl" +) + // Frame-size limits, in total Ethernet frame bytes including the 4-byte FCS. // See the package docs on PacketSize for exactly what that means. const ( @@ -161,6 +169,11 @@ type Config struct { BPS uint64 // on-the-wire bits/sec, 0 means unlimited Queues int // 0 means all available queues + // Machine-readable statistics. StatsFile being empty disables export. + // These describe one invocation and are never remembered by the wizard. + StatsFile string + StatsFormat StatsFormat + // Receive behaviour. RxMode RxMode RxPorts []uint16 @@ -199,6 +212,7 @@ func Default() Config { PayloadByte: 0x5a, Duration: 30 * time.Second, PPS: DefaultPPS, + StatsFormat: StatsCSV, RxMode: RxNone, PCAPTiming: PcapRate, PCAPLoop: true, @@ -352,6 +366,14 @@ func (c *Config) Validate() error { if c.Queues < 0 { bad("--queues must not be negative (0 means all available queues)") } + if c.StatsFile != "" { + if !c.NoTUI { + bad("--stats-file is currently supported only with --no-tui") + } + if c.StatsFormat != StatsCSV && c.StatsFormat != StatsJSONL { + bad("--stats-format %q is not one of csv, jsonl", c.StatsFormat) + } + } errs = append(errs, c.validateRx()...) return errors.Join(errs...) diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 515a6e1..0806e59 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -44,6 +44,9 @@ func TestDefaults(t *testing.T) { if c.Queues != 0 { t.Errorf("default queues = %d, want 0 (all)", c.Queues) } + if c.StatsFile != "" || c.StatsFormat != StatsCSV { + t.Errorf("default stats export = file %q format %q, want disabled csv", c.StatsFile, c.StatsFormat) + } if c.FlowOrder != FlowSequential { t.Errorf("default flow-order = %q, want sequential", c.FlowOrder) } @@ -53,6 +56,39 @@ func TestDefaults(t *testing.T) { } } +func TestStatsExportValidation(t *testing.T) { + tests := []struct { + name string + mutate func(*Config) + wantErr bool + }{ + {"disabled", func(*Config) {}, false}, + {"csv noninteractive", func(c *Config) { + c.NoTUI, c.StatsFile, c.StatsFormat = true, "run.csv", StatsCSV + }, false}, + {"jsonl noninteractive", func(c *Config) { + c.NoTUI, c.StatsFile, c.StatsFormat = true, "run.jsonl", StatsJSONL + }, false}, + {"tui rejected", func(c *Config) { + c.StatsFile, c.StatsFormat = "run.csv", StatsCSV + }, true}, + {"unknown format", func(c *Config) { + c.NoTUI, c.StatsFile, c.StatsFormat = true, "run.bin", StatsFormat("binary") + }, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := valid() + tt.mutate(&cfg) + err := cfg.Validate() + if (err != nil) != tt.wantErr { + t.Fatalf("Validate() error = %v, wantErr %v", err, tt.wantErr) + } + }) + } +} + func TestValidate(t *testing.T) { tests := []struct { name string diff --git a/internal/prefs/prefs.go b/internal/prefs/prefs.go index f00de2f..ff66472 100644 --- a/internal/prefs/prefs.go +++ b/internal/prefs/prefs.go @@ -188,6 +188,8 @@ func sanitise(c config.Config) config.Config { c.SkipWizard = false c.AssumeYes = false c.AllowMatchAll = false + c.StatsFile = "" + c.StatsFormat = config.StatsCSV // Match-all is not remembered either. It is still guarded by a // confirmation, so carrying it forward would not be unsafe — but a later, @@ -273,6 +275,8 @@ func Merge(flags config.Config, saved config.Config, changed func(flag string) b out.SkipWizard = flags.SkipWizard out.AssumeYes = flags.AssumeYes out.AllowMatchAll = flags.AllowMatchAll + out.StatsFile = flags.StatsFile + out.StatsFormat = flags.StatsFormat // A --bps on the command line lifts the remembered packet cap the same way // it lifts the default one. diff --git a/internal/prefs/prefs_test.go b/internal/prefs/prefs_test.go index 4d94e01..703a78c 100644 --- a/internal/prefs/prefs_test.go +++ b/internal/prefs/prefs_test.go @@ -72,6 +72,22 @@ func TestConsentIsNeverRemembered(t *testing.T) { } } +func TestStatsOutputIsNeverRemembered(t *testing.T) { + s := NewAt(t.TempDir()) + c := sample() + c.NoTUI = true + c.StatsFile = "/tmp/run.jsonl" + c.StatsFormat = config.StatsJSONL + if err := s.Save(&c); err != nil { + t.Fatal(err) + } + + got, _ := s.Last() + if got.StatsFile != "" || got.StatsFormat != config.StatsCSV { + t.Errorf("stats output must not be remembered: file=%q format=%q", got.StatsFile, got.StatsFormat) + } +} + // Narrower receive modes are genuinely useful to remember, and are kept. func TestNarrowReceiveModesAreRemembered(t *testing.T) { s := NewAt(t.TempDir()) @@ -246,12 +262,18 @@ func TestMergeNeverInheritsConsent(t *testing.T) { saved.AllowMatchAll = true saved.AssumeYes = true saved.NoTUI = true + saved.StatsFile = "/tmp/old.csv" + saved.StatsFormat = config.StatsJSONL flags := config.Default() + flags.StatsFile = "/tmp/new.csv" got := Merge(flags, saved, func(string) bool { return false }) if got.AllowMatchAll || got.AssumeYes || got.NoTUI { t.Errorf("consent flags must come from this invocation only: %+v", got) } + if got.StatsFile != "/tmp/new.csv" || got.StatsFormat != config.StatsCSV { + t.Errorf("stats output must come from this invocation only: file=%q format=%q", got.StatsFile, got.StatsFormat) + } } // A --bps on the command line has to lift a remembered packet cap, or the new diff --git a/internal/statsexport/export.go b/internal/statsexport/export.go new file mode 100644 index 0000000..aba2a51 --- /dev/null +++ b/internal/statsexport/export.go @@ -0,0 +1,207 @@ +// Package statsexport writes stable, machine-readable run statistics. +package statsexport + +import ( + "encoding/csv" + "encoding/json" + "fmt" + "io" + "strconv" + "time" + + "github.com/atoonk/wireblast/internal/stats" +) + +// SchemaVersion changes only when the record contract changes incompatibly. +const SchemaVersion = 1 + +// Format selects the output encoding. +type Format string + +const ( + FormatCSV Format = "csv" + FormatJSONL Format = "jsonl" +) + +// Writer serializes snapshots. It is not safe for concurrent use. +type Writer struct { + format Format + csv *csv.Writer + json *json.Encoder +} + +// New creates an exporter and writes the CSV header when needed. +func New(format Format, out io.Writer) (*Writer, error) { + w := &Writer{format: format} + switch format { + case FormatCSV: + w.csv = csv.NewWriter(out) + if err := w.csv.Write(csvHeader); err != nil { + return nil, fmt.Errorf("write CSV header: %w", err) + } + w.csv.Flush() + if err := w.csv.Error(); err != nil { + return nil, fmt.Errorf("flush CSV header: %w", err) + } + case FormatJSONL: + w.json = json.NewEncoder(out) + default: + return nil, fmt.Errorf("unknown statistics format %q", format) + } + return w, nil +} + +// Sample writes one periodic aggregate record. +func (w *Writer) Sample(s *stats.Snapshot) error { + return w.write(aggregate("sample", s)) +} + +// Final writes the lifetime aggregate followed by one record per queue. +func (w *Writer) Final(s *stats.Snapshot) error { + if err := w.write(aggregate("final", s)); err != nil { + return err + } + for _, q := range s.Kernel.PerQueue { + queue := q.Queue + r := baseRecord("queue_final", s) + r.QueueID = &queue + r.KernelRXDescriptors = q.RxPackets + r.KernelTXDescriptors = q.TxPackets + r.RXDropped = q.RxDropped + r.RXRingFull = q.RxRingFull + if err := w.write(r); err != nil { + return err + } + } + return nil +} + +// Close flushes buffered output. +func (w *Writer) Close() error { + if w.csv == nil { + return nil + } + w.csv.Flush() + if err := w.csv.Error(); err != nil { + return fmt.Errorf("flush CSV statistics: %w", err) + } + return nil +} + +func (w *Writer) write(r record) error { + switch w.format { + case FormatCSV: + if err := w.csv.Write(r.csv()); err != nil { + return fmt.Errorf("write CSV statistics: %w", err) + } + w.csv.Flush() + if err := w.csv.Error(); err != nil { + return fmt.Errorf("flush CSV statistics: %w", err) + } + case FormatJSONL: + if err := w.json.Encode(r); err != nil { + return fmt.Errorf("write JSONL statistics: %w", err) + } + } + return nil +} + +type record struct { + SchemaVersion int `json:"schema_version"` + RecordType string `json:"record_type"` + TimestampUTC string `json:"timestamp_utc"` + Elapsed float64 `json:"elapsed_seconds"` + State string `json:"state"` + Transmits bool `json:"transmits"` + QueueID *int `json:"queue_id,omitempty"` + + TXPackets uint64 `json:"tx_packets"` + TXBytes uint64 `json:"tx_bytes"` + TXUDP uint64 `json:"tx_udp"` + TXTCP uint64 `json:"tx_tcp"` + TXOther uint64 `json:"tx_other"` + TXErrors uint64 `json:"tx_errors"` + TXPPS float64 `json:"tx_pps"` + TXL1BPS float64 `json:"tx_l1_bps"` + TXL2BPS float64 `json:"tx_l2_bps"` + + RXPackets uint64 `json:"rx_packets"` + RXBytes uint64 `json:"rx_bytes"` + RXUDP uint64 `json:"rx_udp"` + RXTCP uint64 `json:"rx_tcp"` + RXOther uint64 `json:"rx_other"` + RXDrops uint64 `json:"rx_drops"` + RXPPS float64 `json:"rx_pps"` + RXL1BPS float64 `json:"rx_l1_bps"` + RXL2BPS float64 `json:"rx_l2_bps"` + + KernelQueues int `json:"kernel_queues"` + KernelRXDescriptors uint64 `json:"kernel_rx_descriptors"` + KernelTXDescriptors uint64 `json:"kernel_tx_descriptors"` + RXDropped uint64 `json:"rx_dropped"` + RXRingFull uint64 `json:"rx_ring_full"` + RXFillRingEmpty uint64 `json:"rx_fill_ring_empty_descs"` + RXInvalidDescs uint64 `json:"rx_invalid_descs"` + TXInvalidDescs uint64 `json:"tx_invalid_descs"` + TXRingEmpty uint64 `json:"tx_ring_empty_descs"` +} + +func baseRecord(kind string, s *stats.Snapshot) record { + timestamp := "" + if !s.At.IsZero() { + timestamp = s.At.UTC().Format(time.RFC3339Nano) + } + return record{ + SchemaVersion: SchemaVersion, + RecordType: kind, + TimestampUTC: timestamp, + Elapsed: s.Elapsed.Seconds(), + State: s.State.String(), + Transmits: s.Transmits, + } +} + +func aggregate(kind string, s *stats.Snapshot) record { + r := baseRecord(kind, s) + r.TXPackets, r.TXBytes = s.TotalTX.Packets, s.TotalTX.Bytes + r.TXUDP, r.TXTCP, r.TXOther = s.TotalTX.UDP, s.TotalTX.TCP, s.TotalTX.Other + r.TXErrors = s.TotalTX.Errors + r.TXPPS, r.TXL1BPS, r.TXL2BPS = s.TXRate.PPS, s.TXRate.WireBPS, s.TXRate.FrameBPS + r.RXPackets, r.RXBytes = s.TotalRX.Packets, s.TotalRX.Bytes + r.RXUDP, r.RXTCP, r.RXOther = s.TotalRX.UDP, s.TotalRX.TCP, s.TotalRX.Other + r.RXDrops = s.TotalRX.Drops + r.RXPPS, r.RXL1BPS, r.RXL2BPS = s.RXRate.PPS, s.RXRate.WireBPS, s.RXRate.FrameBPS + r.KernelQueues = s.Kernel.Queues + r.KernelRXDescriptors, r.KernelTXDescriptors = s.Kernel.RxPackets, s.Kernel.TxPackets + r.RXDropped, r.RXRingFull = s.Kernel.RxDropped, s.Kernel.RxRingFull + r.RXFillRingEmpty, r.RXInvalidDescs = s.Kernel.RxFillRingEmpty, s.Kernel.RxInvalidDescs + r.TXInvalidDescs, r.TXRingEmpty = s.Kernel.TxInvalidDescs, s.Kernel.TxRingEmpty + return r +} + +var csvHeader = []string{ + "schema_version", "record_type", "timestamp_utc", "elapsed_seconds", "state", "transmits", "queue_id", + "tx_packets", "tx_bytes", "tx_udp", "tx_tcp", "tx_other", "tx_errors", "tx_pps", "tx_l1_bps", "tx_l2_bps", + "rx_packets", "rx_bytes", "rx_udp", "rx_tcp", "rx_other", "rx_drops", "rx_pps", "rx_l1_bps", "rx_l2_bps", + "kernel_queues", "kernel_rx_descriptors", "kernel_tx_descriptors", "rx_dropped", "rx_ring_full", "rx_fill_ring_empty_descs", + "rx_invalid_descs", "tx_invalid_descs", "tx_ring_empty_descs", +} + +func (r record) csv() []string { + queue := "" + if r.QueueID != nil { + queue = strconv.Itoa(*r.QueueID) + } + u := strconv.FormatUint + f := func(v float64) string { return strconv.FormatFloat(v, 'f', 6, 64) } + return []string{ + strconv.Itoa(r.SchemaVersion), r.RecordType, r.TimestampUTC, f(r.Elapsed), r.State, + strconv.FormatBool(r.Transmits), queue, + u(r.TXPackets, 10), u(r.TXBytes, 10), u(r.TXUDP, 10), u(r.TXTCP, 10), u(r.TXOther, 10), u(r.TXErrors, 10), + f(r.TXPPS), f(r.TXL1BPS), f(r.TXL2BPS), + u(r.RXPackets, 10), u(r.RXBytes, 10), u(r.RXUDP, 10), u(r.RXTCP, 10), u(r.RXOther, 10), u(r.RXDrops, 10), + f(r.RXPPS), f(r.RXL1BPS), f(r.RXL2BPS), + strconv.Itoa(r.KernelQueues), u(r.KernelRXDescriptors, 10), u(r.KernelTXDescriptors, 10), u(r.RXDropped, 10), + u(r.RXRingFull, 10), u(r.RXFillRingEmpty, 10), u(r.RXInvalidDescs, 10), u(r.TXInvalidDescs, 10), u(r.TXRingEmpty, 10), + } +} diff --git a/internal/statsexport/export_test.go b/internal/statsexport/export_test.go new file mode 100644 index 0000000..136877b --- /dev/null +++ b/internal/statsexport/export_test.go @@ -0,0 +1,227 @@ +package statsexport + +import ( + "bytes" + "encoding/csv" + "encoding/json" + "errors" + "io" + "strings" + "testing" + "time" + + "github.com/atoonk/wireblast/internal/stats" +) + +func fixture() *stats.Snapshot { + return &stats.Snapshot{ + At: time.Date(2026, 8, 10, 12, 30, 0, 123, time.FixedZone("test", 2*60*60)), + Elapsed: 1500 * time.Millisecond, + State: stats.StateRunning, + Transmits: true, + TotalTX: stats.Totals{ + Packets: 100, Bytes: 6400, UDP: 90, TCP: 5, Other: 5, Errors: 7, + }, + TotalRX: stats.Totals{ + Packets: 80, Bytes: 5120, UDP: 70, TCP: 5, Other: 5, Drops: 11, + }, + TXRate: stats.Rates{PPS: 99.5, FrameBPS: 50_944, WireBPS: 66_864}, + RXRate: stats.Rates{PPS: 79.5, FrameBPS: 40_704, WireBPS: 53_424}, + Kernel: stats.Kernel{ + Queues: 2, RxPackets: 80, TxPackets: 100, + RxDropped: 3, RxRingFull: 8, RxFillRingEmpty: 9, + RxInvalidDescs: 4, TxInvalidDescs: 7, TxRingEmpty: 6, + PerQueue: []stats.KernelQueue{ + {Queue: 0, RxPackets: 50, TxPackets: 60, RxDropped: 1, RxRingFull: 2}, + {Queue: 1, RxPackets: 30, TxPackets: 40, RxDropped: 2, RxRingFull: 6}, + }, + }, + } +} + +func TestCSVWritesStableAggregateAndQueueRecords(t *testing.T) { + var out bytes.Buffer + w, err := New(FormatCSV, &out) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := w.Sample(fixture()); err != nil { + t.Fatalf("Sample: %v", err) + } + if err := w.Final(fixture()); err != nil { + t.Fatalf("Final: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + rows, err := csv.NewReader(strings.NewReader(out.String())).ReadAll() + if err != nil { + t.Fatalf("ReadAll: %v\n%s", err, out.String()) + } + if len(rows) != 5 { + t.Fatalf("rows = %d, want header + sample + final + 2 queues", len(rows)) + } + header := index(rows[0]) + assertCSV(t, header, rows[1], "record_type", "sample") + assertCSV(t, header, rows[1], "timestamp_utc", "2026-08-10T10:30:00.000000123Z") + assertCSV(t, header, rows[1], "elapsed_seconds", "1.500000") + assertCSV(t, header, rows[1], "tx_packets", "100") + assertCSV(t, header, rows[1], "tx_l1_bps", "66864.000000") + assertCSV(t, header, rows[1], "rx_dropped", "3") + assertCSV(t, header, rows[1], "rx_fill_ring_empty_descs", "9") + assertCSV(t, header, rows[1], "tx_ring_empty_descs", "6") + assertCSV(t, header, rows[2], "record_type", "final") + assertCSV(t, header, rows[3], "record_type", "queue_final") + assertCSV(t, header, rows[3], "queue_id", "0") + assertCSV(t, header, rows[4], "queue_id", "1") +} + +func TestSchemaNamesRingProgressAsDescriptors(t *testing.T) { + var out bytes.Buffer + w, err := New(FormatCSV, &out) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := w.Final(fixture()); err != nil { + t.Fatalf("Final: %v", err) + } + + rows, err := csv.NewReader(strings.NewReader(out.String())).ReadAll() + if err != nil { + t.Fatalf("ReadAll: %v", err) + } + header := index(rows[0]) + for _, name := range []string{"kernel_rx_descriptors", "kernel_tx_descriptors"} { + if _, ok := header[name]; !ok { + t.Errorf("CSV header is missing descriptor field %q", name) + } + } + for _, old := range []string{"kernel_rx_packets", "kernel_tx_packets"} { + if _, ok := header[old]; ok { + t.Errorf("CSV header still calls ring descriptors packets: %q", old) + } + } + assertCSV(t, header, rows[1], "kernel_rx_descriptors", "80") + assertCSV(t, header, rows[1], "kernel_tx_descriptors", "100") + assertCSV(t, header, rows[2], "kernel_rx_descriptors", "50") + assertCSV(t, header, rows[2], "kernel_tx_descriptors", "60") +} + +func TestJSONLWritesTheSameVersionedSchema(t *testing.T) { + var out bytes.Buffer + w, err := New(FormatJSONL, &out) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := w.Sample(fixture()); err != nil { + t.Fatalf("Sample: %v", err) + } + if err := w.Final(fixture()); err != nil { + t.Fatalf("Final: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + dec := json.NewDecoder(&out) + var records []map[string]any + for { + var record map[string]any + if err := dec.Decode(&record); errors.Is(err, io.EOF) { + break + } else if err != nil { + t.Fatalf("Decode: %v", err) + } + records = append(records, record) + } + if len(records) != 4 { + t.Fatalf("records = %d, want sample + final + 2 queues", len(records)) + } + if got := records[0]["schema_version"]; got != float64(SchemaVersion) { + t.Errorf("schema_version = %v, want %d", got, SchemaVersion) + } + if got := records[0]["record_type"]; got != "sample" { + t.Errorf("record_type = %v, want sample", got) + } + if got := records[0]["rx_invalid_descs"]; got != float64(4) { + t.Errorf("rx_invalid_descs = %v, want 4", got) + } + if got := records[0]["rx_fill_ring_empty_descs"]; got != float64(9) { + t.Errorf("rx_fill_ring_empty_descs = %v, want 9", got) + } + if got := records[0]["tx_ring_empty_descs"]; got != float64(6) { + t.Errorf("tx_ring_empty_descs = %v, want 6", got) + } + if _, ok := records[0]["rx_fill_ring_empty"]; ok { + t.Error("JSONL record must use the exact rx_fill_ring_empty_descs UAPI name") + } + if _, ok := records[0]["tx_ring_empty"]; ok { + t.Error("JSONL record must use the exact tx_ring_empty_descs UAPI name") + } + if got := records[0]["kernel_rx_descriptors"]; got != float64(80) { + t.Errorf("kernel_rx_descriptors = %v, want 80", got) + } + if _, ok := records[0]["kernel_rx_packets"]; ok { + t.Error("JSONL record must not call ring descriptors packets") + } + if _, ok := records[0]["kernel_tx_packets"]; ok { + t.Error("JSONL record must not call ring descriptors packets") + } + if _, ok := records[0]["queue_id"]; ok { + t.Error("aggregate JSON record should omit queue_id") + } + if got := records[2]["queue_id"]; got != float64(0) { + t.Errorf("first queue_id = %v, want 0", got) + } +} + +func TestNewRejectsUnknownFormat(t *testing.T) { + if _, err := New(Format("xml"), io.Discard); err == nil { + t.Fatal("New(xml) should fail") + } +} + +func TestNewReportsCSVHeaderWriteFailure(t *testing.T) { + if _, err := New(FormatCSV, failWriter{}); err == nil { + t.Fatal("New(csv) should report a header write failure") + } +} + +func TestJSONLReportsRecordWriteFailure(t *testing.T) { + w, err := New(FormatJSONL, failWriter{}) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := w.Sample(fixture()); err == nil { + t.Fatal("Sample should report a record write failure") + } +} + +type failWriter struct{} + +func (failWriter) Write([]byte) (int, error) { + return 0, errors.New("disk full") +} + +func index(header []string) map[string]int { + out := make(map[string]int, len(header)) + for i, name := range header { + out[name] = i + } + return out +} + +func assertCSV(t *testing.T, header map[string]int, row []string, field, want string) { + t.Helper() + i, ok := header[field] + if !ok { + t.Fatalf("missing CSV field %q", field) + } + if i >= len(row) { + t.Fatalf("row has %d columns, field %q needs column %d", len(row), field, i) + } + if got := row[i]; got != want { + t.Errorf("%s = %q, want %q", field, got, want) + } +}