RGW S3

rgw.Run starts a cluster with a gateway, or rgw.Start adds gateways to an existing cluster after the test has stored the RGW options it needs. The SUT gets the gateway's S3 endpoint and the credentials of a user the test created; the test checks buckets, objects, quotas or usage 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.

Helpers

The scenarios share a few helpers:

import (
	"context"
	"testing"
	"time"

	"github.com/JSYoo5B/ceph-testcontainers-go/ceph"
	"github.com/JSYoo5B/ceph-testcontainers-go/override"
	"github.com/JSYoo5B/ceph-testcontainers-go/rgw"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/testcontainers/testcontainers-go"
)

// startCluster starts a cluster without gateways, so that RGW options can be
// applied before a gateway starts.
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
}

// applyConfig stores central configuration and restores it at the end of the
// test. Restoring is optional when the cluster is discarded anyway. The handle
// is never nil and stays tracked after a failed write, so the cleanup is
// registered before the error is checked.
func applyConfig(t *testing.T, cluster *ceph.Container, settings ...ceph.ConfigSetting) {
	t.Helper()
	for _, setting := range settings {
		change, err := override.SetConfig(t.Context(), cluster, setting)
		t.Cleanup(func() {
			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
			defer cancel()
			if err := change.Restore(ctx); err != nil {
				t.Error(err)
			}
		})
		if err != nil {
			t.Fatal(err)
		}
	}
}

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)
	}
}

RGW runs as client.admin, so options go into the client.admin section (or client). Apply them before rgw.Start. An alternative for options that never change during a test is ceph.WithConfigFile with a [client.admin] section.

Test an S3 client with the AWS SDK for Go

The default gateway user is enough for a single principal. Create one user per principal when the test checks permissions between users:

cluster := startCluster(t, ceph.WithClusterConfig(ceph.ClusterConfig{}.WithOSDCount(1)))
gateway, err := rgw.Start(ctx, cluster, rgw.Config{SkipUserCreation: true})
if err != nil {
	t.Fatal(err)
}
endpoint, err := gateway.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}

clients := map[string]*s3.Client{}
for _, id := range []string{"owner", "reader"} {
	user, err := gateway.CreateUser(ctx, rgw.UserConfig{ID: id})
	if err != nil {
		t.Fatal(err)
	}
	accessKey, secretKey, err := user.Credentials()
	if err != nil {
		t.Fatal(err)
	}
	clients[id] = newS3Client(endpoint, gateway.Region, accessKey, secretKey)
}
// clients["owner"] creates a bucket and a bucket policy;
// clients["reader"] checks access denied, then granted.

Notes:

Always use path-style addressing. Virtual-hosted-style bucket names would need DNS for every bucket.
Versioning, multipart uploads, ACLs, bucket policies and object lock need no server preparation. Object lock must be enabled when the bucket is created (ObjectLockEnabledForBucket: aws.Bool(true)).
Tenant users address buckets of their own tenant by plain name. Path-style requests reach a bucket in another tenant as tenant:bucket.
Recent AWS SDK releases add CRC checksums to uploads by default. If a gateway rejects such requests, set RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired and ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired in s3.Options.

Run several gateways in one zone

Gateways of one cluster serve the same data. This tests client failover between endpoints, or credentials shared by several gateways:

cluster, err := rgw.Run(ctx, ceph.DefaultImage,
	rgw.WithGateways(
		rgw.Config{Name: "a"},
		rgw.Config{Name: "b", SkipUserCreation: true},
	),
)
testcontainers.CleanupContainer(t, cluster)
if err != nil {
	t.Fatal(err)
}
gateways, err := rgw.Gateways(ctx, cluster.Container)
if err != nil {
	t.Fatal(err)
}
// gateways is sorted: a, b
a, b := gateways[0], gateways[1]
// Write through a with a's test user, read through b with the same keys.

// Remove a; b keeps serving users, buckets and objects.
if err := a.Remove(ctx); err != nil {
	t.Fatal(err)
}
endpoint, err := b.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}
// Read the objects written through a from endpoint with a's keys; b skipped
// user creation and has none of its own.
_ = newS3Client(endpoint, b.Region, a.AccessKey, a.SecretKey)

Users, tenants and accounts

gateway, err := rgw.Start(ctx, cluster, rgw.Config{SkipUserCreation: true})
if err != nil {
	t.Fatal(err)
}

