Verified Commit 866f912d authored by Chocobozzz's avatar Chocobozzz
Browse files

Refactor video views

Introduce viewers attribute for live videos
Count views for live videos
Reduce delay to see the viewer update for lives
Add ability to configure video views buffer interval and view ip
expiration
parent c49c366a
......@@ -36,7 +36,7 @@ export class JobsComponent extends RestTable implements OnInit {
'video-live-ending',
'video-redundancy',
'video-transcoding',
'videos-views',
'videos-views-stats',
'move-to-object-storage'
]
......
......@@ -658,7 +658,7 @@ export class VideoWatchComponent implements OnInit, OnDestroy {
return this.peertubeSocket.getLiveVideosObservable()
.subscribe(({ type, payload }) => {
if (type === 'state-change') return this.handleLiveStateChange(payload.state)
if (type === 'views-change') return this.handleLiveViewsChange(payload.views)
if (type === 'views-change') return this.handleLiveViewsChange(payload.viewers)
})
}
......@@ -677,7 +677,7 @@ export class VideoWatchComponent implements OnInit, OnDestroy {
this.loadVideo(videoUUID)
}
private handleLiveViewsChange (newViews: number) {
private handleLiveViewsChange (newViewers: number) {
if (!this.video) {
console.error('Cannot update video live views because video is no defined.')
return
......@@ -685,7 +685,7 @@ export class VideoWatchComponent implements OnInit, OnDestroy {
console.log('Updating live views.')
this.video.views = newViews
this.video.viewers = newViewers
}
private initHotkeys () {
......
......@@ -57,6 +57,9 @@ export class Video implements VideoServerModel {
url: string
views: number
// If live
viewers?: number
likes: number
dislikes: number
nsfw: boolean
......@@ -150,6 +153,7 @@ export class Video implements VideoServerModel {
this.url = hash.url
this.views = hash.views
this.viewers = hash.viewers
this.likes = hash.likes
this.dislikes = hash.dislikes
......
......@@ -4,6 +4,6 @@
</ng-container>
<ng-container i18n *ngIf="video.isLive">
{video.views, plural, =1 {1 viewer} other {{{ video.views | myNumberFormatter }} viewers}}
{video.viewers, plural, =1 {1 viewer} other {{{ video.viewers | myNumberFormatter }} viewers}}
</ng-container>
</span>
......@@ -232,6 +232,11 @@ views:
remote:
max_age: '30 days'
# PeerTube buffers local video views before updating and federating the video
local_buffer_update_interval: '30 minutes'
ip_view_expiration: '1 hour'
plugins:
# The website PeerTube will ask for available PeerTube plugins and themes
# This is an unmoderated plugin index, so only install plugins/themes you trust
......
......@@ -230,6 +230,11 @@ views:
remote:
max_age: '30 days'
# PeerTube buffers local video views before updating and federating the video
local_buffer_update_interval: '30 minutes'
ip_view_expiration: '1 hour'
plugins:
# The website PeerTube will ask for available PeerTube plugins and themes
# This is an unmoderated plugin index, so only install plugins/themes you trust
......
......@@ -160,3 +160,6 @@ views:
videos:
remote:
max_age: -1
local_buffer_update_interval: '5 seconds'
ip_view_expiration: '1 second'
......@@ -117,6 +117,7 @@ import { VideosRedundancyScheduler } from './server/lib/schedulers/videos-redund
import { RemoveOldHistoryScheduler } from './server/lib/schedulers/remove-old-history-scheduler'
import { AutoFollowIndexInstances } from './server/lib/schedulers/auto-follow-index-instances'
import { RemoveDanglingResumableUploadsScheduler } from './server/lib/schedulers/remove-dangling-resumable-uploads-scheduler'
import { VideoViewsBufferScheduler } from './server/lib/schedulers/video-views-buffer-scheduler'
import { isHTTPSignatureDigestValid } from './server/helpers/peertube-crypto'
import { PeerTubeSocket } from './server/lib/peertube-socket'
import { updateStreamingPlaylistsInfohashesIfNeeded } from './server/lib/hls'
......@@ -128,6 +129,7 @@ import { LiveManager } from './server/lib/live'
import { HttpStatusCode } from './shared/models/http/http-error-codes'
import { VideosTorrentCache } from '@server/lib/files-cache/videos-torrent-cache'
import { ServerConfigManager } from '@server/lib/server-config-manager'
import { VideoViews } from '@server/lib/video-views'
// ----------- Command line -----------
......@@ -296,11 +298,11 @@ async function startApplication () {
PeerTubeVersionCheckScheduler.Instance.enable()
AutoFollowIndexInstances.Instance.enable()
RemoveDanglingResumableUploadsScheduler.Instance.enable()
VideoViewsBufferScheduler.Instance.enable()
// Redis initialization
Redis.Instance.init()
PeerTubeSocket.Instance.init(server)
VideoViews.Instance.init()
updateStreamingPlaylistsInfohashesIfNeeded()
.catch(err => logger.error('Cannot update streaming playlist infohashes.', { err }))
......
......@@ -2,7 +2,7 @@ import express from 'express'
import toInt from 'validator/lib/toInt'
import { pickCommonVideoQuery } from '@server/helpers/query'
import { doJSONRequest } from '@server/helpers/requests'
import { LiveManager } from '@server/lib/live'
import { VideoViews } from '@server/lib/video-views'
import { openapiOperationDoc } from '@server/middlewares/doc'
import { getServerActor } from '@server/models/application/application'
import { guessAdditionalAttributesFromQuery } from '@server/models/video/formatter/video-format-utils'
......@@ -17,7 +17,6 @@ import { sequelizeTypescript } from '../../../initializers/database'
import { sendView } from '../../../lib/activitypub/send/send-view'
import { JobQueue } from '../../../lib/job-queue'
import { Hooks } from '../../../lib/plugins/hooks'
import { Redis } from '../../../lib/redis'
import {
asyncMiddleware,
asyncRetryTransactionMiddleware,
......@@ -107,7 +106,7 @@ videosRouter.get('/:id',
)
videosRouter.post('/:id/views',
openapiOperationDoc({ operationId: 'addView' }),
asyncMiddleware(videosCustomGetValidator('only-immutable-attributes')),
asyncMiddleware(videosCustomGetValidator('only-video')),
asyncMiddleware(viewVideo)
)
......@@ -153,44 +152,17 @@ function getVideo (_req: express.Request, res: express.Response) {
}
async function viewVideo (req: express.Request, res: express.Response) {
const immutableVideoAttrs = res.locals.onlyImmutableVideo
const video = res.locals.onlyVideo
const ip = req.ip
const exists = await Redis.Instance.doesVideoIPViewExist(ip, immutableVideoAttrs.uuid)
if (exists) {
logger.debug('View for ip %s and video %s already exists.', ip, immutableVideoAttrs.uuid)
return res.status(HttpStatusCode.NO_CONTENT_204).end()
}
const video = await VideoModel.load(immutableVideoAttrs.id)
const promises: Promise<any>[] = [
Redis.Instance.setIPVideoView(ip, video.uuid, video.isLive)
]
let federateView = true
// Increment our live manager
if (video.isLive && video.isOwned()) {
LiveManager.Instance.addViewTo(video.id)
// Views of our local live will be sent by our live manager
federateView = false
}
// Increment our video views cache counter
if (!video.isLive) {
promises.push(Redis.Instance.addVideoView(video.id))
}
const success = await VideoViews.Instance.processView({ video, ip })
if (federateView) {
if (success) {
const serverActor = await getServerActor()
promises.push(sendView(serverActor, video, undefined))
}
await Promise.all(promises)
await sendView(serverActor, video, undefined)
Hooks.runAction('action:api.video.viewed', { video, ip })
Hooks.runAction('action:api.video.viewed', { video: video, ip })
}
return res.status(HttpStatusCode.NO_CONTENT_204).end()
}
......
......@@ -38,7 +38,7 @@ function checkMissedConfig () {
'services.twitter.username', 'services.twitter.whitelisted',
'followers.instance.enabled', 'followers.instance.manual_approval',
'tracker.enabled', 'tracker.private', 'tracker.reject_too_many_announces',
'history.videos.max_age', 'views.videos.remote.max_age',
'history.videos.max_age', 'views.videos.remote.max_age', 'views.videos.local_buffer_update_interval', 'views.videos.ip_view_expiration',
'rates_limit.login.window', 'rates_limit.login.max', 'rates_limit.ask_send_email.window', 'rates_limit.ask_send_email.max',
'theme.default',
'remote_redundancy.videos.accept_from',
......
......@@ -182,7 +182,9 @@ const CONFIG = {
VIDEOS: {
REMOTE: {
MAX_AGE: parseDurationToMs(config.get('views.videos.remote.max_age'))
}
},
LOCAL_BUFFER_UPDATE_INTERVAL: parseDurationToMs(config.get('views.videos.local_buffer_update_interval')),
IP_VIEW_EXPIRATION: parseDurationToMs(config.get('views.videos.ip_view_expiration'))
}
},
PLUGINS: {
......
......@@ -148,7 +148,7 @@ const JOB_ATTEMPTS: { [id in JobType]: number } = {
'video-import': 1,
'email': 5,
'actor-keys': 3,
'videos-views': 1,
'videos-views-stats': 1,
'activitypub-refresher': 1,
'video-redundancy': 1,
'video-live-ending': 1,
......@@ -164,7 +164,7 @@ const JOB_CONCURRENCY: { [id in Exclude<JobType, 'video-transcoding' | 'video-im
'video-file-import': 1,
'email': 5,
'actor-keys': 1,
'videos-views': 1,
'videos-views-stats': 1,
'activitypub-refresher': 1,
'video-redundancy': 1,
'video-live-ending': 10,
......@@ -181,14 +181,14 @@ const JOB_TTL: { [id in JobType]: number } = {
'video-import': 1000 * 3600 * 2, // 2 hours
'email': 60000 * 10, // 10 minutes
'actor-keys': 60000 * 20, // 20 minutes
'videos-views': undefined, // Unlimited
'videos-views-stats': undefined, // Unlimited
'activitypub-refresher': 60000 * 10, // 10 minutes
'video-redundancy': 1000 * 3600 * 3, // 3 hours
'video-live-ending': 1000 * 60 * 10, // 10 minutes
'move-to-object-storage': 1000 * 60 * 60 * 3 // 3 hours
}
const REPEAT_JOBS: { [ id: string ]: EveryRepeatOptions | CronRepeatOptions } = {
'videos-views': {
const REPEAT_JOBS: { [ id in JobType ]?: EveryRepeatOptions | CronRepeatOptions } = {
'videos-views-stats': {
cron: randomInt(1, 20) + ' * * * *' // Between 1-20 minutes past the hour
},
'activitypub-cleaner': {
......@@ -211,6 +211,7 @@ const SCHEDULER_INTERVALS_MS = {
REMOVE_OLD_JOBS: 60000 * 60, // 1 hour
UPDATE_VIDEOS: 60000, // 1 minute
YOUTUBE_DL_UPDATE: 60000 * 60 * 24, // 1 day
VIDEO_VIEWS_BUFFER_UPDATE: CONFIG.VIEWS.VIDEOS.LOCAL_BUFFER_UPDATE_INTERVAL,
CHECK_PLUGINS: CONFIG.PLUGINS.INDEX.CHECK_LATEST_VERSIONS_INTERVAL,
CHECK_PEERTUBE_VERSION: 60000 * 60 * 24, // 1 day
AUTO_FOLLOW_INDEX_INSTANCES: 60000 * 60 * 24, // 1 day
......@@ -343,8 +344,8 @@ const CONSTRAINTS_FIELDS = {
}
const VIEW_LIFETIME = {
VIDEO: 60000 * 60, // 1 hour
LIVE: 60000 * 5 // 5 minutes
VIEW: CONFIG.VIEWS.VIDEOS.IP_VIEW_EXPIRATION,
VIEWER: 60000 * 5 // 5 minutes
}
let CONTACT_FORM_LIFETIME = 60000 * 60 // 1 hour
......@@ -788,13 +789,12 @@ if (isTestInstance() === true) {
SCHEDULER_INTERVALS_MS.AUTO_FOLLOW_INDEX_INSTANCES = 5000
SCHEDULER_INTERVALS_MS.UPDATE_INBOX_STATS = 5000
SCHEDULER_INTERVALS_MS.CHECK_PEERTUBE_VERSION = 2000
REPEAT_JOBS['videos-views'] = { every: 5000 }
REPEAT_JOBS['videos-views-stats'] = { every: 5000 }
REPEAT_JOBS['activitypub-cleaner'] = { every: 5000 }
REDUNDANCY.VIDEOS.RANDOMIZED_FACTOR = 1
VIEW_LIFETIME.VIDEO = 1000 // 1 second
VIEW_LIFETIME.LIVE = 1000 * 5 // 5 second
VIEW_LIFETIME.VIEWER = 1000 * 5 // 5 second
CONTACT_FORM_LIFETIME = 1000 // 1 second
JOB_ATTEMPTS['email'] = 1
......
import { getOrCreateAPVideo } from '../videos'
import { forwardVideoRelatedActivity } from '../send/utils'
import { Redis } from '../../redis'
import { ActivityCreate, ActivityView, ViewObject } from '../../../../shared/models/activitypub'
import { VideoViews } from '@server/lib/video-views'
import { ActivityView } from '../../../../shared/models/activitypub'
import { APProcessorOptions } from '../../../types/activitypub-processor.model'
import { MActorSignature } from '../../../types/models'
import { LiveManager } from '@server/lib/live/live-manager'
import { forwardVideoRelatedActivity } from '../send/utils'
import { getOrCreateAPVideo } from '../videos'
async function processViewActivity (options: APProcessorOptions<ActivityCreate | ActivityView>) {
async function processViewActivity (options: APProcessorOptions<ActivityView>) {
const { activity, byActor } = options
return processCreateView(activity, byActor)
}
......@@ -19,10 +19,8 @@ export {
// ---------------------------------------------------------------------------
async function processCreateView (activity: ActivityView | ActivityCreate, byActor: MActorSignature) {
const videoObject = activity.type === 'View'
? activity.object
: (activity.object as ViewObject).object
async function processCreateView (activity: ActivityView, byActor: MActorSignature) {
const videoObject = activity.object
const { video } = await getOrCreateAPVideo({
videoObject,
......@@ -30,17 +28,13 @@ async function processCreateView (activity: ActivityView | ActivityCreate, byAct
allowRefresh: false
})
if (!video.isLive) {
await Redis.Instance.addVideoView(video.id)
}
const viewerExpires = activity.expires
? new Date(activity.expires)
: undefined
if (video.isOwned()) {
// Our live manager will increment the counter and send the view to followers
if (video.isLive) {
LiveManager.Instance.addViewTo(video.id)
return
}
await VideoViews.Instance.processView({ video, ip: null, viewerExpires })
if (video.isOwned()) {
// Forward the view but don't resend the activity to the sender
const exceptions = [ byActor ]
await forwardVideoRelatedActivity(activity, undefined, exceptions, video)
......
import { Transaction } from 'sequelize'
import { VideoViews } from '@server/lib/video-views'
import { MActorAudience, MVideoImmutable, MVideoUrl } from '@server/types/models'
import { ActivityAudience, ActivityView } from '../../../../shared/models/activitypub'
import { logger } from '../../../helpers/logger'
......@@ -27,7 +28,8 @@ function buildViewActivity (url: string, byActor: MActorAudience, video: MVideoU
id: url,
type: 'View' as 'View',
actor: byActor.url,
object: video.url
object: video.url,
expires: new Date(VideoViews.Instance.buildViewerExpireTime()).toISOString()
},
audience
)
......
......@@ -81,7 +81,6 @@ export class APVideoUpdater extends APVideoAbstractBuilder {
if (videoUpdated.isLive) {
PeerTubeSocket.Instance.sendVideoLiveNewState(videoUpdated)
PeerTubeSocket.Instance.sendVideoViewsUpdate(videoUpdated)
}
logger.info('Remote video with uuid %s updated', this.videoObject.uuid, this.lTags())
......
import { Redis } from '../../redis'
import { isTestInstance } from '../../../helpers/core-utils'
import { logger } from '../../../helpers/logger'
import { VideoModel } from '../../../models/video/video'
import { VideoViewModel } from '../../../models/video/video-view'
import { isTestInstance } from '../../../helpers/core-utils'
import { federateVideoIfNeeded } from '../../activitypub/videos'
import { Redis } from '../../redis'
async function processVideosViews () {
async function processVideosViewsStats () {
const lastHour = new Date()
// In test mode, we run this function multiple times per hour, so we don't want the values of the previous hour
......@@ -15,23 +14,23 @@ async function processVideosViews () {
const startDate = lastHour.setMinutes(0, 0, 0)
const endDate = lastHour.setMinutes(59, 59, 999)
const videoIds = await Redis.Instance.getVideosIdViewed(hour)
const videoIds = await Redis.Instance.listVideosViewedForStats(hour)
if (videoIds.length === 0) return
logger.info('Processing videos views in job for hour %d.', hour)
logger.info('Processing videos views stats in job for hour %d.', hour)
for (const videoId of videoIds) {
try {
const views = await Redis.Instance.getVideoViews(videoId, hour)
await Redis.Instance.deleteVideoViews(videoId, hour)
const views = await Redis.Instance.getVideoViewsStats(videoId, hour)
await Redis.Instance.deleteVideoViewsStats(videoId, hour)
if (views) {
logger.debug('Adding %d views to video %d in hour %d.', views, videoId, hour)
logger.debug('Adding %d views to video %d stats in hour %d.', views, videoId, hour)
try {
const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
const video = await VideoModel.load(videoId)
if (!video) {
logger.debug('Video %d does not exist anymore, skipping videos view addition.', videoId)
logger.debug('Video %d does not exist anymore, skipping videos view stats.', videoId)
continue
}
......@@ -41,21 +40,12 @@ async function processVideosViews () {
views,
videoId
})
if (video.isOwned()) {
// If this is a remote video, the origin instance will send us an update
await VideoModel.incrementViews(videoId, views)
// Send video update
video.views += views
await federateVideoIfNeeded(video, false)
}
} catch (err) {
logger.error('Cannot create video views for video %d in hour %d.', videoId, hour, { err })
logger.error('Cannot create video views stats for video %d in hour %d.', videoId, hour, { err })
}
}
} catch (err) {
logger.error('Cannot update video views of video %d in hour %d.', videoId, hour, { err })
logger.error('Cannot update video views stats of video %d in hour %d.', videoId, hour, { err })
}
}
}
......@@ -63,5 +53,5 @@ async function processVideosViews () {
// ---------------------------------------------------------------------------
export {
processVideosViews
processVideosViewsStats
}
......@@ -36,7 +36,7 @@ import { processVideoFileImport } from './handlers/video-file-import'
import { processVideoImport } from './handlers/video-import'
import { processVideoLiveEnding } from './handlers/video-live-ending'
import { processVideoTranscoding } from './handlers/video-transcoding'
import { processVideosViews } from './handlers/video-views'
import { processVideosViewsStats } from './handlers/video-views-stats'
type CreateJobArgument =
{ type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
......@@ -49,7 +49,7 @@ type CreateJobArgument =
{ type: 'email', payload: EmailPayload } |
{ type: 'video-import', payload: VideoImportPayload } |
{ type: 'activitypub-refresher', payload: RefreshPayload } |
{ type: 'videos-views', payload: {} } |
{ type: 'videos-views-stats', payload: {} } |
{ type: 'video-live-ending', payload: VideoLiveEndingPayload } |
{ type: 'actor-keys', payload: ActorKeysPayload } |
{ type: 'video-redundancy', payload: VideoRedundancyPayload } |
......@@ -71,7 +71,7 @@ const handlers: { [id in JobType]: (job: Job) => Promise<any> } = {
'video-transcoding': processVideoTranscoding,
'email': processEmail,
'video-import': processVideoImport,
'videos-views': processVideosViews,
'videos-views-stats': processVideosViewsStats,
'activitypub-refresher': refreshAPObject,
'video-live-ending': processVideoLiveEnding,
'actor-keys': processActorKeys,
......@@ -89,7 +89,7 @@ const jobTypes: JobType[] = [
'video-transcoding',
'video-file-import',
'video-import',
'videos-views',
'videos-views-stats',
'activitypub-refresher',
'video-redundancy',
'actor-keys',
......@@ -241,8 +241,8 @@ class JobQueue {
}
private addRepeatableJobs () {
this.queues['videos-views'].add({}, {
repeat: REPEAT_JOBS['videos-views']
this.queues['videos-views-stats'].add({}, {
repeat: REPEAT_JOBS['videos-views-stats']
}).catch(err => logger.error('Cannot add repeatable job.', { err }))
if (CONFIG.FEDERATION.VIDEOS.CLEANUP_REMOTE_INTERACTIONS) {
......
......@@ -2,7 +2,6 @@
import { readFile } from 'fs-extra'
import { createServer, Server } from 'net'
import { createServer as createServerTLS, Server as ServerTLS } from 'tls'
import { isTestInstance } from '@server/helpers/core-utils'
import {
computeResolutionsToTranscode,
ffprobePromise,
......@@ -12,7 +11,7 @@ import {
} from '@server/helpers/ffprobe-utils'
import { logger, loggerTagsFactory } from '@server/helpers/logger'
import { CONFIG, registerConfigChangedHandler } from '@server/initializers/config'
import { P2P_MEDIA_LOADER_PEER_VERSION, VIDEO_LIVE, VIEW_LIFETIME } from '@server/initializers/constants'
import { P2P_MEDIA_LOADER_PEER_VERSION, VIDEO_LIVE } from '@server/initializers/constants'
import { UserModel } from '@server/models/user/user'
import { VideoModel } from '@server/models/video/video'
import { VideoLiveModel } from '@server/models/video/video-live'
......@@ -53,8 +52,6 @@ class LiveManager {
private readonly muxingSessions = new Map<string, MuxingSession>()
private readonly videoSessions = new Map<number, string>()
// Values are Date().getTime()
private readonly watchersPerVideo = new Map<number, number[]>()
private rtmpServer: Server
private rtmpsServer: ServerTLS
......@@ -99,8 +96,6 @@ class LiveManager {
// Cleanup broken lives, that were terminated by a server restart for example
this.handleBrokenLives()
.catch(err => logger.error('Cannot handle broken lives.', { err, ...lTags() }))
setInterval(() => this.updateLiveViews(), VIEW_LIFETIME.LIVE)
}
async run () {
......@@ -184,19 +179,6 @@ class LiveManager {
this.abortSession(sessionId)
}
addViewTo (videoId: number) {
if (this.videoSessions.has(videoId) === false) return
let watchers = this.watchersPerVideo.get(videoId)
if (!watchers) {
watchers = []
this.watchersPerVideo.set(videoId, watchers)
}
watchers.push(new Date().getTime())
}
private getContext () {
return context
}
......@@ -377,7 +359,6 @@ class LiveManager {
}
private onMuxingFFmpegEnd (videoId: number) {
this.watchersPerVideo.delete(videoId)
this.videoSessions.delete(videoId)
}
......@@ -411,34 +392,6 @@ class LiveManager {
}
}
private async updateLiveViews () {
if (!this.isRunning()) return
if (!isTestInstance()) logger.info('Updating live video views.', lTags())
for (const videoId of this.watchersPerVideo.keys()) {
const notBefore = new Date().getTime() - VIEW_LIFETIME.LIVE
const watchers = this.watchersPerVideo.get(videoId)
const numWatchers = watchers.length
const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoId)
video.views = numWatchers
await video.save()
await federateVideoIfNeeded(video, false)
PeerTubeSocket.Instance.sendVideoViewsUpdate(video)
// Only keep not expired watchers
const newWatchers = watchers.filter(w => w > notBefore)
this.watchersPerVideo.set(videoId, newWatchers)
logger.debug('New live video views for %s is %d.', video.url, numWatchers, lTags())
}
}
private async handleBrokenLives () {
const videoUUIDs = await VideoModel.listPublishedLiveUUIDs()
......
import { Server as HTTPServer } from 'http'
import { Namespace, Server as SocketServer, Socket } from 'socket.io'
import { isIdValid } from '@server/helpers/custom-validators/misc'