[Task Manager] Ensure putTemplate will only create/update the index template

x-pack/plugins/task_manager/constants.ts

@@ -0,0 +1,11 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import xPackage from '../../package.json';
+import { getTemplateVersion } from './lib/get_template_version';
+
+export const TASK_MANAGER_API_VERSION = 1;
+export const TASK_MANAGER_TEMPLATE_VERSION = getTemplateVersion(xPackage.version);

x-pack/plugins/task_manager/lib/get_template_version.test.ts

@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { getTemplateVersion } from './get_template_version';
+
+describe('getTemplateVersion', () => {
+  it('converts a release build version string into an integer', () => {
+    const versionStr1 = '7.1.2';
+    expect(getTemplateVersion(versionStr1)).toBe(7010299);
+
+    const versionStr2 = '10.1.0';
+    expect(getTemplateVersion(versionStr2)).toBe(10010099);
+  });
+
+  it('converts a alpha build version string into an integer', () => {
+    const versionStr1 = '7.0.0-alpha1';
+    expect(getTemplateVersion(versionStr1)).toBe(7000001);
+
+    const versionStr2 = '7.0.0-alpha3';
+    expect(getTemplateVersion(versionStr2)).toBe(7000003);
+  });
+
+  it('converts a beta build version string into an integer', () => {
+    const versionStr1 = '7.0.0-beta4';
+    expect(getTemplateVersion(versionStr1)).toBe(7000029);
+
+    const versionStr5 = '7.0.0-beta8';
+    expect(getTemplateVersion(versionStr5)).toBe(7000033);
+  });
+
+  it('converts a snapshot build version string into an integer', () => {
+    expect(getTemplateVersion('8.0.0-alpha1')).toBe(8000001);
+    expect(getTemplateVersion('8.0.0-alpha1-snapshot')).toBe(8000001);
+  });
+
+  it('not intended to handle any version parts with 3-digits: it will create malformed integer values', () => {
+    expect(getTemplateVersion('60.61.1') === getTemplateVersion('6.6.101')).toBe(true); // both produce 60610199
+    expect(getTemplateVersion('1.32.0') < getTemplateVersion('1.3.223423')).toBe(true);
+  });
+});

x-pack/plugins/task_manager/lib/get_template_version.ts

@@ -0,0 +1,56 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { padLeft } from 'lodash';
+
+/*
+ * The logic for ID is: XXYYZZAA, where XX is major version, YY is minor
+ * version, ZZ is revision, and AA is alpha/beta/rc indicator.
+ *
+ * AA values below 25 are for alpha builder (since 5.0), and above 25 and below
+ * 50 are beta builds, and below 99 are RC builds, with 99 indicating a release
+ * the (internal) format of the id is there so we can easily do after/before
+ * checks on the id
+ *
+ * Note: the conversion method is carried over from Elasticsearch:
+ * https://github.com/elastic/elasticsearch/blob/de962b2/server/src/main/java/org/elasticsearch/Version.java
+ */
+export function getTemplateVersion(versionStr: string): number {
+  // break up the string parts
+  const splitted = versionStr.split('.');
+  const minorStr = splitted[2] || '';
+
+  // pad each part with leading 0 to make 2 characters
+  const padded = splitted.map((v: string) => {
+    const vMatches = v.match(/\d+/);
+    if (vMatches) {
+      return padLeft(vMatches[0], 2, '0');
+    }
+    return '00';
+  });
+  const [majorV, minorV, patchV] = padded;
+
+  // append the alpha/beta/rc indicator
+  let buildV;
+  if (minorStr.match('alpha')) {
+    const matches = minorStr.match(/alpha(?<alpha>\d+)/);
+    if (matches != null && matches.groups != null) {
+      const alphaVerInt = parseInt(matches.groups.alpha, 10); // alpha build indicator
+      buildV = padLeft(`${alphaVerInt}`, 2, '0');
+    }
+  } else if (minorStr.match('beta')) {
+    const matches = minorStr.match(/beta(?<beta>\d+)/);
+    if (matches != null && matches.groups != null) {
+      const alphaVerInt = parseInt(matches.groups.beta, 10) + 25; // beta build indicator
+      buildV = padLeft(`${alphaVerInt}`, 2, '0');
+    }
+  } else {
+    buildV = '99'; // release build indicator
+  }
+
+  const joinedParts = [majorV, minorV, patchV, buildV].join('');
+  return parseInt(joinedParts, 10);
+}

x-pack/plugins/task_manager/task.ts

