Fix: Abort file processing when DB is not responsible (#289)

src/server/controllers/item/shared/item-data-processing.ts

@@ -38,6 +38,7 @@ import { extraIntervalMilliseconds } from "./extra-intervals-mapping"
 import { AnalyticsEvent } from "../../../utils/analytics/anyltics-event"
 import { downsampleData } from "../../../utils/lttb"
 import moment = require("moment");
+import { DataProcessingException } from "../../../errors/data-processing-exceptions"
 
 export const itemDataProcessing = async ({ projectName, scenarioName, itemId }) => {
     const MAX_LABEL_CHART_LENGTH = 100000
@@ -163,7 +164,7 @@ export const itemDataProcessing = async ({ projectName, scenarioName, itemId })
         }
 
     } catch(error) {
-        console.log(error)
-        throw new Error(`Error while processing dataId: ${itemId} for item: ${itemId}, error: ${error}`)
+        throw new DataProcessingException(
+            `Error while processing dataId: ${itemId} for item: ${itemId}, error: ${error}`)
     }
 }

src/server/controllers/item/shared/item-error-handler.spec.ts

@@ -0,0 +1,17 @@
+import { db } from "../../../../db/db"
+import { itemErrorHandler } from "./item-error-handler"
+
+jest.mock("../../../../db/db")
+
+describe("itemErrorHandler", () => {
+
+    beforeEach(() => {
+        jest.resetAllMocks()
+    })
+    it("should try set the item to error state", () => {
+        (db.none as any).mockResolvedValueOnce(null)
+        const spy = jest.spyOn(require("../../../queries/items"), "updateItem")
+        itemErrorHandler("item_id", Error("Test Error"))
+        expect(spy).toHaveBeenCalledTimes(1)
+    })
+})

src/server/controllers/item/shared/item-error-handler.ts

@@ -0,0 +1,33 @@
+import { DataStreamingToDatabaseException } from "../../../errors/data-streaming-to-database-exception"
+import { logger } from "../../../../logger"
+import { DataParsingException } from "../../../errors/data-parsing-exception"
+import { DataProcessingException } from "../../../errors/data-processing-exceptions"
+import { UnlinkingFileException } from "../../../errors/unlinking-file-exception"
+import { db } from "../../../../db/db"
+import { updateItem } from "../../../queries/items"
+import { ReportStatus } from "../../../queries/items.model"
+
+export const itemErrorHandler = async (itemId: string, processingError: Error) => {
+    if (processingError instanceof DataStreamingToDatabaseException) {
+        // eslint-disable-next-line max-len
+        logger.error(`Error occurred during data parsing and saving into database: ${processingError.message}, item_id: ${itemId}`)
+    } else if (processingError instanceof DataParsingException) {
+        // eslint-disable-next-line max-len
+        logger.error(`Error occurred during data parsing or manipulation: ${processingError.message}, item_id: ${itemId}`)
+    } else if (processingError instanceof DataProcessingException) {
+        // eslint-disable-next-line max-len
+        logger.error(`Error occurred during samples aggregation in database: ${processingError.message}, item_id: ${itemId}`)
+    } else if (processingError instanceof UnlinkingFileException) {
+        // eslint-disable-next-line max-len
+        logger.error(`Error occurred during deleting of uploaded file: ${processingError.message}, item_id: ${itemId}`)
+    } else {
+        logger.error(`Unexpected error occurred: ${processingError.message}, item_id: ${itemId}`)
+    }
+
+    try {
+        logger.info(`Trying to set item: ${itemId} to error state.`)
+        await db.none(updateItem(itemId, ReportStatus.Error, null))
+    } catch(e) {
+        logger.error(`It was not possible to set item ${itemId} to error state: ${e}`)
+    }
+}

src/server/controllers/item/stop-item-async-controller.ts

@@ -1,10 +1,11 @@
 import { Request, Response } from "express"
 import { db } from "../../../db/db"
 import { logger } from "../../../logger"
