RGW Multisite

RGW multisite

rgw/multisite.Run joins two fresh clusters into one realm with a zone each and returns once both zones have committed the same period. Call WaitSyncReady before relying on replicated state. The SUT writes through one zone's gateway; the test checks what reaches the other. Master failover, interrupted links and sync policies 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

Multisite needs fresh clusters without gateways; the fixture starts the gateways itself. The scenarios use these helpers:

func startCluster(t *testing.T, opts ...testcontainers.ContainerCustomizer) *ceph.Container {
	t.Helper()
	cluster, err := ceph.Run(t.Context(), ceph.DefaultImage, opts...)
	testcontainers.CleanupContainer(t, cluster)
	if err != nil {
		t.Fatal(err)
	}
	return cluster
}

func newS3Client(endpoint, region, accessKey, secretKey string) *s3.Client {
	return s3.New(s3.Options{
		BaseEndpoint: aws.String(endpoint),
		Region:       region,
		Credentials:  credentials.NewStaticCredentialsProvider(accessKey, secretKey, ""),
		UsePathStyle: true,
	})
}

// eventually polls check until it returns true or the timeout expires.
func eventually(t *testing.T, timeout time.Duration, check func() bool) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for !check() {
		if time.Now().After(deadline) {
			t.Fatal("condition not met before the deadline")
		}
		time.Sleep(time.Second)
	}
}

Replicate between two clusters with multisite

source := startCluster(t)
destination := startCluster(t)
ms, err := multisite.Run(ctx, ceph.DefaultImage, multisite.WithConfig(multisite.Config{
	Source: source, Destination: destination,
	SourceZone: "east", DestinationZone: "west",
}))
if ms != nil {
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
		defer cancel()
		if err := ms.Terminate(ctx); err != nil {
			t.Error(err)
		}
	})
}
if err != nil {
	t.Fatal(err)
}

eastEndpoint, err := ms.Source.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}
westEndpoint, err := ms.Destination.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}
east := newS3Client(eastEndpoint, ms.Source.Region, ms.Source.AccessKey, ms.Source.SecretKey)
west := newS3Client(westEndpoint, ms.Destination.Region, ms.Destination.AccessKey, ms.Destination.SecretKey)

// The SUT writes through east; the test waits for the object in west.
if _, err := east.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String("shared")}); err != nil {
	t.Fatal(err)
}
if _, err := east.PutObject(ctx, &s3.PutObjectInput{
	Bucket: aws.String("shared"), Key: aws.String("from-east"), Body: strings.NewReader("payload"),
}); err != nil {
	t.Fatal(err)
}
eventually(t, 4*time.Minute, func() bool {
	_, err := west.HeadObject(ctx, &s3.HeadObjectInput{
		Bucket: aws.String("shared"), Key: aws.String("from-east"),
	})
	return err == nil
})

Both zones accept writes, and deletes replicate too, so a replica is not a backup. Users and buckets are realm metadata owned by the metadata master (east here). Create users through the master (ms.Source.CreateUser or ms.ZoneAdmin(ctx, "east", ...)); a bucket created through another zone's gateway is forwarded to the master, so the master must be reachable. Replication is asynchronous: allow minutes rather than seconds, especially after a gateway restart.

To test a zone outage, stop one gateway (ms.Source.Stop(ctx, &timeout)), keep working through the other one, start it again (ms.Source.Start(ctx)), resolve S3Endpoint again, and wait for catch-up with WaitSyncReady.

Planned metadata master failover and failback

Promotion uses native commands; the fixture moves periods between clusters. On Ceph 20.2.4, run period update and period commit as separate commands: the combined period update --commit can see an empty sync status on the new master and refuse the promotion. Do not use --yes-i-really-mean-it to skip the check.

restart := func(g *rgw.Gateway) {
	timeout := 3 * time.Second
	if err := g.Stop(ctx, &timeout); err != nil {
		t.Fatal(err)
	}
	if err := g.Start(ctx); err != nil {
		t.Fatal(err)
	}
	// Resolve g.S3Endpoint again and wait until signed S3 requests succeed.
}

// 1. The destination must have all metadata, and nothing may write meanwhile.
if _, err := ms.WaitSyncReady(ctx, "west"); err != nil {
	t.Fatal(err)
}
// 2. Fence the old master.
timeout := 3 * time.Second
if err := ms.Source.Stop(ctx, &timeout); err != nil {
	t.Fatal(err)
}
// 3. Promote the destination.
for _, args := range [][]string{
	{"zone", "modify", "--master", "--default"},
	{"zonegroup", "modify", "--master"},
	{"period", "update"},
	{"period", "commit"},
} {
	if _, err := ms.ZoneAdmin(ctx, "west", args...); err != nil {
		t.Fatal(err)
	}
}
restart(ms.Destination)
// New users and buckets are now created through west.

