Verified Commit 56e00f75 authored by Chocobozzz's avatar Chocobozzz
Browse files

Improve move to object storage scripts

parent ea250b19
......@@ -47,6 +47,7 @@
"create-transcoding-job": "node ./dist/scripts/create-transcoding-job.js",
"regenerate-thumbnails": "node ./dist/scripts/regenerate-thumbnails.js",
"create-import-video-file-job": "node ./dist/scripts/create-import-video-file-job.js",
"create-move-video-storage-job": "node ./dist/scripts/create-move-video-storage-job.js",
"print-transcode-command": "node ./dist/scripts/print-transcode-command.js",
"test": "bash ./scripts/test.sh",
"help": "bash ./scripts/help.sh",
......
......@@ -47,7 +47,7 @@ async function run () {
filePath: resolve(options.import)
}
JobQueue.Instance.init()
JobQueue.Instance.init(true)
await JobQueue.Instance.createJobWithPromise({ type: 'video-file-import', payload: dataInput })
console.log('Import job for video %s created.', video.uuid)
}
import { registerTSPaths } from '../server/helpers/register-ts-paths'
registerTSPaths()
import { program } from 'commander'
import { VideoModel } from '@server/models/video/video'
import { initDatabaseModels } from '@server/initializers/database'
import { VideoStorage } from '@shared/models'
import { moveToExternalStorageState } from '@server/lib/video-state'
import { JobQueue } from '@server/lib/job-queue'
import { CONFIG } from '@server/initializers/config'
program
.description('Move videos to another storage.')
.option('-o, --to-object-storage', 'Move videos in object storage')
.option('-v, --video [videoUUID]', 'Move a specific video')
.option('-a, --all-videos', 'Migrate all videos')
.parse(process.argv)
const options = program.opts()
if (!options['toObjectStorage']) {
console.error('You need to choose where to send video files.')
process.exit(-1)
}
if (!options['video'] && !options['allVideos']) {
console.error('You need to choose which videos to move.')
process.exit(-1)
}
if (options['toObjectStorage'] && !CONFIG.OBJECT_STORAGE.ENABLED) {
console.error('Object storage is not enabled on this instance.')
process.exit(-1)
}
run()
.then(() => process.exit(0))
.catch(err => console.error(err))
async function run () {
await initDatabaseModels(true)
JobQueue.Instance.init(true)
let ids: number[] = []
if (options['video']) {
const video = await VideoModel.load(options['video'])
if (!video) {
console.error('Unknown video ' + options['video'])
process.exit(-1)
}
if (video.remote === true) {
console.error('Cannot process a remote video')
process.exit(-1)
}
ids.push(video.id)
} else {
ids = await VideoModel.listLocalIds()
}
for (const id of ids) {
const videoFull = await VideoModel.loadAndPopulateAccountAndServerAndTags(id)
const files = videoFull.VideoFiles || []
const hls = videoFull.getHLSPlaylist()
if (files.some(f => f.storage === VideoStorage.FILE_SYSTEM) || hls?.storage === VideoStorage.FILE_SYSTEM) {
console.log('Processing video %s.', videoFull.name)
await moveToExternalStorageState(videoFull, false, undefined)
}
console.log(`Created move-to-object-storage job for ${videoFull.name}.`)
}
}
......@@ -91,7 +91,7 @@ async function run () {
}
}
JobQueue.Instance.init()
JobQueue.Instance.init(true)
video.state = VideoState.TO_TRANSCODE
await video.save()
......
import { registerTSPaths } from '../server/helpers/register-ts-paths'
registerTSPaths()
import { program } from 'commander'
import { VideoModel } from '@server/models/video/video'
import { initDatabaseModels } from '@server/initializers/database'
import { VideoState, VideoStorage } from '@shared/models'
import { moveToNextState } from '@server/lib/video-state'
import { JobQueue } from '@server/lib/job-queue'
program
.description('Migrate videos to object storage.')
.parse(process.argv)
run()
.then(() => process.exit(0))
.catch(err => console.error(err))
async function run () {
await initDatabaseModels(true)
JobQueue.Instance.init()
const videos = await VideoModel.listLocal()
const withFiles = (await Promise.all(videos.map(video => VideoModel.loadWithFiles(video.id))))
.filter(video =>
video.VideoFiles.find(vf => vf.storage === VideoStorage.FILE_SYSTEM) ||
video.VideoStreamingPlaylists.find(sp => sp.storage === VideoStorage.FILE_SYSTEM)
)
for (const video of withFiles) {
video.state = VideoState.TO_MIGRATE_TO_EXTERNAL_STORAGE
await video.save()
await moveToNextState(video, false)
console.log(`Created move-to-object-storage job for ${video.name}.`)
}
}
import { registerTSPaths } from '../server/helpers/register-ts-paths'
registerTSPaths()
import { program } from 'commander'
import { VideoModel } from '@server/models/video/video'
import { initDatabaseModels } from '@server/initializers/database'
import { JobQueue } from '@server/lib/job-queue'
import { VideoStreamingPlaylistModel } from '../server/models/video/video-streaming-playlist'
import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getHlsResolutionPlaylistFilename } from '@server/lib/paths'
import { VideoPathManager } from '@server/lib/video-path-manager'
import { move } from 'fs-extra'
import { join } from 'path'
import { updateMasterHLSPlaylist, updateSha256VODSegments } from '@server/lib/hls'
import { MStreamingPlaylistFilesVideo } from '@server/types/models'
program
.description('Migrate HLS videos to get unique playlist filenames.')
.parse(process.argv)
run()
.then(() => process.exit(0))
.catch(err => console.error(err))
async function run () {
await initDatabaseModels(true)
JobQueue.Instance.init()
const videos = await VideoModel.listLocal()
const withFiles = (await Promise.all(videos.map(video => VideoModel.loadWithFiles(video.id))))
.filter(video =>
video.VideoStreamingPlaylists.find(sp => sp.playlistFilename === 'master.m3u8')
)
console.log(`Found ${withFiles.length} videos with old naming convention.`)
for (const video of withFiles) {
const playlist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
const hlsOutputPath = VideoPathManager.Instance.getFSHLSOutputPath(video)
playlist.playlistFilename = generateHLSMasterPlaylistFilename(video.isLive)
playlist.segmentsSha256Filename = generateHlsSha256SegmentsFilename(video.isLive)
const moveMap = video.VideoStreamingPlaylists.map(sp =>
sp.VideoFiles.map(vf => ({
[join(hlsOutputPath, `${vf.resolution}.m3u8`)]: join(hlsOutputPath, getHlsResolutionPlaylistFilename(vf.filename))
}))
)
.flat()
.reduce((acc, val) => ({
...acc,
...val
}), {
[join(hlsOutputPath, 'segments-sha256.json')]: join(hlsOutputPath, playlist.segmentsSha256Filename),
[join(hlsOutputPath, 'master.m3u8')]: join(hlsOutputPath, playlist.playlistFilename)
})
const moved = []
try {
for (const from of Object.keys(moveMap)) {
await move(from, moveMap[from])
moved.push(from)
}
await playlist.save()
console.log(`Successfully moved ${video.id} with ${moved.length} files.`)
} catch (error) {
console.error(error)
console.error(`Failed to migrate video ${video.id}.`)
if (moved.length > 0) {
continue
}
console.log(`Trying to restore ${moved.length} files.`)
moved.forEach(async from => {
try {
await move(moveMap[from], from)
console.log(`Successfully restored ${from}.`)
} catch (err) {
console.error(err)
console.error(`Failed to restore ${from}.`)
}
})
}
try {
const playlistWithFiles = playlist as MStreamingPlaylistFilesVideo
playlistWithFiles.Video = await playlist.$get('Video')
playlistWithFiles.VideoFiles = await playlist.$get('VideoFiles')
await updateMasterHLSPlaylist(video, playlistWithFiles)
await updateSha256VODSegments(video, playlistWithFiles)
console.log(`Successfully updated master playlist and SHA 256 segments for video ${video.id}`)
} catch (error) {
console.error(`Failed to update master playlist and SHA 256 segments for video ${video.id}`, error)
}
continue
}
}
import { registerTSPaths } from '../../server/helpers/register-ts-paths'
registerTSPaths()
import { join } from 'path'
import { JobQueue } from '@server/lib/job-queue'
import { initDatabaseModels } from '../../server/initializers/database'
import { generateHLSMasterPlaylistFilename, generateHlsSha256SegmentsFilename, getHlsResolutionPlaylistFilename } from '@server/lib/paths'
import { VideoPathManager } from '@server/lib/video-path-manager'
import { VideoModel } from '@server/models/video/video'
import { VideoStreamingPlaylistModel } from '@server/models/video/video-streaming-playlist'
import { move, readFile, writeFile } from 'fs-extra'
import Bluebird from 'bluebird'
import { federateVideoIfNeeded } from '@server/lib/activitypub/videos'
run()
.then(() => process.exit(0))
.catch(err => {
console.error(err)
process.exit(-1)
})
async function run () {
console.log('Migrate old HLS paths to new format.')
await initDatabaseModels(true)
JobQueue.Instance.init(true)
const ids = await VideoModel.listLocalIds()
await Bluebird.map(ids, async id => {
try {
await processVideo(id)
} catch (err) {
console.error('Cannot process video %s.', { err })
}
}, { concurrency: 5 })
console.log('Migration finished!')
}
async function processVideo (videoId: number) {
const video = await VideoModel.loadWithFiles(videoId)
const hls = video.getHLSPlaylist()
if (!hls || hls.playlistFilename !== 'master.m3u8' || hls.VideoFiles.length === 0) {
return
}
console.log(`Renaming HLS playlist files of video ${video.name}.`)
const playlist = await VideoStreamingPlaylistModel.loadHLSPlaylistByVideo(video.id)
const hlsDirPath = VideoPathManager.Instance.getFSHLSOutputPath(video)
const masterPlaylistPath = join(hlsDirPath, playlist.playlistFilename)
let masterPlaylistContent = await readFile(masterPlaylistPath, 'utf8')
for (const videoFile of hls.VideoFiles) {
const srcName = `${videoFile.resolution}.m3u8`
const dstName = getHlsResolutionPlaylistFilename(videoFile.filename)
const src = join(hlsDirPath, srcName)
const dst = join(hlsDirPath, dstName)
try {
await move(src, dst)
masterPlaylistContent = masterPlaylistContent.replace(new RegExp('^' + srcName + '$', 'm'), dstName)
} catch (err) {
console.error('Cannot move video file %s to %s.', src, dst, err)
}
}
await writeFile(masterPlaylistPath, masterPlaylistContent)
if (playlist.segmentsSha256Filename === 'segments-sha256.json') {
try {
const newName = generateHlsSha256SegmentsFilename(video.isLive)
const dst = join(hlsDirPath, newName)
await move(join(hlsDirPath, playlist.segmentsSha256Filename), dst)
playlist.segmentsSha256Filename = newName
} catch (err) {
console.error(`Cannot rename ${video.name} segments-sha256.json file to a new name`, err)
}
}
if (playlist.playlistFilename === 'master.m3u8') {
try {
const newName = generateHLSMasterPlaylistFilename(video.isLive)
const dst = join(hlsDirPath, newName)
await move(join(hlsDirPath, playlist.playlistFilename), dst)
playlist.playlistFilename = newName
} catch (err) {
console.error(`Cannot rename ${video.name} master.m3u8 file to a new name`, err)
}
}
// Everything worked, we can save the playlist now
await playlist.save()
const allVideo = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.id)
await federateVideoIfNeeded(allVideo, false)
console.log(`Successfully moved HLS files of ${video.name}.`)
}
......@@ -7,7 +7,6 @@ import { pathExists, remove } from 'fs-extra'
import { generateImageFilename, processImage } from '@server/helpers/image-utils'
import { THUMBNAILS_SIZE } from '@server/initializers/constants'
import { VideoModel } from '@server/models/video/video'
import { MVideo } from '@server/types/models'
import { initDatabaseModels } from '@server/initializers/database'
program
......@@ -21,16 +20,16 @@ run()
async function run () {
await initDatabaseModels(true)
const videos = await VideoModel.listLocal()
const ids = await VideoModel.listLocalIds()
await map(videos, v => {
return processVideo(v)
.catch(err => console.error('Cannot process video %s.', v.url, err))
await map(ids, id => {
return processVideo(id)
.catch(err => console.error('Cannot process video %d.', id, err))
}, { concurrency: 20 })
}
async function processVideo (videoArg: MVideo) {
const video = await VideoModel.loadWithFiles(videoArg.id)
async function processVideo (id: number) {
const video = await VideoModel.loadWithFiles(id)
console.log('Processing video %s.', video.name)
......
......@@ -115,9 +115,9 @@ async function run () {
console.log('Updating video and torrent files.')
const localVideos = await VideoModel.listLocal()
for (const localVideo of localVideos) {
const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(localVideo.id)
const ids = await VideoModel.listLocalIds()
for (const id of ids) {
const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(id)
console.log('Updating video ' + video.uuid)
......
......@@ -419,8 +419,7 @@ const VIDEO_STATES: { [ id in VideoState ]: string } = {
[VideoState.TO_IMPORT]: 'To import',
[VideoState.WAITING_FOR_LIVE]: 'Waiting for livestream',
[VideoState.LIVE_ENDED]: 'Livestream ended',
[VideoState.TO_MOVE_TO_EXTERNAL_STORAGE]: 'To move to an external storage',
[VideoState.TO_MIGRATE_TO_EXTERNAL_STORAGE]: 'To migrate to an external storage'
[VideoState.TO_MOVE_TO_EXTERNAL_STORAGE]: 'To move to an external storage'
}
const VIDEO_IMPORT_STATES: { [ id in VideoImportState ]: string } = {
......
import { close, ensureDir, move, open, outputJSON, read, readFile, remove, stat, writeFile } from 'fs-extra'
import { flatten, uniq } from 'lodash'
import { basename, dirname, join } from 'path'
import { MStreamingPlaylistFilesVideo, MVideoWithFile } from '@server/types/models'
import { MStreamingPlaylistFilesVideo, MVideo, MVideoUUID } from '@server/types/models'
import { sha256 } from '../helpers/core-utils'
import { getAudioStreamCodec, getVideoStreamCodec, getVideoStreamSize } from '../helpers/ffprobe-utils'
import { logger } from '../helpers/logger'
......@@ -31,7 +31,7 @@ async function updateStreamingPlaylistsInfohashesIfNeeded () {
}
}
async function updateMasterHLSPlaylist (video: MVideoWithFile, playlist: MStreamingPlaylistFilesVideo) {
async function updateMasterHLSPlaylist (video: MVideo, playlist: MStreamingPlaylistFilesVideo) {
const masterPlaylists: string[] = [ '#EXTM3U', '#EXT-X-VERSION:3' ]
for (const file of playlist.VideoFiles) {
......@@ -63,7 +63,7 @@ async function updateMasterHLSPlaylist (video: MVideoWithFile, playlist: MStream
})
}
async function updateSha256VODSegments (video: MVideoWithFile, playlist: MStreamingPlaylistFilesVideo) {
async function updateSha256VODSegments (video: MVideoUUID, playlist: MStreamingPlaylistFilesVideo) {
const json: { [filename: string]: { [range: string]: string } } = {}
// For all the resolutions available for this video
......
......@@ -108,7 +108,7 @@ class JobQueue {
private constructor () {
}
init () {
init (produceOnly = false) {
// Already initialized
if (this.initialized === true) return
this.initialized = true
......@@ -124,6 +124,12 @@ class JobQueue {
for (const handlerName of (Object.keys(handlers) as JobType[])) {
const queue = new Bull(handlerName, queueOptions)
if (produceOnly) {
queue.pause(true)
.catch(err => logger.error('Cannot pause queue %s in produced only job queue', handlerName, { err }))
}
const handler = handlers[handlerName]
queue.process(this.getJobConcurrency(handlerName), handler)
......
......@@ -11,10 +11,6 @@ import { Notifier } from './notifier'
import { addMoveToObjectStorageJob } from './video'
function buildNextVideoState (currentState?: VideoState) {
if (currentState === VideoState.TO_MIGRATE_TO_EXTERNAL_STORAGE) {
return VideoState.TO_MOVE_TO_EXTERNAL_STORAGE
}
if (currentState === VideoState.PUBLISHED) {
throw new Error('Video is already in its final state')
}
......@@ -61,10 +57,29 @@ function moveToNextState (video: MVideoUUID, isNewVideo = true) {
})
}
async function moveToExternalStorageState (video: MVideoFullLight, isNewVideo: boolean, transaction: Transaction) {
const videoJobInfo = await VideoJobInfoModel.load(video.id, transaction)
const pendingTranscode = videoJobInfo?.pendingTranscode || 0
// We want to wait all transcoding jobs before moving the video on an external storage
if (pendingTranscode !== 0) return
await video.setNewState(VideoState.TO_MOVE_TO_EXTERNAL_STORAGE, isNewVideo, transaction)
logger.info('Creating external storage move job for video %s.', video.uuid, { tags: [ video.uuid ] })
try {
await addMoveToObjectStorageJob(video, isNewVideo)
} catch (err) {
logger.error('Cannot add move to object storage job', { err })
}
}
// ---------------------------------------------------------------------------
export {
buildNextVideoState,
moveToExternalStorageState,
moveToNextState
}
......@@ -86,18 +101,3 @@ async function moveToPublishedState (video: MVideoFullLight, isNewVideo: boolean
Notifier.Instance.notifyOnVideoPublishedAfterTranscoding(video)
}
}
async function moveToExternalStorageState (video: MVideoFullLight, isNewVideo: boolean, transaction: Transaction) {
const videoJobInfo = await VideoJobInfoModel.load(video.id, transaction)
const pendingTranscode = videoJobInfo?.pendingTranscode || 0
// We want to wait all transcoding jobs before moving the video on an external storage
if (pendingTranscode !== 0) return
await video.setNewState(VideoState.TO_MOVE_TO_EXTERNAL_STORAGE, isNewVideo, transaction)
logger.info('Creating external storage move job for video %s.', video.uuid, { tags: [ video.uuid ] })
addMoveToObjectStorageJob(video, isNewVideo)
.catch(err => logger.error('Cannot add move to object storage job', { err }))
}
......@@ -805,14 +805,17 @@ export class VideoModel extends Model<Partial<AttributesOnly<VideoModel>>> {
await Promise.all(tasks)
}
static listLocal (): Promise<MVideo[]> {
static listLocalIds (): Promise<number[]> {
const query = {
attributes: [ 'id' ],
raw: true,
where: {
remote: false
}
}
return VideoModel.findAll(query)
.then(rows => rows.map(r => r.id))
}
static listAllAndSharedByActorForOutbox (actorId: number, start: number, count: number) {
......@@ -1674,6 +1677,8 @@ export class VideoModel extends Model<Partial<AttributesOnly<VideoModel>>> {
if (!this.VideoStreamingPlaylists) return undefined
const playlist = this.VideoStreamingPlaylists.find(p => p.type === VideoStreamingPlaylistType.HLS)
if (!playlist) return undefined
playlist.Video = this
return playlist
......@@ -1785,7 +1790,7 @@ export class VideoModel extends Model<Partial<AttributesOnly<VideoModel>>> {
await this.save({ transaction })
}
getBandwidthBits (videoFile: MVideoFile) {
getBandwidthBits (this: MVideo, videoFile: MVideoFile) {
return Math.ceil((videoFile.size * 8) / this.duration)
}
......
/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
import 'mocha'
import {
areObjectStorageTestsDisabled,
cleanupTests,
createMultipleServers,
doubleFollow,
expectStartWith,
makeRawRequest,
ObjectStorageCommand,
PeerTubeServer,
setAccessTokensToServers,
waitJobs
} from '@shared/extra-utils'
import { HttpStatusCode, VideoDetails } from '@shared/models'
async function checkFiles (origin: PeerTubeServer, video: VideoDetails, inObjectStorage: boolean) {
for (const file of video.files) {
const start = inObjectStorage
? ObjectStorageCommand.getWebTorrentBaseUrl()
: origin.url
expectStartWith(file.fileUrl, start)
await makeRawRequest(file.fileUrl, HttpStatusCode.OK_200)
}
const start = inObjectStorage
? ObjectStorageCommand.getPlaylistBaseUrl()
: origin.url