Abort And Resurrect

mnmkng/abort-and-resurrect

No credit card required

Aborts actor runs in bulk and resurrects them with new input and options.

.dockerignore

# configurations
.idea

# crawlee and apify storage folders
apify_storage
crawlee_storage
storage

# installed files
node_modules

# git folder
.git

.gitignore

storage
node_modules

package.json

{
	"name": "actor-resurrect-with-new-input",
	"version": "0.0.1",
	"type": "module",
	"description": "This is an example of an Apify actor.",
	"dependencies": {
		"apify": "^3.0.0"
	},
	"devDependencies": {},
	"scripts": {
		"start": "node src/main.js",
		"test": "echo \"Error: oops, the actor has no tests yet, sad!\" && exit 1"
	},
	"author": "It's not you it's me",
	"license": "ISC"
}

.actor/Dockerfile

# Specify the base Docker image. You can read more about
# the available images at https://sdk.apify.com/docs/guides/docker-images
# You can also use any other image from Docker Hub.
FROM apify/actor-node:16

# Copy just package.json and package-lock.json
# to speed up the build using Docker layer cache.
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 --omit=dev --omit=optional \
    && echo "Installed NPM packages:" \
    && (npm list --omit=dev --all || true) \
    && echo "Node.js version:" \
    && node --version \
    && echo "NPM version:" \
    && npm --version \
    && rm -r ~/.npm

# Next, copy the remaining files and directories with the source code.
# Since we do this after NPM install, quick build will be really fast
# for most source file changes.
COPY . ./


# Run the image.
CMD npm start --silent

.actor/actor.json

{
	"actorSpecification": 1,
	"name": "abort-and-resurrect",
	"title": "Abort & Resurrect With New Input",
	"description": "Aborts selected actor or task runs ",
	"version": "0.0",
	"buildTag": "version-0",
	"input": "./input-schema.json"
}

.actor/input-schema.json

{
  "title": "Actor input schema",
  "description": "This is actor input schema",
  "type": "object",
  "schemaVersion": 1,
  "properties": {
    "runIds": {
      "title": "Run IDs",
      "type": "array",
      "description": "IDs of individual runs you want to abort and resurrect.",
      "editor": "stringList",
      "default": []
    },
    "actorIds": {
      "title": "Actor IDs",
      "type": "array",
      "description": "When you provide an actor ID, all running runs of the actor will be aborted and resurrected.",
      "editor": "stringList",
      "default": []
    },
    "taskIds": {
      "title": "Task IDs",
      "type": "array",
      "description": "When you provide a task ID, all running runs of the task will be aborted and resurrected.",
      "editor": "stringList",
      "default": []
    },
    "transformInputSource": {
      "title": "Transform Input Function",
      "type": "string",
      "description": "This function will be used to transform the original input of the actor run into the new input. The `input` and `option` arguments are mutable and you should make your changes directly in the provided objects.",
      "editor": "javascript",
      "prefill": "function transformInput({ input, options }) {\n     input.foo = 'bar';\n     options.memory = 2048;\n }"
    }
  },
  "required": [
    "transformInputSource"
  ]
}

.idea/.gitignore

# Default ignored files
/shelf/
/workspace.xml
# Editor-based HTTP Client requests
/httpRequests/

.idea/actor-resurrect-with-new-input.iml

<?xml version="1.0" encoding="UTF-8"?>
<module type="WEB_MODULE" version="4">
  <component name="NewModuleRootManager">
    <content url="file://$MODULE_DIR$">
      <excludeFolder url="file://$MODULE_DIR$/temp" />
      <excludeFolder url="file://$MODULE_DIR$/.tmp" />
      <excludeFolder url="file://$MODULE_DIR$/tmp" />
    </content>
    <orderEntry type="inheritedJdk" />
    <orderEntry type="sourceFolder" forTests="false" />
  </component>
</module>

.idea/jsLibraryMappings.xml

<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
  <component name="JavaScriptLibraryMappings">
    <includedPredefinedLibrary name="Node.js Core" />
  </component>
</project>

.idea/modules.xml

<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
  <component name="ProjectModuleManager">
    <modules>
      <module fileurl="file://$PROJECT_DIR$/.idea/actor-resurrect-with-new-input.iml" filepath="$PROJECT_DIR$/.idea/actor-resurrect-with-new-input.iml" />
    </modules>
  </component>
</project>

.idea/workspace.xml

