Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Reduce concurrency when copying image files to S3, add 1 GB memory #83

Merged
merged 3 commits into from
Aug 8, 2024
Merged

Reduce concurrency when copying image files to S3, add 1 GB memory #83

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9e8d018
Reduce concurrency when copying image files to S3, add 1 GB memory
nathanielrindlaub Aug 7, 2024
d8d3d4b
Reduce concurrency to 100
nathanielrindlaub Aug 7, 2024
94f0e7c
Fix lint errors
nathanielrindlaub Aug 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions ingest-delete/task.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ export async function handler(event) {
new SSM.GetParametersCommand({
Names: [`/api/url-${STAGE}`],
WithDecryption: true,
}),
})
)
).Parameters) {
console.log(`ok - setting ${param.Name}`);
Expand Down Expand Up @@ -117,7 +117,7 @@ async function scheduledDelete(stage) {
const res = await cf.send(
new CloudFormation.ListStacksCommand({
NextToken: nextToken,
}),
})
);

stacks.push(...res.StackSummaries);
Expand All @@ -142,7 +142,7 @@ async function scheduledDelete(stage) {
const alarm = await cw.send(
new CW.DescribeAlarmsCommand({
AlarmNames: [`${stack.StackName}-sqs-empty`],
}),
})
);

if (alarm.MetricAlarms[0].StateValue === 'INSUFFICIENT_DATA') {
Expand Down
228 changes: 116 additions & 112 deletions ingest-delete/test/basic.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,131 +6,135 @@ import { MockAgent, setGlobalDispatcher } from 'undici';
import SSM from '@aws-sdk/client-ssm';

test('Basic - CloudWatch Alarm', async (t) => {
const mockAgent = new MockAgent();
setGlobalDispatcher(mockAgent);

const mockPool = mockAgent.get('http://example.com');

mockPool.intercept({
path: '/',
method: 'POST'
}).reply(200, { message: 'posted' });

const order = [];
Sinon.stub(SSM.SSMClient.prototype, 'send').callsFake((command) => {
if (command instanceof SSM.GetParametersCommand) {
order.push('SSM:GetParametersCommand');

t.deepEquals(command.input, {
Names: ['/api/url-test'],
WithDecryption: true
});

return Promise.resolve({
Parameters: [{
Name: '/api/url-test',
Value: 'http://example.com'
}]
});
} else {
t.fail('Unexpected Command');
}
});

Sinon.stub(CloudFormation.CloudFormationClient.prototype, 'send').callsFake((command) => {
if (command instanceof CloudFormation.DeleteStackCommand) {
order.push('CloudFormation:DeleteStack');
const mockAgent = new MockAgent();
setGlobalDispatcher(mockAgent);

const mockPool = mockAgent.get('http://example.com');

mockPool
.intercept({
path: '/',
method: 'POST',
})
.reply(200, { message: 'posted' });

const order = [];
Sinon.stub(SSM.SSMClient.prototype, 'send').callsFake((command) => {
if (command instanceof SSM.GetParametersCommand) {
order.push('SSM:GetParametersCommand');

t.deepEquals(command.input, {
Names: ['/api/url-test'],
WithDecryption: true,
});

return Promise.resolve({
Parameters: [
{
Name: '/api/url-test',
Value: 'http://example.com',
},
],
});
} else {
t.fail('Unexpected Command');
}
});

t.ok(command.input.StackName.startsWith('test-stack-batch'));
Sinon.stub(CloudFormation.CloudFormationClient.prototype, 'send').callsFake((command) => {
if (command instanceof CloudFormation.DeleteStackCommand) {
order.push('CloudFormation:DeleteStack');

return Promise.resolve({});
} else {
t.fail('Unexpected Command');
}
});
t.ok(command.input.StackName.startsWith('test-stack-batch'));

try {
process.env.STAGE = 'test';
await IngestDelete({
Records: [{
Sns: {
Message: JSON.stringify({
AlarmName: 'test-stack-batch-847400c8-8e54-4a40-ab5c-0d593939c883-sqs-empty'
})
}
}]
});
} catch (err) {
t.error(err);
return Promise.resolve({});
} else {
t.fail('Unexpected Command');
}
});

try {
process.env.STAGE = 'test';
await IngestDelete({
Records: [
{
Sns: {
Message: JSON.stringify({
AlarmName: 'test-stack-batch-847400c8-8e54-4a40-ab5c-0d593939c883-sqs-empty',
}),
},
},
],
});
} catch (err) {
t.error(err);
}

t.deepEquals(order, [
'SSM:GetParametersCommand',
'CloudFormation:DeleteStack'
]);
t.deepEquals(order, ['SSM:GetParametersCommand', 'CloudFormation:DeleteStack']);

Sinon.restore();
t.end();
Sinon.restore();
t.end();
});

test('Basic - StopBatch API', async (t) => {
const mockAgent = new MockAgent();
setGlobalDispatcher(mockAgent);

const mockPool = mockAgent.get('http://example.com');

mockPool.intercept({
path: '/',
method: 'POST'
}).reply(200, { message: 'posted' });

const order = [];
Sinon.stub(SSM.SSMClient.prototype, 'send').callsFake((command) => {
if (command instanceof SSM.GetParametersCommand) {
order.push('SSM:GetParametersCommand');

t.deepEquals(command.input, {
Names: ['/api/url-test'],
WithDecryption: true
});

return Promise.resolve({
Parameters: [{
Name: '/api/url-test',
Value: 'http://example.com'
}]
});
} else {
t.fail('Unexpected Command');
}
});

Sinon.stub(CloudFormation.CloudFormationClient.prototype, 'send').callsFake((command) => {
if (command instanceof CloudFormation.DeleteStackCommand) {
order.push('CloudFormation:DeleteStack');
const mockAgent = new MockAgent();
setGlobalDispatcher(mockAgent);

const mockPool = mockAgent.get('http://example.com');

mockPool
.intercept({
path: '/',
method: 'POST',
})
.reply(200, { message: 'posted' });

const order = [];
Sinon.stub(SSM.SSMClient.prototype, 'send').callsFake((command) => {
if (command instanceof SSM.GetParametersCommand) {
order.push('SSM:GetParametersCommand');

t.deepEquals(command.input, {
Names: ['/api/url-test'],
WithDecryption: true,
});

return Promise.resolve({
Parameters: [
{
Name: '/api/url-test',
Value: 'http://example.com',
},
],
});
} else {
t.fail('Unexpected Command');
}
});

t.ok(command.input.StackName.startsWith('animl-ingest-test-batch-'));
Sinon.stub(CloudFormation.CloudFormationClient.prototype, 'send').callsFake((command) => {
if (command instanceof CloudFormation.DeleteStackCommand) {
order.push('CloudFormation:DeleteStack');

return Promise.resolve({});
} else {
t.fail('Unexpected Command');
}
});
t.ok(command.input.StackName.startsWith('animl-ingest-test-batch-'));

try {
process.env.STAGE = 'test';
await IngestDelete({
batch: 'batch-847400c8-8e54-4a40-ab5c-0d593939c883'
});
} catch (err) {
t.error(err);
return Promise.resolve({});
} else {
t.fail('Unexpected Command');
}
});

try {
process.env.STAGE = 'test';
await IngestDelete({
batch: 'batch-847400c8-8e54-4a40-ab5c-0d593939c883',
});
} catch (err) {
t.error(err);
}

t.deepEquals(order, [
'SSM:GetParametersCommand',
'CloudFormation:DeleteStack'
]);
t.deepEquals(order, ['SSM:GetParametersCommand', 'CloudFormation:DeleteStack']);

Sinon.restore();
t.end();
Sinon.restore();
t.end();
});
32 changes: 22 additions & 10 deletions ingest-zip/index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ export default async function handler() {
new SSM.GetParametersCommand({
Names: [`/api/url-${STAGE}`],
WithDecryption: true,
}),
})
)
).Parameters) {
console.log(`ok - setting ${param.Name}`);
Expand All @@ -59,10 +59,10 @@ export default async function handler() {
new S3.GetObjectCommand({
Bucket: task.Bucket,
Key: task.Key,
}),
})
)
).Body,
fs.createWriteStream(path.resolve(os.tmpdir(), 'input.zip')),
fs.createWriteStream(path.resolve(os.tmpdir(), 'input.zip'))
);

