m/rdp-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Expose VPN fabric smoke fanout

Browse Source
This commit is contained in:
Mikhail
2026-05-16 12:41:53 +03:00
parent db9ea53e6c
commit a6ee9ba26f
2 changed files with 13 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agents/rap-node-agent/cmd/mesh-live-smoke/main.go
+11 -8
View File
@@ -42,6 +42,7 @@ type smokeReport struct {
FabricSessionRoundTrips int `json:"fabric_session_round_trips"`
FabricVPNPacketAccepted bool `json:"fabric_vpn_packet_accepted"`
FabricVPNPacketSharded bool `json:"fabric_vpn_packet_sharded"`
FabricVPNPacketFanout int `json:"fabric_vpn_packet_fanout"`
FabricQUICAccepted bool `json:"fabric_quic_accepted"`
FabricQUICEndpoint string `json:"fabric_quic_endpoint"`
FabricSessionLatencyMS int64 `json:"fabric_session_latency_ms"`
@@ -140,7 +141,7 @@ func run(ctx context.Context) (smokeReport, error) {
string(firstFabricSessionResponse.Payload) == "mesh-live-smoke-fabric-session" &&
secondFabricSessionResponse.Type == fabricproto.FramePong &&
string(secondFabricSessionResponse.Payload) == "mesh-live-smoke-fabric-session-2"
fabricVPNPacketAccepted, fabricVPNPacketSharded, err := smokeFabricVPNPacketOverSession(ctx, fabricSession)
fabricVPNPacketAccepted, fabricVPNPacketSharded, fabricVPNPacketFanout, err := smokeFabricVPNPacketOverSession(ctx, fabricSession)
if err != nil {
return smokeReport{}, fmt.Errorf("fabric vpn packet session smoke: %w", err)
}
@@ -163,6 +164,7 @@ func run(ctx context.Context) (smokeReport, error) {
FabricSessionRoundTrips: 2,
FabricVPNPacketAccepted: fabricVPNPacketAccepted,
FabricVPNPacketSharded: fabricVPNPacketSharded,
FabricVPNPacketFanout: fabricVPNPacketFanout,
FabricQUICAccepted: fabricQUICAccepted,
FabricQUICEndpoint: fabricQUICEndpoint,
FabricSessionLatencyMS: fabricSessionLatency.Milliseconds(),
@@ -243,7 +245,7 @@ func smokeQUICTLSConfig() *tls.Config {
}
}
func smokeFabricVPNPacketOverSession(ctx context.Context, fabricSession *mesh.FabricSessionClient) (bool, bool, error) {
func smokeFabricVPNPacketOverSession(ctx context.Context, fabricSession *mesh.FabricSessionClient) (bool, bool, int, error) {
const interactiveStreamID uint64 = 4400
const bulkStreamID uint64 = 4401
pump := fabricSession.StartPump(ctx, mesh.FabricSessionPumpOptions{
@@ -257,7 +259,7 @@ func smokeFabricVPNPacketOverSession(ctx context.Context, fabricSession *mesh.Fa
{Type: fabricproto.FrameOpenStream, StreamID: bulkStreamID, TrafficClass: fabricproto.TrafficClassBulk},
} {
if err := pump.Send(ctx, frame); err != nil {
return false, false, err
return false, false, 0, err
}
}
transport := &vpnruntime.FabricSessionPacketTransport{
@@ -273,7 +275,7 @@ func smokeFabricVPNPacketOverSession(ctx context.Context, fabricSession *mesh.Fa
bulkPacket := smokeIPv4TCPPacket([4]byte{10, 77, 0, 2}, [4]byte{192, 168, 200, 95}, 51000, 443, 0)
controlPacket := smokeIPv4TCPPacket([4]byte{10, 77, 0, 2}, [4]byte{192, 168, 200, 95}, 51001, 3389, 0x02)
if err := transport.SendGatewayPacketBatch(ctx, [][]byte{bulkPacket, controlPacket}); err != nil {
return false, false, err
return false, false, 0, err
}
timer := time.NewTimer(3 * time.Second)
defer timer.Stop()
@@ -291,15 +293,16 @@ func smokeFabricVPNPacketOverSession(ctx context.Context, fabricSession *mesh.Fa
snapshot["sharding_active"] == true &&
snapshot["send_class_count"] == 2 &&
snapshot["send_stream_count"] == 2
return true, sharded, nil
fanout, _ := snapshot["last_batch_frame_count"].(uint64)
return true, sharded, int(fanout), nil
}
}
case err := <-pump.Errors():
return false, false, err
return false, false, 0, err
case <-timer.C:
return false, false, fmt.Errorf("timed out waiting for fabric vpn packet ack")
return false, false, 0, fmt.Errorf("timed out waiting for fabric vpn packet ack")
case <-ctx.Done():
return false, false, ctx.Err()
return false, false, 0, ctx.Err()
}
}
}