move to task manager module to src/server

src/server/config/schema.js

@@ -238,9 +238,26 @@ export default () => Joi.object({
     }),
     profile: Joi.boolean().default(false)
   }).default(),
+
   status: Joi.object({
     allowAnonymous: Joi.boolean().default(false)
   }).default(),
+
+  taskManager: Joi.object({
+    max_attempts: Joi.number()
+      .description('The maximum number of times a task will be attempted before being abandoned as failed')
+      .default(3),
+    poll_interval: Joi.number()
+      .description('How often, in milliseconds, the task manager will look for more work.')
+      .default(3000),
+    index: Joi.string()
+      .description('The name of the index used to store task information.')
+      .default('.kibana_task_manager'),
+    num_workers: Joi.number()
+      .description('The maximum number of tasks that this Kibana instance will run simultaneously.')
+      .default(10),
+  }).default(),
+
   map: Joi.object({
     includeElasticMapsService: Joi.boolean().default(true),
     tilemap: tilemapSchema,

src/server/kbn_server.js

@@ -36,6 +36,7 @@ import * as Plugins from './plugins';
 import { indexPatternsMixin } from './index_patterns';
 import { savedObjectsMixin } from './saved_objects';
 import { sampleDataMixin } from './sample_data';
+import { taskManagerMixin } from './task_manager';
 import { kibanaIndexMappingsMixin } from './mappings';
 import { urlShorteningMixin } from './url_shortening';
 import { serverExtensionsMixin } from './server_extensions';
@@ -96,6 +97,9 @@ export default class KbnServer {
       // setup routes for installing/uninstalling sample data sets
       sampleDataMixin,
 
+      // task manager service
+      taskManagerMixin,
+
       // setup routes for short urls
       urlShorteningMixin,
 

src/server/task_manager/index.ts

@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+export { taskManagerMixin } from './task_manager';

src/server/task_manager/task_manager.ts

@@ -0,0 +1,93 @@
+/*
+ * 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 Joi from 'joi';
+import {
+  Logger,
+  TaskDefinition,
+  TaskDictionary,
+  TaskManager,
+  TaskPool,
+  TaskStore,
+  validateTaskDefinition,
+} from './task_pool';
+import { SanitizedTaskDefinition } from './task_pool/task';
+
+export async function taskManagerMixin(kbnServer: any, server: any, config: any) {
+  const logger = new Logger((...args) => server.log(...args));
+  const numWorkers = config.get('taskManager.num_workers');
+
+  kbnServer.afterPluginsInit(async () => {
+    const callCluster = server.plugins.elasticsearch.getCluster('admin').callWithInternalUser;
+    const store = new TaskStore({
+      index: config.get('taskManager.index'),
+      callCluster,
+      maxAttempts: config.get('taskManager.max_attempts'),
+    });
+
+    logger.debug('Initializing the task manager index');
+    await store.init();
+
+    const definitions = extractTaskDefinitions(numWorkers, kbnServer.uiExports.taskDefinitions);
+
+    const pool = new TaskPool({
+      logger,
+      callCluster,
+      numWorkers,
+      store,
+      definitions,
+      pollInterval: config.get('taskManager.poll_interval'),
+      kbnServer,
+    });
+
+    pool.start();
+
+    server.decorate(
+      'server',
+      'taskManager',
+      new TaskManager({
+        store,
+        pool,
+      })
+    );
+  });
+}
+
+// TODO, move this to a file and properly test it, validate the taskDefinition via Joi or something
+function extractTaskDefinitions(
+  numWorkers: number,
+  taskDefinitions: TaskDictionary<TaskDefinition> = {}
+): TaskDictionary<SanitizedTaskDefinition> {
+  return Object.keys(taskDefinitions).reduce(
+    (acc, type) => {
+      const rawDefinition = taskDefinitions[type];
+      rawDefinition.type = type;
+      const definition = Joi.attempt(rawDefinition, validateTaskDefinition) as TaskDefinition;
+      const workersOccupied = Math.min(numWorkers, definition.workersOccupied || 1);
+
+      acc[type] = {
+        ...definition,
+        workersOccupied,
+      };
+
+      return acc;
+    },
+    {} as TaskDictionary<SanitizedTaskDefinition>
+  );
+}

src/core_plugins/task_manager/task_pool/task_runner.ts → src/server/task_manager/task_pool/task_runner.ts

@@ -22,6 +22,7 @@ import { intervalFromNow } from './task_intervals';
 import { TaskStore } from './task_store';
 
 interface Logger {
+  info: (msg: string) => void;
   debug: (msg: string) => void;
   warning: (msg: string) => void;
 }

src/utils/index.js

@@ -20,6 +20,7 @@
 export { BinderBase } from './binder';
 export { BinderFor } from './binder_for';
 export { deepCloneWithBuffers } from './deep_clone_with_buffers';
+export { Cancellable } from './cancellable/cancellable';
 export { fromRoot } from './from_root';
 export { pkg } from './package_json';
 export { unset } from './unset';

x-pack/index.js

@@ -44,6 +44,6 @@ module.exports = function (kibana) {
     indexManagement(kibana),
     consoleExtensions(kibana),
     notifications(kibana),
-    kueryAutocomplete(kibana),
+    kueryAutocomplete(kibana)
   ];
 };