From bc3918b2aed033e1c7617c0610e2e363c9e605db Mon Sep 17 00:00:00 2001
From: Chocobozzz <me@florianbigard.com>
Date: Thu, 22 Jun 2023 15:02:27 +0200
Subject: [PATCH] Shuffle servers when checking available jobs

To not favour a specific instance
---
 .../server/process/shared/process-studio.ts   |  4 ++++
 .../server/process/shared/process-vod.ts      | 20 +++++++++++++++++++
 packages/peertube-runner/server/server.ts     |  4 ++--
 shared/core-utils/common/array.ts             | 14 +++++++++++++
 4 files changed, 40 insertions(+), 2 deletions(-)

diff --git a/packages/peertube-runner/server/process/shared/process-studio.ts b/packages/peertube-runner/server/process/shared/process-studio.ts
index 9c745d031..ce014495e 100644
--- a/packages/peertube-runner/server/process/shared/process-studio.ts
+++ b/packages/peertube-runner/server/process/shared/process-studio.ts
@@ -20,10 +20,14 @@ export async function processStudioTranscoding (options: ProcessOptions<RunnerJo
   const { server, job, runnerToken } = options
   const payload = job.payload
 
+  logger.info(`Downloading input file ${payload.input.videoFileUrl} for job ${job.jobToken}`)
+
   let outputPath: string
   const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
   let tmpInputFilePath = inputPath
 
+  logger.info(`Input file ${payload.input.videoFileUrl} downloaded for job ${job.jobToken}. Running studio transcoding tasks.`)
+
   try {
     for (const task of payload.tasks) {
       const outputFilename = 'output-edition-' + buildUUID() + '.mp4'
diff --git a/packages/peertube-runner/server/process/shared/process-vod.ts b/packages/peertube-runner/server/process/shared/process-vod.ts
index 22489afd5..7c1119b50 100644
--- a/packages/peertube-runner/server/process/shared/process-vod.ts
+++ b/packages/peertube-runner/server/process/shared/process-vod.ts
@@ -1,4 +1,5 @@
 import { remove } from 'fs-extra'
+import { logger } from 'packages/peertube-runner/shared'
 import { join } from 'path'
 import { buildUUID } from '@shared/extra-utils'
 import {
@@ -16,8 +17,12 @@ export async function processWebVideoTranscoding (options: ProcessOptions<Runner
   const { server, job, runnerToken } = options
   const payload = job.payload
 
+  logger.info(`Downloading input file ${payload.input.videoFileUrl} for web video transcoding job ${job.jobToken}`)
+
   const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
 
+  logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running web video transcoding.`)
+
   const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
 
   const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `output-${buildUUID()}.mp4`)
@@ -56,7 +61,12 @@ export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVO
   const { server, job, runnerToken } = options
   const payload = job.payload
 
+  logger.info(`Downloading input file ${payload.input.videoFileUrl} for HLS transcoding job ${job.jobToken}`)
+
   const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
+
+  logger.info(`Downloaded input file ${payload.input.videoFileUrl} for job ${job.jobToken}. Running HLS transcoding.`)
+
   const uuid = buildUUID()
 
   const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `${uuid}-${payload.output.resolution}.m3u8`)
@@ -101,9 +111,19 @@ export async function processAudioMergeTranscoding (options: ProcessOptions<Runn
   const { server, job, runnerToken } = options
   const payload = job.payload
 
+  logger.info(
+    `Downloading input files ${payload.input.audioFileUrl} and ${payload.input.previewFileUrl} ` +
+    `for audio merge transcoding job ${job.jobToken}`
+  )
+
   const audioPath = await downloadInputFile({ url: payload.input.audioFileUrl, runnerToken, job })
   const inputPath = await downloadInputFile({ url: payload.input.previewFileUrl, runnerToken, job })
 
+  logger.info(
+    `Downloaded input files ${payload.input.audioFileUrl} and ${payload.input.previewFileUrl} ` +
+    `for job ${job.jobToken}. Running audio merge transcoding.`
+  )
+
   const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `output-${buildUUID()}.mp4`)
 
   const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
diff --git a/packages/peertube-runner/server/server.ts b/packages/peertube-runner/server/server.ts
index e76131c74..5fa86fa1a 100644
--- a/packages/peertube-runner/server/server.ts
+++ b/packages/peertube-runner/server/server.ts
@@ -1,7 +1,7 @@
 import { ensureDir, readdir, remove } from 'fs-extra'
 import { join } from 'path'
 import { io, Socket } from 'socket.io-client'
-import { pick, wait } from '@shared/core-utils'
+import { pick, shuffle, wait } from '@shared/core-utils'
 import { PeerTubeProblemDocument, ServerErrorCode } from '@shared/models'
 import { PeerTubeServer as PeerTubeServerCommand } from '@shared/server-commands'
 import { ConfigManager } from '../shared'
@@ -175,7 +175,7 @@ export class RunnerServer {
 
     let hadAvailableJob = false
 
-    for (const server of this.servers) {
+    for (const server of shuffle([ ...this.servers ])) {
       try {
         logger.info('Checking available jobs on ' + server.url)
 
diff --git a/shared/core-utils/common/array.ts b/shared/core-utils/common/array.ts
index e1b422165..878ed1ffe 100644
--- a/shared/core-utils/common/array.ts
+++ b/shared/core-utils/common/array.ts
@@ -20,8 +20,22 @@ function uniqify <T> (elements: T[]) {
   return Array.from(new Set(elements))
 }
 
+// Thanks: https://stackoverflow.com/a/12646864
+function shuffle <T> (elements: T[]) {
+  const shuffled = [ ...elements ]
+
+  for (let i = shuffled.length - 1; i > 0; i--) {
+    const j = Math.floor(Math.random() * (i + 1));
+
+    [ shuffled[i], shuffled[j] ] = [ shuffled[j], shuffled[i] ]
+  }
+
+  return shuffled
+}
+
 export {
   uniqify,
   findCommonElement,
+  shuffle,
   arrayify
 }