Implement remote runner jobs in server

Move ffmpeg functions to @shared

7 files changed, 183 insertions, 23 deletions

diff --git a/server/initializers/checker-after-init.ts b/server/initializers/checker-after-init.ts
index 14ed82cb4..68dea909d 100644
--- a/server/initializers/checker-after-init.ts
+++ b/server/initializers/checker-after-init.ts
@@ -1,7 +1,7 @@
 import config from 'config'
 import { URL } from 'url'
-import { getFFmpegVersion } from '@server/helpers/ffmpeg'
 import { uniqify } from '@shared/core-utils'
+import { getFFmpegVersion } from '@shared/ffmpeg'
 import { VideoRedundancyConfigFilter } from '@shared/models/redundancy/video-redundancy-config-filter.type'
 import { RecentlyAddedStrategy } from '../../shared/models/redundancy'
 import { isProdInstance, parseBytes, parseSemVersion } from '../helpers/core-utils'
diff --git a/server/initializers/checker-before-init.ts b/server/initializers/checker-before-init.ts
index 49010c059..2361aa1eb 100644
--- a/server/initializers/checker-before-init.ts
+++ b/server/initializers/checker-before-init.ts
@@ -1,5 +1,6 @@
 import { IConfig } from 'config'
-import { parseSemVersion, promisify0 } from '../helpers/core-utils'
+import { promisify0 } from '@shared/core-utils'
+import { parseSemVersion } from '../helpers/core-utils'
 import { logger } from '../helpers/logger'
 
 // Special behaviour for config because we can reload it
@@ -36,7 +37,9 @@ function checkMissedConfig () {
     'transcoding.profile', 'transcoding.concurrency',
     'transcoding.resolutions.0p', 'transcoding.resolutions.144p', 'transcoding.resolutions.240p', 'transcoding.resolutions.360p',
     'transcoding.resolutions.480p', 'transcoding.resolutions.720p', 'transcoding.resolutions.1080p', 'transcoding.resolutions.1440p',
-    'transcoding.resolutions.2160p', 'transcoding.always_transcode_original_resolution', 'video_studio.enabled',
+    'transcoding.resolutions.2160p', 'transcoding.always_transcode_original_resolution', 'transcoding.remote_runners.enabled',
+    'video_studio.enabled',
+    'remote_runners.stalled_jobs.vod', 'remote_runners.stalled_jobs.live',
     'import.videos.http.enabled', 'import.videos.torrent.enabled', 'import.videos.concurrency', 'import.videos.timeout',
     'import.video_channel_synchronization.enabled', 'import.video_channel_synchronization.max_per_user',
     'import.video_channel_synchronization.check_interval', 'import.video_channel_synchronization.videos_limit_per_synchronization',
@@ -74,7 +77,8 @@ function checkMissedConfig () {
     'live.transcoding.enabled', 'live.transcoding.threads', 'live.transcoding.profile',
     'live.transcoding.resolutions.144p', 'live.transcoding.resolutions.240p', 'live.transcoding.resolutions.360p',
     'live.transcoding.resolutions.480p', 'live.transcoding.resolutions.720p', 'live.transcoding.resolutions.1080p',
-    'live.transcoding.resolutions.1440p', 'live.transcoding.resolutions.2160p', 'live.transcoding.always_transcode_original_resolution'
+    'live.transcoding.resolutions.1440p', 'live.transcoding.resolutions.2160p', 'live.transcoding.always_transcode_original_resolution',
+    'live.transcoding.remote_runners.enabled'
   ]
 
   const requiredAlternatives = [
diff --git a/server/initializers/config.ts b/server/initializers/config.ts
index e2442213c..699dd4704 100644
--- a/server/initializers/config.ts
+++ b/server/initializers/config.ts
@@ -304,6 +304,12 @@ const CONFIG = {
       COUNT: config.get<number>('feeds.comments.count')
     }
   },
+  REMOTE_RUNNERS: {
+    STALLED_JOBS: {
+      LIVE: parseDurationToMs(config.get<string>('remote_runners.stalled_jobs.live')),
+      VOD: parseDurationToMs(config.get<string>('remote_runners.stalled_jobs.vod'))
+    }
+  },
   ADMIN: {
     get EMAIL () { return config.get<string>('admin.email') }
   },
@@ -359,6 +365,9 @@ const CONFIG = {
     },
     WEBTORRENT: {
       get ENABLED () { return config.get<boolean>('transcoding.webtorrent.enabled') }
+    },
+    REMOTE_RUNNERS: {
+      get ENABLED () { return config.get<boolean>('transcoding.remote_runners.enabled') }
     }
   },
   LIVE: {
@@ -406,6 +415,9 @@ const CONFIG = {
         get '1080p' () { return config.get<boolean>('live.transcoding.resolutions.1080p') },
         get '1440p' () { return config.get<boolean>('live.transcoding.resolutions.1440p') },
         get '2160p' () { return config.get<boolean>('live.transcoding.resolutions.2160p') }
+      },
+      REMOTE_RUNNERS: {
+        get ENABLED () { return config.get<boolean>('live.transcoding.remote_runners.enabled') }
       }
     }
   },
