Moving docx2text and pptx2text to a shared workerpool (#6087)

connectors/package.json

@@ -66,6 +66,7 @@
     "tsconfig-paths-webpack-plugin": "^4.1.0",
     "turndown": "^7.1.2",
     "uuid": "^9.0.0",
+    "workerpool": "^9.1.3",
     "yargs": "^17.7.2"
   },
   "devDependencies": {

connectors/src/connectors/google_drive/temporal/file.ts

@@ -25,7 +25,7 @@ import {
   sectionLength,
   upsertToDatasource,
 } from "@connectors/lib/data_sources";
-import { docx2text } from "@connectors/lib/docx2text.";
+import { docx2text } from "@connectors/lib/docx2text";
 import { dpdf2text } from "@connectors/lib/dpdf2text";
 import {
   GoogleDriveConfig,

connectors/src/lib/docx2text..ts → connectors/src/lib/docx2text.ts

@@ -2,7 +2,9 @@ import tracer from "dd-trace";
 import mammoth from "mammoth";
 import turndown from "turndown";
 
-export async function docx2text(fileContent: Buffer, filename: string) {
+import { getWorkerPool } from "@connectors/lib/workerpool";
+
+async function _docx2text(fileContent: Buffer, filename: string) {
   return tracer.trace(
     `gdrive`,
     {
@@ -23,3 +25,7 @@ export async function docx2text(fileContent: Buffer, filename: string) {
     }
   );
 }
+
+export async function docx2text(fileContent: Buffer, filename: string) {
+  return getWorkerPool().exec(_docx2text, [fileContent, filename]);
+}

connectors/src/lib/pptx2text.ts

@@ -2,13 +2,15 @@ import tracer from "dd-trace";
 import JSZip from "jszip";
 import turndown from "turndown";
 
+import { getWorkerPool } from "@connectors/lib/workerpool";
+
 type PPTXDocument = {
   pages: {
     content: string;
   }[];
 };
 
-export async function PPTX2Text(
+async function _PPTX2Text(
   fileBuffer: Buffer,
   filename?: string
 ): Promise<PPTXDocument> {
@@ -70,3 +72,10 @@ export async function PPTX2Text(
     }
   );
 }
+
+export async function PPTX2Text(
+  fileBuffer: Buffer,
+  filename?: string
+): Promise<PPTXDocument> {
+  return getWorkerPool().exec(_PPTX2Text, [fileBuffer, filename]);
+}

connectors/src/lib/workerpool.ts

@@ -0,0 +1,9 @@
+import workerpool from "workerpool";
+
+let POOL: ReturnType<typeof workerpool.pool> | null = null;
+export function getWorkerPool() {
+  if (!POOL) {
+    POOL = workerpool.pool({ maxWorkers: 3 });
+  }
+  return POOL;
+}

k8s/deployments/connectors-worker-google-drive-deployment.yaml

@@ -47,12 +47,12 @@ spec:
 
           resources:
             requests:
-              cpu: 2000m
+              cpu: 3000m
               memory: 4Gi
               ephemeral-storage: 4Gi
 
             limits:
-              cpu: 2000m
+              cpu: 3000m
               memory: 4Gi
               ephemeral-storage: 4Gi