paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 36 Packages Projects Releases Wiki Activity

Drop runId, update TaskStore interface

Browse Source 
Co-authored-by: Fredrik Adelöw <freben@gmail.com>
Co-authored-by: blam<ben@blam.sh>
Co-authored-by: Patrik Oldsberg <poldsberg@gmail.com>
This commit is contained in:
Johan Haals
2021-02-02 16:10:13 +01:00
parent 88c31ab16b
commit 6608f9f5ee
7 changed files with 168 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/scaffolder-backend/migrations/20210120143715_init.js
+2 -15
View File
@@ -31,10 +31,6 @@ exports.up = async function up(knex) {
.text('status')
.notNullable()
.comment('The current status of the task');
table
.integer('run_id')
.nullable()
.comment('The current run ID of the task');
table
.dateTime('created_at')
.defaultTo(knex.fn.now())
@@ -44,11 +40,6 @@ exports.up = async function up(knex) {
.dateTime('last_heartbeat_at')
.nullable()
.comment('The last timestamp when a heartbeat was received');
table
.integer('retry_count')
.notNullable()
.defaultTo(0)
.comment('The number of times that this task has been attempted');
});
await knex.schema.createTable('task_events', table => {
@@ -65,17 +56,13 @@ exports.up = async function up(knex) {
.notNullable()
.onDelete('CASCADE')
.comment('The task that generated the event');
table
.integer('run_id')
.nullable()
.comment('The run ID of the task that this event applies to');
table
.text('body')
.notNullable()
.comment('The JSON encoded body of the event');
table.text('event_type').notNullable().comment('The type of event');
table
.dateTime('created_at')
.timestamp('created_at', { precision: 9 })
.defaultTo(knex.fn.now())
.notNullable()
.comment('The timestamp when this event was generated');
@@ -90,7 +77,7 @@ exports.up = async function up(knex) {
exports.down = async function down(knex) {
if (knex.client.config.client !== 'sqlite3') {
await knex.schema.alterTable('task_events', table => {
table.dropIndex([], 'task_events_task_id_idx');
table.dropIndex([], 'ctask_events_task_id_idx');
});
}
await knex.schema.dropTable('task_events');
plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts
+46 -39
View File
@@ -43,14 +43,11 @@ export type RawDbTaskRow = {
spec: string;
status: Status;
last_heartbeat_at?: string;
retry_count: number;
created_at: string;
run_id?: string;
};
export type RawDbTaskEventRow = {
id: number;
run_id: string;
task_id: string;
body: string;
event_type: TaskEventType;
@@ -80,10 +77,8 @@ export class DatabaseTaskStore implements TaskStore {
id: result.id,
spec,
status: result.status,
lastHeartbeat: result.last_heartbeat_at,
retryCount: result.retry_count,
lastHeartbeatAt: result.last_heartbeat_at,
createdAt: result.created_at,
runId: result.run_id,
};
} catch (error) {
throw new Error(`Failed to parse spec of task '${taskId}', ${error}`);
@@ -96,7 +91,6 @@ export class DatabaseTaskStore implements TaskStore {
id: taskId,
spec: JSON.stringify(spec),
status: 'open',
retry_count: 0,
});
return { taskId };
}
@@ -114,12 +108,10 @@ export class DatabaseTaskStore implements TaskStore {
return undefined;
}
const runId = uuid();
const updateCount = await tx<RawDbTaskRow>('tasks')
.where({ id: task.id, status: 'open' })
.update({
status: 'processing',
run_id: runId,
last_heartbeat_at: this.db.fn.now(),
});
@@ -133,10 +125,8 @@ export class DatabaseTaskStore implements TaskStore {
id: task.id,
spec,
status: 'processing',
lastHeartbeat: task.last_heartbeat_at,
retryCount: task.retry_count,
lastHeartbeatAt: task.last_heartbeat_at,
createdAt: task.created_at,
runId: runId,
};
} catch (error) {
throw new Error(`Failed to parse spec of task '${task.id}', ${error}`);
@@ -144,58 +134,76 @@ export class DatabaseTaskStore implements TaskStore {
});
}
async heartbeat(runId: string): Promise<void> {
async heartbeatTask(taskId: string): Promise<void> {
const updateCount = await this.db<RawDbTaskRow>('tasks')
.where({ run_id: runId, status: 'processing' })
.where({ id: taskId, status: 'processing' })
.update({
last_heartbeat_at: this.db.fn.now(),
});
if (updateCount === 0) {
throw new Error(`No running task with runId ${runId} found`);
throw new Error(`No running task with taskId ${taskId} found`);
}
}
async listStaleTasks(): Promise<{ tasks: DbTaskRow }> {
const rows = await this.db<RawDbTaskRow>('tasks')
async listStaleTasks({
timeoutS,
}: {
timeoutS: number;
}): Promise<{
tasks: { taskId: string }[];
}> {
const rawRows = await this.db<RawDbTaskRow>('tasks')
.where('status', 'processing')
.andWhere(
'last_heartbeat_at',
'<',
this.db.type === 'sqlite'
? this.db.raw("datetime('now', '-2 seconds')")
: this.db.raw("dateadd('second', -2, now())"),
'<=',
this.db.client.config.client === 'sqlite3'
? this.db.raw(`datetime('now', '-${Number(timeoutS)} seconds')`)
: this.db.raw(`dateadd('second', -${Number(timeoutS)}, now())`),
);
const tasks = rawRows.map(row => ({
taskId: row.id,
}));
return { tasks };
}
async setStatus(runId: string, status: Status): Promise<void> {
async completeTask({
taskId,
status,
eventBody,
}: {
taskId: string;
status: Status;
eventBody: JsonObject;
}): Promise<void> {
let oldStatus: string;
if (status === 'failed' || status === 'completed') {
oldStatus = 'processing';
} else {
throw new Error(
`Invalid status update of run '${runId}' to status '${status}'`,
`Invalid status update of run '${taskId}' to status '${status}'`,
);
}
await this.db.transaction(async tx => {
const [task] = await tx<RawDbTaskRow>('tasks')
.where({
run_id: runId,
id: taskId,
})
.limit(1)
.select();
if (!task) {
throw new Error(`No task with runId ${runId} found`);
throw new Error(`No task with taskId ${taskId} found`);
}
if (task.status !== oldStatus) {
throw new ConflictError(
`Refusing to update status of run '${runId}' to status '${status}' ` +
`Refusing to update status of run '${taskId}' to status '${status}' ` +
`as it is currently '${task.status}', expected '${oldStatus}'`,
);
}
const updateCount = await tx<RawDbTaskRow>('tasks')
.where({
run_id: runId,
id: taskId,
status: oldStatus,
})
.update({
@@ -203,28 +211,28 @@ export class DatabaseTaskStore implements TaskStore {
});
if (updateCount !== 1) {
throw new Error(
`Failed to update status to '${status}' for runId ${runId}`,
`Failed to update status to '${status}' for taskId ${taskId}`,
);
}
await tx<RawDbTaskEventRow>('task_events').insert({
task_id: taskId,
event_type: 'completion',
body: JSON.stringify(eventBody),
});
});
}
async emit({
taskId,
runId,
body,
type,
}: TaskStoreEmitOptions): Promise<void> {
async emitLogEvent({ taskId, body }: TaskStoreEmitOptions): Promise<void> {
const serliazedBody = JSON.stringify(body);
await this.db<RawDbTaskEventRow>('task_events').insert({
task_id: taskId,
run_id: runId,
event_type: type,
event_type: 'log',
body: serliazedBody,
});
}
async getEvents({
async listEvents({
taskId,
after,
}: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }> {
@@ -244,8 +252,7 @@ export class DatabaseTaskStore implements TaskStore {
const body = JSON.parse(event.body) as JsonObject;
return {
id: event.id,
runId: event.run_id,
taskId: event.task_id,
taskId,
body,
type: event.event_type,
createdAt: event.created_at,
plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts
+33 -24
View File
@@ -14,13 +14,16 @@
* limitations under the License.
*/
import { SingleConnectionDatabaseManager } from '@backstage/backend-common';
import {
getVoidLogger,
SingleConnectionDatabaseManager,
} from '@backstage/backend-common';
import { ConfigReader } from '@backstage/config';
import { DatabaseTaskStore } from './DatabaseTaskStore';
import { StorageTaskBroker, TaskAgent } from './StorageTaskBroker';
import { TaskStore, TaskSpec, DbTaskEventRow } from './types';
import { TaskSpec, DbTaskEventRow } from './types';
async function createStore(): Promise<TaskStore> {
async function createStore(): Promise<DatabaseTaskStore> {
const manager = SingleConnectionDatabaseManager.fromConfig(
new ConfigReader({
backend: {
@@ -35,20 +38,21 @@ async function createStore(): Promise<TaskStore> {
}
describe('StorageTaskBroker', () => {
let storage: TaskStore;
let storage: DatabaseTaskStore;
beforeAll(async () => {
storage = await createStore();
});
const logger = getVoidLogger();
it('should claim a dispatched work item', async () => {
const broker = new StorageTaskBroker(storage);
const broker = new StorageTaskBroker(storage, logger);
await broker.dispatch({ steps: [] });
await expect(broker.claim()).resolves.toEqual(expect.any(TaskAgent));
});
it('should wait for a dispatched work item', async () => {
const broker = new StorageTaskBroker(storage);
const broker = new StorageTaskBroker(storage, logger);
const promise = broker.claim();
await expect(Promise.race([promise, 'waiting'])).resolves.toBe('waiting');
@@ -58,7 +62,7 @@ describe('StorageTaskBroker', () => {
});
it('should dispatch multiple items and claim them in order', async () => {
const broker = new StorageTaskBroker(storage);
const broker = new StorageTaskBroker(storage, logger);
await broker.dispatch({ steps: [{ id: 'a' }] } as TaskSpec);
await broker.dispatch({ steps: [{ id: 'b' }] } as TaskSpec);
await broker.dispatch({ steps: [{ id: 'c' }] } as TaskSpec);
@@ -75,16 +79,16 @@ describe('StorageTaskBroker', () => {
});
it('should complete a task', async () => {
const broker = new StorageTaskBroker(storage);
const broker = new StorageTaskBroker(storage, logger);
const dispatchResult = await broker.dispatch({ steps: [] });
const task = await broker.claim();
await task.complete('completed');
const taskRow = await storage.get(dispatchResult.taskId);
expect(taskRow.status).toBe('completed');
});
}, 10000);
it('should fail a task', async () => {
const broker = new StorageTaskBroker(storage);
const broker = new StorageTaskBroker(storage, logger);
const dispatchResult = await broker.dispatch({ steps: [] });
const task = await broker.claim();
await task.complete('failed');
@@ -93,8 +97,8 @@ describe('StorageTaskBroker', () => {
});
it('multiple brokers should be able to observe a single task', async () => {
const broker1 = new StorageTaskBroker(storage);
const broker2 = new StorageTaskBroker(storage);
const broker1 = new StorageTaskBroker(storage, logger);
const broker2 = new StorageTaskBroker(storage, logger);
const { taskId } = await broker1.dispatch({ steps: [] });
@@ -115,7 +119,7 @@ describe('StorageTaskBroker', () => {
await task.complete('completed');
const logs = await logPromise;
expect(logs.map(l => l.body.message)).toEqual([
expect(logs.map(l => l.body.message, logger)).toEqual([
'log 1',
'log 2',
'log 3',
@@ -134,7 +138,7 @@ describe('StorageTaskBroker', () => {
});
it('should heartbeat', async () => {
const broker = new StorageTaskBroker(storage);
const broker = new StorageTaskBroker(storage, logger);
const { taskId } = await broker.dispatch({ steps: [] });
const task = await broker.claim();
@@ -142,7 +146,7 @@ describe('StorageTaskBroker', () => {
for (;;) {
const maybeTask = await storage.get(taskId);
if (maybeTask.lastHeartbeat !== initialTask.lastHeartbeat) {
if (maybeTask.lastHeartbeatAt !== initialTask.lastHeartbeatAt) {
break;
}
await new Promise(resolve => setTimeout(resolve, 50));
@@ -151,24 +155,29 @@ describe('StorageTaskBroker', () => {
expect.assertions(0);
});
it('should be cancelled if heartbeat stops', async () => {
const broker = new StorageTaskBroker(storage);
it('should be update the status to failed if heartbeat fails', async () => {
const broker = new StorageTaskBroker(storage, logger);
const { taskId } = await broker.dispatch({ steps: [] });
console.log('DEBUG: taskId =', taskId);
const task = await broker.claim();
clearInterval((task as any).heartbeatInterval);
setTimeout(() => {
storage.listStaleTasks();
}, 4000);
jest
.spyOn((task as any).storage, 'heartbeatTask')
.mockRejectedValue(new Error('nah m8'));
const intervalId = setInterval(() => {
broker.vacuumTasks({ timeoutS: 2 }).catch(fail);
}, 500);
for (;;) {
const maybeTask = await storage.get(taskId);
if (maybeTask.status === 'cancelled') {
if (maybeTask.status === 'failed') {
break;
}
await new Promise(resolve => setTimeout(resolve, 50));
}
expect.assertions(0);
clearInterval(intervalId);
expect(task.done).toBe(true);
});
});
plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts
+58 -28
View File
@@ -13,7 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import { Logger } from 'winston';
import {
CompletedTaskState,
Task,
@@ -25,11 +25,13 @@ import {
} from './types';
export class TaskAgent implements Task {
private heartbeatInterval?: ReturnType<typeof setInterval>;
private isDone = false;
static create(state: TaskState, storage: TaskStore) {
const agent = new TaskAgent(state, storage);
agent.start();
private heartbeatTimeoutId?: ReturnType<typeof setInterval>;
static create(state: TaskState, storage: TaskStore, logger: Logger) {
const agent = new TaskAgent(state, storage, logger);
agent.startTimeout();
return agent;
}
@@ -37,6 +39,7 @@ export class TaskAgent implements Task {
private constructor(
private readonly state: TaskState,
private readonly storage: TaskStore,
private readonly logger: Logger,
) {}
get spec() {
@@ -44,40 +47,45 @@ export class TaskAgent implements Task {
}
async getWorkspaceName() {
return `${this.state.taskId}_${this.state.runId}`;
return this.state.taskId;
}
get done() {
return this.isDone;
}
async emitLog(message: string): Promise<void> {
await this.storage.emit({
await this.storage.emitLogEvent({
taskId: this.state.taskId,
runId: this.state.runId,
body: { message },
type: 'log',
});
}
async complete(result: CompletedTaskState): Promise<void> {
await this.storage.setStatus(
this.state.runId,
result === 'failed' ? 'failed' : 'completed',
);
this.storage.emit({
await this.storage.completeTask({
taskId: this.state.taskId,
runId: this.state.runId,
body: { message: `Run completed with status: ${result}` },
type: 'completion',
status: result === 'failed' ? 'failed' : 'completed',
eventBody: { message: `Run completed with status: ${result}` },
});
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
this.isDone = true;
if (this.heartbeatTimeoutId) {
clearTimeout(this.heartbeatTimeoutId);
}
}
private start() {
this.heartbeatInterval = setInterval(() => {
if (!this.state.runId) {
throw new Error('no run id provided');
private startTimeout() {
this.heartbeatTimeoutId = setTimeout(async () => {
try {
await this.storage.heartbeatTask(this.state.taskId);
this.startTimeout();
} catch (error) {
this.isDone = true;
this.logger.error(
`Heartbeat for task ${this.state.taskId} failed`,
error,
);
}
this.storage.heartbeat(this.state.runId);
}, 1000);
}
}
@@ -85,7 +93,6 @@ export class TaskAgent implements Task {
interface TaskState {
spec: TaskSpec;
taskId: string;
runId: string;
}
function defer() {
@@ -97,7 +104,10 @@ function defer() {
}
export class StorageTaskBroker implements TaskBroker {
constructor(private readonly storage: TaskStore) {}
constructor(
private readonly storage: TaskStore,
private readonly logger: Logger,
) {}
private deferredDispatch = defer();
async claim(): Promise<Task> {
@@ -106,11 +116,11 @@ export class StorageTaskBroker implements TaskBroker {
if (pendingTask) {
return TaskAgent.create(
{
runId: pendingTask.runId!,
taskId: pendingTask.id,
spec: pendingTask.spec,
},
this.storage,
this.logger,
);
}
@@ -146,7 +156,7 @@ export class StorageTaskBroker implements TaskBroker {
(async () => {
let after = options.after;
while (!cancelled) {
const result = await this.storage.getEvents({ taskId, after: after });
const result = await this.storage.listEvents({ taskId, after: after });
const { events } = result;
if (events.length) {
after = events[events.length - 1].id;
@@ -164,6 +174,26 @@ export class StorageTaskBroker implements TaskBroker {
return unsubscribe;
}
async vacuumTasks(timeoutS: { timeoutS: number }): Promise<void> {
const { tasks } = await this.storage.listStaleTasks(timeoutS);
await Promise.all(
tasks.map(async task => {
try {
await this.storage.completeTask({
taskId: task.taskId,
status: 'failed',
eventBody: {
message:
'The task was cancelled because the task worker lost connection to the task broker',
},
});
} catch (error) {
this.logger.warn(`Failed to cancel task '${task.taskId}', ${error}`);
}
}),
);
}
private waitForDispatch() {
return this.deferredDispatch.promise;
}
plugins/scaffolder-backend/src/scaffolder/tasks/TaskWorker.ts
-1
View File
@@ -73,7 +73,6 @@ export class TaskWorker {
// Give us some time to curl observe
task.emitLog('Task claimed, waiting ...');
await new Promise(resolve => setTimeout(resolve, 5000));
console.log('DEBUG: task.spec =', JSON.stringify(task.spec, null, 2));
task.emitLog(`Starting up work with ${task.spec.steps.length} steps`);
const outputs: { [name: string]: JsonValue } = {};
plugins/scaffolder-backend/src/scaffolder/tasks/types.ts
+27 -12
View File
@@ -29,16 +29,13 @@ export type DbTaskRow = {
id: string;
spec: TaskSpec;
status: Status;
lastHeartbeat?: string;
retryCount: number;
createdAt: string;
runId?: string;
lastHeartbeatAt?: string;
};
export type TaskEventType = 'completion' | 'log';
export type DbTaskEventRow = {
id: number;
runId: string;
taskId: string;
body: JsonObject;
type: TaskEventType;
@@ -60,6 +57,7 @@ export type DispatchResult = {
export interface Task {
spec: TaskSpec;
done: boolean;
emitLog(message: string): Promise<void>;
complete(result: CompletedTaskState): Promise<void>;
getWorkspaceName(): Promise<string>;
@@ -68,13 +66,22 @@ export interface Task {
export interface TaskBroker {
claim(): Promise<Task>;
dispatch(spec: TaskSpec): Promise<DispatchResult>;
vacuumTasks(timeoutS: { timeoutS: number }): Promise<void>;
observe(
options: {
taskId: string;
after: number | undefined;
},
callback: (
error: Error | undefined,
result: { events: DbTaskEventRow[] },
) => void,
): () => void;
}
export type TaskStoreEmitOptions = {
taskId: string;
runId: string;
body: JsonObject;
type: TaskEventType;
};
export type TaskStoreGetEventsOptions = {
@@ -82,14 +89,22 @@ export type TaskStoreGetEventsOptions = {
after?: number | undefined;
};
export interface TaskStore {
get(taskId: string): Promise<DbTaskRow>;
createTask(task: TaskSpec): Promise<{ taskId: string }>;
claimTask(): Promise<DbTaskRow | undefined>;
heartbeat(runId: string): Promise<void>;
listStaleTasks(): Promise<{ tasks: DbTaskRow }>;
setStatus(runId: string, status: Status): Promise<void>;
emit({ taskId, runId, body, type }: TaskStoreEmitOptions): Promise<void>;
getEvents({
completeTask(options: {
taskId: string;
status: Status;
eventBody: JsonObject;
}): Promise<void>;
heartbeatTask(taskId: string): Promise<void>;
listStaleTasks(options: {
timeoutS: number;
}): Promise<{
tasks: { taskId: string }[];
}>;
emitLogEvent({ taskId, body }: TaskStoreEmitOptions): Promise<void>;
listEvents({
taskId,
after,
}: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }>;
plugins/scaffolder-backend/src/service/router.ts
+2 -2
View File
@@ -82,7 +82,7 @@ export async function createRouter(
const databaseTaskStore = await DatabaseTaskStore.create(
await database.getClient(),
);
const taskBroker = new StorageTaskBroker(databaseTaskStore);
const taskBroker = new StorageTaskBroker(databaseTaskStore, logger);
const actionRegistry = new TemplateActionRegistry();
const worker = new TaskWorker({
logger,
@@ -124,7 +124,7 @@ export async function createRouter(
error: job.error,
});
})
// curl -X POST -d '{"templateName":"springboot-template","values": {"storePath":"https://github.com/jhaals/foo", "component_id":"woop", "description": "apa", "owner": "me" }}' -H 'Content-Type: application/json' localhost:7000/api/scaffolder/v2/tasks
.post('/v2/tasks', async (req, res) => {
const templateName: string = req.body.templateName;
const values: TemplaterValues = {