Commit 6d690721 authored by Ramon Nou's avatar Ramon Nou
Browse files

test: improve malleability integration test cleanup 

Propagate workspace PATH for shell clients while composing library paths
consistently, and avoid leaking LD_PRELOAD when interception is disabled.

Make malleability performance tests more robust by retrying transient storage
cleanup and file creation failures, including EBUSY, EDQUOT, and ENOSPC, so
successive runs are less likely to fail due to backend teardown races.
parent ec8e43b8

tests/integration/harness/gkfs.py

+15 −4
Original line number Diff line number Diff line
@@ -893,9 +893,16 @@ class ShellClient: 
        self._env = os.environ.copy()

        self._proxy = proxy



        libdirs = ':'.join(

                filter(None, [os.environ.get('LD_LIBRARY_PATH', '')] +

                             [str(p) for p in self._workspace.libdirs]))

        bindirs = os.pathsep.join(str(p) for p in self._workspace.bindirs)

        if bindirs:

            self._env['PATH'] = os.pathsep.join(

                filter(None, [bindirs, os.environ.get('PATH', '')]))



        libdirs = _compose_ld_library_path(

            preferred_dirs=[],

            workspace_dirs=self._workspace.libdirs,

            inherited_ld_path=os.environ.get('LD_LIBRARY_PATH', ''),

        )



        # ensure the client interception library is available:

        # to avoid running code with potentially installed libraries,
@@ -1008,8 +1015,12 @@ class ShellClient: 
            logger.debug(f"patched env: {self._patched_env}")



        

        script_env = self._env.copy()

        if not intercept_shell:

            script_env.pop('LD_PRELOAD', None)



        proc = subprocess.Popen(['bash', '-c', code],

            env = (self._env if intercept_shell else os.environ),

            env = script_env,

            stdout=subprocess.PIPE,

            stderr=subprocess.PIPE,

            )

tests/integration/malleability/test_malleability_performance.py

+49 −41
Original line number Diff line number Diff line
@@ -43,6 +43,7 @@ import stat 
import time

import math

import hashlib

import errno

import json

import shutil

import statistics
@@ -67,9 +68,7 @@ def _shutdown_daemons(daemons, timeout=10): 
    for daemon in daemons:

        try:

            proc = getattr(daemon, "_proc", None)

            if proc is None or proc.poll() is not None:

                continue



            if proc is not None and proc.poll() is None:

                logger.debug(f"terminating daemon pid {proc.pid}")

                proc.terminate()

                try:
@@ -80,14 +79,8 @@ def _shutdown_daemons(daemons, timeout=10): 
                    proc.wait(timeout=timeout)

        except Exception as exc:

            logger.warning(f"daemon cleanup failed: {exc}")



        shutil.rmtree(daemon.rootdir.as_posix(), ignore_errors=True)



    # Give the backend a moment to release the mount and RPC state before the

    # next performance run starts. Without this pause, the next run may hit

    # transient EBUSY failures on file creation.

    time.sleep(2)



        finally:

            _cleanup_daemon_storage(daemon)



