11const async = require ( 'async' ) ;
22const { errors } = require ( 'arsenal' ) ;
33const monitoring = require ( '../../../utilities/monitoringHandler' ) ;
4- const {
5- actionNeedQuotaCheck,
6- actionWithDataDeletion,
7- } = require ( 'arsenal' ) . policies ;
4+ const { actionNeedQuotaCheck, actionWithDataDeletion } = require ( 'arsenal' ) . policies ;
85const { config } = require ( '../../../Config' ) ;
96const QuotaService = require ( '../../../utilization/instance' ) ;
107
@@ -43,7 +40,7 @@ function processBytesToWrite(apiMethod, bucket, versionId, contentLength, objMD,
4340 // but it also replaces the target, which decreases storage
4441 bytes -= getHotContentLength ( destObjMD ) ;
4542 }
46- } else if ( ! bucket . isVersioningEnabled ( ) || bucket . isVersioningEnabled ( ) && versionId ) {
43+ } else if ( ! bucket . isVersioningEnabled ( ) || ( bucket . isVersioningEnabled ( ) && versionId ) ) {
4744 // object is being deleted (non versioned) or hard-deleted (versioned, as indicated by
4845 // the `versionId` field)
4946 bytes = - getHotContentLength ( objMD ) ;
@@ -67,8 +64,7 @@ function processBytesToWrite(apiMethod, bucket, versionId, contentLength, objMD,
6764 * @returns {boolean } Returns true if the metric is stale, false otherwise.
6865 */
6966function isMetricStale ( metric , resourceType , resourceName , action , inflight , log ) {
70- if ( metric . date && Date . now ( ) - new Date ( metric . date ) . getTime ( ) >
71- QuotaService . maxStaleness ) {
67+ if ( metric . date && Date . now ( ) - new Date ( metric . date ) . getTime ( ) > QuotaService . maxStaleness ) {
7268 log . warn ( 'Stale metrics from the quota service, allowing the request' , {
7369 resourceType,
7470 resourceName,
@@ -109,70 +105,87 @@ function _evaluateQuotas(
109105 let bucketQuotaExceeded = false ;
110106 let accountQuotaExceeded = false ;
111107 const creationDate = new Date ( bucket . getCreationDate ( ) ) . getTime ( ) ;
112- return async . parallel ( {
113- bucketQuota : parallelDone => {
114- if ( bucketQuota > 0 ) {
115- return QuotaService . getUtilizationMetrics ( 'bucket' ,
116- `${ bucket . getName ( ) } _${ creationDate } ` , null , {
117- action,
118- inflight,
119- } , ( err , bucketMetrics ) => {
120- if ( err || inflight < 0 ) {
121- return parallelDone ( err ) ;
122- }
123- if ( ! isMetricStale ( bucketMetrics , 'bucket' , bucket . getName ( ) , action , inflight , log ) &&
124- BigInt ( bucketMetrics . bytesTotal || 0 ) + BigInt ( inflightForCheck || 0 ) > bucketQuota ) {
125- log . debug ( 'Bucket quota exceeded' , {
126- bucket : bucket . getName ( ) ,
108+ return async . parallel (
109+ {
110+ bucketQuota : parallelDone => {
111+ if ( bucketQuota > 0 ) {
112+ return QuotaService . getUtilizationMetrics (
113+ 'bucket' ,
114+ `${ bucket . getName ( ) } _${ creationDate } ` ,
115+ null ,
116+ {
127117 action,
128118 inflight,
129- quota : bucketQuota ,
130- bytesTotal : bucketMetrics . bytesTotal ,
131- } ) ;
132- bucketQuotaExceeded = true ;
133- }
134- return parallelDone ( ) ;
135- } ) ;
136- }
137- return parallelDone ( ) ;
138- } ,
139- accountQuota : parallelDone => {
140- if ( accountQuota > 0 && account ?. account ) {
141- return QuotaService . getUtilizationMetrics ( 'account' ,
142- account . account , null , {
143- action,
144- inflight,
145- } , ( err , accountMetrics ) => {
146- if ( err || inflight < 0 ) {
147- return parallelDone ( err ) ;
148- }
149- // Metrics are served as BigInt strings
150- if ( ! isMetricStale ( accountMetrics , 'account' , account . account , action , inflight , log ) &&
151- BigInt ( accountMetrics . bytesTotal || 0 ) + BigInt ( inflightForCheck || 0 ) > accountQuota ) {
152- log . debug ( 'Account quota exceeded' , {
153- accountId : account . account ,
119+ } ,
120+ ( err , bucketMetrics ) => {
121+ if ( err || inflight < 0 ) {
122+ return parallelDone ( err ) ;
123+ }
124+ if (
125+ ! isMetricStale ( bucketMetrics , 'bucket' , bucket . getName ( ) , action , inflight , log ) &&
126+ BigInt ( bucketMetrics . bytesTotal || 0 ) + BigInt ( inflightForCheck || 0 ) > bucketQuota
127+ ) {
128+ log . debug ( 'Bucket quota exceeded' , {
129+ bucket : bucket . getName ( ) ,
130+ action,
131+ inflight,
132+ quota : bucketQuota ,
133+ bytesTotal : bucketMetrics . bytesTotal ,
134+ } ) ;
135+ bucketQuotaExceeded = true ;
136+ }
137+ return parallelDone ( ) ;
138+ } ,
139+ ) ;
140+ }
141+ return parallelDone ( ) ;
142+ } ,
143+ accountQuota : parallelDone => {
144+ if ( accountQuota > 0 && account ?. account ) {
145+ return QuotaService . getUtilizationMetrics (
146+ 'account' ,
147+ account . account ,
148+ null ,
149+ {
154150 action,
155151 inflight,
156- quota : accountQuota ,
157- bytesTotal : accountMetrics . bytesTotal ,
158- } ) ;
159- accountQuotaExceeded = true ;
160- }
161- return parallelDone ( ) ;
152+ } ,
153+ ( err , accountMetrics ) => {
154+ if ( err || inflight < 0 ) {
155+ return parallelDone ( err ) ;
156+ }
157+ // Metrics are served as BigInt strings
158+ if (
159+ ! isMetricStale ( accountMetrics , 'account' , account . account , action , inflight , log ) &&
160+ BigInt ( accountMetrics . bytesTotal || 0 ) + BigInt ( inflightForCheck || 0 ) > accountQuota
161+ ) {
162+ log . debug ( 'Account quota exceeded' , {
163+ accountId : account . account ,
164+ action,
165+ inflight,
166+ quota : accountQuota ,
167+ bytesTotal : accountMetrics . bytesTotal ,
168+ } ) ;
169+ accountQuotaExceeded = true ;
170+ }
171+ return parallelDone ( ) ;
172+ } ,
173+ ) ;
174+ }
175+ return parallelDone ( ) ;
176+ } ,
177+ } ,
178+ err => {
179+ if ( err ) {
180+ log . warn ( 'Error evaluating quotas' , {
181+ error : err . name ,
182+ description : err . message ,
183+ isInflightDeletion : inflight < 0 ,
162184 } ) ;
163185 }
164- return parallelDone ( ) ;
186+ return callback ( err , bucketQuotaExceeded , accountQuotaExceeded ) ;
165187 } ,
166- } , err => {
167- if ( err ) {
168- log . warn ( 'Error evaluating quotas' , {
169- error : err . name ,
170- description : err . message ,
171- isInflightDeletion : inflight < 0 ,
172- } ) ;
173- }
174- return callback ( err , bucketQuotaExceeded , accountQuotaExceeded ) ;
175- } ) ;
188+ ) ;
176189}
177190
178191/**
@@ -185,11 +198,13 @@ function _evaluateQuotas(
185198 * @returns {undefined } - Returns nothing.
186199 */
187200function monitorQuotaEvaluationDuration ( apiMethod , type , code , duration ) {
188- monitoring . quotaEvaluationDuration . labels ( {
189- action : apiMethod ,
190- type,
191- code,
192- } ) . observe ( duration / 1e9 ) ;
201+ monitoring . quotaEvaluationDuration
202+ . labels ( {
203+ action : apiMethod ,
204+ type,
205+ code,
206+ } )
207+ . observe ( duration / 1e9 ) ;
193208}
194209
195210/**
@@ -247,79 +262,107 @@ function validateQuotas(request, bucket, account, apiNames, apiMethod, inflight,
247262 inflight = 0 ;
248263 }
249264
250- return async . forEach ( apiNames , ( apiName , done ) => {
251- // Object copy operations first check the target object,
252- // meaning the source object, containing the current bytes,
253- // is checked second. This logic handles these APIs calls by
254- // ensuring the bytes are positives (i.e., not an object
255- // replacement). The source is checked with objectGetVersion
256- // instead of objectGet when the copy specifies its version.
257- const isCopySource = ( apiName === 'objectGet' || apiName === 'objectGetVersion' )
258- && ( apiMethod === 'objectCopy' || apiMethod === 'objectPutCopyPart' ) ;
259- if ( isCopySource ) {
260- // eslint-disable-next-line no-param-reassign
261- inflight = Math . abs ( inflight ) ;
262- } else if ( ! actionNeedQuotaCheck [ apiName ] && ! actionWithDataDeletion [ apiName ] ) {
263- return done ( ) ;
264- }
265- // When inflights are disabled, the sum of the current utilization metrics
266- // and the current bytes are compared with the quota. The current bytes
267- // are not sent to the utilization service. When inflights are enabled,
268- // the sum of the current utilization metrics only are compared with the
269- // quota. They include the current inflight bytes sent in the request.
270- let _inflights = shouldSendInflights ? inflight : undefined ;
271- const inflightForCheck = shouldSendInflights ? 0 : inflight ;
272- return _evaluateQuotas ( bucketQuota , accountQuota , bucket , account , _inflights ,
273- inflightForCheck , apiName , log ,
274- ( err , _bucketQuotaExceeded , _accountQuotaExceeded ) => {
275- if ( err ) {
276- return done ( err ) ;
277- }
265+ return async . forEach (
266+ apiNames ,
267+ ( apiName , done ) => {
268+ // Object copy operations first check the target object,
269+ // meaning the source object, containing the current bytes,
270+ // is checked second. This logic handles these APIs calls by
271+ // ensuring the bytes are positives (i.e., not an object
272+ // replacement). The source is checked with objectGetVersion
273+ // instead of objectGet when the copy specifies its version.
274+ const isCopySource =
275+ ( apiName === 'objectGet' || apiName === 'objectGetVersion' ) &&
276+ ( apiMethod === 'objectCopy' || apiMethod === 'objectPutCopyPart' ) ;
277+ if ( isCopySource ) {
278+ // eslint-disable-next-line no-param-reassign
279+ inflight = Math . abs ( inflight ) ;
280+ } else if ( ! actionNeedQuotaCheck [ apiName ] && ! actionWithDataDeletion [ apiName ] ) {
281+ return done ( ) ;
282+ }
283+ // When inflights are disabled, the sum of the current utilization metrics
284+ // and the current bytes are compared with the quota. The current bytes
285+ // are not sent to the utilization service. When inflights are enabled,
286+ // the sum of the current utilization metrics only are compared with the
287+ // quota. They include the current inflight bytes sent in the request.
288+ let _inflights = shouldSendInflights ? inflight : undefined ;
289+ const inflightForCheck = shouldSendInflights ? 0 : inflight ;
290+ return _evaluateQuotas (
291+ bucketQuota ,
292+ accountQuota ,
293+ bucket ,
294+ account ,
295+ _inflights ,
296+ inflightForCheck ,
297+ apiName ,
298+ log ,
299+ ( err , _bucketQuotaExceeded , _accountQuotaExceeded ) => {
300+ if ( err ) {
301+ return done ( err ) ;
302+ }
278303
279- bucketQuotaExceeded = _bucketQuotaExceeded ;
280- accountQuotaExceeded = _accountQuotaExceeded ;
304+ bucketQuotaExceeded = _bucketQuotaExceeded ;
305+ accountQuotaExceeded = _accountQuotaExceeded ;
281306
282- // Inflights are inverted: in case of cleanup, we just re-issue
283- // the same API call.
284- if ( _inflights ) {
285- _inflights = - _inflights ;
286- }
307+ // Inflights are inverted: in case of cleanup, we just re-issue
308+ // the same API call.
309+ if ( _inflights ) {
310+ _inflights = - _inflights ;
311+ }
287312
288- request . finalizerHooks . push ( ( errorFromAPI , _done ) => {
289- const code = ( bucketQuotaExceeded || accountQuotaExceeded ) ? 429 : 200 ;
290- const quotaCleanUpStartTime = process . hrtime . bigint ( ) ;
291- // Quotas are cleaned only in case of error in the API
292- async . waterfall ( [
293- cb => {
294- if ( errorFromAPI ) {
295- return _evaluateQuotas ( bucketQuota , accountQuota , bucket , account , _inflights ,
296- null , apiName , log , cb ) ;
297- }
298- return cb ( ) ;
299- } ,
300- ] , ( ) => {
301- monitorQuotaEvaluationDuration ( apiMethod , type , code , quotaEvaluationDuration +
302- Number ( process . hrtime . bigint ( ) - quotaCleanUpStartTime ) ) ;
303- return _done ( ) ;
313+ request . finalizerHooks . push ( ( errorFromAPI , _done ) => {
314+ const code = bucketQuotaExceeded || accountQuotaExceeded ? 429 : 200 ;
315+ const quotaCleanUpStartTime = process . hrtime . bigint ( ) ;
316+ // Quotas are cleaned only in case of error in the API
317+ async . waterfall (
318+ [
319+ cb => {
320+ if ( errorFromAPI ) {
321+ return _evaluateQuotas (
322+ bucketQuota ,
323+ accountQuota ,
324+ bucket ,
325+ account ,
326+ _inflights ,
327+ null ,
328+ apiName ,
329+ log ,
330+ cb ,
331+ ) ;
332+ }
333+ return cb ( ) ;
334+ } ,
335+ ] ,
336+ ( ) => {
337+ monitorQuotaEvaluationDuration (
338+ apiMethod ,
339+ type ,
340+ code ,
341+ quotaEvaluationDuration + Number ( process . hrtime . bigint ( ) - quotaCleanUpStartTime ) ,
342+ ) ;
343+ return _done ( ) ;
344+ } ,
345+ ) ;
304346 } ) ;
305- } ) ;
306347
307- return done ( ) ;
308- } ) ;
309- } , err => {
310- quotaEvaluationDuration = Number ( process . hrtime . bigint ( ) - requestStartTime ) ;
311- if ( err ) {
312- log . warn ( 'Error getting metrics from the quota service, allowing the request' , {
313- error : err . name ,
314- description : err . message ,
315- } ) ;
316- }
317- if ( ! actionWithDataDeletion [ apiMethod ] &&
318- ( bucketQuotaExceeded || accountQuotaExceeded ) ) {
319- return callback ( errors . QuotaExceeded ) ;
320- }
321- return callback ( ) ;
322- } ) ;
348+ return done ( ) ;
349+ } ,
350+ ) ;
351+ } ,
352+ err => {
353+ quotaEvaluationDuration = Number ( process . hrtime . bigint ( ) - requestStartTime ) ;
354+ if ( err ) {
355+ log . warn ( 'Error getting metrics from the quota service, allowing the request' , {
356+ error : err . name ,
357+ description : err . message ,
358+ } ) ;
359+ }
360+ if ( ! actionWithDataDeletion [ apiMethod ] && ( bucketQuotaExceeded || accountQuotaExceeded ) ) {
361+ return callback ( errors . QuotaExceeded ) ;
362+ }
363+ return callback ( ) ;
364+ } ,
365+ ) ;
323366}
324367
325368module . exports = {
0 commit comments