Time travel control (#2524)

* Hacky working prototype.

* Operate with time.Duration offset instead of fixed timestamp.

* Polished the backend code.

* Made a nicer UI component.

* Small refactorings of the websockets code.

* Fixed the backend tests.

* Better websocketing and smoother transitions

* Small styling refactoring.

* Detecting empty topologies.

* Improved error messaging.

* Addressed some of David's comments.

* Moved nodesDeltaBuffer to a global state to fix the paused status rendering bug.

* Small styling changes

* Changed the websocket global state variables a bit.

* Polishing & refactoring.

* More polishing.

* Final refactoring.

* Addressed a couple of bugs.

* Hidden the timeline control behind Cloud context and a feature flag.

* Addressed most of @davkal's comments.

* Added mixpanel tracking.

client/app/scripts/utils/__tests__/web-api-utils-test.js

@@ -1,6 +1,6 @@
 
 import {OrderedMap as makeOrderedMap} from 'immutable';
-import { buildOptionsQuery, basePath, getApiPath, getWebsocketUrl } from '../web-api-utils';
+import { buildUrlQuery, basePath, getApiPath, getWebsocketUrl } from '../web-api-utils';
 
 describe('WebApiUtils', () => {
   describe('basePath', () => {
@@ -21,13 +21,13 @@ describe('WebApiUtils', () => {
     });
   });
 
-  describe('buildOptionsQuery', () => {
+  describe('buildUrlQuery', () => {
     it('should handle empty options', () => {
-      expect(buildOptionsQuery(makeOrderedMap({}))).toBe('');
+      expect(buildUrlQuery(makeOrderedMap({}))).toBe('');
     });
 
     it('should combine multiple options', () => {
-      expect(buildOptionsQuery(makeOrderedMap([
+      expect(buildUrlQuery(makeOrderedMap([
         ['foo', 2],
         ['bar', 4]
       ]))).toBe('foo=2&bar=4');

client/app/scripts/utils/nodes-delta-utils.js

@@ -0,0 +1,60 @@
+import debug from 'debug';
+import { union, size, map, find, reject, each } from 'lodash';
+
+const log = debug('scope:nodes-delta-utils');
+
+
+// TODO: It would be nice to have a unit test for this function.
+export function consolidateNodesDeltas(first, second) {
+  let toAdd = union(first.add, second.add);
+  let toUpdate = union(first.update, second.update);
+  let toRemove = union(first.remove, second.remove);
+  log('Consolidating delta buffer',
+    'add', size(toAdd),
+    'update', size(toUpdate),
+    'remove', size(toRemove));
+
+  // check if an added node in first was updated in second -> add second update
+  toAdd = map(toAdd, (node) => {
+    const updateNode = find(second.update, {id: node.id});
+    if (updateNode) {
+      toUpdate = reject(toUpdate, {id: node.id});
+      return updateNode;
+    }
+    return node;
+  });
+
+  // check if an updated node in first was updated in second -> updated second update
+  // no action needed, successive updates are fine
+
+  // check if an added node in first was removed in second -> dont add, dont remove
+  each(first.add, (node) => {
+    const removedNode = find(second.remove, {id: node.id});
+    if (removedNode) {
+      toAdd = reject(toAdd, {id: node.id});
+      toRemove = reject(toRemove, {id: node.id});
+    }
+  });
+
+  // check if an updated node in first was removed in second -> remove
+  each(first.update, (node) => {
+    const removedNode = find(second.remove, {id: node.id});
+    if (removedNode) {
+      toUpdate = reject(toUpdate, {id: node.id});
+    }
+  });
+
+  // check if an removed node in first was added in second -> update
+  // remove -> add is fine for the store
+
+  log('Consolidated delta buffer',
+    'add', size(toAdd),
+    'update', size(toUpdate),
+    'remove', size(toRemove));
+
+  return {
+    add: toAdd.length > 0 ? toAdd : null,
+    update: toUpdate.length > 0 ? toUpdate : null,
+    remove: toRemove.length > 0 ? toRemove : null
+  };
+}

client/app/scripts/utils/topology-utils.js

@@ -1,6 +1,7 @@
 import { endsWith } from 'lodash';
 import { Set as makeSet, List as makeList } from 'immutable';
 
+import { isWebsocketQueryingCurrentSelector } from '../selectors/time-travel';
 import { isResourceViewModeSelector } from '../selectors/topology';
 import { pinnedMetricSelector } from '../selectors/node-metric';
 
@@ -133,15 +134,23 @@ export function getCurrentTopologyOptions(state) {
   return state.getIn(['currentTopology', 'options']);
 }
 
-export function isTopologyEmpty(state) {
-  // Consider a topology in the resource view empty if it has no pinned metric.
-  const resourceViewEmpty = isResourceViewModeSelector(state) && !pinnedMetricSelector(state);
-  // Otherwise (in graph and table view), we only look at the node count.
+export function isTopologyNodeCountZero(state) {
   const nodeCount = state.getIn(['currentTopology', 'stats', 'node_count'], 0);
-  const nodesEmpty = nodeCount === 0 && state.get('nodes').size === 0;
-  return resourceViewEmpty || nodesEmpty;
+  return nodeCount === 0 && isWebsocketQueryingCurrentSelector(state);
 }
 
+export function isNodesDisplayEmpty(state) {
+  // Consider a topology in the resource view empty if it has no pinned metric.
+  if (isResourceViewModeSelector(state)) {
+    return !pinnedMetricSelector(state);
+  }
+  // Otherwise (in graph and table view), we only look at the nodes content.
+  return state.get('nodes').isEmpty();
+}
+
+export function isTopologyEmpty(state) {
+  return isTopologyNodeCountZero(state) || isNodesDisplayEmpty(state);
+}
 
 export function getAdjacentNodes(state, originNodeId) {
   let adjacentNodes = makeSet();

client/app/scripts/utils/update-buffer-utils.js

@@ -1,114 +0,0 @@
-import debug from 'debug';
-import Immutable from 'immutable';
-import { union, size, map, find, reject, each } from 'lodash';
-
-import { receiveNodesDelta } from '../actions/app-actions';
-
-const log = debug('scope:update-buffer-utils');
-const makeList = Immutable.List;
-const feedInterval = 1000;
-const bufferLength = 100;
-
-let deltaBuffer = makeList();
-let updateTimer = null;
-
-function isPaused(getState) {
-  return getState().get('updatePausedAt') !== null;
-}
-
-export function resetUpdateBuffer() {
-  clearTimeout(updateTimer);
-  deltaBuffer = deltaBuffer.clear();
-}
-
-function maybeUpdate(getState) {
-  if (isPaused(getState)) {
-    clearTimeout(updateTimer);
-    resetUpdateBuffer();
-  } else {
-    if (deltaBuffer.size > 0) {
-      const delta = deltaBuffer.first();
-      deltaBuffer = deltaBuffer.shift();
-      receiveNodesDelta(delta);
-    }
-    if (deltaBuffer.size > 0) {
-      updateTimer = setTimeout(() => maybeUpdate(getState), feedInterval);
-    }
-  }
-}
-
-// consolidate first buffer entry with second
-function consolidateBuffer() {
-  const first = deltaBuffer.first();
-  deltaBuffer = deltaBuffer.shift();
-  const second = deltaBuffer.first();
-  let toAdd = union(first.add, second.add);
-  let toUpdate = union(first.update, second.update);
-  let toRemove = union(first.remove, second.remove);
-  log('Consolidating delta buffer', 'add', size(toAdd), 'update',
-    size(toUpdate), 'remove', size(toRemove));
-
-  // check if an added node in first was updated in second -> add second update
-  toAdd = map(toAdd, (node) => {
-    const updateNode = find(second.update, {id: node.id});
-    if (updateNode) {
-      toUpdate = reject(toUpdate, {id: node.id});
-      return updateNode;
-    }
-    return node;
-  });
-
-  // check if an updated node in first was updated in second -> updated second update
-  // no action needed, successive updates are fine
-
-  // check if an added node in first was removed in second -> dont add, dont remove
-  each(first.add, (node) => {
-    const removedNode = find(second.remove, {id: node.id});
-    if (removedNode) {
-      toAdd = reject(toAdd, {id: node.id});
-      toRemove = reject(toRemove, {id: node.id});
-    }
-  });
-
-  // check if an updated node in first was removed in second -> remove
-  each(first.update, (node) => {
-    const removedNode = find(second.remove, {id: node.id});
-    if (removedNode) {
-      toUpdate = reject(toUpdate, {id: node.id});
-    }
-  });
-
-  // check if an removed node in first was added in second ->  update
-  // remove -> add is fine for the store
-
-  // update buffer
-  log('Consolidated delta buffer', 'add', size(toAdd), 'update',
-    size(toUpdate), 'remove', size(toRemove));
-  deltaBuffer.set(0, {
-    add: toAdd.length > 0 ? toAdd : null,
-    update: toUpdate.length > 0 ? toUpdate : null,
-    remove: toRemove.length > 0 ? toRemove : null
-  });
-}
-
-export function bufferDeltaUpdate(delta) {
-  if (delta.add === null && delta.update === null && delta.remove === null) {
-    log('Discarding empty nodes delta');
-    return;
-  }
-
-  if (deltaBuffer.size >= bufferLength) {
-    consolidateBuffer();
-  }
-
-  deltaBuffer = deltaBuffer.push(delta);
-  log('Buffering node delta, new size', deltaBuffer.size);
-}
-
-export function getUpdateBufferSize() {
-  return deltaBuffer.size;
-}
-
-export function resumeUpdate(getState) {
-  maybeUpdate(getState);
-}

client/app/scripts/utils/web-api-utils.js

@@ -1,7 +1,8 @@
 import debug from 'debug';
+import moment from 'moment';
 import reqwest from 'reqwest';
-import defaults from 'lodash/defaults';
-import { Map as makeMap, List } from 'immutable';
+import { defaults } from 'lodash';
+import { fromJS, Map as makeMap, List } from 'immutable';
 
 import { blurSearch, clearControlError, closeWebsocket, openWebsocket, receiveError,
   receiveApiDetails, receiveNodesDelta, receiveNodeDetails, receiveControlError,
@@ -9,8 +10,11 @@ import { blurSearch, clearControlError, closeWebsocket, openWebsocket, receiveEr
   receiveControlSuccess, receiveTopologies, receiveNotFound,
   receiveNodesForTopology } from '../actions/app-actions';
 
+import { getCurrentTopologyUrl } from '../utils/topology-utils';
 import { layersTopologyIdsSelector } from '../selectors/resource-view/layout';
-import { API_INTERVAL, TOPOLOGY_INTERVAL } from '../constants/timer';
+import { activeTopologyOptionsSelector } from '../selectors/topology';
+import { isWebsocketQueryingCurrentSelector } from '../selectors/time-travel';
+import { API_REFRESH_INTERVAL, TOPOLOGY_REFRESH_INTERVAL } from '../constants/timer';
 
 const log = debug('scope:web-api-utils');
 
@@ -34,25 +38,25 @@ const csrfToken = (() => {
 
 let socket;
 let reconnectTimer = 0;
-let currentUrl = null;
-let currentOptions = null;
 let topologyTimer = 0;
 let apiDetailsTimer = 0;
 let controlErrorTimer = 0;
-let createWebsocketAt = 0;
-let firstMessageOnWebsocketAt = 0;
+let currentUrl = null;
+let createWebsocketAt = null;
+let firstMessageOnWebsocketAt = null;
 let continuePolling = true;
 
-export function buildOptionsQuery(options) {
-  if (options) {
-    return options.map((value, param) => {
-      if (List.isList(value)) {
-        value = value.join(',');
-      }
-      return `${param}=${value}`;
-    }).join('&');
-  }
-  return '';
+export function buildUrlQuery(params) {
+  if (!params) return '';
+
+  // Ignore the entries with values `null` or `undefined`.
+  return params.map((value, param) => {
+    if (value === undefined || value === null) return null;
+    if (List.isList(value)) {
+      value = value.join(',');
+    }
+    return `${param}=${value}`;
+  }).filter(s => s).join('&');
 }
 
 export function basePath(urlPath) {
@@ -93,7 +97,16 @@ export function getWebsocketUrl(host = window.location.host, pathname = window.l
   return `${wsProto}://${host}${process.env.SCOPE_API_PREFIX || ''}${basePath(pathname)}`;
 }
 
-function createWebsocket(topologyUrl, optionsQuery, dispatch) {
+function buildWebsocketUrl(topologyUrl, topologyOptions = makeMap(), queryTimestamp) {
+  const query = buildUrlQuery(fromJS({
+    t: updateFrequency,
+    timestamp: queryTimestamp,
+    ...topologyOptions.toJS(),
+  }));
+  return `${getWebsocketUrl()}${topologyUrl}/ws?${query}`;
+}
+
+function createWebsocket(websocketUrl, dispatch) {
   if (socket) {
     socket.onclose = null;
     socket.onerror = null;
@@ -104,30 +117,31 @@ function createWebsocket(topologyUrl, optionsQuery, dispatch) {
 
   // profiling
   createWebsocketAt = new Date();
-  firstMessageOnWebsocketAt = 0;
+  firstMessageOnWebsocketAt = null;
 
-  socket = new WebSocket(`${getWebsocketUrl()}${topologyUrl}/ws?t=${updateFrequency}&${optionsQuery}`);
+  socket = new WebSocket(websocketUrl);
 
   socket.onopen = () => {
+    log(`Opening websocket to ${websocketUrl}`);
     dispatch(openWebsocket());
   };
 
   socket.onclose = () => {
     clearTimeout(reconnectTimer);
-    log(`Closing websocket to ${topologyUrl}`, socket.readyState);
+    log(`Closing websocket to ${websocketUrl}`, socket.readyState);
     socket = null;
     dispatch(closeWebsocket());
 
     if (continuePolling) {
       reconnectTimer = setTimeout(() => {
-        createWebsocket(topologyUrl, optionsQuery, dispatch);
+        createWebsocket(websocketUrl, dispatch);
       }, reconnectTimerInterval);
     }
   };
 
   socket.onerror = () => {
-    log(`Error in websocket to ${topologyUrl}`);
-    dispatch(receiveError(currentUrl));
+    log(`Error in websocket to ${websocketUrl}`);
+    dispatch(receiveError(websocketUrl));
   };
 
   socket.onmessage = (event) => {
@@ -170,7 +184,7 @@ function getNodesForTopologies(getState, dispatch, topologyIds, topologyOptions
   getState().get('topologyUrlsById')
     .filter((_, topologyId) => topologyIds.contains(topologyId))
     .reduce((sequence, topologyUrl, topologyId) => sequence.then(() => {
-      const optionsQuery = buildOptionsQuery(topologyOptions.get(topologyId));
+      const optionsQuery = buildUrlQuery(topologyOptions.get(topologyId));
       return doRequest({ url: `${getApiPath()}${topologyUrl}?${optionsQuery}` });
     })
     .then(json => dispatch(receiveNodesForTopology(json.nodes, topologyId))),
@@ -200,7 +214,7 @@ export function getTopologies(options, dispatch, initialPoll) {
   // Used to resume polling when navigating between pages in Weave Cloud.
   continuePolling = initialPoll === true ? true : continuePolling;
   clearTimeout(topologyTimer);
-  const optionsQuery = buildOptionsQuery(options);
+  const optionsQuery = buildUrlQuery(options);
   const url = `${getApiPath()}/api/topology?${optionsQuery}`;
   doRequest({
     url,
@@ -209,7 +223,7 @@ export function getTopologies(options, dispatch, initialPoll) {
         dispatch(receiveTopologies(res));
         topologyTimer = setTimeout(() => {
           getTopologies(options, dispatch);
-        }, TOPOLOGY_INTERVAL);
+        }, TOPOLOGY_REFRESH_INTERVAL);
       }
     },
     error: (req) => {
@@ -219,26 +233,32 @@ export function getTopologies(options, dispatch, initialPoll) {
       if (continuePolling) {
         topologyTimer = setTimeout(() => {
           getTopologies(options, dispatch);
-        }, TOPOLOGY_INTERVAL);
+        }, TOPOLOGY_REFRESH_INTERVAL);
       }
     }
   });
 }
 
-// TODO: topologyUrl and options are always used for the current topology so they as arguments
-// can be replaced by the `state` and then retrieved here internally from selectors.
-export function getNodesDelta(topologyUrl, options, dispatch) {
-  const optionsQuery = buildOptionsQuery(options);
+function getWebsocketQueryTimestamp(state) {
+  // The timestamp query parameter will be used only if it's in the past.
+  if (isWebsocketQueryingCurrentSelector(state)) return null;
+
+  const millisecondsInPast = state.get('websocketQueryMillisecondsInPast');
+  return moment().utc().subtract(millisecondsInPast).toISOString();
+}
+
+export function updateWebsocketChannel(state, dispatch) {
+  const topologyUrl = getCurrentTopologyUrl(state);
+  const topologyOptions = activeTopologyOptionsSelector(state);
+  const queryTimestamp = getWebsocketQueryTimestamp(state);
+  const websocketUrl = buildWebsocketUrl(topologyUrl, topologyOptions, queryTimestamp);
   // Only recreate websocket if url changed or if forced (weave cloud instance reload);
-  // Check for truthy options and that options have changed.
-  const isNewOptions = currentOptions && currentOptions !== optionsQuery;
-  const isNewUrl = topologyUrl !== currentUrl || isNewOptions;
+  const isNewUrl = websocketUrl !== currentUrl;
   // `topologyUrl` can be undefined initially, so only create a socket if it is truthy
   // and no socket exists, or if we get a new url.
-  if ((topologyUrl && !socket) || (topologyUrl && isNewUrl)) {
-    createWebsocket(topologyUrl, optionsQuery, dispatch);
-    currentUrl = topologyUrl;
-    currentOptions = optionsQuery;
+  if (topologyUrl && (!socket || isNewUrl)) {
+    createWebsocket(websocketUrl, dispatch);
+    currentUrl = websocketUrl;
   }
 }
 
@@ -250,7 +270,7 @@ export function getNodeDetails(topologyUrlsById, currentTopologyId, options, nod
     let urlComponents = [getApiPath(), topologyUrl, '/', encodeURIComponent(obj.id)];
     if (currentTopologyId === obj.topologyId) {
       // Only forward filters for nodes in the current topology
-      const optionsQuery = buildOptionsQuery(options);
+      const optionsQuery = buildUrlQuery(options);
       urlComponents = urlComponents.concat(['?', optionsQuery]);
     }
     const url = urlComponents.join('');
@@ -288,7 +308,7 @@ export function getApiDetails(dispatch) {
       if (continuePolling) {
         apiDetailsTimer = setTimeout(() => {
           getApiDetails(dispatch);
-        }, API_INTERVAL);
+        }, API_REFRESH_INTERVAL);
       }
     },
     error: (req) => {
@@ -297,7 +317,7 @@ export function getApiDetails(dispatch) {
       if (continuePolling) {
         apiDetailsTimer = setTimeout(() => {
           getApiDetails(dispatch);
-        }, API_INTERVAL / 2);
+        }, API_REFRESH_INTERVAL / 2);
       }
     }
   });
@@ -407,6 +427,6 @@ export function teardownWebsockets() {
     socket.onopen = null;
     socket.close();
     socket = null;
-    currentOptions = null;
+    currentUrl = null;
   }
 }