Actor Costs

1import { Actor, log } from 'apify';
2import { useState } from 'crawlee';
3import { processRuns } from './process-runs.js';
4
5interface Input {
6    actorIdOrName: string;
7    onlyRunsNewerThan?: string;
8    onlyRunsOlderThan?: string;
9    getCostBreakdown?: boolean;
10}
11
12interface DateAggregation {
13    date: string,
14    runCount: number,
15    cost: number,
16    costDetail: Record<string, number>,
17    firstRunDate: string,
18    lastRunDate: string,
19    buildNumbers: Record<string, number>,
20    statuses: Record<string, number>,
21    origins: Record<string, number>,
22}
23
24type DateAggregations = Record<string, DateAggregation>;
25
26// { date: stats }
27export interface State {
28    dateAggregations: DateAggregations;
29    lastProcessedRunId: string | null;
30    lastProcessedOffset: number;
31}
32
33// The init() call configures the Actor for its environment. It's recommended to start every Actor with an init()
34await Actor.init();
35
36const { actorIdOrName, onlyRunsNewerThan, onlyRunsOlderThan, getCostBreakdown = false } = (await Actor.getInput<Input>())!;
37
38let onlyRunsNewerThanDate;
39
40if (onlyRunsNewerThan) {
41    onlyRunsNewerThanDate = new Date(onlyRunsNewerThan);
42    if (Number.isNaN(onlyRunsNewerThanDate.getTime())) {
43        throw Actor.fail('Invalid date format for onlyRunsNewerThan, use YYYY-MM-DD or with time YYYY-MM-DDTHH:mm:ss');
44    }
45}
46
47let onlyRunsOlderThanDate;
48
49if (onlyRunsOlderThan) {
50    onlyRunsOlderThanDate = new Date(onlyRunsOlderThan);
51    if (Number.isNaN(onlyRunsOlderThanDate.getTime())) {
52        throw Actor.fail('Invalid date format for onlyRunsOlderThan, use YYYY-MM-DD or with time YYYY-MM-DDTHH:mm:ss');
53    }
54}
55
56const runsClient = Actor.apifyClient.actor(actorIdOrName).runs();
57
58const state = await useState<State>(
59    'STATE',
60    { lastProcessedOffset: 0, lastProcessedRunId: null, dateAggregations: {} },
61);
62
63const LIMIT = 1000;
64let offset = state.lastProcessedOffset;
65for (; ;) {
66    const runs = await runsClient.list({ desc: true, limit: 1000, offset }).then((res) => res.items);
67
68    log.info(`Loaded ${runs.length} runs (offset from now: ${offset}), newest: ${runs[0]?.startedAt}, `
69        + `oldest: ${runs[runs.length - 1]?.startedAt} processing them now`);
70
71    const { stopLoop } = await processRuns({
72        runs,
73        state,
74        onlyRunsOlderThanDate,
75        onlyRunsNewerThanDate,
76        getCostBreakdown,
77    });
78
79    state.lastProcessedOffset = offset;
80
81    if (stopLoop) {
82        log.warning(`Reached onlyRunsNewerThanDate ${onlyRunsNewerThanDate}, stopping loading runs`);
83        break;
84    }
85
86    if (runs.length < LIMIT) {
87        log.warning('No more runs to process, stopping loading runs');
88        break;
89    }
90
91    offset += LIMIT;
92}
93
94const totalStats: Omit<DateAggregation, 'date'> = {
95    runCount: 0,
96    cost: 0,
97    costDetail: {},
98    firstRunDate: '',
99    lastRunDate: '',
100    buildNumbers: {},
101    statuses: {},
102    origins: {},
103};
104
105await Actor.pushData(Object.values(state.dateAggregations)
106    .map((aggregation: DateAggregation) => {
107        totalStats.runCount += aggregation.runCount;
108        totalStats.cost += aggregation.cost;
109        if (!totalStats.lastRunDate) {
110            totalStats.lastRunDate = aggregation.lastRunDate;
111        }
112        totalStats.firstRunDate = aggregation.firstRunDate;
113        for (const [buildNumber, count] of Object.entries(aggregation.buildNumbers)) {
114            totalStats.buildNumbers[buildNumber] = (totalStats.buildNumbers[buildNumber] ?? 0) + count;
115        }
116        for (const [status, count] of Object.entries(aggregation.statuses)) {
117            totalStats.statuses[status] = (totalStats.statuses[status] ?? 0) + count;
118        }
119        for (const [origin, count] of Object.entries(aggregation.origins)) {
120            totalStats.origins[origin] = (totalStats.origins[origin] ?? 0) + count;
121        }
122
123        const cleanedCostDetail: Record<string, number> = {};
124
125        for (const [usageType, usageUsd] of Object.entries(aggregation.costDetail)) {
126            cleanedCostDetail[usageType] = Number(usageUsd.toFixed(4));
127            totalStats.costDetail[usageType] ??= 0
128            totalStats.costDetail[usageType] += Number(usageUsd.toFixed(4))
129        }
130
131        return { ...aggregation, cost: Number(aggregation.cost.toFixed(4)), costDetail: cleanedCostDetail };
132    }));
133
134await Actor.setValue('STATE', state);
135await Actor.setValue('TOTAL_STATS', totalStats);
136
137const store = await Actor.openKeyValueStore();
138const url = store.getPublicUrl('TOTAL_STATS');
139await Actor.exit(`Total stats for whole period are available at ${url}`);

