Skip to content

RBD Mirror

Jump to bottom
JaeSang Yoo edited this page Nov 1, 2026 · 2 revisions

RBD mirror

rbd/mirror.Run links two clusters: it configures the pool policies and the peer and starts rbd-mirror receivers on the destination. The service under test works against the source (or the destination after a failover), and the test verifies the replicated state on the other side afterwards. Failover, split-brain and receiver loss are condition steps that the service must survive.

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 these helpers. run executes a command in any container and returns its exit code and combined output. waitReplicated compares the exported bytes of the source and destination images; stop writers and create the checkpoint first. waitTransmitPeer waits until the source pool knows the destination as a peer it sends to (see the pitfalls). promote retries a non-forced promotion until the demotion has propagated.

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"strings"
	"testing"
	"time"

	"github.com/JSYoo5B/ceph-testcontainers-go/ceph"
	"github.com/JSYoo5B/ceph-testcontainers-go/rbd"
	rbdmirror "github.com/JSYoo5B/ceph-testcontainers-go/rbd/mirror"
	"github.com/testcontainers/testcontainers-go"
	tcexec "github.com/testcontainers/testcontainers-go/exec"
)

func run(ctx context.Context, ctr testcontainers.Container, cmd ...string) (int, string, error) {
	code, reader, err := ctr.Exec(ctx, cmd, tcexec.Multiplexed())
	if err != nil {
		return 0, "", err
	}
	out, err := io.ReadAll(reader)
	return code, string(out), err
}

func poll(ctx context.Context, t *testing.T, what string, done func() bool) {
	t.Helper()
	for !done() {
		select {
		case <-ctx.Done():
			t.Fatalf("%s: %v", what, ctx.Err())
		case <-time.After(2 * time.Second):
		}
	}
}

func waitReplicated(ctx context.Context, t *testing.T, link *rbdmirror.Mirror, sourceSpec, destinationSpec string) {
	t.Helper()
	want, err := link.SourceRBD(ctx, "export", "--no-progress", sourceSpec, "-")
	if err != nil {
		t.Fatal(err)
	}
	poll(ctx, t, destinationSpec+" did not match the source", func() bool {
		got, err := link.DestinationRBD(ctx, "export", "--no-progress", destinationSpec, "-")
		return err == nil && bytes.Equal(got, want)
	})
}

func waitTransmitPeer(ctx context.Context, t *testing.T, link *rbdmirror.Mirror, pool, destinationSite string) {
	t.Helper()
	poll(ctx, t, "source never listed "+destinationSite+" as a transmit peer", func() bool {
		out, err := link.SourceRBD(ctx, "mirror", "pool", "info", pool, "--format", "json")
		if err != nil {
			return false
		}
		var info struct {
			Peers []struct {
				SiteName   string `json:"site_name"`
				MirrorUUID string `json:"mirror_uuid"`
				Direction  string `json:"direction"`
			} `json:"peers"`
		}
		if json.Unmarshal(out, &info) != nil {
			return false
		}
		for _, p := range info.Peers {
			if p.SiteName == destinationSite && p.MirrorUUID != "" &&
				(p.Direction == "tx-only" || p.Direction == "rx-tx") {
				return true
			}
		}
		return false
	})
}

// rbdCLI is link.SourceRBD or link.DestinationRBD of the cluster to promote.
func promote(ctx context.Context, t *testing.T, rbdCLI func(context.Context, ...string) ([]byte, error), spec string) {
	t.Helper()
	poll(ctx, t, "promote "+spec, func() bool {
		_, err := rbdCLI(ctx, "mirror", "image", "promote", spec)
		return err == nil
	})
}

The examples write test data with rbd bench --io-type write, which goes through librbd like any application writer. Use your own client in a real test.

Mirror an image with snapshot mirroring

func TestSnapshotMirroring(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
	defer cancel()
	newCluster := func() *ceph.Container {
		cluster, err := ceph.Run(ctx, ceph.DefaultImage,
			rbd.WithPools(ceph.PoolConfig{Name: "volumes"}),
		)
		testcontainers.CleanupContainer(t, cluster)
		if err != nil {
			t.Fatal(err)
		}
		return cluster
	}
	source, destination := newCluster(), newCluster()

	destinationFacts, err := ceph.FactsOf(destination)
	if err != nil {
		t.Fatal(err)
	}
	link, err := rbdmirror.Run(ctx, destinationFacts.Images.RBDAux, rbdmirror.WithConfig(rbdmirror.Config{
		Source: source, Destination: destination, Pool: "volumes",
	}))
	testcontainers.CleanupContainer(t, link) // runs before the clusters' cleanup
	if err != nil {
		t.Fatal(err)
	}
	waitTransmitPeer(ctx, t, link, "volumes", "destination")

	const spec = "volumes/disk"
	write := func() {
		if _, err := link.SourceRBD(ctx, "bench", "--io-type", "write",
			"--io-size", "64K", "--io-total", "2M", "--io-pattern", "rand", spec); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := link.SourceRBD(ctx, "create", "--size", "8M",
		"--image-feature", "layering,exclusive-lock", spec); err != nil {
		t.Fatal(err)
	}
	write()

	// Enabling snapshot mirroring takes the first mirror snapshot.
	if err := link.EnableImage(ctx, "disk"); err != nil {
		t.Fatal(err)
	}
	if _, err := link.WaitReplayReady(ctx, "disk"); err != nil {
		t.Fatal(err)
	}
	waitReplicated(ctx, t, link, spec, spec)

	// Later writes travel only with the next mirror snapshot.
	write()
	if _, err := link.SourceRBD(ctx, "mirror", "image", "snapshot", spec); err != nil {
		t.Fatal(err)
	}
	waitReplicated(ctx, t, link, spec, spec)
}

Snapshot mirroring is asynchronous and copies the image as of each mirror snapshot. Writes after the latest mirror snapshot are not on the destination. Instead of explicit snapshots you can let the MGR rbd_support module take them (rbd mirror snapshot schedule add --pool volumes 1m), but explicit snapshots make tests deterministic.

Mirror an image with journal mirroring

Journal mirroring replays every write. No mirror snapshots are involved.

destinationFacts, err := ceph.FactsOf(destination)
if err != nil {
	t.Fatal(err)
}
link, err := rbdmirror.Run(ctx, destinationFacts.Images.RBDAux, rbdmirror.WithConfig(rbdmirror.Config{
	Source: source, Destination: destination, Pool: "volumes",
	Mode: rbdmirror.ModeJournal,
}))
testcontainers.CleanupContainer(t, link)
if err != nil {
	t.Fatal(err)
}

// Journal mode needs exclusive-lock; EnableImage adds the journaling feature.
if _, err := link.SourceRBD(ctx, "create", "--size", "8M",
	"--image-feature", "layering,exclusive-lock", "volumes/disk"); err != nil {
	t.Fatal(err)
}
if err := link.EnableImage(ctx, "disk"); err != nil {
	t.Fatal(err)
}
if _, err := link.WaitReplayReady(ctx, "disk"); err != nil {
	t.Fatal(err)
}
write() // any librbd writer
waitReplicated(ctx, t, link, "volumes/disk", "volumes/disk")

With pool scope, every image that has the journaling feature is mirrored without EnableImage (which is rejected in pool scope):

destinationFacts, err := ceph.FactsOf(destination)
if err != nil {
	t.Fatal(err)
}
link, err := rbdmirror.Run(ctx, destinationFacts.Images.RBDAux, rbdmirror.WithConfig(rbdmirror.Config{
	Source: source, Destination: destination, Pool: "volumes",
	Scope: rbdmirror.ScopePool, // Mode defaults to journal
}))
// ...
_, err = link.SourceRBD(ctx, "create", "--size", "8M",
	"--image-feature", "layering,exclusive-lock,journaling", "volumes/disk")

Pool scope also enrolls existing journaled images, so quiesce their writers before creating the link.

A stopped receiver does not lose journaled writes: they wait in the source journal and are replayed when the daemon starts again. While the daemon is stopped, ImageStatus reports ReplayReady == false with the same image IDs, and WaitReplayReady gives up after four minutes, or at the caller's earlier deadline, and returns the last observation.

Fail over and fail back between two sites

Planned failover moves the primary image from A to B and back without --force. Each direction needs its own link, so each cluster runs receivers for the other. Give each cluster one site name and keep it in both links.

// The receivers run on each link's destination, so each link uses its
// destination's image.
aFacts, err := ceph.FactsOf(a)
if err != nil {
	t.Fatal(err)
}
bFacts, err := ceph.FactsOf(b)
if err != nil {
	t.Fatal(err)
}
forward, err := rbdmirror.Run(ctx, bFacts.Images.RBDAux, rbdmirror.WithConfig(rbdmirror.Config{
	Source: a, Destination: b, Pool: "volumes",
	SourceSite: "site-a", DestinationSite: "site-b",
}))
testcontainers.CleanupContainer(t, forward)
if err != nil {
	t.Fatal(err)
}
reverse, err := rbdmirror.Run(ctx, aFacts.Images.RBDAux, rbdmirror.WithConfig(rbdmirror.Config{
	Source: b, Destination: a, Pool: "volumes",
	SourceSite: "site-b", DestinationSite: "site-a",
}))
testcontainers.CleanupContainer(t, reverse)
if err != nil {
	t.Fatal(err)
}
waitTransmitPeer(ctx, t, forward, "volumes", "site-b")
waitTransmitPeer(ctx, t, reverse, "volumes", "site-a")

// forward.SourceRBD and reverse.DestinationRBD run on A;
// forward.DestinationRBD and reverse.SourceRBD run on B.
const spec = "volumes/disk"
// ... create the image on A, write, forward.EnableImage(ctx, "disk") ...
waitReplicated(ctx, t, forward, spec, spec)

// Failover A -> B. Stop A's writers first.
if _, err := forward.SourceRBD(ctx, "mirror", "image", "demote", spec); err != nil {
	t.Fatal(err)
}
promote(ctx, t, forward.DestinationRBD, spec) // retries until the demotion arrives

// B is primary. Its changes reach A through the reverse link.
// ... write on B ...
if _, err := reverse.SourceRBD(ctx, "mirror", "image", "snapshot", spec); err != nil {
	t.Fatal(err)
}
waitReplicated(ctx, t, reverse, spec, spec)

// Failback B -> A.
if _, err := reverse.SourceRBD(ctx, "mirror", "image", "demote", spec); err != nil {
	t.Fatal(err)
}
promote(ctx, t, reverse.DestinationRBD, spec)

// A replicates to B again.
// ... write on A ...
if _, err := forward.SourceRBD(ctx, "mirror", "image", "snapshot", spec); err != nil {
	t.Fatal(err)
}
waitReplicated(ctx, t, forward, spec, spec)

User snapshots taken on either side travel with the image and keep their content through both moves. With journal mode the sequence is the same without the mirror image snapshot calls; the promotion retries until the receiver has consumed the demotion marker from the journal.

When the reverse link is created, A already has a tx-only peer for B, which the forward link's daemon registered. Run turns it into rx-tx, so the forward transmission keeps working. Using different site names for the same cluster in the two links is rejected.

Recover from split-brain with resync

A forced promotion while the other side is still primary forks the image. Ceph reports split-brain, and you recover by choosing one side and resynchronizing the other. This discards the losing side's changes; it is not a merge.

const spec = "volumes/disk"
// The image is enabled and replicated from A (source) to B (destination).

// Fork it on purpose: stop B's receiver and force-promote B.
grace := 3 * time.Second
if err := link.Stop(ctx, &grace); err != nil { // the initial daemon
	t.Fatal(err)
}
out, err := link.DestinationRBD(ctx, "mirror", "pool", "info", "volumes", "--format", "json")
if err != nil {
	t.Fatal(err)
}
var info struct{ Peers []struct{ UUID string } }
if err := json.Unmarshal(out, &info); err != nil || len(info.Peers) != 1 {
	t.Fatalf("expected one peer: %s %v", out, err)
}
// In snapshot mode B needs a transmit-capable peer to create its own mirror snapshots.
if _, err := link.DestinationRBD(ctx, "mirror", "pool", "peer", "set", "volumes",
	info.Peers[0].UUID, "direction", "rx-tx"); err != nil {
	t.Fatal(err)
}
if _, err := link.DestinationRBD(ctx, "mirror", "image", "promote", "--force", spec); err != nil {
	t.Fatal(err)
}
// ... both sides are primary now; write different data on A and on B ...

// Keep A: demote B, restart its receiver and wait for the split-brain report.
if _, err := link.DestinationRBD(ctx, "mirror", "image", "demote", spec); err != nil {
	t.Fatal(err)
}
if err := link.Start(ctx); err != nil {
	t.Fatal(err)
}
poll(ctx, t, "split-brain report", func() bool {
	out, err := link.DestinationRBD(ctx, "mirror", "image", "status", spec, "--format", "json")
	var status struct{ State, Description string }
	return err == nil && json.Unmarshal(out, &status) == nil &&
		strings.Contains(status.State, "error") &&
		strings.Contains(strings.ToLower(status.Description), "split-brain")
})
if _, err := link.DestinationRBD(ctx, "mirror", "image", "resync", spec); err != nil {
	t.Fatal(err)
}
waitReplicated(ctx, t, link, spec, spec)

After the resync, B holds A's data and A's snapshots, B's own snapshots are gone, and the next mirror snapshot from A replicates normally. Snapshot mirroring never protects writes made after the latest checkpoint, and nothing in the module fails over automatically.

Remove and re-create the receiving peer

// Remove the destination's receiving peer while the receiver is stopped.
grace := 3 * time.Second
if err := link.Stop(ctx, &grace); err != nil {
	t.Fatal(err)
}
if _, err := link.DestinationRBD(ctx, "mirror", "pool", "peer", "remove", "volumes", peerUUID); err != nil {
	t.Fatal(err)
}
if err := link.Start(ctx); err != nil {
	t.Fatal(err)
}
// The daemon runs, but new checkpoints no longer reach the destination.
// Existing destination data is kept.

if err := link.Rebootstrap(ctx); err != nil { // new peer with a new UUID
	t.Fatal(err)
}
if _, err := link.DestinationRBD(ctx, "mirror", "image", "resync", "volumes/disk"); err != nil {
	t.Fatal(err)
}
waitReplicated(ctx, t, link, "volumes/disk", "volumes/disk")

Configure mirroring before starting receivers

NoInitialDaemons sets up policies and the peer but starts no daemon, so you can prepare data on the source and choose when replication begins.

destinationFacts, err := ceph.FactsOf(destination)
if err != nil {
	t.Fatal(err)
}
link, err := rbdmirror.Run(ctx, destinationFacts.Images.RBDAux, rbdmirror.WithConfig(rbdmirror.Config{
	Source: source, Destination: destination, Pool: "volumes",
	NoInitialDaemons: true,
}))
testcontainers.CleanupContainer(t, link)
if err != nil {
	t.Fatal(err)
}
status, err := link.ReceiverStatus(ctx) // Ready == false, no daemons
if err != nil {
	t.Fatal(err)
}
_ = status

// ... create images and write on the source; nothing replicates yet ...

if _, err := link.AddDaemon(ctx, "receiver"); err != nil {
	t.Fatal(err) // a non-nil daemon returned with the error is still owned by link
}
if _, err := link.WaitReceiverReady(ctx); err != nil {
	t.Fatal(err)
}
waitTransmitPeer(ctx, t, link, "volumes", "destination")
if err := link.EnableImage(ctx, "disk"); err != nil {
	t.Fatal(err)
}
if _, err := link.SourceRBD(ctx, "mirror", "image", "snapshot", "volumes/disk"); err != nil {
	t.Fatal(err)
}
waitReplicated(ctx, t, link, "volumes/disk", "volumes/disk")

The link still owns its two setup containers. Daemons() is empty and the embedded container is nil until the first AddDaemon. A freshly configured link differs from one whose daemons ran and were removed: the source does not know the destination as a transmit peer until a daemon has connected, which is why the example waits for it before enabling the image. Pool scope with journaled images needs neither EnableImage nor a mirror snapshot. The option controls only this link's daemons; it does not prove that no other rbd-mirror serves the pool.

Run several receivers and survive the loss of one

destinationFacts, err := ceph.FactsOf(destination)
if err != nil {
	t.Fatal(err)
}
link, err := rbdmirror.Run(ctx, destinationFacts.Images.RBDAux, rbdmirror.WithConfig(rbdmirror.Config{
	Source: source, Destination: destination, Pool: "volumes",
	DaemonCount: 2, // daemons "a" and "b"
}))
testcontainers.CleanupContainer(t, link)
if err != nil {
	t.Fatal(err)
}
status, err := link.WaitReceiverReady(ctx)
if err != nil {
	t.Fatal(err)
}

// Stop the leader and wait until the survivor leads alone.
survivor := "a"
if status.LeaderDaemonName == "a" {
	survivor = "b"
}
grace := 3 * time.Second
for _, d := range link.Daemons() {
	if d.DaemonName == status.LeaderDaemonName {
		if err := d.Stop(ctx, &grace); err != nil {
			t.Fatal(err)
		}
	}
}
if _, err := link.WaitReceiverReady(ctx, survivor); err != nil {
	t.Fatal(err)
}

Without names, the stopped daemon stays expected and the wait never succeeds. While the stopped member is still in the native election, the wait keeps polling until Ceph drops it. Add a replacement with AddDaemon and start a new wait afterwards; an ongoing wait does not adopt new daemons.

Mirror several namespace pairs with one set of daemons

One link owns the daemons and the peer for a pool. Additional namespace pairs in the same pool are configured with native commands and observed through BindNamespace. Different source and destination namespace names need Ceph 20.

// The namespaces exist on both clusters (rbd.CreateNamespace on each).
destinationFacts, err := ceph.FactsOf(destination)
if err != nil {
	t.Fatal(err)
}
owner, err := rbdmirror.Run(ctx, destinationFacts.Images.RBDAux, rbdmirror.WithConfig(rbdmirror.Config{
	Source: source, Destination: destination, Pool: "volumes",
	SourceNamespace: "app", DestinationNamespace: "app-replica",
}))
testcontainers.CleanupContainer(t, owner)
if err != nil {
	t.Fatal(err)
}

// A second pair with the same site names and an explicit remote namespace.
if _, err := owner.SourceRBD(ctx, "mirror", "pool", "enable", "--site-name", "source",
	"volumes/batch", "image", "--remote-namespace", "batch-replica"); err != nil {
	t.Fatal(err)
}
if _, err := owner.DestinationRBD(ctx, "mirror", "pool", "enable", "--site-name", "destination",
	"volumes/batch-replica", "image", "--remote-namespace", "batch"); err != nil {
	t.Fatal(err)
}

view, err := owner.BindNamespace(ctx, "batch", "batch-replica")
if err != nil {
	t.Fatal(err)
}
if _, err := view.WaitReceiverReady(ctx); err != nil {
	t.Fatal(err)
}
_, _ = owner.SourceRBD(ctx, "create", "--size", "8M", "volumes/batch/disk")
_, _ = owner.SourceRBD(ctx, "mirror", "image", "enable", "volumes/batch/disk", "snapshot")
if _, err := view.WaitReplayReady(ctx, "disk"); err != nil {
	t.Fatal(err)
}
waitReplicated(ctx, t, owner, "volumes/batch/disk", "volumes/batch-replica/disk")

The two native commands are your own writes; they can leave partial configuration if the second one fails, and nothing rolls them back. BindNamespace only reads. Separate views can become ready at different times; their results are not one atomic observation.

References

Home

Clone this wiki locally