diff options
Diffstat (limited to 'server/models')
-rw-r--r-- | server/models/pod.js | 37 | ||||
-rw-r--r-- | server/models/request.js | 110 |
2 files changed, 105 insertions, 42 deletions
diff --git a/server/models/pod.js b/server/models/pod.js index 8e7dd1fd8..b3c6db8e8 100644 --- a/server/models/pod.js +++ b/server/models/pod.js | |||
@@ -50,6 +50,7 @@ module.exports = function (sequelize, DataTypes) { | |||
50 | incrementScores, | 50 | incrementScores, |
51 | list, | 51 | list, |
52 | listAllIds, | 52 | listAllIds, |
53 | listRandomPodIdsWithRequest, | ||
53 | listBadPods, | 54 | listBadPods, |
54 | load, | 55 | load, |
55 | loadByHost, | 56 | loadByHost, |
@@ -134,6 +135,42 @@ function listAllIds (transaction, callback) { | |||
134 | }) | 135 | }) |
135 | } | 136 | } |
136 | 137 | ||
138 | function listRandomPodIdsWithRequest (limit, callback) { | ||
139 | const self = this | ||
140 | |||
141 | self.count().asCallback(function (err, count) { | ||
142 | if (err) return callback(err) | ||
143 | |||
144 | // Optimization... | ||
145 | if (count === 0) return callback(null, []) | ||
146 | |||
147 | let start = Math.floor(Math.random() * count) - limit | ||
148 | if (start < 0) start = 0 | ||
149 | |||
150 | const query = { | ||
151 | attributes: [ 'id' ], | ||
152 | order: [ | ||
153 | [ 'id', 'ASC' ] | ||
154 | ], | ||
155 | offset: start, | ||
156 | limit: limit, | ||
157 | where: { | ||
158 | id: { | ||
159 | $in: [ | ||
160 | this.sequelize.literal('SELECT "podId" FROM "RequestToPods"') | ||
161 | ] | ||
162 | } | ||
163 | } | ||
164 | } | ||
165 | |||
166 | return this.findAll(query).asCallback(function (err, pods) { | ||
167 | if (err) return callback(err) | ||
168 | |||
169 | return callback(null, map(pods, 'id')) | ||
170 | }) | ||
171 | }) | ||
172 | } | ||
173 | |||
137 | function listBadPods (callback) { | 174 | function listBadPods (callback) { |
138 | const query = { | 175 | const query = { |
139 | where: { | 176 | where: { |
diff --git a/server/models/request.js b/server/models/request.js index 1d6038044..26953e5f5 100644 --- a/server/models/request.js +++ b/server/models/request.js | |||
@@ -138,9 +138,9 @@ function makeRequests () { | |||
138 | const self = this | 138 | const self = this |
139 | const RequestToPod = this.sequelize.models.RequestToPod | 139 | const RequestToPod = this.sequelize.models.RequestToPod |
140 | 140 | ||
141 | // We limit the size of the requests (REQUESTS_LIMIT) | 141 | // We limit the size of the requests |
142 | // We don't want to stuck with the same failing requests so we get a random list | 142 | // We don't want to stuck with the same failing requests so we get a random list |
143 | listWithLimitAndRandom.call(self, constants.REQUESTS_LIMIT, function (err, requests) { | 143 | listWithLimitAndRandom.call(self, constants.REQUESTS_LIMIT_PODS, constants.REQUESTS_LIMIT_PER_POD, function (err, requests) { |
144 | if (err) { | 144 | if (err) { |
145 | logger.error('Cannot get the list of requests.', { err: err }) | 145 | logger.error('Cannot get the list of requests.', { err: err }) |
146 | return // Abort | 146 | return // Abort |
@@ -156,13 +156,15 @@ function makeRequests () { | |||
156 | 156 | ||
157 | // We want to group requests by destinations pod and endpoint | 157 | // We want to group requests by destinations pod and endpoint |
158 | const requestsToMakeGrouped = {} | 158 | const requestsToMakeGrouped = {} |
159 | Object.keys(requests).forEach(function (toPodId) { | ||
160 | requests[toPodId].forEach(function (data) { | ||
161 | const request = data.request | ||
162 | const pod = data.pod | ||
163 | const hashKey = toPodId + request.endpoint | ||
159 | 164 | ||
160 | requests.forEach(function (request) { | ||
161 | request.Pods.forEach(function (toPod) { | ||
162 | const hashKey = toPod.id + request.endpoint | ||
163 | if (!requestsToMakeGrouped[hashKey]) { | 165 | if (!requestsToMakeGrouped[hashKey]) { |
164 | requestsToMakeGrouped[hashKey] = { | 166 | requestsToMakeGrouped[hashKey] = { |
165 | toPodId: toPod.id, | 167 | toPod: pod, |
166 | endpoint: request.endpoint, | 168 | endpoint: request.endpoint, |
167 | ids: [], // request ids, to delete them from the DB in the future | 169 | ids: [], // request ids, to delete them from the DB in the future |
168 | datas: [] // requests data, | 170 | datas: [] // requests data, |
@@ -179,36 +181,29 @@ function makeRequests () { | |||
179 | 181 | ||
180 | eachLimit(Object.keys(requestsToMakeGrouped), constants.REQUESTS_IN_PARALLEL, function (hashKey, callbackEach) { | 182 | eachLimit(Object.keys(requestsToMakeGrouped), constants.REQUESTS_IN_PARALLEL, function (hashKey, callbackEach) { |
181 | const requestToMake = requestsToMakeGrouped[hashKey] | 183 | const requestToMake = requestsToMakeGrouped[hashKey] |
184 | const toPod = requestToMake.toPod | ||
182 | 185 | ||
183 | // FIXME: SQL request inside a loop :/ | 186 | // Maybe the pod is not our friend anymore so simply remove it |
184 | self.sequelize.models.Pod.load(requestToMake.toPodId, function (err, toPod) { | 187 | if (!toPod) { |
185 | if (err) { | 188 | const requestIdsToDelete = requestToMake.ids |
186 | logger.error('Error finding pod by id.', { err: err }) | ||
187 | return callbackEach() | ||
188 | } | ||
189 | |||
190 | // Maybe the pod is not our friend anymore so simply remove it | ||
191 | if (!toPod) { | ||
192 | const requestIdsToDelete = requestToMake.ids | ||
193 | 189 | ||
194 | logger.info('Removing %d requests of unexisting pod %s.', requestIdsToDelete.length, requestToMake.toPodId) | 190 | logger.info('Removing %d requests of unexisting pod %s.', requestIdsToDelete.length, requestToMake.toPod.id) |
195 | RequestToPod.removePodOf.call(self, requestIdsToDelete, requestToMake.toPodId) | 191 | RequestToPod.removePodOf.call(self, requestIdsToDelete, requestToMake.toPod.id) |
196 | return callbackEach() | 192 | return callbackEach() |
197 | } | 193 | } |
198 | 194 | ||
199 | makeRequest(toPod, requestToMake.endpoint, requestToMake.datas, function (success) { | 195 | makeRequest(toPod, requestToMake.endpoint, requestToMake.datas, function (success) { |
200 | if (success === true) { | 196 | if (success === true) { |
201 | logger.debug('Removing requests for pod %s.', requestToMake.toPodId, { requestsIds: requestToMake.ids }) | 197 | logger.debug('Removing requests for pod %s.', requestToMake.toPod.id, { requestsIds: requestToMake.ids }) |
202 | 198 | ||
203 | goodPods.push(requestToMake.toPodId) | 199 | goodPods.push(requestToMake.toPod.id) |
204 | 200 | ||
205 | // Remove the pod id of these request ids | 201 | // Remove the pod id of these request ids |
206 | RequestToPod.removePodOf(requestToMake.ids, requestToMake.toPodId, callbackEach) | 202 | RequestToPod.removePodOf(requestToMake.ids, requestToMake.toPod.id, callbackEach) |
207 | } else { | 203 | } else { |
208 | badPods.push(requestToMake.toPodId) | 204 | badPods.push(requestToMake.toPod.id) |
209 | callbackEach() | 205 | callbackEach() |
210 | } | 206 | } |
211 | }) | ||
212 | }) | 207 | }) |
213 | }, function () { | 208 | }, function () { |
214 | // All the requests were made, we update the pods score | 209 | // All the requests were made, we update the pods score |
@@ -275,29 +270,60 @@ function updatePodsScore (goodPods, badPods) { | |||
275 | } | 270 | } |
276 | } | 271 | } |
277 | 272 | ||
278 | function listWithLimitAndRandom (limit, callback) { | 273 | function listWithLimitAndRandom (limitPods, limitRequestsPerPod, callback) { |
279 | const self = this | 274 | const self = this |
275 | const Pod = this.sequelize.models.Pod | ||
280 | 276 | ||
281 | self.count().asCallback(function (err, count) { | 277 | Pod.listRandomPodIdsWithRequest(limitPods, function (err, podIds) { |
282 | if (err) return callback(err) | 278 | if (err) return callback(err) |
283 | 279 | ||
284 | // Optimization... | 280 | // We don't have friends that have requests |
285 | if (count === 0) return callback(null, []) | 281 | if (podIds.length === 0) return callback(null, []) |
286 | |||
287 | let start = Math.floor(Math.random() * count) - limit | ||
288 | if (start < 0) start = 0 | ||
289 | 282 | ||
283 | // The the first x requests of these pods | ||
284 | // It is very important to sort by id ASC to keep the requests order! | ||
290 | const query = { | 285 | const query = { |
291 | order: [ | 286 | order: [ |
292 | [ 'id', 'ASC' ] | 287 | [ 'id', 'ASC' ] |
293 | ], | 288 | ], |
294 | // offset: start, | 289 | include: [ |
295 | // limit: limit, | 290 | { |
296 | include: [ this.sequelize.models.Pod ] | 291 | model: self.sequelize.models.Pod, |
292 | where: { | ||
293 | id: { | ||
294 | $in: podIds | ||
295 | } | ||
296 | } | ||
297 | } | ||
298 | ] | ||
297 | } | 299 | } |
298 | 300 | ||
299 | self.findAll(query).asCallback(callback) | 301 | self.findAll(query).asCallback(function (err, requests) { |
302 | if (err) return callback(err) | ||
303 | |||
304 | const requestsGrouped = groupAndTruncateRequests(requests, limitRequestsPerPod) | ||
305 | return callback(err, requestsGrouped) | ||
306 | }) | ||
307 | }) | ||
308 | } | ||
309 | |||
310 | function groupAndTruncateRequests (requests, limitRequestsPerPod) { | ||
311 | const requestsGrouped = {} | ||
312 | |||
313 | requests.forEach(function (request) { | ||
314 | request.Pods.forEach(function (pod) { | ||
315 | if (!requestsGrouped[pod.id]) requestsGrouped[pod.id] = [] | ||
316 | |||
317 | if (requestsGrouped[pod.id].length < limitRequestsPerPod) { | ||
318 | requestsGrouped[pod.id].push({ | ||
319 | request, | ||
320 | pod | ||
321 | }) | ||
322 | } | ||
323 | }) | ||
300 | }) | 324 | }) |
325 | |||
326 | return requestsGrouped | ||
301 | } | 327 | } |
302 | 328 | ||
303 | function removeAll (callback) { | 329 | function removeAll (callback) { |