diff --git a/third_party/meshnet/daemon/grpcwire/grpcwire.go b/third_party/meshnet/daemon/grpcwire/grpcwire.go index ca00d871..11a9e490 100644 --- a/third_party/meshnet/daemon/grpcwire/grpcwire.go +++ b/third_party/meshnet/daemon/grpcwire/grpcwire.go @@ -328,7 +328,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 @@ -343,6 +353,19 @@ func RecvFrmLocalPodThread(wire *GRPCWire, locIfNm string) error { } func forwardPackets(ctx context.Context, reader io.Reader, wireClient mpb.WireProtocolClient, wire *GRPCWire, locIfNm string) error { + 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 + } + type readResult struct { buf *[]byte n int @@ -365,6 +388,9 @@ func forwardPackets(ctx context.Context, reader io.Reader, wireClient mpb.WirePr 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: bufPtr := res.buf @@ -411,12 +437,18 @@ func forwardPackets(ctx context.Context, reader io.Reader, wireClient mpb.WirePr 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) - 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) + st, err := getStream() + if err != nil { + packetPool.Put(bufPtr) + 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 + } + packetPool.Put(bufPtr) } } } diff --git a/third_party/meshnet/daemon/grpcwire/grpcwire_test.go b/third_party/meshnet/daemon/grpcwire/grpcwire_test.go index 6694bfe9..dd628e1a 100644 --- a/third_party/meshnet/daemon/grpcwire/grpcwire_test.go +++ b/third_party/meshnet/daemon/grpcwire/grpcwire_test.go @@ -33,8 +33,29 @@ func (m *mockWireProtocolClient) SendToOnce(ctx context.Context, in *mpb.Packet, return &mpb.BoolResponse{Response: true}, nil } +type mockStreamClient struct { + grpc.ClientStream + mockClient *mockWireProtocolClient +} + +func (s *mockStreamClient) Send(in *mpb.Packet) error { + if s.mockClient.sendDelay > 0 { + time.Sleep(s.mockClient.sendDelay) + } + s.mockClient.mu.Lock() + defer s.mockClient.mu.Unlock() + frameCopy := make([]byte, len(in.Frame)) + copy(frameCopy, in.Frame) + s.mockClient.receivedFrames = append(s.mockClient.receivedFrames, frameCopy) + return nil +} + +func (s *mockStreamClient) CloseAndRecv() (*mpb.BoolResponse, error) { + 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") + return &mockStreamClient{mockClient: m}, nil } func TestForwardPackets_NoCorruption(t *testing.T) { 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{