Repository: ceph/go-ceph
Pull request #519 (2021-06-21 18:35:16)

cephfs admin: add initial support for cephfs mirroring
Add functions to configure and inspect cephfs mirroring to cephfs admin package.

The first few patches add a few adjustments to non-API code in preparation for testing cephfs mirroring code.

Support for setting up the cephfs mirroring daemon is added to micro-osd. This is disabled on nautilus and octopus.

Functions are added that configure needed ceph mgr modules.

Functions to set up cephfs mirroring and mirror directories are added.

Two functions for inspecting the state of mirroring are added. I know of one additional function that I did not expose. The json returned by that function is a bit more complex and I thought the PR was already too big. I will look into it for a follow up PR.

Basic tests for the JSON parsing, as well as a large "end to end" test that actually sets up mirroring and verifies directories get mirroring are added.

Fixes:  #444

## Checklist
- [x] Added tests for features and functional changes
- [x] Public functions and types are documented
- [x] Standard formatting is applied to Go code


phlogistonjohn: Looks like ceph has changed the output of some of the json commands (again). Need to update to match. I'm not going to support the "old" way unless someone asks for it.
phlogistonjohn: @mergifyio rebase
mergify[bot]: **Command `rebase`: success**
> **Branch has been successfully rebased**
> 

phlogistonjohn: @mergifyio rebase
mergify[bot]: **Command `rebase`: success**
> **Branch has been successfully rebased**
> 

phlogistonjohn: @mergifyio rebase
mergify[bot]: **Command `rebase`: success**
> **Branch has been successfully rebased**
> 

phlogistonjohn: @mergifyio rebase

mergify[bot]: **Command `rebase`: success**
> **Branch has been successfully rebased**
> 


