Skip to content

Commit ebe08f7

Browse files
sagostinclaude
andcommitted
Fix bidirectional TrafficSim reverse path and add offline test coverage
- Resolve the client probe ID from bidirectional server probe metadata (client_probe_id) instead of the virtual probe's own ID (0), which every reverse-path gate treated as disabled — reverse traffic never started. - Parse the legacy bidirectional_receiver marker in extractVoIPOptions so legacy dual-probe receivers can match again. - Infer VoIP mode from interval_ms <= 50 in extractVoIPOptions to mirror NewTrafficSim, keeping reverse traffic pacing equal to the client's. - Restrict the targetless TrafficSim worker key to server probes; client probes now key on targets so a target IP change restarts the worker. - Guard updateServerAllowedAgents against per-client bidirectional server probes wiping the allowed-agents list each reconcile cycle. - Add unit tests for metadata parsing, probe-ID resolution, client matching, worker key semantics, and wire-format server decision paths (initBidirectional / RefreshBidirectional) for both new and legacy formats. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 parent e4e7345 commit ebe08f7

5 files changed

Lines changed: 600 additions & 25 deletions

File tree

probes/trafficsim.go

Lines changed: 27 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -494,7 +494,7 @@ func (ts *TrafficSim) initBidirectional(connection *AgentConnection) {
494494
// 1. NEW: The bidirectional flag is set in metadata
495495
// 2. LEGACY: The target has ":bidir" suffix (dual-probe approach)
496496
if shouldEnableBidirectional(clientProbe) {
497-
connection.ClientProbeID = clientProbe.ID
497+
connection.ClientProbeID = resolveClientProbeID(clientProbe, connection.ReverseTrafficOptions)
498498
connection.ReverseCycle = &CycleTracker{
499499
StartSeq: 1,
500500
StartTime: time.Now(),
@@ -503,7 +503,7 @@ func (ts *TrafficSim) initBidirectional(connection *AgentConnection) {
503503
receivedSeqs: make(map[int]int),
504504
}
505505
log.Infof("[trafficsim] Bidirectional mode ENABLED for agent %d using client probe %d (VoIP: %v, DSCP: %d, Interval: %dms, BiDir flag: %v)",
506-
connection.AgentID, clientProbe.ID, connection.ReverseTrafficOptions.VoIPMode,
506+
connection.AgentID, connection.ClientProbeID, connection.ReverseTrafficOptions.VoIPMode,
507507
connection.ReverseTrafficOptions.DSCPValue, connection.ReverseTrafficOptions.IntervalMs,
508508
connection.ReverseTrafficOptions.Bidirectional)
509509
} else {
@@ -594,6 +594,12 @@ func extractVoIPOptions(metadata json.RawMessage) TrafficSimOptions {
594594
opts.BidirectionalServer = b
595595
}
596596
}
597+
// Legacy bidirectional receiver marker (dual-probe approach)
598+
if bidirRecv, ok := tsConfig["bidirectional_receiver"]; ok {
599+
if b, ok := bidirRecv.(bool); ok {
600+
opts.BidirectionalReceiver = b
601+
}
602+
}
597603
if clientProbeID, ok := tsConfig["client_probe_id"]; ok {
598604
if f, ok := clientProbeID.(float64); ok {
599605
opts.ClientProbeID = uint(f)
@@ -605,8 +611,10 @@ func extractVoIPOptions(metadata json.RawMessage) TrafficSimOptions {
605611
}
606612
}
607613

608-
// Apply VoIP defaults if enabled
609-
if opts.VoIPMode {
614+
// Apply VoIP defaults if enabled or inferred from interval, mirroring NewTrafficSim
615+
// so reverse traffic paces the same as the client's forward traffic.
616+
if opts.VoIPMode || (opts.IntervalMs > 0 && opts.IntervalMs <= 50) {
617+
opts.VoIPMode = true
610618
if opts.PayloadSize == 0 {
611619
opts.PayloadSize = VoIPPayloadSize
612620
}
@@ -638,7 +646,7 @@ func (ts *TrafficSim) RefreshBidirectional() {
638646
if clientProbe != nil && shouldEnableBidirectional(clientProbe) {
639647
// Extract VoIP options including bidirectional flag
640648
opts := extractVoIPOptions(clientProbe.Metadata)
641-
connection.ClientProbeID = clientProbe.ID
649+
connection.ClientProbeID = resolveClientProbeID(clientProbe, opts)
642650
connection.ReverseCycle = &CycleTracker{
643651
StartSeq: 1,
644652
StartTime: time.Now(),
@@ -648,7 +656,7 @@ func (ts *TrafficSim) RefreshBidirectional() {
648656
}
649657
connection.ReverseTrafficOptions = opts
650658
log.Infof("[trafficsim] Bidirectional mode (refresh) enabled for agent %d using client probe %d (VoIP: %v, BiDir: %v)",
651-
connection.AgentID, clientProbe.ID, opts.VoIPMode, opts.Bidirectional)
659+
connection.AgentID, connection.ClientProbeID, opts.VoIPMode, opts.Bidirectional)
652660
refreshed++
653661
}
654662
}
@@ -711,6 +719,19 @@ func (ts *TrafficSim) GetClientProbeForAgent(targetAgentID uint) *Probe {
711719
return nil
712720
}
713721

722+
// resolveClientProbeID returns the probe ID to use for reverse-path attribution.
723+
// For dynamically generated bidirectional SERVER probes the probe itself is virtual
724+
// (ID=0) and the client's real probe ID is carried in metadata as client_probe_id —
725+
// that ID must be used so reverse stats land on the same probe as the client's
726+
// forward stats. For real client probes (mutual-server / legacy) the probe's own ID
727+
// is already the correct one.
728+
func resolveClientProbeID(probe *Probe, opts TrafficSimOptions) uint {
729+
if (opts.BidirectionalServer || opts.BidirectionalReceiver) && opts.ClientProbeID != 0 {
730+
return opts.ClientProbeID
731+
}
732+
return probe.ID
733+
}
734+
714735
// shouldEnableBidirectional checks if bidirectional mode should be enabled for a probe.
715736
// Returns true if:
716737
// 1. The probe has bidirectional=true in metadata (NEW single-probe approach)
Lines changed: 230 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,230 @@
1+
package probes
2+
3+
import (
4+
"encoding/json"
5+
"testing"
6+
)
7+
8+
func tsMetadata(t *testing.T, m map[string]any) json.RawMessage {
9+
t.Helper()
10+
b, err := json.Marshal(m)
11+
if err != nil {
12+
t.Fatalf("marshal metadata: %v", err)
13+
}
14+
return b
15+
}
16+
17+
func TestExtractVoIPOptionsParsesBidirectionalServerFields(t *testing.T) {
18+
md := tsMetadata(t, map[string]any{
19+
"bidirectional": true,
20+
"trafficsim": map[string]any{
21+
"bidirectional_server": true,
22+
"client_probe_id": float64(42),
23+
"client_agent_id": float64(7),
24+
"voip_mode": true,
25+
"interval_ms": float64(20),
26+
"dscp": float64(46),
27+
},
28+
})
29+
30+
opts := extractVoIPOptions(md)
31+
32+
if !opts.Bidirectional {
33+
t.Error("Bidirectional not parsed from top-level flag")
34+
}
35+
if !opts.BidirectionalServer {
36+
t.Error("BidirectionalServer not parsed")
37+
}
38+
if opts.ClientProbeID != 42 {
39+
t.Errorf("ClientProbeID = %d, want 42", opts.ClientProbeID)
40+
}
41+
if opts.ClientAgentID != 7 {
42+
t.Errorf("ClientAgentID = %d, want 7", opts.ClientAgentID)
43+
}
44+
if !opts.VoIPMode || opts.IntervalMs != 20 || opts.DSCPValue != 46 {
45+
t.Errorf("VoIP settings not parsed: %+v", opts)
46+
}
47+
}
48+
49+
// Regression: the legacy bidirectional_receiver marker was parsed by NewTrafficSim
50+
// but NOT by extractVoIPOptions, so GetClientProbeForAgent could never match
51+
// legacy receiver probes.
52+
func TestExtractVoIPOptionsParsesLegacyReceiver(t *testing.T) {
53+
md := tsMetadata(t, map[string]any{
54+
"trafficsim": map[string]any{
55+
"bidirectional_receiver": true,
56+
"client_probe_id": float64(42),
57+
"client_agent_id": float64(7),
58+
},
59+
})
60+
61+
opts := extractVoIPOptions(md)
62+
if !opts.BidirectionalReceiver {
63+
t.Error("legacy bidirectional_receiver not parsed")
64+
}
65+
if opts.ClientProbeID != 42 || opts.ClientAgentID != 7 {
66+
t.Errorf("client ids not parsed: %+v", opts)
67+
}
68+
}
69+
70+
// Regression: NewTrafficSim infers VoIP mode from interval_ms <= 50, but
71+
// extractVoIPOptions didn't — the server's reverse traffic paced differently
72+
// from the client's forward traffic.
73+
func TestExtractVoIPOptionsInfersVoIPFromFastInterval(t *testing.T) {
74+
md := tsMetadata(t, map[string]any{
75+
"trafficsim": map[string]any{
76+
"interval_ms": float64(20), // fast interval, voip_mode NOT set
77+
},
78+
})
79+
80+
opts := extractVoIPOptions(md)
81+
if !opts.VoIPMode {
82+
t.Error("VoIP mode not inferred from interval_ms <= 50")
83+
}
84+
if opts.PayloadSize != VoIPPayloadSize {
85+
t.Errorf("PayloadSize = %d, want VoIP default %d", opts.PayloadSize, VoIPPayloadSize)
86+
}
87+
if opts.PacketsPerSec != 1000/20 {
88+
t.Errorf("PacketsPerSec = %d, want %d", opts.PacketsPerSec, 1000/20)
89+
}
90+
}
91+
92+
func TestExtractVoIPOptionsEmptyMetadataDefaults(t *testing.T) {
93+
opts := extractVoIPOptions(nil)
94+
if opts.VoIPMode || opts.Bidirectional || opts.BidirectionalServer || opts.BidirectionalReceiver {
95+
t.Errorf("unexpected flags from empty metadata: %+v", opts)
96+
}
97+
if opts.IntervalMs != TrafficSimDataInterval {
98+
t.Errorf("IntervalMs = %d, want default %d", opts.IntervalMs, TrafficSimDataInterval)
99+
}
100+
}
101+
102+
// Regression: the dynamically generated bidirectional server probe is virtual
103+
// (ID=0); the client's real probe ID rides in metadata. Using the probe's own
104+
// ID set ClientProbeID=0, which every reverse-path gate treats as "disabled".
105+
func TestResolveClientProbeID(t *testing.T) {
106+
cases := []struct {
107+
name string
108+
probe Probe
109+
opts TrafficSimOptions
110+
want uint
111+
}{
112+
{
113+
name: "virtual bidir server probe uses metadata client_probe_id",
114+
probe: Probe{ID: 0, Server: true},
115+
opts: TrafficSimOptions{BidirectionalServer: true, ClientProbeID: 42},
116+
want: 42,
117+
},
118+
{
119+
name: "legacy receiver probe uses metadata client_probe_id",
120+
probe: Probe{ID: 9, Server: true},
121+
opts: TrafficSimOptions{BidirectionalReceiver: true, ClientProbeID: 42},
122+
want: 42,
123+
},
124+
{
125+
name: "real client probe uses its own ID",
126+
probe: Probe{ID: 7},
127+
opts: TrafficSimOptions{Bidirectional: true},
128+
want: 7,
129+
},
130+
{
131+
name: "server probe without metadata id falls back to own ID",
132+
probe: Probe{ID: 5, Server: true},
133+
opts: TrafficSimOptions{BidirectionalServer: true},
134+
want: 5,
135+
},
136+
}
137+
for _, tc := range cases {
138+
t.Run(tc.name, func(t *testing.T) {
139+
if got := resolveClientProbeID(&tc.probe, tc.opts); got != tc.want {
140+
t.Errorf("resolveClientProbeID() = %d, want %d", got, tc.want)
141+
}
142+
})
143+
}
144+
}
145+
146+
func TestShouldEnableBidirectional(t *testing.T) {
147+
bidirMD := tsMetadata(t, map[string]any{"bidirectional": true})
148+
149+
if !shouldEnableBidirectional(&Probe{Metadata: bidirMD}) {
150+
t.Error("metadata bidirectional flag not honored (new format)")
151+
}
152+
if !shouldEnableBidirectional(&Probe{
153+
Targets: []ProbeTarget{{Target: "1.2.3.4:bidir"}},
154+
}) {
155+
t.Error(":bidir target suffix not honored (legacy format)")
156+
}
157+
if shouldEnableBidirectional(&Probe{
158+
Targets: []ProbeTarget{{Target: "1.2.3.4:5000"}},
159+
}) {
160+
t.Error("plain probe should not enable bidirectional")
161+
}
162+
if shouldEnableBidirectional(nil) {
163+
t.Error("nil probe should not enable bidirectional")
164+
}
165+
}
166+
167+
func TestGetClientProbeForAgent(t *testing.T) {
168+
clientAgent := uint(5)
169+
170+
bidirClientMD := tsMetadata(t, map[string]any{
171+
"bidirectional": true,
172+
"trafficsim": map[string]any{"bidirectional": true},
173+
})
174+
bidirServerMD := tsMetadata(t, map[string]any{
175+
"bidirectional": true,
176+
"trafficsim": map[string]any{
177+
"bidirectional_server": true,
178+
"client_probe_id": float64(42),
179+
"client_agent_id": float64(clientAgent),
180+
},
181+
})
182+
183+
t.Run("matches local client probe targeting the agent (mutual servers)", func(t *testing.T) {
184+
ts := &TrafficSim{}
185+
ts.SetAllProbes([]Probe{{
186+
ID: 7,
187+
Type: ProbeType_TRAFFICSIM,
188+
Metadata: bidirClientMD,
189+
Targets: []ProbeTarget{{Target: "1.2.3.4:5000", AgentID: &clientAgent}},
190+
}})
191+
p := ts.GetClientProbeForAgent(clientAgent)
192+
if p == nil || p.ID != 7 {
193+
t.Fatalf("got %+v, want client probe 7", p)
194+
}
195+
})
196+
197+
t.Run("matches virtual bidirectional server probe by client_agent_id", func(t *testing.T) {
198+
ts := &TrafficSim{}
199+
ts.SetAllProbes([]Probe{{
200+
ID: 0,
201+
Type: ProbeType_TRAFFICSIM,
202+
Server: true,
203+
Metadata: bidirServerMD,
204+
Targets: []ProbeTarget{{Target: "0.0.0.0:5000", AgentID: &clientAgent}},
205+
}})
206+
p := ts.GetClientProbeForAgent(clientAgent)
207+
if p == nil {
208+
t.Fatal("bidirectional server probe not matched")
209+
}
210+
if got := resolveClientProbeID(p, extractVoIPOptions(p.Metadata)); got != 42 {
211+
t.Errorf("resolved client probe ID = %d, want 42 from metadata", got)
212+
}
213+
})
214+
215+
t.Run("no match for unrelated agent or non-bidirectional probe", func(t *testing.T) {
216+
ts := &TrafficSim{}
217+
ts.SetAllProbes([]Probe{{
218+
ID: 7,
219+
Type: ProbeType_TRAFFICSIM,
220+
Targets: []ProbeTarget{{Target: "1.2.3.4:5000", AgentID: &clientAgent}},
221+
// no bidirectional metadata
222+
}})
223+
if p := ts.GetClientProbeForAgent(clientAgent); p != nil {
224+
t.Errorf("non-bidirectional client probe should not match, got %+v", p)
225+
}
226+
if p := ts.GetClientProbeForAgent(999); p != nil {
227+
t.Errorf("unrelated agent should not match, got %+v", p)
228+
}
229+
})
230+
}

0 commit comments

Comments
 (0)