From 62a0838e9b587c995e2ae78cb9892eda4148a565 Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 31 Jul 2026 19:47:08 +0000 Subject: [PATCH 1/8] [meshnet] Replace veth + libpcap with TAP devices for gRPC wires Replace the veth pair and libpcap packet capture implementation in meshnetd with persistent Linux TAP devices created directly inside container network namespaces. Reason for Change: - Eliminates packet loss under high traffic loads caused by libpcap's kernel- to-userspace drop policy. - Provides native Linux socket buffer flow control and backpressure (`txqueuelen`) up to the container application when gRPC processing falls behind. - Guarantees protocol transparent packet handling (including LACP and LLDP frames). - Ensures process crash resilience: setting `TUNSETPERSIST=1` keeps TAP interfaces alive in the container netns across daemon restarts without link flaps. - Synchronizes CNI plugin pod readiness with complete end-to-end gRPC wire setup, preventing test/ping race conditions upon pod startup. - Eliminates CGO and `libpcap-dev` build dependencies, producing a pure Go static binary (`CGO_ENABLED=0`). Key Changes: - Added `CreateOrAttachTAP` in `wireutil` using `TUNSETIFF` & `TUNSETPERSIST`. - Replaced `pcap.Handle` with `*os.File` in `gwire_map.go`, `grpcwire.go`, and `handler.go`. - Updated `ReconcilePodLinks` in `controller.go` to use TAP interfaces directly without host-side veth creation. - Updated `GRPCWireExists` and CNI `cmdAdd` readiness check to block until gRPC wire handshakes are fully established. - Removed `libpcap-dev` and updated Dockerfile to build `meshnetd` with `CGO_ENABLED=0`. --- .../meshnet/daemon/grpcwire/grpcwire.go | 161 ++++++++---------- .../meshnet/daemon/grpcwire/gwire_map.go | 23 +-- .../meshnet/daemon/grpcwire/gwire_recon.go | 21 +-- .../daemon/grpcwire/gwire_rpc_handlers.go | 110 ++---------- .../meshnet/daemon/meshnet/controller.go | 88 ++-------- third_party/meshnet/daemon/meshnet/handler.go | 14 +- third_party/meshnet/docker/Dockerfile | 6 +- third_party/meshnet/plugin/meshnet.go | 56 ++++-- third_party/meshnet/utils/wireutil/tap.go | 82 +++++++++ 9 files changed, 253 insertions(+), 308 deletions(-) create mode 100644 third_party/meshnet/utils/wireutil/tap.go diff --git a/third_party/meshnet/daemon/grpcwire/grpcwire.go b/third_party/meshnet/daemon/grpcwire/grpcwire.go index e9d2181e..689df968 100644 --- a/third_party/meshnet/daemon/grpcwire/grpcwire.go +++ b/third_party/meshnet/daemon/grpcwire/grpcwire.go @@ -4,15 +4,14 @@ import ( "context" "fmt" "io" - "net" + "os" "strings" "sync" - "github.com/google/gopacket" - "github.com/google/gopacket/pcap" + "github.com/containernetworking/plugins/pkg/ns" "github.com/openconfig/gnmi/errlist" - koko "github.com/redhat-nfvpe/koko/api" log "github.com/sirupsen/logrus" + "github.com/vishvananda/netlink" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -32,13 +31,10 @@ type intfIndex struct { } /* - In a given node a veth-pair connects a pod with the meshnet daemon hosted in the node. This meshnet + In a given node a TAP interface connects a pod with the meshnet daemon hosted in the node. This meshnet -daemon provides the grpc-wire service to connect the local pod with the remote pod over grpc. The node -end of the veth-pair must have unique name with in the node. A node can have multiple pods. So there -will be multiple veth-pairs for connecting multiple nodes to meshnet daemon and each of them (the node end) must have unique -names. IntfIndex provides the sequentially increasing number which makes the name unique when added as -suffix to the name. +daemon provides the grpc-wire service to connect the local pod with the remote pod over grpc. IntfIndex provides +the sequentially increasing number which makes the name unique when added as suffix to the name. */ var indexGen intfIndex @@ -49,7 +45,6 @@ func NextIndex() int64 { return indexGen.currId } -/*+++tbf: These constants has no utility other that helping in debugging. These can be removed later. */ type grpcWireOriginator int func (g grpcWireOriginator) String() string { @@ -190,13 +185,8 @@ func WireDownByUID(namespace string, linkUID int) error { } // ------------------------------------------------------------------------------------------------- -func AddWireInMemNDataStore(wire *GRPCWire, handle *pcap.Handle) int { +func AddWireInMemNDataStore(wire *GRPCWire, handle *os.File) int { /* Populate the active wire map and returns the number of currently added active wires. */ - - /* if this wire is already present in the map then it will be overwritten. - It seems to be ok to overwrite. Think more in what situation this may - not be the desired behavior and we need to throw an error. */ - wires.AddInMemNDataStore(wire, handle) return len(wires.wires) } @@ -242,7 +232,7 @@ func DeletePodWires(namespace string, podName string) error { func RemoveWireAcrosAll(wire *GRPCWire, inMem bool) error { if wire == nil { - grpcOvrlyLogger.Infof("[WIRE-DELETE]:Null wire. This ware is already removed") + grpcOvrlyLogger.Infof("[WIRE-DELETE]:Null wire. This wire is already removed") return nil } @@ -252,19 +242,24 @@ func RemoveWireAcrosAll(wire *GRPCWire, inMem bool) error { } wire.IsReady = false - /* Remove the veth from the node */ - intf, err := net.InterfaceByIndex(int(wire.LocalNodeIfaceID)) - if err != nil { - grpcOvrlyLogger.Infof("[WIRE-DELETE]:Interface index %d for wire %d, is already cleaned up.", wire.LocalNodeIfaceID, wire.UID) - } else { - myVeth := koko.VEth{} - myVeth.LinkName = intf.Name - if err = myVeth.RemoveVethLink(); err != nil { - return fmt.Errorf("[WIRE-DELETE]:failed to remove veth link: %w", err) - } + // Close the TAP file handle if open + if handle, ok := wires.GetHandle(wire.LocalNodeIfaceID); ok && handle != nil { + _ = handle.Close() + } + + // Remove the TAP link from the container netns if present + podNs, err := ns.GetNS(wire.LocalPodNetNS) + if err == nil { + _ = podNs.Do(func(_ ns.NetNS) error { + if link, err := netlink.LinkByName(wire.LocalPodIfaceName); err == nil { + return netlink.LinkDel(link) + } + return nil + }) + podNs.Close() } - // clean up im-memory wire-map + // clean up in-memory wire-map if inMem { wires.AtomicDelete(wire) // Deleting the wire from in-memory data } @@ -277,22 +272,8 @@ func RemoveWireAcrosAll(wire *GRPCWire, inMem bool) error { // ----------------------------------------------------------------------------------------------------------- // Generate the name of the interface to be placed on the node func GenNodeIfaceName(podName string, podIfaceName string) (string, error) { - // Linux has issue if interface name is too long. Generate a smaller name. - // In recent kernel versions this is defined by IFNAMSIZ to be 16 bytes, so 15 user-visible bytes - // (assuming it includes a trailing null). IFNAMSIZ is used in defining struct net_device's name. - // The name must not contain / or any whitespace characters - // - //TODO: This method needs to be robust. It monotonically increases the index and never - // decreases it, even if the interfaces are deleted. So far this will work for accumulated - // 1K interfaces per node under the current naming scheme. This is too small. - // Using 14 digit random number and checking if any interface with generated name exists and if - // exists then generate another random number (try 3 times before giving up). This will make it robust. - // This reduces the readability and correlation between the “pod-interface” and corresponding - // “node-interface”, for example eth1host1-<3-digit-index> will become "12345678901234". id := NextIndex() - ifaceName := fmt.Sprintf("%.5s%.5s-%04d", podName, podIfaceName, id) - return ifaceName, nil } @@ -300,37 +281,11 @@ func GenNodeIfaceName(podName string, podIfaceName string) (string, error) { func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { defaultPort := wireutil.GRPCDefaultPort - pktBuffSz := int32(1024 * 64 * 10) //keep buffer for MAX 10 64K frames - url := strings.TrimSpace(fmt.Sprintf("%s:%d", wire.PeerNodeIP, defaultPort)) - /* Utilizing google gopacket for polling for packets from the node. This seems to be the - simplest way to get all packets. - As an alternative to google gopacket(pcap), a socket based implementation is possible. - Not sure if socket based implementation can bring any advantage or not. - - Near term will replace pcap by socket. - */ - - // in some rare cases by the time the thread starts K8S may decide to move the pod somewhere else. - // in that case the local interfaced will be cleaned up asynchronously. Detect the situation and return. - _, err := net.InterfaceByName(locIfNm) - if err != nil { - grpcOvrlyLogger.Errorf("[Packet Receive thread]For pod %s failed to retrieve interface %s/%d. error: %v", wire.LocalPodName, wire.LocalNodeIfaceName, wire.LocalNodeIfaceID, err) - return err - } - - rdHandl, err := pcap.OpenLive(wire.LocalNodeIfaceName, pktBuffSz, true, pcap.BlockForever) - if err != nil { - // let the caller handle the error - grpcOvrlyLogger.Errorf("Receive Thread for local pod failed to open interface: %s/%d, PCAP ERROR: %v", wire.LocalNodeIfaceName, wire.LocalNodeIfaceID, err) - return err - } - defer rdHandl.Close() - err = rdHandl.SetDirection(pcap.Direction(pcap.DirectionIn)) + tapFile, err := GetHostIntfHndl(wire.LocalNodeIfaceID) if err != nil { - // let the caller handle the error - grpcOvrlyLogger.Errorf("Receive Thread for local pod failed to set up capture direction: %s/%d, PCAP ERROR: %v", wire.LocalNodeIfaceName, wire.LocalNodeIfaceID, err) + grpcOvrlyLogger.Errorf("[Packet Receive thread] For pod %s failed to retrieve TAP handle for interface %s/%d. error: %v", wire.LocalPodName, wire.LocalNodeIfaceName, wire.LocalNodeIfaceID, err) return err } @@ -344,45 +299,67 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - source := gopacket.NewPacketSource(rdHandl, rdHandl.LinkType()) wireClient := mpb.NewWireProtocolClient(remote) - in := source.Packets() - var packet gopacket.Packet + buf := make([]byte, 65535) + type readResult struct { + n int + err error + } + readChan := make(chan readResult, 1) + go func() { + for { + n, err := tapFile.Read(buf) + readChan <- readResult{n: n, err: err} + if err != nil { + return + } + } + }() + for { select { case <-wire.StopC: grpcOvrlyLogger.Infof("RecvFrmLocalPodThread: closing connection with remote peer-iface@peer-node-ip: %d@%s/%d from %s@%s", wire.WireIfaceIDOnPeerNode, wire.PeerNodeIP, wire.LocalNodeIfaceID, wire.LocalPodName, wire.LocalPodIfaceName) return io.EOF - case packet = <-in: - data := packet.Data() + case res := <-readChan: + if res.err != nil { + select { + case <-wire.StopC: + return io.EOF + default: + grpcOvrlyLogger.Errorf("RecvFrmLocalPodThread: error reading from TAP interface %s: %v", locIfNm, res.err) + return res.err + } + } + n := res.n + if n <= 0 { + continue + } + + frame := make([]byte, n) + copy(frame, buf[:n]) + + if !wire.IsReady || wire.WireIfaceIDOnPeerNode <= 0 { + // Remote peer handshake is still in progress; skip sending to unassigned wire ID 0 + continue + } + payload := &mpb.Packet{ RemotIntfId: wire.WireIfaceIDOnPeerNode, - Frame: data, + Frame: frame, } - /*+++TODO: Ethernet has a minimum frame size of 64 bytes, comprising an 18-byte header and a payload of 46 bytes. - It also has a maximum frame size of 1518 bytes, in which case the payload is 1500 bytes. - This logic needs to be better, take the interface MTU not hardcoded value of 1518. - This is a very unusual condition to receive an packet from the pod with size > MTU. This can only happens if - things gets really messed up. */ - if len(data) > 1518 { + if n > 1518 { pktType := DecodeFrame(payload.Frame) - grpcOvrlyLogger.Infof("RecvFrmLocalPodThread: unusually large packet received from local pod (may be GRO enabled). size: %d, pkt:%s", len(data), pktType) - /* When Generic Receive Offload (GRO) is enabled then containers can send packets larger than MTU size packet. Do not drop these - packets, deliver it to the receiving container to process. - */ - //continue + grpcOvrlyLogger.Infof("RecvFrmLocalPodThread: unusually large packet received from local pod (may be GRO enabled). size: %d, pkt:%s", n, pktType) } ok, err := wireClient.SendToOnce(ctx, payload) if err != nil || !ok.Response { - grpcOvrlyLogger.Infof("RecvFrmLocalPodThread: Could not deliver pkt %s@%s@%s. Peer not ready, remote iface id %d. err=%v", + grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not deliver pkt %s@%s@%s. Peer not ready, remote iface id %d. err=%v", wire.LocalPodName, wire.LocalPodIfaceName, wire.LocalNodeIfaceName, wire.WireIfaceIDOnPeerNode, err) - /* we generate information and continue. As the above errors will happen when the remote end is not yet ready. - It will eventually get ready and if it can't then someone else will stop this thread. - */ } } } diff --git a/third_party/meshnet/daemon/grpcwire/gwire_map.go b/third_party/meshnet/daemon/grpcwire/gwire_map.go index b58fbc33..baf1461e 100644 --- a/third_party/meshnet/daemon/grpcwire/gwire_map.go +++ b/third_party/meshnet/daemon/grpcwire/gwire_map.go @@ -2,15 +2,14 @@ package grpcwire import ( "fmt" + "os" "sync" - - "github.com/google/gopacket/pcap" ) type wireMap struct { mu sync.Mutex wires map[linkKey]*GRPCWire - handles map[int64]*pcap.Handle + handles map[int64]*os.File } func (w *wireMap) GetWire(namespace string, linkUID int) (*GRPCWire, bool) { @@ -23,14 +22,14 @@ func (w *wireMap) GetWire(namespace string, linkUID int) (*GRPCWire, bool) { return wire, ok } -func (w *wireMap) GetHandle(key int64) (*pcap.Handle, bool) { +func (w *wireMap) GetHandle(key int64) (*os.File, bool) { w.mu.Lock() defer w.mu.Unlock() handle, ok := w.handles[key] return handle, ok } -func (w *wireMap) AddInMem(wire *GRPCWire, handle *pcap.Handle) error { +func (w *wireMap) AddInMem(wire *GRPCWire, handle *os.File) error { w.mu.Lock() defer w.mu.Unlock() w.wires[linkKey{ @@ -42,7 +41,7 @@ func (w *wireMap) AddInMem(wire *GRPCWire, handle *pcap.Handle) error { return nil } -func (w *wireMap) AddInMemNDataStore(wire *GRPCWire, handle *pcap.Handle) error { +func (w *wireMap) AddInMemNDataStore(wire *GRPCWire, handle *os.File) error { w.mu.Lock() defer w.mu.Unlock() w.wires[linkKey{ @@ -87,11 +86,8 @@ func (w *wireMap) DeleteWoLock(wire *GRPCWire) error { * trigger is received. This situation occurs when both the host triggers wire creation almost simultaneously. */ var wires = &wireMap{ - wires: map[linkKey]*GRPCWire{}, - /* Used when a packet is received, then we know the id of the interface to which the packet to be delivered. - This map take interface-id as key and returns the corresponding handle for delivering the packet. - map[interface-id]->handle */ - handles: map[int64]*pcap.Handle{}, + wires: map[linkKey]*GRPCWire{}, + handles: map[int64]*os.File{}, } // FindWiresByPod returns a list of wires matching the namespace and pod. @@ -113,7 +109,6 @@ func GetWiresByPod(namespace string, podName string) ([]*GRPCWire, bool) { func ExtractOneWireByPod(namespace string, podName string) (*GRPCWire, bool) { wires.mu.Lock() defer wires.mu.Unlock() - //var rWires *GRPCWire for _, wire := range wires.wires { if wire.LocalPodName == podName && wire.TopoNamespace == namespace { @@ -123,7 +118,7 @@ func ExtractOneWireByPod(namespace string, podName string) (*GRPCWire, bool) { linkUID: wire.UID, }) - // also clean up the pcap handle for the wire that is extracted from the wire-map + // also clean up the handle for the wire that is extracted from the wire-map delete(wires.handles, wire.LocalNodeIfaceID) return wire, true } @@ -131,7 +126,7 @@ func ExtractOneWireByPod(namespace string, podName string) (*GRPCWire, bool) { return nil, true // no wire found is not a failure, so return true } -func GetHostIntfHndl(intfID int64) (*pcap.Handle, error) { +func GetHostIntfHndl(intfID int64) (*os.File, error) { val, ok := wires.GetHandle(intfID) if ok { diff --git a/third_party/meshnet/daemon/grpcwire/gwire_recon.go b/third_party/meshnet/daemon/grpcwire/gwire_recon.go index a97cb8bd..a5d932b1 100644 --- a/third_party/meshnet/daemon/grpcwire/gwire_recon.go +++ b/third_party/meshnet/daemon/grpcwire/gwire_recon.go @@ -3,13 +3,12 @@ package grpcwire import ( "context" "fmt" - "net" "os" "reflect" - "github.com/google/gopacket/pcap" grpcwirev1 "github.com/openconfig/kne/third_party/meshnet/api/types/v1beta1" mpb "github.com/openconfig/kne/third_party/meshnet/daemon/proto/meshnet/v1beta1" + "github.com/openconfig/kne/third_party/meshnet/utils/wireutil" log "github.com/sirupsen/logrus" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -422,24 +421,18 @@ func reCreateGWire(wStatus grpcwirev1.GWireStatus, _ context.Context) error { // ----------------------------------------------------------------------------------------------------------- // Recreate the wire in-memory wire-map and start the pod to daemon packet receive thread for this wire. func reconLocalGRPCWire(wireDef *mpb.WireDef) error { - locInf, err := net.InterfaceByName(wireDef.WireIfNameOnLocalNode) + tapFile, err := wireutil.CreateOrAttachTAP(wireDef.LocalPodNetNs, wireDef.IntfNameInPod, wireDef.LocalPodIp) if err != nil { - grpcOvrlyLogger.Errorf("[RECONCILE:LOCAL-END]For pod %s failed to retrieve interface ID for interface %v. error:%v", wireDef.LocalPodName, wireDef.WireIfNameOnLocalNode, err) + grpcOvrlyLogger.Errorf("[RECONCILE:LOCAL-END] For pod %s failed to create/attach TAP interface %s in netns %s: %v", + wireDef.LocalPodName, wireDef.IntfNameInPod, wireDef.LocalPodNetNs, err) return err } - - //Using google gopacket for packet receive. An alternative could be using socket. Not sure it it provides any advantage over gopacket. - wrHandle, err := pcap.OpenLive(wireDef.WireIfNameOnLocalNode, 65365, true, pcap.BlockForever) - if err != nil { - grpcOvrlyLogger.Errorf("[RECONCILE:LOCAL-END]Could not open interface for send/recv packets for containers local iface id %d. error:%v", locInf.Index, err) - return err - } - aWire := CreateGWire(locInf.Index, wireDef.WireIfNameOnLocalNode, make(chan struct{}), wireDef) + wireID := NextIndex() + aWire := CreateGWire(int(wireID), wireDef.IntfNameInPod, make(chan struct{}), wireDef) aWire.IsReady = true // reconciling, so add only in memory - wires.AddInMem(aWire, wrHandle) + wires.AddInMem(aWire, tapFile) - // TODO: handle error here go RecvFrmLocalPodThread(aWire, aWire.LocalNodeIfaceName) return nil diff --git a/third_party/meshnet/daemon/grpcwire/gwire_rpc_handlers.go b/third_party/meshnet/daemon/grpcwire/gwire_rpc_handlers.go index a5d91ebd..01863394 100644 --- a/third_party/meshnet/daemon/grpcwire/gwire_rpc_handlers.go +++ b/third_party/meshnet/daemon/grpcwire/gwire_rpc_handlers.go @@ -2,60 +2,38 @@ package grpcwire import ( "context" - "fmt" - "net" log "github.com/sirupsen/logrus" - "github.com/containernetworking/plugins/pkg/ns" - "github.com/google/gopacket/pcap" mpb "github.com/openconfig/kne/third_party/meshnet/daemon/proto/meshnet/v1beta1" "github.com/openconfig/kne/third_party/meshnet/utils/wireutil" - koko "github.com/redhat-nfvpe/koko/api" ) func CreateGRPCWireLocal(ctx context.Context, wireDef *mpb.WireDef) (*mpb.BoolResponse, error) { - locInf, err := net.InterfaceByName(wireDef.WireIfNameOnLocalNode) + tapFile, err := wireutil.CreateOrAttachTAP(wireDef.LocalPodNetNs, wireDef.IntfNameInPod, wireDef.LocalPodIp) if err != nil { log.WithFields(log.Fields{ "daemon": "meshnetd", "overlay": "gRPC", - }).Errorf("[ADD-WIRE:LOCAL-END]For pod %s failed to retrieve interface ID for interface %v. error:%v", wireDef.LocalPodName, wireDef.WireIfNameOnLocalNode, err) + }).Errorf("[ADD-WIRE:LOCAL-END] For pod %s failed to create/attach TAP interface %s in netns %s: %v", + wireDef.LocalPodName, wireDef.IntfNameInPod, wireDef.LocalPodNetNs, err) return &mpb.BoolResponse{Response: false}, err } - // update tx checksumming to off - err = wireutil.SetTxChecksumOff(wireDef.IntfNameInPod, wireDef.LocalPodNetNs) - if err != nil { - log.Errorf("Error in setting tx checksum-off on interface %s, ns %s, pod %s: %v", wireDef.IntfNameInPod, wireDef.LocalPodNetNs, wireDef.LocalPodName, err) - // generate error and continue - } else { - log.Infof("Setting tx checksum-off on interface %s, pod %s is successful", wireDef.IntfNameInPod, wireDef.LocalPodName) - } - - //Using google gopacket for packet receive. An alternative could be using socket. Not sure it it provides any advantage over gopacket. - wrHandle, err := pcap.OpenLive(wireDef.WireIfNameOnLocalNode, 65365, true, pcap.BlockForever) - if err != nil { - log.WithFields(log.Fields{ - "daemon": "meshnetd", - "overlay": "gRPC", - }).Errorf("[ADD-WIRE:LOCAL-END]Could not open interface for send/recv packets for containers local iface id %d. error:%v", locInf.Index, err) - return &mpb.BoolResponse{Response: false}, err - } - - aWire := CreateGWire(locInf.Index, wireDef.WireIfNameOnLocalNode, make(chan struct{}), wireDef) + wireID := NextIndex() + aWire := CreateGWire(int(wireID), wireDef.IntfNameInPod, make(chan struct{}), wireDef) aWire.IsReady = false aWire.Originator = HOST_CREATED_WIRE aWire.OriginatorIP = "unknown" // Add the newly created wire in the in memory wire-map and k8S data store - AddWireInMemNDataStore(aWire, wrHandle) + AddWireInMemNDataStore(aWire, tapFile) log.WithFields(log.Fields{ "daemon": "meshnetd", "overlay": "gRPC", - }).Infof("[ADD-WIRE:LOCAL-END]For pod %s@%s, node iface id %d starting the local packet receive thread", wireDef.LocalPodName, wireDef.IntfNameInPod, locInf.Index) - // TODO: handle error here + }).Infof("[ADD-WIRE:LOCAL-END] For pod %s@%s, wire id %d starting local packet receive thread", wireDef.LocalPodName, wireDef.IntfNameInPod, wireID) + go RecvFrmLocalPodThread(aWire, aWire.LocalNodeIfaceName) return &mpb.BoolResponse{Response: true}, nil @@ -67,8 +45,6 @@ func CreateGRPCWireLocal(ctx context.Context, wireDef *mpb.WireDef) (*mpb.BoolRe // a pod from node A to node B dynamically func CreateUpdateGRPCWireRemoteTriggered(wireDef *mpb.WireDef, stopC chan struct{}) (*GRPCWire, error) { - var err error - // If this wire is already created, then only update the already created wire properties like stopC. // This can happen due to a race between the local and remote peer. // This can also happen when a pod in one end of the wire is deleted and created again. @@ -79,73 +55,21 @@ func CreateUpdateGRPCWireRemoteTriggered(wireDef *mpb.WireDef, stopC chan struct return grpcWire, nil } - outIfNm, err := GenNodeIfaceName(wireDef.LocalPodName, wireDef.IntfNameInPod) - if err != nil { - return nil, fmt.Errorf("[ADD-WIRE:REMOTE-END] could not get current network namespace: %v", err) - } - - currNs, err := ns.GetCurrentNS() + tapFile, err := wireutil.CreateOrAttachTAP(wireDef.LocalPodNetNs, wireDef.IntfNameInPod, wireDef.LocalPodIp) if err != nil { - return nil, fmt.Errorf("[ADD-WIRE:REMOTE-END] could not get current network namespace: %v", err) - } - - /* Create the veth to connect the pod with the meshnet daemon running on the node */ - hostEndVeth := koko.VEth{ - NsName: currNs.Path(), - LinkName: outIfNm, - } - - inIfNm := wireDef.IntfNameInPod - inContainerVeth := koko.VEth{ - NsName: wireDef.LocalPodNetNs, - LinkName: inIfNm, + grpcOvrlyLogger.Errorf("[ADD-WIRE:REMOTE-END] Error creating/attaching TAP interface %s in netns %s: %v", + wireDef.IntfNameInPod, wireDef.LocalPodNetNs, err) + return nil, err } - if wireDef.LocalPodIp != "" { - ipAddr, ipSubnet, err := net.ParseCIDR(wireDef.LocalPodIp) - if err != nil { - return nil, fmt.Errorf("failed to create remote end of GRPC wire(%s@%s), failed to parse CIDR %s: %w", - inIfNm, wireDef.LocalPodName, wireDef.LocalPodIp, err) - } - inContainerVeth.IPAddr = []net.IPNet{{ - IP: ipAddr, - Mask: ipSubnet.Mask, - }} - } + wireID := NextIndex() + grpcOvrlyLogger.Infof("[ADD-WIRE:REMOTE-END] Trigger from %s:%d : Successfully created/attached TAP interface %s@%s (wire id %d).", + wireDef.PeerNodeIp, wireDef.WireIfIdOnPeerNode, wireDef.IntfNameInPod, wireDef.LocalPodName, wireID) - if err = koko.MakeVeth(inContainerVeth, hostEndVeth); err != nil { - grpcOvrlyLogger.Errorf("[ADD-WIRE:REMOTE-END] Error creating vEth pair (in:%s <--> out:%s). Error-> %s", inIfNm, outIfNm, err) - return nil, err - } - if err := wireutil.SetTxChecksumOff(inContainerVeth.LinkName, inContainerVeth.NsName); err != nil { - grpcOvrlyLogger.Errorf("Error in setting tx checksum-off on interface %s, pod %s: %v", inContainerVeth.LinkName, wireDef.LocalPodName, err) - // not returning - } - locIface, err := net.InterfaceByName(hostEndVeth.LinkName) - if err != nil { - // let the caller handle the error - grpcOvrlyLogger.Errorf("[ADD-WIRE:REMOTE-END] Remote end could not get interface index for %s. error:%v", hostEndVeth.LinkName, err) - return nil, err - } - grpcOvrlyLogger.Infof("[ADD-WIRE:REMOTE-END] Trigger from %s:%d : Successfully created remote pod to node vEth pair %s@%s <--> %s(%d).", - wireDef.PeerNodeIp, wireDef.WireIfIdOnPeerNode, inIfNm, wireDef.LocalPodName, outIfNm, locIface.Index) - aWire := CreateGWire(locIface.Index, hostEndVeth.LinkName, stopC, wireDef) - /* Utilizing google gopacket for polling for packets from the node. This seems to be the - simplest way to get all packets. - As an alternative to google gopacket(pcap), a socket based implementation is possible. - Not sure if socket based implementation can bring any advantage or not. - - Near term will replace pcap by socket. - */ - wrHandle, err := pcap.OpenLive(hostEndVeth.LinkName, 65365, true, pcap.BlockForever) - if err != nil { - // let the caller handle the error - grpcOvrlyLogger.Errorf("[ADD-WIRE:REMOTE-END] At remote end could not open interface (%d) for sed/recv packets for containers. error:%v", locIface.Index, err) - return nil, err - } + aWire := CreateGWire(int(wireID), wireDef.IntfNameInPod, stopC, wireDef) // Add the created wire in the in memory wire-map and k8S data store - AddWireInMemNDataStore(aWire, wrHandle) + AddWireInMemNDataStore(aWire, tapFile) return aWire, nil } diff --git a/third_party/meshnet/daemon/meshnet/controller.go b/third_party/meshnet/daemon/meshnet/controller.go index 1c6879b7..cb701757 100644 --- a/third_party/meshnet/daemon/meshnet/controller.go +++ b/third_party/meshnet/daemon/meshnet/controller.go @@ -3,16 +3,13 @@ package meshnet import ( "context" "fmt" - "net" "strings" "time" - "github.com/containernetworking/plugins/pkg/ns" - mpb "github.com/openconfig/kne/third_party/meshnet/daemon/proto/meshnet/v1beta1" "github.com/openconfig/kne/third_party/meshnet/daemon/grpcwire" + mpb "github.com/openconfig/kne/third_party/meshnet/daemon/proto/meshnet/v1beta1" "github.com/openconfig/kne/third_party/meshnet/daemon/vxlan" "github.com/openconfig/kne/third_party/meshnet/utils/wireutil" - koko "github.com/redhat-nfvpe/koko/api" "github.com/vishvananda/netlink" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -171,70 +168,28 @@ func (m *Meshnet) reconcilePodLinksInternal(ctx context.Context, topo *unstructu mnetdLogger.Infof("ReconcilePodLinks: initiating gRPC wire for pod %s <-> peer %s (UID %d)", topo.GetName(), link.PeerPodName, link.LinkUID) - // 1. Generate local host interface name - outIfNm, err := grpcwire.GenNodeIfaceName(topo.GetName(), link.LocalIntf) - if err != nil { - mnetdLogger.Errorf("ReconcilePodLinks: failed to generate node interface name: %v", err) - return err - } - - currNs, err := ns.GetCurrentNS() - if err != nil { - mnetdLogger.Errorf("ReconcilePodLinks: failed to get current host ns: %v", err) - return err - } - - // 2. Create VEth pair (pod namespace <-> host namespace) - inContainerVeth := koko.VEth{ - NsName: netNS, - LinkName: link.LocalIntf, - } - if link.LocalIP != "" { - ipAddr, ipSubnet, err := net.ParseCIDR(link.LocalIP) - if err != nil { - mnetdLogger.Errorf("ReconcilePodLinks: failed to parse local IP CIDR %s: %v", link.LocalIP, err) - return err - } - inContainerVeth.IPAddr = []net.IPNet{{ - IP: ipAddr, - Mask: ipSubnet.Mask, - }} - } - - hostEndVeth := koko.VEth{ - NsName: currNs.Path(), - LinkName: outIfNm, - } - - // Make Veth - if err := koko.MakeVeth(inContainerVeth, hostEndVeth); err != nil { - mnetdLogger.Errorf("ReconcilePodLinks: failed to create local vEth pair (in:%s, out:%s) for gRPC wire: %v", - inContainerVeth.LinkName, hostEndVeth.LinkName, err) - return err - } - - // Disable TX checksum - if err := wireutil.SetTxChecksumOff(inContainerVeth.LinkName, inContainerVeth.NsName); err != nil { - mnetdLogger.Errorf("ReconcilePodLinks: failed to disable Tx checksum on %s: %v", inContainerVeth.LinkName, err) - } - - // 3. Register local end in meshnet daemon (acts as CreateGRPCWireLocal) + // 1. Register local end in meshnet daemon (creates/attaches TAP interface in container netns) wireDefLocal := &mpb.WireDef{ - LocalPodNetNs: netNS, - LinkUid: link.LinkUID, - TopoNs: link.KubeNs, - WireIfNameOnLocalNode: outIfNm, - LocalPodName: topo.GetName(), - IntfNameInPod: link.LocalIntf, - LocalPodIp: link.LocalIP, - PeerNodeIp: peerSrcIP, + LocalPodNetNs: netNS, + LinkUid: link.LinkUID, + TopoNs: link.KubeNs, + LocalPodName: topo.GetName(), + IntfNameInPod: link.LocalIntf, + LocalPodIp: link.LocalIP, + PeerNodeIp: peerSrcIP, } if _, err := grpcwire.CreateGRPCWireLocal(ctx, wireDefLocal); err != nil { mnetdLogger.Errorf("ReconcilePodLinks: failed to register local GRPC wire: %v", err) return err } - // 4. Dial remote daemon and trigger remote end creation + locWire, ok := grpcwire.GetWireByUID(netNS, int(link.LinkUID)) + if !ok || locWire == nil { + mnetdLogger.Errorf("ReconcilePodLinks: failed to get local wire for link UID %d", link.LinkUID) + return fmt.Errorf("local wire not found for link UID %d", link.LinkUID) + } + + // 2. Dial remote daemon and trigger remote end creation url := fmt.Sprintf("%s:%d", peerSrcIP, wireutil.GRPCDefaultPort) url = strings.TrimSpace(url) remoteConn, err := grpc.Dial(url, grpc.WithTransportCredentials(insecure.NewCredentials())) @@ -243,15 +198,8 @@ func (m *Meshnet) reconcilePodLinksInternal(ctx context.Context, topo *unstructu return err } - locInf, err := net.InterfaceByName(outIfNm) - if err != nil { - remoteConn.Close() - mnetdLogger.Errorf("ReconcilePodLinks: failed to get local interface by name %s: %v", outIfNm, err) - return err - } - wireDefRemote := &mpb.WireDef{ - WireIfIdOnPeerNode: int64(locInf.Index), + WireIfIdOnPeerNode: locWire.LocalNodeIfaceID, PeerNodeIp: srcIP, IntfNameInPod: link.PeerIntf, LocalPodNetNs: peerNetNS, @@ -271,7 +219,7 @@ func (m *Meshnet) reconcilePodLinksInternal(ctx context.Context, topo *unstructu } remoteConn.Close() - // 5. Update local end with the peer's host interface ID returned by Node 2 + // 3. Update local end with the peer's host interface ID returned by Node 2 grpcwire.UpdateWireByUID(netNS, int(link.LinkUID), creatResp.PeerIntfId, make(chan struct{})) } else { remotePod := &mpb.RemotePod{ diff --git a/third_party/meshnet/daemon/meshnet/handler.go b/third_party/meshnet/daemon/meshnet/handler.go index ac18e1d1..6f796461 100644 --- a/third_party/meshnet/daemon/meshnet/handler.go +++ b/third_party/meshnet/daemon/meshnet/handler.go @@ -380,6 +380,14 @@ func (m *Meshnet) AddGRPCWireLocal(ctx context.Context, wireDef *mpb.WireDef) (* // ------------------------------------------------------------------------------------------------------ func (m *Meshnet) SendToOnce(ctx context.Context, pkt *mpb.Packet) (*mpb.BoolResponse, error) { + if pkt.RemotIntfId <= 0 { + log.WithFields(log.Fields{ + "daemon": "meshnetd", + "overlay": "gRPC", + }).Debugf("SendToOnce: received packet for uninitialized wire id %d, peer not ready yet", pkt.RemotIntfId) + return &mpb.BoolResponse{Response: false}, nil + } + wrHandle, err := grpcwire.GetHostIntfHndl(pkt.RemotIntfId) if err != nil { log.WithFields(log.Fields{ @@ -394,7 +402,7 @@ func (m *Meshnet) SendToOnce(ctx context.Context, pkt *mpb.Packet) (*mpb.BoolRes // log.Printf("Daemon(SendToOnce): Received [pkt: %s, bytes: %d, for local interface id: %d]. Sending it to local container", pktType, len(pkt.Frame), pkt.RemotIntfId) // log.Printf("Daemon(SendToOnce): Received [bytes: %d, for local interface id: %d]. Sending it to local container", len(pkt.Frame), pkt.RemotIntfId) - err = wrHandle.WritePacketData(pkt.Frame) + _, err = wrHandle.Write(pkt.Frame) if err != nil { log.WithFields(log.Fields{ "daemon": "meshnetd", @@ -447,10 +455,10 @@ func (m *Meshnet) GRPCWireDownRemote(ctx context.Context, wireDef *mpb.WireDef) // GRPCWireExists will return the wire if it exists. func (m *Meshnet) GRPCWireExists(ctx context.Context, wireDef *mpb.WireDef) (*mpb.WireCreateResponse, error) { wire, ok := grpcwire.GetWireByUID(wireDef.LocalPodNetNs, int(wireDef.LinkUid)) - if !ok || wire == nil { + if !ok || wire == nil || !wire.IsReady || wire.WireIfaceIDOnPeerNode <= 0 { return &mpb.WireCreateResponse{Response: false, PeerIntfId: wireDef.WireIfIdOnPeerNode}, nil } - return &mpb.WireCreateResponse{Response: ok, PeerIntfId: wire.WireIfaceIDOnPeerNode}, nil + return &mpb.WireCreateResponse{Response: true, PeerIntfId: wire.WireIfaceIDOnPeerNode}, nil } // --------------------------------------------------------------------------------------------------------------- diff --git a/third_party/meshnet/docker/Dockerfile b/third_party/meshnet/docker/Dockerfile index 16b66d9b..ec2efd25 100644 --- a/third_party/meshnet/docker/Dockerfile +++ b/third_party/meshnet/docker/Dockerfile @@ -26,7 +26,6 @@ COPY go.sum . RUN go mod download && \ apt-get update -y && \ apt-get install -y --no-install-recommends \ - libpcap-dev \ libsystemd-dev && \ rm -rf /var/lib/apt/lists/* @@ -44,7 +43,7 @@ COPY --from=proto_base /src/ . RUN GOOS=${TARGETOS} GOARCH=${TARGETARCH} go build -o meshnet plugin/meshnet.go plugin/grpcwires-plugin.go && \ - GOOS=${TARGETOS} CGO_ENABLED=1 GOARCH=${TARGETARCH} go build -ldflags="-s -w" -o meshnetd daemon/main.go + GOOS=${TARGETOS} CGO_ENABLED=0 GOARCH=${TARGETARCH} go build -ldflags="-s -w" -o meshnetd daemon/main.go #----------------------------------------------------- Final Container --------------------------------- @@ -54,8 +53,7 @@ FROM debian:13-slim # hadolint ignore=DL3008 RUN apt-get update && \ apt-get install -y --no-install-recommends \ - jq \ - libpcap-dev && \ + jq && \ rm -rf /var/lib/apt/lists/* COPY --from=build /go/src/github.com/openconfig/kne/third_party/meshnet/meshnet / diff --git a/third_party/meshnet/plugin/meshnet.go b/third_party/meshnet/plugin/meshnet.go index 2eff874e..6b3ebc0b 100644 --- a/third_party/meshnet/plugin/meshnet.go +++ b/third_party/meshnet/plugin/meshnet.go @@ -208,32 +208,52 @@ func cmdAdd(args *skel.CmdArgs) error { log.Infof("Add[%s]: Successfully registered pod alive status with meshnet daemon", string(cniArgs.K8S_POD_NAME)) if len(localPod.Links) > 0 { - waitCtx, cancel := context.WithDeadline(ctx, startTime.Add(15*time.Second)) + waitCtx, cancel := context.WithDeadline(ctx, startTime.Add(30*time.Second)) defer cancel() - _ = ns.WithNetNSPath(args.Netns, func(_ ns.NetNS) error { - ticker := time.NewTicker(50 * time.Millisecond) - defer ticker.Stop() - for { - ready := true - for _, link := range localPod.Links { + + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + + for { + allReady := true + for _, link := range localPod.Links { + // Check if interface exists in container netns + _ = ns.WithNetNSPath(args.Netns, func(_ ns.NetNS) error { if _, err := netlink.LinkByName(link.LocalIntf); err != nil { - ready = false - break + allReady = false } - } - if ready { - log.Infof("Add[%s]: All %d interfaces ready in container namespace", string(cniArgs.K8S_POD_NAME), len(localPod.Links)) return nil + }) + if !allReady { + break } - select { - case <-waitCtx.Done(): - log.Infof("Add[%s]: Bounded readiness wait expired (%d links); asynchronous completion will continue", string(cniArgs.K8S_POD_NAME), len(localPod.Links)) - return nil - case <-ticker.C: + + // For gRPC links, check if the gRPC wire is fully established on the daemon + wireDef := &mpb.WireDef{ + LocalPodNetNs: args.Netns, + LinkUid: link.Uid, + } + resp, err := meshnetClient.GRPCWireExists(waitCtx, wireDef) + if err != nil || !resp.Response { + allReady = false + break } } - }) + + if allReady { + log.Infof("Add[%s]: All %d interfaces and gRPC wires are ready", string(cniArgs.K8S_POD_NAME), len(localPod.Links)) + break + } + + select { + case <-waitCtx.Done(): + log.Warnf("Add[%s]: Readiness wait timed out (%d links); proceeding asynchronously", string(cniArgs.K8S_POD_NAME), len(localPod.Links)) + goto WaitDone + case <-ticker.C: + } + } } +WaitDone: return types.PrintResult(result, n.CNIVersion) } diff --git a/third_party/meshnet/utils/wireutil/tap.go b/third_party/meshnet/utils/wireutil/tap.go new file mode 100644 index 00000000..37b11650 --- /dev/null +++ b/third_party/meshnet/utils/wireutil/tap.go @@ -0,0 +1,82 @@ +package wireutil + +import ( + "fmt" + "os" + "unsafe" + + "github.com/containernetworking/plugins/pkg/ns" + log "github.com/sirupsen/logrus" + "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" +) + +const tunDevice = "/dev/net/tun" + +type ifreq struct { + Name [unix.IFNAMSIZ]byte + Flags uint16 + _ [22]byte +} + +// CreateOrAttachTAP opens an existing persistent TAP device or creates a new persistent TAP device +// with the given ifName inside the specified network namespace at podNsPath. +// Returns the open *os.File handle to the TAP device. +func CreateOrAttachTAP(podNsPath string, ifName string, ipCIDR string) (*os.File, error) { + podNs, err := ns.GetNS(podNsPath) + if err != nil { + return nil, fmt.Errorf("could not open netns %s: %w", podNsPath, err) + } + defer podNs.Close() + + var tapFile *os.File + + err = podNs.Do(func(_ ns.NetNS) error { + fd, err := unix.Open(tunDevice, unix.O_RDWR, 0) + if err != nil { + return fmt.Errorf("failed to open %s in netns %s: %w", tunDevice, podNsPath, err) + } + + var ifr ifreq + copy(ifr.Name[:], []byte(ifName)) + ifr.Flags = unix.IFF_TAP | unix.IFF_NO_PI + + _, _, errno := unix.Syscall(unix.SYS_IOCTL, uintptr(fd), uintptr(unix.TUNSETIFF), uintptr(unsafe.Pointer(&ifr))) + if errno != 0 { + unix.Close(fd) + return fmt.Errorf("TUNSETIFF failed for %s in netns %s: %v", ifName, podNsPath, errno) + } + + // Make device persistent so it survives process crashes/restarts + _, _, _ = unix.Syscall(unix.SYS_IOCTL, uintptr(fd), uintptr(unix.TUNSETPERSIST), 1) + + link, err := netlink.LinkByName(ifName) + if err != nil { + unix.Close(fd) + return fmt.Errorf("failed to find link %s inside netns %s: %w", ifName, podNsPath, err) + } + + if err := netlink.LinkSetUp(link); err != nil { + log.Warnf("CreateOrAttachTAP: failed to set %s UP in netns %s: %v", ifName, podNsPath, err) + } + + if ipCIDR != "" { + addr, err := netlink.ParseAddr(ipCIDR) + if err == nil { + _ = netlink.AddrAdd(link, addr) + } + } + + tapFile = os.NewFile(uintptr(fd), ifName) + return nil + }) + + if err != nil { + return nil, err + } + + // Disable tx offload inside the netns (ignore error if non-fatal) + _ = SetTxChecksumOff(ifName, podNsPath) + + return tapFile, nil +} From e2dc222f9fe7bba272a036843b6f832b378963c2 Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 31 Jul 2026 22:22:48 +0000 Subject: [PATCH 2/8] Lint / format fixes Add package & public method comments, and format Go --- third_party/meshnet/daemon/cni/cni.go | 1 + .../meshnet/daemon/grpcwire/grpcwire.go | 9 ++++- .../meshnet/daemon/grpcwire/gwire_recon.go | 14 +++---- third_party/meshnet/daemon/meshnet/meshnet.go | 38 +++++++++++-------- third_party/meshnet/daemon/vxlan/vxlan.go | 2 + .../meshnet/plugin/grpcwires-plugin.go | 4 +- third_party/meshnet/plugin/meshnet.go | 1 + .../meshnet/utils/wireutil/wire-util.go | 4 ++ 8 files changed, 46 insertions(+), 27 deletions(-) diff --git a/third_party/meshnet/daemon/cni/cni.go b/third_party/meshnet/daemon/cni/cni.go index b3611855..f0df080b 100644 --- a/third_party/meshnet/daemon/cni/cni.go +++ b/third_party/meshnet/daemon/cni/cni.go @@ -1,3 +1,4 @@ +// Package cni handles CNI configuration file installation and cleanup for meshnet. package cni import ( diff --git a/third_party/meshnet/daemon/grpcwire/grpcwire.go b/third_party/meshnet/daemon/grpcwire/grpcwire.go index 689df968..8fed1aeb 100644 --- a/third_party/meshnet/daemon/grpcwire/grpcwire.go +++ b/third_party/meshnet/daemon/grpcwire/grpcwire.go @@ -1,3 +1,5 @@ +// Package grpcwire provides gRPC overlay wire creation, TAP interface management, +// packet multiplexing, and CRD reconciliation for meshnet daemon. package grpcwire import ( @@ -21,6 +23,7 @@ import ( var grpcOvrlyLogger *log.Entry = nil +// InitLogger initializes logrus logging for the gRPC overlay daemon. func InitLogger() { grpcOvrlyLogger = log.WithFields(log.Fields{"daemon": "meshnetd", "overlay": "gRPC"}) } @@ -38,6 +41,7 @@ the sequentially increasing number which makes the name unique when added as suf */ var indexGen intfIndex +// NextIndex generates a node-wide, monotonically increasing unique wire ID for TAP device handle indexing. func NextIndex() int64 { indexGen.mu.Lock() defer indexGen.mu.Unlock() @@ -93,6 +97,7 @@ type linkKey struct { linkUID int } +// CreateGWire constructs a new GRPCWire struct from the provided wire definition. func CreateGWire(locIfIndex int, locIfNm string, stopC chan struct{}, wireDef *mpb.WireDef) *GRPCWire { return &GRPCWire{ @@ -184,7 +189,7 @@ func WireDownByUID(namespace string, linkUID int) error { return nil } -// ------------------------------------------------------------------------------------------------- +// AddWireInMemNDataStore populates the active wire map and updates K8s status store. func AddWireInMemNDataStore(wire *GRPCWire, handle *os.File) int { /* Populate the active wire map and returns the number of currently added active wires. */ wires.AddInMemNDataStore(wire, handle) @@ -277,7 +282,7 @@ func GenNodeIfaceName(podName string, podIfaceName string) (string, error) { return ifaceName, nil } -// ----------------------------------------------------------------------------------------------------------- +// RecvFrmLocalPodThread reads packets from the local TAP interface and forwards them over the gRPC stream. func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { defaultPort := wireutil.GRPCDefaultPort diff --git a/third_party/meshnet/daemon/grpcwire/gwire_recon.go b/third_party/meshnet/daemon/grpcwire/gwire_recon.go index a5d932b1..bfb4386d 100644 --- a/third_party/meshnet/daemon/grpcwire/gwire_recon.go +++ b/third_party/meshnet/daemon/grpcwire/gwire_recon.go @@ -33,7 +33,7 @@ const ( kGrpcWireItems = "grpcWireItems" // json name of GWireKItems of gwire_type, +++TBD: can we make it dynamic ) -// ----------------------------------------------------------------------------------------------------------- +// SetGWireClient initializes the dynamic K8s client for gRPC wire CRD management. func SetGWireClient(gClient *dynamic.DynamicClient) { // identifier of grpc wire object in k8s apis gWClient.gvr = schema.GroupVersionResource{ @@ -44,12 +44,12 @@ func SetGWireClient(gClient *dynamic.DynamicClient) { gWClient.di = gClient.Resource(gWClient.gvr) } -// ----------------------------------------------------------------------------------------------------------- +// SetGWireClientInterface sets the K8s dynamic resource interface (used for unit testing). func SetGWireClientInterface(gClient dynamic.NamespaceableResourceInterface) { gWClient.di = gClient } -// ------------------------------------------------------------------------------------------------------------ +// GetWireObjListUS lists unstructured GWireKObj resources for a specified node. func (gc GWireClient) GetWireObjListUS(ctx context.Context, ndName string) (*unstructured.UnstructuredList, error) { return gc.di.Namespace("").List(ctx, metav1.ListOptions{ TypeMeta: metav1.TypeMeta{ @@ -61,19 +61,17 @@ func (gc GWireClient) GetWireObjListUS(ctx context.Context, ndName string) (*uns }) } -// ------------------------------------------------------------------------------------------------------------ +// CreatWireObj creates a new unstructured GWireKObj resource in K8s. func (gc GWireClient) CreatWireObj(ctx context.Context, nSpace string, uWbj map[string]interface{}) (*unstructured.Unstructured, error) { return gc.di.Namespace(nSpace).Create(ctx, &unstructured.Unstructured{Object: uWbj}, metav1.CreateOptions{}) } -// ------------------------------------------------------------------------------------------------------------ +// UpdateWireObj updates an existing unstructured GWireKObj resource in K8s. func (gc GWireClient) UpdateWireObj(ctx context.Context, nSpace string, wObjsOnNd *unstructured.Unstructured) (*unstructured.Unstructured, error) { return gc.di.Namespace(nSpace).Update(ctx, wObjsOnNd, metav1.UpdateOptions{}) - } -//------------------------------------------------------------------------------------------------------------ - +// GetWireObjGrpUS retrieves the GWireKObj for a given node and status. func (gc GWireClient) GetWireObjGrpUS(ctx context.Context, wStatus *grpcwirev1.GWireStatus) (*unstructured.Unstructured, error) { return gc.di.Namespace(wStatus.TopoNamespace).Get(ctx, wStatus.LocalNodeName, metav1.GetOptions{}) } diff --git a/third_party/meshnet/daemon/meshnet/meshnet.go b/third_party/meshnet/daemon/meshnet/meshnet.go index 341764a3..a61e711c 100644 --- a/third_party/meshnet/daemon/meshnet/meshnet.go +++ b/third_party/meshnet/daemon/meshnet/meshnet.go @@ -1,3 +1,5 @@ +// Package meshnet implements the meshnet daemon controller loop, K8s topology resource watching, +// and gRPC wire/vxLAN link reconciliation. package meshnet import ( @@ -27,22 +29,24 @@ import ( mpb "github.com/openconfig/kne/third_party/meshnet/daemon/proto/meshnet/v1beta1" ) +// Config defines configuration options for initializing the Meshnet daemon server. type Config struct { Port int GRPCOpts []grpc.ServerOption } +// Meshnet represents the main daemon service instance handling Kubernetes topology reconciliation and gRPC wire protocol RPCs. type Meshnet struct { mpb.UnimplementedLocalServer mpb.UnimplementedRemoteServer mpb.UnimplementedWireProtocolServer - config Config - kClient kubernetes.Interface - tClient topologyclientv1.Interface - GWireDynClient *dynamic.DynamicClient - rCfg *rest.Config - s *grpc.Server - lis net.Listener + config Config + kClient kubernetes.Interface + tClient topologyclientv1.Interface + GWireDynClient *dynamic.DynamicClient + rCfg *rest.Config + s *grpc.Server + lis net.Listener nodeIP string dirtyChan chan struct{} interNodeLinkType string @@ -50,6 +54,7 @@ type Meshnet struct { var mnetdLogger *log.Entry = nil +// InitLogger initializes the logrus logger for the meshnet daemon. func InitLogger() { mnetdLogger = log.WithFields(log.Fields{"daemon": "meshnetd"}) } @@ -72,6 +77,7 @@ func restConfig() (*rest.Config, error) { return rCfg, nil } +// New creates and initializes a new Meshnet daemon instance with gRPC server options and K8s clientsets. func New(cfg Config) (*Meshnet, error) { rCfg, err := restConfig() if err != nil { @@ -106,16 +112,16 @@ func New(cfg Config) (*Meshnet, error) { } m := &Meshnet{ - config: cfg, - rCfg: rCfg, - kClient: kClient, - tClient: tClient, - GWireDynClient: gwireDynClient, - lis: lis, - s: svr, + config: cfg, + rCfg: rCfg, + kClient: kClient, + tClient: tClient, + GWireDynClient: gwireDynClient, + lis: lis, + s: svr, nodeIP: os.Getenv("HOST_IP"), dirtyChan: make(chan struct{}, 1), - interNodeLinkType: lnkTyp, + interNodeLinkType: lnkTyp, } mpb.RegisterLocalServer(m.s, m) mpb.RegisterRemoteServer(m.s, m) @@ -134,11 +140,13 @@ func New(cfg Config) (*Meshnet, error) { return m, nil } +// Serve starts the gRPC server listening on the configured port. func (m *Meshnet) Serve() error { mnetdLogger.Infof("GRPC server has started on port: %d", m.config.Port) return m.s.Serve(m.lis) } +// Stop gracefully stops the gRPC server instance. func (m *Meshnet) Stop() { m.s.Stop() } diff --git a/third_party/meshnet/daemon/vxlan/vxlan.go b/third_party/meshnet/daemon/vxlan/vxlan.go index c1eaa300..496cab76 100644 --- a/third_party/meshnet/daemon/vxlan/vxlan.go +++ b/third_party/meshnet/daemon/vxlan/vxlan.go @@ -1,3 +1,4 @@ +// Package vxlan implements VXLAN overlay link creation and network interface management for meshnet daemon. package vxlan import ( @@ -16,6 +17,7 @@ import ( var vxLanOvrlyLogger *log.Entry = nil +// InitLogger initializes the logrus logger for the VXLAN overlay daemon. func InitLogger() { vxLanOvrlyLogger = log.WithFields(log.Fields{"daemon": "meshnetd", "overlay": "vxLAN"}) } diff --git a/third_party/meshnet/plugin/grpcwires-plugin.go b/third_party/meshnet/plugin/grpcwires-plugin.go index ad8a4545..3de6f281 100644 --- a/third_party/meshnet/plugin/grpcwires-plugin.go +++ b/third_party/meshnet/plugin/grpcwires-plugin.go @@ -22,7 +22,7 @@ const ( skipStatusRetryCount = skipStatusRetryWarnCount * 4 // how many times to retry ) -// -------------------------------------------------------------------------------------------------------- +// CreatGRPCChan sets up the local and remote ends of a gRPC wire channel between two pods on different nodes. func CreatGRPCChan(link *mpb.Link, localPod *mpb.Pod, peerPod *mpb.Pod, localClient mpb.LocalClient, cniArgs *k8sArgs, ctx context.Context) error { // At this point pods attached to both end of this link are both up. They have got the management IP already. @@ -253,7 +253,7 @@ func CreatGRPCChan(link *mpb.Link, localPod *mpb.Pod, peerPod *mpb.Pod, localCli return nil } -// This function is called when a K8S pod is getting deleted. +// MakeGRPCChanDown signals the remote peer node to tear down the remote gRPC wire end when a pod is deleted. func MakeGRPCChanDown(link *mpb.Link, localPod *mpb.Pod, peerPod *mpb.Pod, ctx context.Context) error { if link == nil { return fmt.Errorf("can't remove remote grpc info. link not provided. link:%p", link) diff --git a/third_party/meshnet/plugin/meshnet.go b/third_party/meshnet/plugin/meshnet.go index 6b3ebc0b..163f0c05 100644 --- a/third_party/meshnet/plugin/meshnet.go +++ b/third_party/meshnet/plugin/meshnet.go @@ -382,6 +382,7 @@ func cmdDel(args *skel.CmdArgs) error { return nil } +// SetInterNodeLinkType reads the inter-node link configuration file to set the default overlay mode (GRPC or VXLAN). func SetInterNodeLinkType() { // TODO: Find a more appropriate (if any) way to figure out intended link type // As of today, daemon gets the intended link type from env INTER_NODE_LINK_TYPE diff --git a/third_party/meshnet/utils/wireutil/wire-util.go b/third_party/meshnet/utils/wireutil/wire-util.go index 9fda21a7..9994d51a 100644 --- a/third_party/meshnet/utils/wireutil/wire-util.go +++ b/third_party/meshnet/utils/wireutil/wire-util.go @@ -1,3 +1,5 @@ +// Package wireutil provides low-level network interface creation, TAP/veth management, +// checksum offload tuning, and OS performance utilities for meshnet. package wireutil import ( @@ -34,6 +36,8 @@ const ( INTER_NODE_LINK_GRPC = "GRPC" ) +// SetTxChecksumOff disables TX checksum and segmentation offloading on the specified interface +// inside the target network namespace to prevent checksum corruption during packet forwarding. func SetTxChecksumOff(intfName, nsName string) error { var vethNs ns.NetNS var err error From f297d78b5b99fca718cd25767a6efdb335787912 Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 31 Jul 2026 22:41:43 +0000 Subject: [PATCH 3/8] Resolve codespell complaint --- third_party/meshnet/plugin/meshnet.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/third_party/meshnet/plugin/meshnet.go b/third_party/meshnet/plugin/meshnet.go index 163f0c05..2aa99a07 100644 --- a/third_party/meshnet/plugin/meshnet.go +++ b/third_party/meshnet/plugin/meshnet.go @@ -215,16 +215,16 @@ func cmdAdd(args *skel.CmdArgs) error { defer ticker.Stop() for { - allReady := true + allAreReady := true for _, link := range localPod.Links { // Check if interface exists in container netns _ = ns.WithNetNSPath(args.Netns, func(_ ns.NetNS) error { if _, err := netlink.LinkByName(link.LocalIntf); err != nil { - allReady = false + allAreReady = false } return nil }) - if !allReady { + if !allAreReady { break } @@ -235,12 +235,12 @@ func cmdAdd(args *skel.CmdArgs) error { } resp, err := meshnetClient.GRPCWireExists(waitCtx, wireDef) if err != nil || !resp.Response { - allReady = false + allAreReady = false break } } - if allReady { + if allAreReady { log.Infof("Add[%s]: All %d interfaces and gRPC wires are ready", string(cniArgs.K8S_POD_NAME), len(localPod.Links)) break } From e7eacce44e4de5b409181f51d90fc304466340ff Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 7 Aug 2026 23:03:34 +0000 Subject: [PATCH 4/8] Review comment - be more diligent with error checking in CreateOrAttachTAP --- third_party/meshnet/utils/wireutil/tap.go | 29 ++++++++++++++++++----- 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/third_party/meshnet/utils/wireutil/tap.go b/third_party/meshnet/utils/wireutil/tap.go index 37b11650..b0bdde76 100644 --- a/third_party/meshnet/utils/wireutil/tap.go +++ b/third_party/meshnet/utils/wireutil/tap.go @@ -1,8 +1,10 @@ package wireutil import ( + "errors" "fmt" "os" + "strings" "unsafe" "github.com/containernetworking/plugins/pkg/ns" @@ -48,7 +50,11 @@ func CreateOrAttachTAP(podNsPath string, ifName string, ipCIDR string) (*os.File } // Make device persistent so it survives process crashes/restarts - _, _, _ = unix.Syscall(unix.SYS_IOCTL, uintptr(fd), uintptr(unix.TUNSETPERSIST), 1) + _, _, errno = unix.Syscall(unix.SYS_IOCTL, uintptr(fd), uintptr(unix.TUNSETPERSIST), 1) + if errno != 0 { + unix.Close(fd) + return fmt.Errorf("TUNSETPERSIST failed for %s in netns %s: %v", ifName, podNsPath, errno) + } link, err := netlink.LinkByName(ifName) if err != nil { @@ -57,13 +63,21 @@ func CreateOrAttachTAP(podNsPath string, ifName string, ipCIDR string) (*os.File } if err := netlink.LinkSetUp(link); err != nil { - log.Warnf("CreateOrAttachTAP: failed to set %s UP in netns %s: %v", ifName, podNsPath, err) + unix.Close(fd) + return fmt.Errorf("failed to set %s UP in netns %s: %w", ifName, podNsPath, err) } if ipCIDR != "" { addr, err := netlink.ParseAddr(ipCIDR) - if err == nil { - _ = netlink.AddrAdd(link, addr) + if err != nil { + unix.Close(fd) + return fmt.Errorf("failed to parse CIDR %q for %s: %w", ipCIDR, ifName, err) + } + if err := netlink.AddrAdd(link, addr); err != nil { + if !os.IsExist(err) && !strings.Contains(err.Error(), "file exists") && !errors.Is(err, unix.EEXIST) { + unix.Close(fd) + return fmt.Errorf("failed to add IP %s to %s inside netns %s: %w", ipCIDR, ifName, podNsPath, err) + } } } @@ -75,8 +89,11 @@ func CreateOrAttachTAP(podNsPath string, ifName string, ipCIDR string) (*os.File return nil, err } - // Disable tx offload inside the netns (ignore error if non-fatal) - _ = SetTxChecksumOff(ifName, podNsPath) + // Disable tx offload inside the netns (log warning if non-fatal) + if err := SetTxChecksumOff(ifName, podNsPath); err != nil { + log.Warnf("CreateOrAttachTAP: failed to disable tx checksum on %s inside netns %s: %v", ifName, podNsPath, err) + } return tapFile, nil } + From 9ecc22bd310c482eb94297a0694385afc7a51c4a Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 7 Aug 2026 23:10:17 +0000 Subject: [PATCH 5/8] Hold wire.mu and preserve wire.StopC in UpdateWireByUID --- .../meshnet/daemon/grpcwire/grpcwire.go | 57 ++++++++++++++----- .../meshnet/daemon/meshnet/controller.go | 2 +- 2 files changed, 44 insertions(+), 15 deletions(-) diff --git a/third_party/meshnet/daemon/grpcwire/grpcwire.go b/third_party/meshnet/daemon/grpcwire/grpcwire.go index 8fed1aeb..83cd6fc8 100644 --- a/third_party/meshnet/daemon/grpcwire/grpcwire.go +++ b/third_party/meshnet/daemon/grpcwire/grpcwire.go @@ -99,7 +99,9 @@ type linkKey struct { // CreateGWire constructs a new GRPCWire struct from the provided wire definition. func CreateGWire(locIfIndex int, locIfNm string, stopC chan struct{}, wireDef *mpb.WireDef) *GRPCWire { - + if stopC == nil { + stopC = make(chan struct{}) + } return &GRPCWire{ UID: int(wireDef.LinkUid), @@ -123,11 +125,17 @@ func CreateGWire(locIfIndex int, locIfNm string, stopC chan struct{}, wireDef *m } -// update the ware with the given input and mark the wire ready +// update the wire with the given input and mark the wire ready func (wire *GRPCWire) UpdateWire(peerIntfId int64, stopC chan struct{}) { wire.mu.Lock() defer wire.mu.Unlock() - wire.StopC = stopC + if wire.StopC == nil { + if stopC != nil { + wire.StopC = stopC + } else { + wire.StopC = make(chan struct{}) + } + } if !wire.IsReady { wire.WireIfaceIDOnPeerNode = peerIntfId } @@ -151,13 +159,21 @@ func GetWireByUID(namespace string, linkUID int) (*GRPCWire, bool) { // Returns true if a wire exists, also the wire structure that got modified func UpdateWireByUID(namespace string, linkUID int, peerIntfId int64, stopC chan struct{}) (*GRPCWire, bool) { wires.mu.Lock() - defer wires.mu.Unlock() wire, ok := wires.wires[linkKey{ namespace: namespace, linkUID: linkUID, }] + wires.mu.Unlock() if ok { - wire.StopC = stopC + wire.mu.Lock() + defer wire.mu.Unlock() + if wire.StopC == nil { + if stopC != nil { + wire.StopC = stopC + } else { + wire.StopC = make(chan struct{}) + } + } if !wire.IsReady { wire.WireIfaceIDOnPeerNode = peerIntfId } @@ -169,19 +185,23 @@ func UpdateWireByUID(namespace string, linkUID int, peerIntfId int64, stopC chan // WireDownByUID - stops packet collection from the connected pod func WireDownByUID(namespace string, linkUID int) error { wires.mu.Lock() - defer wires.mu.Unlock() - wire, ok := wires.wires[linkKey{ namespace: namespace, linkUID: linkUID, }] + wires.mu.Unlock() + if ok { + wire.mu.Lock() + defer wire.mu.Unlock() grpcOvrlyLogger.Infof("WireDownByUID: Making wire down from db, %s@%s-%s@%d, peer fid %d, link uid %d", wire.LocalPodName, wire.LocalPodIfaceName, wire.LocalNodeIfaceName, wire.LocalNodeIfaceID, wire.WireIfaceIDOnPeerNode, linkUID) if wire.IsReady { - close(wire.StopC) + if wire.StopC != nil { + close(wire.StopC) + } + wire.IsReady = false } - wire.IsReady = false } else { grpcOvrlyLogger.Infof("WireDownByUID: Did not find entry to make down from db, uid %d, ns %s", linkUID, namespace) @@ -242,10 +262,14 @@ func RemoveWireAcrosAll(wire *GRPCWire, inMem bool) error { } // stop the packet receive thread for this pod + wire.mu.Lock() if wire.IsReady { - close(wire.StopC) + if wire.StopC != nil { + close(wire.StopC) + } + wire.IsReady = false } - wire.IsReady = false + wire.mu.Unlock() // Close the TAP file handle if open if handle, ok := wires.GetHandle(wire.LocalNodeIfaceID); ok && handle != nil { @@ -346,13 +370,18 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { frame := make([]byte, n) copy(frame, buf[:n]) - if !wire.IsReady || wire.WireIfaceIDOnPeerNode <= 0 { + wire.mu.Lock() + isReady := wire.IsReady + peerIntfID := wire.WireIfaceIDOnPeerNode + wire.mu.Unlock() + + if !isReady || peerIntfID <= 0 { // Remote peer handshake is still in progress; skip sending to unassigned wire ID 0 continue } payload := &mpb.Packet{ - RemotIntfId: wire.WireIfaceIDOnPeerNode, + RemotIntfId: peerIntfID, Frame: frame, } @@ -364,7 +393,7 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { ok, err := wireClient.SendToOnce(ctx, payload) if err != nil || !ok.Response { grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not deliver pkt %s@%s@%s. Peer not ready, remote iface id %d. err=%v", - wire.LocalPodName, wire.LocalPodIfaceName, wire.LocalNodeIfaceName, wire.WireIfaceIDOnPeerNode, err) + wire.LocalPodName, wire.LocalPodIfaceName, wire.LocalNodeIfaceName, peerIntfID, err) } } } diff --git a/third_party/meshnet/daemon/meshnet/controller.go b/third_party/meshnet/daemon/meshnet/controller.go index cb701757..9b609df6 100644 --- a/third_party/meshnet/daemon/meshnet/controller.go +++ b/third_party/meshnet/daemon/meshnet/controller.go @@ -220,7 +220,7 @@ func (m *Meshnet) reconcilePodLinksInternal(ctx context.Context, topo *unstructu remoteConn.Close() // 3. Update local end with the peer's host interface ID returned by Node 2 - grpcwire.UpdateWireByUID(netNS, int(link.LinkUID), creatResp.PeerIntfId, make(chan struct{})) + grpcwire.UpdateWireByUID(netNS, int(link.LinkUID), creatResp.PeerIntfId, nil) } else { remotePod := &mpb.RemotePod{ NetNs: netNS, From eb1cbea6106fb46c20bf7001b5cbc8506fa694fd Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 7 Aug 2026 23:13:44 +0000 Subject: [PATCH 6/8] Only check GRPCWireExists for inter-node gRPC links in CNI cmdAdd --- third_party/meshnet/plugin/meshnet.go | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/third_party/meshnet/plugin/meshnet.go b/third_party/meshnet/plugin/meshnet.go index 2aa99a07..17337852 100644 --- a/third_party/meshnet/plugin/meshnet.go +++ b/third_party/meshnet/plugin/meshnet.go @@ -228,15 +228,24 @@ func cmdAdd(args *skel.CmdArgs) error { break } - // For gRPC links, check if the gRPC wire is fully established on the daemon - wireDef := &mpb.WireDef{ - LocalPodNetNs: args.Netns, - LinkUid: link.Uid, - } - resp, err := meshnetClient.GRPCWireExists(waitCtx, wireDef) - if err != nil || !resp.Response { - allAreReady = false - break + // For inter-node gRPC links, check if the gRPC wire is fully established on the daemon + if interNodeLinkType == wireutil.INTER_NODE_LINK_GRPC { + peerPod, err := meshnetClient.Get(ctx, &mpb.PodQuery{ + Name: link.PeerPod, + KubeNs: string(cniArgs.K8S_POD_NAMESPACE), + }) + // Only check gRPC wire readiness if peer is on a different node + if err == nil && peerPod != nil && peerPod.SrcIp != "" && peerPod.SrcIp != localPod.SrcIp { + wireDef := &mpb.WireDef{ + LocalPodNetNs: args.Netns, + LinkUid: link.Uid, + } + resp, err := meshnetClient.GRPCWireExists(waitCtx, wireDef) + if err != nil || !resp.Response { + allAreReady = false + break + } + } } } From e26791ea2114949239bc5b239720cc229df952ee Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 7 Aug 2026 23:29:41 +0000 Subject: [PATCH 7/8] =?UTF-8?q?=20=20=20=20=20=20=E2=80=A2=20grpcwire.go?= =?UTF-8?q?=20uses=20sync.Pool=20to=20eliminate=20per-packet=20heap=20allo?= =?UTF-8?q?cations=20and=20prevent=20concurrent=20reads=20from=20overwriti?= =?UTF-8?q?ng=20=20=20=20=20=20=20in-flight=20packet=20buffers.=20=20=20?= =?UTF-8?q?=20=20=20=20=E2=80=A2=20grpcwire=5Ftest.go=20tests=20that=2050?= =?UTF-8?q?=20concurrent=20in-flight=20packets=20arrive=20without=20data?= =?UTF-8?q?=20corruption=20or=20race=20detector=20warnings.=20=20=20=20=20?= =?UTF-8?q?=20=20*=20Include=20-race=20instrumentation=20in=20make=20test?= =?UTF-8?q?=20for=20meshnet?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- third_party/meshnet/Makefile | 2 +- .../meshnet/daemon/grpcwire/grpcwire.go | 34 +++++- .../meshnet/daemon/grpcwire/grpcwire_test.go | 103 ++++++++++++++++++ 3 files changed, 132 insertions(+), 7 deletions(-) create mode 100644 third_party/meshnet/daemon/grpcwire/grpcwire_test.go diff --git a/third_party/meshnet/Makefile b/third_party/meshnet/Makefile index 59c09243..a99dbdd5 100644 --- a/third_party/meshnet/Makefile +++ b/third_party/meshnet/Makefile @@ -16,7 +16,7 @@ all: docker ## Run unit tests test: - go test ./... + go test -race ./... ## Run unit tests for Reconciliation recon-test: diff --git a/third_party/meshnet/daemon/grpcwire/grpcwire.go b/third_party/meshnet/daemon/grpcwire/grpcwire.go index 83cd6fc8..ca00d871 100644 --- a/third_party/meshnet/daemon/grpcwire/grpcwire.go +++ b/third_party/meshnet/daemon/grpcwire/grpcwire.go @@ -28,6 +28,17 @@ func InitLogger() { grpcOvrlyLogger = log.WithFields(log.Fields{"daemon": "meshnetd", "overlay": "gRPC"}) } +var packetPool = sync.Pool{ + New: func() any { + b := make([]byte, 65535) + return &b + }, +} + +func init() { + InitLogger() +} + type intfIndex struct { mu sync.Mutex currId int64 @@ -308,7 +319,6 @@ func GenNodeIfaceName(podName string, podIfaceName string) (string, error) { // RecvFrmLocalPodThread reads packets from the local TAP interface and forwards them over the gRPC stream. func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { - defaultPort := wireutil.GRPCDefaultPort url := strings.TrimSpace(fmt.Sprintf("%s:%d", wire.PeerNodeIP, defaultPort)) @@ -329,17 +339,21 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { defer cancel() wireClient := mpb.NewWireProtocolClient(remote) + return forwardPackets(ctx, tapFile, wireClient, wire, locIfNm) +} - buf := make([]byte, 65535) +func forwardPackets(ctx context.Context, reader io.Reader, wireClient mpb.WireProtocolClient, wire *GRPCWire, locIfNm string) error { type readResult struct { + buf *[]byte n int err error } readChan := make(chan readResult, 1) go func() { for { - n, err := tapFile.Read(buf) - readChan <- readResult{n: n, err: err} + bufPtr := packetPool.Get().(*[]byte) + n, err := reader.Read(*bufPtr) + readChan <- readResult{buf: bufPtr, n: n, err: err} if err != nil { return } @@ -353,7 +367,11 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { wire.WireIfaceIDOnPeerNode, wire.PeerNodeIP, wire.LocalNodeIfaceID, wire.LocalPodName, wire.LocalPodIfaceName) return io.EOF case res := <-readChan: + bufPtr := res.buf if res.err != nil { + if bufPtr != nil { + packetPool.Put(bufPtr) + } select { case <-wire.StopC: return io.EOF @@ -364,11 +382,13 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { } n := res.n if n <= 0 { + if bufPtr != nil { + packetPool.Put(bufPtr) + } continue } - frame := make([]byte, n) - copy(frame, buf[:n]) + frame := (*bufPtr)[:n] wire.mu.Lock() isReady := wire.IsReady @@ -377,6 +397,7 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { if !isReady || peerIntfID <= 0 { // Remote peer handshake is still in progress; skip sending to unassigned wire ID 0 + packetPool.Put(bufPtr) continue } @@ -391,6 +412,7 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { } ok, err := wireClient.SendToOnce(ctx, payload) + packetPool.Put(bufPtr) if err != nil || !ok.Response { grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not deliver pkt %s@%s@%s. Peer not ready, remote iface id %d. err=%v", wire.LocalPodName, wire.LocalPodIfaceName, wire.LocalNodeIfaceName, peerIntfID, err) diff --git a/third_party/meshnet/daemon/grpcwire/grpcwire_test.go b/third_party/meshnet/daemon/grpcwire/grpcwire_test.go new file mode 100644 index 00000000..6694bfe9 --- /dev/null +++ b/third_party/meshnet/daemon/grpcwire/grpcwire_test.go @@ -0,0 +1,103 @@ +package grpcwire + +import ( + "bytes" + "context" + "errors" + "io" + "sync" + "testing" + "time" + + mpb "github.com/openconfig/kne/third_party/meshnet/daemon/proto/meshnet/v1beta1" + "google.golang.org/grpc" +) + +type mockWireProtocolClient struct { + mpb.UnimplementedWireProtocolServer + mu sync.Mutex + receivedFrames [][]byte + sendDelay time.Duration +} + +func (m *mockWireProtocolClient) SendToOnce(ctx context.Context, in *mpb.Packet, opts ...grpc.CallOption) (*mpb.BoolResponse, error) { + if m.sendDelay > 0 { + time.Sleep(m.sendDelay) + } + m.mu.Lock() + defer m.mu.Unlock() + // Copy frame to verify exact payload received + frameCopy := make([]byte, len(in.Frame)) + copy(frameCopy, in.Frame) + m.receivedFrames = append(m.receivedFrames, frameCopy) + return &mpb.BoolResponse{Response: true}, nil +} + +func (m *mockWireProtocolClient) SendToStream(ctx context.Context, opts ...grpc.CallOption) (grpc.ClientStreamingClient[mpb.Packet, mpb.BoolResponse], error) { + return nil, errors.New("unimplemented") +} + +func TestForwardPackets_NoCorruption(t *testing.T) { + InitLogger() + + pr, pw := io.Pipe() + defer pr.Close() + + mockClient := &mockWireProtocolClient{ + // Simulate network latency so reader goroutine reads next packet while SendToOnce is busy + sendDelay: 5 * time.Millisecond, + } + + stopC := make(chan struct{}) + wireDef := &mpb.WireDef{ + LinkUid: 1, + IntfNameInPod: "eth1", + LocalPodName: "podA", + LocalPodNetNs: "nsA", + PeerNodeIp: "1.2.3.4", + } + wire := CreateGWire(1, "eth1-0001", stopC, wireDef) + wire.WireIfaceIDOnPeerNode = 42 + wire.IsReady = true + + errCh := make(chan error, 1) + go func() { + errCh <- forwardPackets(context.Background(), pr, mockClient, wire, "eth1-0001") + }() + + numPackets := 50 + expectedPackets := make([][]byte, numPackets) + for i := 0; i < numPackets; i++ { + payload := bytes.Repeat([]byte{byte(i + 1)}, 1024) + expectedPackets[i] = payload + if _, err := pw.Write(payload); err != nil { + t.Fatalf("failed to write packet %d: %v", i, err) + } + } + + // Allow all packets to be transmitted + time.Sleep(350 * time.Millisecond) + + // Stop wire + close(stopC) + _ = pw.Close() + + err := <-errCh + if err != nil && !errors.Is(err, io.EOF) { + t.Fatalf("forwardPackets failed with unexpected error: %v", err) + } + + mockClient.mu.Lock() + received := mockClient.receivedFrames + mockClient.mu.Unlock() + + if len(received) != numPackets { + t.Fatalf("expected %d packets, got %d", numPackets, len(received)) + } + + for i, exp := range expectedPackets { + if !bytes.Equal(received[i], exp) { + t.Errorf("packet %d corrupted: expected all 0x%02x, got mismatch", i, exp[0]) + } + } +} From cc4e4657e1c5d44b0d0dd4e5877ca368fea23b3f Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 7 Aug 2026 23:30:13 +0000 Subject: [PATCH 8/8] * Enable -race for all of KNE for make test. * Extend the timeouts for tests to allow for slower test -race --- Makefile | 2 +- topo/node/arista/arista_test.go | 2 +- topo/node/cisco/cisco_test.go | 4 ++-- topo/node/juniper/juniper_test.go | 6 +++--- topo/node/nokia/nokia_test.go | 6 +++--- 5 files changed, 10 insertions(+), 10 deletions(-) diff --git a/Makefile b/Makefile index ba006d6a..96f4c93b 100644 --- a/Makefile +++ b/Makefile @@ -17,7 +17,7 @@ all: docker ## Run unit tests ## Ignore all tests under the cloudbuild/ tree as these targets are end-to-end test: - go test `go list ./... | grep -v /cloudbuild` + go test -race `go list ./... | grep -v /cloudbuild` ## Targets below are for integration testing only diff --git a/topo/node/arista/arista_test.go b/topo/node/arista/arista_test.go index 9a2ede06..d717d6ec 100644 --- a/topo/node/arista/arista_test.go +++ b/topo/node/arista/arista_test.go @@ -539,7 +539,7 @@ func TestResetCfg(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(), diff --git a/topo/node/cisco/cisco_test.go b/topo/node/cisco/cisco_test.go index 7950127b..1ca241c1 100644 --- a/topo/node/cisco/cisco_test.go +++ b/topo/node/cisco/cisco_test.go @@ -1019,7 +1019,7 @@ func TestResetCfg(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(), @@ -1085,7 +1085,7 @@ func TestPushCfg(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(), diff --git a/topo/node/juniper/juniper_test.go b/topo/node/juniper/juniper_test.go index 8288d46c..97db615e 100644 --- a/topo/node/juniper/juniper_test.go +++ b/topo/node/juniper/juniper_test.go @@ -203,7 +203,7 @@ func TestGenerateSelfSigned(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(), @@ -367,7 +367,7 @@ func TestConfigPush(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(), @@ -466,7 +466,7 @@ func TestResetCfg(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(), diff --git a/topo/node/nokia/nokia_test.go b/topo/node/nokia/nokia_test.go index afafded7..90efc309 100644 --- a/topo/node/nokia/nokia_test.go +++ b/topo/node/nokia/nokia_test.go @@ -273,7 +273,7 @@ func TestGenerateSelfSigned(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(), @@ -352,7 +352,7 @@ func TestResetCfg(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(), @@ -434,7 +434,7 @@ func TestConfigPush(t *testing.T) { n.testOpts = []scrapliutil.Option{ scrapliopts.WithTransportType(scraplitransport.FileTransport), scrapliopts.WithFileTransportFile(tt.testFile), - scrapliopts.WithTimeoutOps(2 * time.Second), + scrapliopts.WithTimeoutOps(10 * time.Second), scrapliopts.WithTransportReadSize(1), scrapliopts.WithReadDelay(0), scrapliopts.WithDefaultLogger(),