]>
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 | ||
173 | logger.info('Checking available jobs') | |
174 | ||
175 | this.checkingAvailableJobs = true | |
176 | ||
177 | for (const server of this.servers) { | |
178 | try { | |
179 | const job = await this.requestJob(server) | |
180 | if (!job) continue | |
181 | ||
182 | await this.tryToExecuteJobAsync(server, job) | |
183 | } catch (err) { | |
184 | if ((err.res?.body as PeerTubeProblemDocument)?.code === ServerErrorCode.UNKNOWN_RUNNER_TOKEN) { | |
185 | logger.error({ err }, `Unregistering ${server.url} as the runner token ${server.runnerToken} is invalid`) | |
186 | ||
187 | await this.unregisterRunner({ url: server.url }) | |
188 | return | |
189 | } | |
190 | ||
191 | logger.error({ err }, `Cannot request/accept job on ${server.url} for runner ${server.runnerName}`) | |
192 | } | |
193 | } | |
194 | ||
195 | this.checkingAvailableJobs = false | |
196 | } | |
197 | ||
198 | private async requestJob (server: PeerTubeServer) { | |
199 | logger.debug(`Requesting jobs on ${server.url} for runner ${server.runnerName}`) | |
200 | ||
201 | const { availableJobs } = await server.runnerJobs.request({ runnerToken: server.runnerToken }) | |
202 | ||
5e47f6ab C |
203 | const filtered = availableJobs.filter(j => isJobSupported(j)) |
204 | ||
205 | if (filtered.length === 0) { | |
1772b383 C |
206 | logger.debug(`No job available on ${server.url} for runner ${server.runnerName}`) |
207 | return undefined | |
208 | } | |
209 | ||
5e47f6ab | 210 | return filtered[0] |
1772b383 C |
211 | } |
212 | ||
213 | private async tryToExecuteJobAsync (server: PeerTubeServer, jobToAccept: { uuid: string }) { | |
214 | if (this.processingJobs.length >= ConfigManager.Instance.getConfig().jobs.concurrency) return | |
215 | ||
216 | const { job } = await server.runnerJobs.accept({ runnerToken: server.runnerToken, jobUUID: jobToAccept.uuid }) | |
217 | ||
218 | const processingJob = { job, server } | |
219 | this.processingJobs.push(processingJob) | |
220 | ||
221 | processJob({ server, job, runnerToken: server.runnerToken }) | |
222 | .catch(err => { | |
223 | logger.error({ err }, 'Cannot process job') | |
224 | ||
225 | server.runnerJobs.error({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken: server.runnerToken, message: err.message }) | |
226 | .catch(err2 => logger.error({ err: err2 }, 'Cannot abort job after error')) | |
227 | }) | |
228 | .finally(() => { | |
229 | this.processingJobs = this.processingJobs.filter(p => p !== processingJob) | |
230 | ||
231 | return this.checkAvailableJobs() | |
232 | }) | |
233 | } | |
234 | ||
235 | // --------------------------------------------------------------------------- | |
236 | ||
237 | private saveRegisteredInstancesInConf () { | |
238 | const data = this.servers.map(s => { | |
239 | return pick(s, [ 'url', 'runnerToken', 'runnerName', 'runnerDescription' ]) | |
240 | }) | |
241 | ||
242 | return ConfigManager.Instance.setRegisteredInstances(data) | |
243 | } | |
244 | ||
245 | // --------------------------------------------------------------------------- | |
246 | ||
247 | private async cleanupTMP () { | |
248 | const files = await readdir(ConfigManager.Instance.getTranscodingDirectory()) | |
249 | ||
250 | for (const file of files) { | |
251 | await remove(join(ConfigManager.Instance.getTranscodingDirectory(), file)) | |
252 | } | |
253 | } | |
254 | ||
255 | private async onExit () { | |
3a0c2a77 C |
256 | if (this.cleaningUp) return |
257 | this.cleaningUp = true | |
258 | ||
259 | logger.info('Cleaning up after program exit') | |
260 | ||
1772b383 C |
261 | try { |
262 | for (const { server, job } of this.processingJobs) { | |
263 | await server.runnerJobs.abort({ | |
264 | jobToken: job.jobToken, | |
265 | jobUUID: job.uuid, | |
266 | reason: 'Runner stopped', | |
267 | runnerToken: server.runnerToken | |
268 | }) | |
269 | } | |
270 | ||
271 | await this.cleanupTMP() | |
272 | } catch (err) { | |
3a0c2a77 | 273 | logger.error(err) |
1772b383 C |
274 | process.exit(-1) |
275 | } | |
276 | ||
277 | process.exit() | |
278 | } | |
279 | ||
280 | static get Instance () { | |
281 | return this.instance || (this.instance = new this()) | |
282 | } | |
283 | } |