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

Add tests. Extend storage with fetch staletasks method

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 09:27:18 +01:00
parent 5da5e6b224
commit 88c31ab16b
7 changed files with 182 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/scaffolder-backend/package.json
+2 -1
View File
@@ -58,7 +58,8 @@
"p-queue": "^6.3.0",
"uuid": "^8.2.0",
"winston": "^3.2.1",
"yaml": "^1.10.0"
"yaml": "^1.10.0",
"luxon": "^1.25.0"
},
"devDependencies": {
"@backstage/cli": "^0.5.0",
plugins/scaffolder-backend/src/scaffolder/tasks/DatabaseTaskStore.ts
+24 -11
View File
@@ -20,8 +20,7 @@ import {
NotFoundError,
resolvePackagePath,
} from '@backstage/backend-common';
import Knex, { Transaction } from 'knex';
import { Logger } from 'winston';
import Knex from 'knex';
import { v4 as uuid } from 'uuid';
import {
DbTaskEventRow,
@@ -121,6 +120,7 @@ export class DatabaseTaskStore implements TaskStore {
.update({
status: 'processing',
run_id: runId,
last_heartbeat_at: this.db.fn.now(),
});
if (updateCount < 1) {
@@ -155,6 +155,18 @@ export class DatabaseTaskStore implements TaskStore {
}
}
async listStaleTasks(): Promise<{ tasks: DbTaskRow }> {
const rows = 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())"),
);
}
async setStatus(runId: string, status: Status): Promise<void> {
let oldStatus: string;
if (status === 'failed' || status === 'completed') {
@@ -216,16 +228,17 @@ export class DatabaseTaskStore implements TaskStore {
taskId,
after,
}: TaskStoreGetEventsOptions): Promise<{ events: DbTaskEventRow[] }> {
let query = this.db<RawDbTaskEventRow>('task_events').where({
task_id: taskId,
});
if (typeof after === 'number') {
query = query
.where('task_events.id', '>', after)
.orWhere({ event_type: 'completion' });
}
const rawEvents = await this.db<RawDbTaskEventRow>('task_events')
.where({
task_id: taskId,
})
.andWhere(builder => {
if (typeof after === 'number') {
builder.where('id', '>', after).orWhere('event_type', 'completion');
}
})
.select();
const rawEvents = await query.select();
const events = rawEvents.map(event => {
try {
const body = JSON.parse(event.body) as JsonObject;
plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.test.ts
+117 -28
View File
@@ -14,47 +14,54 @@
* limitations under the License.
*/
import { TemplateEntityV1alpha1 } from '@backstage/catalog-model';
import { TemplaterValues } from '../stages';
import { MemoryTaskStore } from './MemoryTaskStore';
import { 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';
async function createStore(): Promise<TaskStore> {
const manager = SingleConnectionDatabaseManager.fromConfig(
new ConfigReader({
backend: {
database: {
client: 'sqlite3',
connection: ':memory:',
},
},
}),
).forPlugin('scaffolder');
return await DatabaseTaskStore.create(await manager.getClient());
}
describe('StorageTaskBroker', () => {
const storage = new MemoryTaskStore();
const broker = new StorageTaskBroker(storage);
let storage: TaskStore;
const taskSpec = {
values: {} as TemplaterValues,
template: {} as TemplateEntityV1alpha1,
};
beforeAll(async () => {
storage = await createStore();
});
it('should claim a dispatched work item', async () => {
await broker.dispatch(taskSpec);
const broker = new StorageTaskBroker(storage);
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 promise = broker.claim();
await expect(Promise.race([promise, 'waiting'])).resolves.toBe('waiting');
await broker.dispatch(taskSpec);
await broker.dispatch({ steps: [] });
await expect(promise).resolves.toEqual(expect.any(TaskAgent));
});
it('should dispatch multiple items and claim them in order', async () => {
await broker.dispatch({
values: { owner: 'a' } as TemplaterValues,
template: {} as TemplateEntityV1alpha1,
});
await broker.dispatch({
values: { owner: 'b' } as TemplaterValues,
template: {} as TemplateEntityV1alpha1,
});
await broker.dispatch({
values: { owner: 'c' } as TemplaterValues,
template: {} as TemplateEntityV1alpha1,
});
const broker = new StorageTaskBroker(storage);
await broker.dispatch({ steps: [{ id: 'a' }] } as TaskSpec);
await broker.dispatch({ steps: [{ id: 'b' }] } as TaskSpec);
await broker.dispatch({ steps: [{ id: 'c' }] } as TaskSpec);
const taskA = await broker.claim();
const taskB = await broker.claim();
@@ -62,13 +69,14 @@ describe('StorageTaskBroker', () => {
await expect(taskA).toEqual(expect.any(TaskAgent));
await expect(taskB).toEqual(expect.any(TaskAgent));
await expect(taskC).toEqual(expect.any(TaskAgent));
await expect(taskA.spec.values.owner).toBe('a');
await expect(taskB.spec.values.owner).toBe('b');
await expect(taskC.spec.values.owner).toBe('c');
await expect(taskA.spec.steps[0].id).toBe('a');
await expect(taskB.spec.steps[0].id).toBe('b');
await expect(taskC.spec.steps[0].id).toBe('c');
});
it('should complete a task', async () => {
const dispatchResult = await broker.dispatch(taskSpec);
const broker = new StorageTaskBroker(storage);
const dispatchResult = await broker.dispatch({ steps: [] });
const task = await broker.claim();
await task.complete('completed');
const taskRow = await storage.get(dispatchResult.taskId);
@@ -76,10 +84,91 @@ describe('StorageTaskBroker', () => {
});
it('should fail a task', async () => {
const dispatchResult = await broker.dispatch(taskSpec);
const broker = new StorageTaskBroker(storage);
const dispatchResult = await broker.dispatch({ steps: [] });
const task = await broker.claim();
await task.complete('failed');
const taskRow = await storage.get(dispatchResult.taskId);
expect(taskRow.status).toBe('failed');
});
it('multiple brokers should be able to observe a single task', async () => {
const broker1 = new StorageTaskBroker(storage);
const broker2 = new StorageTaskBroker(storage);
const { taskId } = await broker1.dispatch({ steps: [] });
const logPromise = new Promise<DbTaskEventRow[]>(resolve => {
const observedEvents = new Array<DbTaskEventRow>();
broker2.observe({ taskId, after: undefined }, (_err, { events }) => {
observedEvents.push(...events);
if (events.some(e => e.type === 'completion')) {
resolve(observedEvents);
}
});
});
const task = await broker1.claim();
await task.emitLog('log 1');
await task.emitLog('log 2');
await task.emitLog('log 3');
await task.complete('completed');
const logs = await logPromise;
expect(logs.map(l => l.body.message)).toEqual([
'log 1',
'log 2',
'log 3',
'Run completed with status: completed',
]);
const afterLogs = await new Promise<string[]>(resolve => {
broker2.observe({ taskId, after: logs[1].id }, (_err, { events }) =>
resolve(events.map(e => e.body.message as string)),
);
});
expect(afterLogs).toEqual([
'log 3',
'Run completed with status: completed',
]);
});
it('should heartbeat', async () => {
const broker = new StorageTaskBroker(storage);
const { taskId } = await broker.dispatch({ steps: [] });
const task = await broker.claim();
const initialTask = await storage.get(taskId);
for (;;) {
const maybeTask = await storage.get(taskId);
if (maybeTask.lastHeartbeat !== initialTask.lastHeartbeat) {
break;
}
await new Promise(resolve => setTimeout(resolve, 50));
}
await task.complete('completed');
expect.assertions(0);
});
it('should be cancelled if heartbeat stops', async () => {
const broker = new StorageTaskBroker(storage);
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);
for (;;) {
const maybeTask = await storage.get(taskId);
if (maybeTask.status === 'cancelled') {
break;
}
await new Promise(resolve => setTimeout(resolve, 50));
}
expect.assertions(0);
});
});
plugins/scaffolder-backend/src/scaffolder/tasks/StorageTaskBroker.ts
+10 -7
View File
@@ -25,7 +25,7 @@ import {
} from './types';
export class TaskAgent implements Task {
private heartbeartInterval?: ReturnType<typeof setInterval>;
private heartbeatInterval?: ReturnType<typeof setInterval>;
static create(state: TaskState, storage: TaskStore) {
const agent = new TaskAgent(state, storage);
@@ -67,13 +67,13 @@ export class TaskAgent implements Task {
body: { message: `Run completed with status: ${result}` },
type: 'completion',
});
if (this.heartbeartInterval) {
clearInterval(this.heartbeartInterval);
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
}
}
private start() {
this.heartbeartInterval = setInterval(() => {
this.heartbeatInterval = setInterval(() => {
if (!this.state.runId) {
throw new Error('no run id provided');
}
@@ -131,7 +131,10 @@ export class StorageTaskBroker implements TaskBroker {
taskId: string;
after: number | undefined;
},
callback: (result: { events: DbTaskEventRow[] }) => void,
callback: (
error: Error | undefined,
result: { events: DbTaskEventRow[] },
) => void,
): () => void {
const { taskId } = options;
@@ -148,9 +151,9 @@ export class StorageTaskBroker implements TaskBroker {
if (events.length) {
after = events[events.length - 1].id;
try {
callback(result);
callback(undefined, result);
} catch (error) {
console.log('DEBUG: error =', error);
callback(error, { events: [] });
}
}
plugins/scaffolder-backend/src/scaffolder/tasks/types.ts
+1
View File
@@ -86,6 +86,7 @@ export interface TaskStore {
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({
plugins/scaffolder-backend/src/service/router.test.ts
+23 -2
View File
@@ -16,6 +16,7 @@
const mockAccess = jest.fn();
jest.doMock('fs-extra', () => ({
access: mockAccess,
promises: {
access: mockAccess,
},
@@ -27,7 +28,11 @@ jest.doMock('fs-extra', () => ({
remove: jest.fn(),
}));
import { getVoidLogger } from '@backstage/backend-common';
import {
SingleConnectionDatabaseManager,
PluginDatabaseManager,
getVoidLogger,
} from '@backstage/backend-common';
import { ConfigReader } from '@backstage/config';
import express from 'express';
import request from 'supertest';
@@ -41,6 +46,19 @@ const generateEntityClient: any = (template: any) => ({
findTemplate: () => Promise.resolve(template),
});
function createDatabase(): PluginDatabaseManager {
return SingleConnectionDatabaseManager.fromConfig(
new ConfigReader({
backend: {
database: {
client: 'sqlite3',
connection: ':memory:',
},
},
}),
).forPlugin('scaffolder');
}
describe('createRouter - working directory', () => {
const mockPrepare = jest.fn();
const mockPreparers = new Preparers();
@@ -78,7 +96,6 @@ describe('createRouter - working directory', () => {
};
const mockedEntityClient = generateEntityClient(template);
it('should throw an error when working directory does not exist or is not writable', async () => {
mockAccess.mockImplementation(() => {
throw new Error('access error');
@@ -93,6 +110,7 @@ describe('createRouter - working directory', () => {
config: new ConfigReader(workDirConfig('/path')),
dockerClient: new Docker(),
entityClient: mockedEntityClient,
database: createDatabase(),
}),
).rejects.toThrow('access error');
});
@@ -106,6 +124,7 @@ describe('createRouter - working directory', () => {
config: new ConfigReader(workDirConfig('/path')),
dockerClient: new Docker(),
entityClient: mockedEntityClient,
database: createDatabase(),
});
const app = express().use(router);
@@ -134,6 +153,7 @@ describe('createRouter - working directory', () => {
config: new ConfigReader({}),
dockerClient: new Docker(),
entityClient: mockedEntityClient,
database: createDatabase(),
});
const app = express().use(router);
@@ -203,6 +223,7 @@ describe('createRouter', () => {
config: new ConfigReader({}),
dockerClient: new Docker(),
entityClient: generateEntityClient(template),
database: createDatabase(),
});
app = express().use(router);
});
plugins/scaffolder-backend/src/service/router.ts
+5 -1
View File
@@ -164,7 +164,11 @@ export async function createRouter(
// After client opens connection send all nests as string
const unsubscribe = taskBroker.observe(
{ taskId, after },
({ events }) => {
(error, { events }) => {
logger.error(
`Received error from event stream when observing task ${taskId}`,
error,
);
for (const event of events) {
res.write(`event:${JSON.stringify(event)}\n\n`);
if (event.type === 'completion') {