Merge pull request #17 from Nordix/fix-establish-check

integration test fixes

pkg/networkservice/chains/xconnectns/server.go

@@ -97,8 +97,9 @@ func newEndPoint(ctx context.Context, name string, authzServer, resourcePoolServ
 			heal.WithOnRestore(heal.OnRestoreIgnore)),
 		mechanisms.NewServer(map[string]networkservice.NetworkServiceServer{
 			kernelmech.MECHANISM: chain.NewNetworkServiceServer(
-				kernel.NewServer(bridgeName),
+				kernel.NewVethServer(bridgeName),
 				resourcePoolServer,
+				kernel.NewSmartVFServer(bridgeName),
 			),
 			vxlanmech.MECHANISM: vxlan.NewServer(tunnelIP, bridgeName, vxlanInterfacesMutex, vxlanInterfaces),
 		}),

pkg/networkservice/l2ovsconnect/client.go

@@ -47,8 +47,10 @@ func (c *l2ConnectClient) Request(ctx context.Context, request *networkservice.N
 
 	postponeCtxFunc := postpone.ContextWithValues(ctx)
 
+	isEstablished := request.GetConnection().GetNextPathSegment() != nil
+
 	conn, err := next.Client(ctx).Request(ctx, request, opts...)
-	if err != nil || request.GetConnection().GetNextPathSegment() != nil {
+	if err != nil || isEstablished {
 		return conn, err
 	}
 

pkg/networkservice/mechanisms/kernel/client.go

@@ -50,6 +50,8 @@ func NewClient(bridgeName string) networkservice.NetworkServiceClient {
 func (c *kernelClient) Request(ctx context.Context, request *networkservice.NetworkServiceRequest, opts ...grpc.CallOption) (*networkservice.Connection, error) {
 	logger := log.FromContext(ctx).WithField("kernelClient", "Request")
 
+	isEstablished := request.GetConnection().GetNextPathSegment() != nil
+
 	request.MechanismPreferences = append(request.MechanismPreferences, &networkservice.Mechanism{
 		Cls:  cls.LOCAL,
 		Type: kernel.MECHANISM,
@@ -58,7 +60,7 @@ func (c *kernelClient) Request(ctx context.Context, request *networkservice.Netw
 	postponeCtxFunc := postpone.ContextWithValues(ctx)
 
 	conn, err := next.Client(ctx).Request(ctx, request, opts...)
-	if err != nil || request.GetConnection().GetNextPathSegment() != nil {
+	if err != nil || isEstablished {
 		return conn, err
 	}
 

pkg/networkservice/mechanisms/kernel/smartvf_server.go

@@ -0,0 +1,103 @@
+// Copyright (c) 2021 Nordix Foundation.
+//
+// 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.
+
+// +build linux
+
+package kernel
+
+import (
+	"context"
+
+	"github.com/golang/protobuf/ptypes/empty"
+	"github.com/networkservicemesh/api/pkg/api/networkservice"
+	"github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/kernel"
+	"github.com/networkservicemesh/sdk-sriov/pkg/networkservice/common/resourcepool"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
+	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/metadata"
+	"github.com/networkservicemesh/sdk/pkg/tools/log"
+	"github.com/networkservicemesh/sdk/pkg/tools/postpone"
+	"github.com/pkg/errors"
+
+	"github.com/networkservicemesh/sdk-ovs/pkg/tools/ifnames"
+)
+
+type kernelSmartVFServer struct {
+	bridgeName string
+}
+
+// NewSmartVFServer - return a new Smart VF Server chain element for kernel mechanism
+func NewSmartVFServer(bridgeName string) networkservice.NetworkServiceServer {
+	return &kernelSmartVFServer{bridgeName}
+}
+
+// NewClient create a kernel Smart VF server chain element which would be useful to do network plumbing
+// for service client container
+func (k *kernelSmartVFServer) Request(ctx context.Context, request *networkservice.NetworkServiceRequest) (*networkservice.Connection, error) {
+	logger := log.FromContext(ctx).WithField("kernelSmartVFServer", "Request")
+
+	// TODO: remove exists check when switchcase chain element is used
+	_, exists := request.GetConnection().GetMechanism().GetParameters()[resourcepool.TokenIDKey]
+	if !exists {
+		return next.Server(ctx).Request(ctx, request)
+	}
+
+	isEstablished := request.GetConnection().GetNextPathSegment() != nil
+	if !isEstablished {
+		if vfErr := setupVF(ctx, logger, request.GetConnection(), k.bridgeName, metadata.IsClient(k)); vfErr != nil {
+			return nil, vfErr
+		}
+	}
+
+	postponeCtxFunc := postpone.ContextWithValues(ctx)
+
+	conn, err := next.Server(ctx).Request(ctx, request)
+	if err != nil && !isEstablished {
+		closeCtx, cancelClose := postponeCtxFunc()
+		defer cancelClose()
+
+		if _, closeErr := k.Close(closeCtx, conn); closeErr != nil {
+			err = errors.Wrapf(err, "connection closed with error: %s", closeErr.Error())
+		}
+		return nil, err
+	}
+
+	return conn, err
+}
+
+func (k *kernelSmartVFServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
+	logger := log.FromContext(ctx).WithField("kernelSmartVFServer", "Close")
+	_, err := next.Server(ctx).Close(ctx, conn)
+
+	// TODO: remove exists check when switchcase chain element is used
+	_, exists := conn.GetMechanism().GetParameters()[resourcepool.TokenIDKey]
+
+	if mechanism := kernel.ToMechanism(conn.GetMechanism()); mechanism != nil && exists {
+		var kernelServerErr error
+		ovsPortInfo, exists := ifnames.LoadAndDelete(ctx, metadata.IsClient(k))
+		if exists {
+			kernelServerErr = resetVF(logger, ovsPortInfo, k.bridgeName)
+		}
+
+		if err != nil && kernelServerErr != nil {
+			return nil, errors.Wrap(err, kernelServerErr.Error())
+		}
+		if kernelServerErr != nil {
+			return nil, kernelServerErr
+		}
+	}
+
+	return &empty.Empty{}, err
+}

pkg/networkservice/mechanisms/kernel/server.go → pkg/networkservice/mechanisms/kernel/veth_server.go

@@ -28,60 +28,68 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/metadata"
 	"github.com/networkservicemesh/sdk/pkg/tools/log"
+	"github.com/networkservicemesh/sdk/pkg/tools/postpone"
 	"github.com/pkg/errors"
 
 	"github.com/networkservicemesh/sdk-ovs/pkg/tools/ifnames"
 )
 
-type kernelServer struct {
+type kernelVethServer struct {
 	bridgeName string
 }
 
-// NewServer - return a new Server chain element implementing the kernel mechanism with veth pair or smartvf
-func NewServer(bridgeName string) networkservice.NetworkServiceServer {
-	return &kernelServer{bridgeName}
+// NewVethServer - return a new Veth Server chain element for kernel mechanism
+func NewVethServer(bridgeName string) networkservice.NetworkServiceServer {
+	return &kernelVethServer{bridgeName}
 }
 
-// NewClient create a kernel server chain element which would be useful to do network plumbing
+// NewClient create a kernel veth server chain element which would be useful to do network plumbing
 // for service client container
-func (k *kernelServer) Request(ctx context.Context, request *networkservice.NetworkServiceRequest) (*networkservice.Connection, error) {
-	logger := log.FromContext(ctx).WithField("kernelServer", "Request")
-	isEstablished := request.GetConnection().GetNextPathSegment() != nil
+func (k *kernelVethServer) Request(ctx context.Context, request *networkservice.NetworkServiceRequest) (*networkservice.Connection, error) {
+	logger := log.FromContext(ctx).WithField("kernelVethServer", "Request")
+
+	// TODO: remove exists check when switchcase chain element is used
 	_, exists := request.GetConnection().GetMechanism().GetParameters()[resourcepool.TokenIDKey]
-	if !exists && !isEstablished {
+	if exists {
+		return next.Server(ctx).Request(ctx, request)
+	}
+
+	isEstablished := request.GetConnection().GetNextPathSegment() != nil
+	if !isEstablished {
 		if err := setupVeth(ctx, logger, request.Connection, k.bridgeName, metadata.IsClient(k)); err != nil {
 			_ = resetVeth(ctx, logger, request.Connection, k.bridgeName, metadata.IsClient(k))
 			return nil, err
 		}
 	}
+	postponeCtxFunc := postpone.ContextWithValues(ctx)
+
 	conn, err := next.Server(ctx).Request(ctx, request)
-	if err != nil && !exists && !isEstablished {
-		_ = resetVeth(ctx, logger, request.Connection, k.bridgeName, metadata.IsClient(k))
-		return nil, err
-	}
-	if exists && !isEstablished {
-		if vfErr := setupVF(ctx, logger, request.Connection, k.bridgeName, metadata.IsClient(k)); vfErr != nil {
-			return nil, err
+	if err != nil && !isEstablished {
+		closeCtx, cancelClose := postponeCtxFunc()
+		defer cancelClose()
+
+		if _, closeErr := k.Close(closeCtx, conn); closeErr != nil {
+			err = errors.Wrapf(err, "connection closed with error: %s", closeErr.Error())
 		}
+		return nil, err
 	}
+
 	return conn, err
 }
 
-func (k *kernelServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
-	logger := log.FromContext(ctx).WithField("kernelServer", "Close")
+func (k *kernelVethServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
+	logger := log.FromContext(ctx).WithField("kernelVethServer", "Close")
 	_, err := next.Server(ctx).Close(ctx, conn)
 
-	if mechanism := kernel.ToMechanism(conn.GetMechanism()); mechanism != nil {
+	// TODO: remove exists check when switchcase chain element is used
+	_, exists := conn.GetMechanism().GetParameters()[resourcepool.TokenIDKey]
+
+	if mechanism := kernel.ToMechanism(conn.GetMechanism()); mechanism != nil && !exists {
 		var kernelServerErr error
-		ovsPortInfo, exists := ifnames.LoadAndDelete(ctx, metadata.IsClient(k))
+		_, exists := ifnames.LoadAndDelete(ctx, metadata.IsClient(k))
 		if exists {
-			if !ovsPortInfo.IsVfRepresentor {
-				kernelServerErr = resetVeth(ctx, logger, conn, k.bridgeName, metadata.IsClient(k))
-			} else {
-				kernelServerErr = resetVF(logger, ovsPortInfo, k.bridgeName)
-			}
+			kernelServerErr = resetVeth(ctx, logger, conn, k.bridgeName, metadata.IsClient(k))
 		}
-
 		if err != nil && kernelServerErr != nil {
 			return nil, errors.Wrap(err, kernelServerErr.Error())
 		}

pkg/networkservice/mechanisms/vxlan/client.go

@@ -59,10 +59,12 @@ func (c *vxlanClient) Request(ctx context.Context, request *networkservice.Netwo
 		Type: vxlan.MECHANISM,
 	})
 
+	isEstablished := request.GetConnection().GetNextPathSegment() != nil
+
 	postponeCtxFunc := postpone.ContextWithValues(ctx)
 
 	conn, err := next.Client(ctx).Request(ctx, request, opts...)
-	if err != nil || request.GetConnection().GetNextPathSegment() != nil {
+	if err != nil || isEstablished {
 		return conn, err
 	}
 

pkg/networkservice/mechanisms/vxlan/server.go

@@ -28,6 +28,7 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/metadata"
 	"github.com/networkservicemesh/sdk/pkg/tools/log"
+	"github.com/networkservicemesh/sdk/pkg/tools/postpone"
 
 	"github.com/networkservicemesh/sdk-ovs/pkg/tools/ifnames"
 
@@ -64,10 +65,17 @@ func (v *vxlanServer) Request(ctx context.Context, request *networkservice.Netwo
 		}
 	}
 
+	postponeCtxFunc := postpone.ContextWithValues(ctx)
+
 	conn, err := next.Server(ctx).Request(ctx, request)
 	if err != nil && !isEstablished {
-		_ = remove(conn, v.bridgeName, v.vxlanInterfacesMutex, v.vxlanInterfacesMap, metadata.IsClient(v))
-		return conn, err
+		closeCtx, cancelClose := postponeCtxFunc()
+		defer cancelClose()
+
+		if _, closeErr := v.Close(closeCtx, conn); closeErr != nil {
+			err = errors.Wrapf(err, "connection closed with error: %s", closeErr.Error())
+		}
+		return nil, err
 	}
 
 	return conn, err