// The same local ID and bucket names in two tenants stay isolated.
alpha, err := gateway.CreateUser(ctx, rgw.UserConfig{ID: "app", Tenant: "alpha"})
if err != nil {
	t.Fatal(err)
}
beta, err := gateway.CreateUser(ctx, rgw.UserConfig{ID: "app", Tenant: "beta"})
if err != nil {
	t.Fatal(err)
}
if alpha.ID() != "alpha$app" || beta.ID() != "beta$app" {
	t.Fatalf("tenant IDs: %s, %s", alpha.ID(), beta.ID())
}

// An account with two root users and a shared quota.
account, err := gateway.CreateAccount(ctx, rgw.AccountConfig{Name: "team"})
if err != nil {
	t.Fatal(err)
}
rootA, err := account.CreateRootUser(ctx, rgw.UserConfig{ID: "root-a"})
if err != nil {
	t.Fatal(err)
}
rootB, err := account.CreateRootUser(ctx, rgw.UserConfig{ID: "root-b"})
if err != nil {
	t.Fatal(err)
}
err = account.SetQuota(ctx, rgw.Quota{
	Enabled: true, MaxSizeBytes: 1 << 20, MaxObjects: -1,
})
// Objects written by rootA count against the same quota as rootB's.
usage, err := rootB.Usage(ctx) // Scope "account": includes rootA's objects
if err != nil {
	t.Fatal(err)
}
t.Logf("account usage: %+v", usage)

// Clean up in dependency order: objects and buckets through S3, then users,
// then the account.
if err := rootA.Remove(ctx); err != nil {
	t.Fatal(err)
}
if err := rootB.Remove(ctx); err != nil {
	t.Fatal(err)
}
if err := account.Remove(ctx); err != nil {
	t.Fatal(err)
}

Make lifecycle tests fast

Lifecycle rules count in days. rgw_lc_debug_interval makes one lifecycle day last the given number of seconds. Disabling the background lifecycle threads and processing the bucket explicitly makes the test deterministic:

cluster := startCluster(t, ceph.WithClusterConfig(ceph.ClusterConfig{}.WithOSDCount(1)))
applyConfig(t, cluster,
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_lc_debug_interval", Value: "2"},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_enable_lc_threads", Value: "false"},
)
gateway, err := rgw.Start(ctx, cluster, rgw.Config{})
if err != nil {
	t.Fatal(err)
}
endpoint, err := gateway.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}
client := newS3Client(endpoint, gateway.Region, gateway.AccessKey, gateway.SecretKey)

// Create bucket "logs", a rule expiring "tmp/" after 1 day, and objects
// "tmp/a" and "keep/b" with the S3 client.

time.Sleep(3 * time.Second) // more than one lifecycle "day"
if _, err := gateway.Admin(ctx, "lc", "process", "--bucket", "logs"); err != nil {
	t.Fatal(err)
}
eventually(t, 30*time.Second, func() bool {
	_, err := client.HeadObject(ctx, &s3.HeadObjectInput{
		Bucket: aws.String("logs"), Key: aws.String("tmp/a"),
	})
	return err != nil // NotFound once expired
})
// "keep/b" must still be readable.

With the background threads left enabled, the gateway processes lifecycle on its own schedule and the timing is less predictable. This tests the S3 contract with an accelerated clock, not real day boundaries.

Bucket notifications to an HTTP endpoint

RGW pushes bucket notifications to the endpoint of an SNS-compatible topic. The test needs a gateway, a user and an HTTP receiver that the gateway can reach. Run the receiver in a container attached to the cluster:

const receiverScript = `import http.server, json, sys
class Handler(http.server.BaseHTTPRequestHandler):
    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        with open("/tmp/events.log", "a") as log:
            for record in json.loads(body).get("Records", []):
                log.write(record["eventName"] + " " + record["s3"]["object"]["key"] + "\n")
        self.send_response(200)
        self.end_headers()
http.server.HTTPServer(("0.0.0.0", int(sys.argv[1])), Handler).serve_forever()
`

cluster := startCluster(t, ceph.WithClusterConfig(ceph.ClusterConfig{}.WithOSDCount(1)))
gateway, err := rgw.Start(ctx, cluster, rgw.Config{SkipUserCreation: true})
if err != nil {
	t.Fatal(err)
}
user, err := gateway.CreateUser(ctx, rgw.UserConfig{ID: "notifier"})
if err != nil {
	t.Fatal(err)
}
accessKey, secretKey, err := user.Credentials()
if err != nil {
	t.Fatal(err)
}
endpoint, err := gateway.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}

// The default image has Python. WithIdleEntrypoint runs under Docker's init,
// which reaps background processes started with Exec.
receiver, err := testcontainers.Run(ctx, ceph.DefaultImage,
	cluster.WithClient(), ceph.WithIdleEntrypoint())
