From 62a0838e9b587c995e2ae78cb9892eda4148a565 Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 31 Jul 2026 19:47:08 +0000 Subject: [PATCH 1/4] [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/4] 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/4] 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 6c4cadb3e02b85a880015fa7561ae40d2ad12d53 Mon Sep 17 00:00:00 2001 From: Kris Raney Date: Fri, 31 Jul 2026 20:02:48 +0000 Subject: [PATCH 4/4] [meshnet] Use gRPC streaming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. Bidirectional Streaming (SendToStream) Receiver (handler.go): • Implemented SendToStream(stream mpb.WireProtocol_SendToStreamServer) error. • In a loop, stream.Recv() continuously ingests incoming mpb.Packet frames and writes them directly to the destination TAP interface (wrHandle.Write(pkt.Frame)), bypassing per-packet unary RPC overhead. 2. Streaming Sender & Auto-Reconnect (grpcwire.go): • Updated RecvFrmLocalPodThread to establish a persistent client stream (wireClient.SendToStream(ctx)). • Frames read from the local TAP interface are streamed out via st.Send(payload) without blocking for individual RPC responses. • If the stream encounters a network or peer reset, RecvFrmLocalPodThread automatically clears its stream handle and transparently re-establishes SendToStream on the next packet. 3. HTTP/2 Window Size & Buffer Tuning: • Server Configuration (meshnet.go): • Stream window size: 4 MB (grpc.InitialWindowSize(4 * 1024 * 1024)). • Connection window size: 16 MB (grpc.InitialConnWindowSize(16 * 1024 * 1024)). • Max message payload: 64 MB (grpc.MaxRecvMsgSize / grpc.MaxSendMsgSize). • Client Configuration (grpcwire.go): • Configured matching initial window size and message payload limits on client grpc.Dial. --- .../meshnet/daemon/grpcwire/grpcwire.go | 41 ++++++++++++++++--- third_party/meshnet/daemon/meshnet/handler.go | 34 +++++++++++++++ third_party/meshnet/daemon/meshnet/meshnet.go | 12 +++++- 3 files changed, 80 insertions(+), 7 deletions(-) diff --git a/third_party/meshnet/daemon/grpcwire/grpcwire.go b/third_party/meshnet/daemon/grpcwire/grpcwire.go index 8fed1aeb..91f38bd9 100644 --- a/third_party/meshnet/daemon/grpcwire/grpcwire.go +++ b/third_party/meshnet/daemon/grpcwire/grpcwire.go @@ -294,7 +294,17 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { return err } - remote, err := grpc.Dial(url, grpc.WithTransportCredentials(insecure.NewCredentials())) + dialOpts := []grpc.DialOption{ + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithInitialWindowSize(4 * 1024 * 1024), // 4MB stream window + grpc.WithInitialConnWindowSize(16 * 1024 * 1024), // 16MB connection window + grpc.WithDefaultCallOptions( + grpc.MaxCallRecvMsgSize(64*1024*1024), + grpc.MaxCallSendMsgSize(64*1024*1024), + ), + } + + remote, err := grpc.Dial(url, dialOpts...) if err != nil { grpcOvrlyLogger.Infof("RecvFrmLocalPodThread:Failed to connect to remote %s/%d", url, wire.LocalNodeIfaceID) return err @@ -306,6 +316,19 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { wireClient := mpb.NewWireProtocolClient(remote) + var stream mpb.WireProtocol_SendToStreamClient + getStream := func() (mpb.WireProtocol_SendToStreamClient, error) { + if stream != nil { + return stream, nil + } + st, err := wireClient.SendToStream(ctx) + if err != nil { + return nil, err + } + stream = st + return stream, nil + } + buf := make([]byte, 65535) type readResult struct { n int @@ -327,6 +350,9 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { 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) + if stream != nil { + _, _ = stream.CloseAndRecv() + } return io.EOF case res := <-readChan: if res.err != nil { @@ -361,10 +387,15 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { 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.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) + st, err := getStream() + if err != nil { + grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not get stream for %s@%s: %v", wire.LocalPodName, wire.LocalNodeIfaceName, err) + continue + } + + if err := st.Send(payload); err != nil { + grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not send packet over stream %s@%s: %v", wire.LocalPodName, wire.LocalNodeIfaceName, err) + stream = nil // reset stream for reconnect on next packet } } } diff --git a/third_party/meshnet/daemon/meshnet/handler.go b/third_party/meshnet/daemon/meshnet/handler.go index 6f796461..be4bafd0 100644 --- a/third_party/meshnet/daemon/meshnet/handler.go +++ b/third_party/meshnet/daemon/meshnet/handler.go @@ -3,6 +3,7 @@ package meshnet import ( "context" "fmt" + "io" "os" "github.com/openconfig/kne/third_party/meshnet/api/types/v1beta1" @@ -414,6 +415,39 @@ func (m *Meshnet) SendToOnce(ctx context.Context, pkt *mpb.Packet) (*mpb.BoolRes return &mpb.BoolResponse{Response: true}, nil } +// ------------------------------------------------------------------------------------------------------ +func (m *Meshnet) SendToStream(stream mpb.WireProtocol_SendToStreamServer) error { + for { + pkt, err := stream.Recv() + if err == io.EOF { + return stream.SendAndClose(&mpb.BoolResponse{Response: true}) + } + if err != nil { + return err + } + + if pkt.RemotIntfId <= 0 { + continue + } + + wrHandle, err := grpcwire.GetHostIntfHndl(pkt.RemotIntfId) + if err != nil { + log.WithFields(log.Fields{ + "daemon": "meshnetd", + "overlay": "gRPC", + }).Debugf("SendToStream (wire id - %v): Could not find local handle. err:%v", pkt.RemotIntfId, err) + continue + } + + if _, err := wrHandle.Write(pkt.Frame); err != nil { + log.WithFields(log.Fields{ + "daemon": "meshnetd", + "overlay": "gRPC", + }).Errorf("SendToStream (wire id - %v): Could not write packet(%d bytes) to local interface. err:%v", pkt.RemotIntfId, len(pkt.Frame), err) + } + } +} + // --------------------------------------------------------------------------------------------------------------- func (m *Meshnet) AddGRPCWireRemote(ctx context.Context, wireDef *mpb.WireDef) (*mpb.WireCreateResponse, error) { stopC := make(chan struct{}) diff --git a/third_party/meshnet/daemon/meshnet/meshnet.go b/third_party/meshnet/daemon/meshnet/meshnet.go index a61e711c..065ebf94 100644 --- a/third_party/meshnet/daemon/meshnet/meshnet.go +++ b/third_party/meshnet/daemon/meshnet/meshnet.go @@ -103,12 +103,20 @@ func New(cfg Config) (*Meshnet, error) { // If the link type is GRPC then set the GRPC logging level to LevelNone // Otherwise there will be GRPC log for every packet sent as for link type GRPC, GRPC is also the data-plane. This is too // much of log that does not help in debugging and K8S does log rotation very frequently. + defaultOpts := []grpc.ServerOption{ + grpc.InitialWindowSize(4 * 1024 * 1024), // 4MB stream window + grpc.InitialConnWindowSize(16 * 1024 * 1024), // 16MB connection window + grpc.MaxRecvMsgSize(64 * 1024 * 1024), + grpc.MaxSendMsgSize(64 * 1024 * 1024), + } + allOpts := append(defaultOpts, cfg.GRPCOpts...) + var svr *grpc.Server lnkTyp := os.Getenv("INTER_NODE_LINK_TYPE") if lnkTyp == wireutil.INTER_NODE_LINK_GRPC { - svr = grpc.NewServer(cfg.GRPCOpts...) + svr = grpc.NewServer(allOpts...) } else { - svr = newServerWithLogging(cfg.GRPCOpts...) + svr = newServerWithLogging(allOpts...) } m := &Meshnet{