Rework a cache for NS/NSE registry

.golangci.yml

@@ -36,7 +36,7 @@ linters-settings:
   dupl:
     threshold: 150
   funlen:
-    Lines: 120
+    Lines: 130
     Statements: 60
   goconst:
     min-len: 2

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/edwarnicke/serialize v1.0.7
 	github.com/fsnotify/fsnotify v1.5.4
 	github.com/ghodss/yaml v1.0.0
+	github.com/go-pkgz/expirable-cache/v3 v3.0.0
 	github.com/golang-jwt/jwt/v4 v4.5.1
 	github.com/golang/protobuf v1.5.3
 	github.com/google/go-cmp v0.6.0

go.sum

@@ -79,6 +79,8 @@ github.com/go-logr/logr v1.3.0 h1:2y3SDp0ZXuc6/cjLSZ+Q3ir+QB9T/iG5yYRXqsagWSY=
 github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/go-pkgz/expirable-cache/v3 v3.0.0 h1:u3/gcu3sabLYiTCevoRKv+WzjIn5oo7P8XtiXBeRDLw=
+github.com/go-pkgz/expirable-cache/v3 v3.0.0/go.mod h1:2OQiDyEGQalYecLWmXprm3maPXeVb5/6/X7yRPYTzec=
 github.com/go-sql-driver/mysql v1.6.0 h1:BCTh4TKNUYmOmMUcQ3IipzF5prigylS7XXjEkfCHuOE=
 github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg=
 github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=
@@ -143,6 +145,7 @@ github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/C
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1 h1:0hERBMJE1eitiLkihrMvRVBYAkpHzc/J3QdDN+dAcgU=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
 github.com/hashicorp/raft v1.3.9 h1:9yuo1aR0bFTr1cw7pj3S2Bk6MhJCsnr2NAxvIBrP2x4=
 github.com/hashicorp/raft v1.3.9/go.mod h1:4Ak7FSPnuvmb0GV6vgIAJ4vYT4bek9bb6Q+7HVbyzqM=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=

pkg/networkservice/chains/nsmgr/server.go

@@ -55,6 +55,7 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/registry/common/grpcmetadata"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/localbypass"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/memory"
+	"github.com/networkservicemesh/sdk/pkg/registry/common/querycache"
 	registryrecvfd "github.com/networkservicemesh/sdk/pkg/registry/common/recvfd"
 	registrysendfd "github.com/networkservicemesh/sdk/pkg/registry/common/sendfd"
 	"github.com/networkservicemesh/sdk/pkg/registry/common/updatepath"
@@ -239,6 +240,7 @@ func NewServer(ctx context.Context, tokenGenerator token.GeneratorFunc, options
 			chain.NewNetworkServiceRegistryClient(
 				clienturl.NewNetworkServiceRegistryClient(opts.regURL),
 				begin.NewNetworkServiceRegistryClient(),
+				querycache.NewNetworkServiceRegistryClient(ctx),
 				clientconn.NewNetworkServiceRegistryClient(),
 				opts.authorizeNSRegistryClient,
 				grpcmetadata.NewNetworkServiceRegistryClient(),
@@ -263,6 +265,7 @@ func NewServer(ctx context.Context, tokenGenerator token.GeneratorFunc, options
 		registryconnect.NewNetworkServiceEndpointRegistryServer(
 			chain.NewNetworkServiceEndpointRegistryClient(
 				begin.NewNetworkServiceEndpointRegistryClient(),
+				querycache.NewNetworkServiceEndpointRegistryClient(ctx),
 				clienturl.NewNetworkServiceEndpointRegistryClient(opts.regURL),
 				clientconn.NewNetworkServiceEndpointRegistryClient(),
 				opts.authorizeNSERegistryClient,

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)
@@ -484,6 +492,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

@@ -1,6 +1,6 @@
 // Copyright (c) 2020-2022 Doc.ai and/or its affiliates.
 //
-// Copyright (c) 2023 Cisco and/or its affiliates.
+// Copyright (c) 2023-2024 Cisco and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -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()
 	}

pkg/registry/common/querycache/ns_client.go

@@ -0,0 +1,88 @@
+// Copyright (c) 2024 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 querycache adds possible to cache Find queries
+package querycache
+
+import (
+	"context"
+	"time"
+
+	cache "github.com/go-pkgz/expirable-cache/v3"
+	"github.com/golang/protobuf/ptypes/empty"
+	"google.golang.org/grpc"
+
+	"github.com/networkservicemesh/api/pkg/api/registry"
+
+	"github.com/networkservicemesh/sdk/pkg/registry/core/next"
+	"github.com/networkservicemesh/sdk/pkg/registry/core/streamchannel"
+)
+
+type queryCacheNSClient struct {
+	chainContext context.Context
+	cache        cache.Cache[string, []*registry.NetworkService]
+}
+
+// NewNetworkServiceRegistryClient creates new querycache NS registry client that caches all resolved NSs
+func NewNetworkServiceRegistryClient(ctx context.Context) registry.NetworkServiceRegistryClient {
+	var res = &queryCacheNSClient{
+		chainContext: ctx,
+		cache:        cache.NewCache[string, []*registry.NetworkService]().WithLRU().WithMaxKeys(32).WithTTL(time.Millisecond * 100),
+	}
+	return res
+}
+
+func (q *queryCacheNSClient) Register(ctx context.Context, nse *registry.NetworkService, opts ...grpc.CallOption) (*registry.NetworkService, error) {
+	resp, err := next.NetworkServiceRegistryClient(ctx).Register(ctx, nse, opts...)
+	if err == nil {
+		q.cache.Add(resp.GetName(), []*registry.NetworkService{resp})
+	}
+	return resp, err
+}
+
+func (q *queryCacheNSClient) Find(ctx context.Context, query *registry.NetworkServiceQuery, opts ...grpc.CallOption) (registry.NetworkServiceRegistry_FindClient, error) {
+	if query.Watch {
+		return next.NetworkServiceRegistryClient(ctx).Find(ctx, query, opts...)
+	}
+
+	var list []*registry.NetworkService
+	if v, ok := q.cache.Get(query.GetNetworkService().GetName()); ok {
+		list = v
+	} else {
+		var streamClient, err = next.NetworkServiceRegistryClient(ctx).Find(ctx, query, opts...)
+		if err != nil {
+			return streamClient, err
+		}
+		list = registry.ReadNetworkServiceList(streamClient)
+		for _, item := range list {
+			q.cache.Add(item.GetName(), []*registry.NetworkService{item.Clone()})
+		}
+	}
+	var resultStreamChannel = make(chan *registry.NetworkServiceResponse, len(list))
+	for _, item := range list {
+		resultStreamChannel <- &registry.NetworkServiceResponse{NetworkService: item}
+	}
+	close(resultStreamChannel)
+	return streamchannel.NewNetworkServiceFindClient(ctx, resultStreamChannel), nil
+}
+
+func (q *queryCacheNSClient) Unregister(ctx context.Context, in *registry.NetworkService, opts ...grpc.CallOption) (*empty.Empty, error) {
+	resp, err := next.NetworkServiceRegistryClient(ctx).Unregister(ctx, in, opts...)
+	if err == nil {
+		q.cache.Remove(in.GetName())
+	}
+	return resp, err
+}