Report VPN fabric session stream usage

agents/rap-node-agent/internal/vpnruntime/fabric_session_transport.go

@@ -3,6 +3,7 @@ package vpnruntime
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -34,9 +35,13 @@ type FabricSessionPacketTransport struct {
 	StreamIDsByTrafficClass map[string][]uint64
 	StreamIDs               []uint64
 
-	sequence         uint64
-	sequenceMu       sync.Mutex
-	sequenceByStream map[uint64]uint64
+	sequence           uint64
+	sequenceMu         sync.Mutex
+	sequenceByStream   map[uint64]uint64
+	statsMu            sync.Mutex
+	sendFramesByClass  map[string]uint64
+	sendPacketsByClass map[string]uint64
+	sendFramesByStream map[uint64]uint64
 }
 
 func (t *FabricSessionPacketTransport) SendGatewayPacketBatch(ctx context.Context, packets [][]byte) error {
@@ -66,7 +71,11 @@ func (t *FabricSessionPacketTransport) SendGatewayPacketBatch(ctx context.Contex
 	if err != nil {
 		return err
 	}
-	return t.Sender.Send(ctx, frame)
+	if err := t.Sender.Send(ctx, frame); err != nil {
+		return err
+	}
+	t.recordSend(streamID, trafficClass, len(packets))
+	return nil
 }
 
 func (t *FabricSessionPacketTransport) ReceiveGatewayPacketBatch(ctx context.Context, timeout time.Duration) ([][]byte, error) {
@@ -239,6 +248,49 @@ func (t *FabricSessionPacketTransport) nextSequence(streamID uint64) uint64 {
 	return t.sequenceByStream[streamID]
 }
 
+func (t *FabricSessionPacketTransport) recordSend(streamID uint64, trafficClass string, packetCount int) {
+	if t == nil {
+		return
+	}
+	trafficClass = normalizeFabricTrafficClass(trafficClass)
+	t.statsMu.Lock()
+	defer t.statsMu.Unlock()
+	if t.sendFramesByClass == nil {
+		t.sendFramesByClass = map[string]uint64{}
+	}
+	if t.sendPacketsByClass == nil {
+		t.sendPacketsByClass = map[string]uint64{}
+	}
+	if t.sendFramesByStream == nil {
+		t.sendFramesByStream = map[uint64]uint64{}
+	}
+	t.sendFramesByClass[trafficClass]++
+	t.sendPacketsByClass[trafficClass] += uint64(packetCount)
+	t.sendFramesByStream[streamID]++
+}
+
+func (t *FabricSessionPacketTransport) Snapshot() map[string]any {
+	if t == nil {
+		return nil
+	}
+	t.statsMu.Lock()
+	sendFramesByClass := copyStringUint64Map(t.sendFramesByClass)
+	sendPacketsByClass := copyStringUint64Map(t.sendPacketsByClass)
+	sendFramesByStream := make(map[string]uint64, len(t.sendFramesByStream))
+	for streamID, count := range t.sendFramesByStream {
+		sendFramesByStream[fmt.Sprintf("%d", streamID)] = count
+	}
+	t.statsMu.Unlock()
+	return map[string]any{
+		"schema_version":           "rap.vpn_fabric_session_packet_transport.v1",
+		"stream_id":                t.StreamID,
+		"stream_ids_by_class":      copyStreamIDsByTrafficClass(t.StreamIDsByTrafficClass),
+		"send_frames_by_class":     sendFramesByClass,
+		"send_packets_by_class":    sendPacketsByClass,
+		"send_frames_by_stream_id": sendFramesByStream,
+	}
+}
+
 func fabricSessionTrafficClassForPackets(fallback string, packets [][]byte) string {
 	if fallback = normalizeFabricTrafficClass(fallback); fallback != "" && fallback != FabricTrafficClassBulk {
 		return fallback
@@ -248,3 +300,25 @@ func fabricSessionTrafficClassForPackets(fallback string, packets [][]byte) stri
 	}
 	return FabricTrafficClassBulk
 }
+
+func copyStringUint64Map(values map[string]uint64) map[string]uint64 {
+	if len(values) == 0 {
+		return map[string]uint64{}
+	}
+	out := make(map[string]uint64, len(values))
+	for key, value := range values {
+		out[key] = value
+	}
+	return out
+}
+
+func copyStreamIDsByTrafficClass(values map[string][]uint64) map[string][]uint64 {
+	if len(values) == 0 {
+		return map[string][]uint64{}
+	}
+	out := make(map[string][]uint64, len(values))
+	for key, ids := range values {
+		out[key] = append([]uint64(nil), ids...)
+	}
+	return out
+}