// 4. Bring back the old master as a secondary.
if err := ms.PullSourcePeriod(ctx); err != nil {
	t.Fatal(err)
}
if err := ms.Source.Start(ctx); err != nil {
	t.Fatal(err)
}

// Failback is the same sequence in the other direction: wait for metadata in
// east, fence west, promote with ZoneAdmin(ctx, "east", ...), restart east,
// PullDestinationPeriod, start west.

Verify the current master with period get (master_zone equals the zone ID). Fixture.Source and Destination keep pointing at the original zones. Automatic failover, request routing and recovery point guarantees are outside the fixture.

Interrupt replication between zones

In bridge mode a zone can be cut off from the private multisite network while its S3 endpoint stays up:

cut, err := ms.InterruptZoneLink(ctx, "west")
if err != nil {
	if cut != nil {
		_ = cut.Restore(context.WithoutCancel(ctx))
	}
	t.Fatal(err)
}
// Writes in east and west succeed locally but do not replicate.
if err := cut.Restore(ctx); err != nil {
	t.Fatal(err)
}
if _, err := ms.WaitSyncReady(ctx, "west", "east"); err != nil {
	t.Fatal(err)
}
// Now check the objects written during the interruption in both zones.

InterruptZoneLink cuts the zone's whole link to the multisite bridge. To delay or drop only the traffic one zone sends to one peer, use ms.ApplyZoneLinkProfile(ctx, "west", "east", profile) with a fault.NetworkProfile that names the opt-in netem helper (fault.NetemHelperImage); the peer keeps syncing in the other direction, and restoring the returned lease ends the fault. See guides/fault-injection.md.

More zones and zonegroups

// One zonegroup, three zones, "a" is the metadata master.
topology, err := multisite.Run(ctx, ceph.DefaultImage, multisite.WithTopology(multisite.TopologyConfig{
	Realm: "application", MetadataMaster: "a",
	Zones: []multisite.ZoneConfig{
		{Name: "a", Cluster: clusterA},
		{Name: "b", Cluster: clusterB},
		{Name: "c", Cluster: clusterC},
	},
}))
if err != nil {
	t.Fatal(err)
}
for _, zone := range topology.Zones() {
	endpoint, err := zone.Gateway.S3Endpoint(ctx)
	if err != nil {
		t.Fatal(err)
	}
	// S3 client per zone with zone.Gateway.AccessKey, SecretKey and Region.
	_ = newS3Client(endpoint, zone.Gateway.Region, zone.Gateway.AccessKey, zone.Gateway.SecretKey)
}

// Two zonegroups. Users and bucket metadata are shared; object data
// replicates only inside each zonegroup.
config := multisite.TopologyConfig{
	Realm: "application", MasterZonegroup: "us",
	Zonegroups: []multisite.ZonegroupConfig{
		{Name: "us", MasterZone: "a", Zones: []multisite.ZoneConfig{
			{Name: "a", Cluster: clusterA},
			{Name: "b", Cluster: clusterB},
		}},
		{Name: "eu", MasterZone: "c", Zones: []multisite.ZoneConfig{
			{Name: "c", Cluster: clusterC},
		}},
	},
}
_ = config // start it with multisite.Run(ctx, img, multisite.WithTopology(config))

// Grow and shrink a running topology.
zone, err := topology.AddZone(ctx, ceph.DefaultImage, multisite.ZoneConfig{Name: "d", Cluster: clusterD})
if err != nil {
	t.Fatal(err)
}
t.Log("added zone", zone.Name)
if err := topology.RemoveZone(ctx, "b"); err != nil {
	t.Fatal(err)
}
// Resolve every remaining gateway's S3Endpoint again after both calls.

Selective sync with a sync policy

This recipe replicates only objects under published/ in one bucket, from east to west. A zonegroup group with status allowed permits replication without activating it; once it exists, only bucket groups with status enabled replicate objects. Bucket metadata still replicates, so the bucket exists in both zones; only its objects are filtered.

// ms from multisite.Run with SourceZone "east" and DestinationZone "west".

// 1. Zonegroup permission.
permit, err := ms.CreateSyncGroup(ctx, multisite.SyncPolicyScope{},
	multisite.SyncGroupConfig{ID: "permit", Status: multisite.SyncAllowed})
if err != nil {
	t.Fatal(err)
}
if err := permit.CreateFlow(ctx, multisite.SyncFlowConfig{
	ID: "pair", Zones: []string{"east", "west"},
}); err != nil {
	t.Fatal(err)
}
if err := permit.CreatePipe(ctx, multisite.SyncPipeConfig{
	ID: "east-to-west", SourceZones: []string{"east"}, DestinationZones: []string{"west"},
}); err != nil {
	t.Fatal(err)
}
if err := permit.Apply(ctx); err != nil { // reloads the gateways
	t.Fatal(err)
}
if _, err := ms.WaitSyncReady(ctx, "west", "east"); err != nil {
	t.Fatal(err)
}
// Resolve both S3 endpoints again and rebuild the clients.