-import { updateItem, updateItemStatus } from "../../queries/items"
+import { updateItemStatus } from "../../queries/items"
 import { ReportStatus } from "../../queries/items.model"
 import { itemDataProcessing } from "./shared/item-data-processing"
 import { StatusCode } from "../../utils/status-code"
+import { itemErrorHandler } from "./shared/item-error-handler"
 
 export const stopItemAsyncController = async (req: Request, res: Response) => {
   const { projectName, scenarioName, itemId } = req.params
@@ -26,6 +27,6 @@ export const stopItemAsyncController = async (req: Request, res: Response) => {
     }
   } catch(e) {
     logger.error(`Processing of item ${itemId} failed ${e}`)
-    await db.none(updateItem(itemId, ReportStatus.Error, null))
+    await itemErrorHandler(itemId, e)
   }
 }

src/server/errors/data-parsing-exception.ts

@@ -0,0 +1,9 @@
+export class DataParsingException extends Error {
+    constructor (message) {
+        super(message)
+
+        this.name = this.constructor.name
+
+        Error.captureStackTrace(this, this.constructor)
+    }
+}

src/server/errors/data-processing-exceptions.ts

@@ -0,0 +1,9 @@
+export class DataProcessingException extends Error {
+    constructor (message) {
+        super(message)
+
+        this.name = this.constructor.name
+
+        Error.captureStackTrace(this, this.constructor)
+    }
+}

src/server/errors/data-streaming-to-database-exception.ts

@@ -0,0 +1,9 @@
+export class DataStreamingToDatabaseException extends Error {
+    constructor (message) {
+        super(message)
+
+        this.name = this.constructor.name
+
+        Error.captureStackTrace(this, this.constructor)
+    }
+}

src/server/errors/unlinking-file-exception.ts

@@ -0,0 +1,9 @@
+export class UnlinkingFileException extends Error {
+    constructor (message) {
+        super(message)
+
+        this.name = this.constructor.name
+
+        Error.captureStackTrace(this, this.constructor)
+    }
+}

src/server/queries/items.ts

@@ -1,6 +1,9 @@
 /* eslint-disable max-lines */
 // eslint-disable-next-line max-len
 
+import * as pgp from "pg-promise"
+const pg = pgp()
+
 // eslint-disable-next-line max-len
 export const createNewItem = (scenarioName, startTime, environment, note, status, projectName, hostname, reportStatus, name, resourcesLink) => {
     return {
@@ -644,3 +647,61 @@ WHERE row_n <= 5;`,
         values: [itemId],
     }
 }
+
+
+export const samplesColumnSet = new pg.helpers.ColumnSet([
+    "elapsed", "success", "bytes", "label",
+    {
+        name: "timestamp",
+        prop: "timeStamp",
+    },
+    {
+        name: "sent_bytes",
+        prop: "sentBytes",
+    },
+    {
+        name: "connect",
+        prop: "Connect",
+    }, {
+        name: "hostname",
+        prop: "Hostname",
+        def: null,
+    }, {
+        name: "status_code",
+        prop: "responseCode",
+    },
+    {
+        name: "all_threads",
+        prop: "allThreads",
+    },
+    {
+        name: "grp_threads",
+        prop: "grpThreads",
+    }, {
+        name: "latency",
+        prop: "Latency",
+    },
+    {
+        name: "response_message",
+        prop: "responseMessage",
+    },
+    {
+        name: "item_id",
+        prop: "itemId",
+    },
+    {
+        name: "sut_hostname",
+        prop: "sutHostname",
+        def: null,
+    },
+    {
+        name: "failure_message",
+        prop: "failureMessage",
+        def: null,
+    },
+    {
+        name: "thread_name",
+        prop: "threadName",
+        def: null,
+    },
+], { table: new pg.helpers.TableName({ table: "samples", schema: "jtl" }) })