def _snapshot_chunk_files(rootdir):

    """Snapshot chunk files under a daemon rootdir.
@@ -219,12 +212,31 @@ def _wait_for_unmounted(mountdir, timeout=10): 




def _cleanup_daemon_storage(daemon):

    shutil.rmtree(daemon.rootdir.as_posix(), ignore_errors=True)

    shutil.rmtree(daemon.metadir.as_posix(), ignore_errors=True)

    for path in (daemon.rootdir, daemon.metadir):

        path = Path(path)

        for attempt in range(20):

            shutil.rmtree(path.as_posix(), ignore_errors=True)

            if not path.exists():

                break

            time.sleep(0.25)

        if path.exists():

            logger.warning(f"failed to remove daemon storage {path}")

    #if daemon.logdir.exists():

    #    shutil.rmtree(daemon.logdir.as_posix(), ignore_errors=True)





def _cleanup_iteration_workspace(iter_workspace):

    for path in (iter_workspace.rootdir, iter_workspace.metadir, iter_workspace.mountdir):

        path = Path(path)

        for attempt in range(20):

            shutil.rmtree(path.as_posix(), ignore_errors=True)

            if not path.exists():

                break

            time.sleep(0.25)

        if path.exists():

            logger.warning(f"failed to remove iteration workspace path {path}")





class _IterationWorkspaceAdapter:

    """Per-iteration workspace with fresh mount/root/meta/log dirs."""
@@ -331,16 +343,26 @@ def create_deterministic_file(client, mountdir, filename, size): 


    last_ret = None

    seed = 0x474b4653 ^ sum(ord(ch) for ch in filename) ^ size

    retry_errnos = {errno.EBUSY, errno.EDQUOT, errno.ENOSPC}

    for attempt in range(30):

        ret = client.write_random_and_md5(fpath, size, "--seed", seed, timeout=max(60, int(size / (1024 * 1024)) * 5))

        last_ret = ret

        if ret.retval != -1:

            break

        last_errno = getattr(ret, 'errno', None)

        if last_errno != 16:

        if last_errno not in retry_errnos:

            break

        logger.warning(f"open busy for {fpath} (attempt {attempt + 1}/30), retrying")

        logger.debug(f"  mountdir exists={mountdir.exists()} ismount={os.path.ismount(mountdir)}")

        try:

            usage = shutil.disk_usage(mountdir)

            free = usage.free

        except OSError:

            free = "unknown"

        logger.warning(

            f"transient write failure for {fpath}, errno={last_errno} "

            f"(attempt {attempt + 1}/30), retrying")

        logger.debug(

            f"  mountdir exists={mountdir.exists()} "

            f"ismount={os.path.ismount(mountdir)} free={free}")

        time.sleep(1)



    assert last_ret is not None
@@ -606,6 +628,7 @@ DEFAULT_FILE_SIZE = int(os.environ.get("GKFS_PERF_FILE_SIZE", str(8 * 1024))) 
DEFAULT_OLD_NODES = int(os.environ.get("GKFS_MALLEABILITY_OLD_NODES", "4"))

DEFAULT_NEW_NODES = int(os.environ.get("GKFS_MALLEABILITY_NEW_NODES", "2"))

DEFAULT_TIMEOUT = 340

MUTATE_STATUS_POLL_INTERVAL = 0.25

CI_FAST = os.environ.get("GKFS_MALLEABILITY_CI_FAST", "").lower() in ("1", "on", "true", "yes")

KEEP_ITERATION_WORKSPACES = os.environ.get(

    "GKFS_MALLEABILITY_KEEP_WORKSPACES", "").lower() in ("1", "on", "true", "yes")
@@ -669,7 +692,6 @@ def test_shrink_performance_multi_run(client_fixture, 
    all_daemons = []

    

    for run_idx in range(num_reps):

        time.sleep(10)  # Give the backend time to release resources from previous run

        logger.info("=" * 70)

        logger.info(f"SHRINK PERFORMANCE TEST - RUN {run_idx + 1}/{num_reps}")

        logger.info("=" * 70)
@@ -677,17 +699,18 @@ def test_shrink_performance_multi_run(client_fixture, 
        base_workspace = request.getfixturevalue("test_workspace")

        iter_workspace = _make_iteration_workspace(base_workspace, f"shrink_iter_{run_idx}")

        iter_daemon_factory = FwdDaemonCreator(request.config.getoption('--interface'), iter_workspace, keep_hosts=True)

        daemons = []

        run_mountdir = None

        try:

            if hasattr(client, "_patched_env"):

                client._patched_env["LIBGKFS_HOSTS_FILE"] = str(iter_workspace.twd / "gkfs_hosts.txt")

            if hasattr(client, "_env"):

                client._env["LIBGKFS_HOSTS_FILE"] = str(iter_workspace.twd / "gkfs_hosts.txt")

        

            # Create daemons

        daemons = []

            for i in range(num_start):

                d = iter_daemon_factory.create(enable_forwarding=False)

                daemons.append(d)

        time.sleep(5)

            all_daemons.extend(daemons)

        

            hostfile = Path(daemons[0].hostfile)
@@ -743,7 +766,7 @@ def test_shrink_performance_multi_run(client_fixture, 
                cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

                if "No mutate running/finished." in cmd.stderr.decode():

                    break

            time.sleep(2)

                time.sleep(MUTATE_STATUS_POLL_INTERVAL)

            else:

                pytest.fail(f"Shrink did not complete within 120s (run {run_idx + 1})")
@@ -755,7 +778,6 @@ def test_shrink_performance_multi_run(client_fixture, 
            cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

            assert cmd.exit_code == 0

            _set_client_hostfile(client, workspace_file)

        time.sleep(2)

        

            # Verify

            accessible = count_accessible_files(created_files, client)
@@ -782,11 +804,10 @@ def test_shrink_performance_multi_run(client_fixture, 
        

            logger.info(f"  Run {run_idx + 1}: shrink took {elapsed:.3f}s, files: {accessible}/{num_files}")

        

        # Clean up: shutdown daemons and wipe backend storage for fresh iteration

        finally:

            _shutdown_daemons(daemons)

            if run_mountdir is not None:

                _wait_for_unmounted(run_mountdir)

        for d in daemons:

            _cleanup_daemon_storage(d)

            _cleanup_iteration_workspace(iter_workspace)
@@ -840,7 +861,6 @@ def test_expand_performance_multi_run(client_fixture, 
    expand_on_demand_value = "ON" if EXPAND_ON_DEMAND else "OFF"

    

    for run_idx in range(num_reps):

        time.sleep(10)  # Give the backend time to release resources from previous run

        logger.info("=" * 70)

        logger.info(f"EXPAND PERFORMANCE TEST - RUN {run_idx + 1}/{num_reps}")

        logger.info(f"  GKFS_EXPAND_ON_DEMAND={expand_on_demand_value}")
@@ -861,7 +881,6 @@ def test_expand_performance_multi_run(client_fixture, 
        for i in range(num_start):

            d = iter_daemon_factory.create(enable_forwarding=False)

            daemons.append(d)

        time.sleep(5)

        all_daemons.extend(daemons)

        

        hostfile = Path(daemons[0].hostfile)
@@ -892,7 +911,6 @@ def test_expand_performance_multi_run(client_fixture, 
            d = iter_daemon_factory.create(expand_mode=True,

                                           enable_forwarding=False)

            new_daemons.append(d)

            time.sleep(2)  # Wait for daemon to write '+' to workspace

        all_daemons.extend(new_daemons)

        logger.info(f"  Expand: {len(daemons)} active, {len(new_daemons)} adding")
@@ -931,7 +949,7 @@ def test_expand_performance_multi_run(client_fixture, 
            cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

            if "No mutate running/finished." in cmd.stderr.decode():

                break

            time.sleep(2)

            time.sleep(MUTATE_STATUS_POLL_INTERVAL)

        else:

            pytest.fail(f"Expand did not complete within 120s (run {run_idx + 1})")
@@ -943,7 +961,6 @@ def test_expand_performance_multi_run(client_fixture, 
        cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

        assert cmd.exit_code == 0

        _set_client_hostfile(client, workspace_file)

        time.sleep(2)

        

        # Verify

        accessible = count_accessible_files(created_files, client)
@@ -1025,7 +1042,6 @@ def test_mutate_performance_multi_run(client_fixture, 
    all_daemons = []

    

    for run_idx in range(num_reps):

        time.sleep(10)  # Give the backend time to release resources from previous run

        logger.info("=" * 70)

        logger.info(f"MUTATE PERFORMANCE TEST - RUN {run_idx + 1}/{num_reps}")

        logger.info("=" * 70)
@@ -1043,7 +1059,6 @@ def test_mutate_performance_multi_run(client_fixture, 
        for i in range(num_start):

            d = iter_daemon_factory.create(enable_forwarding=False)

            old_daemons.append(d)

        time.sleep(5)

        all_daemons.extend(old_daemons)

        

        hostfile = Path(old_daemons[0].hostfile)
@@ -1072,7 +1087,6 @@ def test_mutate_performance_multi_run(client_fixture, 
            d = iter_daemon_factory.create(expand_mode=True,

                                           enable_forwarding=False)

            new_daemons.append(d)

            time.sleep(1)

        all_daemons.extend(new_daemons)

        

        # For same-count mutate: all old get '-' markers, all new get '+' markers
@@ -1119,7 +1133,7 @@ def test_mutate_performance_multi_run(client_fixture, 
            cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

            if "No mutate running/finished." in cmd.stderr.decode():

                break

            time.sleep(2)

            time.sleep(MUTATE_STATUS_POLL_INTERVAL)

        

        elapsed = perf_counter() - t0

        wall = time.time() - t_wall
@@ -1129,7 +1143,6 @@ def test_mutate_performance_multi_run(client_fixture, 
        cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

        assert cmd.exit_code == 0

        _set_client_hostfile(client, workspace_file)

        time.sleep(2)

        

        # Verify

        accessible = count_accessible_files(created_files, client)
@@ -1211,7 +1224,6 @@ def test_comprehensive_cycle_performance(client_fixture, 
    all_daemons = []

    

    for rep_idx in range(num_reps):

        time.sleep(10)  # Give the backend time to release resources from previous run

        logger.info("=" * 70)

        logger.info(f"COMPREHENSIVE CYCLE TEST - REPLICATION {rep_idx + 1}/{num_reps}")

        logger.info("=" * 70)
@@ -1229,7 +1241,6 @@ def test_comprehensive_cycle_performance(client_fixture, 
        for i in range(num_initial):

            d = iter_daemon_factory.create(enable_forwarding=False)

            daemons.append(d)

        time.sleep(5)

        all_daemons.extend(daemons)

        

        hostfile = Path(daemons[0].hostfile)
@@ -1277,14 +1288,13 @@ def test_comprehensive_cycle_performance(client_fixture, 
            cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

            if "No mutate running/finished." in cmd.stderr.decode():

                break

            time.sleep(2)

            time.sleep(MUTATE_STATUS_POLL_INTERVAL)

        

        shrink_elapsed = perf_counter() - shrink_t0

        cmd_str = f'bash -c "source {env_file} && gkfs_malleability mutate finalize"'

        cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

        assert cmd.exit_code == 0

        _set_client_hostfile(client, workspace_shrink)

        time.sleep(2)

        

        post_shrink_verify = verify_files_with_md5(file_md5_map, client, run_mountdir,

                                                   max_read_checks=1)
@@ -1312,7 +1322,6 @@ def test_comprehensive_cycle_performance(client_fixture, 
            d = iter_daemon_factory.create(expand_mode=True,

                                           enable_forwarding=False)

            new_daemons.append(d)

            time.sleep(2)  # Wait for daemon to write '+' to workspace

        all_daemons.extend(new_daemons)

        

        workspace_expand = hostfile.parent / f"cycle_expand_workspace_{rep_idx}.txt"
@@ -1350,14 +1359,13 @@ def test_comprehensive_cycle_performance(client_fixture, 
            cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

            if "No mutate running/finished." in cmd.stderr.decode():

                break

            time.sleep(2)

            time.sleep(MUTATE_STATUS_POLL_INTERVAL)

        

        expand_elapsed = perf_counter() - expand_t0

        cmd_str = f'bash -c "source {env_file} && gkfs_malleability mutate finalize"'

        cmd = gkfs_shell.script(cmd_str, intercept_shell=False)

        assert cmd.exit_code == 0

        _set_client_hostfile(client, workspace_expand)

        time.sleep(2)

        

        post_expand_verify = verify_files_with_md5(file_md5_map, client, run_mountdir,

                                                   max_read_checks=1)