Add peertube runner cli

10 files changed, 389 insertions, 0 deletions

diff --git a/packages/peertube-runner/shared/config-manager.ts b/packages/peertube-runner/shared/config-manager.ts
new file mode 100644
index 000000000..352bae1fa
--- /dev/null
+++ b/packages/peertube-runner/shared/config-manager.ts
@@ -0,0 +1,139 @@
+import envPaths from 'env-paths'
+import { ensureDir, pathExists, readFile, remove, writeFile } from 'fs-extra'
+import { merge } from 'lodash'
+import { logger } from 'packages/peertube-runner/shared/logger'
+import { dirname, join } from 'path'
+import { parse, stringify } from '@iarna/toml'
+
+const paths = envPaths('peertube-runner')
+
+type Config = {
+  jobs: {
+    concurrency: number
+  }
+
+  ffmpeg: {
+    threads: number
+    nice: number
+  }
+
+  registeredInstances: {
+    url: string
+    runnerToken: string
+    runnerName: string
+    runnerDescription?: string
+  }[]
+}
+
+export class ConfigManager {
+  private static instance: ConfigManager
+
+  private config: Config = {
+    jobs: {
+      concurrency: 2
+    },
+    ffmpeg: {
+      threads: 2,
+      nice: 20
+    },
+    registeredInstances: []
+  }
+
+  private id: string
+  private configFilePath: string
+
+  private constructor () {}
+
+  init (id: string) {
+    this.id = id
+    this.configFilePath = join(this.getConfigDir(), 'config.toml')
+  }
+
+  async load () {
+    logger.info(`Using ${this.configFilePath} as configuration file`)
+
+    if (this.isTestInstance()) {
+      logger.info('Removing configuration file as we are using the "test" id')
+      await remove(this.configFilePath)
+    }
+
+    await ensureDir(dirname(this.configFilePath))
+
+    if (!await pathExists(this.configFilePath)) {
+      await this.save()
+    }
+
+    const file = await readFile(this.configFilePath, 'utf-8')
+
+    this.config = merge(this.config, parse(file))
+  }
+
+  save () {
+    return writeFile(this.configFilePath, stringify(this.config))
+  }
+
+  // ---------------------------------------------------------------------------
+
+  async setRegisteredInstances (registeredInstances: {
+    url: string
+    runnerToken: string
+    runnerName: string
+    runnerDescription?: string
+  }[]) {
+    this.config.registeredInstances = registeredInstances
+
+    await this.save()
+  }
+
+  // ---------------------------------------------------------------------------
+
+  getConfig () {
+    return this.deepFreeze(this.config)
+  }
+
+  // ---------------------------------------------------------------------------
+
+  getTranscodingDirectory () {
+    return join(paths.cache, this.id, 'transcoding')
+  }
+
+  getSocketDirectory () {
+    return join(paths.data, this.id)
+  }
+
+  getSocketPath () {
+    return join(this.getSocketDirectory(), 'peertube-runner.sock')
+  }
+
+  getConfigDir () {
+    return join(paths.config, this.id)
+  }
+
+  // ---------------------------------------------------------------------------
+
+  isTestInstance () {
+    return this.id === 'test'
+  }
+
+  // ---------------------------------------------------------------------------
+
+  // Thanks: https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Object/freeze
+  private deepFreeze <T extends object> (object: T) {
+    const propNames = Reflect.ownKeys(object)
+
+    // Freeze properties before freezing self
+    for (const name of propNames) {
+      const value = object[name]
+
+      if ((value && typeof value === 'object') || typeof value === 'function') {
+        this.deepFreeze(value)
+      }
+    }
+
+    return Object.freeze({ ...object })
+  }
+
+  static get Instance () {
+    return this.instance || (this.instance = new this())
+  }
+}
diff --git a/packages/peertube-runner/shared/http.ts b/packages/peertube-runner/shared/http.ts
new file mode 100644
index 000000000..d3fff70d1
--- /dev/null
+++ b/packages/peertube-runner/shared/http.ts
@@ -0,0 +1,66 @@
+import { createWriteStream, remove } from 'fs-extra'
+import { request as requestHTTP } from 'http'
+import { request as requestHTTPS, RequestOptions } from 'https'
+import { logger } from './logger'
+
+export function downloadFile (options: {
+  url: string
+  destination: string
+  runnerToken: string
+  jobToken: string
+}) {
+  const { url, destination, runnerToken, jobToken } = options
+
+  logger.debug(`Downloading file ${url}`)
+
+  return new Promise<void>((res, rej) => {
+    const parsed = new URL(url)
+
+    const body = JSON.stringify({
+      runnerToken,
+      jobToken
+    })
+
+    const getOptions: RequestOptions = {
+      method: 'POST',
+      hostname: parsed.hostname,
+      port: parsed.port,
+      path: parsed.pathname,
+      headers: {
+        'Content-Type': 'application/json',
+        'Content-Length': Buffer.byteLength(body, 'utf-8')
+      }
+    }
+
+    const request = getRequest(url)(getOptions, response => {
+      const code = response.statusCode ?? 0
+
+      if (code >= 400) {
+        return rej(new Error(response.statusMessage))
+      }
+
+      const file = createWriteStream(destination)
+      file.on('finish', () => res())
+
+      response.pipe(file)
+    })
+
+    request.on('error', err => {
+      remove(destination)
+        .catch(err => console.error(err))
+
+      return rej(err)
+    })
+
+    request.write(body)
+    request.end()
+  })
+}
+
+// ---------------------------------------------------------------------------
+
+function getRequest (url: string) {
+  if (url.startsWith('https://')) return requestHTTPS
+
+  return requestHTTP
+}
diff --git a/packages/peertube-runner/shared/index.ts b/packages/peertube-runner/shared/index.ts
new file mode 100644
index 000000000..d0b5a2e3e
--- /dev/null
+++ b/packages/peertube-runner/shared/index.ts
@@ -0,0 +1,3 @@
+export * from './config-manager'
+export * from './http'
+export * from './logger'
diff --git a/packages/peertube-runner/shared/ipc/index.ts b/packages/peertube-runner/shared/ipc/index.ts
new file mode 100644
index 000000000..ad4590281
--- /dev/null
+++ b/packages/peertube-runner/shared/ipc/index.ts
@@ -0,0 +1,2 @@
+export * from './ipc-client'
+export * from './ipc-server'
diff --git a/packages/peertube-runner/shared/ipc/ipc-client.ts b/packages/peertube-runner/shared/ipc/ipc-client.ts
new file mode 100644
index 000000000..7f5951157
--- /dev/null
+++ b/packages/peertube-runner/shared/ipc/ipc-client.ts
@@ -0,0 +1,74 @@
+import CliTable3 from 'cli-table3'
+import { ensureDir } from 'fs-extra'
+import { Client as NetIPC } from 'net-ipc'
+import { ConfigManager } from '../config-manager'
+import { IPCReponse, IPCReponseData, IPCRequest } from './shared'
+
+export class IPCClient {
+  private netIPC: NetIPC
+
+  async run () {
+    await ensureDir(ConfigManager.Instance.getSocketDirectory())
+
+    const socketPath = ConfigManager.Instance.getSocketPath()
+    this.netIPC = new NetIPC({ path: socketPath })
+    await this.netIPC.connect()
+  }
+
+  async askRegister (options: {
+    url: string
+    registrationToken: string
+    runnerName: string
+    runnerDescription?: string
+  }) {
+    const req: IPCRequest = {
+      type: 'register',
+      ...options
+    }
+
+    const { success, error } = await this.netIPC.request(req) as IPCReponse
+
+    if (success) console.log('PeerTube instance registered')
+    else console.error('Could not register PeerTube instance on runner server side', error)
+  }
+
+  async askUnregister (options: {
+    url: string
+  }) {
+    const req: IPCRequest = {
+      type: 'unregister',
+      ...options
+    }
+
+    const { success, error } = await this.netIPC.request(req) as IPCReponse
+
+    if (success) console.log('PeerTube instance unregistered')
+    else console.error('Could not unregister PeerTube instance on runner server side', error)
+  }
+
+  async askListRegistered () {
+    const req: IPCRequest = {
+      type: 'list-registered'
+    }
+
+    const { success, error, data } = await this.netIPC.request(req) as IPCReponse<IPCReponseData>
+    if (!success) {
+      console.error('Could not list registered PeerTube instances', error)
+      return
+    }
+
+    const table = new CliTable3({
+      head: [ 'instance', 'runner name', 'runner description' ]
+    })
+
+    for (const server of data.servers) {
+      table.push([ server.url, server.runnerName, server.runnerDescription ])
+    }
+
+    console.log(table.toString())
+  }
+
+  stop () {
+    this.netIPC.destroy()
+  }
+}
diff --git a/packages/peertube-runner/shared/ipc/ipc-server.ts b/packages/peertube-runner/shared/ipc/ipc-server.ts
new file mode 100644
index 000000000..bc340198b
--- /dev/null
+++ b/packages/peertube-runner/shared/ipc/ipc-server.ts
@@ -0,0 +1,61 @@
+import { ensureDir } from 'fs-extra'
+import { Server as NetIPC } from 'net-ipc'
+import { pick } from '@shared/core-utils'
+import { RunnerServer } from '../../server'
+import { ConfigManager } from '../config-manager'
+import { logger } from '../logger'
+import { IPCReponse, IPCReponseData, IPCRequest } from './shared'
+
+export class IPCServer {
+  private netIPC: NetIPC
+  private runnerServer: RunnerServer
+
+  async run (runnerServer: RunnerServer) {
+    this.runnerServer = runnerServer
+
+    await ensureDir(ConfigManager.Instance.getSocketDirectory())
+
+    const socketPath = ConfigManager.Instance.getSocketPath()
+    this.netIPC = new NetIPC({ path: socketPath })
+    await this.netIPC.start()
+
+    logger.info(`IPC socket created on ${socketPath}`)
+
+    this.netIPC.on('request', async (req: IPCRequest, res) => {
+      try {
+        const data = await this.process(req)
+
+        this.sendReponse(res, { success: true, data })
+      } catch (err) {
+        console.error('Cannot execute RPC call', err)
+        this.sendReponse(res, { success: false, error: err.message })
+      }
+    })
+  }
+
+  private async process (req: IPCRequest) {
+    switch (req.type) {
+      case 'register':
+        await this.runnerServer.registerRunner(pick(req, [ 'url', 'registrationToken', 'runnerName', 'runnerDescription' ]))
+        return undefined
+
+      case 'unregister':
+        await this.runnerServer.unregisterRunner({ url: req.url })
+        return undefined
+
+      case 'list-registered':
+        return Promise.resolve(this.runnerServer.listRegistered())
+
+      default:
+        throw new Error('Unknown RPC call ' + (req as any).type)
+    }
+  }
+
+  private sendReponse <T extends IPCReponseData> (
+    response: (data: any) => Promise<void>,
+    body: IPCReponse<T>
+  ) {
+    response(body)
+      .catch(err => console.error('Cannot send response after IPC request', err))
+  }
+}
diff --git a/packages/peertube-runner/shared/ipc/shared/index.ts b/packages/peertube-runner/shared/ipc/shared/index.ts
new file mode 100644
index 000000000..deaaa152e
--- /dev/null
+++ b/packages/peertube-runner/shared/ipc/shared/index.ts
@@ -0,0 +1,2 @@
+export * from './ipc-request.model'
+export * from './ipc-response.model'
diff --git a/packages/peertube-runner/shared/ipc/shared/ipc-request.model.ts b/packages/peertube-runner/shared/ipc/shared/ipc-request.model.ts
new file mode 100644
index 000000000..0f733cdfe
--- /dev/null
+++ b/packages/peertube-runner/shared/ipc/shared/ipc-request.model.ts
@@ -0,0 +1,15 @@
+export type IPCRequest =
+  IPCRequestRegister |
+  IPCRequestUnregister |
+  IPCRequestListRegistered
+
+export type IPCRequestRegister = {
+  type: 'register'
+  url: string
+  registrationToken: string
+  runnerName: string
+  runnerDescription?: string
+}
+
+export type IPCRequestUnregister = { type: 'unregister', url: string }
+export type IPCRequestListRegistered = { type: 'list-registered' }
diff --git a/packages/peertube-runner/shared/ipc/shared/ipc-response.model.ts b/packages/peertube-runner/shared/ipc/shared/ipc-response.model.ts
new file mode 100644
index 000000000..689d6e09a
--- /dev/null
+++ b/packages/peertube-runner/shared/ipc/shared/ipc-response.model.ts
@@ -0,0 +1,15 @@
+export type IPCReponse <T extends IPCReponseData = undefined> = {
+  success: boolean
+  error?: string
+  data?: T
+}
+
+export type IPCReponseData =
+  // list registered
+  {
+    servers: {
+      runnerName: string
+      runnerDescription: string
+      url: string
+    }[]
+  }
diff --git a/packages/peertube-runner/shared/logger.ts b/packages/peertube-runner/shared/logger.ts
new file mode 100644
index 000000000..bf0f41828
--- /dev/null
+++ b/packages/peertube-runner/shared/logger.ts
@@ -0,0 +1,12 @@
+import { pino } from 'pino'
+import pretty from 'pino-pretty'
+
+const logger = pino(pretty({
+  colorize: true
+}))
+
+logger.level = 'info'
+
+export {
+  logger
+}