// 2. Create the bucket through east (the metadata master), then the bucket policy.
selected, err := ms.CreateSyncGroup(ctx, multisite.SyncPolicyScope{Bucket: "app-data"},
	multisite.SyncGroupConfig{ID: "published", Status: multisite.SyncEnabled})
if err != nil {
	t.Fatal(err)
}
if err := selected.CreatePipe(ctx, multisite.SyncPipeConfig{
	ID: "published", SourceZones: []string{"east"}, DestinationZones: []string{"west"},
	Prefix: "published/",
}); err != nil {
	t.Fatal(err)
}

// 3. Wait until west has imported the bucket policy, then write.
if _, err := selected.WaitBucketPolicyReady(ctx, "west"); err != nil {
	t.Fatal(err)
}
// Write app-data/published/report and app-data/private/notes through east.

// 4. Wait for the selected object, then for the bucket checkpoint.
eventually(t, 4*time.Minute, func() bool {
	_, err := west.HeadObject(ctx, &s3.HeadObjectInput{
		Bucket: aws.String("app-data"), Key: aws.String("published/report"),
	})
	return err == nil
})
status, err := selected.WaitBucketSyncReady(ctx, "published", "east", "west")
if err != nil {
	t.Fatal(err)
}
_ = status // State "incremental", CaughtUp true

// 5. Excluded objects must stay absent. Check over a window (for example
// 35 seconds), not once.

Variations:

• Tags: Tags: []multisite.SyncObjectTag{{Key: "class", Value: "public"}} selects objects carrying that tag. The S3 client sets object tags.
• Different destination bucket: SourceBucket and DestinationBucket with concrete names. Both buckets must exist; their instance IDs are captured.
• Destination owner: DestinationOwner: someUser makes replicated objects owned by an ordinary user created with CreateUser.
• Storage class: DestinationStorageClass: "STANDARD_IA" with DestinationPlacements: map[string]*rgw.Placement{"west": westPlacement}. Create, publish, import and reload the placement in both zones first (see RGW S3, "Placement targets and storage classes"). Check the actual class in the S3 listing and the data in the destination pool, not only the stored policy.
• User mode: User: replicator replicates with that user's permissions. The user needs both s3:GetObject and s3:GetObjectAcl on the source objects. On the default image, ordinary-user permissions are not enforced (see the multisite package documentation). With an account root user, a destination bucket policy that denies s3:ReplicateObject blocks replication while ordinary S3 requests keep working.
• Changing the prefix: selected.SetPipePrefix(ctx, "published", "reports/"), then WaitBucketPolicyReady before writing.
• Removing: Remove for the bucket group (already copied objects stay); Remove plus Apply for the zonegroup group.

Pause and resume a bucket policy

Setting a bucket group to forbidden stops replication of new writes. After switching back to enabled, new objects replicate again, but log entries recorded while the policy was forbidden may remain unprocessed. Replay them explicitly on a quiet fixture:

if err := selected.SetStatus(ctx, multisite.SyncForbidden); err != nil {
	t.Fatal(err)
}
if _, err := selected.WaitBucketPolicyReady(ctx, "west"); err != nil {
	t.Fatal(err)
}
// BucketSyncStatus now reports State "stopped" or "disabled_or_no_sources".
// Writes to published/ are not replicated.

if err := selected.SetStatus(ctx, multisite.SyncEnabled); err != nil {
	t.Fatal(err)
}
if _, err := selected.WaitBucketPolicyReady(ctx, "west"); err != nil {
	t.Fatal(err)
}

// Replay the existing bucket logs with the current policy.
before, err := selected.BucketSyncStatus(ctx, "published", "east", "west")
if err != nil || before.State != "incremental" {
	t.Fatalf("state %q: %v", before.State, err)
}
args := []string{"bucket", "sync", "run",
	"--bucket", before.DestinationBucket.Name, "--bucket-id", before.DestinationBucket.ID,
	"--source-zone", before.SourceZoneID,
	"--source-bucket", before.SourceBucket.Name, "--source-bucket-id", before.SourceBucket.ID,
}
if _, err := ms.ZoneAdmin(ctx, before.Zone, args...); err != nil {
	t.Fatal(err)
}
if _, err := selected.WaitBucketSyncReady(ctx, "published", "east", "west"); err != nil {
	t.Fatal(err)
}

• For tenant buckets, pass tenant/name as --bucket and add --source-tenant.
• Pass the captured native zone ID with --source-zone. On Ceph 20.2.4, --source-zone-id is a sync-policy option and does not select the source for bucket sync run.
• The replay processes the existing log entries of all matching pipes, including deletes, so objects written while the policy was forbidden may replicate now.
• Do not use bucket sync init, marker edits, log trims or forced unlocks. An exit status of 0 in the stopped state can be a no-op and proves nothing.
• Canceling the Docker exec does not stop a running native worker. Do not retry an uncertain replay automatically; discard the fixture instead.

References

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