Skip to content

Forward X-Scal-Request-Uids on Scuba quota requests #6220

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
delthas wants to merge 3 commits into
base: development/9.4
Choose a base branch
from
Open

Forward X-Scal-Request-Uids on Scuba quota requests #6220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9f145ea
Prettify files touched by this change
delthas Jul 6, 2026
53199f1
Forward X-Scal-Request-Uids on Scuba quota requests
delthas Jul 6, 2026
c0b8f6f
Migrate ScubaClientImpl.getUtilizationMetrics to async/await
delthas Jul 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
313 changes: 178 additions & 135 deletions lib/api/apiUtils/quotas/quotaUtils.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,7 @@
const async = require('async');
const { errors } = require('arsenal');
const monitoring = require('../../../utilities/monitoringHandler');
const {
actionNeedQuotaCheckCopy,
actionNeedQuotaCheck,
actionWithDataDeletion,
} = require('arsenal').policies;
const { actionNeedQuotaCheckCopy, actionNeedQuotaCheck, actionWithDataDeletion } = require('arsenal').policies;
const { config } = require('../../../Config');
const QuotaService = require('../../../utilization/instance');

Expand Down Expand Up @@ -44,7 +40,7 @@
// but it also replaces the target, which decreases storage
bytes -= getHotContentLength(destObjMD);
}
} else if (!bucket.isVersioningEnabled() || bucket.isVersioningEnabled() && versionId) {
} else if (!bucket.isVersioningEnabled() || (bucket.isVersioningEnabled() && versionId)) {
// object is being deleted (non versioned) or hard-deleted (versioned, as indicated by
// the `versionId` field)
bytes = -getHotContentLength(objMD);
Expand All @@ -68,8 +64,7 @@
* @returns {boolean} Returns true if the metric is stale, false otherwise.
*/
function isMetricStale(metric, resourceType, resourceName, action, inflight, log) {
if (metric.date && Date.now() - new Date(metric.date).getTime() >
QuotaService.maxStaleness) {
if (metric.date && Date.now() - new Date(metric.date).getTime() > QuotaService.maxStaleness) {
log.warn('Stale metrics from the quota service, allowing the request', {
resourceType,
resourceName,
Expand Down Expand Up @@ -110,70 +105,89 @@
let bucketQuotaExceeded = false;
let accountQuotaExceeded = false;
const creationDate = new Date(bucket.getCreationDate()).getTime();
return async.parallel({
bucketQuota: parallelDone => {
if (bucketQuota > 0) {
return QuotaService.getUtilizationMetrics('bucket',
`${bucket.getName()}_${creationDate}`, null, {
action,
inflight,
}, (err, bucketMetrics) => {
if (err || inflight < 0) {
return parallelDone(err);
}
if (!isMetricStale(bucketMetrics, 'bucket', bucket.getName(), action, inflight, log) &&
BigInt(bucketMetrics.bytesTotal || 0) + BigInt(inflightForCheck || 0) > bucketQuota) {
log.debug('Bucket quota exceeded', {
bucket: bucket.getName(),
return async.parallel(
{
bucketQuota: parallelDone => {
if (bucketQuota > 0) {
return QuotaService.getUtilizationMetrics(
'bucket',
`${bucket.getName()}_${creationDate}`,
null,
{
action,
inflight,
quota: bucketQuota,
bytesTotal: bucketMetrics.bytesTotal,
});
bucketQuotaExceeded = true;
}
return parallelDone();
});
}
return parallelDone();
},
accountQuota: parallelDone => {
if (accountQuota > 0 && account?.account) {
return QuotaService.getUtilizationMetrics('account',
account.account, null, {
action,
inflight,
}, (err, accountMetrics) => {
if (err || inflight < 0) {
return parallelDone(err);
}
// Metrics are served as BigInt strings
if (!isMetricStale(accountMetrics, 'account', account.account, action, inflight, log) &&
BigInt(accountMetrics.bytesTotal || 0) + BigInt(inflightForCheck || 0) > accountQuota) {
log.debug('Account quota exceeded', {
accountId: account.account,
},
log,
(err, bucketMetrics) => {
if (err || inflight < 0) {
return parallelDone(err);
}
if (
!isMetricStale(bucketMetrics, 'bucket', bucket.getName(), action, inflight, log) &&
BigInt(bucketMetrics.bytesTotal || 0) + BigInt(inflightForCheck || 0) > bucketQuota
) {
log.debug('Bucket quota exceeded', {
bucket: bucket.getName(),
action,
inflight,
quota: bucketQuota,
bytesTotal: bucketMetrics.bytesTotal,
});
bucketQuotaExceeded = true;
}
return parallelDone();
},
);
}
return parallelDone();
},
accountQuota: parallelDone => {
if (accountQuota > 0 && account?.account) {
return QuotaService.getUtilizationMetrics(
'account',
account.account,
null,
{
action,
inflight,
quota: accountQuota,
bytesTotal: accountMetrics.bytesTotal,
});
accountQuotaExceeded = true;
}
return parallelDone();
},
log,
(err, accountMetrics) => {
if (err || inflight < 0) {
return parallelDone(err);
}
// Metrics are served as BigInt strings
if (
!isMetricStale(accountMetrics, 'account', account.account, action, inflight, log) &&
BigInt(accountMetrics.bytesTotal || 0) + BigInt(inflightForCheck || 0) > accountQuota
) {
log.debug('Account quota exceeded', {
accountId: account.account,
action,
inflight,
quota: accountQuota,
bytesTotal: accountMetrics.bytesTotal,
});
accountQuotaExceeded = true;
}
return parallelDone();
},
);
}
return parallelDone();
},
},
err => {
if (err) {
log.warn('Error evaluating quotas', {
error: err.name,
description: err.message,
isInflightDeletion: inflight < 0,
});
}
return parallelDone();
return callback(err, bucketQuotaExceeded, accountQuotaExceeded);
},
}, err => {
if (err) {
log.warn('Error evaluating quotas', {
error: err.name,
description: err.message,
isInflightDeletion: inflight < 0,
});
}
return callback(err, bucketQuotaExceeded, accountQuotaExceeded);
});
);
}

/**
Expand All @@ -186,11 +200,13 @@
* @returns {undefined} - Returns nothing.
*/
function monitorQuotaEvaluationDuration(apiMethod, type, code, duration) {
monitoring.quotaEvaluationDuration.labels({
action: apiMethod,
type,
code,
}).observe(duration / 1e9);
monitoring.quotaEvaluationDuration
.labels({
action: apiMethod,
type,
code,
})
.observe(duration / 1e9);
}

/**
Expand Down Expand Up @@ -248,76 +264,103 @@
inflight = 0;
}

return async.forEach(apiNames, (apiName, done) => {
// Object copy operations first check the target object,
// meaning the source object, containing the current bytes,
// is checked second. This logic handles these APIs calls by
// ensuring the bytes are positives (i.e., not an object
// replacement).
if (actionNeedQuotaCheckCopy(apiName, apiMethod)) {
// eslint-disable-next-line no-param-reassign
inflight = Math.abs(inflight);
} else if (!actionNeedQuotaCheck[apiName] && !actionWithDataDeletion[apiName]) {
return done();
}
// When inflights are disabled, the sum of the current utilization metrics
// and the current bytes are compared with the quota. The current bytes
// are not sent to the utilization service. When inflights are enabled,
// the sum of the current utilization metrics only are compared with the
// quota. They include the current inflight bytes sent in the request.
let _inflights = shouldSendInflights ? inflight : undefined;
const inflightForCheck = shouldSendInflights ? 0 : inflight;
return _evaluateQuotas(bucketQuota, accountQuota, bucket, account, _inflights,
inflightForCheck, apiName, log,
(err, _bucketQuotaExceeded, _accountQuotaExceeded) => {
if (err) {
return done(err);
}
return async.forEach(
apiNames,
(apiName, done) => {
// Object copy operations first check the target object,
// meaning the source object, containing the current bytes,
// is checked second. This logic handles these APIs calls by
// ensuring the bytes are positives (i.e., not an object
// replacement).
if (actionNeedQuotaCheckCopy(apiName, apiMethod)) {
// eslint-disable-next-line no-param-reassign
inflight = Math.abs(inflight);
} else if (!actionNeedQuotaCheck[apiName] && !actionWithDataDeletion[apiName]) {
return done();
}
// When inflights are disabled, the sum of the current utilization metrics
// and the current bytes are compared with the quota. The current bytes
// are not sent to the utilization service. When inflights are enabled,
// the sum of the current utilization metrics only are compared with the
// quota. They include the current inflight bytes sent in the request.
let _inflights = shouldSendInflights ? inflight : undefined;
const inflightForCheck = shouldSendInflights ? 0 : inflight;
return _evaluateQuotas(
bucketQuota,
accountQuota,
bucket,
account,
_inflights,
inflightForCheck,
apiName,
log,
(err, _bucketQuotaExceeded, _accountQuotaExceeded) => {
if (err) {
return done(err);
}

bucketQuotaExceeded = _bucketQuotaExceeded;
accountQuotaExceeded = _accountQuotaExceeded;
bucketQuotaExceeded = _bucketQuotaExceeded;
accountQuotaExceeded = _accountQuotaExceeded;

// Inflights are inverted: in case of cleanup, we just re-issue
// the same API call.
if (_inflights) {
_inflights = -_inflights;
}
// Inflights are inverted: in case of cleanup, we just re-issue
// the same API call.
if (_inflights) {
_inflights = -_inflights;
}

request.finalizerHooks.push((errorFromAPI, _done) => {
const code = (bucketQuotaExceeded || accountQuotaExceeded) ? 429 : 200;
const quotaCleanUpStartTime = process.hrtime.bigint();
// Quotas are cleaned only in case of error in the API
async.waterfall([
cb => {
if (errorFromAPI) {
return _evaluateQuotas(bucketQuota, accountQuota, bucket, account, _inflights,
null, apiName, log, cb);
}
return cb();
},
], () => {
monitorQuotaEvaluationDuration(apiMethod, type, code, quotaEvaluationDuration +
Number(process.hrtime.bigint() - quotaCleanUpStartTime));
return _done();
request.finalizerHooks.push((errorFromAPI, _done) => {
const code = bucketQuotaExceeded || accountQuotaExceeded ? 429 : 200;
const quotaCleanUpStartTime = process.hrtime.bigint();
// Quotas are cleaned only in case of error in the API
async.waterfall(
[
cb => {
if (errorFromAPI) {
return _evaluateQuotas(
bucketQuota,
accountQuota,
bucket,
account,
_inflights,
null,
apiName,
log,
cb,
);
}
return cb();
},

Check notice

Code scanning / CodeQL

Callback-style function (async migration) Note

This function uses a callback parameter ('cb'). Refactor to async/await.
Comment on lines +317 to +332
],
() => {
monitorQuotaEvaluationDuration(
apiMethod,
type,
code,
quotaEvaluationDuration + Number(process.hrtime.bigint() - quotaCleanUpStartTime),
);
return _done();
},
);
});
});

return done();
});
}, err => {
quotaEvaluationDuration = Number(process.hrtime.bigint() - requestStartTime);
if (err) {
log.warn('Error getting metrics from the quota service, allowing the request', {
error: err.name,
description: err.message,
});
}
if (!actionWithDataDeletion[apiMethod] &&
(bucketQuotaExceeded || accountQuotaExceeded)) {
return callback(errors.QuotaExceeded);
}
return callback();
});
return done();
},
);
},

Check notice

Code scanning / CodeQL

Callback-style function (async migration) Note

This function uses a callback parameter ('done'). Refactor to async/await.
Comment on lines +269 to +349
err => {
quotaEvaluationDuration = Number(process.hrtime.bigint() - requestStartTime);
if (err) {
log.warn('Error getting metrics from the quota service, allowing the request', {
error: err.name,
description: err.message,
});
}
if (!actionWithDataDeletion[apiMethod] && (bucketQuotaExceeded || accountQuotaExceeded)) {
return callback(errors.QuotaExceeded);
}
return callback();
},
);
}

module.exports = {
Expand Down
Loading
Loading