[TD-7352] - Add better logging for flushing and uploading failures

packages/integration-sdk-runtime/src/execution/__tests__/jobState.test.ts

@@ -25,6 +25,7 @@ import {
 } from '../uploader';
 import { FlushedGraphObjectData } from '../../storage/types';
 import pMap from 'p-map';
+import { createMockIntegrationLogger } from '../../../test/util/fixtures';
 
 jest.mock('fs');
 
@@ -35,11 +36,14 @@ function entitiesToEntityKeySet(entities: Entity[]): Set<string> {
 }
 
 function createInMemoryStepGraphObjectDataUploaderCollector(
-  partial?: CreateQueuedStepGraphObjectDataUploaderParams,
+  partial?: Partial<CreateQueuedStepGraphObjectDataUploaderParams>,
 ) {
   const graphObjectDataCollection: FlushedGraphObjectData[] = [];
 
+  const logger = createMockIntegrationLogger();
+
   const uploader = createQueuedStepGraphObjectDataUploader({
+    logger,
     stepId: uuid(),
     uploadConcurrency: 5,
     upload(graphObjectData) {
@@ -307,8 +311,10 @@ describe('upload callbacks', () => {
 
   test('#waitUntilUploadsComplete should resolve when all uploads completed', async () => {
     const graphObjectDataCollection: FlushedGraphObjectData[] = [];
+    const logger = createMockIntegrationLogger();
 
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId: uuid(),
       uploadConcurrency: 5,
       async upload(graphObjectData) {

packages/integration-sdk-runtime/src/execution/executeIntegration.ts

@@ -250,7 +250,9 @@ export async function executeWithContext<
       }
 
       const {
-        graphObjectStore = new FileSystemGraphObjectStore(),
+        graphObjectStore = new FileSystemGraphObjectStore({
+          logger,
+        }),
         createStepGraphObjectDataUploader,
         resultsCallback,
       } = options;

packages/integration-sdk-runtime/src/execution/uploader.test.ts

@@ -52,7 +52,10 @@ describe('#createQueuedStepGraphObjectDataUploader', () => {
     const uploaded: FlushedGraphObjectData[] = [];
     let numQueued = 0;
 
+    const logger = createMockIntegrationLogger();
+
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId: uuid(),
       uploadConcurrency: Infinity,
       async upload(d) {
@@ -77,7 +80,10 @@ describe('#createQueuedStepGraphObjectDataUploader', () => {
     const uploaded: FlushedGraphObjectData[] = [];
     let throttleCount = 0;
 
+    const logger = createMockIntegrationLogger();
+
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId: uuid(),
       uploadConcurrency: 2,
       async upload(d) {
@@ -101,7 +107,10 @@ describe('#createQueuedStepGraphObjectDataUploader', () => {
 
     let numQueued = 0;
 
+    const logger = createMockIntegrationLogger();
+
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId,
       uploadConcurrency: 2,
       async upload(d) {
@@ -143,7 +152,10 @@ describe('#createQueuedStepGraphObjectDataUploader', () => {
 
     let numQueued = 0;
 
+    const logger = createMockIntegrationLogger();
+
     const uploader = createQueuedStepGraphObjectDataUploader({
+      logger,
       stepId,
       uploadConcurrency: 2,
       async upload(d) {

packages/integration-sdk-runtime/src/execution/uploader.ts

@@ -1,4 +1,7 @@
-import { UploadError } from '@jupiterone/integration-sdk-core';
+import {
+  IntegrationLogger,
+  UploadError,
+} from '@jupiterone/integration-sdk-core';
 import PQueue from 'p-queue/dist';
 import { FlushedGraphObjectData } from '../storage/types';
 import {
@@ -19,13 +22,15 @@ export type CreateStepGraphObjectDataUploaderFunction = (
 ) => StepGraphObjectDataUploader;
 
 export interface CreateQueuedStepGraphObjectDataUploaderParams {
+  logger: IntegrationLogger;
   stepId: string;
   uploadConcurrency: number;
   upload: (graphObjectData: FlushedGraphObjectData) => Promise<void>;
   onThrottleEnqueue?: () => void;
 }
 
 export function createQueuedStepGraphObjectDataUploader({
+  logger,
   stepId,
   uploadConcurrency: maximumQueueSize,
   upload,
@@ -69,6 +74,10 @@ export function createQueuedStepGraphObjectDataUploader({
       queue
         .add(() => upload(graphObjectData))
         .catch((err) => {
+          logger.warn(
+            { err, stepId, graphObjectData },
+            'Error uploading graph object data batch',
+          );
           // Do not pause the queue entirely. We will try to prevent additional
           // tasks from being added to the queue, but even if an error occurs,
           // we should try uploading the remaining data that we have queued up.
@@ -100,6 +109,15 @@ export function createQueuedStepGraphObjectDataUploader({
         // this time, we could be receiving additional tasks in our queue that
         // will grow the queue.
         completed = true;
+
+        logger.debug(
+          {
+            stepId,
+            uploadErrorCount: uploadErrors.length,
+            typesInvolvedInFailures: Array.from(typesInvolvedInFailures),
+          },
+          'Upload queue processing complete',
+        );
       }
 
       if (uploadErrors.length) {
@@ -150,6 +168,7 @@ export function createPersisterApiStepGraphObjectDataUploader({
   uploadBatchSizeInBytes = DEFAULT_UPLOAD_BATCH_SIZE_IN_BYTES,
 }: CreatePersisterApiStepGraphObjectDataUploaderParams) {
   return createQueuedStepGraphObjectDataUploader({
+    logger: synchronizationJobContext.logger,
     stepId,
     uploadConcurrency,
     async upload(graphObjectData) {

packages/integration-sdk-runtime/src/storage/FileSystemGraphObjectStore/FileSystemGraphObjectStore.ts

@@ -10,6 +10,7 @@ import {
   GetIndexMetadataForGraphObjectTypeParams,
   IntegrationStep,
   GraphObjectIterateeOptions,
+  IntegrationLogger,
 } from '@jupiterone/integration-sdk-core';
 
 import { flushDataToDisk } from './flushDataToDisk';
@@ -52,6 +53,11 @@ export interface FileSystemGraphObjectStoreParams {
    * Whether the files that are written to disk should be minified or not
    */
   prettifyFiles?: boolean;
+
+  /**
+   * Optional logger for debugging and tracking data flow
+   */
+  logger?: IntegrationLogger;
 }
 
 interface GraphObjectIndexMetadataMap {
@@ -135,13 +141,15 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
     string,
     GraphObjectLocationOnDisk
   >(ENTITY_LOCATION_ON_DISK_DEFAULT_MAP_KEY_SPACE);
+  private readonly logger?: IntegrationLogger;
 
   constructor(params?: FileSystemGraphObjectStoreParams) {
     this.semaphore = new Sema(BINARY_SEMAPHORE_CONCURRENCY);
     this.graphObjectFileSize =
       params?.graphObjectFileSize || DEFAULT_GRAPH_OBJECT_FILE_SIZE;
 
     this.prettifyFiles = params?.prettifyFiles || false;
+    this.logger = params?.logger;
     this.graphObjectBufferThresholdInBytes = min([
       params?.graphObjectBufferThresholdInBytes ||
         DEFAULT_UPLOAD_BATCH_SIZE_IN_BYTES,
@@ -325,36 +333,83 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
         });
 
         if (indexable.length) {
-          await Promise.all(
-            chunk(indexable, this.graphObjectFileSize).map(async (data) => {
-              const graphObjectsToFilePaths = await flushDataToDisk({
-                storageDirectoryPath: stepId,
-                collectionType: 'entities',
-                data,
-                pretty: this.prettifyFiles,
-              });
-
-              for (const {
-                graphDataPath,
-                collection,
-              } of graphObjectsToFilePaths) {
-                for (const [index, e] of collection.entries()) {
-                  this.entityOnDiskLocationMap.set(e._key, {
-                    graphDataPath,
-                    index,
-                  });
-                }
-              }
-            }),
+          const chunks = chunk(indexable, this.graphObjectFileSize);
+          this.logger?.debug(
+            {
+              stepId,
+              entityCount: indexable.length,
+              chunkCount: chunks.length,
+              chunkSize: this.graphObjectFileSize,
+            },
+            'Flushing entity chunks to disk',
           );
+
+          try {
+            await Promise.all(
+              chunks.map(async (data, chunkIndex) => {
+                const graphObjectsToFilePaths = await flushDataToDisk({
+                  storageDirectoryPath: stepId,
+                  collectionType: 'entities',
+                  data,
+                  pretty: this.prettifyFiles,
+                  logger: this.logger,
+                });
+
+                for (const {
+                  graphDataPath,
+                  collection,
+                } of graphObjectsToFilePaths) {
+                  for (const [index, e] of collection.entries()) {
+                    this.entityOnDiskLocationMap.set(e._key, {
+                      graphDataPath,
+                      index,
+                    });
+                  }
+                }
+
+                this.logger?.debug(
+                  {
+                    stepId,
+                    chunkIndex,
+                    entitiesInChunk: data.length,
+                    filesCreated: graphObjectsToFilePaths.length,
+                  },
+                  'Entity chunk flushed successfully',
+                );
+              }),
+            );
+          } catch (error) {
+            this.logger?.error(
+              {
+                stepId,
+                entityCount: indexable.length,
+                chunkCount: chunks.length,
+                error: error.message,
+                errorStack: error.stack,
+              },
+              'Failed to flush entity chunks to disk',
+            );
+            throw error;
+          }
         }
 
         this.localGraphObjectStore.flushEntities(entities, stepId);
         entitiesToUpload = entitiesToUpload.concat(entities);
       }
 
       if (onEntitiesFlushed) {
-        await onEntitiesFlushed(entitiesToUpload);
+        try {
+          await onEntitiesFlushed(entitiesToUpload);
+        } catch (err) {
+          this.logger?.error(
+            {
+              entityCount: entitiesToUpload.length,
+              err,
+            },
+            'onEntitiesFlushed callback failed',
+          );
+          throw err;
+        }
       }
     });
   }
@@ -406,24 +461,71 @@ export class FileSystemGraphObjectStore implements GraphObjectStore {
         });
 
         if (indexable.length) {
-          await Promise.all(
-            chunk(indexable, this.graphObjectFileSize).map(async (data) => {
-              await flushDataToDisk({
-                storageDirectoryPath: stepId,
-                collectionType: 'relationships',
-                data,
-                pretty: this.prettifyFiles,
-              });
-            }),
+          const chunks = chunk(indexable, this.graphObjectFileSize);
+          this.logger?.debug(
+            {
+              stepId,
+              relationshipCount: indexable.length,
+              chunkCount: chunks.length,
+              chunkSize: this.graphObjectFileSize,
+            },
+            'Flushing relationship chunks to disk',
           );
+
+          try {
+            await Promise.all(
+              chunks.map(async (data, chunkIndex) => {
+                await flushDataToDisk({
+                  storageDirectoryPath: stepId,
+                  collectionType: 'relationships',
+                  data,
+                  pretty: this.prettifyFiles,
+                  logger: this.logger,
+                });
+
+                this.logger?.debug(
+                  {
+                    stepId,
+                    chunkIndex,
+                    relationshipsInChunk: data.length,
+                  },
+                  'Relationship chunk flushed successfully',
+                );
+              }),
+            );
+          } catch (error) {
+            this.logger?.error(
+              {
+                stepId,
+                relationshipCount: indexable.length,
+                chunkCount: chunks.length,
+                error: error.message,
+                errorStack: error.stack,
+              },
+              'Failed to flush relationship chunks to disk',
+            );
+            throw error;
+          }
         }
 
         this.localGraphObjectStore.flushRelationships(relationships, stepId);
         relationshipsToUpload = relationshipsToUpload.concat(relationships);
       }
 
       if (onRelationshipsFlushed) {
-        await onRelationshipsFlushed(relationshipsToUpload);
+        try {
+          await onRelationshipsFlushed(relationshipsToUpload);
+        } catch (error) {
+          this.logger?.error(
+            {
+              relationshipCount: relationshipsToUpload.length,
+              error: error.message,
+              errorStack: error.stack,
+            },
+            'onRelationshipsFlushed callback failed',
+          );
+          throw error;
+        }
       }
     });
   }