Merge pull request #8455 from nadavMiz/versioning_add_linkat

NSFS | versioning | use linkat in gpfs safe_link to not override existing versions

Author: nadavMiz

.github/workflows/jest-unit-tests.yaml

@@ -14,7 +14,7 @@ jobs:
 
       - name: Run Jest Unit Test
         run: |
-          sudo apt-get -y install nasm
+          sudo apt-get -y install nasm libcap-dev
           npm install
           npm run build
           npm run jest

src/deploy/NVA_build/builder.Dockerfile

@@ -21,7 +21,7 @@ RUN dnf update -y -q --nobest && \
 COPY ./src/deploy/NVA_build/install_arrow_build.sh ./src/deploy/NVA_build/install_arrow_build.sh
 ARG BUILD_S3SELECT_PARQUET=0
 RUN ./src/deploy/NVA_build/install_arrow_build.sh $BUILD_S3SELECT_PARQUET
-RUN dnf install -y -q wget unzip which vim python3 boost-devel && \
+RUN dnf install -y -q wget unzip which vim python3 boost-devel libcap-devel && \
     dnf group install -y -q "Development Tools" && \
     dnf clean all
 RUN version="2.15.05" && \

src/deploy/RPM_build/noobaa.spec

@@ -30,6 +30,7 @@ BuildRequires:  python3
 BuildRequires:  make
 BuildRequires:  gcc-c++
 BuildRequires:  boost-devel
+BuildRequires:  libcap-devel
 
 Recommends:     jemalloc
 

src/native/common.gypi

@@ -40,6 +40,7 @@
                 ],
                 'ldflags': [
                     '-lrt', # librt
+                    '-lcap',
                 ],
             }],
 

src/native/fs/fs_napi.cpp

@@ -559,6 +559,7 @@ struct FSWorker : public Napi::AsyncWorker
     int _warn_threshold_ms;
     double _took_time;
     Napi::FunctionReference _report_fs_stats;
+    bool _should_add_thread_capabilities;
 
     // executes the ctime check in the stat and read file fuctions
     // NOTE: If _do_ctime_check = false, then some functions will fallback to using mtime check
@@ -574,6 +575,7 @@ struct FSWorker : public Napi::AsyncWorker
         , _errno(0)
         , _warn_threshold_ms(0)
         , _took_time(0)
+        , _should_add_thread_capabilities(false)
         , _do_ctime_check(false)
     {
         for (int i = 0; i < (int)info.Length(); ++i) _args_ref.Set(i, info[i]);
@@ -607,6 +609,9 @@ struct FSWorker : public Napi::AsyncWorker
         tx.set_user(_uid, _gid);
         DBG1("FS::FSWorker::Execute: " << _desc << DVAL(_uid) << DVAL(_gid) << DVAL(geteuid()) << DVAL(getegid()) << DVAL(getuid()) << DVAL(getgid()));
 
+        if(_should_add_thread_capabilities) {
+            tx.add_thread_capabilities();
+        }
         auto start_time = std::chrono::high_resolution_clock::now();
         Work();
         auto end_time = std::chrono::high_resolution_clock::now();
@@ -641,6 +646,9 @@ struct FSWorker : public Napi::AsyncWorker
     {
         return gpfs_dl_path != NULL && gpfs_lib_file_exists > -1 && _backend == GPFS_BACKEND;
     }
+    void AddThreadCapabilities() {
+        _should_add_thread_capabilities = true;
+    }
     virtual void OnOK() override
     {
         DBG1("FS::FSWorker::OnOK: undefined " << _desc);
@@ -1619,15 +1627,24 @@ struct LinkFileAt : public FSWrapWorker<FileWrap>
 {
     std::string _filepath;
     int _replace_fd;
+    bool _should_not_override;
     LinkFileAt(const Napi::CallbackInfo& info)
         : FSWrapWorker<FileWrap>(info)
         , _replace_fd(-1)
+        , _should_not_override(false)
     {
         _filepath = info[1].As<Napi::String>();
         if (info.Length() > 2 && !info[2].IsUndefined()) {
             _replace_fd = info[2].As<Napi::Number>();
         }
-        Begin(XSTR() << "LinkFileAt " << DVAL(_wrap->_path) << DVAL(_wrap->_fd) << DVAL(_filepath));
+        if (info.Length() > 3 && !info[3].IsUndefined()) {
+            _should_not_override = info[3].As<Napi::Boolean>();
+        }
+        if(_replace_fd < 0  && _should_not_override) {
+            //set thread capabilities to allow linkat from user other than root.
+            AddThreadCapabilities();
+        }
+        Begin(XSTR() << "LinkFileAt " << DVAL(_wrap->_path) << DVAL(_wrap->_fd) << DVAL(_filepath) << DVAL(_should_not_override));
     }
     virtual void Work()
     {
@@ -1638,6 +1655,8 @@ struct LinkFileAt : public FSWrapWorker<FileWrap>
         // Linux will fail the linkat() if the file already exist and we want to replace it if it existed.
         if (_replace_fd >= 0) {
             SYSCALL_OR_RETURN(dlsym_gpfs_linkatif(fd, "", AT_FDCWD, _filepath.c_str(), AT_EMPTY_PATH, _replace_fd));
+        } else if (_should_not_override){
+            SYSCALL_OR_RETURN(linkat(fd, "", AT_FDCWD, _filepath.c_str(), AT_EMPTY_PATH));
         } else {
             SYSCALL_OR_RETURN(dlsym_gpfs_linkat(fd, "", AT_FDCWD, _filepath.c_str(), AT_EMPTY_PATH));
         }

src/native/util/os.h

@@ -38,6 +38,8 @@ class ThreadScope
         change_user();
     }
 
+    int add_thread_capabilities();
+
     const static uid_t orig_uid;
     const static gid_t orig_gid;
     const static std::vector<gid_t> orig_groups;

src/native/util/os_darwin.cpp

@@ -54,6 +54,14 @@ ThreadScope::restore_user()
     }
 }
 
+int
+ThreadScope::add_thread_capabilities()
+{
+    //set capabilities not used in darwin
+    LOG("function set_capabilities_linkat is unsupported in darwin");
+    return -1;
+}
+
 } // namespace noobaa
 
 #endif

