CephFS

CephFS

cephfs.Run starts a cluster with a filesystem and its MDS daemons; cephfs.Filesystems(cluster.Container) returns the filesystems for the test. Subvolumes, restricted principals and snapshots prepare the state the SUT mounts; MDS failover and client eviction are condition steps.

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.

Helpers

The examples share a small helper that runs a command and fails the test on a non-zero exit, and a helper that attaches a client container:

import (
	"context"
	"io"
	"slices"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/JSYoo5B/ceph-testcontainers-go/ceph"
	"github.com/JSYoo5B/ceph-testcontainers-go/cephfs"
	"github.com/JSYoo5B/ceph-testcontainers-go/fault"
	"github.com/JSYoo5B/ceph-testcontainers-go/override"
	"github.com/testcontainers/testcontainers-go"
	tcexec "github.com/testcontainers/testcontainers-go/exec"
)

func mustExec(t *testing.T, ctr testcontainers.Container, args ...string) string {
	t.Helper()
	code, reader, err := ctr.Exec(t.Context(), args, tcexec.Multiplexed())
	if err != nil {
		t.Fatal(err)
	}
	output, err := io.ReadAll(reader)
	if err != nil {
		t.Fatal(err)
	}
	if code != 0 {
		t.Fatalf("%v exited %d: %s", args, code, output)
	}
	return strings.TrimSpace(string(output))
}

func startClient(t *testing.T, cluster *cephfs.Container) testcontainers.Container {
	t.Helper()
	client, err := testcontainers.Run(t.Context(), ceph.DefaultImage,
		cluster.WithClient(), ceph.WithIdleEntrypoint())
	testcontainers.CleanupContainer(t, client)
	if err != nil {
		t.Fatal(err)
	}
	return client
}

Mount a filesystem with libcephfs in a client container

Ceph clients talk to the MON, the MDS and every OSD directly, so the client must reach the addresses Ceph advertises. In the default bridge mode, run the client in a container attached with cluster.WithClient() (admin credentials) or client.WithIdentity() (one principal). The container needs the client library you test; ceph.DefaultImage has Python libcephfs, and go-ceph programs can run in any Linux image with libcephfs installed.

const listRoot = `import cephfs, sys
fsname, root = sys.argv[1], sys.argv[2]
fs = cephfs.LibCephFS(conffile='/etc/ceph/ceph.conf', auth_id='admin')
fs.conf_set('client_mount_timeout', '30')
fs.mount(mount_root=root.encode(), filesystem_name=fsname.encode())
try:
    d = fs.opendir('/')
    entry = fs.readdir(d)
    while entry:
        print(entry.d_name.decode())
        entry = fs.readdir(d)
    fs.closedir(d)
finally:
    fs.shutdown()
`

cluster, err := cephfs.Run(ctx, ceph.DefaultImage,
	cephfs.WithFilesystems(cephfs.Config{Name: "app-fs"}, cephfs.Config{Name: "logs"}))
testcontainers.CleanupContainer(t, cluster)
if err != nil {
	t.Fatal(err)
}
fs := cephfs.Filesystems(cluster.Container)[0] // sorted by name: app-fs
volume, err := fs.CreateSubvolume(ctx, cephfs.SubvolumeConfig{Name: "data"})
if err != nil {
	t.Fatal(err)
}
client := startClient(t, cluster)
t.Log(mustExec(t, client, "python3", "-c", listRoot, fs.FilesystemName, volume.Path))

Notes:

• With several filesystems, always pass filesystem_name; otherwise the client mounts Ceph's default filesystem.
• Each python3 process opens a new session. Start a new process for each phase of a test (before and after a failure) so cached data cannot hide a problem.
• Long scripts can be copied once with client.CopyToContainer and run by path.
• No kernel or FUSE mount is used, so the client container needs no privileges or /dev/fuse.
• For a native client in the test process, start the cluster with ceph.WithHostNetwork() on a Linux Docker host and pass the bytes from cluster.ConnectionConfig(ctx) to the client. See Host network.

Test quota enforcement (EDQUOT)

A subvolume's SizeBytes becomes the ceph.quota.max_bytes attribute of its directory. libcephfs refuses a write that would cross the limit with EDQUOT (errno 122), which the Python binding raises as cephfs.DiskQuotaExceeded.

const quotaProbe = `import cephfs, errno, os, sys
fsname, path, size = sys.argv[1], sys.argv[2], int(sys.argv[3])
fs = cephfs.LibCephFS(conffile='/etc/ceph/ceph.conf', auth_id='admin')
fs.conf_set('client_quota', 'true')
fs.mount(filesystem_name=fsname.encode())
fd = fs.open(path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
try:
    fs.write(fd, b'x' * size, 0)
    fs.fsync(fd, 0)
    print('written')
except cephfs.DiskQuotaExceeded as e:
    assert e.errno == errno.EDQUOT
    print('EDQUOT')
finally:
    fs.close(fd)
    fs.shutdown()
`

// fs and client as in "Mount a filesystem with libcephfs in a client container".
volume, err := fs.CreateSubvolume(ctx, cephfs.SubvolumeConfig{Name: "limited", SizeBytes: 1 << 20})
if err != nil {
	t.Fatal(err)
}
file := volume.Path + "/big"
if out := mustExec(t, client, "python3", "-c", quotaProbe, fs.FilesystemName, file, "2097152"); out != "EDQUOT" {
	t.Fatalf("2 MiB write into a 1 MiB quota: %s", out)
}
if err := volume.Resize(ctx, 8<<20); err != nil {
	t.Fatal(err)
}
if out := mustExec(t, client, "python3", "-c", quotaProbe, fs.FilesystemName, file, "2097152"); out != "written" {
	t.Fatalf("write after resize: %s", out)
}

Notes:

• Quotas are enforced by the client from the usage it knows. A single write that would cross the limit fails reliably; many concurrent writers on several clients can overshoot before enforcement catches up.
• BytesUsed from SubvolumeInfo is recursive accounting and lags writes. Poll it until it reflects your data before testing that a resize below usage is refused.
• A group quota (SubvolumeGroupConfig.SizeBytes) limits all subvolumes in the group; the smaller of the two limits applies.

Give each subvolume its own restricted principal

Each tenant gets a namespace-isolated subvolume and a principal that can only reach it. The client container attached with ClientConfig.WithIdentity holds only that principal's keyring.

const tenantProbe = `import cephfs, errno, os, sys
fsname, user, keyring, own, other = sys.argv[1:]
def mount(root):
    fs = cephfs.LibCephFS(conffile='/etc/ceph/ceph.conf', auth_id=user)
    fs.conf_set('keyring', keyring)
    fs.conf_set('client_mount_timeout', '10')
    fs.mount(mount_root=root.encode(), filesystem_name=fsname.encode())
    return fs
fs = mount(own)
fd = fs.open('/owned', os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
fs.write(fd, b'tenant data', 0)
fs.fsync(fd, 0)
fs.close(fd)
fs.shutdown()
try:
    mount(other)
except cephfs.Error as e:
    assert abs(e.args[0]) in (errno.EPERM, errno.EACCES), e
    print('denied')
else:
    raise AssertionError('mounted another tenant')
`

group, err := fs.CreateSubvolumeGroup(ctx, cephfs.SubvolumeGroupConfig{Name: "tenants"})
if err != nil {
	t.Fatal(err)
}
volumes := map[string]*cephfs.Subvolume{}
for _, name := range []string{"tenant-a", "tenant-b"} {
	volumes[name], err = fs.CreateSubvolume(ctx, cephfs.SubvolumeConfig{
		Name: name, GroupName: group.Name, SizeBytes: 64 << 20, NamespaceIsolated: true,
	})
	if err != nil {
		t.Fatal(err)
	}
}
grant, err := volumes["tenant-a"].Authorize(ctx,
	cephfs.SubvolumeAuthorizationConfig{ClientID: "tenant-a", Access: "rw"})
if err != nil {
	t.Fatal(err) // a non-nil grant may be partial; do not authorize the same ID again
}
tenant, err := testcontainers.Run(ctx, ceph.DefaultImage,
	grant.Client.WithIdentity(), ceph.WithIdleEntrypoint())
testcontainers.CleanupContainer(t, tenant)
if err != nil {
	t.Fatal(err)
}
out := mustExec(t, tenant, "python3", "-c", tenantProbe, fs.FilesystemName,
	grant.Client.User(), grant.Client.KeyringPath(), grant.Path, volumes["tenant-b"].Path)
if out != "denied" {
	t.Fatalf("tenant probe: %s", out)
}

// Revoke: new connections are refused, then established sessions are evicted.
if err := grant.Deauthorize(ctx); err != nil {
	t.Fatal(err)
}
if err := grant.EvictClients(ctx); err != nil {
	t.Fatal(err)
}
if err := grant.Client.Delete(ctx); err != nil { // optional
	t.Fatal(err)
}

Notes:

• SubvolumeAuthorizedClients(ctx, "tenant-a", "tenants") lists the principals Ceph recorded for the subvolume.
• Test revocation with a new process. A principal left with no caps at all can time out during mount instead of failing with EPERM, so do not use a timeout alone as proof of revocation.
• rw does not allow changing layouts (p) or managing snapshots (s). For other cap combinations, create the principal with cluster.CreateClient and ceph.ClientCaps.
• Terminate tenant containers before the cluster; CleanupContainer does this when the containers are registered after the cluster.

Take a consistent snapshot of several subvolumes

Snapshots are taken one subvolume at a time and do not pause writers. To get one checkpoint across several subvolumes, quiesce them, take the snapshots and release.

q, err := fs.QuiesceSubvolumes(ctx, []*cephfs.Subvolume{orders, payments},
	cephfs.QuiesceConfig{Timeout: 20 * time.Second, Expiration: time.Minute})
if err != nil {
	// A non-nil q may exist after an uncertain creation; Ceph releases it
	// when Expiration passes.
	t.Fatal(err)
}
for _, volume := range []*cephfs.Subvolume{orders, payments} {
	if _, err := volume.CreateSnapshot(ctx, "checkpoint-1"); err != nil {
		t.Fatal(err) // Expiration resumes I/O even if the test stops here.
	}
}
if err := q.Release(ctx); err != nil {
	// The set expired, timed out or was changed: the snapshots are not a
	// consistent checkpoint.
	t.Fatal(err)
}

Notes:

• Choose Expiration longer than the snapshot step. While the set is QUIESCED, client I/O on the members is paused.
• q.Status(ctx) shows the state without extending the expiration.

Fail over an MDS and replace a stopped MDS

With one active and one standby daemon, stopping the active daemon makes the standby take rank 0. The test then either restarts the old daemon or retires it and creates a new one.

cluster, err := cephfs.Run(ctx, ceph.DefaultImage,
	cephfs.WithFilesystems(cephfs.Config{ActiveMDS: 1, StandbyMDS: 1}))
testcontainers.CleanupContainer(t, cluster)
if err != nil {
	t.Fatal(err)
}
fs := cephfs.Filesystems(cluster.Container)[0]

before, err := fs.MDSStatus(ctx)
if err != nil {
	t.Fatal(err)
}
var old *cephfs.MDS
for _, mds := range fs.MDSs() {
	if mds.ID == before.Active[0].Name {
		old = mds
	}
}
timeout := 10 * time.Second
if err := old.Stop(ctx, &timeout); err != nil {
	t.Fatal(err)
}
// Fail the old GID now instead of waiting for mds_beacon_grace.
if _, err := cluster.Ceph(ctx, "mds", "fail", strconv.FormatUint(before.Active[0].GID, 10)); err != nil {
	t.Fatal(err)
}
deadline := time.Now().Add(2 * time.Minute)
for {
	status, err := fs.MDSStatus(ctx)
	if err == nil && len(status.Active) == 1 && status.Active[0].Name != old.ID &&
		status.Active[0].State == "up:active" {
		break
	}
	if time.Now().After(deadline) {
		t.Fatalf("no takeover: %+v %v", status, err)
	}
	time.Sleep(time.Second)
}
// Clients can mount again. WaitReady would fail now: standby capacity is missing.

// Option 1: restart the old daemon; it rejoins as a standby with a new GID.
//     err = old.Start(ctx)
// Option 2: retire it and let ScaleMDS start a daemon with a new name.
if err := old.RemoveStopped(ctx); err != nil {
	t.Fatal(err)
}
if err := fs.ScaleMDS(ctx, 1, 1); err != nil {
	t.Fatal(err)
}
if err := fs.WaitReady(ctx); err != nil {
	t.Fatal(err)
}

With StandbyReplay: true the follower takes over faster because it already replays the active daemon's journal; a restarted daemon rejoins as the new follower.

When the filesystem has a single MDS and no standby, nothing can take over. The filesystem is unavailable: Ceph reports FS_WITH_FAILED_MDS and FS_DEGRADED (warning) and MDS_ALL_DOWN (error), and new mounts fail after client_mount_timeout with errno 110. MDS.AddReplacement starts a new daemon for the failed rank:

fs := cephfs.Filesystems(cluster.Container)[0] // 1 active, 0 standby
status, err := fs.MDSStatus(ctx)
if err != nil {
	t.Fatal(err)
}
old := fs.MDSs()[0]
timeout := 10 * time.Second
if err := old.Stop(ctx, &timeout); err != nil {
	t.Fatal(err)
}
// Without a standby Ceph may keep a dead daemon registered as laggy.
if _, err := cluster.Ceph(ctx, "mds", "fail", strconv.FormatUint(status.Active[0].GID, 10)); err != nil {
	t.Fatal(err)
}
replacement, err := old.AddReplacement(ctx)
if err != nil {
	t.Fatal(err) // a non-nil replacement stays owned by the cluster
}
if err := old.RemoveStopped(ctx); err != nil {
	t.Fatal(err)
}
if err := fs.WaitReady(ctx); err != nil {
	t.Fatal(err)
}
t.Logf("rank 0 is now mds.%s", replacement.ID)

The old daemon's Cephx key stays in the cluster. Files written before the failure are readable through the new daemon.

Instead of a replacement, the same container can be started again with old.Start(ctx) after mds fail. The restarted daemon takes rank 0 with a new GID, and a client that stayed mounted reconnects: its session keeps the same client ID, PID and address, and a file it opened before the failure keeps working. See MDS sessions.

Evict a frozen client

A client that stops responding keeps its session and capabilities until the MDS gives up on it. Freezing the client container with fault.Pause reproduces this. With shortened session limits, the test can observe the eviction, the blocklist entry and the loss of unsynced data.

// The holder writes "durable" and syncs it, then caches "-volatile" without
// syncing, publishes its client ID and waits.
const holderScript = `import cephfs, os, sys, time
fs = cephfs.LibCephFS(conffile='/etc/ceph/ceph.conf', auth_id='admin')
fs.mount(filesystem_name=sys.argv[1].encode())
fd = fs.open('/held', os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o644)
fs.write(fd, b'durable', 0)
fs.fsync(fd, 0)
fs.write(fd, b'-volatile', 7)
with open('/tmp/holder-id', 'w') as out:
    out.write(str(fs.get_instance_id()))
time.sleep(3600)
`
const readerScript = `import cephfs, os, sys
fs = cephfs.LibCephFS(conffile='/etc/ceph/ceph.conf', auth_id='admin')
fs.conf_set('client_mount_timeout', '30')
fs.mount(filesystem_name=sys.argv[1].encode())
fd = fs.open('/held', os.O_RDONLY)
print(fs.read(fd, 0, 64).decode())
fs.close(fd)
fs.shutdown()
`

findSession := func(id string) *cephfs.Session {
	sessions, err := fs.Sessions(ctx)
	if err != nil {
		t.Fatal(err)
	}
	for _, session := range sessions {
		if strconv.FormatUint(session.ID, 10) == id {
			return &session
		}
	}
	return nil
}

change, err := fs.OverrideSessionTimeouts(ctx, cephfs.SessionTimeouts{
	Timeout: 30 * time.Second, Autoclose: time.Minute,
})
if change != nil {
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
		defer cancel()
		_ = change.Restore(ctx)
	})
}
if err != nil {
	t.Fatal(err)
}

held := startClient(t, cluster)
mustExec(t, held, "sh", "-c", `nohup python3 -c "$1" "$2" >/tmp/holder.log 2>&1 &`,
	"sh", holderScript, fs.FilesystemName)
var id string
for deadline := time.Now().Add(time.Minute); id == ""; time.Sleep(200 * time.Millisecond) {
	if time.Now().After(deadline) {
		t.Fatal(mustExec(t, held, "cat", "/tmp/holder.log"))
	}
	id = mustExec(t, held, "sh", "-c", "cat /tmp/holder-id 2>/dev/null || true")
}
session := findSession(id)
if session == nil {
	t.Fatal("holder has no session")
}

