Add forwarder reuse (#1232)

* add forwarder matching Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * use path index instead of magic number Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * minor fixes for tests Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * fix linter issue Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * make samples Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * fix linter issues Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * fix heal tests Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * change iter count Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * fix timeout value Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * change iter count again Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * change count Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * increase timeout Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * apply suggestions Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add debug info Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * forwarder health check after nsmgr restart Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * disable traces Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * cleanup Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * rework health check Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * enable tracing Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * disable tracing Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * increase timeout Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * rework forwarder health check again Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add simple sleep Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * disable tracing Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * increase sleep time Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add healing Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * tracing Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add delete Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * disable race Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * revert 'add delete' commit Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * enable tracing Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add client conn delete Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * delete registry restart Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test with forwarder health check Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * remove health check + increase sleep Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * change iter count Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add forwarder registration Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * fix linter issue Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * enable all tests Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test with health check again Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * run only one test Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add forwarder register Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test with clientconn delete again Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * revert last commit Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * increase timeout Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add log Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test with forwarder register again Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * increase forwarder count Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add race flag + disable tracing Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * move test to heal_test.go file Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add clientconn.Delete(ctx) again Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test only local case Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test with registry restart Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * cleanup Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test retry patch Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add logs Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * wake CI Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test with 100 ms Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test with 1s Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * test without retry Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add manual forwarder heal Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add remote test Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * reduce tryTimeout Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * add nse reregistration Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * cleanup Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * fix linter issue Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * enable all tests Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * fix linter issue Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com> * apply suggestions Signed-off-by: Nikita Skrynnik <nikita.skrynnik@xored.com>
networkservicemesh · Mar 14, 2022 · 37b596e · 37b596e
1 parent 793dfca
commit 37b596e
Show file tree

Hide file tree

Showing 2 changed files with 125 additions and 0 deletions.
diff --git a/pkg/networkservice/chains/nsmgr/heal_test.go b/pkg/networkservice/chains/nsmgr/heal_test.go
@@ -18,6 +18,7 @@ package nsmgr_test
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -445,3 +446,116 @@ func checkSecondRequestsReceived(requestsDone func() int) func() bool {
 		return requestsDone() >= 2
 	}
 }
+
+func Test_ForwarderShouldBeSelectedCorrectlyOnNSMgrRestart(t *testing.T) {
+	var samples = []struct {
+		name             string
+		nodeNum          int
+		pathSegmentCount int
+	}{
+		{
+			name:             "Local",
+			nodeNum:          0,
+			pathSegmentCount: 4,
+		},
+		{
+			name:             "Remote",
+			nodeNum:          1,
+			pathSegmentCount: 6,
+		},
+	}
+
+	for _, sample := range samples {
+		t.Run(sample.name, func(t *testing.T) {
+			// nolint:scopelint
+			testForwarderShouldBeSelectedCorrectlyOnNSMgrRestart(t, sample.nodeNum, sample.pathSegmentCount)
+		})
+	}
+}
+
+func testForwarderShouldBeSelectedCorrectlyOnNSMgrRestart(t *testing.T, nodeNum, pathSegmentCount int) {
+	t.Cleanup(func() { goleak.VerifyNone(t) })
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
+	defer cancel()
+
+	domain := sandbox.NewBuilder(ctx, t).
+		SetNodesCount(nodeNum + 1).
+		SetRegistryProxySupplier(nil).
+		SetNSMgrProxySupplier(nil).
+		Build()
+
+	var expectedForwarderName string
+
+	require.Len(t, domain.Nodes[nodeNum].Forwarders, 1)
+	for k := range domain.Nodes[nodeNum].Forwarders {
+		expectedForwarderName = k
+	}
+
+	nsRegistryClient := domain.NewNSRegistryClient(ctx, sandbox.GenerateTestToken)
+
+	_, err := nsRegistryClient.Register(ctx, &registry.NetworkService{
+		Name: "my-ns",
+	})
+	require.NoError(t, err)
+
+	nseReg := &registry.NetworkServiceEndpoint{
+		Name:                "my-nse-1",
+		NetworkServiceNames: []string{"my-ns"},
+	}
+
+	nseEntry := domain.Nodes[nodeNum].NewEndpoint(ctx, nseReg, sandbox.GenerateTestToken)
+
+	nsc := domain.Nodes[0].NewClient(ctx, sandbox.GenerateTestToken)
+
+	request := defaultRequest("my-ns")
+
+	conn, err := nsc.Request(ctx, request.Clone())
+	require.NoError(t, err)
+	require.NotNil(t, conn)
+	require.Equal(t, pathSegmentCount, len(conn.Path.PathSegments))
+	require.Equal(t, expectedForwarderName, conn.GetPath().GetPathSegments()[pathSegmentCount-2].Name)
+
+	for i := 0; i < 10; i++ {
+		request.Connection = conn.Clone()
+		conn, err = nsc.Request(ctx, request.Clone())
+
+		require.NoError(t, err)
+		require.Equal(t, expectedForwarderName, conn.GetPath().GetPathSegments()[pathSegmentCount-2].Name)
+
+		domain.Nodes[nodeNum].NSMgr.Restart()
+
+		domain.Nodes[nodeNum].NewForwarder(ctx, &registry.NetworkServiceEndpoint{
+			Name:                sandbox.UniqueName(fmt.Sprintf("%v-forwarder", i)),
+			NetworkServiceNames: []string{"forwarder"},
+			NetworkServiceLabels: map[string]*registry.NetworkServiceLabels{
+				"forwarder": {
+					Labels: map[string]string{
+						"p2p": "true",
+					},
+				},
+			},
+		}, sandbox.GenerateTestToken)
+
+		_, err = domain.Nodes[nodeNum].NSMgr.NetworkServiceEndpointRegistryServer().Register(ctx, &registry.NetworkServiceEndpoint{
+			Name:                expectedForwarderName,
+			Url:                 domain.Nodes[nodeNum].Forwarders[expectedForwarderName].URL.String(),
+			NetworkServiceNames: []string{"forwarder"},
+			NetworkServiceLabels: map[string]*registry.NetworkServiceLabels{
+				"forwarder": {
+					Labels: map[string]string{
+						"p2p": "true",
+					},
+				},
+			},
+		})
+		require.NoError(t, err)
+
+		_, err = domain.Nodes[nodeNum].NSMgr.NetworkServiceEndpointRegistryServer().Register(ctx, &registry.NetworkServiceEndpoint{
+			Name:                nseReg.Name,
+			Url:                 nseEntry.URL.String(),
+			NetworkServiceNames: nseReg.NetworkServiceNames,
+		})
+		require.NoError(t, err)
+	}
+}
diff --git a/pkg/networkservice/common/discoverforwarder/server.go b/pkg/networkservice/common/discoverforwarder/server.go
@@ -93,6 +93,17 @@ func (d *discoverForwarderServer) Request(ctx context.Context, request *networks
 			return nil, errors.New("no candidates found")
 		}
 
+		segments := request.Connection.GetPath().GetPathSegments()
+		if pathIndex := int(request.Connection.GetPath().Index); len(segments) > pathIndex+1 {
+			datapathForwarder := segments[pathIndex+1].Name
+			for i, candidate := range nses {
+				if candidate.Name == datapathForwarder {
+					nses[0], nses[i] = nses[i], nses[0]
+					break
+				}
+			}
+		}
+
 		var candidatesErr = errors.New("all forwarders have failed")
 
 		// TODO: Should we consider about load balancing?