Merge pull request #8727 from romayalon/romy-5.15-backports1

5.15.11 Backports | Pre-signed URL fixes + FileWriter addition (ChunkFS removal)

config.js

@@ -821,6 +821,8 @@ config.NSFS_LOW_FREE_SPACE_MB_UNLEASH = 10 * 1024;
 // operations safely.
 config.NSFS_LOW_FREE_SPACE_PERCENT_UNLEASH = 0.10;
 
+config.NFSF_UPLOAD_STREAM_MEM_THRESHOLD = 8 * 1024 * 1024;
+
 ////////////////////////////
 // NSFS NON CONTAINERIZED //
 ////////////////////////////

src/endpoint/s3/s3_errors.js

@@ -519,6 +519,21 @@ S3Error.InvalidEncodingType = Object.freeze({
     message: 'Invalid Encoding Method specified in Request',
     http_code: 400,
 });
+S3Error.AuthorizationQueryParametersError = Object.freeze({
+    code: 'AuthorizationQueryParametersError',
+    message: 'X-Amz-Expires must be less than a week (in seconds); that is, the given X-Amz-Expires must be less than 604800 seconds',
+    http_code: 400,
+});
+S3Error.RequestExpired = Object.freeze({
+    code: 'AccessDenied',
+    message: 'Request has expired',
+    http_code: 403,
+});
+S3Error.RequestNotValidYet = Object.freeze({
+    code: 'AccessDenied',
+    message: 'request is not valid yet',
+    http_code: 403,
+});
 
 ////////////////////////////////////////////////////////////////
 // S3 Select                                                  //

src/sdk/namespace_fs.js

@@ -18,7 +18,7 @@ const stream_utils = require('../util/stream_utils');
 const buffer_utils = require('../util/buffer_utils');
 const size_utils = require('../util/size_utils');
 const native_fs_utils = require('../util/native_fs_utils');
-const ChunkFS = require('../util/chunk_fs');
+const FileWriter = require('../util/file_writer');
 const LRUCache = require('../util/lru_cache');
 const nb_native = require('../util/nb_native');
 const RpcError = require('../rpc/rpc_error');