src/native/util/os_linux.cpp

@@ -5,6 +5,7 @@
 
 #include <sys/syscall.h>
 #include <sys/types.h>
+#include <sys/capability.h>
 #include <grp.h>
 #include <limits.h>
 
@@ -35,7 +36,7 @@ const std::vector<gid_t> ThreadScope::orig_groups = [] {
 
 /**
  * set the effective uid/gid of the current thread using direct syscalls
- * unsets the supplementary group IDs for the current thread using direct syscall 
+ * unsets the supplementary group IDs for the current thread using direct syscall
  * we have to bypass the libc wrappers because posix requires it to syncronize
  * uid & gid to all threads which is undesirable in our case.
  */
@@ -64,6 +65,41 @@ ThreadScope::restore_user()
     }
 }
 
+int
+ThreadScope::add_thread_capabilities() {
+    cap_t caps = cap_get_proc();
+    cap_flag_value_t cap_flag_value;
+    if(caps == NULL) {
+        LOG("ThreadScope::set_thread_capabilities - cap_get_proc failed");
+        return -1;
+    }
+    if(cap_get_flag(caps, CAP_DAC_READ_SEARCH, CAP_EFFECTIVE, &cap_flag_value) < 0) {
+        LOG("ThreadScope::set_thread_capabilities - cap_get_flag failed");
+        cap_free(caps);
+        return -1;
+    }
+    if(cap_flag_value == CAP_SET) {
+        LOG("ThreadScope::cap_flag_value - capability already exists");
+        cap_free(caps);
+        return 0;
+    }
+    cap_value_t newcaps[1] = { CAP_DAC_READ_SEARCH, };
+    if(cap_set_flag(caps, CAP_EFFECTIVE, 1, newcaps, CAP_SET) < 0) {
+        LOG("ThreadScope::set_thread_capabilities - cap_set_flag failed");
+        cap_free(caps);
+        return -1;
+    }
+    if(cap_set_proc(caps) < 0) {
+        LOG("ThreadScope::set_thread_capabilities - cap_set_proc failed");
+        cap_free(caps);
+        return -1;
+    }
+    if(cap_free(caps) < 0) {
+        LOG("cap_free failed");
+    }
+    return 0;
+}
+
 } // namespace noobaa
 
 #endif

src/sdk/namespace_fs.js

