feat: Add redirected actor logs

package.json

@@ -43,10 +43,6 @@
         "url": "https://github.com/apify/apify-client-js/issues"
     },
     "homepage": "https://docs.apify.com/api/client/js/",
-    "files": [
-        "dist",
-        "!dist/*.tsbuildinfo"
-    ],
     "scripts": {
         "build": "npm run clean && npm run build:node && npm run build:browser",
         "postbuild": "gen-esm-wrapper dist/index.js dist/index.mjs",
@@ -67,6 +63,7 @@
         "@apify/utilities": "^2.18.0",
         "@crawlee/types": "^3.3.0",
         "agentkeepalive": "^4.2.1",
+        "ansi-colors": "^4.1.1",
         "async-retry": "^1.3.3",
         "axios": "^1.6.7",
         "content-type": "^1.0.5",

src/resource_clients/actor.ts

@@ -3,6 +3,7 @@ import ow from 'ow';
 
 import type { RUN_GENERAL_ACCESS } from '@apify/consts';
 import { ACT_JOB_STATUSES, ACTOR_PERMISSION_LEVEL, META_ORIGINS } from '@apify/consts';
+import { Log } from '@apify/log';
 
 import type { ApiClientSubResourceOptions } from '../base/api_client';
 import { ResourceClient } from '../base/resource_client';
@@ -139,18 +140,30 @@ export class ActorClient extends ResourceClient {
                 webhooks: ow.optional.array.ofType(ow.object),
                 maxItems: ow.optional.number.not.negative,
                 maxTotalChargeUsd: ow.optional.number.not.negative,
+                log: ow.optional.any(ow.null, ow.object.instanceOf(Log)),
                 restartOnError: ow.optional.boolean,
                 forcePermissionLevel: ow.optional.string.oneOf(Object.values(ACTOR_PERMISSION_LEVEL)),
             }),
         );
 
-        const { waitSecs, ...startOptions } = options;
+        const { waitSecs, log, ...startOptions } = options;
         const { id } = await this.start(input, startOptions);
 
         // Calling root client because we need access to top level API.
         // Creating a new instance of RunClient here would only allow
         // setting it up as a nested route under actor API.
-        return this.apifyClient.run(id).waitForFinish({ waitSecs });
+        const newRunClient = this.apifyClient.run(id);
+
+        const streamedLog = await newRunClient.getStreamedLog({ toLog: options?.log });
+        let actorRun: Promise<ActorRun> | undefined;
+        try {
+            streamedLog?.start();
+            actorRun = this.apifyClient.run(id).waitForFinish({ waitSecs });
+            return actorRun;
+        } finally {
+            await actorRun;
+            await streamedLog?.stop();
+        }
     }
 
     /**
@@ -426,6 +439,7 @@ export interface ActorStartOptions {
 
 export interface ActorCallOptions extends Omit<ActorStartOptions, 'waitForFinish'> {
     waitSecs?: number;
+    log?: Log | null;
 }
 
 export interface ActorRunListItem {

src/resource_clients/log.ts

@@ -1,5 +1,11 @@
+// eslint-disable-next-line max-classes-per-file
 import type { Readable } from 'node:stream';
 
+import c from 'ansi-colors';
+
+import type { Log } from '@apify/log';
+import { Logger, LogLevel } from '@apify/log';
+
 import type { ApifyApiError } from '../apify_api_error';
 import type { ApiClientSubResourceOptions } from '../base/api_client';
 import { ResourceClient } from '../base/resource_client';
@@ -20,11 +26,11 @@ export class LogClient extends ResourceClient {
     /**
      * https://docs.apify.com/api/v2#/reference/logs/log/get-log
      */
