diff --git a/src/client/dfs/common.c b/src/client/dfs/common.c
index c66de734411..8f5b86fb4d6 100644
--- a/src/client/dfs/common.c
+++ b/src/client/dfs/common.c
@@ -629,8 +629,8 @@ entry_stat(dfs_t *dfs, daos_handle_t th, daos_handle_t oh, const char *name, siz
 }
 
 /*
- * create a dir object. If caller passes parent obj, we check for existence of
- * object first.
+ * Create a dir object. If caller passes parent obj, and cid is not set,
+ * the child oclass is taken from the parent.
  */
 int
 create_dir(dfs_t *dfs, dfs_obj_t *parent, daos_oclass_id_t cid, dfs_obj_t *dir)
diff --git a/src/client/dfs/dfs_sys.c b/src/client/dfs/dfs_sys.c
index 9dcd7799382..32f2217f2ac 100644
--- a/src/client/dfs/dfs_sys.c
+++ b/src/client/dfs/dfs_sys.c
@@ -1399,6 +1399,74 @@ dfs_sys_mkdir(dfs_sys_t *dfs_sys, const char *dir, mode_t mode,
 	return rc;
 }
 
+static int
+check_existing_dir(dfs_sys_t *dfs_sys, const char *dir_path)
+{
+	struct stat st = {0};
+	int         rc;
+
+	rc = dfs_sys_stat(dfs_sys, dir_path, 0, &st);
+	if (rc != 0) {
+		D_DEBUG(DB_TRACE, "failed to stat %s: (%d)\n", dir_path, rc);
+		return rc;
+	}
+
+	/* if it's not a directory, fail */
+	if (!S_ISDIR(st.st_mode))
+		return EEXIST;
+
+	/* if it is a directory, then it's not an error */
+	return 0;
+}
+
+int
+dfs_sys_mkdir_p(dfs_sys_t *dfs_sys, const char *dir_path, mode_t mode, daos_oclass_id_t cid)
+{
+	int   path_len = strnlen(dir_path, PATH_MAX);
+	char *_path    = NULL;
+	char *ptr      = NULL;
+	int   rc       = 0;
+
+	if (dfs_sys == NULL)
+		return EINVAL;
+	if (dir_path == NULL)
+		return EINVAL;
+	if (path_len == PATH_MAX)
+		return ENAMETOOLONG;
+
+	D_STRNDUP(_path, dir_path, path_len);
+	if (_path == NULL)
+		return ENOMEM;
+
+	/* iterate through the parent directories and create them if necessary */
+	for (ptr = _path + 1; *ptr != '\0'; ptr++) {
+		if (*ptr != '/')
+			continue;
+
+		/* truncate the string here to create the parent */
+		*ptr = '\0';
+		rc   = dfs_sys_mkdir(dfs_sys, _path, mode, cid);
+		if (rc != 0) {
+			if (rc != EEXIST)
+				D_GOTO(out_free, rc);
+			rc = check_existing_dir(dfs_sys, _path);
+			if (rc != 0)
+				D_GOTO(out_free, rc);
+		}
+		/* reset to keep going */
+		*ptr = '/';
+	}
+
+	/* create the final directory */
+	rc = dfs_sys_mkdir(dfs_sys, _path, mode, cid);
+	if (rc == EEXIST)
+		rc = check_existing_dir(dfs_sys, _path);
+
+out_free:
+	D_FREE(_path);
+	return rc;
+}
+
 int
 dfs_sys_opendir(dfs_sys_t *dfs_sys, const char *dir, int flags, DIR **_dirp)
 {
diff --git a/src/include/daos_fs_sys.h b/src/include/daos_fs_sys.h
index a1fc6fd0f61..e60dd00a5e2 100644
--- a/src/include/daos_fs_sys.h
+++ b/src/include/daos_fs_sys.h
@@ -542,6 +542,19 @@ int
 dfs_sys_mkdir(dfs_sys_t *dfs_sys, const char *dir, mode_t mode,
 	      daos_oclass_id_t cid);
 
+/**
+ * Create a directory and all of its parent directories.
+ *
+ * \param[in]	dfs_sys  Pointer to the mounted file system.
+ * \param[in]	dir_path Link path of new dir.
+ * \param[in]	mode	 mkdir mode.
+ * \param[in]	cid	 DAOS object class id (pass 0 for default MAX_RW).
+ *
+ * \return		 0 on success, errno code on failure.
+ */
+int
+dfs_sys_mkdir_p(dfs_sys_t *dfs_sys, const char *dir_path, mode_t mode, daos_oclass_id_t cid);
+
 /**
  * Open a directory.
  * The directory must be closed with dfs_sys_closedir().
diff --git a/src/tests/suite/dfs_sys_unit_test.c b/src/tests/suite/dfs_sys_unit_test.c
index 3b296b8977c..e9228567e52 100644
--- a/src/tests/suite/dfs_sys_unit_test.c
+++ b/src/tests/suite/dfs_sys_unit_test.c
@@ -1,5 +1,5 @@
 /**
- * (C) Copyright 2019-2022 Intel Corporation.
+ * (C) Copyright 2019-2024 Intel Corporation.
  *
  * SPDX-License-Identifier: BSD-2-Clause-Patent
  */
@@ -816,31 +816,100 @@ dfs_sys_test_chown(void **state)
 	assert_int_equal(rc, 0);
 }
 
