"Add tests for proxy and malleability to increase coverage" (!222) · Merge requests · hpc / gekkofs

src/proxy/util.cpp

+1 −1

Changes for src/proxy/util.cpp: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -83,7 +83,7 @@ load_hostfile(const std::string& lfpath) {
		uri.find("na+sm") != std::string::npos) {
		PROXY_DATA->use_auto_sm(true);
		PROXY_DATA->log()->info(
		"{}() auto_sm detected in daemon hosefile. Enabling it on proxy ...",
		"{}() auto_sm detected in daemon hostfile. Enabling it on proxy ...",
		__func__);
		}

tests/integration/data/test_truncate.py

+88 −0

Changes for tests/integration/data/test_truncate.py: 88 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -158,3 +158,91 @@ def test_fail_truncate(gkfs_daemon, gkfs_client):
		assert ret.statbuf.st_size == buf_length+1


		def test_truncate_proxy(gkfs_daemon_proxy, gkfs_proxy, gkfs_client_proxy):
		"""Testing truncate:
		1. create a large file over multiple chunks
		2. truncate it in the middle and compare it with a fresh file with equal contents (exactly at chunk border)
		3. truncate it again so that in truncates in the middle of the chunk and compare with fresh file
		TODO chunksize needs to be respected to make sure chunk border and in the middle of chunk truncates are honored
		"""
		truncfile = gkfs_daemon_proxy.mountdir / "trunc_file"

		# open and create test file
		ret = gkfs_client_proxy.open(truncfile, os.O_CREAT \| os.O_WRONLY, stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)

		assert ret.retval != -1

		# write a multi MB file (16mb)
		buf_length = 16777216
		ret = gkfs_client_proxy.write_random(truncfile, buf_length)

		assert ret.retval == buf_length

		ret = gkfs_client_proxy.stat(truncfile)

		assert ret.statbuf.st_size == buf_length

		# truncate it
		# split exactly in the middle
		trunc_size = buf_length // 2
		ret = gkfs_client_proxy.truncate(truncfile, trunc_size)

		assert ret.retval == 0
		# check file length
		ret = gkfs_client_proxy.stat(truncfile)

		assert ret.statbuf.st_size == trunc_size

		# verify contents by writing a new file (random content is seeded) and checksum both
		truncfile_verify = gkfs_daemon_proxy.mountdir / "trunc_file_verify"

		# open and create test file
		ret = gkfs_client_proxy.open(truncfile_verify, os.O_CREAT \| os.O_WRONLY, stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)

		assert ret.retval != -1

		# write trunc_size of data to new file
		ret = gkfs_client_proxy.write_random(truncfile_verify, trunc_size)

		assert ret.retval == trunc_size

		ret = gkfs_client_proxy.stat(truncfile_verify)

		assert ret.statbuf.st_size == trunc_size

		ret = gkfs_client_proxy.file_compare(truncfile, truncfile_verify, trunc_size)

		assert ret.retval == 0

		# trunc at byte 712345 (middle of chunk)
		# TODO feed chunksize into test to make sure it is always in the middle of the chunk
		trunc_size = 712345
		ret = gkfs_client_proxy.truncate(truncfile, trunc_size)

		assert ret.retval == 0

		# check file length
		ret = gkfs_client_proxy.stat(truncfile)

		assert ret.statbuf.st_size == trunc_size

		# verify contents by writing a new file (random content is seeded) and checksum both
		truncfile_verify_2 = gkfs_daemon_proxy.mountdir / "trunc_file_verify_2"

		# open and create test file
		ret = gkfs_client_proxy.open(truncfile_verify_2, os.O_CREAT \| os.O_WRONLY, stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)

		assert ret.retval != -1

		# write trunc_size of data to new file
		ret = gkfs_client_proxy.write_random(truncfile_verify_2, trunc_size)

		assert ret.retval == trunc_size

		ret = gkfs_client_proxy.stat(truncfile_verify_2)

		assert ret.statbuf.st_size == trunc_size

		ret = gkfs_client_proxy.file_compare(truncfile, truncfile_verify_2, trunc_size)

		assert ret.retval == 0
		No newline at end of file

