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   listWithLimit.call(self, constants.REQUESTS_LIMIT, function (err, requests) {
 132     if (err) {
 133       logger.error('Cannot get the list of requests.', { err: err })
 134       return // Abort
 135     }
 136
 137     // If there are no requests, abort
 138     if (requests.length === 0) {
 139       logger.info('No requests to make.')
 140       return
 141     }
 142
 143     logger.info('Making requests to friends.')
 144
 145     // Requests by pods id
 146     const requestsToMake = {}
 147
 148     requests.forEach(function (poolRequest) {
 149       poolRequest.to.forEach(function (toPodId) {
 150         if (!requestsToMake[toPodId]) {
 151           requestsToMake[toPodId] = {
 152             ids: [],
 153             datas: []
 154           }
 155         }
 156
 157         requestsToMake[toPodId].ids.push(poolRequest._id)
 158         requestsToMake[toPodId].datas.push(poolRequest.request)
 159       })
 160     })
 161
 162     const goodPods = []
 163     const badPods = []
 164
 165     eachLimit(Object.keys(requestsToMake), constants.REQUESTS_IN_PARALLEL, function (toPodId, callbackEach) {
 166       const requestToMake = requestsToMake[toPodId]
 167
 168       // FIXME: mongodb request inside a loop :/
 169       Pod.load(toPodId, function (err, toPod) {
 170         if (err) {
 171           logger.error('Error finding pod by id.', { err: err })
 172           return callbackEach()
 173         }
 174
 175         // Maybe the pod is not our friend anymore so simply remove it
 176         if (!toPod) {
 177           logger.info('Removing %d requests of unexisting pod %s.', requestToMake.ids.length, toPodId)
 178           removePodOf.call(self, requestToMake.ids, toPodId)
 179           return callbackEach()
 180         }
 181
 182         makeRequest(toPod, requestToMake.datas, function (success) {
 183           if (success === true) {
 184             logger.debug('Removing requests for %s pod.', toPodId, { requestsIds: requestToMake.ids })
 185
 186             goodPods.push(toPodId)
 187
 188             // Remove the pod id of these request ids
 189             removePodOf.call(self, requestToMake.ids, toPodId, callbackEach)
 190           } else {
 191             badPods.push(toPodId)
 192             callbackEach()
 193           }
 194         })
 195       })
 196     }, function () {
 197       // All the requests were made, we update the pods score
 198       updatePodsScore(goodPods, badPods)
 199       // Flush requests with no pod
 200       removeWithEmptyTo.call(self)
 201     })
 202   })
 203 }
 204
 205 // Remove pods with a score of 0 (too many requests where they were unreachable)
 206 function removeBadPods () {
 207   waterfall([
 208     function findBadPods (callback) {
 209       Pod.listBadPods(function (err, pods) {
 210         if (err) {
 211           logger.error('Cannot find bad pods.', { error: err })
 212           return callback(err)
 213         }
 214
 215         return callback(null, pods)
 216       })
 217     },
 218
 219     function removeTheseBadPods (pods, callback) {
 220       if (pods.length === 0) return callback(null, 0)
 221
 222       each(pods, function (pod, callbackEach) {
 223         pod.remove(callbackEach)
 224       }, function (err) {
 225         return callback(err, pods.length)
 226       })
 227     }
 228   ], function (err, numberOfPodsRemoved) {
 229     if (err) {
 230       logger.error('Cannot remove bad pods.', { error: err })
 231     } else if (numberOfPodsRemoved) {
 232       logger.info('Removed %d pods.', numberOfPodsRemoved)
 233     } else {
 234       logger.info('No need to remove bad pods.')
 235     }
 236   })
 237 }
 238
 239 function updatePodsScore (goodPods, badPods) {
 240   logger.info('Updating %d good pods and %d bad pods scores.', goodPods.length, badPods.length)
 241
 242   Pod.incrementScores(goodPods, constants.PODS_SCORE.BONUS, function (err) {
 243     if (err) logger.error('Cannot increment scores of good pods.')
 244   })
 245
 246   Pod.incrementScores(badPods, constants.PODS_SCORE.MALUS, function (err) {
 247     if (err) logger.error('Cannot decrement scores of bad pods.')
 248     removeBadPods()
 249   })
 250 }
 251
 252 function listWithLimit (limit, callback) {
 253   this.find({ }, { _id: 1, request: 1, to: 1 }).sort({ _id: 1 }).limit(limit).exec(callback)
 254 }
 255
 256 function removeAll (callback) {
 257   this.remove({ }, callback)
 258 }
 259
 260 function removePodOf (requestsIds, podId, callback) {
 261   if (!callback) callback = function () {}
 262
 263   this.update({ _id: { $in: requestsIds } }, { $pull: { to: podId } }, { multi: true }, callback)
 264 }
 265
 266 function removeWithEmptyTo (callback) {
 267   if (!callback) callback = function () {}
 268
 269   this.remove({ to: { $size: 0 } }, callback)
 270 }