Add vl3mtu chain element (networkservicemesh#1325)

Signed-off-by: Artem Glazychev <artem.glazychev@xored.com>
Signed-off-by: anastasia.malysheva <anastasia.malysheva@xored.com>

pkg/networkservice/common/monitor/eventloop.go

@@ -29,11 +29,11 @@ import (
 type eventLoop struct {
 	eventLoopCtx  context.Context
 	conn          *networkservice.Connection
-	eventConsumer eventConsumer
+	eventConsumer EventConsumer
 	client        networkservice.MonitorConnection_MonitorConnectionsClient
 }
 
-func newEventLoop(ctx context.Context, ec eventConsumer, cc grpc.ClientConnInterface, conn *networkservice.Connection) (context.CancelFunc, error) {
+func newEventLoop(ctx context.Context, ec EventConsumer, cc grpc.ClientConnInterface, conn *networkservice.Connection) (context.CancelFunc, error) {
 	conn = conn.Clone()
 	// Is another chain element asking for events?  If not, no need to monitor
 	if ec == nil {

pkg/networkservice/common/monitor/metadata.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2021 Cisco and/or its affiliates.
+// Copyright (c) 2021-2022 Cisco and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -23,6 +23,7 @@ import (
 )
 
 type key struct{}
+type keyEventConsumer struct{}
 
 // store sets the context.CancelFunc stored in per Connection.Id metadata.
 func store(ctx context.Context, isClient bool, cancel context.CancelFunc) {
@@ -39,3 +40,19 @@ func loadAndDelete(ctx context.Context, isClient bool) (value context.CancelFunc
 	value, ok = rawValue.(context.CancelFunc)
 	return value, ok
 }
+
+// storeEventConsumer sets the eventConsumer stored in per Connection.Id metadata.
+func storeEventConsumer(ctx context.Context, isClient bool, eventConsumer EventConsumer) {
+	metadata.Map(ctx, isClient).Store(keyEventConsumer{}, eventConsumer)
+}
+
+// LoadEventConsumer loads EventConsumer stored in per Connection.Id metadata.
+// The loaded result reports whether the key was present.
+func LoadEventConsumer(ctx context.Context, isClient bool) (value EventConsumer, ok bool) {
+	rawValue, ok := metadata.Map(ctx, isClient).Load(keyEventConsumer{})
+	if !ok {
+		return
+	}
+	value, ok = rawValue.(EventConsumer)
+	return value, ok
+}

pkg/networkservice/connectioncontext/mtu/vl3mtu/client.go

@@ -0,0 +1,82 @@
+// Copyright (c) 2022 Cisco and/or its affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package vl3mtu
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+
+	"github.com/networkservicemesh/api/pkg/api/networkservice"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
+)
+
+type vl3MtuClient struct {
+	minMtu uint32
+	m      sync.Mutex
+}
+
+// NewClient - returns a new vl3mtu client chain element.
+// It stores a minimum mtu of the vl3 mtu and updates connection context if required
+func NewClient() networkservice.NetworkServiceClient {
+	return &vl3MtuClient{
+		minMtu: jumboFrameSize,
+	}
+}
+
+func (v *vl3MtuClient) Request(ctx context.Context, request *networkservice.NetworkServiceRequest, opts ...grpc.CallOption) (*networkservice.Connection, error) {
+	c := request.GetConnection()
+	if c.GetContext() == nil {
+		c.Context = &networkservice.ConnectionContext{}
+	}
+
+	// Check MTU of the connection
+	minMTU := atomic.LoadUint32(&v.minMtu)
+	if minMTU < c.GetContext().GetMTU() || c.GetContext().GetMTU() == 0 {
+		c.GetContext().MTU = minMTU
+	}
+
+	conn, err := next.Client(ctx).Request(ctx, request, opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	// Update MTU of the vl3
+	v.updateMinMTU(conn)
+
+	return conn, nil
+}
+
+func (v *vl3MtuClient) Close(ctx context.Context, conn *networkservice.Connection, opts ...grpc.CallOption) (*emptypb.Empty, error) {
+	return next.Client(ctx).Close(ctx, conn, opts...)
+}
+
+func (v *vl3MtuClient) updateMinMTU(conn *networkservice.Connection) {
+	if atomic.LoadUint32(&v.minMtu) <= conn.GetContext().GetMTU() {
+		return
+	}
+
+	v.m.Lock()
+	defer v.m.Unlock()
+	if atomic.LoadUint32(&v.minMtu) <= conn.GetContext().GetMTU() {
+		return
+	}
+	atomic.StoreUint32(&v.minMtu, conn.GetContext().GetMTU())
+}

pkg/networkservice/connectioncontext/mtu/vl3mtu/client_test.go

@@ -0,0 +1,76 @@
+// Copyright (c) 2022 Cisco and/or its affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package vl3mtu_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/networkservicemesh/api/pkg/api/networkservice"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+
+	"github.com/networkservicemesh/sdk/pkg/networkservice/connectioncontext/mtu/vl3mtu"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
+)
+
+func Test_vl3MtuClient(t *testing.T) {
+	t.Cleanup(func() { goleak.VerifyNone(t) })
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	client := next.NewNetworkServiceClient(
+		vl3mtu.NewClient(),
+	)
+
+	// Send the first Request
+	resp1, err := client.Request(ctx, &networkservice.NetworkServiceRequest{
+		Connection: &networkservice.Connection{
+			Id:      "1",
+			Context: &networkservice.ConnectionContext{MTU: 2000},
+		},
+	})
+	require.NoError(t, err)
+	require.Equal(t, uint32(2000), resp1.GetContext().GetMTU())
+
+	// Send request without MTU
+	resp2, err := client.Request(ctx, &networkservice.NetworkServiceRequest{
+		Connection: &networkservice.Connection{
+			Id: "2",
+		},
+	})
+	require.NoError(t, err)
+	require.Equal(t, uint32(2000), resp2.GetContext().GetMTU())
+
+	// Send request with lower MTU
+	resp3, err := client.Request(ctx, &networkservice.NetworkServiceRequest{
+		Connection: &networkservice.Connection{
+			Id:      "3",
+			Context: &networkservice.ConnectionContext{MTU: 1500},
+		},
+	})
+	require.NoError(t, err)
+	require.Equal(t, uint32(1500), resp3.GetContext().GetMTU())
+
+	// Refresh the first connection
+	resp1, err = client.Request(ctx, &networkservice.NetworkServiceRequest{Connection: resp1})
+
+	require.NoError(t, err)
+	require.Equal(t, uint32(1500), resp1.GetContext().GetMTU())
+}

pkg/networkservice/connectioncontext/mtu/vl3mtu/doc.go

@@ -0,0 +1,18 @@
+// Copyright (c) 2022 Cisco and/or its affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package vl3mtu provides networkservice.NetworkService chain elements to store minimum value of vl3 MTU.
+package vl3mtu

pkg/networkservice/connectioncontext/mtu/vl3mtu/server.go

@@ -0,0 +1,103 @@
+// Copyright (c) 2022 Cisco and/or its affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package vl3mtu
+
+import (
+	"context"
+
+	"github.com/edwarnicke/serialize"
+	"github.com/golang/protobuf/ptypes/empty"
+	"github.com/networkservicemesh/api/pkg/api/networkservice"
+
+	"github.com/networkservicemesh/sdk/pkg/networkservice/common/monitor"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/metadata"
+	"github.com/networkservicemesh/sdk/pkg/tools/log"
+)
+
+const (
+	jumboFrameSize = 9000
+)
+
+type vl3MtuServer struct {
+	connections map[string]*networkservice.Connection
+	executor    serialize.Executor
+	minMtu      uint32
+}
+
+// NewServer - returns a new vl3mtu server chain element.
+// It stores a minimum mtu of the vl3 and sends connectionEvent if refresh required
+func NewServer() networkservice.NetworkServiceServer {
+	return &vl3MtuServer{
+		minMtu:      jumboFrameSize,
+		connections: make(map[string]*networkservice.Connection),
+	}
+}
+
+func (v *vl3MtuServer) Request(ctx context.Context, request *networkservice.NetworkServiceRequest) (*networkservice.Connection, error) {
+	// Check MTU of the connection
+	if request.GetConnection().GetContext() == nil {
+		request.GetConnection().Context = &networkservice.ConnectionContext{}
+	}
+	<-v.executor.AsyncExec(func() {
+		if request.GetConnection().GetContext().MTU > v.minMtu || request.GetConnection().GetContext().MTU == 0 {
+			request.GetConnection().GetContext().MTU = v.minMtu
+		}
+	})
+
+	conn, err := next.Server(ctx).Request(ctx, request)
+	if err != nil {
+		return conn, err
+	}
+
+	v.executor.AsyncExec(func() {
+		// We need to update minimum mtu of the vl3 network and send notifications to the already connected clients.
+		logger := log.FromContext(ctx).WithField("vl3MtuServer", "Request")
+		if conn.GetContext().MTU < v.minMtu {
+			v.minMtu = conn.GetContext().MTU
+			logger.Debug("MTU was updated")
+
+			connections := make(map[string]*networkservice.Connection)
+			for _, c := range v.connections {
+				if c.GetId() == conn.GetId() {
+					continue
+				}
+				connections[c.GetId()] = c.Clone()
+				connections[c.GetId()].GetContext().MTU = v.minMtu
+				connections[c.GetId()].State = networkservice.State_REFRESH_REQUESTED
+			}
+			if eventConsumer, ok := monitor.LoadEventConsumer(ctx, metadata.IsClient(v)); ok {
+				_ = eventConsumer.Send(&networkservice.ConnectionEvent{
+					Type:        networkservice.ConnectionEventType_UPDATE,
+					Connections: connections,
+				})
+			} else {
+				logger.Debug("eventConsumer is not presented")
+			}
+		}
+		v.connections[conn.GetId()] = conn
+	})
+
+	return conn, err
+}
+
+func (v *vl3MtuServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
+	v.executor.AsyncExec(func() {
+		delete(v.connections, conn.GetId())
+	})
+	return next.Server(ctx).Close(ctx, conn)
+}