+static void
+dfs_sys_test_mkdir(void **state)
+{
+	test_arg_t *arg    = *state;
+	const char *parent = "/a";
+	const char *child  = "/a/b";
+	const char *file   = "/a/b/whoops";
+	dfs_obj_t  *obj;
+	int         rc;
+
+	if (arg->myrank != 0)
+		return;
+
+	/* create the parent */
+	rc = dfs_sys_mkdir(dfs_sys_mt, parent, S_IWUSR | S_IRUSR, 0);
+	assert_int_equal(rc, 0);
+
+	/* trying to create the parent again should fail */
+	rc = dfs_sys_mkdir(dfs_sys_mt, parent, S_IWUSR | S_IRUSR, 0);
+	assert_int_equal(rc, EEXIST);
+
+	rc = dfs_sys_mkdir(dfs_sys_mt, child, S_IWUSR | S_IRUSR, 0);
+	assert_int_equal(rc, 0);
+
+	rc = dfs_sys_open(dfs_sys_mt, file, S_IFREG, O_CREAT | O_RDWR, 0, 0, NULL, &obj);
+	assert_int_equal(rc, 0);
+	dfs_sys_close(obj);
+
+	/* this shouldn't work */
+	rc = dfs_sys_mkdir(dfs_sys_mt, file, S_IWUSR | S_IRUSR, 0);
+	assert_int_equal(rc, EEXIST);
+
+	rc = dfs_sys_remove(dfs_sys_mt, parent, true, NULL);
+	assert_int_equal(rc, 0);
+}
+
+static void
+dfs_sys_test_mkdir_p(void **state)
+{
+	test_arg_t *arg    = *state;
+	const char *parent = "/a";
+	const char *child  = "/a/b";
+	const char *file   = "/a/b/whoops";
+	dfs_obj_t  *obj;
+	int         rc;
+
+	if (arg->myrank != 0)
+		return;
+
+	/* create the child and its parents */
+	rc = dfs_sys_mkdir_p(dfs_sys_mt, child, S_IWUSR | S_IRUSR, 0);
+	assert_int_equal(rc, 0);
+
+	/* creating the parent shouldn't fail even though it exists */
+	rc = dfs_sys_mkdir_p(dfs_sys_mt, parent, S_IWUSR | S_IRUSR, 0);
+	assert_int_equal(rc, 0);
+
+	rc = dfs_sys_open(dfs_sys_mt, file, S_IFREG, O_CREAT | O_RDWR, 0, 0, NULL, &obj);
+	assert_int_equal(rc, 0);
+	dfs_sys_close(obj);
+
+	/* this shouldn't work */
+	rc = dfs_sys_mkdir_p(dfs_sys_mt, file, S_IWUSR | S_IRUSR, 0);
+	assert_int_equal(rc, EEXIST);
+
+	rc = dfs_sys_remove(dfs_sys_mt, parent, true, NULL);
+	assert_int_equal(rc, 0);
+}
+
 static const struct CMUnitTest dfs_sys_unit_tests[] = {
-	{ "DFS_SYS_UNIT_TEST1:  DFS Sys mount / umount",
-	  dfs_sys_test_mount, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST2:  DFS Sys2base",
-	  dfs_sys_test_sys2base, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST3:  DFS Sys create / remove",
-	  dfs_sys_test_create_remove, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST4:  DFS Sys access / chmod",
-	  dfs_sys_test_access_chmod, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST5:  DFS Sys open / stat",
-	  dfs_sys_test_open_stat, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST6:  DFS Sys readlink",
-	  dfs_sys_test_readlink, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST7:  DFS Sys setattr",
-	  dfs_sys_test_setattr, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST8:  DFS Sys read / write",
-	  dfs_sys_test_read_write, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST9:  DFS Sys opendir / readdir",
-	  dfs_sys_test_open_readdir, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST10: DFS Sys xattr",
-	  dfs_sys_test_xattr, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST11: DFS Sys l2g/g2l handles",
-	  dfs_sys_test_handles, async_disable, test_case_teardown},
-	{ "DFS_SYS_UNIT_TEST12: DFS Sys chown",
-	  dfs_sys_test_chown, async_disable, test_case_teardown},
+    {"DFS_SYS_UNIT_TEST1:  DFS Sys mount / umount", dfs_sys_test_mount, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST2:  DFS Sys2base", dfs_sys_test_sys2base, async_disable, test_case_teardown},
+    {"DFS_SYS_UNIT_TEST3:  DFS Sys create / remove", dfs_sys_test_create_remove, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST4:  DFS Sys access / chmod", dfs_sys_test_access_chmod, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST5:  DFS Sys open / stat", dfs_sys_test_open_stat, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST6:  DFS Sys readlink", dfs_sys_test_readlink, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST7:  DFS Sys setattr", dfs_sys_test_setattr, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST8:  DFS Sys read / write", dfs_sys_test_read_write, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST9:  DFS Sys opendir / readdir", dfs_sys_test_open_readdir, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST10: DFS Sys xattr", dfs_sys_test_xattr, async_disable, test_case_teardown},
+    {"DFS_SYS_UNIT_TEST11: DFS Sys l2g/g2l handles", dfs_sys_test_handles, async_disable,
+     test_case_teardown},
+    {"DFS_SYS_UNIT_TEST12: DFS Sys chown", dfs_sys_test_chown, async_disable, test_case_teardown},
+    {"DFS_SYS_UNIT_TEST13: DFS Sys mkdir", dfs_sys_test_mkdir, async_disable},
+    {"DFS_SYS_UNIT_TEST14: DFS Sys mkdir_p", dfs_sys_test_mkdir_p, async_disable,
+     test_case_teardown},
 };
 
 static int