rgw/admin: Add bucket notification and topic management APIs

[anoopcs9]

Adds SNS topic management and S3 bucket notification API support to the rgw admin client. It also introduces a doRequest() helper to share sign+send+read logic across all call variants.

fixes #547
fixes #548

Checklist

• Added tests for features and functional changes
• Public functions and types are documented
• Standard formatting is applied to Go code
• Is this a new API? Added a new file that begins with //go:build ceph_preview
• Ran make api-update to record new APIs

[phlogistonjohn]

looks OK to me, but of course RGW is my weak area so take this with a grain of salt.

[mergify]

This pull request does not currently match the merge queue conditions, so it cannot be queued from here. The box comes back if it matches again.

[fitbeard]

Thanks for upstreaming this — it'd let downstream consumers drop a good chunk of custom SNS/SigV4 plumbing, so I'd love to see it land solid. Two issues I ran into implementing the same against RGW (confirmed on Tentacle 20.2.4), plus a couple of minor notes:

1. CreateTopic sends Attributes in the query string, and RGW drops them. callSNS sets u.RawQuery = params.Encode() with a nil body, but RGW only parses Attributes.entry.N.key/value from the POST form body (application/x-www-form-urlencoded). The call returns 200 and creates the topic, but the attributes are silently discarded — so push-endpoint/persistent/Policy/etc. never take effect. GetTopicAttributes/DeleteTopic/ListTopics are unaffected (no Attributes), so only CreateTopic needs to move its params into the body. Minimal repro (build with -tags ceph_preview):

arn, _ := co.CreateTopic(ctx, "repro-topic", map[string]string{"push-endpoint": "http://push.example:8080"})
t, _ := co.GetTopicAttributes(ctx, arn)
// t.EndPoint.EndpointAddress == ""

2. The attribute loop reuses a fixed index:

for k, v := range attrs {
    params.Set("Attributes.entry.1.key", k)
    params.Set("Attributes.entry.1.value", v)
}

Every entry overwrites entry.1, so only one (map-order-dependent) attribute is ever sent — it needs an incrementing index (entry.1, entry.2, …). Topics commonly set several attributes at once.

3. Both of the above would be caught by having topic_test.go pass 2–3 attributes and assert they round-trip via GetTopicAttributes (e.g. EndPoint.EndpointAddress); it passes a single attribute and only checks TopicArn/Name, which come back regardless of whether the attributes were stored.

Minor / optional:

• Transient errors. Under concurrent topic churn RGW's shared topic metadata is eventually consistent — CreateTopic/DeleteTopic can return transient NoSuchKey/5xx/ConcurrentModification, and GetTopicAttributes can briefly 404 right after a successful CreateTopic. Not something a low-level client must retry, but the integration tests may flake without a small retry/backoff.
• Error body format. SNS returns S3-style bodies (rather than the IAM-style ); worth confirming handleStatusError parses that so callers get a meaningful code/message rather than an opaque error.

[fitbeard]

Real-world topics almost always carry several attributes at once. A few examples from our downstream provider's acceptance suite, expressed against this API — with the current loop each of these keeps only one (map-order-dependent) attribute:

// Persistent topic with retry tuning (4 attributes)
co.CreateTopic(ctx, "orders-persistent", map[string]string{
    "push-endpoint": "http://consumer.internal:10900",
    "persistent":    "true",
    "time_to_live":  "300",
    "max_retries":   "5",
})

// Topic with opaque data + persistence (4 attributes)
co.CreateTopic(ctx, "orders-opaque", map[string]string{
    "push-endpoint": "http://consumer.internal:10900",
    "persistent":    "true",
    "OpaqueData":    "tenant=acme;pipeline=ingest",
    "time_to_live":  "600",
})

// CloudEvents-formatted delivery (2 attributes)
co.CreateTopic(ctx, "orders-cloudevents", map[string]string{
    "push-endpoint": "http://consumer.internal:10900",
    "cloudevents":   "true",
})

// Kafka endpoint with broker list + ack level (4 attributes)
co.CreateTopic(ctx, "orders-kafka", map[string]string{
    "push-endpoint":   "kafka://kafka.internal:9092",
    "kafka-brokers":   "kafka.internal:9092",
    "kafka-ack-level": "broker",
    "persistent":      "true",
})

With the fixed entry.1 index only one of these survives per call and even that is dropped while it's sent in the query string. An incrementing index + POST-body encoding makes all of them round-trip.

Pull request has been modified.

[anoopcs9]

