server/models/request.js

   1 'use strict'
   2
   3 const each = require('async/each')
   4 const eachLimit = require('async/eachLimit')
   5 const map = require('lodash/map')
   6 const mongoose = require('mongoose')
   7 const waterfall = require('async/waterfall')
   8
   9 const constants = require('../initializers/constants')
  10 const logger = require('../helpers/logger')
  11 const requests = require('../helpers/requests')
  12
  13 const Pod = mongoose.model('Pod')
  14 const Video = mongoose.model('Video')
  15
  16 let timer = null
  17 let lastRequestTimestamp = 0
  18
  19 // ---------------------------------------------------------------------------
  20
  21 const RequestSchema = mongoose.Schema({
  22   request: mongoose.Schema.Types.Mixed,
  23   to: [ { type: mongoose.Schema.Types.ObjectId, ref: 'Pod' } ]
  24 })
  25
  26 RequestSchema.statics = {
  27   activate,
  28   deactivate,
  29   flush,
  30   forceSend,
  31   list,
  32   remainingMilliSeconds
  33 }
  34
  35 RequestSchema.pre('save', function (next) {
  36   const self = this
  37
  38   if (self.to.length === 0) {
  39     Pod.listAllIds(function (err, podIds) {
  40       if (err) return next(err)
  41
  42       // No friends
  43       if (podIds.length === 0) return
  44
  45       self.to = podIds
  46       return next()
  47     })
  48   } else {
  49     return next()
  50   }
  51 })
  52
  53 mongoose.model('Request', RequestSchema)
  54
  55 // ------------------------------ STATICS ------------------------------
  56
  57 function activate () {
  58   logger.info('Requests scheduler activated.')
  59   lastRequestTimestamp = Date.now()
  60
  61   const self = this
  62   timer = setInterval(function () {
  63     lastRequestTimestamp = Date.now()
  64     makeRequests.call(self)
  65   }, constants.REQUESTS_INTERVAL)
  66 }
  67
  68 function deactivate () {
  69   logger.info('Requests scheduler deactivated.')
  70   clearInterval(timer)
  71   timer = null
  72 }
  73
  74 function flush () {
  75   removeAll.call(this, function (err) {
  76     if (err) logger.error('Cannot flush the requests.', { error: err })
  77   })
  78 }
  79
  80 function forceSend () {
  81   logger.info('Force requests scheduler sending.')
  82   makeRequests.call(this)
  83 }
  84
  85 function list (callback) {
  86   this.find({ }, callback)
  87 }
  88
  89 function remainingMilliSeconds () {
  90   if (timer === null) return -1
  91
  92   return constants.REQUESTS_INTERVAL - (Date.now() - lastRequestTimestamp)
  93 }
  94
  95 // ---------------------------------------------------------------------------
  96
  97 // Make a requests to friends of a certain type
  98 function makeRequest (toPod, requestsToMake, callback) {
  99   if (!callback) callback = function () {}
 100
 101   const params = {
 102     toPod: toPod,
 103     encrypt: true, // Security
 104     sign: true, // To prove our identity
 105     method: 'POST',
 106     path: '/api/' + constants.API_VERSION + '/remote/videos',
 107     data: requestsToMake // Requests we need to make
 108   }
 109
 110   // Make multiple retry requests to all of pods
 111   // The function fire some useful callbacks
 112   requests.makeSecureRequest(params, function (err, res) {
 113     if (err || (res.statusCode !== 200 && res.statusCode !== 201 && res.statusCode !== 204)) {
 114       logger.error(
 115         'Error sending secure request to %s pod.',
 116         toPod.url,
 117         {
 118           error: err || new Error('Status code not 20x : ' + res.statusCode)
 119         }
 120       )
 121
 122       return callback(false)
 123     }
 124
 125     return callback(true)
 126   })
 127 }
 128
 129 // Make all the requests of the scheduler
 130 function makeRequests () {
 131   const self = this
 132
 133   listWithLimit.call(self, constants.REQUESTS_LIMIT, function (err, requests) {
 134     if (err) {
 135       logger.error('Cannot get the list of requests.', { err: err })
 136       return // Abort
 137     }
 138
 139     // If there are no requests, abort
 140     if (requests.length === 0) {
 141       logger.info('No requests to make.')
 142       return
 143     }
 144
 145     logger.info('Making requests to friends.')
 146
 147     // Requests by pods id
 148     const requestsToMake = {}
 149
 150     requests.forEach(function (poolRequest) {
 151       poolRequest.to.forEach(function (toPodId) {
 152         if (!requestsToMake[toPodId]) {
 153           requestsToMake[toPodId] = {
 154             ids: [],
 155             datas: []
 156           }
 157         }
 158
 159         requestsToMake[toPodId].ids.push(poolRequest._id)
 160         requestsToMake[toPodId].datas.push(poolRequest.request)
 161       })
 162     })
 163
 164     const goodPods = []
 165     const badPods = []
 166
 167     eachLimit(Object.keys(requestsToMake), constants.REQUESTS_IN_PARALLEL, function (toPodId, callbackEach) {
 168       const requestToMake = requestsToMake[toPodId]
 169
 170       // FIXME: mongodb request inside a loop :/
 171       Pod.load(toPodId, function (err, toPod) {
 172         if (err) {
 173           logger.error('Error finding pod by id.', { err: err })
 174           return callbackEach()
 175         }
 176
 177         // Maybe the pod is not our friend anymore so simply remove it
 178         if (!toPod) {
 179           logger.info('Removing %d requests of unexisting pod %s.', requestToMake.ids.length, toPodId)
 180           removePodOf.call(self, requestToMake.ids, toPodId)
 181           return callbackEach()
 182         }
 183
 184         makeRequest(toPod, requestToMake.datas, function (success) {
 185           if (success === true) {
 186             logger.debug('Removing requests for %s pod.', toPodId, { requestsIds: requestToMake.ids })
 187
 188             // Remove the pod id of these request ids
 189             removePodOf.call(self, requestToMake.ids, toPodId)
 190             goodPods.push(toPodId)
 191           } else {
 192             badPods.push(toPodId)
 193           }
 194
 195           callbackEach()
 196         })
 197       })
 198     }, function () {
 199       // All the requests were made, we update the pods score
 200       updatePodsScore(goodPods, badPods)
 201       // Flush requests with no pod
 202       removeWithEmptyTo.call(self)
 203     })
 204   })
 205 }
 206
 207 // Remove pods with a score of 0 (too many requests where they were unreachable)
 208 function removeBadPods () {
 209   waterfall([
 210     function findBadPods (callback) {
 211       Pod.listBadPods(function (err, pods) {
 212         if (err) {
 213           logger.error('Cannot find bad pods.', { error: err })
 214           return callback(err)
 215         }
 216
 217         return callback(null, pods)
 218       })
 219     },
 220
 221     function listVideosOfTheseBadPods (pods, callback) {
 222       if (pods.length === 0) return callback(null)
 223
 224       const urls = map(pods, 'url')
 225
 226       Video.listByUrls(urls, function (err, videosList) {
 227         if (err) {
 228           logger.error('Cannot list videos urls.', { error: err, urls: urls })
 229           return callback(null, pods, [])
 230         }
 231
 232         return callback(null, pods, videosList)
 233       })
 234     },
 235
 236     function removeVideosOfTheseBadPods (pods, videosList, callback) {
 237       // We don't have to remove pods, skip
 238       if (typeof pods === 'function') {
 239         callback = pods
 240         return callback(null)
 241       }
 242
 243       each(videosList, function (video, callbackEach) {
 244         video.remove(callbackEach)
 245       }, function (err) {
 246         if (err) {
 247           // Don't stop the process
 248           logger.error('Error while removing videos of bad pods.', { error: err })
 249           return
 250         }
 251
 252         return callback(null, pods)
 253       })
 254     },
 255
 256     function removeBadPodsFromDB (pods, callback) {
 257       // We don't have to remove pods, skip
 258       if (typeof pods === 'function') {
 259         callback = pods
 260         return callback(null)
 261       }
 262
 263       each(pods, function (pod, callbackEach) {
 264         pod.remove(callbackEach)
 265       }, function (err) {
 266         if (err) return callback(err)
 267
 268         return callback(null, pods.length)
 269       })
 270     }
 271   ], function (err, numberOfPodsRemoved) {
 272     if (err) {
 273       logger.error('Cannot remove bad pods.', { error: err })
 274     } else if (numberOfPodsRemoved) {
 275       logger.info('Removed %d pods.', numberOfPodsRemoved)
 276     } else {
 277       logger.info('No need to remove bad pods.')
 278     }
 279   })
 280 }
 281
 282 function updatePodsScore (goodPods, badPods) {
 283   logger.info('Updating %d good pods and %d bad pods scores.', goodPods.length, badPods.length)
 284
 285   Pod.incrementScores(goodPods, constants.PODS_SCORE.BONUS, function (err) {
 286     if (err) logger.error('Cannot increment scores of good pods.')
 287   })
 288
 289   Pod.incrementScores(badPods, constants.PODS_SCORE.MALUS, function (err) {
 290     if (err) logger.error('Cannot decrement scores of bad pods.')
 291     removeBadPods()
 292   })
 293 }
 294
 295 function listWithLimit (limit, callback) {
 296   this.find({ }, { _id: 1, request: 1, to: 1 }).sort({ _id: 1 }).limit(limit).exec(callback)
 297 }
 298
 299 function removeAll (callback) {
 300   this.remove({ }, callback)
 301 }
 302
 303 function removePodOf (requestsIds, podId, callback) {
 304   if (!callback) callback = function () {}
 305
 306   this.update({ _id: { $in: requestsIds } }, { $pull: { to: podId } }, { multi: true }, callback)
 307 }
 308
 309 function removeWithEmptyTo (callback) {
 310   if (!callback) callback = function () {}
 311
 312   this.remove({ to: { $size: 0 } }, callback)
 313 }