// Preparse Zip to get a general sense of how many items are in the zip
Expand Down Expand Up @@ -116,7 +116,7 @@ export default async function handler() {
ParameterValue: `s3://${task.Bucket}/${task.Key}`,
},
],
}),
})
);

await monitor(StackName);
Expand All @@ -143,34 +143,46 @@ export default async function handler() {
images.push(entry);
}

for await (const ms of asyncPool(1000, images, async (entry) => {
console.time('copying images to S3 @ 100 concurrency');

let maxMemoryUsed = 0;
for await (const ms of asyncPool(100, images, async (entry) => {
const parsed = path.parse(entry.name);
if (!parsed.ext) return;
if (parsed.base[0] === '.') return;
if (!SUPPORTED_FILE_TYPES.includes(parsed.ext.toLowerCase())) return;
if (!parsed.ext) return `not ok - no extension: ${entry.name}`;
if (parsed.base[0] === '.') return `not ok - hidden file: ${entry.name}`;
if (!SUPPORTED_FILE_TYPES.includes(parsed.ext.toLowerCase()))
return `not ok - unsupported file type: ${entry.name}`;

const data = await prezip.entryData(entry);

// NOTE: just for logging memory usage
if (process.memoryUsage().rss > maxMemoryUsed) {
maxMemoryUsed = process.memoryUsage().rss;
}

await s3.send(
new S3.PutObjectCommand({
Bucket: task.Bucket,
Key: `${batch}/${entry.name}`,
Body: data,
}),
})
);

return `ok - written: ${batch}/${entry.name}`;
})) {
console.log(ms);
}

console.timeEnd('copying images to S3 @ 100 concurrency');
console.log(`max memory used: ${maxMemoryUsed / 1024 / 1024} MB`);

prezip.close();

await s3.send(
new S3.DeleteObjectCommand({
Bucket: task.Bucket,
Key: task.Key,
}),
})
);

console.log('ok - extraction complete');
Expand Down
Loading
Loading