testcontainers.CleanupContainer(t, receiver)
if err != nil {
	t.Fatal(err)
}
if err := receiver.CopyToContainer(ctx, []byte(receiverScript), "/tmp/receiver.py", 0o644); err != nil {
	t.Fatal(err)
}
if _, _, err := receiver.Exec(ctx, []string{"sh", "-c",
	"nohup python3 /tmp/receiver.py 8080 >/tmp/receiver.out 2>&1 &"}); err != nil {
	t.Fatal(err)
}

// The address the gateway uses to reach the receiver.
facts, err := ceph.FactsOf(cluster)
if err != nil {
	t.Fatal(err)
}
host := facts.PublicAddress // host mode
if !facts.UsesHostNetwork {
	if host, err = receiver.ContainerIP(ctx); err != nil {
		t.Fatal(err)
	}
}
pushEndpoint := "http://" + net.JoinHostPort(host, "8080")

topics := sns.New(sns.Options{
	BaseEndpoint: aws.String(endpoint),
	Region:       gateway.Region,
	Credentials:  credentials.NewStaticCredentialsProvider(accessKey, secretKey, ""),
})
topic, err := topics.CreateTopic(ctx, &sns.CreateTopicInput{
	Name:       aws.String("uploads"),
	Attributes: map[string]string{"push-endpoint": pushEndpoint},
})
if err != nil {
	t.Fatal(err)
}

client := newS3Client(endpoint, gateway.Region, accessKey, secretKey)
// Create bucket "inbox", then:
if _, err := client.PutBucketNotificationConfiguration(ctx, &s3.PutBucketNotificationConfigurationInput{
	Bucket: aws.String("inbox"),
	NotificationConfiguration: &types.NotificationConfiguration{
		TopicConfigurations: []types.TopicConfiguration{{
			Id:       aws.String("incoming"),
			TopicArn: topic.TopicArn,
			Events:   []types.Event{"s3:ObjectCreated:*", "s3:ObjectRemoved:*"},
			Filter: &types.NotificationConfigurationFilter{Key: &types.S3KeyFilter{
				FilterRules: []types.FilterRule{{
					Name: types.FilterRuleNamePrefix, Value: aws.String("in/"),
				}},
			}},
		}},
	},
}); err != nil {
	t.Fatal(err)
}
// Write "in/a" and "out/b", delete "in/a", then poll /tmp/events.log with
// receiver.Exec until it contains "ObjectCreated:Put in/a" and
// "ObjectRemoved:Delete in/a". "out/b" is outside the filter.

Notes:

The topic ARN has the form arn:aws:sns:<zonegroup>::<name>.
In host mode the receiver binds a port on the Docker host; choose a free one.
Add the topic attributes persistent=true and, optionally, retry_sleep_duration=<seconds> for persistent delivery. RGW then queues events and delivers them asynchronously, so S3 writes do not wait for the receiver, and events written while the receiver is down arrive after it restarts on the same address.
The topic API is a form-encoded POST / on the gateway endpoint. Any SigV4-signed client works; the SNS client of the AWS SDK is the simplest.
Remove notification rules with the RGW extension DELETE /<bucket>?notification and topics with SNS DeleteTopic if the test reuses the bucket.

See the Ceph bucket notifications documentation for event types and topic attributes.

STS AssumeRole

STS issues temporary credentials for roles. Every gateway that should accept the session tokens needs the same rgw_sts_key, so set it before starting the gateways:

cluster := startCluster(t, ceph.WithClusterConfig(ceph.ClusterConfig{}.WithOSDCount(1)))

// Generate a CryptoKey for session tokens. Never log it.
control, err := cluster.ControlContainer(ctx)
if err != nil {
	t.Fatal(err)
}
_, out, err := control.Exec(ctx,
	[]string{"ceph-authtool", "--gen-print-key"}, tcexec.Multiplexed())
if err != nil {
	t.Fatal(err)
}
key, err := io.ReadAll(out)
if err != nil {
	t.Fatal(err)
}
applyConfig(t, cluster,
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_sts_key", Value: strings.TrimSpace(string(key))},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_s3_auth_use_sts", Value: "true"},
)
gateway, err := rgw.Start(ctx, cluster, rgw.Config{SkipUserCreation: true})
if err != nil {
	t.Fatal(err)
}
owner, err := gateway.CreateUser(ctx, rgw.UserConfig{ID: "owner"})
if err != nil {
	t.Fatal(err)
}
assumer, err := gateway.CreateUser(ctx, rgw.UserConfig{ID: "assumer"})
if err != nil {
	t.Fatal(err)
}

