server/models/request.js

   1 'use strict'
   2
   3 const each = require('async/each')
   4 const eachLimit = require('async/eachLimit')
   5 const mongoose = require('mongoose')
   6 const waterfall = require('async/waterfall')
   7
   8 const constants = require('../initializers/constants')
   9 const logger = require('../helpers/logger')
  10 const requests = require('../helpers/requests')
  11
  12 const Pod = mongoose.model('Pod')
  13
  14 let timer = null
  15 let lastRequestTimestamp = 0
  16
  17 // ---------------------------------------------------------------------------
  18
  19 const RequestSchema = mongoose.Schema({
  20   request: mongoose.Schema.Types.Mixed,
  21   to: [ { type: mongoose.Schema.Types.ObjectId, ref: 'Pod' } ]
  22 })
  23
  24 RequestSchema.statics = {
  25   activate,
  26   deactivate,
  27   flush,
  28   forceSend,
  29   list,
  30   remainingMilliSeconds
  31 }
  32
  33 RequestSchema.pre('save', function (next) {
  34   const self = this
  35
  36   if (self.to.length === 0) {
  37     Pod.listAllIds(function (err, podIds) {
  38       if (err) return next(err)
  39
  40       // No friends
  41       if (podIds.length === 0) return
  42
  43       self.to = podIds
  44       return next()
  45     })
  46   } else {
  47     return next()
  48   }
  49 })
  50
  51 mongoose.model('Request', RequestSchema)
  52
  53 // ------------------------------ STATICS ------------------------------
  54
  55 function activate () {
  56   logger.info('Requests scheduler activated.')
  57   lastRequestTimestamp = Date.now()
  58
  59   const self = this
  60   timer = setInterval(function () {
  61     lastRequestTimestamp = Date.now()
  62     makeRequests.call(self)
  63   }, constants.REQUESTS_INTERVAL)
  64 }
  65
  66 function deactivate () {
  67   logger.info('Requests scheduler deactivated.')
  68   clearInterval(timer)
  69   timer = null
  70 }
  71
  72 function flush () {
  73   removeAll.call(this, function (err) {
  74     if (err) logger.error('Cannot flush the requests.', { error: err })
  75   })
  76 }
  77
  78 function forceSend () {
  79   logger.info('Force requests scheduler sending.')
  80   makeRequests.call(this)
  81 }
  82
  83 function list (callback) {
  84   this.find({ }, callback)
  85 }
  86
  87 function remainingMilliSeconds () {
  88   if (timer === null) return -1
  89
  90   return constants.REQUESTS_INTERVAL - (Date.now() - lastRequestTimestamp)
  91 }
  92
  93 // ---------------------------------------------------------------------------
  94
  95 // Make a requests to friends of a certain type
  96 function makeRequest (toPod, requestsToMake, callback) {
  97   if (!callback) callback = function () {}
  98
  99   const params = {
 100     toPod: toPod,
 101     encrypt: true, // Security
 102     sign: true, // To prove our identity
 103     method: 'POST',
 104     path: '/api/' + constants.API_VERSION + '/remote/videos',
 105     data: requestsToMake // Requests we need to make
 106   }
 107
 108   // Make multiple retry requests to all of pods
 109   // The function fire some useful callbacks
 110   requests.makeSecureRequest(params, function (err, res) {
 111     if (err || (res.statusCode !== 200 && res.statusCode !== 201 && res.statusCode !== 204)) {
 112       logger.error(
 113         'Error sending secure request to %s pod.',
 114         toPod.url,
 115         {
 116           error: err || new Error('Status code not 20x : ' + res.statusCode)
 117         }
 118       )
 119
 120       return callback(false)
 121     }
 122
 123     return callback(true)
 124   })
 125 }
 126
 127 // Make all the requests of the scheduler
 128 function makeRequests () {
 129   const self = this
 130
 131   // We limit the size of the requests (REQUESTS_LIMIT)
 132   // We don't want to stuck with the same failing requests so we get a random list
 133   listWithLimitAndRandom.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             goodPods.push(toPodId)
 189
 190             // Remove the pod id of these request ids
 191             removePodOf.call(self, requestToMake.ids, toPodId, callbackEach)
 192           } else {
 193             badPods.push(toPodId)
 194             callbackEach()
 195           }
 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 removeTheseBadPods (pods, callback) {
 222       if (pods.length === 0) return callback(null, 0)
 223
 224       each(pods, function (pod, callbackEach) {
 225         pod.remove(callbackEach)
 226       }, function (err) {
 227         return callback(err, pods.length)
 228       })
 229     }
 230   ], function (err, numberOfPodsRemoved) {
 231     if (err) {
 232       logger.error('Cannot remove bad pods.', { error: err })
 233     } else if (numberOfPodsRemoved) {
 234       logger.info('Removed %d pods.', numberOfPodsRemoved)
 235     } else {
 236       logger.info('No need to remove bad pods.')
 237     }
 238   })
 239 }
 240
 241 function updatePodsScore (goodPods, badPods) {
 242   logger.info('Updating %d good pods and %d bad pods scores.', goodPods.length, badPods.length)
 243
 244   Pod.incrementScores(goodPods, constants.PODS_SCORE.BONUS, function (err) {
 245     if (err) logger.error('Cannot increment scores of good pods.')
 246   })
 247
 248   Pod.incrementScores(badPods, constants.PODS_SCORE.MALUS, function (err) {
 249     if (err) logger.error('Cannot decrement scores of bad pods.')
 250     removeBadPods()
 251   })
 252 }
 253
 254 function listWithLimitAndRandom (limit, callback) {
 255   const self = this
 256
 257   self.count(function (err, count) {
 258     if (err) return callback(err)
 259
 260     let start = Math.floor(Math.random() * count) - limit
 261     if (start < 0) start = 0
 262
 263     self.find({ }, { _id: 1, request: 1, to: 1 }).sort({ _id: 1 }).skip(start).limit(limit).exec(callback)
 264   })
 265 }
 266
 267 function removeAll (callback) {
 268   this.remove({ }, callback)
 269 }
 270
 271 function removePodOf (requestsIds, podId, callback) {
 272   if (!callback) callback = function () {}
 273
 274   this.update({ _id: { $in: requestsIds } }, { $pull: { to: podId } }, { multi: true }, callback)
 275 }
 276
 277 function removeWithEmptyTo (callback) {
 278   if (!callback) callback = function () {}
 279
 280   this.remove({ to: { $size: 0 } }, callback)
 281 }