@@ -158,6 +158,12 @@ export interface TaskInstance {
    */
   taskType: string;
 
+  /**
+   * The date and time that this task was originally scheduled. This is used
+   * for convenience to task run functions, and for troubleshooting.
+   */
+  scheduledAt?: Date;
+
   /**
    * The date and time that this task is scheduled to be run. It is not
    * guaranteed to run at this time, but it is guaranteed not to run earlier
@@ -210,6 +216,12 @@ export interface ConcreteTaskInstance extends TaskInstance {
    */
   version: number;
 
+  /**
+   * The date and time that this task was originally scheduled. This is used
+   * for convenience to task run functions, and for troubleshooting.
+   */
+  scheduledAt: Date;
+
   /**
    * The number of unsuccessful attempts since the last successful run. This
    * will be zeroed out after a successful run.

x-pack/plugins/task_manager/task_manager.test.ts

@@ -75,6 +75,9 @@ describe('TaskManager', () => {
   });
 
   function testOpts() {
+    const callCluster = sinon.stub();
+    callCluster.withArgs('indices.getTemplate').returns(Promise.resolve({ tasky: {} }));
+
     const $test = {
       events: {} as any,
       afterPluginsInit: _.noop,
@@ -100,7 +103,7 @@ describe('TaskManager', () => {
         plugins: {
           elasticsearch: {
             getCluster() {
-              return { callWithInternalUser: _.noop };
+              return { callWithInternalUser: callCluster };
             },
             status: {
               on(eventName: string, callback: () => any) {

x-pack/plugins/task_manager/task_manager.ts

@@ -53,11 +53,15 @@ export class TaskManager {
 
     const logger = new TaskManagerLogger((...args: any[]) => server.log(...args));
 
+    /* Kibana UUID needs to be pulled live (not cached), as it takes a long time
+     * to initialize, and can change after startup */
     const store = new TaskStore({
       callCluster: server.plugins.elasticsearch.getCluster('admin').callWithInternalUser,
       index: config.get('xpack.task_manager.index'),
       maxAttempts: config.get('xpack.task_manager.max_attempts'),
       supportedTypes: Object.keys(this.definitions),
+      logger,
+      getKibanaUuid: () => config.get('server.uuid'),
     });
     const pool = new TaskPool({
       logger,

x-pack/plugins/task_manager/task_poller.test.ts

@@ -14,9 +14,13 @@ let store: TaskStore;
 
 describe('TaskPoller', () => {
   beforeEach(() => {
-    const callCluster = sinon.spy();
+    const callCluster = sinon.stub();
+    callCluster.withArgs('indices.getTemplate').returns(Promise.resolve({ tasky: {} }));
+    const getKibanaUuid = sinon.stub().returns('kibana-123-uuid-test');
     store = new TaskStore({
       callCluster,
+      getKibanaUuid,
+      logger: mockLogger(),
       index: 'tasky',
       maxAttempts: 2,
       supportedTypes: ['a', 'b', 'c'],

x-pack/plugins/task_manager/task_store.test.ts

@@ -6,23 +6,33 @@
 
 import _ from 'lodash';
 import sinon from 'sinon';
+import {
+  TASK_MANAGER_API_VERSION as API_VERSION,
+  TASK_MANAGER_TEMPLATE_VERSION as TEMPLATE_VERSION,
+} from './constants';
 import { TaskInstance, TaskStatus } from './task';
 import { FetchOpts, TaskStore } from './task_store';
+import { mockLogger } from './test_utils';
+
+const getKibanaUuid = sinon.stub().returns('kibana-uuid-123-test');
 
 describe('TaskStore', () => {
   describe('init', () => {
     test('creates the task manager index', async () => {
-      const callCluster = sinon.spy();
+      const callCluster = sinon.stub();
+      callCluster.withArgs('indices.getTemplate').returns(Promise.resolve({ tasky: {} }));
       const store = new TaskStore({
         callCluster,
+        getKibanaUuid,
+        logger: mockLogger(),
         index: 'tasky',
         maxAttempts: 2,
         supportedTypes: ['a', 'b', 'c'],
       });
 
       await store.init();
 
-      sinon.assert.calledOnce(callCluster);
+      sinon.assert.calledTwice(callCluster); // store.init calls twice: once to check for existing template, once to put the template (if needed)
 
       sinon.assert.calledWithMatch(callCluster, 'indices.putTemplate', {
         body: {
@@ -35,27 +45,64 @@ describe('TaskStore', () => {
         name: 'tasky',
       });
     });
+
+    test('logs a warning if newer index template exists', async () => {
+      const callCluster = sinon.stub();
+      callCluster
+        .withArgs('indices.getTemplate')
+        .returns(Promise.resolve({ tasky: { version: Infinity } }));
+
+      const logger = {
+        info: sinon.spy(),
+        debug: sinon.spy(),
+        warning: sinon.spy(),
+        error: sinon.spy(),
+      };
+
+      const store = new TaskStore({
+        callCluster,
+        getKibanaUuid,
+        logger,
+        index: 'tasky',
+        maxAttempts: 2,
+        supportedTypes: ['a', 'b', 'c'],
+      });
+
+      await store.init();
+      const loggingCall = logger.warning.getCall(0);
+      expect(loggingCall.args[0]).toBe(
+        `This Kibana instance defines an older template version (${TEMPLATE_VERSION}) than is currently in Elasticsearch (Infinity). ` +
+          `Because of the potential for non-backwards compatible changes, this Kibana instance will only be able to claim scheduled tasks with ` +
+          `"kibana.apiVersion" <= ${API_VERSION} in the task metadata.`
+      );
+      expect(logger.warning.calledOnce).toBe(true);
+    });
   });
 
   describe('schedule', () => {
     async function testSchedule(task: TaskInstance) {
-      const callCluster = sinon.spy(() =>
+      const callCluster = sinon.stub();
+      callCluster.withArgs('index').returns(
         Promise.resolve({
           _id: 'testid',
           _version: 3344,
         })
       );
+      callCluster.withArgs('indices.getTemplate').returns(Promise.resolve({ tasky: {} }));
       const store = new TaskStore({
         callCluster,
+        getKibanaUuid,
+        logger: mockLogger(),
         index: 'tasky',
         maxAttempts: 2,
         supportedTypes: ['report', 'dernstraight', 'yawn'],
       });
+      await store.init();
       const result = await store.schedule(task);
 
-      sinon.assert.calledTwice(callCluster);
+      sinon.assert.calledThrice(callCluster);
 
-      return { result, callCluster, arg: callCluster.args[1][1] };
+      return { result, callCluster, arg: callCluster.args[2][1] };
     }
 
     test('serializes the params and state', async () => {
@@ -80,7 +127,7 @@ describe('TaskStore', () => {
       });
     });
 
-    test('retiurns a concrete task instance', async () => {
+    test('returns a concrete task instance', async () => {
       const task = {
         params: { hello: 'world' },
         state: { foo: 'bar' },
@@ -119,6 +166,8 @@ describe('TaskStore', () => {
       const callCluster = sinon.spy(async () => ({ hits: { hits } }));
       const store = new TaskStore({
         callCluster,
+        getKibanaUuid,
+        logger: mockLogger(),
         index: 'tasky',
         maxAttempts: 2,
         supportedTypes: ['a', 'b', 'c'],
@@ -283,6 +332,7 @@ describe('TaskStore', () => {
       const callCluster = sinon.spy(async () => ({ hits: { hits } }));
       const store = new TaskStore({
         callCluster,
+        logger: mockLogger(),
         supportedTypes: ['a', 'b', 'c'],
         index: 'tasky',
         maxAttempts: 2,
@@ -304,6 +354,8 @@ describe('TaskStore', () => {
       const callCluster = sinon.spy(async () => ({ hits: { hits: [] } }));
       const store = new TaskStore({
         callCluster,
+        getKibanaUuid,
+        logger: mockLogger(),
         supportedTypes: ['a', 'b', 'c'],
         index: 'tasky',
         maxAttempts: 2,
@@ -339,6 +391,7 @@ describe('TaskStore', () => {
                       { terms: { 'task.taskType': ['foo', 'bar'] } },
                       { range: { 'task.attempts': { lte: maxAttempts } } },
                       { range: { 'task.runAt': { lte: 'now' } } },
+                      { range: { 'kibana.apiVersion': { lte: 1 } } },
                     ],
                   },
                 },
@@ -432,6 +485,7 @@ describe('TaskStore', () => {
       const runAt = new Date();
       const task = {
         runAt,
+        scheduledAt: runAt,
         id: 'task:324242',
         params: { hello: 'world' },
         state: { foo: 'bar' },
@@ -444,6 +498,8 @@ describe('TaskStore', () => {
       const callCluster = sinon.spy(async () => ({ _version: task.version + 1 }));
       const store = new TaskStore({
         callCluster,
+        getKibanaUuid,
+        logger: mockLogger(),
         index: 'tasky',
         maxAttempts: 2,
         supportedTypes: ['a', 'b', 'c'],
@@ -488,6 +544,8 @@ describe('TaskStore', () => {
       );
       const store = new TaskStore({
         callCluster,
+        getKibanaUuid,
+        logger: mockLogger(),
         index: 'myindex',
         maxAttempts: 2,
         supportedTypes: ['a'],