[meshnet] Manage system-level kernel and link tunables for packet performance

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 (

third_party/meshnet/daemon/grpcwire/grpcwire.go

@@ -1,27 +1,24 @@
+// Package grpcwire provides gRPC overlay wire creation, TAP interface management,
+// packet multiplexing, and CRD reconciliation for meshnet daemon.
 package grpcwire
 
 import (
-	"context"
 	"fmt"
 	"io"
-	"net"
-	"strings"
+	"os"
 	"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"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
+	"github.com/vishvananda/netlink"
 
 	mpb "github.com/openconfig/kne/third_party/meshnet/daemon/proto/meshnet/v1beta1"
-	"github.com/openconfig/kne/third_party/meshnet/utils/wireutil"
 )
 
 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"})
 }
@@ -32,24 +29,21 @@ 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
 
+// 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()
 	indexGen.currId++
 	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 {
@@ -98,6 +92,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{
@@ -189,14 +184,9 @@ func WireDownByUID(namespace string, linkUID int) error {
 	return nil
 }
 
-// -------------------------------------------------------------------------------------------------
-func AddWireInMemNDataStore(wire *GRPCWire, handle *pcap.Handle) int {
+// 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. */
-
-	/* 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,112 +272,79 @@ 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
 }
 
-// -----------------------------------------------------------------------------------------------------------
+// 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
-	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)
+	tapFile, err := GetHostIntfHndl(wire.LocalNodeIfaceID)
 	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)
+		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
 	}
 
-	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()
+	nodeStream := streamMgr.GetOrCreateStream(wire.TopoNamespace, wire.PeerNodeIP)
+	defer streamMgr.ReleaseStream(wire.TopoNamespace, wire.PeerNodeIP)
 
-	err = rdHandl.SetDirection(pcap.Direction(pcap.DirectionIn))
-	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)
-		return err
+	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
+			}
+		}
+	}()
 
-	remote, err := grpc.Dial(url, grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		grpcOvrlyLogger.Infof("RecvFrmLocalPodThread:Failed to connect to remote %s/%d", url, wire.LocalNodeIfaceID)
-		return err
-	}
-	defer remote.Close()
-
-	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
 	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",
-					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.
-				*/
+			if !nodeStream.Send(payload) {
+				grpcOvrlyLogger.Debugf("RecvFrmLocalPodThread: Could not queue packet over stream %s@%s (queue full)", wire.LocalPodName, wire.LocalNodeIfaceName)
 			}
 		}
 	}

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,17 +41,16 @@ 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{
 		namespace: wire.LocalPodNetNS,
 		linkUID:   wire.UID,
 	}] = wire
-
-	wire.K8sStoreGWire()
-
 	w.handles[wire.LocalNodeIfaceID] = handle
+	w.mu.Unlock()
+
+	go wire.K8sStoreGWire()
 	return nil
 }
 
@@ -87,11 +85,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 +108,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,15 +117,15 @@ 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
 		}
 	}
 	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 {