Patch:
diff --git a/cephfs/admin/mgrmodule.go b/cephfs/admin/mgrmodule.go
new file mode 100644
index 0000000..50a9157
--- /dev/null
+++ b/cephfs/admin/mgrmodule.go
@@ -0,0 +1,54 @@
+package admin
+
+import (
+	"github.com/ceph/go-ceph/internal/commands"
+)
+
+const mirroring = "mirroring"
+
+// EnableModule will enable the specified manager module.
+//
+// Similar To:
+//  ceph mgr module enable <module> [--force]
+func (fsa *FSAdmin) EnableModule(module string, force bool) error {
+	m := map[string]string{
+		"prefix": "mgr module enable",
+		"module": module,
+		"format": "json",
+	}
+	if force {
+		m["force"] = "--force"
+	}
+	// Why is this _only_ part of the mon command json? You'd think a mgr
+	// command would be available as a MgrCommand but I couldn't figure it out.
+	return commands.MarshalMonCommand(fsa.conn, m).NoData().End()
+}
+
+// DisableModule will disable the specified manager module.
+//
+// Similar To:
+//  ceph mgr module disable <module>
+func (fsa *FSAdmin) DisableModule(module string) error {
+	m := map[string]string{
+		"prefix": "mgr module disable",
+		"module": module,
+		"format": "json",
+	}
+	return commands.MarshalMonCommand(fsa.conn, m).NoData().End()
+}
+
+// EnableMirroringModule will enable the mirroring module for cephfs.
+//
+// Similar To:
+//  ceph mgr module enable mirroring [--force]
+func (fsa *FSAdmin) EnableMirroringModule(force bool) error {
+	return fsa.EnableModule(mirroring, force)
+}
+
+// DisableMirroringModule will disable the mirroring module for cephfs.
+//
+// Similar To:
+//  ceph mgr module disable mirroring
+func (fsa *FSAdmin) DisableMirroringModule() error {
+	return fsa.DisableModule(mirroring)
+}
diff --git a/cephfs/admin/mirror.go b/cephfs/admin/mirror.go
new file mode 100644
index 0000000..b1627ac
--- /dev/null
+++ b/cephfs/admin/mirror.go
@@ -0,0 +1,234 @@
+package admin
+
+import (
+	ccom "github.com/ceph/go-ceph/common/commands"
+	"github.com/ceph/go-ceph/internal/commands"
+)
+
+// SnapshotMirrorAdmin helps administer the snapshot mirroring features of
+// cephfs. Snapshot mirroring is only available in ceph pacific and later.
+type SnapshotMirrorAdmin struct {
+	conn ccom.MgrCommander
+}
+
+// SnapshotMirror returns a new SnapshotMirrorAdmin to be used for the
+// administration of snapshot mirroring features.
+func (fsa *FSAdmin) SnapshotMirror() *SnapshotMirrorAdmin {
+	return &SnapshotMirrorAdmin{conn: fsa.conn}
+}
+
+// Enable snapshot mirroring for the given file system.
+//
+// Similar To:
+//  ceph fs snapshot mirror enable <fs_name>
+func (sma *SnapshotMirrorAdmin) Enable(fsname string) error {
+	m := map[string]string{
+		"prefix":  "fs snapshot mirror enable",
+		"fs_name": fsname,
+		"format":  "json",
+	}
+	return commands.MarshalMgrCommand(sma.conn, m).NoStatus().EmptyBody().End()
+}
+
+// Disable snapshot mirroring for the given file system.
+//
+// Similar To:
+//  ceph fs snapshot mirror disable <fs_name>
+func (sma *SnapshotMirrorAdmin) Disable(fsname string) error {
+	m := map[string]string{
+		"prefix":  "fs snapshot mirror disable",
+		"fs_name": fsname,
+		"format":  "json",
+	}
+	return commands.MarshalMgrCommand(sma.conn, m).NoStatus().EmptyBody().End()
+}
+
+// Add a path in the file system to be mirrored.
+//
+// Similar To:
+//  ceph fs snapshot mirror add <fs_name> <path>
+func (sma *SnapshotMirrorAdmin) Add(fsname, path string) error {
+	m := map[string]string{
+		"prefix":  "fs snapshot mirror add",
+		"fs_name": fsname,
+		"path":    path,
+		"format":  "json",
+	}
+	return commands.MarshalMgrCommand(sma.conn, m).NoStatus().EmptyBody().End()
+}
+
+// Remove a path in the file system from mirroring.
+//
+// Similar To:
+//  ceph fs snapshot mirror remove <fs_name> <path>
+func (sma *SnapshotMirrorAdmin) Remove(fsname, path string) error {
+	m := map[string]string{
+		"prefix":  "fs snapshot mirror remove",
+		"fs_name": fsname,
+		"path":    path,
+		"format":  "json",
+	}
+	return commands.MarshalMgrCommand(sma.conn, m).NoStatus().EmptyBody().End()
+}
+
+type bootstrapTokenResponse struct {
+	Token string `json:"token"`
+}
+
+// CreatePeerBootstrapToken returns a token that can be used to create
+// a peering association between this site an another site.
+//
+// Similar To:
+//  ceph fs snapshot mirror peer_bootstrap create <fs_name> <client_entity> <site-name>
+func (sma *SnapshotMirrorAdmin) CreatePeerBootstrapToken(
+	fsname, client, site string) (string, error) {
+	m := map[string]string{
+		"prefix":      "fs snapshot mirror peer_bootstrap create",
+		"fs_name":     fsname,
+		"client_name": client,
+		"format":      "json",
+	}
+	if site != "" {
+		m["site_name"] = site
+	}
+	var bt bootstrapTokenResponse
+	err := commands.MarshalMgrCommand(sma.conn, m).NoStatus().Unmarshal(&bt).End()
+	return bt.Token, err
+}
+
+// ImportPeerBoostrapToken creates an association between another site, one
+// that has provided a token, with the current site.
+//
+// Similar To:
+//  ceph fs snapshot mirror peer_bootstrap import <fs_name> <token>
+func (sma *SnapshotMirrorAdmin) ImportPeerBoostrapToken(fsname, token string) error {
+	m := map[string]string{
+		"prefix":  "fs snapshot mirror peer_bootstrap import",
+		"fs_name": fsname,
+		"token":   token,
+		"format":  "json",
+	}
+	return commands.MarshalMgrCommand(sma.conn, m).NoStatus().EmptyBody().End()
+}
+
+// DaemonID represents the ID of a cephfs mirroring daemon.
+type DaemonID uint
+
+// FileSystemID represents the ID of a cephfs file system.
+type FileSystemID uint
+
+// PeerUUID represents the UUID of a cephfs mirroring peer.
+type PeerUUID string
+
+// DaemonStatusPeerRemote contains fields detailing a remote peer.
+type DaemonStatusPeerRemote struct {
+	ClientName  string `json:"client_name"`
+	ClusterName string `json:"cluster_name"`
+	FSName      string `json:"fs_name"`
+}
+
+// DaemonStatusPeerStats contains fields detailing the a remote peer's stats.
+type DaemonStatusPeerStats struct {
+	FailureCount  uint64 `json:"failure_count"`
+	RecoveryCount uint64 `json:"recovery_count"`
+}
+
+// DaemonStatusPeerInfo contains fields representing information about a remote peer.
+type DaemonStatusPeerInfo struct {
+	UUID   PeerUUID               `json:"uuid"`
+	Remote DaemonStatusPeerRemote `json:"remote"`
+	Stats  DaemonStatusPeerStats  `json:"stats"`
+}
+
+// DaemonStatusFileSystemInfo represents information about a mirrored file system.
+type DaemonStatusFileSystemInfo struct {
+	FileSystemID   FileSystemID           `json:"filesystem_id"`
+	Name           string                 `json:"name"`
+	DirectoryCount int64                  `json:"directory_count"`
+	Peers          []DaemonStatusPeerInfo `json:"peers"`
+}
+
+// DaemonStatusInfo maps file system IDs to information about that file system.
+type DaemonStatusInfo struct {
+	DaemonID    DaemonID                     `json:"daemon_id"`
+	FileSystems []DaemonStatusFileSystemInfo `json:"filesystems"`
+}
+
+// DaemonStatusResults maps mirroring daemon IDs to information about that
+// mirroring daemon.
+type DaemonStatusResults []DaemonStatusInfo
+
+func parseDaemonStatus(res response) (DaemonStatusResults, error) {
+	var dsr DaemonStatusResults
+	if err := res.NoStatus().Unmarshal(&dsr).End(); err != nil {
+		return nil, err
+	}
+	return dsr, nil
+}
+
+// DaemonStatus returns information on the status of cephfs mirroring daemons
+// associated with the given file system.
+//
+// Similar To:
+//  ceph fs snapshot mirror daemon status <fs_name>
+func (sma *SnapshotMirrorAdmin) DaemonStatus(fsname string) (
+	DaemonStatusResults, error) {
+	// ---
+	m := map[string]string{
+		"prefix":  "fs snapshot mirror daemon status",
+		"fs_name": fsname,
+		"format":  "json",
+	}
+	return parseDaemonStatus(commands.MarshalMgrCommand(sma.conn, m))
+}
+
+// PeerInfo includes information about a cephfs mirroring peer.
+type PeerInfo struct {
+	ClientName string `json:"client_name"`
+	SiteName   string `json:"site_name"`
+	FSName     string `json:"fs_name"`
+	MonHost    string `json:"mon_host"`
+}
+
+// PeerListResults maps a peer's UUID to information about that peer.
+type PeerListResults map[PeerUUID]PeerInfo
+
+func parsePeerList(res response) (PeerListResults, error) {
+	var plr PeerListResults
+	if err := res.NoStatus().Unmarshal(&plr).End(); err != nil {
+		return nil, err
+	}
+	return plr, nil
+}
+
+// PeerList returns information about peers associated with the given file system.
+//
+// Similar To:
+//  ceph fs snapshot mirror peer_list <fs_name>
+func (sma *SnapshotMirrorAdmin) PeerList(fsname string) (
+	PeerListResults, error) {
+	// ---
+	m := map[string]string{
+		"prefix":  "fs snapshot mirror peer_list",
+		"fs_name": fsname,
+		"format":  "json",
+	}
+	return parsePeerList(commands.MarshalMgrCommand(sma.conn, m))
+}
+
+/*
+ DirMap - figure out what last_shuffled is supposed to mean and, if it is a time
+          like it seems to be, how best to represent in Go.
+
+ DirMap TODO
+ceph fs snapshot mirror dirmap
+func (sma *SnapshotMirrorAdmin) DirMap(fsname, path string) error {
+	m := map[string]string{
+		"prefix":  "fs snapshot mirror dirmap",
+		"fs_name": fsname,
+		"path":    path,
+		"format":  "json",
+	}
+	return commands.MarshalMgrCommand(sma.conn, m).NoStatus().EmptyBody().End()
+}
+*/
diff --git a/internal/commands/response.go b/internal/commands/response.go
index 86a56d3..cafd4ba 100644
--- a/internal/commands/response.go
+++ b/internal/commands/response.go
@@ -110,6 +110,23 @@ func (r Response) NoBody() Response {
 	return r
 }
 
+// EmptyBody is similar to NoBody but also accepts an empty JSON object.
+func (r Response) EmptyBody() Response {
+	if !r.Ok() {
+		return r
+	}
+	if len(r.body) != 0 {
+		d := map[string]interface{}{}
+		if err := json.Unmarshal(r.body, &d); err != nil {
+			return Response{r.body, r.status, err}
+		}
+		if len(d) != 0 {
+			return Response{r.body, r.status, ErrBodyNotEmpty}
+		}
+	}
+	return r
+}
+
 // NoData asserts that the input response has no status or body values.
 func (r Response) NoData() Response {
 	return r.NoStatus().NoBody()
diff --git a/micro-osd.sh b/micro-osd.sh
index cc683d9..832c9e6 100755
--- a/micro-osd.sh
+++ b/micro-osd.sh
@@ -114,6 +114,27 @@ ceph-mgr --id ${MGR_NAME}
 ceph auth get-or-create client.rbd-mirror.${MIRROR_ID} mon 'profile rbd-mirror' osd 'profile rbd'
 rbd-mirror --id ${MIRROR_ID} --log-file ${LOG_DIR}/rbd-mirror.log
 
+# start cephfs-mirror
+# Skip on "nautilus" and "octopus" the supported ceph versions that
+# don't have it.
+case "${CEPH_VERSION}" in
+    nautilus|octopus)
+        echo "Skipping cephfs-mirror on ${CEPH_VERSION} ..."
+    ;;
+    *)
+        ceph auth get-or-create "client.cephfs-mirror.${MIRROR_ID}" \
+            mon 'profile cephfs-mirror' \
+            mds 'allow r' \
+            osd 'allow rw tag cephfs metadata=*, allow r tag cephfs data=*' \
+            mgr 'allow r'
+        cephfs-mirror --id "cephfs-mirror.${MIRROR_ID}" \
+            --log-file "${LOG_DIR}/cephfs-mirror.log"
+        ceph fs authorize cephfs client.cephfs-mirror-remote / rwps > "${DIR}/cephfs-mirror-remote.out"
+        # the .out file above is not used by the scripts but can be used for debugging
+    ;;
+esac
+
+
 # start an rgw
 ceph auth get-or-create client.rgw."${RGW_ID}" osd 'allow rwx' mon 'allow rw' -o ${RGW_DATA}/keyring
 radosgw -n client.rgw."${RGW_ID}" -k ${RGW_DATA}/keyring


