Repository: vmware-tanzu/octant
Pull request #632 (2020-02-05 16:05:06)

Remove extra actions from poller and websocket client
**What this PR does / why we need it**:

There are a number of duplicated items in the poller and websocket:
* closing websockets
* the poller loop itself

This change attempts to remove some of these duplications.

This change also removes thrashing the websocket client.
The client was being closed too often.

**Which issue(s) this PR fixes**
 #631 

**Special notes for your reviewer**:



Patch:
diff --git a/changelogs/unreleased/631-bryanl b/changelogs/unreleased/631-bryanl
new file mode 100644
index 00000000..62e88659
--- /dev/null
+++ b/changelogs/unreleased/631-bryanl
@@ -0,0 +1,2 @@
+Remove extra actions from poller and websocket client
+
diff --git a/internal/api/content_manager.go b/internal/api/content_manager.go
index 210dcd1f..4b99fc59 100644
--- a/internal/api/content_manager.go
+++ b/internal/api/content_manager.go
@@ -31,7 +31,18 @@ type ContentManagerOption func(manager *ContentManager)
 
 // ContentGenerateFunc is a function that generates content. It returns `rerun=true`
 // if the action should be be immediately rerun.
-type ContentGenerateFunc func(ctx context.Context, state octant.State) (component.ContentResponse, bool, error)
+type ContentGenerateFunc func(ctx context.Context, state octant.State) (Content, bool, error)
+
+type Content struct {
+	Response component.ContentResponse
+	Path     string
+}
+
+var (
+	emptyContent = Content{
+		Response: component.EmptyContentResponse,
+	}
+)
 
 // WithContentGenerator configures the content generate function.
 func WithContentGenerator(fn ContentGenerateFunc) ContentManagerOption {
@@ -77,7 +88,11 @@ var _ StateManager = (*ContentManager)(nil)
 
 // Start starts the manager.
 func (cm *ContentManager) Start(ctx context.Context, state octant.State, s OctantClient) {
+	logger := log.From(ctx)
+	logger.Debugf("starting content manager")
+
 	defer func() {
+		logger.Debugf("stopping content manager")
 		close(cm.updateContentCh)
 	}()
 
@@ -96,7 +111,7 @@ func (cm *ContentManager) runUpdate(state octant.State, s OctantClient) PollerFu
 			return false
 		}
 
-		contentResponse, _, err := cm.contentGenerateFunc(ctx, state)
+		content, _, err := cm.contentGenerateFunc(ctx, state)
 		if err != nil {
 			var ae *oerrors.AccessError
 			if errors.As(err, &ae) {
@@ -112,14 +127,17 @@ func (cm *ContentManager) runUpdate(state octant.State, s OctantClient) PollerFu
 		}
 
 		if ctx.Err() == nil {
-			s.Send(CreateContentEvent(contentResponse, state.GetNamespace(), contentPath, state.GetQueryParams()))
+			if content.Path == state.GetContentPath() {
+				s.Send(CreateContentEvent(content.Response, state.GetNamespace(), contentPath, state.GetQueryParams()))
+			}
+
 		}
 
 		return false
 	}
 }
 
