]>
Commit | Line | Data |
---|---|---|
1772b383 C |
1 | import { ensureDir, readdir, remove } from 'fs-extra' |
2 | import { join } from 'path' | |
3 | import { io, Socket } from 'socket.io-client' | |
4 | import { pick } from '@shared/core-utils' | |
5 | import { PeerTubeProblemDocument, ServerErrorCode } from '@shared/models' | |
6 | import { PeerTubeServer as PeerTubeServerCommand } from '@shared/server-commands' | |
7 | import { ConfigManager } from '../shared' | |
8 | import { IPCServer } from '../shared/ipc' | |
9 | import { logger } from '../shared/logger' | |
10 | import { JobWithToken, processJob } from './process' | |
5e47f6ab | 11 | import { isJobSupported } from './shared' |
1772b383 C |
12 | |
13 | type PeerTubeServer = PeerTubeServerCommand & { | |
14 | runnerToken: string | |
15 | runnerName: string | |
16 | runnerDescription?: string | |
17 | } | |
18 | ||
19 | export class RunnerServer { | |
20 | private static instance: RunnerServer | |
21 | ||
22 | private servers: PeerTubeServer[] = [] | |
23 | private processingJobs: { job: JobWithToken, server: PeerTubeServer }[] = [] | |
24 | ||
25 | private checkingAvailableJobs = false | |
26 | ||
3a0c2a77 C |
27 | private cleaningUp = false |
28 | ||
1772b383 C |
29 | private readonly sockets = new Map<PeerTubeServer, Socket>() |
30 | ||
31 | private constructor () {} | |
32 | ||
33 | async run () { | |
34 | logger.info('Running PeerTube runner in server mode') | |
35 | ||
36 | await ConfigManager.Instance.load() | |
37 | ||
38 | for (const registered of ConfigManager.Instance.getConfig().registeredInstances) { | |
39 | const serverCommand = new PeerTubeServerCommand({ url: registered.url }) | |
40 | ||
41 | this.loadServer(Object.assign(serverCommand, registered)) | |
42 | ||
43 | logger.info(`Loading registered instance ${registered.url}`) | |
44 | } | |
45 | ||
46 | // Run IPC | |
47 | const ipcServer = new IPCServer() | |
48 | try { | |
49 | await ipcServer.run(this) | |
50 | } catch (err) { | |
3a0c2a77 | 51 | logger.error('Cannot start local socket for IPC communication', err) |
1772b383 C |
52 | process.exit(-1) |
53 | } | |
54 | ||
55 | // Cleanup on exit | |
56 | for (const code of [ 'SIGINT', 'SIGUSR1', 'SIGUSR2', 'uncaughtException' ]) { | |
3a0c2a77 C |
57 | process.on(code, async (err, origin) => { |
58 | if (code === 'uncaughtException') { | |
59 | logger.error({ err, origin }, 'uncaughtException') | |
60 | } | |
61 | ||
1772b383 C |
62 | await this.onExit() |
63 | }) | |
64 | } | |
65 | ||
66 | // Process jobs | |
67 | await ensureDir(ConfigManager.Instance.getTranscodingDirectory()) | |
68 | await this.cleanupTMP() | |
69 | ||
70 | logger.info(`Using ${ConfigManager.Instance.getTranscodingDirectory()} for transcoding directory`) | |
71 | ||
72 | await this.checkAvailableJobs() | |
73 | } | |
74 | ||
75 | // --------------------------------------------------------------------------- | |
76 | ||
77 | async registerRunner (options: { | |
78 | url: string | |
79 | registrationToken: string | |
80 | runnerName: string | |
81 | runnerDescription?: string | |
82 | }) { | |
83 | const { url, registrationToken, runnerName, runnerDescription } = options | |
84 | ||
85 | logger.info(`Registering runner ${runnerName} on ${url}...`) | |
86 | ||
87 | const serverCommand = new PeerTubeServerCommand({ url }) | |
88 | const { runnerToken } = await serverCommand.runners.register({ name: runnerName, description: runnerDescription, registrationToken }) | |
89 | ||
90 | const server: PeerTubeServer = Object.assign(serverCommand, { | |
91 | runnerToken, | |
92 | runnerName, | |
93 | runnerDescription | |
94 | }) | |
95 | ||
96 | this.loadServer(server) | |
97 | await this.saveRegisteredInstancesInConf() | |
98 | ||
99 | logger.info(`Registered runner ${runnerName} on ${url}`) | |
100 | ||
101 | await this.checkAvailableJobs() | |
102 | } | |
103 | ||
104 | private loadServer (server: PeerTubeServer) { | |
105 | this.servers.push(server) | |
106 | ||
107 | const url = server.url + '/runners' | |
108 | const socket = io(url, { | |
109 | auth: { | |
110 | runnerToken: server.runnerToken | |
111 | }, | |
112 | transports: [ 'websocket' ] | |
113 | }) | |
114 | ||
115 | socket.on('connect_error', err => logger.warn({ err }, `Cannot connect to ${url} socket`)) | |
116 | socket.on('connect', () => logger.info(`Connected to ${url} socket`)) | |
117 | socket.on('available-jobs', () => this.checkAvailableJobs()) | |
118 | ||
119 | this.sockets.set(server, socket) | |
120 | } | |
121 | ||
122 | async unregisterRunner (options: { | |
123 | url: string | |
124 | }) { | |
125 | const { url } = options | |
126 | ||
127 | const server = this.servers.find(s => s.url === url) | |
128 | if (!server) { | |
129 | logger.error(`Unknown server ${url} to unregister`) | |
130 | return | |
131 | } | |
132 | ||
133 | logger.info(`Unregistering runner ${server.runnerName} on ${url}...`) | |
134 | ||
135 | try { | |
136 | await server.runners.unregister({ runnerToken: server.runnerToken }) | |
137 | } catch (err) { | |
138 | logger.error({ err }, `Cannot unregister runner ${server.runnerName} on ${url}`) | |
139 | } | |
140 | ||
141 | this.unloadServer(server) | |
142 | await this.saveRegisteredInstancesInConf() | |
143 | ||
144 | logger.info(`Unregistered runner ${server.runnerName} on ${server.url}`) | |
145 | } | |
146 | ||
147 | private unloadServer (server: PeerTubeServer) { | |
148 | this.servers = this.servers.filter(s => s !== server) | |
149 | ||
150 | const socket = this.sockets.get(server) | |
151 | socket.disconnect() | |
152 | ||
153 | this.sockets.delete(server) | |
154 | } | |
155 | ||
156 | listRegistered () { | |
157 | return { | |
158 | servers: this.servers.map(s => { | |
159 | return { | |
160 | url: s.url, | |
161 | runnerName: s.runnerName, | |
162 | runnerDescription: s.runnerDescription | |
163 | } | |
164 | }) | |
165 | } | |
166 | } | |
167 | ||
168 | // --------------------------------------------------------------------------- | |
169 | ||
170 | private async checkAvailableJobs () { | |
171 | if (this.checkingAvailableJobs) return | |
172 | ||
1772b383 C |
173 | this.checkingAvailableJobs = true |
174 | ||
fe7019b2 C |
175 | let hadAvailableJob = false |
176 | ||
1772b383 C |
177 | for (const server of this.servers) { |
178 | try { | |
fe7019b2 C |
179 | logger.info('Checking available jobs on ' + server.url) |
180 | ||
1772b383 C |
181 | const job = await this.requestJob(server) |
182 | if (!job) continue | |
183 | ||
fe7019b2 C |
184 | hadAvailableJob = true |
185 | ||
1772b383 C |
186 | await this.tryToExecuteJobAsync(server, job) |
187 | } catch (err) { | |
fe7019b2 C |
188 | const code = (err.res?.body as PeerTubeProblemDocument)?.code |
189 | ||
190 | if (code === ServerErrorCode.RUNNER_JOB_NOT_IN_PROCESSING_STATE) { | |
191 | logger.debug({ err }, 'Runner job is not in processing state anymore, retry later') | |
192 | return | |
193 | } | |
194 | ||
195 | if (code === ServerErrorCode.UNKNOWN_RUNNER_TOKEN) { | |
1772b383 C |
196 | logger.error({ err }, `Unregistering ${server.url} as the runner token ${server.runnerToken} is invalid`) |
197 | ||
198 | await this.unregisterRunner({ url: server.url }) | |
199 | return | |
200 | } | |
201 | ||
202 | logger.error({ err }, `Cannot request/accept job on ${server.url} for runner ${server.runnerName}`) | |
203 | } | |
204 | } | |
205 | ||
206 | this.checkingAvailableJobs = false | |
fe7019b2 C |
207 | |
208 | if (hadAvailableJob && this.canProcessMoreJobs()) { | |
209 | this.checkAvailableJobs() | |
210 | .catch(err => logger.error({ err }, 'Cannot check more available jobs')) | |
211 | } | |
1772b383 C |
212 | } |
213 | ||
214 | private async requestJob (server: PeerTubeServer) { | |
215 | logger.debug(`Requesting jobs on ${server.url} for runner ${server.runnerName}`) | |
216 | ||
217 | const { availableJobs } = await server.runnerJobs.request({ runnerToken: server.runnerToken }) | |
218 | ||
5e47f6ab C |
219 | const filtered = availableJobs.filter(j => isJobSupported(j)) |
220 | ||
221 | if (filtered.length === 0) { | |
1772b383 C |
222 | logger.debug(`No job available on ${server.url} for runner ${server.runnerName}`) |
223 | return undefined | |
224 | } | |
225 | ||
5e47f6ab | 226 | return filtered[0] |
1772b383 C |
227 | } |
228 | ||
229 | private async tryToExecuteJobAsync (server: PeerTubeServer, jobToAccept: { uuid: string }) { | |
fe7019b2 | 230 | if (!this.canProcessMoreJobs()) return |
1772b383 C |
231 | |
232 | const { job } = await server.runnerJobs.accept({ runnerToken: server.runnerToken, jobUUID: jobToAccept.uuid }) | |
233 | ||
234 | const processingJob = { job, server } | |
235 | this.processingJobs.push(processingJob) | |
236 | ||
237 | processJob({ server, job, runnerToken: server.runnerToken }) | |
238 | .catch(err => { | |
239 | logger.error({ err }, 'Cannot process job') | |
240 | ||
241 | server.runnerJobs.error({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken: server.runnerToken, message: err.message }) | |
242 | .catch(err2 => logger.error({ err: err2 }, 'Cannot abort job after error')) | |
243 | }) | |
244 | .finally(() => { | |
245 | this.processingJobs = this.processingJobs.filter(p => p !== processingJob) | |
246 | ||
247 | return this.checkAvailableJobs() | |
248 | }) | |
249 | } | |
250 | ||
251 | // --------------------------------------------------------------------------- | |
252 | ||
253 | private saveRegisteredInstancesInConf () { | |
254 | const data = this.servers.map(s => { | |
255 | return pick(s, [ 'url', 'runnerToken', 'runnerName', 'runnerDescription' ]) | |
256 | }) | |
257 | ||
258 | return ConfigManager.Instance.setRegisteredInstances(data) | |
259 | } | |
260 | ||
fe7019b2 C |
261 | private canProcessMoreJobs () { |
262 | return this.processingJobs.length < ConfigManager.Instance.getConfig().jobs.concurrency | |
263 | } | |
264 | ||
1772b383 C |
265 | // --------------------------------------------------------------------------- |
266 | ||
267 | private async cleanupTMP () { | |
268 | const files = await readdir(ConfigManager.Instance.getTranscodingDirectory()) | |
269 | ||
270 | for (const file of files) { | |
271 | await remove(join(ConfigManager.Instance.getTranscodingDirectory(), file)) | |
272 | } | |
273 | } | |
274 | ||
275 | private async onExit () { | |
3a0c2a77 C |
276 | if (this.cleaningUp) return |
277 | this.cleaningUp = true | |
278 | ||
279 | logger.info('Cleaning up after program exit') | |
280 | ||
1772b383 C |
281 | try { |
282 | for (const { server, job } of this.processingJobs) { | |
283 | await server.runnerJobs.abort({ | |
284 | jobToken: job.jobToken, | |
285 | jobUUID: job.uuid, | |
286 | reason: 'Runner stopped', | |
287 | runnerToken: server.runnerToken | |
288 | }) | |
289 | } | |
290 | ||
291 | await this.cleanupTMP() | |
292 | } catch (err) { | |
3a0c2a77 | 293 | logger.error(err) |
1772b383 C |
294 | process.exit(-1) |
295 | } | |
296 | ||
297 | process.exit() | |
298 | } | |
299 | ||
300 | static get Instance () { | |
301 | return this.instance || (this.instance = new this()) | |
302 | } | |
303 | } |