diff --git a/server/initializers/constants.ts b/server/initializers/constants.ts
index 6cad4eb23..279e77421 100644
--- a/server/initializers/constants.ts
+++ b/server/initializers/constants.ts
@@ -6,6 +6,7 @@ import { randomInt, root } from '@shared/core-utils'
 import {
   AbuseState,
   JobType,
+  RunnerJobState,
   UserRegistrationState,
   VideoChannelSyncState,
   VideoImportState,
@@ -26,7 +27,7 @@ import { CONFIG, registerConfigChangedHandler } from './config'
 
 // ---------------------------------------------------------------------------
 
-const LAST_MIGRATION_VERSION = 760
+const LAST_MIGRATION_VERSION = 765
 
 // ---------------------------------------------------------------------------
 
@@ -81,6 +82,10 @@ const SORTABLE_COLUMNS = {
 
   USER_REGISTRATIONS: [ 'createdAt', 'state' ],
 
+  RUNNERS: [ 'createdAt' ],
+  RUNNER_REGISTRATION_TOKENS: [ 'createdAt' ],
+  RUNNER_JOBS: [ 'updatedAt', 'createdAt', 'priority', 'state', 'progress' ],
+
   VIDEOS: [ 'name', 'duration', 'createdAt', 'publishedAt', 'originallyPublishedAt', 'views', 'likes', 'trending', 'hot', 'best' ],
 
   // Don't forget to update peertube-search-index with the same values
@@ -139,6 +144,8 @@ const REMOTE_SCHEME = {
   WS: 'wss'
 }
 
+// ---------------------------------------------------------------------------
+
 const JOB_ATTEMPTS: { [id in JobType]: number } = {
   'activitypub-http-broadcast': 1,
   'activitypub-http-broadcast-parallel': 1,
@@ -160,6 +167,7 @@ const JOB_ATTEMPTS: { [id in JobType]: number } = {
   'video-channel-import': 1,
   'after-video-channel-import': 1,
   'move-to-object-storage': 3,
+  'transcoding-job-builder': 1,
   'notify': 1,
   'federate-video': 1
 }
@@ -183,6 +191,7 @@ const JOB_CONCURRENCY: { [id in Exclude<JobType, 'video-transcoding' | 'video-im
   'move-to-object-storage': 1,
   'video-channel-import': 1,
   'after-video-channel-import': 1,
+  'transcoding-job-builder': 1,
   'notify': 5,
   'federate-video': 3
 }
@@ -207,6 +216,7 @@ const JOB_TTL: { [id in JobType]: number } = {
   'move-to-object-storage': 1000 * 60 * 60 * 3, // 3 hours
   'video-channel-import': 1000 * 60 * 60 * 4, // 4 hours
   'after-video-channel-import': 60000 * 5, // 5 minutes
+  'transcoding-job-builder': 60000, // 1 minute
   'notify': 60000 * 5, // 5 minutes
   'federate-video': 60000 * 5 // 5 minutes
 }
@@ -222,21 +232,6 @@ const JOB_PRIORITY = {
   TRANSCODING: 100
 }
 
-const BROADCAST_CONCURRENCY = 30 // How many requests in parallel we do in activitypub-http-broadcast job
-const CRAWL_REQUEST_CONCURRENCY = 1 // How many requests in parallel to fetch remote data (likes, shares...)
-
-const AP_CLEANER = {
-  CONCURRENCY: 10, // How many requests in parallel we do in activitypub-cleaner job
-  UNAVAILABLE_TRESHOLD: 3, // How many attempts we do before removing an unavailable remote resource
-  PERIOD: parseDurationToMs('1 week') // /!\ Has to be sync with REPEAT_JOBS
-}
-
-const REQUEST_TIMEOUTS = {
-  DEFAULT: 7000, // 7 seconds
-  FILE: 30000, // 30 seconds
-  REDUNDANCY: JOB_TTL['video-redundancy']
-}
-
 const JOB_REMOVAL_OPTIONS = {
   COUNT: 10000, // Max jobs to store
 
@@ -256,7 +251,29 @@ const JOB_REMOVAL_OPTIONS = {
 
 const VIDEO_IMPORT_TIMEOUT = Math.floor(JOB_TTL['video-import'] * 0.9)
 
+const RUNNER_JOBS = {
+  MAX_FAILURES: 5
+}
+
+// ---------------------------------------------------------------------------
+
+const BROADCAST_CONCURRENCY = 30 // How many requests in parallel we do in activitypub-http-broadcast job
+const CRAWL_REQUEST_CONCURRENCY = 1 // How many requests in parallel to fetch remote data (likes, shares...)
+
+const AP_CLEANER = {
+  CONCURRENCY: 10, // How many requests in parallel we do in activitypub-cleaner job
+  UNAVAILABLE_TRESHOLD: 3, // How many attempts we do before removing an unavailable remote resource
+  PERIOD: parseDurationToMs('1 week') // /!\ Has to be sync with REPEAT_JOBS
+}
+
+const REQUEST_TIMEOUTS = {
+  DEFAULT: 7000, // 7 seconds
+  FILE: 30000, // 30 seconds
+  REDUNDANCY: JOB_TTL['video-redundancy']
+}
+
 const SCHEDULER_INTERVALS_MS = {
+  RUNNER_JOB_WATCH_DOG: Math.min(CONFIG.REMOTE_RUNNERS.STALLED_JOBS.VOD, CONFIG.REMOTE_RUNNERS.STALLED_JOBS.LIVE),
   ACTOR_FOLLOW_SCORES: 60000 * 60, // 1 hour
   REMOVE_OLD_JOBS: 60000 * 60, // 1 hour
   UPDATE_VIDEOS: 60000, // 1 minute
@@ -410,6 +427,17 @@ const CONSTRAINTS_FIELDS = {
     CLIENT_STACK_TRACE: { min: 1, max: 15000 }, // Length
     CLIENT_META: { min: 1, max: 5000 }, // Length
     CLIENT_USER_AGENT: { min: 1, max: 200 } // Length
+  },
+  RUNNERS: {
+    TOKEN: { min: 1, max: 1000 }, // Length
+    NAME: { min: 1, max: 100 }, // Length
+    DESCRIPTION: { min: 1, max: 1000 } // Length
+  },
+  RUNNER_JOBS: {
+    TOKEN: { min: 1, max: 1000 }, // Length
+    REASON: { min: 1, max: 5000 }, // Length
+    ERROR_MESSAGE: { min: 1, max: 5000 }, // Length
+    PROGRESS: { min: 0, max: 100 } // Value
   }
 }
 
@@ -540,6 +568,17 @@ const VIDEO_PLAYLIST_TYPES: { [ id in VideoPlaylistType ]: string } = {
   [VideoPlaylistType.WATCH_LATER]: 'Watch later'
 }
 
+const RUNNER_JOB_STATES: { [ id in RunnerJobState ]: string } = {
+  [RunnerJobState.PROCESSING]: 'Processing',
+  [RunnerJobState.COMPLETED]: 'Completed',
+  [RunnerJobState.PENDING]: 'Pending',
+  [RunnerJobState.ERRORED]: 'Errored',
+  [RunnerJobState.WAITING_FOR_PARENT_JOB]: 'Waiting for parent job to finish',
+  [RunnerJobState.CANCELLED]: 'Cancelled',
+  [RunnerJobState.PARENT_ERRORED]: 'Parent job failed',
+  [RunnerJobState.PARENT_CANCELLED]: 'Parent job cancelled'
+}
+
 const MIMETYPES = {
   AUDIO: {
     MIMETYPE_EXT: {
@@ -594,6 +633,11 @@ const MIMETYPES = {
     MIMETYPE_EXT: {
       'application/x-bittorrent': '.torrent'
     }
+  },
+  M3U8: {
+    MIMETYPE_EXT: {
+      'application/vnd.apple.mpegurl': '.m3u8'
+    }
   }
 }
 MIMETYPES.AUDIO.EXT_MIMETYPE = invert(MIMETYPES.AUDIO.MIMETYPE_EXT)
@@ -1027,6 +1071,7 @@ export {
   SEARCH_INDEX,
   DIRECTORIES,
   RESUMABLE_UPLOAD_SESSION_LIFETIME,
+  RUNNER_JOB_STATES,
   P2P_MEDIA_LOADER_PEER_VERSION,
   ACTOR_IMAGES_SIZE,
   ACCEPT_HEADERS,
@@ -1085,6 +1130,7 @@ export {
   USER_REGISTRATION_STATES,
   LRU_CACHE,
   REQUEST_TIMEOUTS,
+  RUNNER_JOBS,
   MAX_LOCAL_VIEWER_WATCH_SECTIONS,
   USER_PASSWORD_RESET_LIFETIME,
   USER_PASSWORD_CREATE_LIFETIME,
diff --git a/server/initializers/database.ts b/server/initializers/database.ts
index 3f31099ed..14dd8c379 100644
--- a/server/initializers/database.ts
+++ b/server/initializers/database.ts
@@ -1,6 +1,9 @@
 import { QueryTypes, Transaction } from 'sequelize'
 import { Sequelize as SequelizeTypescript } from 'sequelize-typescript'
 import { ActorCustomPageModel } from '@server/models/account/actor-custom-page'
+import { RunnerModel } from '@server/models/runner/runner'
+import { RunnerJobModel } from '@server/models/runner/runner-job'
+import { RunnerRegistrationTokenModel } from '@server/models/runner/runner-registration-token'
 import { TrackerModel } from '@server/models/server/tracker'
 import { VideoTrackerModel } from '@server/models/server/video-tracker'
 import { UserModel } from '@server/models/user/user'
@@ -9,6 +12,7 @@ import { UserRegistrationModel } from '@server/models/user/user-registration'
 import { UserVideoHistoryModel } from '@server/models/user/user-video-history'
 import { VideoChannelSyncModel } from '@server/models/video/video-channel-sync'
 import { VideoJobInfoModel } from '@server/models/video/video-job-info'
+import { VideoLiveReplaySettingModel } from '@server/models/video/video-live-replay-setting'
 import { VideoLiveSessionModel } from '@server/models/video/video-live-session'
 import { VideoSourceModel } from '@server/models/video/video-source'
 import { LocalVideoViewerModel } from '@server/models/view/local-video-viewer'
@@ -52,7 +56,6 @@ import { VideoStreamingPlaylistModel } from '../models/video/video-streaming-pla
 import { VideoTagModel } from '../models/video/video-tag'
 import { VideoViewModel } from '../models/view/video-view'
 import { CONFIG } from './config'
-import { VideoLiveReplaySettingModel } from '@server/models/video/video-live-replay-setting'
 
 require('pg').defaults.parseInt8 = true // Avoid BIGINT to be converted to string
 
@@ -159,7 +162,10 @@ async function initDatabaseModels (silent: boolean) {
     ActorCustomPageModel,
     VideoJobInfoModel,
     VideoChannelSyncModel,
-    UserRegistrationModel
+    UserRegistrationModel,
+    RunnerRegistrationTokenModel,
+    RunnerModel,
+    RunnerJobModel
   ])
 
   // Check extensions exist in the database
diff --git a/server/initializers/installer.ts b/server/initializers/installer.ts
index f48f348a7..2406a5936 100644
--- a/server/initializers/installer.ts
+++ b/server/initializers/installer.ts
@@ -2,7 +2,9 @@ import { ensureDir, readdir, remove } from 'fs-extra'
 import passwordGenerator from 'password-generator'
 import { join } from 'path'
 import { isTestOrDevInstance } from '@server/helpers/core-utils'
+import { generateRunnerRegistrationToken } from '@server/helpers/token-generator'
 import { getNodeABIVersion } from '@server/helpers/version'
+import { RunnerRegistrationTokenModel } from '@server/models/runner/runner-registration-token'
 import { UserRole } from '@shared/models'
 import { logger } from '../helpers/logger'
 import { buildUser, createApplicationActor, createUserAccountAndChannelAndPlaylist } from '../lib/user'
@@ -22,7 +24,8 @@ async function installApplication () {
           return Promise.all([
             createApplicationIfNotExist(),
             createOAuthClientIfNotExist(),
-            createOAuthAdminIfNotExist()
+            createOAuthAdminIfNotExist(),
+            createRunnerRegistrationTokenIfNotExist()
           ])
         }),
 
@@ -183,3 +186,14 @@ async function createApplicationIfNotExist () {
 
   return createApplicationActor(application.id)
 }
+
+async function createRunnerRegistrationTokenIfNotExist () {
+  const total = await RunnerRegistrationTokenModel.countTotal()
+  if (total !== 0) return undefined
+
+  const token = new RunnerRegistrationTokenModel({
+    registrationToken: generateRunnerRegistrationToken()
+  })
+
+  await token.save()
+}
diff --git a/server/initializers/migrations/0765-remote-transcoding.ts b/server/initializers/migrations/0765-remote-transcoding.ts
new file mode 100644
index 000000000..40cca03b4
--- /dev/null
+++ b/server/initializers/migrations/0765-remote-transcoding.ts
@@ -0,0 +1,78 @@
+import * as Sequelize from 'sequelize'
+
+async function up (utils: {
+  transaction: Sequelize.Transaction
+  queryInterface: Sequelize.QueryInterface
+  sequelize: Sequelize.Sequelize
+}): Promise<void> {
+  {
+    const query = `
+    CREATE TABLE IF NOT EXISTS "runnerRegistrationToken"(
+      "id" serial,
+      "registrationToken" varchar(255) NOT NULL,
+      "createdAt" timestamp with time zone NOT NULL,
+      "updatedAt" timestamp with time zone NOT NULL,
+      PRIMARY KEY ("id")
+    );
+    `
+
+    await utils.sequelize.query(query, { transaction : utils.transaction })
+  }
+
+  {
+    const query = `
+    CREATE TABLE IF NOT EXISTS "runner"(
+      "id" serial,
+      "runnerToken" varchar(255) NOT NULL,
+      "name" varchar(255) NOT NULL,
+      "description" varchar(1000),
+      "lastContact" timestamp with time zone NOT NULL,
+      "ip" varchar(255) NOT NULL,
+      "runnerRegistrationTokenId" integer REFERENCES "runnerRegistrationToken"("id") ON DELETE CASCADE ON UPDATE CASCADE,
+      "createdAt" timestamp with time zone NOT NULL,
+      "updatedAt" timestamp with time zone NOT NULL,
+      PRIMARY KEY ("id")
+    );
+    `
+
+    await utils.sequelize.query(query, { transaction : utils.transaction })
+  }
+
+  {
+    const query = `
+    CREATE TABLE IF NOT EXISTS "runnerJob"(
+      "id" serial,
+      "uuid" uuid NOT NULL,
+      "type" varchar(255) NOT NULL,
+      "payload" jsonb NOT NULL,
+      "privatePayload" jsonb NOT NULL,
+      "state" integer NOT NULL,
+      "failures" integer NOT NULL DEFAULT 0,
+      "error" varchar(5000),
+      "priority" integer NOT NULL,
+      "processingJobToken" varchar(255),
+      "progress" integer,
+      "startedAt" timestamp with time zone,
+      "finishedAt" timestamp with time zone,
+      "dependsOnRunnerJobId" integer REFERENCES "runnerJob"("id") ON DELETE CASCADE ON UPDATE CASCADE,
+      "runnerId" integer REFERENCES "runner"("id") ON DELETE SET NULL ON UPDATE CASCADE,
+      "createdAt" timestamp with time zone NOT NULL,
+      "updatedAt" timestamp with time zone NOT NULL,
+      PRIMARY KEY ("id")
+    );
+
+
+    `
+
+    await utils.sequelize.query(query, { transaction : utils.transaction })
+  }
+}
+
+function down (options) {
+  throw new Error('Not implemented.')
+}
+
+export {
+  up,
+  down
+}