1import { Actor, log } from 'apify';
2
3import type { ActorRunListItem, ActorRun } from 'apify-client';
4import { sleep } from 'crawlee';
5import type { State } from './main.js';
6
7interface ProcessRunsInputs {
8    runs: ActorRunListItem[];
9    state: State;
10    onlyRunsOlderThanDate?: Date;
11    onlyRunsNewerThanDate?: Date;
12    getCostBreakdown: boolean;
13}
14
15let isMigrating = false;
16Actor.on('migrating', () => {
17    isMigrating = true;
18});
19
20let foundLastProcessedRun = false;
21
22export const processRuns = async ({ runs, state, onlyRunsOlderThanDate, onlyRunsNewerThanDate, getCostBreakdown }: ProcessRunsInputs): Promise<{ stopLoop: boolean }> => {
23    // Runs are in decs mode
24    for (let run of runs) {
25        if (getCostBreakdown) {
26            run = (await Actor.apifyClient.run(run.id).get())! as ActorRun
27        }
28
29        if (isMigrating) {
30            log.warning('Actor is migrating, pausing all processing and storing last state to continue where we left of');
31            state.lastProcessedRunId = run.id;
32            await sleep(999999);
33        }
34
35        // If we load after migration, we need to find run we already processed
36        if (state.lastProcessedRunId && !foundLastProcessedRun) {
37            const isLastProcessed = state.lastProcessedRunId === run.id;
38            if (isLastProcessed) {
39                foundLastProcessedRun = true;
40                state.lastProcessedRunId = null;
41            } else {
42                log.warning(`Skipping run we already processed before migration ${run.id}`);
43                continue;
44            }
45        }
46
47        if (onlyRunsOlderThanDate && run.startedAt > onlyRunsOlderThanDate) {
48            continue;
49        }
50        if (onlyRunsNewerThanDate && run.startedAt < onlyRunsNewerThanDate) {
51            // We are going from present to past so at this point we can exit
52            return { stopLoop: true };
53        }
54
55        const runDate = run.startedAt.toISOString().split('T')[0];
56        state.dateAggregations[runDate] ??= {
57            date: runDate,
58            runCount: 0,
59            cost: 0,
60            costDetail: {},
61            firstRunDate: run.startedAt.toISOString(),
62            lastRunDate: run.startedAt.toISOString(),
63            buildNumbers: {},
64            statuses: {},
65            origins: {},
66        };
67
68        state.dateAggregations[runDate].runCount++;
69        state.dateAggregations[runDate].cost += run.usageTotalUsd ?? 0;
70
71
72        if ((run as ActorRun).usageUsd) {
73            for (const [usageType, usageUsd] of Object.entries((run as ActorRun).usageUsd as Record<string, number>)) {
74                state.dateAggregations[runDate].costDetail[usageType] ??= 0;
75                state.dateAggregations[runDate].costDetail[usageType] += usageUsd;
76            }
77        }
78
79        // lastRunDate is always the first we encounter because we go desc so we don't have to update it
80        state.dateAggregations[runDate].firstRunDate = run.startedAt.toISOString();
81
82        state.dateAggregations[runDate].buildNumbers[run.buildNumber] ??= 0;
83        state.dateAggregations[runDate].buildNumbers[run.buildNumber]++;
84
85        state.dateAggregations[runDate].statuses[run.status] ??= 0;
86        state.dateAggregations[runDate].statuses[run.status]++;
87
88        state.dateAggregations[runDate].origins[run.meta.origin] ??= 0;
89        state.dateAggregations[runDate].origins[run.meta.origin]++;
90    }
91
92    return { stopLoop: false };
93};