feat: option to persist state using valkey store (#8)

Author: birme

package-lock.json

@@ -27,13 +27,15 @@
     "fastify": "5.3.2",
     "minio": "^8.0.5",
     "nodemon": "^2.0.20",
+    "redis": "^5.5.6",
     "ts-node": "^10.9.1"
   },
   "devDependencies": {
     "@commitlint/cli": "^17.4.2",
     "@commitlint/config-conventional": "^17.4.2",
     "@types/jest": "^29.4.0",
     "@types/node": "^18.11.18",
+    "@types/redis": "^4.0.10",
     "@typescript-eslint/eslint-plugin": "^5.51.0",
     "@typescript-eslint/parser": "^5.51.0",
     "eslint": "^8.33.0",

package.json

@@ -163,6 +163,17 @@ ENCORE_URL=<svtencore-url>
 SUBTITLE_GENERATOR_URL=<subtitle-generator-url>
 ```
 
+To ensure that the orchestrator maintains state during a restart you can provide a URL to a Redis or Valkey store. Create the Valkey for this purpose first.
+
+```bash
+% npx -y @osaas/cli db create valkey orchestrator
+redis://<orchestrator-valkey-ip>:<orchestrator-valkey-port>
+```
+
+```
+REDIS_URL=redis://<orchestrator-valkey-ip>:<orchestrator-valkey-port>
+```
+
 For deployment of this repository you need to first create a GitHub personal access token. Follow the instructions in this [guide](https://docs.osaas.io/osaas.wiki/Service%3A-Web-Runner.html) to create it.
 
 Navigate to the [Web Runner service](https://app.osaas.io/dashboard/service/eyevinn-web-runner) in Open Source Cloud web console and create a service secret `ghtoken` to store the GitHub personal access token and another service secret `osctoken` for your OSC personal access token.

readme.md

@@ -29,10 +29,13 @@ export interface OrchestratorOptions {
   s3AccessKeyId: string;
   s3SecretAccessKey: string;
   api: FastifyInstance;
+  redisUrl?: string;
 }
 
 export default (opts: OrchestratorOptions) => {
-  const workOrderManager = new WorkOrderManager();
+  const workOrderManager = new WorkOrderManager({
+    redisUrl: opts.redisUrl ? new URL(opts.redisUrl) : undefined
+  });
   const ctx = new Context();
 
   const timer = setInterval(async () => {
@@ -54,23 +57,54 @@ export default (opts: OrchestratorOptions) => {
             ) {
               console.log(`[${workOrder.id}]: Starting task ${task.type}`);
               if (task.type === 'ABR_TRANSCODE') {
-                await startAbrTranscodeTask(ctx, task, workOrder, opts);
+                await startAbrTranscodeTask(
+                  ctx,
+                  task,
+                  workOrder,
+                  workOrderManager,
+                  opts
+                );
               } else if (task.type === 'VOD_PACKAGE') {
-                await startVodPackageTask(ctx, task, workOrder, opts);
+                await startVodPackageTask(
+                  ctx,
+                  task,
+                  workOrder,
+                  workOrderManager,
+                  opts
+                );
               } else if (task.type === 'TRANSCRIBE') {
-                await startTranscribeTask(ctx, task, workOrder, opts);
+                await startTranscribeTask(
+                  ctx,
+                  task,
+                  workOrder,
+                  workOrderManager,
+                  opts
+                );
               } else if (task.type === 'CLEANUP') {
-                await startCleanupTask(ctx, task, workOrder, opts);
+                await startCleanupTask(
+                  ctx,
+                  task,
+                  workOrder,
+                  workOrderManager,
+                  opts
+                );
               }
             } else if (task.status === 'IN_PROGRESS') {
               if (task.type === 'VOD_PACKAGE') {
-                await updateVodPackageTask(ctx, task, workOrder, opts);
+                await updateVodPackageTask(
+                  ctx,
+                  task,
+                  workOrder,
+                  workOrderManager,
+                  opts
+                );
               }
             }
           } catch (error) {
             console.error(
               `[${workOrder.id}]: Error processing task ${task.type} for work order: `,
-              error
+              error,
+              workOrder
             );
           }
         }

src/orchestrator.ts

@@ -1,12 +1,14 @@
 import { Context } from '@osaas/client-core';
-import { WorkOrder, WorkOrderTask } from '../workorder';
+import { WorkOrder, WorkOrderManager, WorkOrderTask } from '../workorder';
 import { OrchestratorOptions } from '../../orchestrator';
 import { transcode } from '@osaas/client-transcode';
+import { EncoreJob } from '../encore';
 
 export async function startAbrTranscodeTask(
   ctx: Context,
   task: WorkOrderTask,
   workOrder: WorkOrder,
+  workOrderManager: WorkOrderManager,
   opts: OrchestratorOptions
 ) {
   const encoreServiceAccessToken = await ctx.getServiceAccessToken('encore');
@@ -25,5 +27,10 @@ export async function startAbrTranscodeTask(
       bearerToken: encoreServiceAccessToken
     }
   );
-  task.status = 'IN_PROGRESS';
+  await workOrderManager.updateWorkOrderTask(
+    workOrder.id,
+    task.type,
+    'IN_PROGRESS',
+    job as EncoreJob
+  );
 }

src/orchestrator/tasks/abr_transcode.ts

@@ -1,12 +1,13 @@
 import { Context } from '@osaas/client-core';
-import { WorkOrder, WorkOrderTask } from '../workorder';
+import { WorkOrder, WorkOrderManager, WorkOrderTask } from '../workorder';
 import { OrchestratorOptions } from '../../orchestrator';
 import { removeDir, removeFile } from '../../storage/minio';
 
 export async function startCleanupTask(
   ctx: Context,
   task: WorkOrderTask,
   workOrder: WorkOrder,
+  workOrderManager: WorkOrderManager,
   opts: OrchestratorOptions
 ) {
   // Remove source file
@@ -32,5 +33,9 @@ export async function startCleanupTask(
   console.log(
     `[${workOrder.id}]: Removed ABR transcoded files and subtitles for work order`
   );
-  task.status = 'COMPLETED';
+  await workOrderManager.updateWorkOrderTask(
+    workOrder.id,
+    task.type,
+    'COMPLETED'
+  );
 }

src/orchestrator/tasks/cleanup.ts

@@ -1,12 +1,13 @@
 import { Context } from '@osaas/client-core';
-import { WorkOrder, WorkOrderTask } from '../workorder';
+import { WorkOrder, WorkOrderManager, WorkOrderTask } from '../workorder';
 import { OrchestratorOptions } from '../../orchestrator';
 import { fileExists, readTextFile } from '../../storage/minio';
 
 export async function startTranscribeTask(
   ctx: Context,
   task: WorkOrderTask,
   workOrder: WorkOrder,
+  workOrderManager: WorkOrderManager,
   opts: OrchestratorOptions
 ) {
   const sourceWithoutExtension = workOrder.source
@@ -67,4 +68,9 @@ export async function startTranscribeTask(
   const data = await response.json();
   console.log(`[${workOrder.id}]: Transcription started for work order:`, data);
   task.status = 'IN_PROGRESS';
+  await workOrderManager.updateWorkOrderTask(
+    workOrder.id,
+    task.type,
+    task.status
+  );
 }

src/orchestrator/tasks/transcribe.ts

@@ -4,14 +4,15 @@ import {
   EncoreJob,
   parseInputsFromEncoreJob
 } from '../encore';
-import { WorkOrder, WorkOrderTask } from '../workorder';
+import { WorkOrder, WorkOrderManager, WorkOrderTask } from '../workorder';
 import { OrchestratorOptions } from '../../orchestrator';
 import { ShakaJob } from '../shaka';
 
 export async function startVodPackageTask(
   ctx: Context,
   task: WorkOrderTask,
   workOrder: WorkOrder,
+  workOrderManager: WorkOrderManager,
   opts: OrchestratorOptions
 ) {
   const workOrderTask = workOrder.tasks.find(
@@ -68,17 +69,33 @@ export async function startVodPackageTask(
       s3EndpointUrl: opts.s3EndpointUrl
     }
   );
-  task.taskPayload = shakaJob;
-  task.status = 'IN_PROGRESS';
+  await workOrderManager.updateWorkOrderTask(
+    workOrder.id,
+    task.type,
+    'IN_PROGRESS',
+    shakaJob as ShakaJob
+  );
 }
 
 export async function updateVodPackageTask(
   ctx: Context,
   task: WorkOrderTask,
   workOrder: WorkOrder,
+  workOrderManager: WorkOrderManager,
   opts: OrchestratorOptions
 ) {
   const shakaTaskPayload = task.taskPayload as ShakaJob;
+  if (!shakaTaskPayload) {
+    console.error(
+      `[${workOrder.id}]: No Shaka task payload found for task ${task.type}`
+    );
+    await workOrderManager.updateWorkOrderTask(
+      workOrder.id,
+      task.type,
+      'FAILED'
+    );
+    return;
+  }
   const shakaServiceAccessToken = await ctx.getServiceAccessToken(
     'eyevinn-shaka-packager-s3'
   );
@@ -99,4 +116,9 @@ export async function updateVodPackageTask(
       `[${workOrder.id}]: Shaka job ${shakaTaskPayload.name} failed: ${shakaJob.error}`
     );
   }
+  await workOrderManager.updateWorkOrderTask(
+    workOrder.id,
+    task.type,
+    task.status
+  );
 }