From e5c64188d7e719e65e2beeffab082058752b0f56 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 1/9] config: honour explicitly configured zero values A zero was indistinguishable from an omitted key, so every field whose zero value is a legitimate setting was silently replaced by its default: zipf: 0 -> 1.10 (README documents 0 as "equal base weights") total_noise: 0 -> 0.015 (noise could not be disabled) host_noise: 0 -> 0.25 correlation: 0 -> 0.98 (uncorrelated noise was unreachable) stddev: 0 -> 220 (constant frame size was unreachable) seed: 0 -> 1 sub_agent_id: 0 -> 1 in attack mode Make those fields pointers in the YAML structs and resolve the defaults only when the key is absent. RuntimePacket carries the resolved packet settings so the rest of the code keeps working with plain values. Fields whose zero value is invalid anyway (sizes, sampling_rate, tick, samples_per_datagram, hosts, max_samples_per_tick) are left as they are. Co-Authored-By: Claude Opus 5 --- config.go | 115 ++++++++++++++++++++++++++++--------------------- config_test.go | 64 ++++++++++++++++++++++++++- model_test.go | 4 +- packet.go | 4 +- 4 files changed, 134 insertions(+), 53 deletions(-) diff --git a/config.go b/config.go index 07a9ba3..d815d85 100644 --- a/config.go +++ b/config.go @@ -16,21 +16,31 @@ import ( "go.yaml.in/yaml/v3" ) +// Fields whose zero value is a meaningful setting are pointers, so an explicit +// 0 in the YAML is distinguishable from an omitted key. type ExporterConfig struct { - Collector string `yaml:"collector"` - AgentIP string `yaml:"agent_ip"` - SubAgentID uint32 `yaml:"sub_agent_id"` - SamplingRate uint32 `yaml:"sampling_rate"` - Tick string `yaml:"tick"` - SamplesPerDatagram int `yaml:"samples_per_datagram"` - Seed int64 `yaml:"seed"` + Collector string `yaml:"collector"` + AgentIP string `yaml:"agent_ip"` + SubAgentID *uint32 `yaml:"sub_agent_id"` + SamplingRate uint32 `yaml:"sampling_rate"` + Tick string `yaml:"tick"` + SamplesPerDatagram int `yaml:"samples_per_datagram"` + Seed *int64 `yaml:"seed"` } type PacketConfig struct { - MeanSize int `yaml:"mean_size"` - StdDev float64 `yaml:"stddev"` - MinSize int `yaml:"min_size"` - MaxSize int `yaml:"max_size"` + MeanSize int `yaml:"mean_size"` + StdDev *float64 `yaml:"stddev"` + MinSize int `yaml:"min_size"` + MaxSize int `yaml:"max_size"` +} + +// RuntimePacket is PacketConfig with defaults resolved. +type RuntimePacket struct { + MeanSize int + StdDev float64 + MinSize int + MaxSize int } type ProtocolMix struct { @@ -45,10 +55,10 @@ type AmbientNetworkConfig struct { OutgoingShare float64 `yaml:"outgoing_share"` Rate string `yaml:"rate"` Hosts int `yaml:"hosts"` - Zipf float64 `yaml:"zipf"` - TotalNoise float64 `yaml:"total_noise"` - HostNoise float64 `yaml:"host_noise"` - Correlation float64 `yaml:"correlation"` + Zipf *float64 `yaml:"zipf"` + TotalNoise *float64 `yaml:"total_noise"` + HostNoise *float64 `yaml:"host_noise"` + Correlation *float64 `yaml:"correlation"` PeerNetworks []string `yaml:"peer_networks"` } @@ -87,7 +97,7 @@ type RuntimeCommon struct { Tick time.Duration SamplesPerDatagram int Seed int64 - Packet PacketConfig + Packet RuntimePacket Protocols ProtocolMix LogInterval time.Duration MaxSamplesPerTick int @@ -159,18 +169,11 @@ func loadAmbientConfig(path string) (RuntimeAmbient, error) { if n.Hosts == 0 { n.Hosts = 128 } - if n.Zipf == 0 { - n.Zipf = 1.10 - } - if n.TotalNoise == 0 { - n.TotalNoise = 0.015 - } - if n.HostNoise == 0 { - n.HostNoise = 0.25 - } - if n.Correlation == 0 { - n.Correlation = 0.98 - } + // 0 is a valid setting here: uniform weights, no noise, no correlation. + zipf := valueOr(n.Zipf, 1.10) + totalNoise := valueOr(n.TotalNoise, 0.015) + hostNoise := valueOr(n.HostNoise, 0.25) + correlation := valueOr(n.Correlation, 0.98) prefix, err := netip.ParsePrefix(n.CIDR) if err != nil { @@ -188,13 +191,13 @@ func loadAmbientConfig(path string) (RuntimeAmbient, error) { if n.Hosts < 1 || n.Hosts > 65536 { return RuntimeAmbient{}, fmt.Errorf("networks[%d].hosts must be between 1 and 65536", i) } - if n.Zipf < 0 { + if zipf < 0 { return RuntimeAmbient{}, fmt.Errorf("networks[%d].zipf must be non-negative", i) } - if n.TotalNoise < 0 || n.HostNoise < 0 { + if totalNoise < 0 || hostNoise < 0 { return RuntimeAmbient{}, fmt.Errorf("networks[%d] noise values must be non-negative", i) } - if n.Correlation < 0 || n.Correlation >= 1 { + if correlation < 0 || correlation >= 1 { return RuntimeAmbient{}, fmt.Errorf("networks[%d].correlation must be in [0, 1)", i) } @@ -209,10 +212,10 @@ func loadAmbientConfig(path string) (RuntimeAmbient, error) { OutgoingShare: outgoingShare, RateBPS: rate, Hosts: n.Hosts, - Zipf: n.Zipf, - TotalNoise: n.TotalNoise, - HostNoise: n.HostNoise, - Correlation: n.Correlation, + Zipf: zipf, + TotalNoise: totalNoise, + HostNoise: hostNoise, + Correlation: correlation, PeerPrefixes: peers, }) } @@ -307,8 +310,8 @@ func applyCommonDefaults(exporter *ExporterConfig, packet *PacketConfig, protoco if exporter.AgentIP == "" { exporter.AgentIP = "192.0.2.10" } - if exporter.SubAgentID == 0 { - exporter.SubAgentID = defaultSubAgent + if exporter.SubAgentID == nil { + exporter.SubAgentID = &defaultSubAgent } if exporter.SamplingRate == 0 { exporter.SamplingRate = 4096 @@ -319,14 +322,14 @@ func applyCommonDefaults(exporter *ExporterConfig, packet *PacketConfig, protoco if exporter.SamplesPerDatagram == 0 { exporter.SamplesPerDatagram = 8 } - if exporter.Seed == 0 { - exporter.Seed = 1 + if exporter.Seed == nil { + exporter.Seed = ptr(int64(1)) } if packet.MeanSize == 0 { packet.MeanSize = 900 } - if packet.StdDev == 0 { - packet.StdDev = 220 + if packet.StdDev == nil { + packet.StdDev = ptr(220.0) } if packet.MinSize == 0 { packet.MinSize = 64 @@ -371,7 +374,7 @@ func validateCommon(exporter ExporterConfig, packet PacketConfig, protocols Prot if packet.MinSize < 64 || packet.MaxSize < 78 || packet.MaxSize > 9216 || packet.MinSize > packet.MaxSize { return RuntimeCommon{}, errors.New("packet sizes must satisfy 64 <= min_size <= max_size <= 9216 and max_size >= 78 for IPv6") } - if packet.MeanSize < packet.MinSize || packet.MeanSize > packet.MaxSize || packet.StdDev < 0 { + if packet.MeanSize < packet.MinSize || packet.MeanSize > packet.MaxSize || *packet.StdDev < 0 { return RuntimeCommon{}, errors.New("packet.mean_size/stddev are inconsistent with min_size/max_size") } if _, err := protocols.normalized(); err != nil { @@ -388,18 +391,34 @@ func validateCommon(exporter ExporterConfig, packet PacketConfig, protocols Prot return RuntimeCommon{ Collector: exporter.Collector, AgentIP: agentIP, - SubAgentID: exporter.SubAgentID, + SubAgentID: *exporter.SubAgentID, SamplingRate: exporter.SamplingRate, Tick: tick, SamplesPerDatagram: exporter.SamplesPerDatagram, - Seed: exporter.Seed, - Packet: packet, - Protocols: protocols, - LogInterval: logEvery, - MaxSamplesPerTick: maxSamples, + Seed: *exporter.Seed, + Packet: RuntimePacket{ + MeanSize: packet.MeanSize, + StdDev: *packet.StdDev, + MinSize: packet.MinSize, + MaxSize: packet.MaxSize, + }, + Protocols: protocols, + LogInterval: logEvery, + MaxSamplesPerTick: maxSamples, }, nil } +func ptr[T any](value T) *T { + return &value +} + +func valueOr[T any](value *T, fallback T) T { + if value == nil { + return fallback + } + return *value +} + func (p ProtocolMix) normalized() (ProtocolMix, error) { if p.TCP < 0 || p.UDP < 0 || p.ICMP < 0 { return ProtocolMix{}, errors.New("protocol weights cannot be negative") diff --git a/config_test.go b/config_test.go index be1313e..aab1655 100644 --- a/config_test.go +++ b/config_test.go @@ -1,6 +1,10 @@ package main -import "testing" +import ( + "os" + "path/filepath" + "testing" +) func TestParseBitRate(t *testing.T) { tests := map[string]float64{ @@ -34,3 +38,61 @@ func TestParseAmbientShares(t *testing.T) { t.Fatal("expected zero shares to be rejected") } } + +func TestExplicitZeroValuesAreHonoured(t *testing.T) { + path := filepath.Join(t.TempDir(), "ambient.yaml") + body := `exporter: + seed: 0 + sub_agent_id: 0 +packet: + stddev: 0 +networks: + - cidr: "10.0.0.0/24" + incoming_share: 1 + outgoing_share: 0 + rate: "100Mbps" + zipf: 0 + total_noise: 0 + host_noise: 0 + correlation: 0 +` + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + cfg, err := loadAmbientConfig(path) + if err != nil { + t.Fatal(err) + } + if cfg.Common.Seed != 0 || cfg.Common.Packet.StdDev != 0 { + t.Fatalf("seed=%d stddev=%f, expected both 0", cfg.Common.Seed, cfg.Common.Packet.StdDev) + } + network := cfg.Networks[0] + if network.Zipf != 0 || network.TotalNoise != 0 || network.HostNoise != 0 || network.Correlation != 0 { + t.Fatalf("zipf=%f total_noise=%f host_noise=%f correlation=%f, expected all 0", + network.Zipf, network.TotalNoise, network.HostNoise, network.Correlation) + } +} + +func TestOmittedValuesUseDefaults(t *testing.T) { + path := filepath.Join(t.TempDir(), "ambient.yaml") + body := `networks: + - cidr: "10.0.0.0/24" + incoming_share: 1 + outgoing_share: 0 + rate: "100Mbps" +` + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + cfg, err := loadAmbientConfig(path) + if err != nil { + t.Fatal(err) + } + if cfg.Common.Seed != 1 || cfg.Common.Packet.StdDev != 220 { + t.Fatalf("seed=%d stddev=%f, expected 1/220", cfg.Common.Seed, cfg.Common.Packet.StdDev) + } + network := cfg.Networks[0] + if network.Zipf != 1.10 || network.TotalNoise != 0.015 || network.HostNoise != 0.25 || network.Correlation != 0.98 { + t.Fatalf("defaults not applied: %+v", network) + } +} diff --git a/model_test.go b/model_test.go index fbd78b9..0eb55eb 100644 --- a/model_test.go +++ b/model_test.go @@ -40,7 +40,7 @@ func TestAmbientOneSampleBudget(t *testing.T) { Common: RuntimeCommon{ SamplingRate: 10, Seed: 1, - Packet: PacketConfig{ + Packet: RuntimePacket{ MeanSize: 1000, MinSize: 1000, MaxSize: 1000, @@ -98,7 +98,7 @@ func TestAmbientBidirectionalShares(t *testing.T) { Common: RuntimeCommon{ SamplingRate: 10, Seed: 1, - Packet: PacketConfig{ + Packet: RuntimePacket{ MeanSize: 1000, MinSize: 1000, MaxSize: 1000, diff --git a/packet.go b/packet.go index 8f52477..51a1b74 100644 --- a/packet.go +++ b/packet.go @@ -43,11 +43,11 @@ func (c ProtocolChooser) Next(rng *rand.Rand) Protocol { } type PacketSizeSampler struct { - Config PacketConfig + Config RuntimePacket RNG *rand.Rand } -func NewPacketSizeSampler(cfg PacketConfig, rng *rand.Rand) PacketSizeSampler { +func NewPacketSizeSampler(cfg RuntimePacket, rng *rand.Rand) PacketSizeSampler { return PacketSizeSampler{Config: cfg, RNG: rng} } From 12fc127ba15052117985d97a106a55f63e3592c1 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 2/9] address: drop dead code uint16Bytes had no callers, and both branches of the AddrFromSlice conditional in addressAt were identical. Co-Authored-By: Claude Opus 5 --- address.go | 15 +-------------- 1 file changed, 1 insertion(+), 14 deletions(-) diff --git a/address.go b/address.go index d8b747f..94fe94f 100644 --- a/address.go +++ b/address.go @@ -1,7 +1,6 @@ package main import ( - "encoding/binary" "fmt" "math/big" "math/rand" @@ -90,13 +89,7 @@ func addressAt(prefix netip.Prefix, offset uint64) (netip.Addr, error) { padded := make([]byte, width) copy(padded[width-len(encoded):], encoded) - var addr netip.Addr - var ok bool - if width == 4 { - addr, ok = netip.AddrFromSlice(padded) - } else { - addr, ok = netip.AddrFromSlice(padded) - } + addr, ok := netip.AddrFromSlice(padded) if !ok || !prefix.Contains(addr) { return netip.Addr{}, fmt.Errorf("generated address %s is outside prefix %s", addr, prefix) } @@ -118,9 +111,3 @@ func macFromAddress(addr netip.Addr, discriminator byte) [6]byte { mac[5] ^= v[15] return mac } - -func uint16Bytes(value uint16) []byte { - buf := make([]byte, 2) - binary.BigEndian.PutUint16(buf, value) - return buf -} From a307d4b825dd70b41d51ab6ce9b2727afc7cbd8d Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 3/9] packet: keep one stable MAC address per host The MAC discriminator was chosen by the role of the address in the packet (0x11 source, 0x22 destination) rather than by its role in the topology, so each host appeared with two different MAC addresses depending on direction, and every MAC appeared to move between switch ports. Derive it from the internal/peer role instead, which is stable across directions. Co-Authored-By: Claude Opus 5 --- packet.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/packet.go b/packet.go index 51a1b74..25207a1 100644 --- a/packet.go +++ b/packet.go @@ -86,8 +86,13 @@ func buildPacketHeader(spec PacketSpec, rng *rand.Rand) ([]byte, error) { } frameLength := max(spec.FrameLength, minimumFrameLength(source.Is6(), spec.Protocol)) - sourceMAC := macFromAddress(source, 0x11) - destinationMAC := macFromAddress(destination, 0x22) + // The discriminator follows the address role in the topology, not in this + // packet, so a host keeps the same MAC in both directions. + sourceMAC := macFromAddress(spec.Peer, 0x22) + destinationMAC := macFromAddress(spec.Internal, 0x11) + if spec.Direction == DirectionOutgoing { + sourceMAC, destinationMAC = destinationMAC, sourceMAC + } header := make([]byte, 14) copy(header[0:6], destinationMAC[:]) From 89b619ad558ee118c1a8afdaf660bbc09ef7b2de Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 4/9] runner: report the attack phase after the ramp-down ends attackPhase returned "ramp-down" for everything past the hold phase, including configurations without a ramp_down and the final tick after the attack has finished. Co-Authored-By: Claude Opus 5 --- runner.go | 6 +++++- runner_test.go | 26 ++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) create mode 100644 runner_test.go diff --git a/runner.go b/runner.go index e7dc935..319eda8 100644 --- a/runner.go +++ b/runner.go @@ -160,7 +160,11 @@ func attackPhase(elapsed time.Duration, cfg RuntimeAttack) string { if elapsed < cfg.Hold { return "hold" } - return "ramp-down" + elapsed -= cfg.Hold + if elapsed < cfg.RampDown { + return "ramp-down" + } + return "done" } func measuredRate(bits uint64, interval time.Duration) string { diff --git a/runner_test.go b/runner_test.go new file mode 100644 index 0000000..b0df048 --- /dev/null +++ b/runner_test.go @@ -0,0 +1,26 @@ +package main + +import ( + "testing" + "time" +) + +func TestAttackPhase(t *testing.T) { + cfg := RuntimeAttack{RampUp: 10 * time.Second, Hold: 20 * time.Second, RampDown: 10 * time.Second} + checks := map[time.Duration]string{ + 0: "ramp-up", + 15 * time.Second: "hold", + 35 * time.Second: "ramp-down", + 40 * time.Second: "done", + } + for at, want := range checks { + if got := attackPhase(at, cfg); got != want { + t.Fatalf("attackPhase(%s)=%s, expected %s", at, got, want) + } + } + + // Without a ramp-down phase the attack is done once the hold ends. + if got := attackPhase(1500*time.Millisecond, RuntimeAttack{Hold: time.Second}); got != "done" { + t.Fatalf("attackPhase without ramp_down=%s, expected done", got) + } +} From 14f6f8a90912fba767189a1066d2e40cf9ab02e0 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 5/9] exporter: do not exit when the collector is unreachable net.DialUDP returns a connected socket, so an ICMP port unreachable from the collector surfaces as an error on the next write. That error was propagated out of Flush and terminated the process: mode=ambient collector=127.0.0.1:6343 ... duration=infinite error: send sFlow datagram: write udp ...: connect: connection refused exit status 1 The generator therefore died ~200ms after startup if the collector was not listening yet, and any collector restart killed a running ambient profile. Treat a failed send like a lost sample: count it, drop the datagram, log the first failure and then at most one line every 10s, and log the recovery. Encoding errors stay fatal. The count is reported as send_errors in the periodic log so silent drops remain visible. Co-Authored-By: Claude Opus 5 --- exporter.go | 43 ++++++++++++++++++++++++++++++++++++++++++- exporter_test.go | 47 +++++++++++++++++++++++++++++++++++++++++++++++ runner.go | 8 ++++---- 3 files changed, 93 insertions(+), 5 deletions(-) create mode 100644 exporter_test.go diff --git a/exporter.go b/exporter.go index c53ef51..25bbbce 100644 --- a/exporter.go +++ b/exporter.go @@ -3,6 +3,7 @@ package main import ( "bytes" "fmt" + "log" "math/rand" "net" "time" @@ -12,9 +13,13 @@ import ( const sflowHeaderProtocolEthernet = 1 +// sendErrorLogInterval throttles logging while the collector is unreachable. +const sendErrorLogInterval = 10 * time.Second + type SFlowStats struct { Samples uint64 Datagrams uint64 + SendErrors uint64 EstimatedBits uint64 EstimatedIncomingBits uint64 EstimatedOutgoingBits uint64 @@ -31,6 +36,10 @@ type SFlowExporter struct { pending []sflow.Sample rng *rand.Rand stats SFlowStats + sendFailing bool + lastSendError time.Time + lastSendErrorLog time.Time + suppressedErrors uint64 } func NewSFlowExporter(cfg RuntimeCommon) (*SFlowExporter, error) { @@ -108,14 +117,46 @@ func (e *SFlowExporter) Flush() error { if err := e.encoder.Encode(&payload, e.pending); err != nil { return fmt.Errorf("encode sFlow datagram: %w", err) } + // A connected UDP socket surfaces an ICMP port unreachable as a write + // error, so a collector that is down or restarting must not stop the + // generator: drop the datagram and keep going. if _, err := e.connection.Write(payload.Bytes()); err != nil { - return fmt.Errorf("send sFlow datagram: %w", err) + e.stats.SendErrors++ + e.pending = e.pending[:0] + e.noteSendError(err) + return nil + } + // A single success is not a recovery: the kernel reports an ICMP error on + // the write that follows it, so writes alternate while the collector is + // down. Only a quiet period counts. + if e.sendFailing && time.Since(e.lastSendError) >= sendErrorLogInterval { + log.Printf("sFlow collector reachable again") + e.sendFailing = false + e.suppressedErrors = 0 } e.stats.Datagrams++ e.pending = e.pending[:0] return nil } +func (e *SFlowExporter) noteSendError(err error) { + now := time.Now() + e.lastSendError = now + if !e.sendFailing { + e.sendFailing = true + e.lastSendErrorLog = now + e.suppressedErrors = 0 + log.Printf("warning: send sFlow datagram: %v (continuing; further errors are summarized every %s)", err, sendErrorLogInterval) + return + } + e.suppressedErrors++ + if now.Sub(e.lastSendErrorLog) >= sendErrorLogInterval { + log.Printf("warning: sFlow collector still unreachable, %d datagram(s) dropped in the last %s: %v", e.suppressedErrors, now.Sub(e.lastSendErrorLog).Truncate(time.Second), err) + e.lastSendErrorLog = now + e.suppressedErrors = 0 + } +} + func (e *SFlowExporter) SnapshotAndReset() SFlowStats { stats := e.stats e.stats = SFlowStats{} diff --git a/exporter_test.go b/exporter_test.go new file mode 100644 index 0000000..40665fa --- /dev/null +++ b/exporter_test.go @@ -0,0 +1,47 @@ +package main + +import ( + "net" + "net/netip" + "testing" +) + +// A collector that is down must not terminate the generator. +func TestExporterSurvivesSendFailure(t *testing.T) { + listener, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}) + if err != nil { + t.Fatal(err) + } + defer listener.Close() + + exporter, err := NewSFlowExporter(RuntimeCommon{ + Collector: listener.LocalAddr().String(), + AgentIP: net.ParseIP("192.0.2.10"), + SamplingRate: 4096, + SamplesPerDatagram: 1, + Seed: 1, + }) + if err != nil { + t.Fatal(err) + } + if err := exporter.connection.Close(); err != nil { + t.Fatal(err) + } + + spec := PacketSpec{ + Internal: netip.MustParseAddr("10.0.0.42"), + Peer: netip.MustParseAddr("198.18.0.1"), + Direction: DirectionIncoming, + Protocol: ProtocolUDP, + FrameLength: 900, + } + for i := 0; i < 3; i++ { + if err := exporter.Add(spec); err != nil { + t.Fatalf("send failure must not be fatal: %v", err) + } + } + stats := exporter.SnapshotAndReset() + if stats.SendErrors != 3 || stats.Samples != 3 || stats.Datagrams != 0 { + t.Fatalf("stats=%+v, expected 3 samples, 3 send errors, 0 datagrams", stats) + } +} diff --git a/runner.go b/runner.go index 319eda8..f8337cb 100644 --- a/runner.go +++ b/runner.go @@ -69,10 +69,10 @@ func runAmbient(parent context.Context, cfg RuntimeAmbient) error { if len(top) > 0 { topText = fmt.Sprintf("%s@%s", top[0].Address, formatRate(top[0].LastRate)) } - log.Printf("mode=ambient target=%s target_in=%s target_out=%s estimated=%s estimated_in=%s estimated_out=%s samples=%d datagrams=%d top=%s", + log.Printf("mode=ambient target=%s target_in=%s target_out=%s estimated=%s estimated_in=%s estimated_out=%s samples=%d datagrams=%d send_errors=%d top=%s", formatRate(result.TargetBPS), formatRate(result.TargetIncomingBPS), formatRate(result.TargetOutgoingBPS), measuredRate(stats.EstimatedBits, interval), measuredRate(stats.EstimatedIncomingBits, interval), measuredRate(stats.EstimatedOutgoingBits, interval), - stats.Samples, stats.Datagrams, topText) + stats.Samples, stats.Datagrams, stats.SendErrors, topText) lastLog = time.Now() } } @@ -143,9 +143,9 @@ func runAttack(ctx context.Context, cfg RuntimeAttack) error { if now.Sub(lastLog) >= cfg.Common.LogInterval { stats := exporter.SnapshotAndReset() - log.Printf("mode=attack victim=%s target=%s estimated=%s samples=%d datagrams=%d phase=%s", + log.Printf("mode=attack victim=%s target=%s estimated=%s samples=%d datagrams=%d send_errors=%d phase=%s", cfg.Victim, formatRate(result.TargetBPS), measuredRate(stats.EstimatedBits, now.Sub(lastLog)), - stats.Samples, stats.Datagrams, attackPhase(elapsed, cfg)) + stats.Samples, stats.Datagrams, stats.SendErrors, attackPhase(elapsed, cfg)) lastLog = now } } From dc7b3ef280f46ce1c39014903387b2c2fb39465a Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 6/9] runner: credit ambient ticks with the measured interval runAmbient passed the configured tick to model.Tick while runAttack already used the measured one. time.Ticker drops ticks when the consumer falls behind, so under load ambient credited less time than actually elapsed and permanently under-generated the configured rate with no way to catch up. Co-Authored-By: Claude Opus 5 --- runner.go | 34 +++++++++++++++++++++++++++------- runner_test.go | 20 ++++++++++++++++++++ 2 files changed, 47 insertions(+), 7 deletions(-) diff --git a/runner.go b/runner.go index f8337cb..2ebefd0 100644 --- a/runner.go +++ b/runner.go @@ -7,6 +7,18 @@ import ( "time" ) +// maxCatchUpTicks allows short scheduler delays to be credited without letting +// a long pause turn into a sustained burst after the generator recovers. +const maxCatchUpTicks = 4 + +func boundedCreditInterval(elapsed, tick time.Duration) time.Duration { + maximum := time.Duration(maxCatchUpTicks) * tick + if elapsed > maximum { + return maximum + } + return elapsed +} + func runAmbient(parent context.Context, cfg RuntimeAmbient) error { ctx := parent var cancel context.CancelFunc @@ -35,7 +47,9 @@ func runAmbient(parent context.Context, cfg RuntimeAmbient) error { ticker := time.NewTicker(cfg.Common.Tick) defer ticker.Stop() - lastLog := time.Now() + start := time.Now() + lastTick := start + lastLog := start for { select { @@ -44,8 +58,14 @@ func runAmbient(parent context.Context, cfg RuntimeAmbient) error { return parent.Err() } return nil - case <-ticker.C: - result, err := model.Tick(cfg.Common.Tick, cfg.Common.MaxSamplesPerTick) + case now := <-ticker.C: + // Measured, not nominal: time.Ticker drops ticks when the loop + // falls behind, and crediting cfg.Common.Tick for a longer + // interval silently under-generates the configured rate. + interval := boundedCreditInterval(now.Sub(lastTick), cfg.Common.Tick) + lastTick = now + + result, err := model.Tick(interval, cfg.Common.MaxSamplesPerTick) if err != nil { return err } @@ -61,8 +81,8 @@ func runAmbient(parent context.Context, cfg RuntimeAmbient) error { log.Printf("warning: ambient reached max_samples_per_tick=%d; generated rate may be clipped", cfg.Common.MaxSamplesPerTick) } - if time.Since(lastLog) >= cfg.Common.LogInterval { - interval := time.Since(lastLog) + if now.Sub(lastLog) >= cfg.Common.LogInterval { + interval := now.Sub(lastLog) stats := exporter.SnapshotAndReset() top := sortedTopHosts(model.Networks, 1) topText := "none" @@ -73,7 +93,7 @@ func runAmbient(parent context.Context, cfg RuntimeAmbient) error { formatRate(result.TargetBPS), formatRate(result.TargetIncomingBPS), formatRate(result.TargetOutgoingBPS), measuredRate(stats.EstimatedBits, interval), measuredRate(stats.EstimatedIncomingBits, interval), measuredRate(stats.EstimatedOutgoingBits, interval), stats.Samples, stats.Datagrams, stats.SendErrors, topText) - lastLog = time.Now() + lastLog = now } } } @@ -111,7 +131,7 @@ func runAttack(ctx context.Context, cfg RuntimeAttack) error { return ctx.Err() case now := <-ticker.C: elapsed := now.Sub(start) - interval := now.Sub(lastTick) + interval := boundedCreditInterval(now.Sub(lastTick), cfg.Common.Tick) lastTick = now if elapsed > totalDuration { if err := exporter.Flush(); err != nil { diff --git a/runner_test.go b/runner_test.go index b0df048..8002d8c 100644 --- a/runner_test.go +++ b/runner_test.go @@ -5,6 +5,26 @@ import ( "time" ) +func TestBoundedCreditInterval(t *testing.T) { + tick := 100 * time.Millisecond + checks := []struct { + name string + elapsed time.Duration + want time.Duration + }{ + {name: "on time", elapsed: tick, want: tick}, + {name: "short delay", elapsed: 250 * time.Millisecond, want: 250 * time.Millisecond}, + {name: "long stall", elapsed: 30 * time.Second, want: maxCatchUpTicks * tick}, + } + for _, check := range checks { + t.Run(check.name, func(t *testing.T) { + if got := boundedCreditInterval(check.elapsed, tick); got != check.want { + t.Fatalf("boundedCreditInterval(%s, %s)=%s, expected %s", check.elapsed, tick, got, check.want) + } + }) + } +} + func TestAttackPhase(t *testing.T) { cfg := RuntimeAttack{RampUp: 10 * time.Second, Hold: 20 * time.Second, RampDown: 10 * time.Second} checks := map[time.Duration]string{ From d13e3607b7384098a139a6ca65198fc7d2d5a162 Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 7/9] runner: summarize clipped ticks instead of warning on every tick The max_samples_per_tick warning was logged on every affected tick, which is ten lines per second at the default tick of 100ms. Warn once and report the number of affected ticks as clipped in the periodic log line instead. Co-Authored-By: Claude Opus 5 --- README.md | 8 +++++--- runner.go | 31 +++++++++++++++++++++++-------- 2 files changed, 28 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 20f3f4e..197b335 100644 --- a/README.md +++ b/README.md @@ -378,17 +378,19 @@ protocols: Ambient example: ```text -mode=ambient target=3.01Gbps target_in=2.25Gbps target_out=758.00Mbps estimated=3.00Gbps estimated_in=2.24Gbps estimated_out=760.00Mbps samples=102 datagrams=13 top=10.10.0.17@402.10Mbps +mode=ambient target=3.01Gbps target_in=2.25Gbps target_out=758.00Mbps estimated=3.00Gbps estimated_in=2.24Gbps estimated_out=760.00Mbps samples=102 datagrams=13 clipped=0 send_errors=0 top=10.10.0.17@402.10Mbps ``` Attack example: ```text -mode=attack victim=10.10.0.42 target=9.62Gbps estimated=9.59Gbps samples=326 datagrams=41 phase=ramp-up +mode=attack victim=10.10.0.42 target=9.62Gbps estimated=9.59Gbps samples=326 datagrams=41 clipped=0 send_errors=0 phase=ramp-up ``` `target` is the traffic rate requested by the model. `estimated` is the traffic volume represented by the emitted samples, not the actual bandwidth consumed by the generator's UDP datagrams. +`clipped` counts the ticks in the interval that reached `max_samples_per_tick`. `send_errors` counts datagrams dropped because the collector could not be reached; an unreachable collector is logged but does not stop the generator, which resumes sending as soon as the collector accepts datagrams again. + ## Safety limits `max_samples_per_tick` limits the amount of work performed during a single model update: @@ -397,7 +399,7 @@ mode=attack victim=10.10.0.42 target=9.62Gbps estimated=9.59Gbps samples=326 dat max_samples_per_tick: 100000 ``` -A configuration with a very high represented rate, a low sampling rate, small frames, or a very short tick can reach this limit. The generator then emits a warning and caps the generated sample count for that tick. +A configuration with a very high represented rate, a low sampling rate, small frames, or a very short tick can reach this limit. The generator then warns once, counts the affected ticks in the `clipped` field of the periodic log, and caps the generated sample count for those ticks. Clipped ticks are shared out across hosts and networks, so no network is starved, and the reported `target` still describes the full configuration. ## Current limitations diff --git a/runner.go b/runner.go index 2ebefd0..d2ae793 100644 --- a/runner.go +++ b/runner.go @@ -50,6 +50,8 @@ func runAmbient(parent context.Context, cfg RuntimeAmbient) error { start := time.Now() lastTick := start lastLog := start + clippedTicks := 0 + clippingWarned := false for { select { @@ -78,7 +80,11 @@ func runAmbient(parent context.Context, cfg RuntimeAmbient) error { return err } if len(result.Packets) >= cfg.Common.MaxSamplesPerTick { - log.Printf("warning: ambient reached max_samples_per_tick=%d; generated rate may be clipped", cfg.Common.MaxSamplesPerTick) + clippedTicks++ + if !clippingWarned { + log.Printf("warning: ambient reached max_samples_per_tick=%d; generated rate may be clipped (further ticks are counted in the periodic log)", cfg.Common.MaxSamplesPerTick) + clippingWarned = true + } } if now.Sub(lastLog) >= cfg.Common.LogInterval { @@ -89,11 +95,12 @@ func runAmbient(parent context.Context, cfg RuntimeAmbient) error { if len(top) > 0 { topText = fmt.Sprintf("%s@%s", top[0].Address, formatRate(top[0].LastRate)) } - log.Printf("mode=ambient target=%s target_in=%s target_out=%s estimated=%s estimated_in=%s estimated_out=%s samples=%d datagrams=%d send_errors=%d top=%s", + log.Printf("mode=ambient target=%s target_in=%s target_out=%s estimated=%s estimated_in=%s estimated_out=%s samples=%d datagrams=%d clipped=%d send_errors=%d top=%s", formatRate(result.TargetBPS), formatRate(result.TargetIncomingBPS), formatRate(result.TargetOutgoingBPS), measuredRate(stats.EstimatedBits, interval), measuredRate(stats.EstimatedIncomingBits, interval), measuredRate(stats.EstimatedOutgoingBits, interval), - stats.Samples, stats.Datagrams, stats.SendErrors, topText) + stats.Samples, stats.Datagrams, clippedTicks, stats.SendErrors, topText) lastLog = now + clippedTicks = 0 } } } @@ -122,6 +129,8 @@ func runAttack(ctx context.Context, cfg RuntimeAttack) error { start := time.Now() lastTick := start lastLog := start + clippedTicks := 0 + clippingWarned := false ticker := time.NewTicker(cfg.Common.Tick) defer ticker.Stop() @@ -139,8 +148,9 @@ func runAttack(ctx context.Context, cfg RuntimeAttack) error { } stats := exporter.SnapshotAndReset() if stats.Samples > 0 { - log.Printf("mode=attack target=0bps estimated=%s samples=%d datagrams=%d phase=done", - measuredRate(stats.EstimatedBits, now.Sub(lastLog)), stats.Samples, stats.Datagrams) + log.Printf("mode=attack victim=%s target=0bps estimated=%s samples=%d datagrams=%d clipped=%d send_errors=%d phase=done", + cfg.Victim, measuredRate(stats.EstimatedBits, now.Sub(lastLog)), stats.Samples, stats.Datagrams, + clippedTicks, stats.SendErrors) } return nil } @@ -158,15 +168,20 @@ func runAttack(ctx context.Context, cfg RuntimeAttack) error { return err } if len(result.Packets) >= cfg.Common.MaxSamplesPerTick { - log.Printf("warning: attack reached max_samples_per_tick=%d; generated rate may be clipped", cfg.Common.MaxSamplesPerTick) + clippedTicks++ + if !clippingWarned { + log.Printf("warning: attack reached max_samples_per_tick=%d; generated rate may be clipped (further ticks are counted in the periodic log)", cfg.Common.MaxSamplesPerTick) + clippingWarned = true + } } if now.Sub(lastLog) >= cfg.Common.LogInterval { stats := exporter.SnapshotAndReset() - log.Printf("mode=attack victim=%s target=%s estimated=%s samples=%d datagrams=%d send_errors=%d phase=%s", + log.Printf("mode=attack victim=%s target=%s estimated=%s samples=%d datagrams=%d clipped=%d send_errors=%d phase=%s", cfg.Victim, formatRate(result.TargetBPS), measuredRate(stats.EstimatedBits, now.Sub(lastLog)), - stats.Samples, stats.Datagrams, stats.SendErrors, attackPhase(elapsed, cfg)) + stats.Samples, stats.Datagrams, clippedTicks, stats.SendErrors, attackPhase(elapsed, cfg)) lastLog = now + clippedTicks = 0 } } } From da48b543580b758071d07ba4afcd9d848a2b9e5c Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 8/9] model: cap the sample budget backlog SampleBudget.Bytes grew without bound whenever samples could not be emitted, either because max_samples_per_tick clipped the tick or because the process was delayed. The arrears were spent later as one artificial burst. Cap the backlog at four intervals of credit, with a floor of two maximum-size frames so that a very low-rate host can still accumulate enough for a single sample. Co-Authored-By: Claude Opus 5 --- model.go | 21 ++++++++++++++++----- model_test.go | 21 +++++++++++++++++++++ 2 files changed, 37 insertions(+), 5 deletions(-) diff --git a/model.go b/model.go index 8bb173a..f782cfe 100644 --- a/model.go +++ b/model.go @@ -25,13 +25,23 @@ func (a *AR1) Step(rng *rand.Rand) float64 { return a.Value } +// budgetBacklogTicks bounds the arrears a budget may carry between ticks. +const budgetBacklogTicks = 4 + type SampleBudget struct { Bytes float64 NextFrameSize int } -func (b *SampleBudget) AddRate(rateBPS float64, interval time.Duration, samplingRate uint32) { - b.Bytes += rateBPS / 8 * interval.Seconds() / float64(samplingRate) +// AddRate credits one interval of traffic. The backlog is capped so that a +// clipped or delayed tick is not repaid later as a single burst; the floor of +// two maximum frames keeps low-rate hosts able to reach one sample. +func (b *SampleBudget) AddRate(rateBPS float64, interval time.Duration, samplingRate uint32, maxFrameSize int) { + credit := rateBPS / 8 * interval.Seconds() / float64(samplingRate) + b.Bytes += credit + if limit := math.Max(credit*budgetBacklogTicks, float64(2*maxFrameSize)); b.Bytes > limit { + b.Bytes = limit + } } func (b *SampleBudget) Take(sampler *PacketSizeSampler, minimum int) (int, bool) { @@ -137,6 +147,7 @@ func NewAmbientModel(cfg RuntimeAmbient) (*AmbientModel, error) { func (m *AmbientModel) Tick(interval time.Duration, maxSamples int) (TickResult, error) { result := TickResult{Packets: make([]PacketSpec, 0, 128)} + maxFrameSize := m.Sampler.Config.MaxSize for networkIndex := range m.Networks { network := &m.Networks[networkIndex] @@ -169,8 +180,8 @@ func (m *AmbientModel) Tick(interval time.Duration, maxSamples int) (TickResult, result.TopAddress = host.Address } - host.IncomingBudget.AddRate(host.LastIncomingRate, interval, m.SamplingRate) - host.OutgoingBudget.AddRate(host.LastOutgoingRate, interval, m.SamplingRate) + host.IncomingBudget.AddRate(host.LastIncomingRate, interval, m.SamplingRate, maxFrameSize) + host.OutgoingBudget.AddRate(host.LastOutgoingRate, interval, m.SamplingRate, maxFrameSize) for len(result.Packets) < maxSamples { generated := false @@ -272,7 +283,7 @@ func (m *AttackModel) RateAt(elapsed time.Duration) float64 { func (m *AttackModel) Tick(elapsed, interval time.Duration, maxSamples int) (TickResult, error) { rate := m.RateAt(elapsed) result := TickResult{TargetBPS: rate, TopAddress: m.Config.Victim, TopRateBPS: rate} - m.Budget.AddRate(rate, interval, m.Config.Common.SamplingRate) + m.Budget.AddRate(rate, interval, m.Config.Common.SamplingRate, m.Config.Common.Packet.MaxSize) for len(result.Packets) < maxSamples { protocol := m.Chooser.Next(m.RNG) diff --git a/model_test.go b/model_test.go index 0eb55eb..28a3f3f 100644 --- a/model_test.go +++ b/model_test.go @@ -1,6 +1,7 @@ package main import ( + "math" "math/rand" "net/netip" "testing" @@ -140,3 +141,23 @@ func TestAmbientBidirectionalShares(t *testing.T) { t.Fatalf("target split=%f/%f, expected 240000/80000", result.TargetIncomingBPS, result.TargetOutgoingBPS) } } + +func TestSampleBudgetBacklogIsCapped(t *testing.T) { + var budget SampleBudget + credit := 1e9 / 8 * 0.1 / 1000 + for i := 0; i < 100; i++ { + budget.AddRate(1e9, 100*time.Millisecond, 1000, 1518) + } + if budget.Bytes > credit*budgetBacklogTicks { + t.Fatalf("backlog=%f, expected at most %f", budget.Bytes, credit*budgetBacklogTicks) + } + + // A low-rate host must still be able to accumulate one frame. + var slow SampleBudget + for i := 0; i < 1000; i++ { + slow.AddRate(1000, 100*time.Millisecond, 1000, 1518) + } + if math.Abs(slow.Bytes-12.5) > 1e-9 { + t.Fatalf("low-rate backlog=%f, expected 12.5", slow.Bytes) + } +} From 0618e1b4f4be951a616f04efe0a6311cfac9bfbf Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Sat, 15 Aug 2026 23:53:31 +0300 Subject: [PATCH 9/9] model: stop max_samples_per_tick from starving whole networks The emission loop returned from the middle of the per-host loop once maxSamples was reached, so every host and network after that point never reached AddRate at all. Their traffic was not deferred, it was lost, and because TargetBPS is accumulated in the same loop the reported target dropped with it: the log showed target and estimated in agreement while half the configured traffic was missing. With two identical 1Gbps networks and a clip of 5 samples per tick: before: netA=100 netB=0 target=1.01Gbps (configured total 2Gbps) after: netA=52 netB=48 target=2.00Gbps Account for every host first, then emit in a separate pass that resumes from where the previous tick was cut off, so clipping is shared out across hosts and networks instead of always falling on the last ones. Co-Authored-By: Claude Opus 5 --- model.go | 92 ++++++++++++++++++++++++++++++++++++++++----------- model_test.go | 88 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 160 insertions(+), 20 deletions(-) diff --git a/model.go b/model.go index f782cfe..92de2c2 100644 --- a/model.go +++ b/model.go @@ -80,8 +80,21 @@ type AmbientModel struct { Sampler PacketSizeSampler Chooser ProtocolChooser SamplingRate uint32 + + // emitOrder is a flat view of every host. The cursors remember the next + // host and direction after clipping so neither is persistently starved. + emitOrder []hostRef + emitCursor int + emitDirection int +} + +type hostRef struct { + network int + host int } +var ambientDirections = [...]Direction{DirectionIncoming, DirectionOutgoing} + type PacketSpec struct { Internal netip.Addr Peer netip.Addr @@ -142,9 +155,17 @@ func NewAmbientModel(cfg RuntimeAmbient) (*AmbientModel, error) { } model.Networks = append(model.Networks, state) } + for networkIndex, network := range model.Networks { + for hostIndex := range network.Hosts { + model.emitOrder = append(model.emitOrder, hostRef{network: networkIndex, host: hostIndex}) + } + } return model, nil } +// Tick advances the model by one interval. Every host is accounted for before +// any packet is emitted, so the reported target rates and the credited budgets +// cover the whole configuration even when maxSamples clips the emission. func (m *AmbientModel) Tick(interval time.Duration, maxSamples int) (TickResult, error) { result := TickResult{Packets: make([]PacketSpec, 0, 128)} maxFrameSize := m.Sampler.Config.MaxSize @@ -168,7 +189,6 @@ func (m *AmbientModel) Tick(interval time.Duration, maxSamples int) (TickResult, scores[i] = score scoreTotal += score } - for i := range network.Hosts { host := &network.Hosts[i] hostShare := scores[i] / scoreTotal @@ -182,32 +202,64 @@ func (m *AmbientModel) Tick(interval time.Duration, maxSamples int) (TickResult, host.IncomingBudget.AddRate(host.LastIncomingRate, interval, m.SamplingRate, maxFrameSize) host.OutgoingBudget.AddRate(host.LastOutgoingRate, interval, m.SamplingRate, maxFrameSize) + } + } - for len(result.Packets) < maxSamples { - generated := false - for _, direction := range []Direction{DirectionIncoming, DirectionOutgoing} { - if len(result.Packets) >= maxSamples { - break - } - packet, ok, err := m.takeAmbientPacket(host, network, direction) - if err != nil { - return TickResult{}, err - } - if ok { - result.Packets = append(result.Packets, packet) - generated = true - } - } - if !generated { + if err := m.emit(&result, maxSamples); err != nil { + return TickResult{}, err + } + return result, nil +} + +// emit drains host budgets into samples. When maxSamples clips the tick the +// cursor moves past the host that was cut off, so the next tick starts with the +// hosts that were not reached instead of the same ones every time. +func (m *AmbientModel) emit(result *TickResult, maxSamples int) error { + if len(m.emitOrder) == 0 { + return nil + } + if m.emitCursor >= len(m.emitOrder) { + m.emitCursor = 0 + } + + startDirection := m.emitDirection + lastDirection := startDirection + for offset := 0; offset < len(m.emitOrder); offset++ { + index := (m.emitCursor + offset) % len(m.emitOrder) + ref := m.emitOrder[index] + network := &m.Networks[ref.network] + host := &network.Hosts[ref.host] + + for len(result.Packets) < maxSamples { + generated := false + for directionOffset := range ambientDirections { + if len(result.Packets) >= maxSamples { break } + directionIndex := (startDirection + directionOffset) % len(ambientDirections) + direction := ambientDirections[directionIndex] + packet, ok, err := m.takeAmbientPacket(host, network, direction) + if err != nil { + return err + } + if ok { + result.Packets = append(result.Packets, packet) + generated = true + lastDirection = directionIndex + } } - if len(result.Packets) >= maxSamples { - return result, nil + if !generated { + break } } + if len(result.Packets) >= maxSamples { + m.emitCursor = (index + 1) % len(m.emitOrder) + m.emitDirection = (lastDirection + 1) % len(ambientDirections) + return nil + } } - return result, nil + m.emitCursor = 0 + return nil } func (m *AmbientModel) takeAmbientPacket(host *AmbientHostState, network *AmbientNetworkState, direction Direction) (PacketSpec, bool, error) { diff --git a/model_test.go b/model_test.go index 28a3f3f..9ef1a97 100644 --- a/model_test.go +++ b/model_test.go @@ -161,3 +161,91 @@ func TestSampleBudgetBacklogIsCapped(t *testing.T) { t.Fatalf("low-rate backlog=%f, expected 12.5", slow.Bytes) } } + +func TestAmbientClippingKeepsAllNetworksServed(t *testing.T) { + network := func(cidr string) RuntimeAmbientNetwork { + return RuntimeAmbientNetwork{ + Prefix: netip.MustParsePrefix(cidr), + IncomingShare: 1, + RateBPS: 1e9, + Hosts: 4, + Zipf: 1.1, + PeerPrefixes: []netip.Prefix{netip.MustParsePrefix("198.18.0.0/15")}, + } + } + cfg := RuntimeAmbient{ + Common: RuntimeCommon{ + SamplingRate: 1000, + Seed: 7, + Packet: RuntimePacket{MeanSize: 1000, MinSize: 1000, MaxSize: 1000}, + Protocols: ProtocolMix{UDP: 1}, + }, + Networks: []RuntimeAmbientNetwork{network("10.0.0.0/24"), network("10.1.0.0/24")}, + } + model, err := NewAmbientModel(cfg) + if err != nil { + t.Fatal(err) + } + + served := map[string]int{} + var target float64 + for i := 0; i < 20; i++ { + result, err := model.Tick(100*time.Millisecond, 5) + if err != nil { + t.Fatal(err) + } + target = result.TargetBPS + for _, packet := range result.Packets { + if netip.MustParsePrefix("10.0.0.0/24").Contains(packet.Internal) { + served["first"]++ + } else { + served["second"]++ + } + } + } + if served["first"] == 0 || served["second"] == 0 { + t.Fatalf("clipping starved a network: first=%d second=%d", served["first"], served["second"]) + } + if target != 2e9 { + t.Fatalf("target=%f, expected the full 2e9 even when clipped", target) + } +} + +func TestAmbientClippingKeepsBothDirectionsServed(t *testing.T) { + cfg := RuntimeAmbient{ + Common: RuntimeCommon{ + SamplingRate: 10, + Seed: 1, + Packet: RuntimePacket{MeanSize: 1000, MinSize: 1000, MaxSize: 1000}, + Protocols: ProtocolMix{UDP: 1}, + }, + Networks: []RuntimeAmbientNetwork{{ + Prefix: netip.MustParsePrefix("10.0.0.1/32"), + IncomingShare: 0.5, + OutgoingShare: 0.5, + RateBPS: 160000, + Hosts: 1, + PeerPrefixes: []netip.Prefix{netip.MustParsePrefix("198.18.0.0/15")}, + }}, + } + model, err := NewAmbientModel(cfg) + if err != nil { + t.Fatal(err) + } + + served := map[Direction]int{} + for range 10 { + result, err := model.Tick(time.Second, 1) + if err != nil { + t.Fatal(err) + } + if len(result.Packets) != 1 { + t.Fatalf("generated %d samples, expected 1", len(result.Packets)) + } + served[result.Packets[0].Direction]++ + } + if served[DirectionIncoming] != 5 || served[DirectionOutgoing] != 5 { + t.Fatalf("clipping skewed directions: incoming=%d outgoing=%d, expected 5/5", + served[DirectionIncoming], served[DirectionOutgoing]) + } +}