@fitbeard Thanks for the thorough review - both issues were real and are now hopefully fixed:

1. callSNS() now sends all parameters (including Attributes.entry.N.{key|value}) in the POST form body (application/x-www-form-urlencoded).

2. CreateTopic now uses an incrementing index (entry.1, entry.2, . . .) derived from sorted map keys, so multiple attributes survive and round-trip correctly.

The test has been updated to create a topic with two attributes ('push-endpoint' + 'persistent') and assert they both appear in GetTopicAttributes, which would have caught both bugs.

handleStatusError parses S3-style XML error bodies, so callers get meaningful codes/messages. I haven't added retry/backoff for transient errors in the integration tests yet, but agree it's worth considering if we see flakes.

[fitbeard]

The incrementing/sorted-keys fix, the POST-body switch, and the round-trip assertion in the test all look good — thanks for the quick turnaround. One regression the fix introduced, which is what's currently failing the reef and quincy jobs:

In callSNS, the now non-empty body is signed with the hash of an empty string:

payloadHash := sha256Hex([]byte{})
params.Set("PayloadHash", payloadHash)
body := []byte(params.Encode())
...
return api.doRequest(ctx, req, payloadHash)

So X-Amz-Content-Sha256 advertises e3b0c442…b855 (the hash of empty) while the body isn't empty. Newer RGW (main/tentacle/squid/umbrella) tolerates the mismatch, but reef (18.2.8) and quincy verify it and reject CreateTopic with 403 SignatureDoesNotMatch; the notification tests then cascade (empty topic → InvalidArgument → index out of range panic). From the reef job:

Action=CreateTopic&…&PayloadHash=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855
HTTP/1.1 403 Forbidden
<Error><Code>SignatureDoesNotMatch</Code>…</Error>

Two separate things here:

1. Wrong signing hash — it needs to be the hash of the actual body, not of an empty slice. (Your callNotification signs with UNSIGNED-PAYLOAD and reef accepts that path — the notification PUT gets a business error, not a 403 — so it's a good reference for what reef expects.)
2. Spurious PayloadHash form field — params.Set("PayloadHash", …) puts a signing detail into the SNS form body; it isn't an SNS parameter (you can see it echoed in the failing request line above) and should be removed.

[fitbeard]

For the callSNS signing issue above — here's a minimal fix that keeps the signed-payload approach but hashes the actual encoded body and drops the PayloadHash field:

func (api *API) callSNS(ctx context.Context, action string, params url.Values) ([]byte, error) {
    if params == nil {
            params = url.Values{}
    }
    params.Set("Action", action)

    body := []byte(params.Encode())
    payloadHash := sha256Hex(body)

    req, err := http.NewRequestWithContext(ctx, http.MethodPost, api.Endpoint, bytes.NewReader(body))
    if err != nil {
            return nil, err
    }

    req.Header.Set("Content-Type", "application/x-www-form-urlencoded")

    return api.doRequest(ctx, req, payloadHash)
}

Key changes: compute payloadHash from body after params.Encode() (so X-Amz-Content-Sha256 matches what's sent), and remove the params.Set("PayloadHash", …) line. Worth updating the doc comment too — drop PayloadHash from the "All parameters (…)" list.

Alternatively, mirror callNotification exactly: drop both payloadHash lines and return api.doRequest(ctx, req, unsignedPayload). That also works, but then sha256Hex and its crypto/sha256/encoding/hex imports become unused and should be removed — so hashing the real body is the smaller change.

[anoopcs9]

@fitbeard Updated to hash the actual body with sha256Hex(body) instead. Thanks for the insights.

Regarding the unsignedPayload alternative - I tried it but it fails with "SignatureDoesNotMatch" on the latest main branch build. Looks like the SNS topic endpoint requires a signed body hash even for POST requests, unlike the S3 notification endpoint which accepts "UNSIGNED-PAYLOAD". So signing with the real body hash is the approach that works consistently across all versions.

[anoopcs9]

@Mergifyio rebase

[mergify]

rebase

❌ This pull request comes from a fork and cannot be rebased

Details

GitHub refuses an OAuth token on its rebase API for a fork, so rebasing one means impersonating a GitHub user to force-push the contributor's branch. Mergify does not do that.

Use the update action or the @mergifyio update command instead: it brings the pull request up to date by merging the base branch into it, and needs no impersonation. It only has something to do when the pull request is behind its base branch, so if what the branch needs is a linear history, its author has to rebase it themselves.

[fitbeard]

@anoopcs9 Thanks for the quick iterations!