run sequentially logs without emitting event

Author: eduardomourar

package-lock.json

@@ -34,6 +34,7 @@
     "homepage": "https://github.com/eduardomourar/cloudformation-cli-typescript-plugin#readme",
     "dependencies": {
         "autobind-decorator": "^2.4.0",
+        "promise-sequential": "^1.1.1",
         "reflect-metadata": "^0.1.13",
         "tombok": "https://github.com/eduardomourar/tombok/releases/download/v0.0.1/tombok-0.0.1.tgz",
         "uuid": "^7.0.2"
@@ -42,6 +43,7 @@
         "@types/aws-sdk": "^2.7.0",
         "@types/jest": "^25.2.1",
         "@types/node": "^12.0.0",
+        "@types/promise-sequential": "^1.1.0",
         "@types/uuid": "^7.0.0",
         "@typescript-eslint/eslint-plugin": "^2.29.0",
         "@typescript-eslint/parser": "^2.19.2",
@@ -54,12 +56,12 @@
         "eslint-plugin-prettier": "^3.1.3",
         "jest": "^25.5.2",
         "minimist": ">=1.2.5",
-        "prettier": "2.0.5",
+        "prettier": "^2.0.5",
         "ts-jest": "^25.4.0",
         "typescript": "^3.7.0"
     },
     "optionalDependencies": {
-        "aws-sdk": "^2.656.0"
+        "aws-sdk": "~2.631.0"
     },
     "prettier": {
         "parser": "typescript",

package.json

@@ -37,6 +37,7 @@ export async function reportProgress(options: ProgressOptions): Promise<void> {
     }
     if (currentOperationStatus) {
         request.CurrentOperationStatus = currentOperationStatus;
+        LOGGER.debug('Record Handler Progress Request:', request);
         const response: { [key: string]: any } = await client
             .recordHandlerProgress(request)
             .promise();
@@ -45,8 +46,8 @@ export async function reportProgress(options: ProgressOptions): Promise<void> {
             requestId = response.ResponseMetadata.RequestId;
         }
         LOGGER.debug(
-            `Record Handler Progress with Request Id ${requestId} and Request:`,
-            request
+            `Record handler progress with Request Id ${requestId} and response:`,
+            response
         );
     }
 }

src/callback.ts

@@ -1,16 +1,20 @@
 import { boundMethod } from 'autobind-decorator';
 import { EventEmitter } from 'events';
 import CloudWatchLogs, {
+    DescribeLogStreamsResponse,
+    LogStream,
     InputLogEvent,
     PutLogEventsRequest,
     PutLogEventsResponse,
 } from 'aws-sdk/clients/cloudwatchlogs';
 import S3, { PutObjectRequest, PutObjectOutput } from 'aws-sdk/clients/s3';
+import promiseSequential from 'promise-sequential';
 
 import { SessionProxy } from './proxy';
-import { HandlerRequest, runInSequence } from './utils';
+import { delay, HandlerRequest } from './utils';
 
 type Console = globalThis.Console;