tests/integration/directories/test_directories.py

+86 −0

Changes for tests/integration/directories/test_directories.py: 86 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -299,3 +299,89 @@ def test_opendir(gkfs_daemon, gkfs_client, directory_path):
		assert ret.dirp is None
		assert ret.errno == errno.ENOENT

		def test_finedir_proxy(gkfs_daemon_proxy, gkfs_proxy, gkfs_client_proxy):
		"""Tests several corner cases for directories scan"""

		topdir = gkfs_daemon_proxy.mountdir / "finetop"
		file_a = topdir / "file_"


		# create topdir
		ret = gkfs_client_proxy.mkdir(
		topdir,
		stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)

		assert ret.retval == 0

		ret = gkfs_client_proxy.readdir(topdir)

		# XXX: This might change in the future if we add '.' and '..'
		assert len(ret.dirents) == 0

		# populate top directory

		for files in range (1,4):
		ret = gkfs_client_proxy.directory_validate(
		topdir, 1)
		assert ret.retval == files


		ret = gkfs_client_proxy.directory_validate(
		topdir, 1000)
		assert ret.retval == 1000+3


		def test_extended_proxy(gkfs_daemon_proxy, gkfs_proxy, gkfs_shell_proxy, gkfs_client_proxy):
		topdir = gkfs_daemon_proxy.mountdir / "test_extended"
		dir_a = topdir / "dir_a"
		dir_b = topdir / "dir_b"
		file_a = topdir / "file_a"
		subdir_a = dir_a / "subdir_a"

		# create topdir
		ret = gkfs_client_proxy.mkdir(
		topdir,
		stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)

		assert ret.retval == 0

		ret = gkfs_client_proxy.mkdir(
		dir_a,
		stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)

		assert ret.retval == 0

		ret = gkfs_client_proxy.mkdir(
		dir_b,
		stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)

		assert ret.retval == 0

		ret = gkfs_client_proxy.mkdir(
		subdir_a,
		stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)

		assert ret.retval == 0

		ret = gkfs_client_proxy.open(file_a,
		os.O_CREAT,
		stat.S_IRWXU \| stat.S_IRWXG \| stat.S_IRWXO)
		assert ret.retval != -1

		buf = b'42'
		ret = gkfs_client_proxy.write(file_a, buf, 1)

		assert ret.retval == 1

		cmd = gkfs_shell_proxy.sfind(
		topdir,
		'-M',
		gkfs_daemon_proxy.mountdir,
		'-S',
		1,
		'-name',
		'_k'
		)

		assert cmd.exit_code == 0
		assert cmd.stdout.decode() == "MATCHED 0/4\n"
		No newline at end of file

tests/integration/forwarding/test_map.py

+8 −0

Changes for tests/integration/forwarding/test_map.py: 8 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -138,6 +138,10 @@ def test_two_io_nodes(gkfwd_daemon_factory, gkfwd_client_factory):

		assert ion == '1'

		d00.shutdown()
		d01.shutdown()


		def test_two_io_nodes_remap(gkfwd_daemon_factory, gkfwd_client_factory):
		"""Write files from two clients using two daemons"""

		@@ -201,6 +205,8 @@ def test_two_io_nodes_remap(gkfwd_daemon_factory, gkfwd_client_factory):
		ion = line.split()[-1]

		assert ion == '1'
		d00.shutdown()
		d01.shutdown()

		def test_two_io_nodes_operations(gkfwd_daemon_factory, gkfwd_client_factory):
		"""Write files from one client and read in the other using two daemons"""
		@@ -285,3 +291,5 @@ def test_two_io_nodes_operations(gkfwd_daemon_factory, gkfwd_client_factory):
		ion = line.split()[-1]

		assert ion == '1'
		d00.shutdown()
		d01.shutdown()
		No newline at end of file

tests/integration/harness/gkfs.py

+156 −4