<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
  <component name="AutoImportSettings">
    <option name="autoReloadType" value="SELECTIVE" />
  </component>
  <component name="ChangeListManager">
    <list default="true" id="3df15837-a08d-4cd4-934f-7978ad5b076a" name="Changes" comment="" />
    <option name="SHOW_DIALOG" value="false" />
    <option name="HIGHLIGHT_CONFLICTS" value="true" />
    <option name="HIGHLIGHT_NON_ACTIVE_CHANGELIST" value="false" />
    <option name="LAST_RESOLUTION" value="IGNORE" />
  </component>
  <component name="MarkdownSettingsMigration">
    <option name="stateVersion" value="1" />
  </component>
  <component name="ProjectId" id="2LM98PiybFyoOanD9AfPLRS0ExL" />
  <component name="ProjectViewState">
    <option name="hideEmptyMiddlePackages" value="true" />
    <option name="showLibraryContents" value="true" />
  </component>
  <component name="PropertiesComponent"><![CDATA[{
  "keyToString": {
    "RunOnceActivity.OpenProjectViewOnStart": "true",
    "RunOnceActivity.ShowReadmeOnStart": "true",
    "WebServerToolWindowFactoryState": "false",
    "javascript.nodejs.core.library.configured.version": "16.16.0",
    "javascript.nodejs.core.library.typings.version": "16.18.12",
    "node.js.detected.package.eslint": "true",
    "node.js.detected.package.tslint": "true",
    "node.js.selected.package.eslint": "(autodetect)",
    "node.js.selected.package.tslint": "(autodetect)",
    "nodejs_package_manager_path": "npm",
    "ts.external.directory.path": "/Users/mnmkng/Library/Application Support/JetBrains/Toolbox/apps/WebStorm/ch-0/223.8617.44/WebStorm.app/Contents/plugins/javascript-impl/jsLanguageServicesImpl/external",
    "vue.rearranger.settings.migration": "true"
  }
}]]></component>
  <component name="RecentsManager">
    <key name="MoveFile.RECENT_KEYS">
      <recent name="$PROJECT_DIR$" />
    </key>
  </component>
  <component name="RunManager">
    <configuration name="main.js" type="NodeJSConfigurationType" temporary="true" nameIsGenerated="true" path-to-js-file="src/main.js" working-dir="$PROJECT_DIR$">
      <envs>
        <env name="APIFY_TOKEN" value="apify_api_iPPqEVeHrcqhYcFSFb4NXTnXqCixIm0SBRnJ" />
      </envs>
      <method v="2" />
    </configuration>
    <recent_temporary>
      <list>
        <item itemvalue="Node.js.main.js" />
      </list>
    </recent_temporary>
  </component>
  <component name="SpellCheckerSettings" RuntimeDictionaries="0" Folders="0" CustomDictionaries="0" DefaultDictionary="application-level" UseSingleDictionary="true" transferred="true" />
  <component name="TaskManager">
    <task active="true" id="Default" summary="Default task">
      <changelist id="3df15837-a08d-4cd4-934f-7978ad5b076a" name="Changes" comment="" />
      <created>1675669924902</created>
      <option name="number" value="Default" />
      <option name="presentableId" value="Default" />
      <updated>1675669924902</updated>
      <workItem from="1675669926072" duration="13837000" />
    </task>
    <servers />
  </component>
  <component name="TypeScriptGeneratedFilesManager">
    <option name="version" value="3" />
  </component>
  <component name="XDebuggerManager">
    <breakpoint-manager>
      <breakpoints>
        <line-breakpoint enabled="true" type="javascript">
          <url>file://$PROJECT_DIR$/src/utils.js</url>
          <line>19</line>
          <properties lambdaOrdinal="-1" />
          <option name="timeStamp" value="3" />
        </line-breakpoint>
        <line-breakpoint enabled="true" type="javascript">
          <url>file://$PROJECT_DIR$/src/utils.js</url>
          <line>69</line>
          <properties lambdaOrdinal="-1" />
          <option name="timeStamp" value="7" />
        </line-breakpoint>
        <line-breakpoint enabled="true" type="javascript">
          <url>file://$PROJECT_DIR$/src/utils.js</url>
          <line>74</line>
          <properties lambdaOrdinal="-1" />
          <option name="timeStamp" value="8" />
        </line-breakpoint>
      </breakpoints>
    </breakpoint-manager>
  </component>
</project>

src/main.js

import { Actor } from 'apify';
import { evalFunctionOrThrow, diff, fetchRuns, fetchTaskRuns, fetchActorRuns } from "./utils.js";

await Actor.init();
const client = Actor.newClient();

const { runIds, actorIds, taskIds, transformInputSource } = await Actor.getInput();

// PREP & VALIDATION
/**
 * @param {object} context
 * @param {object} context.input
 * @param {object} context.options
 */
const transformInput = evalFunctionOrThrow(transformInputSource);

if (!(runIds.length || actorIds.length || taskIds.length)) {
    throw new Error('Did not receive any run IDs, actor IDs or task IDs. Aborting.');
}

// FETCH RUNNING RUNS
const collectedRuns = [];

if (runIds.length) {
    const runs = await fetchRuns(client, runIds);
    console.log(`Fetched ${runs.length} runs based on their IDs.`);
    collectedRuns.push(...runs);
}

if (actorIds.length) {
    const runs = await fetchActorRuns(client, actorIds)
    collectedRuns.push(...runs);
}

if (taskIds.length) {
    const runs = await fetchTaskRuns(client, taskIds)
    collectedRuns.push(...runs);
}

// TRANSFORM, ABORT, RESURRECT
const processedRunPromises = collectedRuns.map(async (run) => {
    const runClient = client.run(run.id);
    const kvsClient = runClient.keyValueStore();

    console.log(`[${run.id}] Fetching input.`);
    const { value: input } = await kvsClient.getRecord('INPUT');
    const newInput = JSON.parse(JSON.stringify(input));
    // The run object contains the options in a different
    // format than the API expects them in resurrect.
    const apiOptions = {
        build: run.options.build,
        memory: run.options.memoryMbytes,
        timeout: run.options.timeoutSecs,
    }
    const newOptions = { ...apiOptions };

    console.log(`[${run.id}] Transforming input with transformInput function.`);
    transformInput({
        input: newInput,
        options: newOptions,
    })
    const inputDiff = diff(input, newInput);
    if (inputDiff) {
        console.log(`[${run.id}] Changed input:`, inputDiff);
    }
    const optionsDiff = diff(apiOptions, newOptions);
    if (optionsDiff) {
        console.log(`[${run.id}] Changed options:`, optionsDiff);
    }

    console.log(`[${run.id}] Aborting run.`)
    await runClient.abort({ gracefully: true });
    // Graceful abort can take some time.
    await runClient.waitForFinish();

    console.log(`[${run.id}] Saving new input to key-value store.`)
    await kvsClient.setRecord({
        key: 'INPUT',
        value: newInput,
    })

    console.log(`[${run.id}] Resurrecting run.`)
    const resurrectedRun = await runClient.resurrect(newOptions);

    console.log(`[${run.id}] Saving input and run information to dataset.`)
    await Actor.pushData({
        input,
        newInput,
        abortedRun: run,
        resurrectedRun,
    })
})

const processedRuns = await Promise.all(processedRunPromises);
console.log(`Aborted and resurrected ${processedRuns.length} runs. The end.`);
await Actor.exit();

src/utils.js

import vm from 'node:vm'

export const fetchRuns = async (apifyClient, runIds) => {
    const runPromises = runIds.map(async (runId) => {
        return apifyClient.run(runId).get();
    })
    const runs = await Promise.all(runPromises);
    return runs.filter(run => run.status === 'RUNNING');
}

export const fetchActorRuns = async (apifyClient, actorIds) => {
    const runPromises = actorIds.map(async (actorId) => {
        const { items } = await apifyClient.actor(actorId).runs().list({ status: 'RUNNING'});
        const runIds = items.map(run => run.id);
        const runs = await fetchRuns(apifyClient, runIds);
        console.log(`Fetched ${runs.length} running runs of actor: ${actorId}`);
        return runs;
    })
    const runsInArrays = await Promise.all(runPromises);
    return runsInArrays.flat();
}

export const fetchTaskRuns = async (apifyClient, taskIds) => {
    const runPromises = taskIds.map(async (taskId) => {
        const { items } = await apifyClient.task(taskId).runs().list({ status: 'RUNNING'});
        const runIds = items.map(run => run.id);
        const runs = await fetchRuns(apifyClient, runIds);
        console.log(`Fetched ${runs.length} running runs of actor: ${taskId}`);
        return runs;
    })
    const runsInArrays = await Promise.all(runPromises);
    return runsInArrays.flat();
}

/**
 * Transforms a page function string into a Function object.
 * @param {string} funcString
 * @return {Function}
 */
export const evalFunctionOrThrow = (funcString) => {
    let func;

    try {
        func = vm.runInThisContext(`(${funcString})`);
    } catch (err) {
        throw new Error(`Compilation of transformInput source code failed.\n${err.message}\n${err.stack.substr(err.stack.indexOf('\n'))}`);
    }

    if (typeof func !== 'function') throw new Error('`transformInput` function is not a function.');

    return func;
};

/**
 * Creates a diff between the first and second object.
 * @param {object} obj1
 * @param {object} obj2
 * @returns {object|undefined}
 */
export const diff = (obj1, obj2) => {
    const result = {};
    if (Object.is(obj1, obj2)) {
        return undefined;
    }
    if (!obj2 || typeof obj2 !== 'object') {
        return obj2;
    }
    Object.keys(obj1 || {}).concat(Object.keys(obj2 || {})).forEach(key => {
        if (obj2[key] !== obj1[key] && !Object.is(obj1[key], obj2[key])) {
            result[key] = obj2[key];
        }
        if (typeof obj2[key] === 'object' && typeof obj1[key] === 'object') {
            const value = diff(obj1[key], obj2[key]);
            if (value !== undefined) {
                result[key] = value;
            }
        }
    });
    // clean empty objects
    for (const [k, v] of Object.entries(result)) {
        if (!Object.keys(v).length) {
            delete result[k];
        }
    }
    return result;
}