-    async get(): Promise<string | undefined> {
+    async get(options: LogOptions = {}): Promise<string | undefined> {
         const requestOpts: ApifyRequestConfig = {
             url: this._url(),
             method: 'GET',
-            params: this._params(),
+            params: this._params(options),
         };
 
         try {
@@ -41,9 +47,10 @@ export class LogClient extends ResourceClient {
      * Gets the log in a Readable stream format. Only works in Node.js.
      * https://docs.apify.com/api/v2#/reference/logs/log/get-log
      */
-    async stream(): Promise<Readable | undefined> {
+    async stream(options: LogOptions = {}): Promise<Readable | undefined> {
         const params = {
             stream: true,
+            raw: options.raw,
         };
 
         const requestOpts: ApifyRequestConfig = {
@@ -63,3 +70,215 @@ export class LogClient extends ResourceClient {
         return undefined;
     }
 }
+
+export interface LogOptions {
+    /** @default false */
+    raw?: boolean;
+}
+
+// Temp create it here and ask Martin where to put it
+
+const DEFAULT_OPTIONS = {
+    /** Whether to exclude timestamp of log redirection in redirected logs. */
+    skipTime: true,
+    /** Level of log redirection */
+    level: LogLevel.DEBUG,
+};
+
+/**
+ * Logger for redirected actor logs.
+ */
+export class LoggerActorRedirect extends Logger {
+    constructor(options = {}) {
+        super({ ...DEFAULT_OPTIONS, ...options });
+    }
+
+    override _log(level: LogLevel, message: string, data?: any, exception?: unknown, opts: Record<string, any> = {}) {
+        if (level > this.options.level) {
+            return;
+        }
+        if (data || exception) {
+            throw new Error('Redirect logger does not use other arguments than level and message');
+        }
+        let { prefix } = opts;
+        prefix = prefix ? `${prefix}` : '';
+
+        let maybeDate = '';
+        if (!this.options.skipTime) {
+            maybeDate = `${new Date().toISOString().replace('Z', '').replace('T', ' ')} `;
+        }
+
+        const line = `${c.gray(maybeDate)}${c.cyan(prefix)}${message || ''}`;
+
+        // All redirected logs are logged at info level to avid any console specific formating for non-info levels,
+        // which have already been applied once to the original log. (For example error stack traces etc.)
+        this._outputWithConsole(LogLevel.INFO, line);
+        return line;
+    }
+}
+
+/**
+ * Helper class for redirecting streamed Actor logs to another log.
+ */
+export class StreamedLog {
+    private toLog: Log;
+    private streamBuffer: Buffer[] = [];
+    private splitMarker = /(?:\n|^)(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)/g;
+    private relevancyTimeLimit: Date | null;
+
+    private logClient: LogClient;
+    private streamingTask: Promise<void> | null = null;
+    private stopLogging = false;
+
+    constructor(logClient: LogClient, toLog: Log, fromStart = true) {
+        this.toLog = toLog;
+        this.logClient = logClient;
+        this.relevancyTimeLimit = fromStart ? null : new Date();
+    }
+
+    /**
+     * Start log redirection.
+     */
+    public start(): void {
+        if (this.streamingTask) {
+            throw new Error('Streaming task already active');
+        }
+        this.stopLogging = false;
+        this.streamingTask = this.streamLog();
+    }
+
+    /**
+     * Stop log redirection.
+     */
+    public async stop(): Promise<void> {
+        if (!this.streamingTask) {
+            throw new Error('Streaming task is not active');
+        }
+        this.stopLogging = true;
+        try {
+            await this.streamingTask;
+        } catch (err) {
+            if (!(err instanceof Error && err.name === 'AbortError')) {
+                throw err;
+            }
+        } finally {
+            this.streamingTask = null;
+        }
+    }
+
+    /**
+     * Get log stream from response and redirect it to another log.
+     */
+    private async streamLog(): Promise<void> {
+        const twoBytesLimit = 0xc0;
+        const threeBytesLimit = 0xe0;
+        const fourBytesLimit = 0xf0;
+
+        const logStream = await this.logClient.stream({ raw: true });
+        if (!logStream) {
+            return;
+        }
+        let incompleteCharacter: Uint8Array = new Uint8Array();
+
+        for await (const chunk of logStream) {
+            // Handle possible leftover incomplete multibyte character from previous chunk
+            const chunkPrependedWithIncompleteCharacter = new Uint8Array(incompleteCharacter.length + chunk.length);
+            chunkPrependedWithIncompleteCharacter.set(incompleteCharacter, 0);
+            chunkPrependedWithIncompleteCharacter.set(chunk, incompleteCharacter.length);
+
+            // Extract possible incomplete multibyte character from the end of this chunk
+            if (
+                chunkPrependedWithIncompleteCharacter.length > 1 &&
+                chunkPrependedWithIncompleteCharacter[chunkPrependedWithIncompleteCharacter.length] >= twoBytesLimit
+            ) {
+                incompleteCharacter = chunkPrependedWithIncompleteCharacter.slice(
+                    chunkPrependedWithIncompleteCharacter.length - 1,
+                );
+            } else if (
+                chunkPrependedWithIncompleteCharacter.length > 2 &&
+                chunkPrependedWithIncompleteCharacter[chunkPrependedWithIncompleteCharacter.length] >= threeBytesLimit
+            ) {
+                incompleteCharacter = chunkPrependedWithIncompleteCharacter.slice(
+                    chunkPrependedWithIncompleteCharacter.length - 2,
+                );
+            } else if (
+                chunkPrependedWithIncompleteCharacter.length > 3 &&
+                chunkPrependedWithIncompleteCharacter[chunkPrependedWithIncompleteCharacter.length] >= fourBytesLimit
+            ) {
+                incompleteCharacter = chunkPrependedWithIncompleteCharacter.slice(
+                    chunkPrependedWithIncompleteCharacter.length - 3,
+                );
+            }
+
+            // Keep processing the new data until stopped
+            this.streamBuffer.push(Buffer.from(chunkPrependedWithIncompleteCharacter));
+            // Log data only if the chunk contains the marker as it indicates previous message is complete
+            if (this.splitMarker.test(chunkPrependedWithIncompleteCharacter.toString())) {
+                this.logBufferContent(false);
+            }
+            if (this.stopLogging) {
+                break;
+            }
+        }
+
+        // Process the remaining buffer
+        this.logBufferContent(true);
+    }
+
+    /**
+     * Parse the buffer and log complete messages.
+     */
+    private logBufferContent(includeLastPart = false): void {
+        const allParts = Buffer.concat(this.streamBuffer).toString().split(this.splitMarker).slice(1);
+        let messageMarkers;
+        let messageContents;
+        // Parse the buffer parts into complete messages
+        if (includeLastPart) {
+            // This is final call, so log everything. Do not keep anything in the buffer.
+            messageMarkers = allParts.filter((_, i) => i % 2 === 0);
+            messageContents = allParts.filter((_, i) => i % 2 !== 0);
+            this.streamBuffer = [];
+        } else {
+            messageMarkers = allParts.filter((_, i) => i % 2 === 0).slice(0, -1);
+            messageContents = allParts.filter((_, i) => i % 2 !== 0).slice(0, -1);
+
+            // The last two parts (marker and message) are possibly not complete and will be left in the buffer.
+            this.streamBuffer = [Buffer.from(allParts.slice(-2).join(''))];
+        }
+
+        messageMarkers.forEach((marker, index) => {
+            const decodedMarker = marker;
+            const decodedContent = messageContents[index];
+            if (this.relevancyTimeLimit) {
+                // Log only relevant messages. Ignore too old log messages.
+                const logTime = new Date(decodedMarker);
+                if (logTime < this.relevancyTimeLimit) {
+                    return;
+                }
+            }
+            const message = decodedMarker + decodedContent;
+
+            // Log parsed message at guessed level.
+            this.logAtGuessedLevel(message);
+        });
+    }
+
+    /**
+     * Log messages at appropriate log level guessed from the message content.
+     *
+     * Original log level information does not have to be included in the message at all.
+     * This is methods just guesses, exotic formating or specific keywords can break the guessing logic.
+     */
+    private logAtGuessedLevel(message: string) {
+        message = message.trim();
+
+        if (message.includes('ERROR')) this.toLog.error(message);
+        else if (message.includes('SOFT_FAIL')) this.toLog.softFail(message);
+        else if (message.includes('WARNING')) this.toLog.warning(message);
+        else if (message.includes('INFO')) this.toLog.info(message);
+        else if (message.includes('DEBUG')) this.toLog.debug(message);
+        else if (message.includes('PERF')) this.toLog.perf(message);
+        // Fallback in case original log message does not indicate known log level.
+        else this.toLog.info(message);
+    }
+}

src/resource_clients/run.ts

@@ -2,15 +2,16 @@ import type { AxiosRequestConfig } from 'axios';
 import ow from 'ow';
 
 import type { RUN_GENERAL_ACCESS } from '@apify/consts';
+import { LEVELS, Log } from '@apify/log';
 
 import type { ApiClientOptionsWithOptionalResourcePath } from '../base/api_client';
 import { ResourceClient } from '../base/resource_client';
 import type { ApifyResponse } from '../http_client';
-import { cast, parseDateFields, pluckData } from '../utils';
+import { cast, isNode, parseDateFields, pluckData } from '../utils';
 import type { ActorRun } from './actor';
 import { DatasetClient } from './dataset';
 import { KeyValueStoreClient } from './key_value_store';
-import { LogClient } from './log';
+import { LogClient, LoggerActorRedirect, StreamedLog } from './log';
 import { RequestQueueClient } from './request_queue';
 
 const RUN_CHARGE_IDEMPOTENCY_HEADER = 'idempotency-key';
@@ -266,6 +267,39 @@ export class RunClient extends ResourceClient {
             }),
         );
     }
+
+    /**
+     * Get StreamedLog for convenient streaming of the run log and their redirection.
+     */
+    async getStreamedLog(options: GetStreamedLogOptions = {}): Promise<StreamedLog | undefined> {
+        const { fromStart = true } = options;
+        let { toLog } = options;
+        if (toLog === null || !isNode()) {
+            // Explicitly no logging or not in Node.js
+            return undefined;
+        }
+        if (toLog === undefined) {
+            // Create default StreamedLog
+            // Get actor name and run id
+            const runData = await this.get();
+            const runId = runData ? `${runData.id ?? ''}` : '';
+
+            const actorId = runData?.actId ?? '';
+            const actorData = (await this.apifyClient.actor(actorId).get()) || { name: '' };
+
+            const actorName = runData ? (actorData.name ?? '') : '';
+            const name = [actorName, `runId:${runId}`].filter(Boolean).join(' ');
+
+            toLog = new Log({ level: LEVELS.DEBUG, prefix: `${name} -> `, logger: new LoggerActorRedirect() });
+        }
+
+        return new StreamedLog(this.log(), toLog, fromStart);
+    }
+}
+
+export interface GetStreamedLogOptions {
+    toLog?: Log | null;
+    fromStart?: boolean;
 }
 
 export interface RunGetOptions {

test/_helper.js

@@ -1,6 +1,6 @@
 const { launchPuppeteer, puppeteerUtils } = require('@crawlee/puppeteer');
 
-const mockServer = require('./mock_server/server');
+const { mockServer } = require('./mock_server/server');
 
 class Browser {
     async start() {