@@ -1483,31 +1483,34 @@ class NamespaceFS {
     // Can be finetuned further on if needed and inserting the Semaphore logic inside
     // Instead of wrapping the whole _upload_stream function (q_buffers lives outside of the data scope of the stream)
     async _upload_stream({ fs_context, params, target_file, object_sdk, offset }) {
-        const { source_stream, copy_source } = params;
+        const { copy_source } = params;
         try {
             // Not using async iterators with ReadableStreams due to unsettled promises issues on abort/destroy
             const md5_enabled = config.NSFS_CALCULATE_MD5 || (this.force_md5_etag ||
                 object_sdk?.requesting_account?.force_md5_etag);
-            const chunk_fs = new ChunkFS({
+            const file_writer = new FileWriter({
                 target_file,
                 fs_context,
-                stats: this.stats,
-                namespace_resource_id: this.namespace_resource_id,
-                md5_enabled,
                 offset,
+                md5_enabled,
+                stats: this.stats,
                 bucket: params.bucket,
-                large_buf_size: multi_buffer_pool.get_buffers_pool(undefined).buf_size
+                large_buf_size: multi_buffer_pool.get_buffers_pool(undefined).buf_size,
+                namespace_resource_id: this.namespace_resource_id,
             });
-            chunk_fs.on('error', err1 => dbg.error('namespace_fs._upload_stream: error occured on stream ChunkFS: ', err1));
+            file_writer.on('error', err => dbg.error('namespace_fs._upload_stream: error occured on FileWriter: ', err));
+            file_writer.on('finish', arg => dbg.log1('namespace_fs._upload_stream: finish occured on stream FileWriter: ', arg));
+            file_writer.on('close', arg => dbg.log1('namespace_fs._upload_stream: close occured on stream FileWriter: ', arg));
+
             if (copy_source) {
-                await this.read_object_stream(copy_source, object_sdk, chunk_fs);
+                await this.read_object_stream(copy_source, object_sdk, file_writer);
             } else if (params.source_params) {
-                await params.source_ns.read_object_stream(params.source_params, object_sdk, chunk_fs);
+                await params.source_ns.read_object_stream(params.source_params, object_sdk, file_writer);
             } else {
-                await stream_utils.pipeline([source_stream, chunk_fs]);
-                await stream_utils.wait_finished(chunk_fs);
+                await stream_utils.pipeline([params.source_stream, file_writer]);
+                await stream_utils.wait_finished(file_writer);
             }
-            return { digest: chunk_fs.digest, total_bytes: chunk_fs.total_bytes };
+            return { digest: file_writer.digest, total_bytes: file_writer.total_bytes };
         } catch (error) {
             dbg.error('_upload_stream had error: ', error);
             throw error;

src/test/unit_tests/index.js

@@ -57,7 +57,7 @@ require('./test_bucket_chunks_builder');
 require('./test_mirror_writer');
 require('./test_namespace_fs');
 require('./test_ns_list_objects');
-require('./test_chunk_fs');
+require('./test_file_writer');
 require('./test_namespace_fs_mpu');
 require('./test_nb_native_fs');
 require('./test_s3select');

src/test/unit_tests/nc_coretest.js

@@ -96,6 +96,8 @@ function setup(options = {}) {
         });
 
         // TODO - run health
+        // wait 2 seconds before announcing nc coretes is ready
+        await P.delay(2000);
         await announce(`nc coretest ready... (took ${((Date.now() - start) / 1000).toFixed(1)} sec)`);
     });
 

src/test/unit_tests/nc_index.js

@@ -7,7 +7,7 @@ coretest.setup();
 
 require('./test_namespace_fs');
 require('./test_ns_list_objects');
-require('./test_chunk_fs');
+require('./test_file_writer');
 require('./test_namespace_fs_mpu');
 require('./test_nb_native_fs');
 require('./test_nc_nsfs_cli');

src/test/unit_tests/test_bucketspace.js

@@ -1,5 +1,5 @@
 /* Copyright (C) 2020 NooBaa */
-/*eslint max-lines: ["error", 2200]*/
+/*eslint max-lines: ["error", 2500]*/
 /*eslint max-lines-per-function: ["error", 1300]*/
 /*eslint max-statements: ["error", 80, { "ignoreTopLevelFunctions": true }]*/
 'use strict';
@@ -12,10 +12,16 @@ const util = require('util');
 const http = require('http');
 const mocha = require('mocha');
 const assert = require('assert');
+const http_utils = require('../../util/http_utils');
 const config = require('../../../config');
 const fs_utils = require('../../util/fs_utils');
-const { stat, open } = require('../../util/nb_native')().fs;
+const fetch = require('node-fetch');
+const P = require('../../util/promise');
+const cloud_utils = require('../../util/cloud_utils');
+const SensitiveString = require('../../util/sensitive_string');
+const S3Error = require('../../../src/endpoint/s3/s3_errors').S3Error;
 const test_utils = require('../system_tests/test_utils');
+const { stat, open } = require('../../util/nb_native')().fs;
 const { get_process_fs_context } = require('../../util/native_fs_utils');
 const { TYPES } = require('../../manage_nsfs/manage_nsfs_constants');
 const ManageCLIError = require('../../manage_nsfs/manage_nsfs_cli_errors').ManageCLIError;
@@ -1848,3 +1854,140 @@ async function update_account_nsfs_config(email, default_resource, new_nsfs_acco
     }
 }
 
+mocha.describe('Presigned URL tests', function() {
+    this.timeout(50000); // eslint-disable-line no-invalid-this
+    const nsr = 'presigned_url_nsr';
+    const account_name = 'presigned_url_account';
+    const fs_path = path.join(TMP_PATH, 'presigned_url_tests/');
+    const presigned_url_bucket = 'presigned-url-bucket';
+    const presigned_url_object = 'presigned-url-object.txt';
+    const presigned_body = 'presigned_body';
+    let s3_client;
+    let access_key;
+    let secret_key;
+    CORETEST_ENDPOINT = coretest.get_http_address();
+    let valid_default_presigned_url;
+    let presigned_url_params;
+
+    mocha.before(async function() {
+        await fs_utils.create_fresh_path(fs_path);
+        await rpc_client.pool.create_namespace_resource({ name: nsr, nsfs_config: { fs_root_path: fs_path } });
+        const new_buckets_path = '/'; // manual fix for 5.16/5.15 since we didn't backport the related test refactoring
+        const nsfs_account_config = {
+            uid: process.getuid(), gid: process.getgid(), new_buckets_path, nsfs_only: true
+        };
+        const account_params = { ...new_account_params, email: `${account_name}@noobaa.io`, name: account_name, default_resource: nsr, nsfs_account_config };
+        const res = await rpc_client.account.create_account(account_params);
+        access_key = res.access_keys[0].access_key;
+        secret_key = res.access_keys[0].secret_key;
+        s3_client = generate_s3_client(access_key.unwrap(), secret_key.unwrap(), CORETEST_ENDPOINT);
+        await s3_client.createBucket({ Bucket: presigned_url_bucket });
+        await s3_client.putObject({ Bucket: presigned_url_bucket, Key: presigned_url_object, Body: presigned_body });
+
+        presigned_url_params = {
+            bucket: new SensitiveString(presigned_url_bucket),
+            key: presigned_url_object,
+            endpoint: CORETEST_ENDPOINT,
+            access_key: access_key,
+            secret_key: secret_key
+        };
+        valid_default_presigned_url = cloud_utils.get_signed_url(presigned_url_params);
+    });
+
+    mocha.after(async function() {
+        if (!is_nc_coretest) return;
+        await s3_client.deleteObject({ Bucket: presigned_url_bucket, Key: presigned_url_object });
+        await s3_client.deleteBucket({ Bucket: presigned_url_bucket });
+        await rpc_client.account.delete_account({ email: `${account_name}@noobaa.io` });
+        await fs_utils.folder_delete(fs_path);
+    });
+
+    it('fetch valid presigned URL - 604800 seconds - epoch expiry - should return object data', async () => {
+        const data = await fetchData(valid_default_presigned_url);
+        assert.equal(data, presigned_body);
+    });
+
+    it('fetch valid presigned URL - 604800 seconds - should return object data - with valid date + expiry in seconds', async () => {
+        const now = new Date();
+        const valid_url_with_date = valid_default_presigned_url + '&X-Amz-Date=' + now.toISOString() + '&X-Amz-Expires=' + 604800;
+        const data = await fetchData(valid_url_with_date);
+        assert.equal(data, presigned_body);
+    });
+
+    it('fetch invalid presigned URL - 604800 seconds - epoch expiry + with future date', async () => {
+        const now = new Date();
+        // Add one hour (3600000 milliseconds)
+        const one_hour_in_ms = 60 * 60 * 1000;
+        const one_hour_from_now = new Date(now.getTime() + one_hour_in_ms);
+        const future_presigned_url = valid_default_presigned_url + '&X-Amz-Date=' + one_hour_from_now.toISOString();
+        const expected_err = new S3Error(S3Error.RequestNotValidYet);
+        await assert_throws_async(fetchData(future_presigned_url), expected_err.message);
+    });
+
+    it('fetch invalid presigned URL - 604800 expiry seconds + with future date', async () => {
+        const now = new Date();
+        // Add one hour (3600000 milliseconds)
+        const one_hour_in_ms = 60 * 60 * 1000;
+        const one_hour_from_now = new Date(now.getTime() + one_hour_in_ms);
+        const future_presigned_url = valid_default_presigned_url + '&X-Amz-Date=' + one_hour_from_now.toISOString() + '&X-Amz-Expires=' + 604800;
+        const expected_err = new S3Error(S3Error.RequestNotValidYet);
+        await assert_throws_async(fetchData(future_presigned_url), expected_err.message);
+    });
+
+    it('fetch invalid presigned URL - 604800 seconds - epoch expiry - URL expired', async () => {
+        const expired_presigned_url = cloud_utils.get_signed_url(presigned_url_params, 1);
+        // wait for 2 seconds before fetching the url
+        await P.delay(2000);
+        const expected_err = new S3Error(S3Error.RequestExpired);
+        await assert_throws_async(fetchData(expired_presigned_url), expected_err.message);
+    });
+
+    it('fetch invalid presigned URL - 604800 expiry seconds - URL expired', async () => {
+        const now = new Date();
+        const expired_presigned_url = cloud_utils.get_signed_url(presigned_url_params, 1) + '&X-Amz-Date=' + now.toISOString() + '&X-Amz-Expires=' + 1;
+        // wait for 2 seconds before fetching the url
+        await P.delay(2000);
+        const expected_err = new S3Error(S3Error.RequestExpired);
+        await assert_throws_async(fetchData(expired_presigned_url), expected_err.message);
+    });
+
+    it('fetch invalid presigned URL - expiry expoch - expire in bigger than limit', async () => {
+        const invalid_expiry = 604800 + 10;
+        const invalid_expiry_presigned_url = cloud_utils.get_signed_url(presigned_url_params, invalid_expiry);
+        const expected_err = new S3Error(S3Error.AuthorizationQueryParametersError);
+        await assert_throws_async(fetchData(invalid_expiry_presigned_url), expected_err.message);
+    });
+
+    it('fetch invalid presigned URL - expire in bigger than limit', async () => {
+        const now = new Date();
+        const invalid_expiry = 604800 + 10;
+        const invalid_expiry_presigned_url = cloud_utils.get_signed_url(presigned_url_params, invalid_expiry) + '&X-Amz-Date=' + now.toISOString() + '&X-Amz-Expires=' + invalid_expiry;
+        const expected_err = new S3Error(S3Error.AuthorizationQueryParametersError);
+        await assert_throws_async(fetchData(invalid_expiry_presigned_url), expected_err.message);
+    });
+});
+
+async function fetchData(presigned_url) {
+    const response = await fetch(presigned_url, { agent: new http.Agent({ keepAlive: false }) });
+    let data;
+    if (!response.ok) {
+        data = (await response.text()).trim();
+        const err_json = (await http_utils.parse_xml_to_js(data)).Error;
+        const err = new Error(err_json.Message);
+        err.code = err_json.Code;
+        throw err;
+    }
+    data = await response.text();
+    return data.trim();
+}
+
+async function assert_throws_async(promise, expected_message = 'Access Denied') {
+    try {
+        await promise;
+        assert.fail('Test was suppose to fail on ' + expected_message);
+    } catch (err) {
+        if (err.message !== expected_message) {
+            throw err;
+        }
+    }
+}

src/test/unit_tests/test_file_writer.js

@@ -0,0 +1,69 @@
+/* Copyright (C) 2020 NooBaa */
+/* eslint-disable no-invalid-this */
+'use strict';
+
+const mocha = require('mocha');
+const config = require('../../../config');
+const file_writer_hashing = require('../../tools/file_writer_hashing');
+const orig_iov_max = config.NSFS_DEFAULT_IOV_MAX;
+
+// on iov_max small tests we need to use smaller amount of parts and chunks to ensure that the test will finish
+// in a reasonable period of time because we will flush max 1/2 buffers at a time.
+const small_iov_num_parts = 20;
+
+
+mocha.describe('FileWriter', function() {
+    const RUN_TIMEOUT = 10 * 60 * 1000;
+
+    mocha.afterEach(function() {
+        config.NSFS_DEFAULT_IOV_MAX = orig_iov_max;
+    });
+
+    mocha.it('Concurrent FileWriter with hash target', async function() {
+        const self = this;
+        self.timeout(RUN_TIMEOUT);
+        await file_writer_hashing.hash_target();
+    });
+
+    mocha.it('Concurrent FileWriter with file target', async function() {
+        const self = this;
+        self.timeout(RUN_TIMEOUT);
+        await file_writer_hashing.file_target();
+    });
+
+    mocha.it('Concurrent FileWriter with hash target - iov_max=1', async function() {
+        const self = this;
+        self.timeout(RUN_TIMEOUT);
+        await file_writer_hashing.hash_target(undefined, small_iov_num_parts, 1);
+    });
+
+    mocha.it('Concurrent FileWriter with file target - iov_max=1', async function() {
+        const self = this;
+        self.timeout(RUN_TIMEOUT);
+        await file_writer_hashing.file_target(undefined, small_iov_num_parts, 1);
+    });
+
+    mocha.it('Concurrent FileWriter with hash target - iov_max=2', async function() {
+        const self = this;
+        self.timeout(RUN_TIMEOUT);
+        await file_writer_hashing.hash_target(undefined, small_iov_num_parts, 2);
+    });
+
+    mocha.it('Concurrent FileWriter with file target - iov_max=2', async function() {
+        const self = this;
+        self.timeout(RUN_TIMEOUT);
+        await file_writer_hashing.file_target(undefined, small_iov_num_parts, 2);
+    });
+
+    mocha.it('Concurrent FileWriter with file target - produce num_chunks > 1024 && total_chunks_size < config.NSFS_BUF_SIZE_L', async function() {
+        const self = this;
+        self.timeout(RUN_TIMEOUT);
+        // The goal of this test is to produce num_chunks > 1024 && total_chunks_size < config.NSFS_BUF_SIZE_L
+        // so we will flush buffers because of reaching max num of buffers and not because we reached the max NSFS buf size
+        // chunk size = 100, num_chunks = (10 * 1024 * 1024)/100 < 104587, 104587 = num_chunks > 1024 
+        // chunk size = 100, total_chunks_size after having 1024 chunks is = 100 * 1024 < config.NSFS_BUF_SIZE_L
+        const chunk_size = 100;
+        const parts_s = 50;
+        await file_writer_hashing.file_target(chunk_size, parts_s);
+    });
+});