// Role with a trust policy for "assumer" and an inline permission policy.
// Admin rejects control characters, so keep JSON documents on one line.
trust := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":"arn:aws:iam:::user/assumer"},"Action":"sts:AssumeRole"}]}`
created, err := gateway.Admin(ctx, "role", "create",
	"--role-name", "reader", "--path", "/app/", "--assume-role-policy-doc", trust)
if err != nil {
	t.Fatal(err)
}
var role struct {
	ID  string `json:"RoleId"`
	ARN string `json:"Arn"`
}
if err := json.Unmarshal(created, &role); err != nil {
	t.Fatal(err)
}
permission := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"s3:GetObject","Resource":"arn:aws:s3:::shared/*"}]}`
if _, err := gateway.Admin(ctx, "role-policy", "put", "--role-name", "reader",
	"--policy-name", "read-shared", "--policy-doc", permission); err != nil {
	t.Fatal(err)
}

endpoint, err := gateway.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}
accessKey, secretKey, err := assumer.Credentials()
if err != nil {
	t.Fatal(err)
}
stsClient := sts.New(sts.Options{
	BaseEndpoint: aws.String(endpoint),
	Region:       gateway.Region,
	Credentials:  credentials.NewStaticCredentialsProvider(accessKey, secretKey, ""),
})
assumed, err := stsClient.AssumeRole(ctx, &sts.AssumeRoleInput{
	RoleArn:         aws.String(role.ARN),
	RoleSessionName: aws.String("test"),
})
if err != nil {
	t.Fatal(err)
}
c := assumed.Credentials
temporary := s3.New(s3.Options{
	BaseEndpoint: aws.String(endpoint),
	Region:       gateway.Region,
	Credentials: credentials.NewStaticCredentialsProvider(
		*c.AccessKeyId, *c.SecretAccessKey, *c.SessionToken),
	UsePathStyle: true,
})
// owner creates bucket "shared" and its objects beforehand; temporary can
// read shared/*, while writes and other keys are denied.
_, _ = owner, temporary

Notes:

RGW re-reads the role on each S3 request made with a session token, so a change to the role's permission policy (for example an explicit deny) applies to existing sessions. A trust policy change controls new AssumeRole calls; do not rely on it to revoke tokens already issued.
If you sign STS requests yourself, sign the Content-Type header and all x-amz-* headers. RGW rejects a form POST whose Content-Type is present but unsigned before it evaluates the trust policy.
To check that a role does not exist yet, use role get --role-name <name> through gateway.Exec and expect exit status 2. In a fresh zone role list can fail with ENOENT because the role metadata pool does not exist yet.
The principal of a tenant user is arn:aws:iam::<tenant>:user/<id>.
OIDC/WebIdentity, Keycloak, LDAP and Keystone need separate identity providers and are not covered here.

See the Ceph STS and role documentation.

Swift and S3 on the same gateway

The Swift API is served under /swift next to S3. With rgw_swift_account_in_url=true, storage URLs include the account (/swift/v1/AUTH_<uid>):

cluster := startCluster(t, ceph.WithClusterConfig(ceph.ClusterConfig{}.WithOSDCount(1)))
applyConfig(t, cluster,
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_swift_account_in_url", Value: "true"},
)
gateway, err := rgw.Start(ctx, cluster, rgw.Config{SkipUserCreation: true})
if err != nil {
	t.Fatal(err)
}
user, err := gateway.CreateUser(ctx, rgw.UserConfig{ID: "app"})
if err != nil {
	t.Fatal(err)
}

// A Swift subuser with its own key.
const swiftKey = "swift-test-secret"
swiftUser := user.ID() + ":swift"
if _, err := gateway.Admin(ctx, "subuser", "create", "--uid", user.ID(),
	"--subuser", swiftUser, "--access", "full",
	"--key-type", "swift", "--secret", swiftKey); err != nil {
	t.Fatal(err)
}

endpoint, err := gateway.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, endpoint+"/auth", nil)
req.Header.Set("X-Auth-User", swiftUser)
req.Header.Set("X-Auth-Key", swiftKey)
resp, err := http.DefaultClient.Do(req)
if err != nil {
	t.Fatal(err)
}
resp.Body.Close()
token := resp.Header.Get("X-Auth-Token")
storage, err := url.Parse(resp.Header.Get("X-Storage-Url"))
if err != nil {
	t.Fatal(err)
}
// Use endpoint + storage.Path for container and object requests with the
// X-Auth-Token header. The host in X-Storage-Url may be an internal address.
list, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint+storage.Path, nil)
if err != nil {
	t.Fatal(err)
}
list.Header.Set("X-Auth-Token", token)
listed, err := http.DefaultClient.Do(list)
if err != nil {
	t.Fatal(err)
}
listed.Body.Close() // 200 or 204 with the account's containers

Objects written through Swift are readable through S3 with the user's S3 keys, and the other way round. subuser rm --subuser <uid>:swift --purge-keys revokes the Swift key; tokens issued earlier stop working.

Ceph's default rgw_enable_apis already includes the S3, Swift, STS, IAM and admin APIs. If you set the option, list every API the test uses (for example s3,swift,swift_auth,sts,iam,admin,notifications). Moving Swift to the root prefix (rgw_swift_url_prefix=/) cannot be combined with S3 on the same gateway.

Server-side encryption with Vault

SSE-KMS needs an external key management service. HashiCorp Vault's KV v2 engine works with the Testcontainers Vault module. Vault is a test service of its own; it is not part of the Ceph images.

Order: start the cluster, start Vault where the gateway can reach it, prepare the key and a token, apply the RGW options, start the gateway with the token file.

Start Vault with vault.Run(ctx, "hashicorp/vault:1.21.4", vault.WithToken(rootToken), ...). In bridge mode attach it to the cluster network: read facts, err := ceph.FactsOf(cluster) and pass network.WithNetworkName(nil, facts.NetworkName). In host mode keep it on a Docker bridge (network.WithBridgeNetwork()); the host-network gateway reaches the container IP on the same Docker host. Register testcontainers.CleanupContainer for it.
Through the Vault HTTP API (HttpHostAddress) with the root token: mount a KV v2 engine at tc, store a base64-encoded 256-bit key at tc/data/<key-id> as {"data":{"key":"<base64>"}}, write a policy that only reads that path, and create a token with that policy. Do not give the root token to RGW.
Apply the RGW options and start the gateway with the token file:

// vaultContainer is the Vault container from step 1, scopedToken the token
// from step 2.
vaultIP, err := vaultContainer.ContainerIP(ctx)
if err != nil {
	t.Fatal(err)
}
applyConfig(t, cluster,
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_crypt_s3_kms_backend", Value: "vault"},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_crypt_vault_auth", Value: "token"},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_crypt_vault_token_file", Value: "/tc/vault-token"},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_crypt_vault_addr", Value: "http://" + net.JoinHostPort(vaultIP, "8200")},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_crypt_vault_prefix", Value: "/v1/tc/data"},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_crypt_vault_secret_engine", Value: "kv"},
	// Only because this recipe uses plain HTTP to the gateway.
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_crypt_require_ssl", Value: "false"},
)
gateway, err := rgw.Start(ctx, cluster, rgw.Config{},
	testcontainers.WithFiles(testcontainers.ContainerFile{
		Reader:            strings.NewReader(scopedToken),
		ContainerFilePath: "/tc/vault-token",
		FileMode:          0o600,
	}))

endpoint, err := gateway.S3Endpoint(ctx)
if err != nil {
	t.Fatal(err)
}
client := newS3Client(endpoint, gateway.Region, gateway.AccessKey, gateway.SecretKey)

// Upload with SSE-KMS and the key ID stored in Vault (bucket "secure" exists).
payload := []byte("secret document")
if _, err := client.PutObject(ctx, &s3.PutObjectInput{
	Bucket:               aws.String("secure"),
	Key:                  aws.String("doc"),
	Body:                 bytes.NewReader(payload),
	ServerSideEncryption: types.ServerSideEncryptionAwsKms,
	SSEKMSKeyId:          aws.String("allowed"),
}); err != nil {
	t.Fatal(err)
}

Every gateway that serves encrypted objects needs its own copy of the token file. A key ID that the token's policy does not allow makes the upload fail and leaves no object. Deleting the key in Vault makes reads fail with 404 NoSuchKey while the listing still shows the object; writing the same key material back as a new KV version makes the object readable again.

Production Vault features (HA, seal/unseal, agent token renewal, Transit, TLS to Vault) and other KMS backends need their own setup. See Ceph's Vault integration.

Usage logs and Admin Ops

The usage log records dated request and byte counters per user and category. It is disabled by default. Enable it before the gateway starts and shorten the flush settings:

cluster := startCluster(t, ceph.WithClusterConfig(ceph.ClusterConfig{}.WithOSDCount(1)))
applyConfig(t, cluster,
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_enable_usage_log", Value: "true"},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_usage_log_tick_interval", Value: "1"},
	ceph.ConfigSetting{Section: "client.admin", Name: "rgw_usage_log_flush_threshold", Value: "1"},
)
gateway, err := rgw.Start(ctx, cluster, rgw.Config{SkipUserCreation: true})
if err != nil {
	t.Fatal(err)
}
producer, err := gateway.CreateUser(ctx, rgw.UserConfig{ID: "producer"})
if err != nil {
	t.Fatal(err)
}
operator, err := gateway.CreateUser(ctx, rgw.UserConfig{
	ID: "operator", AdminCaps: "usage=read;ratelimit=read",
})
if err != nil {
	t.Fatal(err)
}
// producer generates S3 traffic; operator reads its usage below.
_, _ = producer, operator

Admin Ops requests are ordinary SigV4 requests (service name s3) signed with the keys of a user that has the needed capability. A small helper with the AWS SDK signer:

import v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"

func adminRequest(ctx context.Context, method, endpoint, pathAndQuery, region string, user *rgw.User) (*http.Response, error) {
	accessKey, secretKey, err := user.Credentials()
	if err != nil {
		return nil, err
	}
	req, err := http.NewRequestWithContext(ctx, method, endpoint+pathAndQuery, nil)
	if err != nil {
		return nil, err
	}
	const emptyHash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
	req.Header.Set("X-Amz-Content-Sha256", emptyHash)
	creds := aws.Credentials{AccessKeyID: accessKey, SecretAccessKey: secretKey}
	err = v4.NewSigner().SignHTTP(ctx, creds, req, emptyHash, "s3", region, time.Now(),
		func(o *v4.SignerOptions) { o.DisableURIPathEscaping = true })
	if err != nil {
		return nil, err
	}
	return http.DefaultClient.Do(req)
}

Read the producer's usage after its S3 traffic. Persistence is asynchronous even with the short flush settings, so poll (up to about 90 seconds):

query := url.Values{
	"uid": {producer.ID()}, "show-entries": {"true"},
	"show-summary": {"true"}, "format": {"json"},
}
resp, err := adminRequest(ctx, http.MethodGet, endpoint, "/admin/usage?"+query.Encode(),
	gateway.Region, operator)
if err != nil {
	t.Fatal(err)
}
defer resp.Body.Close()
// Look for the put_obj and get_obj categories with successful operations and
// bytes received and sent.

Request	Capability
`GET /admin/usage?uid=<uid>&show-entries=true&show-summary=true&format=json`	`usage=read`
`DELETE /admin/usage?uid=<uid>&remove-all=false&format=json`	`usage=write`
`GET /admin/ratelimit?ratelimit-scope=bucket&bucket=<name>&format=json`	`ratelimit=read`
`POST /admin/ratelimit?ratelimit-scope=bucket&bucket=<name>&...`	`ratelimit=write`

A user without the capability gets 403 AccessDenied. Add capabilities later with gateway.Admin(ctx, "caps", "add", "--uid", operator.ID(), "--caps", "usage=write;ratelimit=write") and remove them with caps rm. Capabilities grant an operation type, not access to particular users or buckets: scope each request with uid or bucket. When trimming usage, always pass the uid; never send remove-all=true or an unscoped trim, which clears the log of every user.

See the Ceph Admin Ops API.

Bucket rate limits

A bucket rate limit makes RGW answer excess requests with HTTP 503 SlowDown. It is separate from quotas. Read the current policy first, then post the limit (the operator from the previous example, with ratelimit=write added):

path := func(values url.Values) string {
	values.Set("ratelimit-scope", "bucket")
	values.Set("bucket", "limited")
	values.Set("format", "json")
	return "/admin/ratelimit?" + values.Encode()
}
// GET path(url.Values{}) returns {"bucket_ratelimit": {...}}; keep it for restore.

limit := url.Values{
	"enabled":         {"true"},
	"max-read-ops":    {"1"},
	"max-write-ops":   {"0"},
	"max-read-bytes":  {"0"},
	"max-write-bytes": {"0"},
}
resp, err := adminRequest(ctx, http.MethodPost, endpoint, path(limit), gateway.Region, operator)
if err != nil {
	t.Fatal(err)
}
resp.Body.Close()
if resp.StatusCode != http.StatusOK {
	t.Fatalf("set rate limit: %s", resp.Status)
}
// Repeated GETs of an object in "limited" soon return 503 SlowDown; other
// buckets of the same user are not affected.

On Tentacle, the limits are per-minute token buckets for read and write operations and bytes. 0 means no limit for that dimension. Newer fields such as max-list-ops and max-delete-ops do not exist in Ceph 20.
Each gateway enforces limits for the requests it serves.
To restore, post the original values of all five fields, not only enabled=false.
The same API has other ratelimit-scope values (for example user limits); see the Ceph Admin Ops documentation.

Bucket quota and resharding

User.SetBucketQuota applies to every bucket of a user. To limit one named bucket, and to reshard a bucket on demand, use Admin:

cluster := startCluster(t, ceph.WithClusterConfig(ceph.ClusterConfig{}.WithOSDCount(1)))
// Stop the background resharder from racing the explicit reshard below.
applyConfig(t, cluster, ceph.ConfigSetting{
	Section: "client.admin", Name: "rgw_dynamic_resharding", Value: "false",
})
gateway, err := rgw.Start(ctx, cluster, rgw.Config{})
if err != nil {
	t.Fatal(err)
}

// After creating "limited" and "sibling" with the S3 client:
for _, args := range [][]string{
	{"quota", "set", "--quota-scope", "bucket", "--bucket", "limited", "--max-objects", "2"},
	{"quota", "enable", "--quota-scope", "bucket", "--bucket", "limited"},
} {
	if _, err := gateway.Admin(ctx, args...); err != nil {
		t.Fatal(err)
	}
}
// A third object in "limited" fails with 403 QuotaExceeded (poll: enforcement
// is asynchronous); "sibling" accepts writes. "quota disable" restores writes.

for _, args := range [][]string{
	{"reshard", "add", "--bucket", "limited", "--num-shards", "17"},
	{"reshard", "list"},
	{"reshard", "process"},
} {
	if _, err := gateway.Admin(ctx, args...); err != nil {
		t.Fatal(err)
	}
}
stats, err := gateway.Admin(ctx, "bucket", "stats", "--bucket", "limited")
if err != nil {
	t.Fatal(err)
}
t.Logf("bucket stats: %s", stats) // num_shards is 17

reshard process processes the whole reshard queue, so use it only when the queue contains your own buckets. See the Ceph quota and dynamic resharding documentation.

Native HTTPS endpoint

Config.TLS starts Beast's HTTPS listener next to HTTP. The certificate must name the host that S3SecureEndpoint returns: localhost or 127.0.0.1 for a local Docker engine, or the address given with NetworkConfig.WithHostAddress.

// A test CA and a server certificate for localhost.
caKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
leafKey, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
now := time.Now()
ca := &x509.Certificate{
	SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "test CA"},
	NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour),
	IsCA: true, BasicConstraintsValid: true,
	KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature,
}
caDER, _ := x509.CreateCertificate(rand.Reader, ca, ca, &caKey.PublicKey, caKey)
leaf := &x509.Certificate{
	SerialNumber: big.NewInt(2), Subject: pkix.Name{CommonName: "localhost"},
	DNSNames:    []string{"localhost"},
	IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")},
	NotBefore:   now.Add(-time.Minute), NotAfter: now.Add(time.Hour),
	KeyUsage:    x509.KeyUsageDigitalSignature,
	ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
}
leafDER, _ := x509.CreateCertificate(rand.Reader, leaf, ca, &leafKey.PublicKey, caKey)
keyDER, _ := x509.MarshalPKCS8PrivateKey(leafKey)
caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER})
chainPEM := append(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: leafDER}), caPEM...)
keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER})

gateway, err := rgw.Start(ctx, cluster, rgw.Config{
	Name: "secure",
	TLS:  &rgw.TLSConfig{CertificatePEM: chainPEM, PrivateKeyPEM: keyPEM},
})
if err != nil {
	t.Fatal(err)
}
secureEndpoint, err := gateway.S3SecureEndpoint(ctx)
if err != nil {
	t.Fatal(err)
}

roots := x509.NewCertPool()
roots.AppendCertsFromPEM(caPEM)
client := s3.New(s3.Options{
	BaseEndpoint: aws.String(secureEndpoint),
	Region:       gateway.Region,
	Credentials:  credentials.NewStaticCredentialsProvider(gateway.AccessKey, gateway.SecretKey, ""),
	UsePathStyle: true,
	HTTPClient: &http.Client{Transport: &http.Transport{
		TLSClientConfig: &tls.Config{RootCAs: roots},
	}},
})
// Give the SUT secureEndpoint and caPEM; client checks the listener here.
if _, err := client.ListBuckets(ctx, &s3.ListBucketsInput{}); err != nil {
	t.Fatal(err)
}

The HTTP endpoint (S3Endpoint) keeps working and serves the same data on a different port. A client without the CA fails certificate verification. See Ceph's Beast frontend options.

Placement targets and storage classes

Create the pools, add a target with two storage classes, activate it, and place a bucket in it with LocationConstraint:

// Three OSDs for the 2+1 erasure-coded class.
cluster := startCluster(t, ceph.WithClusterConfig(ceph.ClusterConfig{}.WithOSDCount(3)))
for _, pool := range []ceph.PoolConfig{
	{Name: "app-index", Application: "rgw", PGNum: 1},
	{Name: "app-extra", Application: "rgw", PGNum: 1},
	{Name: "app-standard", Application: "rgw", PGNum: 1},
	{Name: "app-cold", Application: "rgw", PGNum: 1,
		ErasureCode: &ceph.ErasureCodeConfig{K: 2, M: 1}},
} {
	if _, err := cluster.CreatePool(ctx, pool); err != nil {
		t.Fatal(err)
	}
}
if err := cluster.WaitForClean(ctx); err != nil {
	t.Fatal(err)
}

gateway, err := rgw.Start(ctx, cluster, rgw.Config{})
if err != nil {
	t.Fatal(err)
}
inline := false
placement, err := gateway.CreatePlacement(ctx, rgw.PlacementConfig{
	Name: "tiered", IndexPool: "app-index", DataExtraPool: "app-extra",
	InlineData: &inline,
	StorageClasses: []rgw.StorageClassConfig{
		{Name: "STANDARD", DataPool: "app-standard"},
		{Name: "STANDARD_IA", DataPool: "app-cold"},
	},
})
if err != nil {
	t.Fatal(err)
}
if err := gateway.ApplyPlacement(ctx, placement); err != nil { // restarts the gateway
	t.Fatal(err)
}
endpoint, err := gateway.S3Endpoint(ctx) // resolve again after the restart
client := newS3Client(endpoint, gateway.Region, gateway.AccessKey, gateway.SecretKey)

if _, err := client.CreateBucket(ctx, &s3.CreateBucketInput{
	Bucket: aws.String("archive"),
	CreateBucketConfiguration: &types.CreateBucketConfiguration{
		LocationConstraint: types.BucketLocationConstraint(placement.LocationConstraint),
	},
}); err != nil {
	t.Fatal(err)
}
payload := []byte("archived log")
if _, err := client.PutObject(ctx, &s3.PutObjectInput{
	Bucket:       aws.String("archive"),
	Key:          aws.String("old.log"),
	Body:         bytes.NewReader(payload),
	StorageClass: types.StorageClassStandardIa,
}); err != nil {
	t.Fatal(err)
}

To confirm where the data went, list the data pool: client.Exec(ctx, []string{"rados", "-p", "app-cold", "ls"}) with client from cluster.ClientContainer(ctx). RGW object names in the data pool do not always contain the S3 key literally.

With Tags on the target, only users with a matching placement tag can create buckets in it; grant the tag and a default with User.SetPlacement. Lifecycle transitions between classes are configured by the S3 client.

In a realm, every zone needs its own mapping before the master publishes:

// Create the same pools in both clusters first.
sourcePlacement, err := ms.Source.CreatePlacement(ctx, config)
if err != nil {
	t.Fatal(err)
}
destinationPlacement, err := ms.Destination.CreatePlacement(ctx, config)
if err != nil {
	t.Fatal(err)
}
if err := ms.Source.ApplyPlacement(ctx, sourcePlacement); err != nil { // metadata master publishes
	t.Fatal(err)
}
if err := ms.PullDestinationPeriod(ctx); err != nil {
	t.Fatal(err)
}
if err := ms.Destination.ReloadPlacement(ctx, destinationPlacement); err != nil {
	t.Fatal(err)
}
// Resolve both S3 endpoints again. Allow time for sync to resume after the
// gateway restarts before checking replicated objects.

See Ceph's placement and storage classes.

References

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

Home

Navigation Menu

Repository navigation

RGW S3

RGW S3

Helpers

Test an S3 client with the AWS SDK for Go

Run several gateways in one zone

Users, tenants and accounts

Make lifecycle tests fast

Bucket notifications to an HTTP endpoint

STS AssumeRole

Swift and S3 on the same gateway

Server-side encryption with Vault

Usage logs and Admin Ops

Bucket rate limits

Bucket quota and resharding

Native HTTPS endpoint

Placement targets and storage classes

References

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally