CephFS Mirror

CephFS mirror

cephfs/mirror.Run links a source and a destination filesystem in two clusters and starts cephfs-mirror daemons. The fixture seeds and enrolls the source directory; the service under test writes and snapshots on the source (or reads the destination), and the test waits for the exact source snapshot and verifies the destination state afterwards.

Each scenario follows the same shape: the goal of the test, the fixture it builds independently of the system under test (SUT), the hand-off of connection details to the SUT, an optional condition step that reproduces an operational situation, the verification of the collaborator's state after the SUT ran, and cleanup.

Snippets run inside a test function with ctx := t.Context(). Fragments that do not show their own Run use the fixture named in the scenario (cluster, source, destination, link). Every snippet on this page compiles against the current API; replace the command-line writers with your own client in a real test.

The scenario uses the helpers from CephFS.

Mirror a directory and wait for a snapshot

Two independent clusters each get a filesystem. The source directory is created first, mirroring starts, and the test waits for one exact snapshot before reading it on the destination.

const sourceScript = `import cephfs, json, os, sys
fsname, phase = sys.argv[1], sys.argv[2]
fs = cephfs.LibCephFS(conffile='/etc/ceph/ceph.conf', auth_id='admin')
fs.mount(filesystem_name=fsname.encode())
try:
    if phase == 'seed':
        fs.mkdir('/data', 0o755)
        fd = fs.open('/data/file', os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o644)
        fs.write(fd, b'mirrored bytes', 0)
        fs.fsync(fd, 0)
        fs.close(fd)
    elif phase == 'snapshot':
        name = sys.argv[3]
        fs.mkdir('/data/.snap/' + name, 0o755)
        print(json.dumps({'id': fs.snap_info('/data/.snap/' + name)['id'], 'name': name}))
finally:
    fs.shutdown()
`
const readFile = `import cephfs, os, sys
fs = cephfs.LibCephFS(conffile='/etc/ceph/ceph.conf', auth_id='admin')
fs.mount(filesystem_name=sys.argv[1].encode())
try:
    fd = fs.open(sys.argv[2], os.O_RDONLY)
    print(fs.read(fd, 0, 1 << 20).decode())
    fs.close(fd)
finally:
    fs.shutdown()
`

func TestCephFSMirror(t *testing.T) {
	ctx := t.Context()
	sourceCluster, err := cephfs.Run(ctx, ceph.DefaultImage)
	testcontainers.CleanupContainer(t, sourceCluster)
	if err != nil {
		t.Fatal(err)
	}
	destinationCluster, err := cephfs.Run(ctx, ceph.DefaultImage)
	testcontainers.CleanupContainer(t, destinationCluster)
	if err != nil {
		t.Fatal(err)
	}
	sourceFS := cephfs.Filesystems(sourceCluster.Container)[0]
	destinationFS := cephfs.Filesystems(destinationCluster.Container)[0]
	sourceClient := startClient(t, sourceCluster)
	destinationClient := startClient(t, destinationCluster)

	mustExec(t, sourceClient, "python3", "-c", sourceScript, sourceFS.FilesystemName, "seed")

	// Registered after the clusters, so it is terminated before them.
	sourceClusterFacts, err := ceph.FactsOf(sourceCluster.Container)
	if err != nil {
		t.Fatal(err)
	}
	mirror, err := cephfsmirror.Run(ctx, sourceClusterFacts.Images.CephFS, cephfsmirror.WithConfig(cephfsmirror.Config{
		Source:                sourceCluster.Container,
		Destination:           destinationCluster.Container,
		SourceFilesystem:      sourceFS.FilesystemName,
		DestinationFilesystem: destinationFS.FilesystemName,
		Directories:           []string{"/data"},
	}))
	if mirror != nil {
		t.Cleanup(func() {
			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
			defer cancel()
			if err := mirror.Terminate(ctx); err != nil {
				t.Error(err)
			}
		})
	}
	if err != nil {
		t.Fatal(err)
	}
	if _, err := mirror.WaitDirectoryReady(ctx, "/data"); err != nil {
		t.Fatal(err)
	}

	var checkpoint cephfsmirror.Snapshot // decoded from {"id": ..., "name": ...}
	out := mustExec(t, sourceClient, "python3", "-c", sourceScript, sourceFS.FilesystemName, "snapshot", "checkpoint-1")
	if err := json.Unmarshal([]byte(out), &checkpoint); err != nil {
		t.Fatal(err)
	}
	if _, err := mirror.WaitSnapshotSynced(ctx, "/data", checkpoint); err != nil {
		t.Fatal(err)
	}

	got := mustExec(t, destinationClient, "python3", "-c", readFile,
		destinationFS.FilesystemName, "/data/.snap/checkpoint-1/file")
	if got != "mirrored bytes" {
		t.Fatalf("destination snapshot holds %q", got)
	}
}

Notes:

• Deleting a snapshot on the source deletes it on the destination too. Mirroring is replication, not a backup with its own retention.
• To test an outage, stop a daemon (mirror.Daemons()[0].Stop), remove all daemons with Daemon.Remove, or cut the peer link with Daemon.InterruptPeerLink. Snapshots taken meanwhile arrive after recovery.
• With DaemonCount: 2, a stopped daemon's directories move to the survivor. After AddDaemon, call RebalanceDirectories on Ceph 20.2.4.
• Compare file bytes, names, symlinks, modes and owners on the destination. Native mirroring is not guaranteed to reproduce user.* extended attributes.

References

• API reference: pkg.go.dev or go doc on each package
• Repository: README and guides
• Purpose and scope: API purpose and purpose tags