Merge pull request #7 from tsullivan/chrisdavies-alerting/task-manager

Allow hardcoded ID passed in scheduling a task

src/server/task_manager/task.ts

@@ -50,10 +50,9 @@ export interface RunContext {
  */
 export interface RunResult {
  /**
- * Specifies the next run date / time for this task. If unspecified, the system
- * will reschedule the task based on the interval defined by the tasks' definition.
- * If there is no runAt and there is no interval in the task's definition, this
- * is treated as a single-run task, and will not be rescheduled after completion.
+ * Specifies the next run date / time for this task. If unspecified, this is
+ * treated as a single-run task, and will not be rescheduled after
+ * completion.
  */
  runAt?: Date;
 
@@ -158,14 +157,22 @@ export type TaskStatus = 'idle' | 'running';
  * and execute a task.
  */
 export interface TaskInstance {
+ /**
+ * Optional ID that can be passed by the caller. When ID is undefined, ES
+ * will auto-generate a unique id. Otherwise, ID will be used to either
+ * create a new document, or update existing document
+ */
+ id?: string;
+
  /**
  * The task definition type whose run function will execute this instance.
  */
  taskType: string;
 
  /**
- * 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 than this.
+ * 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
+ * than this. Defaults to immediately.
  */
  runAt?: Date;
 
@@ -204,7 +211,8 @@ export interface TaskInstance {
  */
 export interface ConcreteTaskInstance extends TaskInstance {
  /**
- * The id of the Elastic document that stores this instance's data.
+ * The id of the Elastic document that stores this instance's data. This can
+ * be passed by the caller when scheduling the task.
  */
  id: string;
 

src/server/task_manager/task_store.ts

@@ -162,15 +162,16 @@ export class TaskStore {
  throw new Error(`Unsupported task type "${taskInstance.taskType}".`);
  }
 
- const body = rawSource(taskInstance);
- const { task } = body;
+ const { id, ...body } = rawSource(taskInstance);
  const result = await this.callCluster('index', {
+ id,
  body,
  index: this.index,
  type: DOC_TYPE,
  refresh: true,
  });
 
+ const { task } = body;
  return {
  ...taskInstance,
  id: result._id,
@@ -319,8 +320,9 @@ function paginatableSort(sort: any[] = []) {
 }
 
 function rawSource(doc: ConcreteTaskInstance | TaskInstance) {
+ const { id, ...taskFields } = doc;
  const source = {
- ...doc,
+ ...taskFields,
  params: JSON.stringify(doc.params || {}),
  state: JSON.stringify(doc.state || {}),
  attempts: (doc as ConcreteTaskInstance).attempts || 0,
@@ -333,16 +335,19 @@ function rawSource(doc: ConcreteTaskInstance | TaskInstance) {
  delete (source as any).type;
 
  return {
+ id,
  type: 'task',
  task: source,
  };
 }
 
 function taskDocToRaw(doc: ConcreteTaskInstance, index: string): RawTaskDoc {
+ const { type, task } = rawSource(doc);
+
  return {
  _id: doc.id,
  _index: index,
- _source: rawSource(doc),
+ _source: { type, task },
  _type: DOC_TYPE,
  _version: doc.version,
  };