-func (cm *ContentManager) generateContent(ctx context.Context, state octant.State) (component.ContentResponse, bool, error) {
+func (cm *ContentManager) generateContent(ctx context.Context, state octant.State) (Content, bool, error) {
 	contentPath := state.GetContentPath()
 	logger := cm.logger.With("contentPath", contentPath)
 
@@ -130,7 +148,7 @@ func (cm *ContentManager) generateContent(ctx context.Context, state octant.Stat
 
 	m, ok := cm.moduleManager.ModuleForContentPath(contentPath)
 	if !ok {
-		return component.EmptyContentResponse, false, fmt.Errorf("unable to find module for content path %q", contentPath)
+		return emptyContent, false, fmt.Errorf("unable to find module for content path %q", contentPath)
 	}
 	modulePath := strings.TrimPrefix(contentPath, m.Name())
 	options := module.ContentOptions{
@@ -141,13 +159,17 @@ func (cm *ContentManager) generateContent(ctx context.Context, state octant.Stat
 		if nfe, ok := err.(notFound); ok && nfe.NotFound() {
 			logger.Debugf("path not found, redirecting to parent")
 			state.SetContentPath(notFoundRedirectPath(contentPath))
-			return component.EmptyContentResponse, true, nil
+			return emptyContent, true, nil
 		} else {
-			return component.EmptyContentResponse, false, fmt.Errorf("generate content: %w", err)
+			return emptyContent, false, fmt.Errorf("generate content: %w", err)
 		}
 	}
 
-	return contentResponse, false, nil
+	content := Content{
+		Response: contentResponse,
+		Path:     contentPath,
+	}
+	return content, false, nil
 }
 
 // Handlers returns a slice of client request handlers.
@@ -200,6 +222,8 @@ func (cm *ContentManager) SetContentPath(state octant.State, payload action.Payl
 		return fmt.Errorf("extract query params from payload: %w", err)
 	}
 
+	cm.logger.With("content-path", contentPath).Debugf("setting content path")
+
 	state.SetContentPath(contentPath)
 	return nil
 }
diff --git a/internal/api/poller.go b/internal/api/poller.go
index 6d788f3c..d6145bc7 100644
--- a/internal/api/poller.go
+++ b/internal/api/poller.go
@@ -7,7 +7,7 @@ package api
 
 import (
 	"context"
-	"sync"
+	"fmt"
 	"time"
 
 	"github.com/google/uuid"
@@ -24,6 +24,8 @@ type PollerFunc func(context.Context) bool
 
 // Poller is a poller. It runs an action.
 type Poller interface {
+	// Run runs `action` and delays `resetDuration` before starting again. If a message
+	// is sent to `ch`, it will cancel current work and restart `action`.
 	Run(ctx context.Context, ch <-chan struct{}, action PollerFunc, resetDuration time.Duration)
 }
 
@@ -38,7 +40,7 @@ func NewSingleRunPoller() *SingleRunPoller {
 }
 
 // Run runs the poller.
-func (a SingleRunPoller) Run(ctx context.Context, ch <-chan struct{}, action PollerFunc, resetDuration time.Duration) {
+func (a SingleRunPoller) Run(ctx context.Context, _ <-chan struct{}, action PollerFunc, resetDuration time.Duration) {
 	action(ctx)
 }
 
@@ -51,129 +53,45 @@ var _ Poller = (*InterruptiblePoller)(nil)
 
 // NewInterruptiblePoller creates an instance of InterruptiblePoller.
 func NewInterruptiblePoller(name string) *InterruptiblePoller {
-	return &InterruptiblePoller{name: name}
+	return &InterruptiblePoller{
+		name: name,
+	}
 }
 
 // Run runs the poller.
-func (ip *InterruptiblePoller) Run(ctx context.Context, ch <-chan struct{}, action PollerFunc, resetDuration time.Duration) {
-	logger := log.From(ctx).With("poller-name", ip.name)
-	ctx = log.WithLoggerContext(ctx, logger)
-
-	jt := initJobTracker(ctx, action)
-	defer jt.clear()
-
-	pollerQueue := make(chan job, 10)
+func (a *InterruptiblePoller) Run(ctx context.Context, ch <-chan struct{}, action PollerFunc, resetDuration time.Duration) {
+	logger := log.From(ctx).With(
+		"poller-name", a.name,
+		"poller-instance", uuid.New().String())
+	logger.Debugf("starting poller")
+
+	timer := time.NewTimer(0)
+	done := false
+	for !done {
+		func() {
+			cur, cancel := context.WithCancel(log.WithLoggerContext(ctx, logger))
+			defer cancel()
+			canceled := false
 
-	worker := func() {
-		for j := range pollerQueue {
 			select {
-			case <-j.ctx.Done():
-				// Job's context was canceled. Nothing else to do here.
-			case <-j.run():
-				if j.ctx.Err() == nil {
-					<-time.After(resetDuration)
-					pollerQueue <- jt.create()
-				}
+			case <-ctx.Done():
+				logger.Debugf("poller has been canceled")
+				done = true
+			case <-ch:
+				canceled = true
+				logger.Debugf("poller was interrupted")
+			case <-timer.C:
+				logger.Debugf("poller is running action")
+				now := time.Now()
+				action(cur)
+				logger.With("elapsed", fmt.Sprintf("%s", time.Since(now))).
+					Debugf("poller ran action")
 			}
-		}
-	}
-
-	for i := 0; i < pollerWorkerCount; i++ {
-		go worker()
-	}
-
-	go func() {
-		for range ch {
-			// Cancel all existing jobs before creating a new job.
-			jt.clear()
-			pollerQueue <- jt.create()
-		}
-	}()
-
-	pollerQueue <- jt.create()
-
-	<-ctx.Done()
-}
-
-type job struct {
-	id         uuid.UUID
-	ctx        context.Context
-	cancelFunc context.CancelFunc
-	action     PollerFunc
-}
-
-func (j *job) run() <-chan bool {
-	ch := make(chan bool, 1)
-
-	done := make(chan bool, 1)
-
-	go func() {
-		j.action(j.ctx)
-		done <- true
-	}()
-
-	go func() {
-		select {
-		case <-j.ctx.Done():
-			ch <- true
-		case <-done:
-			ch <- true
-		}
-		defer close(ch)
-	}()
-
-	return ch
-}
-
-func createJob(ctx context.Context, action PollerFunc) job {
-	ctx, cancel := context.WithCancel(ctx)
-
-	return job{
-		id:         uuid.New(),
-		cancelFunc: cancel,
-		ctx:        ctx,
-		action:     action,
-	}
-}
-
-func (j *job) cancel() {
-	j.cancelFunc()
-}
-
-type jobTracker struct {
-	jobs   map[uuid.UUID]job
-	action PollerFunc
-	mu     sync.Mutex
-	ctx    context.Context
-	logger log.Logger
-}
 
-func initJobTracker(ctx context.Context, action PollerFunc) *jobTracker {
-	return &jobTracker{
-		ctx:    ctx,
-		action: action,
-		jobs:   make(map[uuid.UUID]job),
-		mu:     sync.Mutex{},
-		logger: log.From(ctx),
-	}
-}
-
-func (jt *jobTracker) create() job {
-	jt.mu.Lock()
-	defer jt.mu.Unlock()
-
-	j := createJob(jt.ctx, jt.action)
-	jt.jobs[j.id] = j
-
-	return j
-}
-
-func (jt *jobTracker) clear() {
-	jt.mu.Lock()
-	defer jt.mu.Unlock()
+			if !canceled {
+				timer.Reset(resetDuration)
+			}
 
-	for id, j := range jt.jobs {
-		j.cancel()
-		delete(jt.jobs, id)
+		}()
 	}
 }
diff --git a/internal/api/websocket_client.go b/internal/api/websocket_client.go
index 8f816891..4bdc3518 100644
--- a/internal/api/websocket_client.go
+++ b/internal/api/websocket_client.go
@@ -57,6 +57,7 @@ var _ OctantClient = (*WebsocketClient)(nil)
 // NewWebsocketClient creates an instance of WebsocketClient.
 func NewWebsocketClient(ctx context.Context, conn *websocket.Conn, manager *WebsocketClientManager, dashConfig config.Dash, actionDispatcher ActionDispatcher, id uuid.UUID) *WebsocketClient {
 	logger := dashConfig.Logger().With("component", "websocket-client", "client-id", id.String())
+	ctx = log.WithLoggerContext(ctx, logger)
 
 	ctx, cancel := context.WithCancel(ctx)
 
@@ -84,6 +85,8 @@ func NewWebsocketClient(ctx context.Context, conn *websocket.Conn, manager *Webs
 		client.RegisterHandler(handler)
 	}
 
+	logger.Debugf("created websocket client")
+
 	return client
 }
 
@@ -95,6 +98,7 @@ func (c *WebsocketClient) ID() string {
 func (c *WebsocketClient) readPump() {
 	defer func() {
 		c.isOpen = false
+		c.logger.Debugf("closing read pump")
 	}()
 
 	go func() {
@@ -180,6 +184,7 @@ func (c *WebsocketClient) writePump() {
 	ticker := time.NewTicker(pingPeriod)
 	defer func() {
 		ticker.Stop()
+		c.logger.Debugf("closing write pump")
 	}()
 
 	done := false
diff --git a/web/src/app/modules/shared/services/websocket/websocket.service.ts b/web/src/app/modules/shared/services/websocket/websocket.service.ts
index ca40aad7..f7976092 100644
--- a/web/src/app/modules/shared/services/websocket/websocket.service.ts
+++ b/web/src/app/modules/shared/services/websocket/websocket.service.ts
@@ -11,7 +11,7 @@ import {
   NotifierSignalType,
 } from '../../notifier/notifier.service';
 import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
-import { delay, retryWhen, tap } from 'rxjs/operators';
+import { delay, retryWhen, take, tap } from 'rxjs/operators';
 
 interface WebsocketPayload {
   type: string;
@@ -38,13 +38,13 @@ interface Alert {
   providedIn: 'root',
 })
 export class WebsocketService implements BackendService {
-  ws: WebSocket;
   handlers: { [key: string]: ({}) => void } = {};
   reconnected = new Subject<Event>();
 
   private notifierSession: NotifierSession;
   private subject: WebSocketSubject<unknown>;
   private connectSignalID = new BehaviorSubject<string>('');
+  private opened = false;
 
   constructor(notifierService: NotifierService) {
     this.notifierSession = notifierService.createSession();
@@ -76,6 +76,9 @@ export class WebsocketService implements BackendService {
   }
 
   open() {
+    if (this.opened) {
+      return;
+    }
     this.createWebSocket()
       .pipe(
         retryWhen(errors =>
@@ -93,22 +96,28 @@ export class WebsocketService implements BackendService {
       )
       .subscribe(
         data => {
-          this.connectSignalID
-            .subscribe(id => {
-              if (id !== '') {
-                this.notifierSession.removeAllSignals();
-                this.connectSignalID.next('');
-              }
-            })
-            .unsubscribe();
+          this.connectSignalID.pipe(take(1)).subscribe(id => {
+            if (id !== '') {
+              this.notifierSession.removeAllSignals();
+              this.connectSignalID.next('');
+            }
+          });
 
           this.parseWebsocketMessage(data);
         },
-        err => console.error(err)
+        err => console.error(err),
+        () => {
+          console.log('web socket is closing');
+        }
       );
+
+    this.opened = true;
   }
 
   close() {
+    if (!this.opened) {
+      this.opened = false;
+    }
     this.subject.unsubscribe();
   }
 
diff --git a/web/src/app/modules/sugarloaf/components/smart/container/container.component.ts b/web/src/app/modules/sugarloaf/components/smart/container/container.component.ts
index c2f118f3..ebb078e8 100644
--- a/web/src/app/modules/sugarloaf/components/smart/container/container.component.ts
+++ b/web/src/app/modules/sugarloaf/components/smart/container/container.component.ts
@@ -55,11 +55,5 @@ export class ContainerComponent implements OnInit, OnDestroy {
     this.websocketService.open();
   }
 
-  closeSocket() {
-    this.websocketService.close();
-  }
-
-  ngOnDestroy(): void {
-    this.closeSocket();
-  }
+  ngOnDestroy(): void {}
 }


Test patch:
diff --git a/internal/api/content_manager_test.go b/internal/api/content_manager_test.go
index 81e029e1..10b90d28 100644
--- a/internal/api/content_manager_test.go
+++ b/internal/api/content_manager_test.go
@@ -46,9 +46,7 @@ func TestContentManager_GenerateContent(t *testing.T) {
 	moduleManager := moduleFake.NewMockManagerInterface(controller)
 	state := octantFake.NewMockState(controller)
 
-	state.EXPECT().GetContentPath().Return("/path")
-	state.EXPECT().GetNamespace().Return("default")
-	state.EXPECT().GetQueryParams().Return(params)
+	state.EXPECT().GetContentPath().Return("/path").AnyTimes()
 	state.EXPECT().OnContentPathUpdate(gomock.Any()).DoAndReturn(func(fn octant.ContentPathUpdateFunc) octant.UpdateCancelFunc {
 		fn("foo")
 		return func() {}
@@ -65,8 +63,8 @@ func TestContentManager_GenerateContent(t *testing.T) {
 
 	poller := api.NewSingleRunPoller()
 
-	contentGenerator := func(ctx context.Context, state octant.State) (component.ContentResponse, bool, error) {
-		return contentResponse, false, nil
+	contentGenerator := func(ctx context.Context, state octant.State) (api.Content, bool, error) {
+		return api.Content{Response: contentResponse}, false, nil
 	}
 	manager := api.NewContentManager(moduleManager, logger,
 		api.WithContentGenerator(contentGenerator),