@@ -1473,15 +1473,23 @@ class NamespaceFS {
         let retries = config.NSFS_RENAME_RETRIES;
         for (;;) {
             try {
-                const new_ver_info = !is_gpfs && await this._get_version_info(fs_context, new_ver_tmp_path);
-                // get latest version_id if exists
-                const latest_ver_info = await this._get_version_info(fs_context, latest_ver_path);
-                const versioned_path = latest_ver_info && this._get_version_path(key, latest_ver_info.version_id_str);
-                const versioned_info = latest_ver_info && await this._get_version_info(fs_context, versioned_path);
-
-                gpfs_options = await this._open_files_gpfs(fs_context, new_ver_tmp_path, latest_ver_path, upload_file,
-                    latest_ver_info, open_mode, undefined, versioned_info);
+                let new_ver_info;
+                let latest_ver_info;
+                if (is_gpfs) {
+                    const latest_ver_info_exist = await native_fs_utils.is_path_exists(fs_context, latest_ver_path);
+                    gpfs_options = await this._open_files_gpfs(fs_context, new_ver_tmp_path, latest_ver_path, upload_file,
+                        latest_ver_info_exist, open_mode, undefined, undefined);
+
+                    //get latest version if exists
+                    const latest_fd = gpfs_options?.move_to_dst?.dst_file;
+                    latest_ver_info = latest_fd && await this._get_version_info(fs_context, undefined, latest_fd);
+                } else {
+                    new_ver_info = await this._get_version_info(fs_context, new_ver_tmp_path);
+                    //get latest version if exists. TODO use fd like in GPFS
+                    latest_ver_info = await this._get_version_info(fs_context, latest_ver_path);
+                }
                 const bucket_tmp_dir_path = this.get_bucket_tmpdir_full_path();
+                const versioned_path = latest_ver_info && this._get_version_path(key, latest_ver_info.version_id_str);
                 dbg.log1('Namespace_fs._move_to_dest_version:', latest_ver_info, new_ver_info, gpfs_options);
 
                 if (this._is_versioning_suspended()) {
@@ -2725,10 +2733,12 @@ class NamespaceFS {
     // path - specifies the path to version
     // if version xattr contains version info - return info by xattr
     // else - it's a null version - return stat
-    async _get_version_info(fs_context, version_path) {
+    // if fd is passed, will use fd instead of path to stat
+    async _get_version_info(fs_context, version_path, fd) {
         try {
-            const stat = await nb_native().fs.stat(fs_context, version_path, { skip_user_xattr: true });
-            dbg.log1('NamespaceFS._get_version_info stat ', stat, version_path);
+            const stat = fd ? await fd.stat(fs_context, { skip_user_xattr: true }) :
+                await nb_native().fs.stat(fs_context, version_path, { skip_user_xattr: true });
+            dbg.log1('NamespaceFS._get_version_info stat ', stat, version_path, fd);
 
             const version_id_str = this._get_version_id_by_xattr(stat);
             const ver_info_by_xattr = this._extract_version_info_from_xattr(version_id_str);
@@ -3152,8 +3162,8 @@ class NamespaceFS {
     // opens the unopened files involved in the version move during upload/deletion
     // returns an object contains the relevant options for the move/unlink flow
     // eslint-disable-next-line max-params
-    async _open_files_gpfs(fs_context, src_path, dst_path, upload_or_dir_file, dst_ver_info, open_mode, delete_version, versioned_info) {
-        dbg.log1('Namespace_fs._open_files_gpfs:', src_path, src_path && path.dirname(src_path), dst_path, upload_or_dir_file, dst_ver_info, open_mode, delete_version, versioned_info);
+    async _open_files_gpfs(fs_context, src_path, dst_path, upload_or_dir_file, dst_ver_exist, open_mode, delete_version, versioned_info) {
+        dbg.log1('Namespace_fs._open_files_gpfs:', src_path, src_path && path.dirname(src_path), dst_path, upload_or_dir_file, Boolean(dst_ver_exist), open_mode, delete_version, versioned_info);
         const is_gpfs = native_fs_utils._is_gpfs(fs_context);
         if (!is_gpfs) return;
 
@@ -3179,13 +3189,13 @@ class NamespaceFS {
                 src_file = upload_or_dir_file;
                 dir_file = await native_fs_utils.open_file(fs_context, this.bucket_path, path.dirname(src_path), 'r');
             }
-            if (dst_ver_info) {
+            if (dst_ver_exist) {
                 dbg.log1('NamespaceFS._open_files_gpfs dst version exist - opening dst version file...');
                 dst_file = await native_fs_utils.open_file(fs_context, this.bucket_path, dst_path, 'r');
             }
             return {
-                move_to_versions: { src_file: dst_file, dir_file, should_override: false },
-                move_to_dst: { src_file, dst_file, dir_file, versioned_file }
+                move_to_versions: { src_file: dst_file, dir_file },
+                move_to_dst: { src_file, dst_file, dir_file}
             };
         } catch (err) {
             dbg.error('NamespaceFS._open_files_gpfs couldn\'t open files', err);

src/sdk/nb.d.ts

@@ -1000,7 +1000,7 @@ interface NativeFile {
     write(fs_context: NativeFSContext, buffer: Buffer, len: number, offset?: number): Promise<void>;
     writev(fs_context: NativeFSContext, buffers: Buffer[], offset?: number): Promise<void>;
     replacexattr(fs_context: NativeFSContext, xattr: NativeFSXattr, clear_prefix?: string): Promise<void>;
-    linkfileat(fs_context: NativeFSContext, path: string, fd?: number): Promise<void>;
+    linkfileat(fs_context: NativeFSContext, path: string, fd?: number, should_not_override?: boolean): Promise<void>;
     fsync(fs_context: NativeFSContext): Promise<void>;
     fd: number;
     flock(fs_context: NativeFSContext, operation: "EXCLUSIVE" | "SHARED" | "UNLOCK"): Promise<void>;