Changes for tests/integration/harness/gkfs.py: 156 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -48,6 +48,12 @@ gkfs_client_log_level = 'all'
		gkfs_client_log_syscall_filter = 'epoll_wait,epoll_create'
		gkfs_daemon_active_log_pattern = r'Startup successful. Daemon is ready.'

		gkfs_proxy_cmd = 'gkfs_proxy'
		gkfs_proxy_pid = 'gkfs_proxy.pid'
		gkfs_proxy_log_file = 'gkfs_proxy.log'
		gkfs_proxy_log_level = '100'
		gkfs_proxy_active_log_pattern = r'Initializing environment'

		gkfwd_daemon_cmd = 'gkfs_daemon'
		gkfwd_client_cmd = 'gkfs.io'
		gkfwd_client_lib_file = 'libgkfs_intercept.so'
		@@ -225,7 +231,7 @@ class FwdClientCreator:


		class Daemon:
		def __init__(self, interface, database, workspace):
		def __init__(self, interface, database, workspace, proxy = False):

		self._address = get_ephemeral_address(interface)
		self._workspace = workspace
		@@ -233,6 +239,8 @@ class Daemon:
		self._cmd = sh.Command(gkfs_daemon_cmd, self._workspace.bindirs)
		self._env = os.environ.copy()
		self._metadir = self.rootdir
		self._proxy = proxy

		libdirs = ':'.join(
		filter(None, [os.environ.get('LD_LIBRARY_PATH', '')] +
		[str(p) for p in self._workspace.libdirs]))
		@@ -257,6 +265,9 @@ class Daemon:
		'--enable-chunkstats']
		if self._database == "parallaxdb" :
		args.append('--clean-rootdir-finish')
		if self._proxy == True:
		args.append('--proxy-protocol')
		args.append('ofi+sockets')

		logger.debug(f"spawning daemon")
		logger.debug(f"cmdline: {self._cmd} " + " ".join(map(str, args)))
		@@ -367,6 +378,136 @@ class Daemon:
		def interface(self):
		return self._interface

		class Proxy:
		def __init__(self, workspace):
		self._parser = IOParser()
		self._workspace = workspace
		self._cmd = sh.Command(gkfs_proxy_cmd, self._workspace.bindirs)
		self._env = os.environ.copy()

		libdirs = ':'.join(
		filter(None, [os.environ.get('LD_LIBRARY_PATH', '')] +
		[str(p) for p in self._workspace.libdirs]))

		self._patched_env = {
		'LD_LIBRARY_PATH' : libdirs,
		'GKFS_PROXY_LOG_PATH' : str(self.logdir / gkfs_proxy_log_file),
		'GKFS_PROXY_LOG_LEVEL': gkfs_proxy_log_level
		}
		self._env.update(self._patched_env)


		def run(self):

		args = ['-H', str(self.cwd / gkfs_hosts_file),
		'--pid-path', str(self.cwd / gkfs_proxy_pid),
		'-p', 'ofi+sockets']


		logger.debug(f"spawning proxy")
		logger.debug(f"cmdline: {self._cmd} " + " ".join(map(str, args)))
		logger.debug(f"patched env:\n{pformat(self._patched_env)}")

		self._proc = self._cmd(
		args,
		_env=self._env,
		_out=sys.stdout,
		_err=sys.stderr,
		_bg=True,
		)

		logger.debug(f"proxy process spawned (PID={self._proc.pid})")
		logger.debug("waiting for proxy to be ready")

		try:
		self.wait_until_active(self._proc.pid, 720.0)
		except Exception as ex:
		logger.error(f"proxy initialization failed: {ex}")

		# if the daemon initialized correctly but took longer than
		# `timeout`, we may be leaving running processes behind
		if _process_exists(self._proc.pid):
		self.shutdown()

		logger.critical(f"proxy was shut down, what is ex? {ex.__repr__}?")
		raise ex

		logger.debug("proxy is ready")
		return self



		def wait_until_active(self, pid, timeout, max_lines=50):
		"""
		Waits until a GKFS daemon is active or until a certain timeout
		has expired. Checks if the daemon is running by searching its
		log for a pre-defined readiness message.

		Parameters
		----------
		pid: `int`
		The PID of the daemon process to wait for.

		timeout: `number`
		The number of seconds to wait for

		max_lines: `int`
		The maximum number of log lines to check for a match.
		"""



		init_time = perf_counter()

		while perf_counter() - init_time < timeout:
		try:
		# logger.debug(f"checking log file")
		with open(self.logdir / gkfs_proxy_log_file) as log:
		for line in islice(log, max_lines):
		if re.search(gkfs_proxy_active_log_pattern, line) is not None:
		return
		except FileNotFoundError:
		# Log is missing, the daemon might have crashed...
		logger.debug(f"proxy log file missing, checking if daemon is alive...")

		pid=self._proc.pid

		if not _process_exists(pid):
		raise RuntimeError(f"process {pid} is not running")

		# ... or it might just be lazy. let's give it some more time
		logger.debug(f"proxy {pid} found, retrying...")
		time.sleep(1)
		raise RuntimeError("initialization timeout exceeded")

		def shutdown(self):
		logger.debug(f"terminating proxy")
		try:
		self._proc.terminate()
		err = self._proc.wait()
		except sh.SignalException_SIGTERM:
		pass
		except Exception:
		raise


		@property
		def cwd(self):
		return self._workspace.twd

		@property
		def rootdir(self):
		return self._workspace.rootdir

		@property
		def logdir(self):
		return self._workspace.logdir

		@property
		def interface(self):
		return self._interface


		class _proxy_exec():
		def __init__(self, client, name):
		self._client = client
		@@ -383,11 +524,12 @@ class Client:
		function calls, be them system calls (e.g. read()) or glibc I/O functions
		(e.g. opendir()).
		"""
		def __init__(self, workspace):
		def __init__(self, workspace, proxy = False):
		self._parser = IOParser()
		self._workspace = workspace
		self._cmd = sh.Command(gkfs_client_cmd, self._workspace.bindirs)
		self._env = os.environ.copy()
		self._proxy = proxy

		libdirs = ':'.join(
		filter(None, [os.environ.get('LD_LIBRARY_PATH', '')] +
		@@ -424,6 +566,9 @@ class Client:
		'LIBGKFS_LOG_SYSCALL_FILTER': gkfs_client_log_syscall_filter
		}

		if (self._proxy == True):
		self._patched_env['LIBGKFS_PROXY_PID_FILE'] = str(self.cwd / gkfs_proxy_pid)

		self._env.update(self._patched_env)

		@property
		@@ -489,10 +634,11 @@ class ShellClient:
		on a GekkoFS instance.
		"""

		def __init__(self, workspace):
		def __init__(self, workspace, proxy = False):
		self._workspace = workspace
		self._search_paths = _find_search_paths(self._workspace.bindirs)
		self._env = os.environ.copy()
		self._proxy = proxy

		libdirs = ':'.join(
		filter(None, [os.environ.get('LD_LIBRARY_PATH', '')] +
		@@ -525,6 +671,9 @@ class ShellClient:
		'LIBGKFS_LOG_SYSCALL_FILTER': gkfs_client_log_syscall_filter
		}

		if (self._proxy == True):
		self._patched_env['LIBGKFS_PROXY_PID_FILE'] = str(self.cwd / gkfs_proxy_pid)

		self._env.update(self._patched_env)

		@property
		@@ -713,7 +862,7 @@ class FwdDaemon:

		self._address = get_ephemeral_address(interface)
		self._workspace = workspace

		self._hostfile = str(self.cwd / gkfwd_hosts_file)
		self._cmd = sh.Command(gkfwd_daemon_cmd, self._workspace.bindirs)
		self._env = os.environ.copy()

		@@ -848,6 +997,9 @@ class FwdDaemon:
		def interface(self):
		return self._interface

		@property
		def hostfile(self):
		return self._hostfile

		class FwdClient:
		"""

"Add tests for proxy and malleability to increase coverage"

Merge request reports