Skip to content

Commit 539868d

Browse files
mrgrainmrgrain
authored
refactor(cli): work-graph is using IoHost (#90)
Relates to aws/aws-cdk#32292 --- By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license
1 parent 58b3534 commit 539868d

File tree

8 files changed

+96
-54
lines changed

8 files changed

+96
-54
lines changed

‎packages/@aws-cdk/toolkit-lib/lib/actions/deploy/private/helpers.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { DeployOptions, HotswapProperties } from '..';
2-
import { Deployments, EcsHotswapProperties, HotswapPropertyOverrides, WorkGraph } from '../../../api/aws-cdk';
2+
import { Deployments, EcsHotswapProperties, HotswapPropertyOverrides, type WorkGraph } from '../../../api/aws-cdk';
33

44
export function buildParameterMap(parameters?: Map<string, string | undefined>): { [name: string]: { [name: string]: string | undefined } } {
55
const parameterMap: {

‎packages/@aws-cdk/toolkit-lib/lib/api/aws-cdk.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ export { DEFAULT_TOOLKIT_STACK_NAME } from '../../../../aws-cdk/lib/api/toolkit-
1010
export { ResourceMigrator } from '../../../../aws-cdk/lib/api/resource-import';
1111
export { StackActivityProgress } from '../../../../aws-cdk/lib/api/stack-events';
1212
export { CloudWatchLogEventMonitor, findCloudWatchLogGroups } from '../../../../aws-cdk/lib/api/logs';
13-
export { WorkGraph, WorkGraphBuilder, AssetBuildNode, AssetPublishNode, StackNode, Concurrency } from '../../../../aws-cdk/lib/api/work-graph';
13+
export { type WorkGraph, WorkGraphBuilder, AssetBuildNode, AssetPublishNode, StackNode, Concurrency } from '../../../../aws-cdk/lib/api/work-graph';
1414

1515
// Context Providers
1616
export * as contextproviders from '../../../../aws-cdk/lib/context-providers';

‎packages/@aws-cdk/toolkit-lib/lib/toolkit/toolkit.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -518,7 +518,7 @@ export class Toolkit extends CloudAssemblySourceBuilder implements AsyncDisposab
518518
stack,
519519
...stack.dependencies.filter(cxapi.AssetManifestArtifact.isAssetManifestArtifact),
520520
]);
521-
const workGraph = new WorkGraphBuilder(prebuildAssets).build(stacksAndTheirAssetManifests);
521+
const workGraph = new WorkGraphBuilder({ ioHost, action }, prebuildAssets).build(stacksAndTheirAssetManifests);
522522

523523
// Unless we are running with '--force', skip already published assets
524524
if (!options.force) {

‎packages/aws-cdk/lib/api/work-graph/work-graph-builder.ts‎

Lines changed: 18 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import * as cxapi from '@aws-cdk/cx-api';
22
import { AssetManifest, type IManifestEntry } from 'cdk-assets';
33
import { WorkGraph } from './work-graph';
44
import { DeploymentState, AssetBuildNode, WorkNode } from './work-graph-types';
5+
import { IoMessaging } from '../../toolkit/cli-io-host';
56
import { ToolkitError } from '../../toolkit/error';
67
import { contentHashAny } from '../../util/content-hash';
78

@@ -20,10 +21,18 @@ export class WorkGraphBuilder {
2021
'asset-publish': 0,
2122
'stack': 5,
2223
};
23-
private readonly graph = new WorkGraph();
24-
25-
constructor(private readonly prebuildAssets: boolean, private readonly idPrefix = '') {
26-
24+
private readonly graph: WorkGraph;
25+
private readonly ioHost: IoMessaging['ioHost'];
26+
private readonly action: IoMessaging['action'];
27+
28+
constructor(
29+
{ ioHost, action }: IoMessaging,
30+
private readonly prebuildAssets: boolean,
31+
private readonly idPrefix = '',
32+
) {
33+
this.graph = new WorkGraph({}, { ioHost, action });
34+
this.ioHost = ioHost;
35+
this.action = action;
2736
}
2837

2938
private addStack(artifact: cxapi.CloudFormationStackArtifact) {
@@ -120,7 +129,11 @@ export class WorkGraphBuilder {
120129
}
121130
} else if (cxapi.NestedCloudAssemblyArtifact.isNestedCloudAssemblyArtifact(artifact)) {
122131
const assembly = new cxapi.CloudAssembly(artifact.fullPath, { topoSort: false });
123-
const nestedGraph = new WorkGraphBuilder(this.prebuildAssets, `${this.idPrefix}${artifact.id}.`).build(assembly.artifacts);
132+
const nestedGraph = new WorkGraphBuilder(
133+
{ ioHost: this.ioHost, action: this.action },
134+
this.prebuildAssets,
135+
`${this.idPrefix}${artifact.id}.`,
136+
).build(assembly.artifacts);
124137
this.graph.absorb(nestedGraph);
125138
} else {
126139
// Ignore whatever else

‎packages/aws-cdk/lib/api/work-graph/work-graph.ts‎

Lines changed: 41 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,30 @@
11
import { WorkNode, DeploymentState, StackNode, AssetBuildNode, AssetPublishNode } from './work-graph-types';
2-
import { debug, trace } from '../../logging';
2+
import { debug, trace } from '../../cli/messages';
3+
import { IoMessaging } from '../../toolkit/cli-io-host';
34
import { ToolkitError } from '../../toolkit/error';
45
import { parallelPromises } from '../../util/parallel';
56

67
export type Concurrency = number | Record<WorkNode['type'], number>;
78

9+
export interface WorkGraphProps {
10+
ioHost: IoMessaging['ioHost'];
11+
action: IoMessaging['action'];
12+
}
13+
814
export class WorkGraph {
915
public readonly nodes: Record<string, WorkNode>;
1016
private readonly readyPool: Array<WorkNode> = [];
1117
private readonly lazyDependencies = new Map<string, string[]>();
18+
private readonly ioHost: IoMessaging['ioHost'];
19+
private readonly action: IoMessaging['action'];
20+
1221
public error?: Error;
1322

14-
public constructor(nodes: Record<string, WorkNode> = {}) {
23+
public constructor(nodes: Record<string, WorkNode>, props: WorkGraphProps) {
1524
this.nodes = { ...nodes };
25+
26+
this.ioHost = props.ioHost;
27+
this.action = props.action;
1628
}
1729

1830
public addNodes(...nodes: WorkNode[]) {
@@ -118,8 +130,8 @@ export class WorkGraph {
118130
/**
119131
* Return the set of unblocked nodes
120132
*/
121-
public ready(): ReadonlyArray<WorkNode> {
122-
this.updateReadyPool();
133+
public async ready(): Promise<ReadonlyArray<WorkNode>> {
134+
await this.updateReadyPool();
123135
return this.readyPool;
124136
}
125137

@@ -149,28 +161,30 @@ export class WorkGraph {
149161
start();
150162

151163
function start() {
152-
graph.updateReadyPool();
153-
154-
for (let i = 0; i < graph.readyPool.length; ) {
155-
const node = graph.readyPool[i];
156-
157-
if (active[node.type] < max[node.type] && totalActive() < totalMax) {
158-
graph.readyPool.splice(i, 1);
159-
startOne(node);
160-
} else {
161-
i += 1;
164+
graph.updateReadyPool().then(() => {
165+
for (let i = 0; i < graph.readyPool.length; ) {
166+
const node = graph.readyPool[i];
167+
168+
if (active[node.type] < max[node.type] && totalActive() < totalMax) {
169+
graph.readyPool.splice(i, 1);
170+
startOne(node);
171+
} else {
172+
i += 1;
173+
}
162174
}
163-
}
164175

165-
if (totalActive() === 0) {
166-
if (graph.done()) {
167-
ok();
176+
if (totalActive() === 0) {
177+
if (graph.done()) {
178+
ok();
179+
}
180+
// wait for other active deploys to finish before failing
181+
if (graph.hasFailed()) {
182+
fail(graph.error);
183+
}
168184
}
169-
// wait for other active deploys to finish before failing
170-
if (graph.hasFailed()) {
171-
fail(graph.error);
172-
}
173-
}
185+
}).catch((e) => {
186+
fail(e);
187+
});
174188
}
175189

176190
function startOne(x: WorkNode) {
@@ -252,7 +266,7 @@ export class WorkGraph {
252266
* Do this in parallel, because there may be a lot of assets in an application (seen in practice: >100 assets)
253267
*/
254268
public async removeUnnecessaryAssets(isUnnecessary: (x: AssetPublishNode) => Promise<boolean>) {
255-
debug('Checking for previously published assets');
269+
await this.ioHost.notify(debug(this.action, 'Checking for previously published assets'));
256270

257271
const publishes = this.nodesOfType('asset-publish');
258272

@@ -265,7 +279,7 @@ export class WorkGraph {
265279
this.removeNode(assetNode);
266280
}
267281

268-
debug(`${publishes.length} total assets, ${publishes.length - alreadyPublished.length} still need to be published`);
282+
await this.ioHost.notify(debug(this.action, `${publishes.length} total assets, ${publishes.length - alreadyPublished.length} still need to be published`));
269283

270284
// Now also remove any asset build steps that don't have any dependencies on them anymore
271285
const unusedBuilds = this.nodesOfType('asset-build').filter(build => this.dependees(build).length === 0);
@@ -274,7 +288,7 @@ export class WorkGraph {
274288
}
275289
}
276290

277-
private updateReadyPool() {
291+
private async updateReadyPool() {
278292
const activeCount = Object.values(this.nodes).filter((x) => x.deploymentState === DeploymentState.DEPLOYING).length;
279293
const pendingCount = Object.values(this.nodes).filter((x) => x.deploymentState === DeploymentState.PENDING).length;
280294

@@ -296,7 +310,7 @@ export class WorkGraph {
296310

297311
if (this.readyPool.length === 0 && activeCount === 0 && pendingCount > 0) {
298312
const cycle = this.findCycle() ?? ['No cycle found!'];
299-
trace(`Cycle ${cycle.join(' -> ')} in graph ${this}`);
313+
await this.ioHost.notify(trace(this.action, `Cycle ${cycle.join(' -> ')} in graph ${this}`));
300314
throw new ToolkitError(`Unable to make progress anymore, dependency cycle between remaining artifacts: ${cycle.join(' -> ')} (run with -vv for full graph)`);
301315
}
302316
}

‎packages/aws-cdk/lib/cli/cdk-toolkit.ts‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ import { CloudWatchLogEventMonitor } from '../api/logs/logs-monitor';
2727
import { ResourceImporter, removeNonImportResources, ResourceMigrator } from '../api/resource-import';
2828
import { StackActivityProgress } from '../api/stack-events';
2929
import { tagsForStack, type Tag } from '../api/tags';
30-
import { type AssetBuildNode, type AssetPublishNode, type Concurrency, type StackNode, WorkGraph } from '../api/work-graph';
30+
import { type AssetBuildNode, type AssetPublishNode, type Concurrency, type StackNode, type WorkGraph } from '../api/work-graph';
3131
import { WorkGraphBuilder } from '../api/work-graph/work-graph-builder';
3232
import {
3333
generateCdkApp,
@@ -583,7 +583,10 @@ export class CdkToolkit {
583583
stack,
584584
...stack.dependencies.filter(cxapi.AssetManifestArtifact.isAssetManifestArtifact),
585585
]);
586-
const workGraph = new WorkGraphBuilder(prebuildAssets).build(stacksAndTheirAssetManifests);
586+
const workGraph = new WorkGraphBuilder({
587+
ioHost: this.ioHost,
588+
action: 'deploy',
589+
}, prebuildAssets).build(stacksAndTheirAssetManifests);
587590

588591
// Unless we are running with '--force', skip already published assets
589592
if (!options.force) {

‎packages/aws-cdk/test/api/work-graph/work-graph-builder.test.ts‎

Lines changed: 19 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,14 @@ import { CloudAssemblyBuilder } from '@aws-cdk/cx-api';
77
import { expect } from '@jest/globals';
88
import { WorkGraph, WorkGraphBuilder } from '../../../lib/api/work-graph';
99
import type { AssetBuildNode, AssetPublishNode, StackNode, WorkNode } from '../../../lib/api/work-graph';
10+
import { CliIoHost, IoMessaging } from '../../../lib/toolkit/cli-io-host';
1011

1112
let rootBuilder: CloudAssemblyBuilder;
13+
let mockMsg: IoMessaging = {
14+
ioHost: CliIoHost.instance(),
15+
action: 'deploy'
16+
};
17+
1218
beforeEach(() => {
1319
rootBuilder = new CloudAssemblyBuilder();
1420
});
@@ -44,7 +50,7 @@ describe('with some stacks and assets', () => {
4450
});
4551

4652
test('stack depends on the asset publishing step', () => {
47-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
53+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
4854

4955
expect(assertableNode(graph.node('stack2'))).toEqual(expect.objectContaining({
5056
type: 'stack',
@@ -53,7 +59,7 @@ describe('with some stacks and assets', () => {
5359
});
5460

5561
test('asset publishing step depends on asset building step', () => {
56-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
62+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
5763

5864
expect(graph.node('publish-F1-add54bdbcb')).toEqual(expect.objectContaining({
5965
type: 'asset-publish',
@@ -62,7 +68,7 @@ describe('with some stacks and assets', () => {
6268
});
6369

6470
test('with prebuild off, asset building inherits dependencies from their parent stack', () => {
65-
const graph = new WorkGraphBuilder(false).build(assembly.artifacts);
71+
const graph = new WorkGraphBuilder(mockMsg, false).build(assembly.artifacts);
6672

6773
expect(graph.node('build-F1-a533139934')).toEqual(expect.objectContaining({
6874
type: 'asset-build',
@@ -71,7 +77,7 @@ describe('with some stacks and assets', () => {
7177
});
7278

7379
test('with prebuild on, assets only have their own dependencies', () => {
74-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
80+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
7581

7682
expect(graph.node('build-F1-a533139934')).toEqual(expect.objectContaining({
7783
type: 'asset-build',
@@ -90,8 +96,8 @@ test('tree metadata is ignored', async () => {
9096

9197
const assembly = rootBuilder.buildAssembly();
9298

93-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
94-
expect(graph.ready().length).toEqual(0);
99+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
100+
expect((await graph.ready()).length).toEqual(0);
95101
});
96102

97103
test('can handle nested assemblies', async () => {
@@ -103,7 +109,7 @@ test('can handle nested assemblies', async () => {
103109
const assembly = rootBuilder.buildAssembly();
104110

105111
let workDone = 0;
106-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
112+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
107113

108114
await graph.doParallel(10, {
109115
deployStack: async () => { workDone += 1; },
@@ -126,8 +132,8 @@ test('dependencies on unselected artifacts are silently ignored', async () => {
126132
});
127133

128134
const asm = rootBuilder.buildAssembly();
129-
const graph = new WorkGraphBuilder(true).build([asm.getStackArtifact('stackB')]);
130-
expect(graph.ready()[0]).toEqual(expect.objectContaining({
135+
const graph = new WorkGraphBuilder(mockMsg, true).build([asm.getStackArtifact('stackB')]);
136+
expect((await graph.ready())[0]).toEqual(expect.objectContaining({
131137
id: 'stackB',
132138
dependencies: new Set(),
133139
}));
@@ -162,7 +168,7 @@ describe('tests that use assets', () => {
162168

163169
const assembly = rootBuilder.buildAssembly();
164170

165-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
171+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
166172
const traversal = await traverseAndRecord(graph);
167173

168174
expect(traversal).toEqual([
@@ -184,7 +190,7 @@ describe('tests that use assets', () => {
184190
addAssets(rootBuilder, 'StackC.assets', { files });
185191

186192
const assembly = rootBuilder.buildAssembly();
187-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
193+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
188194

189195
// THEN
190196
expect(graph.findCycle()).toBeUndefined();
@@ -224,7 +230,7 @@ describe('tests that use assets', () => {
224230

225231
const assembly = rootBuilder.buildAssembly();
226232

227-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
233+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
228234
const traversal = await traverseAndRecord(graph);
229235

230236
expect(traversal).toEqual([
@@ -270,7 +276,7 @@ describe('tests that use assets', () => {
270276

271277
const assembly = rootBuilder.buildAssembly();
272278

273-
const graph = new WorkGraphBuilder(true).build(assembly.artifacts);
279+
const graph = new WorkGraphBuilder(mockMsg, true).build(assembly.artifacts);
274280
const traversal = await traverseAndRecord(graph);
275281

276282
expect(traversal).toEqual([

‎packages/aws-cdk/test/api/work-graph/work-graph.test.ts‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import { WorkGraph, DeploymentState } from '../../../lib/api/work-graph';
22
import type { AssetBuildNode, AssetPublishNode, StackNode } from '../../../lib/api/work-graph';
3+
import { CliIoHost, IoMessaging } from '../../../lib/toolkit/cli-io-host';
34

45
const DUMMY: any = 'DUMMY';
56

@@ -9,6 +10,11 @@ const sleep = async (duration: number) => new Promise<void>((resolve) => setTime
910
// a chance to start new tasks.
1011
const SLOW = 200;
1112

13+
let mockMsg: IoMessaging = {
14+
ioHost: CliIoHost.instance(),
15+
action: 'deploy'
16+
}
17+
1218
/**
1319
* Repurposing unused stack attributes to create specific test scenarios
1420
* - stack.name = deployment duration
@@ -243,7 +249,7 @@ describe('WorkGraph', () => {
243249
expected: ['c-build', 'c-publish', 'A', 'b-build', 'b-publish', 'B'],
244250
},
245251
])('Success - Concurrency: $concurrency - $scenario', async ({ concurrency, expected, toDeploy }) => {
246-
const graph = new WorkGraph();
252+
const graph = new WorkGraph({}, mockMsg);
247253
addTestArtifactsToGraph(toDeploy, graph);
248254

249255
await graph.doParallel(concurrency, callbacks);
@@ -252,7 +258,7 @@ describe('WorkGraph', () => {
252258
});
253259

254260
test('can remove unnecessary assets', async () => {
255-
const graph = new WorkGraph();
261+
const graph = new WorkGraph({}, mockMsg);
256262
addTestArtifactsToGraph([
257263
{ id: 'a', type: 'asset' },
258264
{ id: 'b', type: 'asset' },
@@ -375,7 +381,7 @@ describe('WorkGraph', () => {
375381
expected: ['b-build', 'C'],
376382
},
377383
])('Failure - Concurrency: $concurrency - $scenario', async ({ concurrency, expectedError, toDeploy, expected }) => {
378-
const graph = new WorkGraph();
384+
const graph = new WorkGraph({}, mockMsg);
379385
addTestArtifactsToGraph(toDeploy, graph);
380386

381387
await expect(graph.doParallel(concurrency, callbacks)).rejects.toThrow(expectedError);
@@ -411,7 +417,7 @@ describe('WorkGraph', () => {
411417
expectedError: 'B -> C -> D -> B',
412418
},
413419
])('Failure - Graph Circular Dependencies - $scenario', async ({ toDeploy, expectedError }) => {
414-
const graph = new WorkGraph();
420+
const graph = new WorkGraph({}, mockMsg);
415421
addTestArtifactsToGraph(toDeploy, graph);
416422

417423
await expect(graph.doParallel(1, callbacks)).rejects.toThrow(new RegExp(`Unable to make progress.*${expectedError}`));

0 commit comments

Comments
 (0)