pause, err := fault.Pause(ctx, cluster.Container, held)
if pause != nil {
	t.Cleanup(func() { _ = pause.Resume(context.Background()) })
}
if err != nil {
	t.Fatal(err)
}

// The reader needs the frozen holder's capabilities, so the MDS evicts the
// holder as soon as Timeout passes. The reader gets only the synced bytes.
other := startClient(t, cluster)
if got := mustExec(t, other, "python3", "-c", readerScript, fs.FilesystemName); got != "durable" {
	t.Fatalf("reader got %q", got)
}
for deadline := time.Now().Add(2 * time.Minute); findSession(id) != nil; time.Sleep(2 * time.Second) {
	if time.Now().After(deadline) {
		t.Fatal("the MDS kept the frozen client's session")
	}
}
observer, err := ceph.Observe(cluster.Container)
if err != nil {
	t.Fatal(err)
}
entries, err := observer.BlocklistEntries(ctx)
if err != nil {
	t.Fatal(err)
}
if !slices.ContainsFunc(entries, func(e ceph.BlocklistEntry) bool { return e.Address == session.Address }) {
	t.Fatalf("%s is not blocklisted", session.Address)
}
if err := pause.Resume(ctx); err != nil {
	t.Fatal(err)
}

What to expect:

Situation	Eviction
Another client wants the capabilities	After about Timeout (30 s here). The waiting client blocks until then.
Nobody wants them	After about Autoclose (60 s here). The session stays open meanwhile.

After Resume, the holder's old session fails with ESHUTDOWN (errno 108); a new mount from the same container gets a new client ID and works. The cluster health returns to HEALTH_OK; Ceph evicts before it would raise MDS_CLIENT_LATE_RELEASE. The blocklist entry expires by itself (one hour by default) and is not removed by Restore. To test eviction without blocklisting, set mds_session_blocklist_on_timeout to false with override.SetConfig (section mds).

Keep an MDS active while it drops client messages

Pausing or stopping an MDS stops its beacons, so the MON marks it failed and a standby takes over. To test a client against an MDS that stays up:active but does not answer, enable the messenger option ms_blackhole_client on that daemon. The MDS keeps talking to the MON, no failover happens, and new client sessions cannot open. See Fault injection for the other messenger options.

const mountProbe = `import cephfs, sys
fs = cephfs.LibCephFS(conffile='/etc/ceph/ceph.conf', auth_id='admin')
fs.conf_set('client_mount_timeout', '5')
try:
    fs.mount(filesystem_name=sys.argv[1].encode())
    print('mounted')
except cephfs.Error as e:
    print('failed', abs(e.args[0]))
finally:
    fs.shutdown()
`

// cluster, fs and client as in "Mount a filesystem with libcephfs in a client container".
status, err := fs.MDSStatus(ctx)
if err != nil {
	t.Fatal(err)
}
active := status.Active[0]
hold, err := override.SetConfig(ctx, cluster.Container, ceph.ConfigSetting{
	Section: "mds." + active.Name, Name: "ms_blackhole_client", Value: "true",
})
t.Cleanup(func() { _ = hold.Restore(context.Background()) })
if err != nil {
	t.Fatal(err)
}
if out := mustExec(t, client, "python3", "-c", mountProbe, fs.FilesystemName); out != "failed 110" {
	t.Fatalf("mount against a silent MDS: %s", out)
}
after, err := fs.MDSStatus(ctx)
if err != nil || after.Active[0].Name != active.Name || after.Active[0].GID != active.GID {
	t.Fatalf("the MDS was replaced: %+v %v", after, err)
}
if err := hold.Restore(ctx); err != nil {
	t.Fatal(err)
}
if out := mustExec(t, client, "python3", "-c", mountProbe, fs.FilesystemName); out != "mounted" {
	t.Fatalf("mount after restore: %s", out)
}

Notes:

• Use a new client process for each phase. The effect on sessions that were already open is not covered by this recipe.
• Never set ms_blackhole_client on the MON, the MGR or the client section: the fixture's own CLI is a client, so override.SetConfig and Restore would hang.
• ms_inject_delay_type, ms_inject_delay_probability and ms_inject_delay_max add random delays instead of dropping messages.

References

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