From 9e167724f7e933f41d9ea2e1c31772bf4c560a28 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Tue, 21 Feb 2017 21:35:59 +0100 Subject: Server: make a basic "quick and dirty update" for videos This system will be useful to to update some int video attributes (likes, dislikes, views...) The classic system is not used because we need some optimization for scaling --- server/lib/base-request-scheduler.js | 140 +++++++++++++++++++++++ server/lib/friends.js | 26 ++++- server/lib/request-scheduler.js | 178 ++++++----------------------- server/lib/request-video-qadu-scheduler.js | 116 +++++++++++++++++++ 4 files changed, 319 insertions(+), 141 deletions(-) create mode 100644 server/lib/base-request-scheduler.js create mode 100644 server/lib/request-video-qadu-scheduler.js (limited to 'server/lib') diff --git a/server/lib/base-request-scheduler.js b/server/lib/base-request-scheduler.js new file mode 100644 index 000000000..d15680c25 --- /dev/null +++ b/server/lib/base-request-scheduler.js @@ -0,0 +1,140 @@ +'use strict' + +const eachLimit = require('async/eachLimit') + +const constants = require('../initializers/constants') +const db = require('../initializers/database') +const logger = require('../helpers/logger') +const requests = require('../helpers/requests') + +module.exports = class BaseRequestScheduler { + + constructor (options) { + this.lastRequestTimestamp = 0 + this.timer = null + } + + activate () { + logger.info('Requests scheduler activated.') + this.lastRequestTimestamp = Date.now() + + this.timer = setInterval(() => { + this.lastRequestTimestamp = Date.now() + this.makeRequests() + }, constants.REQUESTS_INTERVAL) + } + + deactivate () { + logger.info('Requests scheduler deactivated.') + clearInterval(this.timer) + this.timer = null + } + + forceSend () { + logger.info('Force requests scheduler sending.') + this.makeRequests() + } + + remainingMilliSeconds () { + if (this.timer === null) return -1 + + return constants.REQUESTS_INTERVAL - (Date.now() - this.lastRequestTimestamp) + } + + // --------------------------------------------------------------------------- + + // Make a requests to friends of a certain type + makeRequest (toPod, requestEndpoint, requestsToMake, callback) { + if (!callback) callback = function () {} + + const params = { + toPod: toPod, + sign: true, // Prove our identity + method: 'POST', + path: '/api/' + constants.API_VERSION + '/remote/' + requestEndpoint, + data: requestsToMake // Requests we need to make + } + + // Make multiple retry requests to all of pods + // The function fire some useful callbacks + requests.makeSecureRequest(params, (err, res) => { + if (err || (res.statusCode !== 200 && res.statusCode !== 201 && res.statusCode !== 204)) { + err = err ? err.message : 'Status code not 20x : ' + res.statusCode + logger.error('Error sending secure request to %s pod.', toPod.host, { error: err }) + + return callback(false) + } + + return callback(true) + }) + } + + // Make all the requests of the scheduler + makeRequests () { + this.getRequestModel().listWithLimitAndRandom(this.limitPods, this.limitPerPod, (err, requests) => { + if (err) { + logger.error('Cannot get the list of "%s".', this.description, { err: err }) + return // Abort + } + + // If there are no requests, abort + if (requests.length === 0) { + logger.info('No "%s" to make.', this.description) + return + } + + // We want to group requests by destinations pod and endpoint + const requestsToMakeGrouped = this.buildRequestObjects(requests) + + logger.info('Making "%s" to friends.', this.description) + + const goodPods = [] + const badPods = [] + + eachLimit(Object.keys(requestsToMakeGrouped), constants.REQUESTS_IN_PARALLEL, (hashKey, callbackEach) => { + const requestToMake = requestsToMakeGrouped[hashKey] + const toPod = requestToMake.toPod + + // Maybe the pod is not our friend anymore so simply remove it + if (!toPod) { + const requestIdsToDelete = requestToMake.ids + + logger.info('Removing %d "%s" of unexisting pod %s.', requestIdsToDelete.length, this.description, requestToMake.toPod.id) + return this.getRequestToPodModel().removePodOf(requestIdsToDelete, requestToMake.toPod.id, callbackEach) + } + + this.makeRequest(toPod, requestToMake.endpoint, requestToMake.datas, (success) => { + if (success === false) { + badPods.push(requestToMake.toPod.id) + return callbackEach() + } + + logger.debug('Removing requests for pod %s.', requestToMake.toPod.id, { requestsIds: requestToMake.ids }) + goodPods.push(requestToMake.toPod.id) + + // Remove the pod id of these request ids + this.getRequestToPodModel().removeByRequestIdsAndPod(requestToMake.ids, requestToMake.toPod.id, callbackEach) + + this.afterRequestHook() + }) + }, () => { + // All the requests were made, we update the pods score + db.Pod.updatePodsScore(goodPods, badPods) + + this.afterRequestsHook() + }) + }) + } + + flush (callback) { + this.getRequestModel().removeAll(callback) + } + + afterRequestHook () { + // Nothing to do, let children reimplement it + } + + afterRequestsHook () { + // Nothing to do, let children reimplement it + } +} diff --git a/server/lib/friends.js b/server/lib/friends.js index d53ab4553..424a30801 100644 --- a/server/lib/friends.js +++ b/server/lib/friends.js @@ -12,15 +12,19 @@ const logger = require('../helpers/logger') const peertubeCrypto = require('../helpers/peertube-crypto') const requests = require('../helpers/requests') const RequestScheduler = require('./request-scheduler') +const RequestVideoQaduScheduler = require('./request-video-qadu-scheduler') const ENDPOINT_ACTIONS = constants.REQUEST_ENDPOINT_ACTIONS[constants.REQUEST_ENDPOINTS.VIDEOS] + const requestScheduler = new RequestScheduler() +const requestSchedulerVideoQadu = new RequestVideoQaduScheduler() const friends = { activate, addVideoToFriends, updateVideoToFriends, reportAbuseVideoToFriend, + quickAndDirtyUpdateVideoToFriends, hasFriends, makeFriends, quitFriends, @@ -30,6 +34,7 @@ const friends = { function activate () { requestScheduler.activate() + requestSchedulerVideoQadu.activate() } function addVideoToFriends (videoData, transaction, callback) { @@ -71,6 +76,15 @@ function reportAbuseVideoToFriend (reportData, video) { createRequest(options) } +function quickAndDirtyUpdateVideoToFriends (videoId, type, transaction, callback) { + const options = { + videoId, + type, + transaction + } + return createVideoQaduRequest(options, callback) +} + function hasFriends (callback) { db.Pod.countAll(function (err, count) { if (err) return callback(err) @@ -110,7 +124,11 @@ function quitFriends (callback) { waterfall([ function flushRequests (callbackAsync) { - requestScheduler.flush(callbackAsync) + requestScheduler.flush(err => callbackAsync(err)) + }, + + function flushVideoQaduRequests (callbackAsync) { + requestSchedulerVideoQadu.flush(err => callbackAsync(err)) }, function getPodsList (callbackAsync) { @@ -310,6 +328,12 @@ function createRequest (options, callback) { }) } +function createVideoQaduRequest (options, callback) { + if (!callback) callback = function () {} + + requestSchedulerVideoQadu.createRequest(options, callback) +} + function isMe (host) { return host === constants.CONFIG.WEBSERVER.HOST } diff --git a/server/lib/request-scheduler.js b/server/lib/request-scheduler.js index 28dabe339..6b6535519 100644 --- a/server/lib/request-scheduler.js +++ b/server/lib/request-scheduler.js @@ -1,44 +1,54 @@ 'use strict' -const eachLimit = require('async/eachLimit') - const constants = require('../initializers/constants') +const BaseRequestScheduler = require('./base-request-scheduler') const db = require('../initializers/database') const logger = require('../helpers/logger') -const requests = require('../helpers/requests') -module.exports = class RequestScheduler { +module.exports = class RequestScheduler extends BaseRequestScheduler { constructor () { - this.lastRequestTimestamp = 0 - this.timer = null - } + super() - activate () { - logger.info('Requests scheduler activated.') - this.lastRequestTimestamp = Date.now() + // We limit the size of the requests + this.limitPods = constants.REQUESTS_LIMIT_PODS + this.limitPerPod = constants.REQUESTS_LIMIT_PER_POD - this.timer = setInterval(() => { - this.lastRequestTimestamp = Date.now() - this.makeRequests() - }, constants.REQUESTS_INTERVAL) + this.description = 'requests' } - deactivate () { - logger.info('Requests scheduler deactivated.') - clearInterval(this.timer) - this.timer = null + getRequestModel () { + return db.Request } - forceSend () { - logger.info('Force requests scheduler sending.') - this.makeRequests() + getRequestToPodModel () { + return db.RequestToPod } - remainingMilliSeconds () { - if (this.timer === null) return -1 + buildRequestObjects (requests) { + const requestsToMakeGrouped = {} + + Object.keys(requests).forEach(toPodId => { + requests[toPodId].forEach(data => { + const request = data.request + const pod = data.pod + const hashKey = toPodId + request.endpoint + + if (!requestsToMakeGrouped[hashKey]) { + requestsToMakeGrouped[hashKey] = { + toPod: pod, + endpoint: request.endpoint, + ids: [], // request ids, to delete them from the DB in the future + datas: [] // requests data, + } + } + + requestsToMakeGrouped[hashKey].ids.push(request.id) + requestsToMakeGrouped[hashKey].datas.push(request.request) + }) + }) - return constants.REQUESTS_INTERVAL - (Date.now() - this.lastRequestTimestamp) + return requestsToMakeGrouped } // { type, endpoint, data, toIds, transaction } @@ -79,122 +89,10 @@ module.exports = class RequestScheduler { // --------------------------------------------------------------------------- - // Make all the requests of the scheduler - makeRequests () { - // We limit the size of the requests - // We don't want to stuck with the same failing requests so we get a random list - db.Request.listWithLimitAndRandom(constants.REQUESTS_LIMIT_PODS, constants.REQUESTS_LIMIT_PER_POD, (err, requests) => { - if (err) { - logger.error('Cannot get the list of requests.', { err: err }) - return // Abort - } - - // If there are no requests, abort - if (requests.length === 0) { - logger.info('No requests to make.') - return - } - - // We want to group requests by destinations pod and endpoint - const requestsToMakeGrouped = this.buildRequestObjects(requests) - - logger.info('Making requests to friends.') - - const goodPods = [] - const badPods = [] - - eachLimit(Object.keys(requestsToMakeGrouped), constants.REQUESTS_IN_PARALLEL, (hashKey, callbackEach) => { - const requestToMake = requestsToMakeGrouped[hashKey] - const toPod = requestToMake.toPod - - // Maybe the pod is not our friend anymore so simply remove it - if (!toPod) { - const requestIdsToDelete = requestToMake.ids - - logger.info('Removing %d requests of unexisting pod %s.', requestIdsToDelete.length, requestToMake.toPod.id) - return db.RequestToPod.removePodOf(requestIdsToDelete, requestToMake.toPod.id, callbackEach) - } - - this.makeRequest(toPod, requestToMake.endpoint, requestToMake.datas, (success) => { - if (success === false) { - badPods.push(requestToMake.toPod.id) - return callbackEach() - } - - logger.debug('Removing requests for pod %s.', requestToMake.toPod.id, { requestsIds: requestToMake.ids }) - goodPods.push(requestToMake.toPod.id) - - // Remove the pod id of these request ids - db.RequestToPod.removePodOf(requestToMake.ids, requestToMake.toPod.id, callbackEach) - }) - }, () => { - // All the requests were made, we update the pods score - db.Request.updatePodsScore(goodPods, badPods) - // Flush requests with no pod - db.Request.removeWithEmptyTo(err => { - if (err) logger.error('Error when removing requests with no pods.', { error: err }) - }) - }) - }) - } - - // Make a requests to friends of a certain type - makeRequest (toPod, requestEndpoint, requestsToMake, callback) { - if (!callback) callback = function () {} - - const params = { - toPod: toPod, - sign: true, // Prove our identity - method: 'POST', - path: '/api/' + constants.API_VERSION + '/remote/' + requestEndpoint, - data: requestsToMake // Requests we need to make - } - - // Make multiple retry requests to all of pods - // The function fire some useful callbacks - requests.makeSecureRequest(params, (err, res) => { - if (err || (res.statusCode !== 200 && res.statusCode !== 201 && res.statusCode !== 204)) { - err = err ? err.message : 'Status code not 20x : ' + res.statusCode - logger.error('Error sending secure request to %s pod.', toPod.host, { error: err }) - - return callback(false) - } - - return callback(true) - }) - } - - buildRequestObjects (requests) { - const requestsToMakeGrouped = {} - - Object.keys(requests).forEach(toPodId => { - requests[toPodId].forEach(data => { - const request = data.request - const pod = data.pod - const hashKey = toPodId + request.endpoint - - if (!requestsToMakeGrouped[hashKey]) { - requestsToMakeGrouped[hashKey] = { - toPod: pod, - endpoint: request.endpoint, - ids: [], // request ids, to delete them from the DB in the future - datas: [] // requests data, - } - } - - requestsToMakeGrouped[hashKey].ids.push(request.id) - requestsToMakeGrouped[hashKey].datas.push(request.request) - }) - }) - - return requestsToMakeGrouped - } - - flush (callback) { - db.Request.removeAll(err => { - if (err) logger.error('Cannot flush the requests.', { error: err }) - - return callback(err) + afterRequestsHook () { + // Flush requests with no pod + this.getRequestModel().removeWithEmptyTo(err => { + if (err) logger.error('Error when removing requests with no pods.', { error: err }) }) } } diff --git a/server/lib/request-video-qadu-scheduler.js b/server/lib/request-video-qadu-scheduler.js new file mode 100644 index 000000000..401b2fb44 --- /dev/null +++ b/server/lib/request-video-qadu-scheduler.js @@ -0,0 +1,116 @@ +'use strict' + +const BaseRequestScheduler = require('./base-request-scheduler') +const constants = require('../initializers/constants') +const db = require('../initializers/database') +const logger = require('../helpers/logger') + +module.exports = class RequestVideoQaduScheduler extends BaseRequestScheduler { + + constructor () { + super() + + // We limit the size of the requests + this.limitPods = constants.REQUESTS_VIDEO_QADU_LIMIT_PODS + this.limitPerPod = constants.REQUESTS_VIDEO_QADU_LIMIT_PODS + + this.description = 'video QADU requests' + } + + getRequestModel () { + return db.RequestVideoQadu + } + + getRequestToPodModel () { + return db.RequestVideoQadu + } + + buildRequestObjects (requests) { + const requestsToMakeGrouped = {} + + Object.keys(requests).forEach(toPodId => { + requests[toPodId].forEach(data => { + const request = data.request + const video = data.video + const pod = data.pod + const hashKey = toPodId + + if (!requestsToMakeGrouped[hashKey]) { + requestsToMakeGrouped[hashKey] = { + toPod: pod, + endpoint: constants.REQUEST_ENDPOINTS.QADU, + ids: [], // request ids, to delete them from the DB in the future + datas: [], // requests data + videos: {} + } + } + + if (!requestsToMakeGrouped[hashKey].videos[video.id]) { + requestsToMakeGrouped[hashKey].videos[video.id] = {} + } + + const videoData = requestsToMakeGrouped[hashKey].videos[video.id] + + switch (request.type) { + case constants.REQUEST_VIDEO_QADU_TYPES.LIKES: + videoData.likes = video.likes + break + + case constants.REQUEST_VIDEO_QADU_TYPES.DISLIKES: + videoData.likes = video.dislikes + break + + case constants.REQUEST_VIDEO_QADU_TYPES.VIEWS: + videoData.views = video.views + break + + default: + logger.error('Unknown request video QADU type %s.', request.type) + return + } + + // Do not forget the remoteId so the remote pod can identify the video + videoData.remoteId = video.id + requestsToMakeGrouped[hashKey].ids.push(request.id) + requestsToMakeGrouped[hashKey].videos[video.id] = videoData + }) + }) + + Object.keys(requestsToMakeGrouped).forEach(hashKey => { + Object.keys(requestsToMakeGrouped[hashKey].videos).forEach(videoId => { + const videoData = requestsToMakeGrouped[hashKey].videos[videoId] + + requestsToMakeGrouped[hashKey].datas.push({ + data: videoData + }) + }) + + // We don't need it anymore, it was just to build our datas array + delete requestsToMakeGrouped[hashKey].videos + }) + + return requestsToMakeGrouped + } + + // { type, videoId, transaction? } + createRequest (options, callback) { + const type = options.type + const videoId = options.videoId + const transaction = options.transaction + + const dbRequestOptions = {} + if (transaction) dbRequestOptions.transaction = transaction + + // Send the update to all our friends + db.Pod.listAllIds(options.transaction, function (err, podIds) { + if (err) return callback(err) + + const queries = [] + podIds.forEach(podId => { + queries.push({ type, videoId, podId }) + }) + + return db.RequestVideoQadu.bulkCreate(queries, dbRequestOptions).asCallback(callback) + }) + } +} -- cgit v1.2.3