+type PromiseFunction = () => Promise<any>;
 
 interface LogOptions {
     groupName: string;
@@ -26,13 +30,14 @@ export class ProviderLogHandler {
     private static instance: ProviderLogHandler;
     public emitter: LogEmitter;
     public client: CloudWatchLogs;
-    public sequenceToken = '';
+    public sequenceToken: string = null;
     public accountId: string;
     public groupName: string;
     public stream: string;
     public logger: Console;
     public clientS3: S3;
-    private stack: Array<Promise<any>> = [];
+    private stack: Array<PromiseFunction> = [];
+    private isProcessing = false;
 
     /**
      * The ProviderLogHandler's constructor should always be private to prevent direct
@@ -50,28 +55,48 @@ export class ProviderLogHandler {
         const logger = options.logger || global.console;
         this.logger = logger;
         this.emitter.on('log', (...args: any[]) => {
-            this.stack.push(this.deliverLog(args));
+            // this.logger.debug('Emitting log event...');
         });
         // Create maps of each logger method and then alias that.
         Object.entries(this.logger).forEach(([key, val]) => {
             if (typeof val === 'function') {
                 if (['log', 'error', 'warn', 'info'].includes(key)) {
-                    this.logger[key as 'log' | 'error' | 'warn' | 'info'] = function (
+                    this.logger[key as 'log' | 'error' | 'warn' | 'info'] = (
                         ...args: any[]
-                    ): void {
-                        // For adding other event watchers later.
-                        setImmediate(() => emitter.emit('log', ...args));
+                    ): void => {
+                        if (!this.isProcessing) {
+                            const logLevel = key.toUpperCase();
+                            // Add log level when not present
+                            if (
+                                args.length &&
+                                args[0].substring(0, logLevel.length).toUpperCase() !==
+                                    logLevel
+                            ) {
+                                args.unshift(logLevel);
+                            }
+                            this.stack.push(() =>
+                                this.deliverLog(args).catch(this.logger.debug)
+                            );
+                            // For adding other event watchers later.
+                            setImmediate(() => {
+                                this.emitter.emit('log', ...args);
+                            });
+                        } else {
+                            this.logger.debug(
+                                'Logs are being delivered at the moment...'
+                            );
+                        }
 
                         // Calls the logger method.
-                        val.apply(this, args);
+                        val.apply(this.logger, args);
                     };
                 }
             }
         });
     }
 
     private async initialize(): Promise<void> {
-        this.sequenceToken = '';
+        this.sequenceToken = null;
         this.stack = [];
         try {
             await this.deliverLogCloudWatch(['Initialize CloudWatch']);
@@ -142,11 +167,13 @@ export class ProviderLogHandler {
 
     @boundMethod
     public async processLogs(): Promise<void> {
+        this.isProcessing = true;
         if (this.stack.length > 0) {
-            this.stack.push(this.deliverLog(['Log delivery finalized.']));
+            this.stack.push(() => this.deliverLog(['Log delivery finalized.']));
         }
-        await runInSequence(this.stack);
+        await promiseSequential(this.stack);
         this.stack = [];
+        this.isProcessing = false;
     }
 
     private async createLogGroup(): Promise<void> {
@@ -199,27 +226,48 @@ export class ProviderLogHandler {
             const response: PutLogEventsResponse = await this.client
                 .putLogEvents(logEventsParams)
                 .promise();
-            this.sequenceToken = response?.nextSequenceToken;
+            this.sequenceToken = response?.nextSequenceToken || null;
             this.logger.debug('Response from "putLogEvents"', response);
             return response;
         } catch (err) {
             const errorCode = err.code || err.name;
-            this.logger.debug('Error from "deliverLogCloudWatch"', err);
-            this.logger.debug(`Error from 'putLogEvents' ${JSON.stringify(err)}`);
+            this.logger.debug(
+                `Error from "putLogEvents" with sequence token ${this.sequenceToken}`,
+                err
+            );
             if (
                 errorCode === 'DataAlreadyAcceptedException' ||
                 errorCode === 'InvalidSequenceTokenException'
             ) {
-                this.sequenceToken = (err.message || '').split(' ').pop();
-                this.putLogEvents(record);
+                this.sequenceToken = null;
+                // Delay to avoid throttling
+                await delay(1);
+                try {
+                    const response: DescribeLogStreamsResponse = await this.client
+                        .describeLogStreams({
+                            logGroupName: this.groupName,
+                            logStreamNamePrefix: this.stream,
+                            limit: 1,
+                        })
+                        .promise();
+                    this.logger.debug('Response from "describeLogStreams"', response);
+                    if (response.logStreams && response.logStreams.length) {
+                        const logStream = response.logStreams[0] as LogStream;
+                        this.sequenceToken = logStream.uploadSequenceToken;
+                    }
+                } catch (err) {
+                    this.logger.debug('Error from "describeLogStreams"', err);
+                }
             } else {
                 throw err;
             }
         }
     }
 
     @boundMethod
-    private async deliverLogCloudWatch(messages: any[]): Promise<PutLogEventsResponse> {
+    private async deliverLogCloudWatch(
+        messages: any[]
+    ): Promise<PutLogEventsResponse | void> {
         const currentTime = new Date(Date.now());
         const record: InputLogEvent = {
             message: JSON.stringify({ messages }),
@@ -236,8 +284,20 @@ export class ProviderLogHandler {
                     await this.createLogGroup();
                 }
                 await this.createLogStream();
-                return this.putLogEvents(record);
-            } else {
+            } else if (
+                errorCode !== 'DataAlreadyAcceptedException' &&
+                errorCode !== 'InvalidSequenceTokenException'
+            ) {
+                throw err;
+            }
+            try {
+                const response = await this.putLogEvents(record);
+                return response;
+            } catch (err) {
+                // Additional retry for sequence token error
+                if (this.sequenceToken) {
+                    return this.putLogEvents(record);
+                }
                 throw err;
             }
         }
@@ -316,7 +376,7 @@ export class ProviderLogHandler {
     @boundMethod
     private async deliverLog(
         messages: any[]
-    ): Promise<PutLogEventsResponse | PutObjectOutput> {
+    ): Promise<PutLogEventsResponse | PutObjectOutput | void> {
         if (this.clientS3) {
             return this.deliverLogS3(messages);
         }

src/log-delivery.ts

@@ -122,6 +122,11 @@ export abstract class BaseResource<T extends BaseResourceModel = BaseResourceMod
         }
         // Modify requestContext in-place, so that invoke count is bumped on local
         // reinvoke too.
+        if (!handlerRequest.requestContext) {
+            handlerRequest.requestContext = {
+                invocation: 0,
+            } as RequestContext<Map<string, any>>;
+        }
         const reinvokeContext: RequestContext<Map<string, any>> =
             handlerRequest.requestContext;
         reinvokeContext.invocation = (reinvokeContext.invocation || 0) + 1;
@@ -241,11 +246,15 @@ export abstract class BaseResource<T extends BaseResourceModel = BaseResourceMod
                 callbackContext
             );
         } catch (err) {
+            if (!err.stack) {
+                Error.captureStackTrace(err);
+            }
+            err.stack = `${new Error().stack}\n${err.stack}`;
             if (err instanceof BaseHandlerException) {
-                LOGGER.error('Handler error');
+                LOGGER.error(`Handler error: ${err.message}`, err);
                 progress = err.toProgressEvent();
             } else {
-                LOGGER.error('Exception caught');
+                LOGGER.error(`Exception caught: ${err.message}`, err);
                 msg = err.message || msg;
                 progress = ProgressEvent.failed(HandlerErrorCode.InternalFailure, msg);
             }
@@ -336,12 +345,11 @@ export abstract class BaseResource<T extends BaseResourceModel = BaseResourceMod
         let isLogSetup = false;
         let progress: ProgressEvent;
 
-        const printOrLog = (message: string): void => {
+        const printOrLog = (...args: any[]): void => {
             if (isLogSetup) {
-                LOGGER.error(message);
+                LOGGER.error(...args);
             } else {
-                console.log(message);
-                console.trace();
+                console.log(...args);
             }
         };
 
@@ -351,7 +359,7 @@ export abstract class BaseResource<T extends BaseResourceModel = BaseResourceMod
             );
             const [callerSession, platformSession, providerSession] = sessions;
             isLogSetup = await ProviderLogHandler.setup(event, providerSession);
-
+            // LOGGER.debug('entrypoint eventData', eventData.toObject());
             const request = this.castResourceRequest(event);
 
             const metrics = new MetricsPublisherProxy(
@@ -435,11 +443,15 @@ export abstract class BaseResource<T extends BaseResourceModel = BaseResourceMod
                 );
             }
         } catch (err) {
+            if (!err.stack) {
+                Error.captureStackTrace(err);
+            }
+            err.stack = `${new Error().stack}\n${err.stack}`;
             if (err instanceof BaseHandlerException) {
-                printOrLog('Handler error');
+                printOrLog(`Handler error: ${err.message}`, err);
                 progress = err.toProgressEvent();
             } else {
-                printOrLog('Exception caught');
+                printOrLog(`Exception caught: ${err.message}`, err);
                 progress = ProgressEvent.failed(
                     HandlerErrorCode.InternalFailure,
                     err.message

src/resource.ts

@@ -35,19 +35,6 @@ export async function delay(seconds: number): Promise<void> {
     return new Promise((_) => setTimeout(() => _(), seconds * 1000));
 }
 
-/**
- * Execute promises in sequence (not in parallel as Promise.all)
- *
- * @param functions Array of functions that return a promise
- */
-export async function runInSequence(functions: Array<Promise<any>>): Promise<any[]> {
-    const results = [];
-    for (const fn of functions) {
-        results.push(await fn);
-    }
-    return results;
-}
-
 @allArgsConstructor
 export class TestEvent {
     credentials: Credentials;