Optimize Canvas REST calls via batching

package.json

@@ -215,7 +215,6 @@
     "rxjs": "^6.2.1",
     "script-loader": "0.7.2",
     "semver": "^5.5.0",
-    "socket.io": "^2.1.1",
     "stream-stream": "^1.2.6",
     "style-loader": "0.23.1",
     "tar": "2.2.0",

packages/kbn-interpreter/package.json

@@ -12,7 +12,6 @@
     "@kbn/i18n": "1.0.0",
     "lodash": "npm:@elastic/lodash@3.10.1-kibana1",
     "lodash.clone": "^4.5.0",
-    "socket.io-client": "^2.1.1",
     "uuid": "3.0.1"
   },
   "devDependencies": {

packages/kbn-interpreter/src/common/index.js

@@ -20,7 +20,7 @@
 export { FunctionsRegistry } from './lib/functions_registry';
 export { TypesRegistry } from './lib/types_registry';
 export { createError } from './interpreter/create_error';
-export { interpretProvider } from './interpreter/interpret';
+export { interpreterProvider } from './interpreter/interpret';
 export { serializeProvider } from './lib/serialize';
 export { fromExpression, toExpression, safeElementFromExpression } from './lib/ast';
 export { Fn } from './lib/fn';

packages/kbn-interpreter/src/common/interpreter/interpret.js

@@ -25,8 +25,8 @@ import { getByAlias } from '../lib/get_by_alias';
 import { castProvider } from './cast';
 import { createError } from './create_error';
 
-export function interpretProvider(config) {
-  const { functions, onFunctionNotFound, types } = config;
+export function interpreterProvider(config) {
+  const { functions, types } = config;
   const handlers = { ...config.handlers, types };
   const cast = castProvider(types);
 
@@ -54,15 +54,8 @@ export function interpretProvider(config) {
     const { function: fnName, arguments: fnArgs } = link;
     const fnDef = getByAlias(functions, fnName);
 
-    // if the function is not found, pass the expression chain to the not found handler
-    // in this case, it will try to execute the function in another context
     if (!fnDef) {
-      chain.unshift(link);
-      try {
-        return await onFunctionNotFound({ type: 'expression', chain: chain }, context);
-      } catch (e) {
-        return createError(e);
-      }
+      return createError({ message: `Function ${fnName} could not be found.` });
     }
 
     try {

packages/kbn-interpreter/src/public/batched_fetch.js

@@ -0,0 +1,111 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you 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.
+ */
+
+import { FUNCTIONS_URL } from './consts';
+
+/**
+ * Create a function which executes a Canvas function on the
+ * server as part of a larger batch of executions.
+ */
+export function batchedFetch({ kfetch, serialize, ms = 10 }) {
+  // Uniquely identifies each function call in a batch operation
+  // so that the appropriate promise can be resolved / rejected later.
+  let id = 0;
+
+  // A map like { id: { future, request } }, which is used to
+  // track all of the function calls in a batch operation.
+  let batch = {};
+  let timeout;
+
+  const nextId = () => ++id;
+
+  const reset = () => {
+    id = 0;
+    batch = {};
+    timeout = undefined;
+  };
+
+  const runBatch = () => {
+    processBatch(kfetch, batch);
+    reset();
+  };
+
+  return ({ functionName, context, args }) => {
+    if (!timeout) {
+      timeout = setTimeout(runBatch, ms);
+    }
+
+    const id = nextId();
+    const future = createFuture();
+
+    batch[id] = {
+      future,
+      request: { id, functionName, args, context: serialize(context) },
+    };
+
+    return future.promise;
+  };
+}
+
+/**
+ * An externally resolvable / rejectable promise, used to make sure
+ * individual batch responses go to the correct caller.
+ */
+function createFuture() {
+  let resolve;
+  let reject;
+
+  return {
+    resolve(val) { return resolve(val); },
+    reject(val) { return reject(val); },
+    promise: new Promise((res, rej) => {
+      resolve = res;
+      reject = rej;
+    }),
+  };
+}
+
+/**
+ * Runs the specified batch of functions on the server, then resolves
+ * the related promises.
+ */
+async function processBatch(kfetch, batch) {
+  try {
+    const { results } = await kfetch({
+      pathname: FUNCTIONS_URL,
+      method: 'POST',
+      body: JSON.stringify({
+        functions: Object.values(batch).map(({ request }) => request),
+      }),
+    });
+
+    results.forEach(({ id, result }) => {
+      const { future } = batch[id];
+      if (result.statusCode && result.err) {
+        future.reject(result);
+      } else {
+        future.resolve(result);
+      }
+    });
+  } catch (err) {
+    Object.values(batch).forEach(({ future }) => {
+      future.reject(err);
+    });
+  }
+}

src/legacy/core_plugins/interpreter/server/lib/route_expression/thread/babeled.js → packages/kbn-interpreter/src/public/consts.js

@@ -17,5 +17,5 @@
  * under the License.
  */
 
-require('../../../../../../../setup_node_env');
-require('./worker');
+// The server endpoint for retrieiving and running Canvas functions.
+export const FUNCTIONS_URL = '/api/canvas/fns';

packages/kbn-interpreter/src/public/create_handlers.js

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-export function createHandlers(/*socket*/) {
+export function createHandlers() {
   return {
     environment: 'client',
   };

packages/kbn-interpreter/src/public/index.js

@@ -18,6 +18,5 @@
  */
 
 export { loadBrowserRegistries } from './browser_registries';
-export { createSocket } from './socket';
 export { initializeInterpreter } from './interpreter';
 export { RenderFunctionsRegistry } from './render_functions_registry';

packages/kbn-interpreter/src/public/interpreter.js

@@ -17,55 +17,37 @@
  * under the License.
  */
 
-import { socketInterpreterProvider } from '../common/interpreter/socket_interpret';
+import { interpreterProvider } from '../common/interpreter/interpret';
 import { serializeProvider } from '../common/lib/serialize';
 import { createHandlers } from './create_handlers';
-
-export async function initializeInterpreter(socket, typesRegistry, functionsRegistry) {
-  let resolve;
-  const functionList = new Promise(_resolve => (resolve = _resolve));
-
-  const getInitializedFunctions = async () => {
-    return functionList;
-  };
+import { batchedFetch } from './batched_fetch';
+import { FUNCTIONS_URL } from './consts';
+
+export async function initializeInterpreter(kfetch, typesRegistry, functionsRegistry) {
+  const serverFunctionList = await kfetch({ pathname: FUNCTIONS_URL });
+  const types = typesRegistry.toJS();
+  const { serialize } = serializeProvider(types);
+  const batch = batchedFetch({ kfetch, serialize });
+
+  // For every sever-side function, register a client-side
+  // function that matches its definition, but which simply
+  // calls the server-side function endpoint.
+  Object.keys(serverFunctionList).forEach(functionName => {
+    functionsRegistry.register(() => ({
+      ...serverFunctionList[functionName],
+      fn: (context, args) => batch({ functionName, args, context }),
+    }));
+  });
 
   const interpretAst = async (ast, context, handlers) => {
-    // Load plugins before attempting to get functions, otherwise this gets racey
-    const serverFunctionList = await functionList;
-    const interpretFn = await socketInterpreterProvider({
+    const interpretFn = await interpreterProvider({
       types: typesRegistry.toJS(),
-      handlers: { ...handlers, ...createHandlers(socket) },
+      handlers: { ...handlers, ...createHandlers() },
       functions: functionsRegistry.toJS(),
-      referableFunctions: serverFunctionList,
-      socket: socket,
     });
     return interpretFn(ast, context);
   };
 
-  // Listen for interpreter runs
-  socket.on('run', ({ ast, context, id }) => {
-    const types = typesRegistry.toJS();
-    const { serialize, deserialize } = serializeProvider(types);
-    interpretAst(ast, deserialize(context)).then(value => {
-      socket.emit(`resp:${id}`, { value: serialize(value) });
-    });
-  });
-
-  // Create the function list
-  let gotFunctionList = false;
-  socket.once('functionList', (fl) => {
-    gotFunctionList = true;
-    resolve(fl);
-  });
-
-  const interval = setInterval(() => {
-    if (gotFunctionList) {
-      clearInterval(interval);
-      return;
-    }
-    socket.emit('getFunctionList');
-  }, 1000);
-
-  return { getInitializedFunctions, interpretAst };
+  return { interpretAst };
 }