Skip to content

Commit

Permalink
refactor(tasks): disable logs on task definition (#4273)
Browse files Browse the repository at this point in the history
  • Loading branch information
brunozoric authored Sep 19, 2024
1 parent 53fc875 commit 0a32c9c
Show file tree
Hide file tree
Showing 5 changed files with 76 additions and 23 deletions.
6 changes: 5 additions & 1 deletion packages/api-elasticsearch-tasks/__tests__/mocks/store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,5 +13,9 @@ export const createTaskManagerStoreMock = (params?: Params) => {
const context = params?.context || createContextMock();
const task = params?.task || createTaskMock();
const log = params?.log || createTaskLogMock(task);
return new TaskManagerStore(context, task, log);
return new TaskManagerStore({
context,
task,
log
});
};
6 changes: 5 additions & 1 deletion packages/tasks/__tests__/live/store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,11 @@ export const createLiveStore = async <C extends Context = Context>(
) => {
const context = params.context || (await createLiveContext(params));

const store = new TaskManagerStore(context, params.task, params.taskLog);
const store = new TaskManagerStore({
context,
task: params.task,
log: params.taskLog
});

return {
store,
Expand Down
57 changes: 40 additions & 17 deletions packages/tasks/src/runner/TaskControl.ts
Original file line number Diff line number Diff line change
Expand Up @@ -45,12 +45,28 @@ export class TaskControl implements ITaskControl {
});
}
/**
* As this as a run of the task, we need to create a new log entry.
* Let's get the task definition.
*/
const definition = this.context.tasks.getDefinition(task.definitionId);
if (!definition) {
return this.response.error({
error: {
message: `Task "${task.id}" cannot be executed because there is no "${task.definitionId}" definition plugin.`,
code: "TASK_DEFINITION_ERROR",
data: {
definitionId: task.definitionId
}
}
});
}
const disableDatabaseLogs = !!definition.disableDatabaseLogs;

/**
* As this as a run of the task, we need to create a new log entry.
*/
let taskLog: ITaskLog;
try {
taskLog = await this.getTaskLog(task);
taskLog = await this.getTaskLog(task, disableDatabaseLogs);
} catch (error) {
return this.response.error({
error
Expand Down Expand Up @@ -85,8 +101,14 @@ export class TaskControl implements ITaskControl {
}
});
}

const taskResponse = new TaskResponse(this.response);
const store = new TaskManagerStore(this.context, task, taskLog);
const store = new TaskManagerStore({
context: this.context,
task,
log: taskLog,
disableDatabaseLogs
});

const manager = new TaskManager(
this.runner,
Expand All @@ -98,19 +120,6 @@ export class TaskControl implements ITaskControl {

const databaseResponse = new DatabaseResponse(this.response, store);

const definition = this.context.tasks.getDefinition(task.definitionId);
if (!definition) {
return await databaseResponse.error({
error: {
message: `Task "${task.id}" cannot be executed because there is no "${task.definitionId}" definition plugin.`,
code: "TASK_DEFINITION_ERROR",
data: {
definitionId: task.definitionId
}
}
});
}

try {
const result = await manager.run(definition);

Expand Down Expand Up @@ -189,7 +198,21 @@ export class TaskControl implements ITaskControl {
});
}

private async getTaskLog(task: ITask): Promise<ITaskLog> {
private async getTaskLog(task: ITask, disableDatabaseLogs: boolean): Promise<ITaskLog> {
/**
* If logs are disabled, let's return a mocked one.
*/
if (disableDatabaseLogs) {
return {
id: `${task.id}-log`,
createdOn: task.createdOn,
createdBy: task.createdBy,
executionName: task.executionName,
task: task.id,
iteration: task.iterations,
items: []
};
}
let taskLog: ITaskLog | null = null;
/**
* First we are trying to get existing latest log.
Expand Down
26 changes: 22 additions & 4 deletions packages/tasks/src/runner/TaskManagerStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,13 @@ export interface TaskManagerStoreContext {
tasks: Pick<ITasksContextObject, "updateTask" | "updateLog">;
}

export interface ITaskManagerStoreParams {
context: TaskManagerStoreContext;
task: ITask;
log: ITaskLog;
disableDatabaseLogs?: boolean;
}

export class TaskManagerStore<
T extends ITaskDataInput = ITaskDataInput,
O extends ITaskResponseDoneResultOutput = ITaskResponseDoneResultOutput
Expand All @@ -51,14 +58,16 @@ export class TaskManagerStore<
private readonly context: TaskManagerStoreContext;
private task: ITask<T, O>;
private taskLog: ITaskLog;
private readonly disableDatabaseLogs: boolean;

private readonly taskUpdater = new ObjectUpdater<ITask<T, O>>();
private readonly taskLogUpdater = new ObjectUpdater<ITaskLog>();

public constructor(context: TaskManagerStoreContext, task: ITask, log: ITaskLog) {
this.context = context;
this.task = task as ITask<T, O>;
this.taskLog = log;
public constructor(params: ITaskManagerStoreParams) {
this.context = params.context;
this.task = params.task as ITask<T, O>;
this.taskLog = params.log;
this.disableDatabaseLogs = !!params.disableDatabaseLogs;
}

public getStatus(): TaskDataStatus {
Expand Down Expand Up @@ -139,6 +148,9 @@ export class TaskManagerStore<
log: ITaskManagerStoreInfoLog,
options?: ITaskManagerStoreAddLogOptions
): Promise<void> {
if (this.disableDatabaseLogs) {
return;
}
this.taskLogUpdater.update({
items: [
{
Expand All @@ -163,6 +175,9 @@ export class TaskManagerStore<
log: ITaskManagerStoreErrorLog,
options?: ITaskManagerStoreAddLogOptions
): Promise<void> {
if (this.disableDatabaseLogs) {
return;
}
/**
* Let's log the error to the console as well.
*/
Expand Down Expand Up @@ -196,6 +211,9 @@ export class TaskManagerStore<
this.taskUpdater.fetch()
);
}
if (this.disableDatabaseLogs) {
return;
}
if (this.taskLogUpdater.isDirty()) {
this.taskLog = await this.context.tasks.updateLog(
this.taskLog.id,
Expand Down
4 changes: 4 additions & 0 deletions packages/tasks/src/types.ts
Original file line number Diff line number Diff line change
Expand Up @@ -393,6 +393,10 @@ export interface ITaskDefinition<
* Maximum number a step function can call the Lambda.
*/
maxIterations: number;
/**
* Disable storing logs in database for this task.
*/
disableDatabaseLogs?: boolean;
/**
* Task run method.
*/
Expand Down

0 comments on commit 0a32c9c

Please sign in to comment.