Compute Per Actor

# This is a template for a Dockerfile used to run acts in Actor system.
# The base image name below is set during the act build, based on user settings.
# IMPORTANT: The base image must set a correct working directory, such as /usr/src/app or /home/user
FROM apify/actor-node-basic:v0.21.10

# Second, copy just package.json and package-lock.json since it should be
# the only file that affects "npm install" in the next step, to speed up the build
COPY package*.json ./

# Install NPM packages, skip optional and development dependencies to
# keep the image small. Avoid logging too much and print the dependency
# tree for debugging
RUN npm --quiet set progress=false \
 && npm install --only=prod --no-optional \
 && echo "Installed NPM packages:" \
 && (npm list --all || true) \
 && echo "Node.js version:" \
 && node --version \
 && echo "NPM version:" \
 && npm --version

# Copy source code to container
# Do this in the last step, to have fast build if only the source code changed
COPY  . ./

# NOTE: The CMD is already defined by the base image.
# Uncomment this for local node inspector debugging:
# CMD [ "node", "--inspect=0.0.0.0:9229", "main.js" ]

{
    "name": "apify-project",
    "version": "0.0.1",
    "description": "",
    "author": "It's not you it's me",
    "license": "ISC",
    "dependencies": {
        "apify": "0.21.10"
    },
    "scripts": {
        "start": "node main.js"
    }
}

1const Apify = require('apify');
2
3const { client, utils: { log, sleep } } = Apify;
4
5Apify.main(async () => {
6    log.info('Getting actors.');
7    const actors = await getActors();
8    log.info('Getting actor runs.');
9    const actorsWithRuns = await getActorsWithRuns(actors);
10    log.info('Getting stats for individual runs.');
11    const pool = createRunPool(actorsWithRuns);
12    await pool.run();
13    await Apify.setValue('ACTORS', actorsWithRuns);
14    log.info('Calculating CU usage.');
15    const results = calculateComputeUnitUsage(actorsWithRuns);
16    log.info('Saving output.');
17    await Apify.setValue('OUTPUT', { actors: results });
18});
19
20async function getActors() {
21    const { items } = await client.acts.listActs();
22    return items;
23}
24
25async function getActorsWithRuns(actors) {
26    const actorsWithRunsPromises = actors.map(async (actor, actorIdx) => {
27        await sleep(actorIdx * 33);
28        log.info(`Getting runs for actor: ${actor.id}`);
29        const { items } = await client.acts.listRuns({ actId: actor.id })
30        return { ...actor, runs: items };
31    }, {});
32    return Promise.all(actorsWithRunsPromises);
33}
34
35async function getRunStats(actorId, run) {
36    return client.acts.getRun({ actId: actorId, runId: run.id })
37}
38
39function createRunPool(actorsWithRuns) {
40    const allTasks = actorsWithRuns.reduce((tasks, actor, actorIndex) => {
41        const runs = actor.runs.map((run, runIndex) => ({
42            actorId: actor.id,
43            actorIndex,
44            run,
45            runIndex,
46        }));
47        return tasks.concat(runs);
48    }, []);
49    setInterval(() => {
50        log.info(`There are ${allTasks.length} remaining runs to process.`);
51    }, 10000)
52    return new Apify.AutoscaledPool({
53        minConcurrency: 5,
54        runTaskFunction: async () => {
55            const task = allTasks.shift();
56            const runWithStats = await getRunStats(task.actorId, task.run);
57            actorsWithRuns[task.actorIndex].runs[task.runIndex] = runWithStats;
58        },
59        isTaskReadyFunction: () => !!allTasks.length,
60        isFinishedFunction: () => !allTasks.length
61    })
62}
63
64function calculateComputeUnitUsage(actors) {
65    return actors.map((actor) => {
66        const computeUnitUsage = Array.isArray(actor.runs)
67            ? actor.runs.reduce((sum, run) => sum + (run.stats.computeUnits || 0), 0)
68            : null;
69        return {
70            id: actor.id,
71            name: actor.name,
72            computeUnitUsage,
73        }
74    })
75}