From 167abe71017e09f57628fe3ef36c92943d27e77f Mon Sep 17 00:00:00 2001 From: Vladimir Popov Date: Tue, 21 Sep 2021 11:58:59 +0700 Subject: [PATCH] Use watchers with 1m timeout Signed-off-by: Vladimir Popov --- pkg/registry/etcd/ns_server.go | 65 ++++++++++++++++++---------- pkg/registry/etcd/nse_server.go | 77 ++++++++++++++++++++------------- 2 files changed, 88 insertions(+), 54 deletions(-) diff --git a/pkg/registry/etcd/ns_server.go b/pkg/registry/etcd/ns_server.go index 0c693be..ff8f927 100644 --- a/pkg/registry/etcd/ns_server.go +++ b/pkg/registry/etcd/ns_server.go @@ -20,6 +20,7 @@ import ( "context" "errors" "io" + "time" "github.com/golang/protobuf/ptypes/empty" "github.com/networkservicemesh/sdk/pkg/tools/log" @@ -74,38 +75,54 @@ func (n *etcdNSRegistryServer) Register(ctx context.Context, request *registry.N } func (n *etcdNSRegistryServer) watch(query *registry.NetworkServiceQuery, s registry.NetworkServiceRegistry_FindServer) error { - logger := log.FromContext(n.chainContext).WithField("etcdNSRegistryServer", "watch") - - for { - watcher, err := n.client.NetworkservicemeshV1().NetworkServices(n.ns).Watch(s.Context(), metav1.ListOptions{}) + var watchErr error + for watchErr == nil { + timeoutSeconds := int64(time.Minute / time.Second) + watcher, err := n.client.NetworkservicemeshV1().NetworkServices(n.ns).Watch(s.Context(), metav1.ListOptions{ + TimeoutSeconds: &timeoutSeconds, + }) if err != nil { return err } - var event watch.Event - for watcherOpened := true; watcherOpened; { - select { - case <-s.Context().Done(): - return s.Context().Err() - case event, watcherOpened = <-watcher.ResultChan(): - if !watcherOpened { - logger.Warn("watcher is closed, retrying") - continue - } - if event.Type != watch.Added { - continue - } - model := event.Object.(*v1.NetworkService) - item := (*registry.NetworkService)(&model.Spec) - if matchutils.MatchNetworkServices(query.NetworkService, item) { - err := s.Send(item) - if err != nil { - return err - } + watchErr = n.handleWatcher(watcher, query, s) + + watcher.Stop() + } + return watchErr +} + +func (n *etcdNSRegistryServer) handleWatcher( + watcher watch.Interface, + query *registry.NetworkServiceQuery, + s registry.NetworkServiceRegistry_FindServer, +) error { + logger := log.FromContext(n.chainContext).WithField("etcdNSRegistryServer", "handleWatcher") + + var event watch.Event + for watcherOpened := true; watcherOpened; { + select { + case <-s.Context().Done(): + return s.Context().Err() + case event, watcherOpened = <-watcher.ResultChan(): + if !watcherOpened { + logger.Warn("watcher is closed, retrying") + continue + } + if event.Type != watch.Added { + continue + } + model := event.Object.(*v1.NetworkService) + item := (*registry.NetworkService)(&model.Spec) + if matchutils.MatchNetworkServices(query.NetworkService, item) { + err := s.Send(item) + if err != nil { + return err } } } } + return nil } func (n *etcdNSRegistryServer) Find(query *registry.NetworkServiceQuery, s registry.NetworkServiceRegistry_FindServer) error { diff --git a/pkg/registry/etcd/nse_server.go b/pkg/registry/etcd/nse_server.go index 0ceb249..a62efc4 100644 --- a/pkg/registry/etcd/nse_server.go +++ b/pkg/registry/etcd/nse_server.go @@ -20,6 +20,7 @@ import ( "context" "errors" "io" + "time" "github.com/golang/protobuf/ptypes/empty" "github.com/networkservicemesh/sdk/pkg/tools/log" @@ -116,44 +117,60 @@ func (n *etcdNSERegistryServer) Unregister(ctx context.Context, request *registr } func (n *etcdNSERegistryServer) watch(query *registry.NetworkServiceEndpointQuery, s registry.NetworkServiceEndpointRegistry_FindServer) error { - logger := log.FromContext(n.chainContext).WithField("etcdNSERegistryServer", "watch") - - for { - watcher, err := n.client.NetworkservicemeshV1().NetworkServiceEndpoints(n.ns).Watch(s.Context(), metav1.ListOptions{}) + var watchErr error + for watchErr == nil { + timeoutSeconds := int64(time.Minute / time.Second) + watcher, err := n.client.NetworkservicemeshV1().NetworkServiceEndpoints(n.ns).Watch(s.Context(), metav1.ListOptions{ + TimeoutSeconds: &timeoutSeconds, + }) if err != nil { return err } - var event watch.Event - for watcherOpened := true; watcherOpened; { - select { - case <-n.chainContext.Done(): - return n.chainContext.Err() - case <-s.Context().Done(): - return s.Context().Err() - case event, watcherOpened = <-watcher.ResultChan(): - if !watcherOpened { - logger.Warn("watcher is closed, retrying") - continue - } - model, ok := event.Object.(*v1.NetworkServiceEndpoint) - if !ok { - logger.Errorf("event: %v", event) - continue - } - item := (*registry.NetworkServiceEndpoint)(&model.Spec) - if event.Type == watch.Deleted { - item.ExpirationTime.Seconds = -1 - } - if matchutils.MatchNetworkServiceEndpoints(query.NetworkServiceEndpoint, item) { - err := s.Send(item) - if err != nil { - return err - } + watchErr = n.handleWatcher(watcher, query, s) + + watcher.Stop() + } + return watchErr +} + +func (n *etcdNSERegistryServer) handleWatcher( + watcher watch.Interface, + query *registry.NetworkServiceEndpointQuery, + s registry.NetworkServiceEndpointRegistry_FindServer, +) error { + logger := log.FromContext(n.chainContext).WithField("etcdNSERegistryServer", "handleWatcher") + + var event watch.Event + for watcherOpened := true; watcherOpened; { + select { + case <-n.chainContext.Done(): + return n.chainContext.Err() + case <-s.Context().Done(): + return s.Context().Err() + case event, watcherOpened = <-watcher.ResultChan(): + if !watcherOpened { + logger.Warn("watcher is closed, retrying") + continue + } + model, ok := event.Object.(*v1.NetworkServiceEndpoint) + if !ok { + logger.Errorf("event: %v", event) + continue + } + item := (*registry.NetworkServiceEndpoint)(&model.Spec) + if event.Type == watch.Deleted { + item.ExpirationTime.Seconds = -1 + } + if matchutils.MatchNetworkServiceEndpoints(query.NetworkServiceEndpoint, item) { + err := s.Send(item) + if err != nil { + return err } } } } + return nil } // NewNetworkServiceEndpointRegistryServer creates new registry.NetworkServiceRegistryServer that is using etcd to store network services.