Seed zero-value quota metric on bucket create / quota enable (9.3 backport)

lib/api/apiUtils/bucket/bucketCreation.js

@@ -1,5 +1,6 @@
 const assert = require('assert');
 const async = require('async');
+const { promisify } = require('util');
 const { errors } = require('arsenal');
 
 const acl = require('../../../metadata/acl');
@@ -85,27 +86,46 @@
     return metadata.updateBucket(bucketName, bucket, log, callback);
 }
 
-function freshStartCreateBucket(bucket, canonicalID, log, callback) {
+const createBucketMD = promisify((...args) => metadata.createBucket(...args));
+const initializeBucketCapacity = promisify((...args) => metadata.initializeBucketCapacity(...args));
+const addToUsersBucketAsync = promisify(addToUsersBucket);
+const removeTransientOrDeletedLabelAsync = promisify(removeTransientOrDeletedLabel);
+
+async function seedBucketQuotaCapacity(bucket, log) {
+    if (!config.isQuotaEnabled()) {
+        return;
+    }
+    try {
+        await initializeBucketCapacity(bucket.getName(), bucket.getCreationDate(), log);
+    } catch (err) {
+        // Best-effort: a missing metric self-heals at the next count-items run,
+        // so a seeding failure must never fail the bucket operation.
+        log.warn('error seeding bucket quota capacity metric', { error: err });
+    }
+}
+
+async function freshStartCreateBucket(bucket, canonicalID, log, callback) {
+    if (callback) {
+        return freshStartCreateBucket(bucket, canonicalID, log).then(() => callback(null), callback);
+    }
     const bucketName = bucket.getName();
-    metadata.createBucket(bucketName, bucket, log, err => {
+    try {
+        await createBucketMD(bucketName, bucket, log);
+    } catch (err) {
         if (err?.is?.BucketAlreadyExists) {
             // The concurrent creator owns users-bucket registration and
             // cleanup of the transient flag.
             log.trace('bucket already exists in metadata');
-            return callback();
-        }
-        if (err) {
-            log.debug('error from metadata', { error: err });
-            return callback(err);
+            return undefined;
         }
-        log.trace('created bucket in metadata');
-        return addToUsersBucket(canonicalID, bucketName, bucket, log, err => {
-            if (err) {
-                return callback(err);
-            }
-            return removeTransientOrDeletedLabel(bucket, log, callback);
-        });
-    });
+        throw err;
+    }
+    log.trace('created bucket in metadata');
+    await addToUsersBucketAsync(canonicalID, bucketName, bucket, log);
+    // Seed before clearing the transient/deleted flag so the bucket is never usable without its metric doc.
+    await seedBucketQuotaCapacity(bucket, log);
+    await removeTransientOrDeletedLabelAsync(bucket, log);
+    return undefined;
 }
 
 /**
@@ -120,14 +140,16 @@
  * @param {function} callback - callback with error or null as arguments
  * @return {undefined}
  */
-function cleanUpBucket(bucketMD, canonicalID, log, callback) {
+async function cleanUpBucket(bucketMD, canonicalID, log, callback) {
+    if (callback) {
+        return cleanUpBucket(bucketMD, canonicalID, log).then(() => callback(null), callback);
+    }
     const bucketName = bucketMD.getName();
-    return addToUsersBucket(canonicalID, bucketName, bucketMD, log, err => {
-        if (err) {
-            return callback(err);
-        }
-        return removeTransientOrDeletedLabel(bucketMD, log, callback);
-    });
+    await addToUsersBucketAsync(canonicalID, bucketName, bucketMD, log);
+    // Seed before clearing the transient/deleted flag so the bucket is never usable without its metric doc.
+    await seedBucketQuotaCapacity(bucketMD, log);
+    await removeTransientOrDeletedLabelAsync(bucketMD, log);
+    return undefined;
 }
 
 /**

lib/api/bucketUpdateQuota.js

@@ -1,95 +1,139 @@
-const { waterfall } = require('async');
+const { promisify } = require('util');
 const { errorInstances } = require('arsenal');
 const collectCorsHeaders = require('../utilities/collectCorsHeaders');
 const { standardMetadataValidateBucket } = require('../metadata/metadataUtils');
 const metadata = require('../metadata/wrapper');
 const { pushMetric } = require('../utapi/utilities');
 const monitoring = require('../utilities/monitoringHandler');
 const { parseString } = require('xml2js');
+const { config } = require('../Config');
+const constants = require('../../constants');
 
-function validateBucketQuotaProperty(requestBody, next) {
+const listObjects = promisify((...args) => metadata.listObject(...args));
+const updateBucketMD = promisify((...args) => metadata.updateBucket(...args));
+const initializeBucketCapacity = promisify((...args) => metadata.initializeBucketCapacity(...args));
+const validateBucket = promisify(standardMetadataValidateBucket);
+const parseStringAsync = promisify(parseString);
+
+function validateBucketQuotaProperty(requestBody) {
     let quota = requestBody.quota;
     if (quota === undefined) {
         quota = requestBody.QuotaConfiguration?.Quota;
     }
     const quotaValue = parseInt(quota, 10);
     if (Number.isNaN(quotaValue)) {
-        return next(errorInstances.InvalidArgument.customizeDescription('Quota Value should be a number'));
+        throw errorInstances.InvalidArgument.customizeDescription('Quota Value should be a number');
     }
     if (quotaValue <= 0) {
-        return next(errorInstances.InvalidArgument.customizeDescription('Quota value must be a positive number'));
+        throw errorInstances.InvalidArgument.customizeDescription('Quota value must be a positive number');
     }
-    return next(null, quotaValue);
+    return quotaValue;
 }
 
-function parseRequestBody(requestBody, contentType, next) {
-    switch (contentType) {
-        case 'application/xml':
-            return parseString(requestBody, { explicitArray: false }, (xmlError, xmlData) => {
-                if (xmlError) {
-                    return next(errorInstances.InvalidArgument.customizeDescription('Invalid XML format'));
-                }
-                return next(null, xmlData);
-            });
-        case 'application/json':
-        default:
-            try {
-                const jsonData = JSON.parse(requestBody);
-                if (typeof jsonData !== 'object') {
-                    throw new Error('Invalid JSON');
-                }
-                return next(null, jsonData);
-            } catch {
-                return next(errorInstances.InvalidArgument.customizeDescription('Request body must be a JSON object'));
-            }
+async function parseRequestBody(requestBody, contentType) {
+    if (contentType === 'application/xml') {
+        try {
+            return await parseStringAsync(requestBody, { explicitArray: false });
+        } catch {
+            throw errorInstances.InvalidArgument.customizeDescription('Invalid XML format');
+        }
+    }
+    try {
+        const jsonData = JSON.parse(requestBody);
+        if (typeof jsonData !== 'object') {
+            throw new Error('Invalid JSON');
+        }
+        return jsonData;
+    } catch {
+        throw errorInstances.InvalidArgument.customizeDescription('Request body must be a JSON object');
     }
 }
 
-function bucketUpdateQuota(authInfo, request, log, callback) {
-    log.debug('processing request', { method: 'bucketUpdateQuota' });
+async function bucketHasInProgressMpus(bucketName, log) {
+    const mpuBucketName = `${constants.mpuBucketPrefix}${bucketName}`;
+    try {
+        // Mirror bucketDelete's check: one overview key per in-progress upload,
+        // bounded to a single key. A missing shadow bucket means none.
+        const list = await listObjects(mpuBucketName, { prefix: 'overview', maxKeys: 1 }, log);
+        return (list.Contents?.length ?? 0) > 0;
+    } catch (err) {
+        if (err.is?.NoSuchBucket) {
+            return false;
+        }
+        throw err;
+    }
+}
+
+async function seedEmptyBucketCapacity(bucket, log) {
+    if (!config.isQuotaEnabled()) {
+        return;
+    }
+    const bucketName = bucket.getName();
+    try {
+        const list = await listObjects(bucketName, { maxKeys: 1, listingType: 'DelimiterVersions' }, log);
+        const hasObjects = (list.Versions?.length ?? 0) + (list.DeleteMarkers?.length ?? 0) > 0;
+        if (hasObjects) {
+            return;
+        }
+        // A DelimiterVersions listing does not see in-progress MPU parts (shadow
+        // bucket), which hold real storage and are counted by count-items; seeding
+        // zero for a bucket with only uncommitted uploads would under-enforce.
+        if (await bucketHasInProgressMpus(bucketName, log)) {
+            return;
+        }
+        await initializeBucketCapacity(bucketName, bucket.getCreationDate(), log);
+    } catch (err) {
+        // Best-effort: a missing metric self-heals at the next count-items run,
+        // so a seeding failure must never fail the quota update.
+        log.warn('error seeding bucket quota capacity metric', { error: err });
+    }
+}
 
+async function bucketUpdateQuota(authInfo, request, log, callback) {
+    if (callback) {
+        return bucketUpdateQuota(authInfo, request, log).then(
+            corsHeaders => callback(null, corsHeaders),
+            err => callback(err, err.code, err.additionalResHeaders),
+        );
+    }
+
+    log.debug('processing request', { method: 'bucketUpdateQuota' });
     const { bucketName } = request;
     const metadataValParams = {
         authInfo,
         bucketName,
         requestType: request.apiMethods || 'bucketUpdateQuota',
         request,
     };
-    let bucket = null;
-    return waterfall([
-        next => standardMetadataValidateBucket(metadataValParams, request.actionImplicitDenies, log,
-            (err, b) => {
-                bucket = b;
-                return next(err, bucket);
-            }),
-        (bucket, next) => parseRequestBody(request.post, request.headers['content-type'], (err, requestBody) =>
-            next(err, bucket, requestBody)),
-        (bucket, requestBody, next) => validateBucketQuotaProperty(requestBody, (err, quotaValue) =>
-            next(err, bucket, quotaValue)),
-        (bucket, quotaValue, next) => {
-            bucket.setQuota(quotaValue);
-            return metadata.updateBucket(bucket.getName(), bucket, log, next);
-        },
-    ], (err, bucket) => {
-        const corsHeaders = collectCorsHeaders(request.headers.origin,
-            request.method, bucket);
-        if (err) {
-            log.debug('error processing request', {
-                error: err,
-                method: 'bucketUpdateQuota'
-            });
-            monitoring.promMetrics('PUT', bucketName, err.code,
-                'updateBucketQuota');
-            return callback(err, err.code, corsHeaders);
-        }
-        monitoring.promMetrics(
-            'PUT', bucketName, '200', 'updateBucketQuota');
-        pushMetric('updateBucketQuota', log, {
-            authInfo,
-            bucket: bucketName,
+
+    let bucket;
+    try {
+        bucket = await validateBucket(metadataValParams, request.actionImplicitDenies, log);
+        const requestBody = await parseRequestBody(request.post, request.headers['content-type']);
+        const quotaValue = validateBucketQuotaProperty(requestBody);
+        // Seed before enabling the quota so the metric exists before quotas are enforced. A PUT racing
+        // between the emptiness check and this call leaves a stale zero that self-heals at the next count-items run.
+        await seedEmptyBucketCapacity(bucket, log);
+        bucket.setQuota(quotaValue);
+        await updateBucketMD(bucket.getName(), bucket, log);
+    } catch (err) {
+        const corsHeaders = collectCorsHeaders(request.headers.origin, request.method, bucket);
+        log.debug('error processing request', {
+            error: err,
+            method: 'bucketUpdateQuota',
         });
-        return callback(null, corsHeaders);
+        monitoring.promMetrics('PUT', bucketName, err.code, 'updateBucketQuota');
+        err.additionalResHeaders = err.additionalResHeaders || corsHeaders;
+        throw err;
+    }
+
+    const corsHeaders = collectCorsHeaders(request.headers.origin, request.method, bucket);
+    monitoring.promMetrics('PUT', bucketName, '200', 'updateBucketQuota');
+    pushMetric('updateBucketQuota', log, {
+        authInfo,
+        bucket: bucketName,
     });
+    return corsHeaders;
 }
 
 module.exports = bucketUpdateQuota;

package.json

@@ -30,7 +30,7 @@
         "@azure/storage-blob": "^12.28.0",
         "@hapi/joi": "^17.1.1",
         "@smithy/node-http-handler": "^3.0.0",
-        "arsenal": "git+https://github.com/scality/Arsenal#8.4.21",
+        "arsenal": "git+https://github.com/scality/Arsenal#8.4.22",
         "async": "2.6.4",
         "bucketclient": "scality/bucketclient#8.2.7",
         "bufferutil": "^4.0.8",