Skip to content

Commit 6b2fe2f

Browse files
romayalonnadavMiz
romayalon
authored and
nadavMiz
committed
Merge pull request noobaa#8405 from romayalon/romy-concurrency-nested-key
NSFS | Versioning | Concurrency tests
2 parents e24e59a + f84510e commit 6b2fe2f

File tree

4 files changed

+409
-85
lines changed

4 files changed

+409
-85
lines changed

‎docs/NooBaaNonContainerized/CI&Tests.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,8 @@ The following is a list of `NC jest tests` files -
104104
9. `test_nc_nsfs_account_schema_validation.test.js` - Tests NC account schema validation.
105105
10. `test_nc_nsfs_new_buckets_path_validation.test.js` - Tests new_buckets_path RW access.
106106
11. `test_config_fs.test.js` - Tests ConfigFS methods.
107+
12. `test_nsfs_concurrency` - Tests concurrent operations.
108+
13. `test_versioning_concurrency` - Tests concurrent operations on versioned enabled bucket.
107109

108110
#### nc_index.js File
109111
* The `nc_index.js` is a file that runs several NC and NSFS mocha related tests.

‎src/sdk/namespace_fs.js‎

Lines changed: 106 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -920,22 +920,39 @@ class NamespaceFS {
920920
async read_object_md(params, object_sdk) {
921921
const fs_context = this.prepare_fs_context(object_sdk);
922922
let file_path;
923+
let stat;
924+
let isDir;
925+
let retries = (this._is_versioning_enabled() || this._is_versioning_suspended()) ? config.NSFS_RENAME_RETRIES : 0;
926+
const is_gpfs = native_fs_utils._is_gpfs(fs_context);
923927
try {
924-
file_path = await this._find_version_path(fs_context, params, true);
925-
await this._check_path_in_bucket_boundaries(fs_context, file_path);
926-
await this._load_bucket(params, fs_context);
927-
let stat = await nb_native().fs.stat(fs_context, file_path);
928-
929-
const isDir = native_fs_utils.isDirectory(stat);
930-
if (isDir) {
931-
if (!stat.xattr?.[XATTR_DIR_CONTENT] || !params.key.endsWith('/')) {
932-
throw error_utils.new_error_code('ENOENT', 'NoSuchKey');
933-
} else if (stat.xattr?.[XATTR_DIR_CONTENT] !== '0') {
934-
// find dir object content file path and return its stat + xattr of its parent directory
935-
const dir_content_path = await this._find_version_path(fs_context, params);
936-
const dir_content_path_stat = await nb_native().fs.stat(fs_context, dir_content_path);
937-
const xattr = stat.xattr;
938-
stat = { ...dir_content_path_stat, xattr };
928+
for (;;) {
929+
try {
930+
file_path = await this._find_version_path(fs_context, params, true);
931+
await this._check_path_in_bucket_boundaries(fs_context, file_path);
932+
await this._load_bucket(params, fs_context);
933+
stat = await nb_native().fs.stat(fs_context, file_path);
934+
935+
isDir = native_fs_utils.isDirectory(stat);
936+
if (isDir) {
937+
if (!stat.xattr?.[XATTR_DIR_CONTENT] || !params.key.endsWith('/')) {
938+
throw error_utils.new_error_code('ENOENT', 'NoSuchKey');
939+
} else if (stat.xattr?.[XATTR_DIR_CONTENT] !== '0') {
940+
// find dir object content file path and return its stat + xattr of its parent directory
941+
const dir_content_path = await this._find_version_path(fs_context, params);
942+
const dir_content_path_stat = await nb_native().fs.stat(fs_context, dir_content_path);
943+
const xattr = stat.xattr;
944+
stat = { ...dir_content_path_stat, xattr };
945+
}
946+
}
947+
if (this._is_mismatch_version_id(stat, params.version_id)) {
948+
dbg.warn('NamespaceFS.read_object_md mismatch version_id', file_path, params.version_id, this._get_version_id_by_xattr(stat));
949+
throw error_utils.new_error_code('MISMATCH_VERSION', 'file version does not match the version we asked for');
950+
}
951+
break;
952+
} catch (err) {
953+
dbg.warn(`NamespaceFS.read_object_md: retrying retries=${retries} file_path=${file_path}`, err);
954+
retries -= 1;
955+
if (retries <= 0 || !native_fs_utils.should_retry_link_unlink(is_gpfs, err)) throw err;
939956
}
940957
}
941958
this._throw_if_delete_marker(stat, params);
@@ -948,6 +965,23 @@ class NamespaceFS {
948965
}
949966
}
950967

968+
async _is_empty_directory_content(file_path, fs_context, params) {
969+
const is_dir_content = this._is_directory_content(file_path, params.key);
970+
if (is_dir_content) {
971+
try {
972+
const md_path = this._get_file_md_path(params);
973+
const dir_stat = await nb_native().fs.stat(fs_context, md_path);
974+
if (dir_stat && dir_stat.xattr[XATTR_DIR_CONTENT] === '0') return true;
975+
} catch (err) {
976+
//failed to get object
977+
new NoobaaEvent(NoobaaEvent.OBJECT_GET_FAILED).create_event(params.key,
978+
{bucket_path: this.bucket_path, object_name: params.key}, err);
979+
dbg.log0('NamespaceFS: read_object_stream couldnt find dir content xattr', err);
980+
}
981+
}
982+
return false;
983+
}
984+
951985
// eslint-disable-next-line max-statements
952986
async read_object_stream(params, object_sdk, res) {
953987
let file;
@@ -956,34 +990,41 @@ class NamespaceFS {
956990
let file_path;
957991
try {
958992
await this._load_bucket(params, fs_context);
959-
file_path = await this._find_version_path(fs_context, params);
960-
await this._check_path_in_bucket_boundaries(fs_context, file_path);
961-
962-
// NOTE: don't move this code after the open
963-
// this can lead to ENOENT failures due to file not exists when content size is 0
964-
// if entry is a directory object and its content size = 0 - return empty response
965-
const is_dir_content = this._is_directory_content(file_path, params.key);
966-
if (is_dir_content) {
993+
let retries = (this._is_versioning_disabled() || this._is_versioning_suspended()) ? config.NSFS_RENAME_RETRIES : 0;
994+
const is_gpfs = native_fs_utils._is_gpfs(fs_context);
995+
let stat;
996+
for (;;) {
967997
try {
968-
const md_path = this._get_file_md_path(params);
969-
const dir_stat = await nb_native().fs.stat(fs_context, md_path);
970-
if (dir_stat && dir_stat.xattr[XATTR_DIR_CONTENT] === '0') return null;
998+
file_path = await this._find_version_path(fs_context, params);
999+
await this._check_path_in_bucket_boundaries(fs_context, file_path);
1000+
1001+
// NOTE: don't move this code after the open
1002+
// this can lead to ENOENT failures due to file not exists when content size is 0
1003+
// if entry is a directory object and its content size = 0 - return empty response
1004+
if (await this._is_empty_directory_content(file_path, fs_context, params)) return null;
1005+
1006+
file = await nb_native().fs.open(
1007+
fs_context,
1008+
file_path,
1009+
config.NSFS_OPEN_READ_MODE,
1010+
native_fs_utils.get_umasked_mode(config.BASE_MODE_FILE),
1011+
);
1012+
stat = await file.stat(fs_context);
1013+
if (this._is_mismatch_version_id(stat, params.version_id)) {
1014+
dbg.warn('NamespaceFS.read_object_stream mismatch version_id', params.version_id, this._get_version_id_by_xattr(stat));
1015+
throw error_utils.new_error_code('MISMATCH_VERSION', 'file version does not match the version we asked for');
1016+
}
1017+
break;
9711018
} catch (err) {
972-
//failed to get object
973-
new NoobaaEvent(NoobaaEvent.OBJECT_GET_FAILED).create_event(params.key,
974-
{bucket_path: this.bucket_path, object_name: params.key}, err);
975-
dbg.log0('NamespaceFS: read_object_stream couldnt find dir content xattr', err);
1019+
dbg.warn(`NamespaceFS.read_object_stream: retrying retries=${retries} file_path=${file_path}`, err);
1020+
retries -= 1;
1021+
if (retries <= 0 || !native_fs_utils.should_retry_link_unlink(is_gpfs, err)) {
1022+
new NoobaaEvent(NoobaaEvent.OBJECT_GET_FAILED).create_event(params.key,
1023+
{bucket_path: this.bucket_path, object_name: params.key}, err);
1024+
throw err;
1025+
}
9761026
}
9771027
}
978-
979-
file = await nb_native().fs.open(
980-
fs_context,
981-
file_path,
982-
config.NSFS_OPEN_READ_MODE,
983-
native_fs_utils.get_umasked_mode(config.BASE_MODE_FILE),
984-
);
985-
986-
const stat = await file.stat(fs_context);
9871028
this._throw_if_delete_marker(stat, params);
9881029
// await this._fail_if_archived_or_sparse_file(fs_context, file_path, stat);
9891030

@@ -1244,14 +1285,14 @@ class NamespaceFS {
12441285

12451286
/**
12461287
* _check_copy_storage_class returns true if a copy is needed to be forced.
1247-
*
1288+
*
12481289
* This might be needed if we need to manage xattr separately on the source
12491290
* object and target object (eg. GLACIER objects).
1250-
*
1291+
*
12511292
* NOTE: The function will throw S3 error if source object storage class is
12521293
* "GLACIER" but it is not in restored state (AWS behaviour).
1253-
* @param {nb.NativeFSContext} fs_context
1254-
* @param {Record<any, any>} params
1294+
* @param {nb.NativeFSContext} fs_context
1295+
* @param {Record<any, any>} params
12551296
* @returns {Promise<boolean>}
12561297
*/
12571298
async _check_copy_storage_class(fs_context, params) {
@@ -1391,8 +1432,8 @@ class NamespaceFS {
13911432
}
13921433

13931434
// 1. get latest version_id
1394-
// 2. if versioning is suspended -
1395-
// 2.1 if version ID of the latest version is null -
1435+
// 2. if versioning is suspended -
1436+
// 2.1 if version ID of the latest version is null -
13961437
// 2.1.1 remove latest version
13971438
// 2.2 else (version ID of the latest version is unique or there is no latest version) -
13981439
// 2.2.1 remove a version (or delete marker) with null version ID from .versions/ (if exists)
@@ -1557,7 +1598,7 @@ class NamespaceFS {
15571598
const delimiter_idx = create_params_parsed.key.indexOf(params.delimiter, start_idx);
15581599
if (delimiter_idx > 0) {
15591600
common_prefixes_set.add(create_params_parsed.key.substring(0, delimiter_idx + 1));
1560-
// if key has common prefix it should not be returned as an upload object
1601+
// if key has common prefix it should not be returned as an upload object
15611602
return undefined;
15621603
}
15631604
}
@@ -1606,7 +1647,7 @@ class NamespaceFS {
16061647
return path.join(params.mpu_path, `part-${params.num}`);
16071648
}
16081649

1609-
// optimized version of upload_multipart -
1650+
// optimized version of upload_multipart -
16101651
// 1. if size is pre known -
16111652
// 1.1. calc offset
16121653
// 1.2. upload data to by_size file in offset position
@@ -1710,13 +1751,13 @@ class NamespaceFS {
17101751
}
17111752
}
17121753

1713-
// iterate over multiparts array -
1754+
// iterate over multiparts array -
17141755
// 1. if num of unique sizes is 1
17151756
// 1.1. if this is the last part - link the size file and break the loop
17161757
// 1.2. else, continue the loop
17171758
// 2. if num of unique sizes is 2
17181759
// 2.1. if should_copy_file_prefix
1719-
// 2.1.1. if the cur part is the last, link the previous part file to upload_path and copy the last part (tail) to upload_path
1760+
// 2.1.1. if the cur part is the last, link the previous part file to upload_path and copy the last part (tail) to upload_path
17201761
// 2.1.2. else - copy the prev part size file prefix to upload_path
17211762
// 3. copy bytes of the current's part size file
17221763
async complete_object_upload(params, object_sdk) {
@@ -2069,12 +2110,12 @@ class NamespaceFS {
20692110
/**
20702111
* restore_object simply sets the restore request xattr
20712112
* which should be picked by another mechanism.
2072-
*
2113+
*
20732114
* restore_object internally relies on 2 xattrs:
20742115
* - XATTR_RESTORE_REQUEST
20752116
* - XATTR_RESTORE_EXPIRY
2076-
* @param {*} params
2077-
* @param {nb.ObjectSDK} object_sdk
2117+
* @param {*} params
2118+
* @param {nb.ObjectSDK} object_sdk
20782119
* @returns {Promise<boolean>}
20792120
*/
20802121
async restore_object(params, object_sdk) {
@@ -2195,7 +2236,7 @@ class NamespaceFS {
21952236
}
21962237

21972238
/**
2198-
*
2239+
*
21992240
* @param {*} fs_context - fs context object
22002241
* @param {string} file_path - path to file
22012242
* @param {*} set - the xattr object to be set
@@ -2533,7 +2574,7 @@ class NamespaceFS {
25332574
}
25342575
try {
25352576
// Returns the real path of the entry.
2536-
// The entry path may point to regular file or directory, but can have symbolic links
2577+
// The entry path may point to regular file or directory, but can have symbolic links
25372578
const full_path = await nb_native().fs.realpath(fs_context, entry_path);
25382579
if (!full_path.startsWith(this.bucket_path)) {
25392580
dbg.log0('check_bucket_boundaries: the path', entry_path, 'is not in the bucket', this.bucket_path, 'boundaries');
@@ -2725,6 +2766,10 @@ class NamespaceFS {
27252766
}
27262767
}
27272768

2769+
_is_mismatch_version_id(stat, version_id) {
2770+
return version_id && !this._is_versioning_disabled() && this._get_version_id_by_xattr(stat) !== version_id;
2771+
}
2772+
27282773
/**
27292774
* @param {nb.NativeFSContext} fs_context
27302775
* @param {string} key
@@ -2771,9 +2816,9 @@ class NamespaceFS {
27712816
// there are a few concurrency scenarios that might happen we should retry for -
27722817
// 1. the version id is the latest, concurrent put will might move the version id from being the latest to .versions/ -
27732818
// will throw safe unlink failed on non matching fd (on GPFS) or inode/mtime (on POSIX).
2774-
// 2. the version id is the second latest and stays under .versions/ - on concurrent delete of the latest,
2819+
// 2. the version id is the second latest and stays under .versions/ - on concurrent delete of the latest,
27752820
// the version id might move to be the latest and we will get ENOENT
2776-
// 3. concurrent delete of this version - will get ENOENT, doing a retry will return successfully
2821+
// 3. concurrent delete of this version - will get ENOENT, doing a retry will return successfully
27772822
// after we will see that the version was already deleted
27782823
if (retries <= 0 || !native_fs_utils.should_retry_link_unlink(is_gpfs, err)) throw err;
27792824
} finally {
@@ -2930,8 +2975,8 @@ class NamespaceFS {
29302975
const bucket_tmp_dir_path = this.get_bucket_tmpdir_full_path();
29312976
if (this._is_versioning_enabled() || suspended_and_latest_is_not_null) {
29322977
await native_fs_utils._make_path_dirs(versioned_path, fs_context);
2933-
await native_fs_utils.safe_move(fs_context, latest_ver_path, versioned_path, latest_ver_info,
2934-
gpfs_options && gpfs_options.delete_version, bucket_tmp_dir_path);
2978+
await native_fs_utils.safe_move_posix(fs_context, latest_ver_path, versioned_path, latest_ver_info,
2979+
bucket_tmp_dir_path);
29352980
if (suspended_and_latest_is_not_null) {
29362981
// remove a version (or delete marker) with null version ID from .versions/ (if exists)
29372982
await this._delete_null_version_from_versions_directory(params.key, fs_context);
@@ -2945,9 +2990,9 @@ class NamespaceFS {
29452990
}
29462991
break;
29472992
} catch (err) {
2993+
dbg.warn(`NamespaceFS._delete_latest_version: Retrying retries=${retries} latest_ver_path=${latest_ver_path}`, err);
29482994
retries -= 1;
29492995
if (retries <= 0 || !native_fs_utils.should_retry_link_unlink(is_gpfs, err)) throw err;
2950-
dbg.warn(`NamespaceFS._delete_latest_version: Retrying retries=${retries} latest_ver_path=${latest_ver_path}`, err);
29512996
} finally {
29522997
if (gpfs_options) await this._close_files_gpfs(fs_context, gpfs_options.delete_version, undefined, true);
29532998
}
@@ -2962,7 +3007,7 @@ class NamespaceFS {
29623007

29633008
// We can have only one versioned object with null version ID per key.
29643009
// It can be latest version, old version in .version/ directory or delete marker
2965-
// This function removes an object version or delete marker with a null version ID inside .version/ directory
3010+
// This function removes an object version or delete marker with a null version ID inside .version/ directory
29663011
async _delete_null_version_from_versions_directory(key, fs_context) {
29673012
const is_gpfs = native_fs_utils._is_gpfs(fs_context);
29683013
let retries = config.NSFS_RENAME_RETRIES;
@@ -3098,7 +3143,7 @@ class NamespaceFS {
30983143
dst_file = await native_fs_utils.open_file(fs_context, this.bucket_path, dst_path, 'r');
30993144
}
31003145
return {
3101-
move_to_versions: { src_file: dst_file, dir_file, dst_file: versioned_file },
3146+
move_to_versions: { src_file: dst_file, dir_file, is_move_to_versions: true },
31023147
move_to_dst: { src_file, dst_file, dir_file, versioned_file }
31033148
};
31043149
} catch (err) {

0 commit comments

Comments
 (0)