Test patch:
diff --git a/cephfs/admin/fsadmin_test.go b/cephfs/admin/fsadmin_test.go
index ef0f466..63dc5e7 100644
--- a/cephfs/admin/fsadmin_test.go
+++ b/cephfs/admin/fsadmin_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/ceph/go-ceph/internal/commands"
+	"github.com/ceph/go-ceph/rados"
 )
 
 var (
@@ -61,20 +62,34 @@ func getFSAdmin(t *testing.T) *FSAdmin {
 	if cachedFSAdmin != nil {
 		return cachedFSAdmin
 	}
-	fsa, err := New()
+	cachedFSAdmin = newFSAdmin(t, "")
+	return cachedFSAdmin
+}
+
+func newFSAdmin(t *testing.T, configFile string) *FSAdmin {
+	conn, err := rados.NewConn()
+	require.NoError(t, err)
+	if configFile == "" {
+		err = conn.ReadDefaultConfigFile()
+		require.NoError(t, err)
+	} else {
+		err = conn.ReadConfigFile(configFile)
+		require.NoError(t, err)
+	}
+	err = conn.Connect()
 	require.NoError(t, err)
-	require.NotNil(t, fsa)
-	// We steal the connection set up by the New() method and wrap it in an
-	// optional tracer.
-	c := fsa.conn
+
+	var cmdr RadosCommander = conn
 	if debugTrace {
-		c = commands.NewTraceCommander(c)
+		// We wrap the "real" connection, which meets the RadosCommander interface,
+		// with a trace commander when debugTrace is set.
+		cmdr = commands.NewTraceCommander(conn)
 	}
-	cachedFSAdmin = NewFromConn(c)
+
 	// We sleep briefly before returning in order to ensure we have a mgr map
 	// before we start executing the tests.
 	time.Sleep(50 * time.Millisecond)
-	return cachedFSAdmin
+	return NewFromConn(cmdr)
 }
 
 func TestInvalidFSAdmin(t *testing.T) {
diff --git a/cephfs/admin/mirror_test.go b/cephfs/admin/mirror_test.go
new file mode 100644
index 0000000..cf46150
--- /dev/null
+++ b/cephfs/admin/mirror_test.go
@@ -0,0 +1,168 @@
+package admin
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// sampleOldDaemonStatus1 and sampleOldDaemonStatus2 are examples of the json
+// returned for daemon status before Ceph v16.2.5. It is not used by the tests
+// as the code now only supports the format of v16.2.5 and later. This is
+// retained for reference and the off chance that someone asks go-ceph to
+// support the older format.
+
+var sampleOldDaemonStatus1 = `
+{"4157": {"1": {"name": "cephfs", "directory_count": 0, "peers": {}}}}
+`
+
+var sampleOldDaemonStatus2 = `
+{
+  "4154": {
+    "1": {
+      "name": "cephfs",
+      "directory_count": 1,
+      "peers": {
+        "d284fccd-6110-4e94-843c-78ecda3aef38": {
+          "remote": {"client_name": "client.mirror_remote", "cluster_name": "ceph_b", "fs_name": "cephfs"},
+          "stats": {"failure_count": 1, "recovery_count": 0}
+        }
+      }
+    }
+  }
+}
+`
+
+var sampleDaemonStatus1 = `
+[
+  {
+    "daemon_id": 4115,
+    "filesystems": [
+      {
+        "filesystem_id": 1,
+        "name": "cephfs",
+        "directory_count": 0,
+        "peers": []
+      }
+    ]
+  }
+]
+`
+
+var sampleDaemonStatus2 = `
+[
+  {
+    "daemon_id": 4143,
+    "filesystems": [
+      {
+        "filesystem_id": 1,
+        "name": "cephfs",
+        "directory_count": 1,
+        "peers": [
+          {
+            "uuid": "43c50942-9dba-4f66-8f9b-102378fa863e",
+            "remote": {
+              "client_name": "client.mirror_remote",
+              "cluster_name": "ceph_b",
+              "fs_name": "cephfs"
+            },
+            "stats": {
+              "failure_count": 1,
+              "recovery_count": 0
+            }
+          }
+        ]
+      }
+    ]
+  }
+]
+`
+
+var samplePeerList1 = `
+{
+  "f138660d-7b22-4036-95ba-0fda727bff40": {
+    "client_name": "client.mirror_remote",
+    "site_name": "ceph_b",
+    "fs_name": "cephfs",
+    "mon_host": "test_ceph_b"
+  }
+}
+`
+
+func TestParseDaemonStatus(t *testing.T) {
+	t.Run("error", func(t *testing.T) {
+		r := newResponse(nil, "", errors.New("snark"))
+		_, err := parseDaemonStatus(r)
+		assert.Error(t, err)
+		assert.Equal(t, "snark", err.Error())
+	})
+	t.Run("statusSet", func(t *testing.T) {
+		r := newResponse(nil, "oopsie", nil)
+		_, err := parseDaemonStatus(r)
+		assert.Error(t, err)
+	})
+	t.Run("ok", func(t *testing.T) {
+		r := newResponse([]byte(sampleDaemonStatus1), "", nil)
+		ds, err := parseDaemonStatus(r)
+		assert.NoError(t, err)
+		if assert.NotNil(t, ds) && assert.Len(t, ds, 1) {
+			dsi := ds[0]
+			assert.Equal(t, DaemonID(4115), dsi.DaemonID)
+			if assert.Len(t, dsi.FileSystems, 1) {
+				fs := dsi.FileSystems[0]
+				assert.Len(t, fs.Peers, 0)
+				assert.Equal(t, "cephfs", fs.Name)
+				assert.Equal(t, int64(0), fs.DirectoryCount)
+			}
+		}
+	})
+	t.Run("ok2", func(t *testing.T) {
+		r := newResponse([]byte(sampleDaemonStatus2), "", nil)
+		ds, err := parseDaemonStatus(r)
+		assert.NoError(t, err)
+		if assert.NotNil(t, ds) && assert.Len(t, ds, 1) {
+			dsi := ds[0]
+			assert.Equal(t, DaemonID(4143), dsi.DaemonID)
+			if assert.Len(t, dsi.FileSystems, 1) {
+				fs := dsi.FileSystems[0]
+				assert.Equal(t, "cephfs", fs.Name)
+				assert.Equal(t, int64(1), fs.DirectoryCount)
+				if assert.Len(t, fs.Peers, 1) {
+					p := fs.Peers[0]
+					assert.Equal(t, "ceph_b", p.Remote.ClusterName)
+					assert.Equal(t, "cephfs", p.Remote.FSName)
+					assert.Equal(t, uint64(1), p.Stats.FailureCount)
+					assert.Equal(t, uint64(0), p.Stats.RecoveryCount)
+				}
+			}
+		}
+	})
+}
+
+func TestParsePeerList(t *testing.T) {
+	t.Run("error", func(t *testing.T) {
+		r := newResponse(nil, "", errors.New("snark"))
+		_, err := parsePeerList(r)
+		assert.Error(t, err)
+		assert.Equal(t, "snark", err.Error())
+	})
+	t.Run("statusSet", func(t *testing.T) {
+		r := newResponse(nil, "oopsie", nil)
+		_, err := parsePeerList(r)
+		assert.Error(t, err)
+	})
+	t.Run("ok", func(t *testing.T) {
+		r := newResponse([]byte(samplePeerList1), "", nil)
+		plr, err := parsePeerList(r)
+		assert.NoError(t, err)
+		if assert.NotNil(t, plr) && assert.Len(t, plr, 1) {
+			p, ok := plr[PeerUUID("f138660d-7b22-4036-95ba-0fda727bff40")]
+			assert.True(t, ok)
+			assert.Equal(t, "client.mirror_remote", p.ClientName)
+			assert.Equal(t, "ceph_b", p.SiteName)
+			assert.Equal(t, "cephfs", p.FSName)
+			assert.Equal(t, "test_ceph_b", p.MonHost)
+		}
+	})
+}
diff --git a/cephfs/admin/mirror_workflow_test.go b/cephfs/admin/mirror_workflow_test.go
new file mode 100644
index 0000000..9ca799b
--- /dev/null
+++ b/cephfs/admin/mirror_workflow_test.go
@@ -0,0 +1,234 @@
+// +build !nautilus,!octopus
+
+package admin
+
+import (
+	"errors"
+	"os"
+	pth "path"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ceph/go-ceph/cephfs"
+)
+
+func mirrorConfig() string {
+	return os.Getenv("MIRROR_CONF")
+}
+
+const (
+	noForce      = false
+	mirrorClient = "client.mirror_remote"
+)
+
+func TestMirroring(t *testing.T) {
+	if mirrorConfig() == "" {
+		t.Skip("no mirror config available")
+	}
+
+	fsa1 := getFSAdmin(t)
+	fsname := "cephfs"
+
+	require.NotNil(t, fsa1.conn)
+	err := fsa1.EnableMirroringModule(noForce)
+	assert.NoError(t, err)
+	defer func() {
+		err := fsa1.DisableMirroringModule()
+		assert.NoError(t, err)
+	}()
+	require.NoError(t, err)
+
+	smadmin1 := fsa1.SnapshotMirror()
+	err = smadmin1.Enable(fsname)
+	require.NoError(t, err)
+	defer func() {
+		err := smadmin1.Disable(fsname)
+		require.NoError(t, err)
+	}()
+
+	fsa2 := newFSAdmin(t, mirrorConfig())
+	err = fsa2.EnableMirroringModule(noForce)
+	require.NoError(t, err)
+	defer func() {
+		err := fsa2.DisableMirroringModule()
+		assert.NoError(t, err)
+	}()
+
+	smadmin2 := fsa2.SnapshotMirror()
+	err = smadmin2.Enable(fsname)
+	require.NoError(t, err)
+	defer func() {
+		err := smadmin2.Disable(fsname)
+		require.NoError(t, err)
+	}()
+
+	// from https://docs.ceph.com/en/pacific/dev/cephfs-mirroring/
+	//  "Peer bootstrap involves creating a bootstrap token on the peer cluster"
+	// and "Import the bootstrap token in the primary cluster"
+	token, err := smadmin2.CreatePeerBootstrapToken(fsname, mirrorClient, "ceph_b")
+	require.NoError(t, err)
+	err = smadmin1.ImportPeerBoostrapToken(fsname, token)
+	require.NoError(t, err)
+
+	// we need a path to mirror
+	path := "/wonderland"
+
+	mount1 := fsConnect(t, "")
+	defer func(mount *cephfs.MountInfo) {
+		assert.NoError(t, mount.Unmount())
+		assert.NoError(t, mount.Release())
+	}(mount1)
+
+	mount2 := fsConnect(t, mirrorConfig())
+	defer func(mount *cephfs.MountInfo) {
+		assert.NoError(t, mount.Unmount())
+		assert.NoError(t, mount.Release())
+	}(mount2)
+
+	err = mount1.MakeDir(path, 0770)
+	require.NoError(t, err)
+	defer func() {
+		err = mount2.ChangeDir("/")
+		assert.NoError(t, err)
+		err = mount1.RemoveDir(path)
+		assert.NoError(t, err)
+	}()
+	err = mount2.MakeDir(path, 0770)
+	require.NoError(t, err)
+	defer func() {
+		err = mount2.ChangeDir("/")
+		assert.NoError(t, err)
+		err = mount2.RemoveDir(path)
+		assert.NoError(t, err)
+	}()
+
+	err = smadmin1.Add(fsname, path)
+	require.NoError(t, err)
+
+	err = mount1.ChangeDir(path)
+	require.NoError(t, err)
+
+	// write some dirs & files
+	err = mount1.MakeDir("drink_me", 0770)
+	require.NoError(t, err)
+	err = mount1.MakeDir("eat_me", 0770)
+	require.NoError(t, err)
+	writeFile(t, mount1, "drink_me/bottle1.txt",
+		[]byte("magic potions #1\n"))
+
+	snapname1 := "alice"
+	err = mount1.MakeDir(pth.Join(snapDir, snapname1), 0700)
+	require.NoError(t, err)
+	defer func() {
+		err := mount1.RemoveDir(pth.Join(snapDir, snapname1))
+		assert.NoError(t, err)
+		err = mount2.RemoveDir(pth.Join(snapDir, snapname1))
+		assert.NoError(t, err)
+	}()
+
+	err = mount2.ChangeDir(path)
+	require.NoError(t, err)
+
+	// wait a bit for the snapshot to propagate and the dirs to be created on
+	// the remote fs.
+	for i := 0; i < 60; i++ {
+		time.Sleep(500 * time.Millisecond)
+		_, err1 := mount2.Statx("drink_me", cephfs.StatxBasicStats, 0)
+		_, err2 := mount2.Statx("eat_me", cephfs.StatxBasicStats, 0)
+		if err1 == nil && err2 == nil {
+			break
+		}
+	}
+
+waitforpeers:
+	for i := 0; i < 60; i++ {
+		time.Sleep(500 * time.Millisecond)
+		dstatus, err := smadmin1.DaemonStatus(fsname)
+		assert.NoError(t, err)
+		for _, dsinfo := range dstatus {
+			for _, fsinfo := range dsinfo.FileSystems {
+				if len(fsinfo.Peers) > 0 {
+					break waitforpeers
+				}
+			}
+		}
+	}
+
+	p, err := smadmin1.PeerList(fsname)
+	assert.NoError(t, err)
+	assert.Len(t, p, 1)
+	for _, peer := range p {
+		assert.Equal(t, "cephfs", peer.FSName)
+	}
+
+	stx, err := mount2.Statx("drink_me", cephfs.StatxBasicStats, 0)
+	if assert.NoError(t, err) {
+		assert.Equal(t, uint16(0040000), stx.Mode&0040000) // is dir?
+	}
+
+	stx, err = mount2.Statx("eat_me", cephfs.StatxBasicStats, 0)
+	if assert.NoError(t, err) {
+		assert.Equal(t, uint16(0040000), stx.Mode&0040000) // is dir?
+	}
+
+	stx, err = mount2.Statx("drink_me/bottle1.txt", cephfs.StatxBasicStats, 0)
+	if assert.NoError(t, err) {
+		assert.Equal(t, uint16(0100000), stx.Mode&0100000) // is reg?
+		assert.Equal(t, uint64(17), stx.Size)
+	}
+	data := readFile(t, mount2, "drink_me/bottle1.txt")
+	assert.Equal(t, "magic potions #1\n", string(data))
+
+	err = mount1.Unlink("drink_me/bottle1.txt")
+	require.NoError(t, err)
+	err = mount1.RemoveDir("drink_me")
+	require.NoError(t, err)
+	err = mount1.RemoveDir("eat_me")
+	require.NoError(t, err)
+
+	snapname2 := "rabbit"
+	err = mount1.MakeDir(pth.Join(snapDir, snapname2), 0700)
+	require.NoError(t, err)
+	defer func() {
+		err := mount1.RemoveDir(pth.Join(snapDir, snapname2))
+		assert.NoError(t, err)
+		err = mount2.RemoveDir(pth.Join(snapDir, snapname2))
+		assert.NoError(t, err)
+	}()
+
+	// wait a bit for the snapshot to propagate and the dirs to be removed on
+	// the remote fs.
+	for i := 0; i < 60; i++ {
+		time.Sleep(500 * time.Millisecond)
+		_, err1 := mount2.Statx("drink_me", cephfs.StatxBasicStats, 0)
+		_, err2 := mount2.Statx("eat_me", cephfs.StatxBasicStats, 0)
+		if err1 != nil && err2 != nil {
+			break
+		}
+
+	}
+	_, err = mount2.Statx("drink_me", cephfs.StatxBasicStats, 0)
+	if assert.Error(t, err) {
+		var ec errorWithCode
+		if assert.True(t, errors.As(err, &ec)) {
+			assert.Equal(t, -2, ec.ErrorCode())
+		}
+	}
+	_, err = mount2.Statx("eat_me", cephfs.StatxBasicStats, 0)
+	if assert.Error(t, err) {
+		var ec errorWithCode
+		if assert.True(t, errors.As(err, &ec)) {
+			assert.Equal(t, -2, ec.ErrorCode())
+		}
+	}
+
+	err = smadmin1.Remove(fsname, path)
+	assert.NoError(t, err)
+}
+
+type errorWithCode interface {
+	ErrorCode() int
+}
diff --git a/cephfs/admin/workflow_test.go b/cephfs/admin/workflow_test.go
index f5a1c0e..d0fcfed 100644
--- a/cephfs/admin/workflow_test.go
+++ b/cephfs/admin/workflow_test.go
@@ -19,13 +19,18 @@ import (
 
 var snapDir = ".snapshots"
 
-func fsConnect(t *testing.T) *cephfs.MountInfo {
+func fsConnect(t *testing.T, configFile string) *cephfs.MountInfo {
 	mount, err := cephfs.CreateMount()
 	require.NoError(t, err)
 	require.NotNil(t, mount)
 
-	err = mount.ReadDefaultConfigFile()
-	require.NoError(t, err)
+	if configFile == "" {
+		err = mount.ReadDefaultConfigFile()
+		require.NoError(t, err)
+	} else {
+		err = mount.ReadConfigFile(configFile)
+		require.NoError(t, err)
+	}
 	err = mount.SetConfigOption("client_snapdir", snapDir)
 	require.NoError(t, err)
 
@@ -114,7 +119,7 @@ func TestWorkflow(t *testing.T) {
 	require.NotEqual(t, "", subPath)
 
 	// connect to volume, cd to path (?)
-	mount := fsConnect(t)
+	mount := fsConnect(t, "")
 	defer func(mount *cephfs.MountInfo) {
 		assert.NoError(t, mount.Unmount())
 		assert.NoError(t, mount.Release())