update caches after testing

Signed-off-by: denis-tingaikin <[email protected]>

pkg/networkservice/chains/nsmgr/server.go

@@ -240,7 +240,7 @@ func NewServer(ctx context.Context, tokenGenerator token.GeneratorFunc, options
 			chain.NewNetworkServiceRegistryClient(
 				clienturl.NewNetworkServiceRegistryClient(opts.regURL),
 				begin.NewNetworkServiceRegistryClient(),
-				//querycache.NewNetworkServiceRegistryClient(ctx),
+				querycache.NewNetworkServiceRegistryClient(ctx),
 				clientconn.NewNetworkServiceRegistryClient(),
 				opts.authorizeNSRegistryClient,
 				grpcmetadata.NewNetworkServiceRegistryClient(),

pkg/networkservice/chains/nsmgr/single_test.go

@@ -373,6 +373,14 @@ func Test_UsecasePoint2MultiPoint(t *testing.T) {
 	_, err = nsc.Close(ctx, conn)
 	require.NoError(t, err)
 
+	require.Eventually(t, func() bool {
+		conn, err = nsc.Request(ctx, request.Clone())
+		if err != nil {
+			return false
+		}
+		return len(conn.Path.PathSegments) == 4 && conn.GetPath().GetPathSegments()[2].Name == "p2p forwarder"
+	}, time.Second, time.Second/10)
+
 	conn, err = nsc.Request(ctx, request.Clone())
 	require.NoError(t, err)
 	require.NotNil(t, conn)
@@ -383,9 +391,12 @@ func Test_UsecasePoint2MultiPoint(t *testing.T) {
 func Test_RemoteUsecase_Point2MultiPoint(t *testing.T) {
 	t.Cleanup(func() { goleak.VerifyNone(t) })
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Hour*5)
 	defer cancel()
 
+	// log.EnableTracing(true)
+	// logrus.SetLevel(logrus.TraceLevel)
+
 	const nodeCount = 2
 
 	domain := sandbox.NewBuilder(ctx, t).
@@ -484,6 +495,14 @@ func Test_RemoteUsecase_Point2MultiPoint(t *testing.T) {
 	_, err = nsc.Close(ctx, conn)
 	require.NoError(t, err)
 
+	require.Eventually(t, func() bool {
+		conn, err = nsc.Request(ctx, request.Clone())
+		if err != nil {
+			return false
+		}
+		return len(conn.Path.PathSegments) == 6 && conn.GetPath().GetPathSegments()[2].Name == "p2p forwarder-0" && conn.GetPath().GetPathSegments()[4].Name == "p2p forwarder-1"
+	}, time.Second, time.Second/10)
+
 	conn, err = nsc.Request(ctx, request.Clone())
 	require.NoError(t, err)
 	require.NotNil(t, conn)

pkg/networkservice/common/discoverforwarder/server.go

@@ -197,9 +197,10 @@ func (d *discoverForwarderServer) matchForwarders(nsLabels map[string]string, ns
 
 		var matchLabels = match.GetMetadata().GetLabels()
 		if matchLabels == nil {
-			matchLabels = map[string]string{
-				"p2p": "true",
-			}
+			matchLabels = make(map[string]string)
+		}
+		if len(matchLabels) == 0 {
+			matchLabels["p2p"] = "true"
 		}
 		for _, nse := range nses {
 			var forwarderLabels = nse.GetNetworkServiceLabels()[d.forwarderServiceName]
@@ -217,7 +218,6 @@ func (d *discoverForwarderServer) matchForwarders(nsLabels map[string]string, ns
 
 		break
 	}
-
 	return result
 }
 

pkg/registry/chains/client/ns_client.go

@@ -32,6 +32,7 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/registry/common/grpcmetadata"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/heal"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/null"
+	"github.com/networkservicemesh/sdk/pkg/registry/common/querycache"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/retry"
 	"github.com/networkservicemesh/sdk/pkg/registry/core/chain"
 	"github.com/networkservicemesh/sdk/pkg/registry/utils/metadata"
@@ -53,6 +54,7 @@ func NewNetworkServiceRegistryClient(ctx context.Context, opts ...Option) regist
 			[]registry.NetworkServiceRegistryClient{
 				begin.NewNetworkServiceRegistryClient(),
 				metadata.NewNetworkServiceClient(),
+				querycache.NewNetworkServiceRegistryClient(ctx),
 				retry.NewNetworkServiceRegistryClient(ctx),
 				clientOpts.authorizeNSRegistryClient,
 				heal.NewNetworkServiceRegistryClient(ctx),

pkg/registry/chains/client/nse_client.go

@@ -33,6 +33,7 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/registry/common/grpcmetadata"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/heal"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/null"
+	"github.com/networkservicemesh/sdk/pkg/registry/common/querycache"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/refresh"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/retry"
 	"github.com/networkservicemesh/sdk/pkg/registry/core/chain"
@@ -55,6 +56,7 @@ func NewNetworkServiceEndpointRegistryClient(ctx context.Context, opts ...Option
 			[]registry.NetworkServiceEndpointRegistryClient{
 				begin.NewNetworkServiceEndpointRegistryClient(),
 				metadata.NewNetworkServiceEndpointClient(),
+				querycache.NewNetworkServiceEndpointRegistryClient(ctx),
 				retry.NewNetworkServiceEndpointRegistryClient(ctx),
 				heal.NewNetworkServiceEndpointRegistryClient(ctx),
 				refresh.NewNetworkServiceEndpointRegistryClient(ctx),

pkg/registry/common/localbypass/server.go

@@ -39,6 +39,7 @@ type localBypassNSEFindServer struct {
 }
 
 func (s *localBypassNSEFindServer) Send(nseResp *registry.NetworkServiceEndpointResponse) error {
+	nseResp = nseResp.Clone()
 	if u, ok := s.nseURLs.Load(nseResp.NetworkServiceEndpoint.Name); ok {
 		nseResp.NetworkServiceEndpoint.Url = u.String()
 	}