mirrors/versitygw
1
0
Fork 0
mirror of https://github.com/versity/versitygw.git synced 2026-09-06 08:06:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1448 from versity/ben/rabbitmq-event

Browse Source 
feat: add rabbitmq s3 event notification support
This commit is contained in:
Ben McClelland
2025-08-13 16:34:49 -07:00
committed by GitHub
parent 634396c3c5 36d2a55162
commit 6be62f189d
6 changed files with 204 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/versitygw/main.go
+26
View File
@@ -45,6 +45,8 @@ var (
certFile, keyFile string
kafkaURL, kafkaTopic, kafkaKey string
natsURL, natsTopic string
rabbitmqURL, rabbitmqExchange string
rabbitmqRoutingKey string
eventWebhookURL string
eventConfigFilePath string
logWebhookURL, accessLog string
@@ -288,6 +290,27 @@ func initFlags() []cli.Flag {
Destination: &natsTopic,
Aliases: []string{"ent"},
},
&cli.StringFlag{
Name: "event-rabbitmq-url",
Usage: "rabbitmq server url to send the bucket notifications (amqp or amqps scheme)",
EnvVars: []string{"VGW_EVENT_RABBITMQ_URL"},
Destination: &rabbitmqURL,
Aliases: []string{"eru"},
},
&cli.StringFlag{
Name: "event-rabbitmq-exchange",
Usage: "rabbitmq exchange to publish bucket notifications to (blank for default)",
EnvVars: []string{"VGW_EVENT_RABBITMQ_EXCHANGE"},
Destination: &rabbitmqExchange,
Aliases: []string{"ere"},
},
&cli.StringFlag{
Name: "event-rabbitmq-routing-key",
Usage: "rabbitmq routing key when publishing bucket notifications (defaults to bucket name when blank)",
EnvVars: []string{"VGW_EVENT_RABBITMQ_ROUTING_KEY"},
Destination: &rabbitmqRoutingKey,
Aliases: []string{"errk"},
},
&cli.StringFlag{
Name: "event-webhook-url",
Usage: "webhook url to send bucket notifications",
@@ -693,6 +716,9 @@ func runGateway(ctx context.Context, be backend.Backend) error {
KafkaTopicKey: kafkaKey,
NatsURL: natsURL,
NatsTopic: natsTopic,
RabbitmqURL: rabbitmqURL,
RabbitmqExchange: rabbitmqExchange,
RabbitmqRoutingKey: rabbitmqRoutingKey,
WebhookURL: eventWebhookURL,
FilterConfigFilePath: eventConfigFilePath,
})
extra/example.conf
+13
View File
@@ -169,6 +169,19 @@ ROOT_SECRET_ACCESS_KEY=
#VGW_EVENT_NATS_URL=
#VGW_EVENT_NATS_TOPIC=
# Bucket events can be sent to a RabbitMQ messaging service. When
# VGW_EVENT_RABBITMQ_URL is specified, events will be published to the specified
# exchange (VGW_EVENT_RABBITMQ_EXCHANGE) using the routing key
# (VGW_EVENT_RABBITMQ_ROUTING_KEY). If exchange is blank the default exchange is
# used. If routing key is blank, it will be left empty (the server can bind a
# queue with an empty binding key or you can set an explicit key).
# Example URL formats:
# amqp://user:pass@rabbitmq:5672/
# amqps://user:pass@rabbitmq:5671/vhost
#VGW_EVENT_RABBITMQ_URL=
#VGW_EVENT_RABBITMQ_EXCHANGE=
#VGW_EVENT_RABBITMQ_ROUTING_KEY=
# Bucket events can be sent to a webhook. When VGW_EVENT_WEBHOOK_URL is
# specified, all configured bucket events will be sent to the webhook.
#VGW_EVENT_WEBHOOK_URL=
go.mod
+1
View File
@@ -20,6 +20,7 @@ require (
github.com/nats-io/nats.go v1.44.0
github.com/oklog/ulid/v2 v2.1.1
github.com/pkg/xattr v0.4.12
github.com/rabbitmq/amqp091-go v1.10.0
github.com/segmentio/kafka-go v0.4.48
github.com/smira/go-statsd v1.3.4
github.com/stretchr/testify v1.10.0
go.sum
+4
View File
@@ -147,6 +147,8 @@ github.com/pkg/xattr v0.4.12 h1:rRTkSyFNTRElv6pkA3zpjHpQ90p/OdHQC1GmGh1aTjM=
github.com/pkg/xattr v0.4.12/go.mod h1:di8WF84zAKk8jzR1UBTEWh9AUlIZZ7M/JNt8e9B6ktU=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw=
github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
@@ -192,6 +194,8 @@ github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZ
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
s3event/event.go
+10 -3
View File
@@ -72,9 +72,10 @@ type ConfigurationId string
// This field will be changed after implementing per bucket notifications
const (
ConfigurationIdKafka ConfigurationId = "kafka-global"
ConfigurationIdNats ConfigurationId = "nats-global"
ConfigurationIdWebhook ConfigurationId = "webhook-global"
ConfigurationIdKafka ConfigurationId = "kafka-global"
ConfigurationIdNats ConfigurationId = "nats-global"
ConfigurationIdWebhook ConfigurationId = "webhook-global"
ConfigurationIdRabbitMQ ConfigurationId = "rabbitmq-global"
)
type EventS3Data struct {
@@ -113,6 +114,9 @@ type EventConfig struct {
KafkaTopicKey string
NatsURL string
NatsTopic string
RabbitmqURL string
RabbitmqExchange string
RabbitmqRoutingKey string
WebhookURL string
FilterConfigFilePath string
}
@@ -133,6 +137,9 @@ func InitEventSender(cfg *EventConfig) (S3EventSender, error) {
case cfg.NatsURL != "":
evSender, err = InitNatsEventService(cfg.NatsURL, cfg.NatsTopic, filter)
fmt.Printf("initializing S3 Event Notifications with Nats. URL: %v, topic: %v\n", cfg.NatsURL, cfg.NatsTopic)
case cfg.RabbitmqURL != "":
evSender, err = InitRabbitmqEventService(cfg.RabbitmqURL, cfg.RabbitmqExchange, cfg.RabbitmqRoutingKey, filter)
fmt.Printf("initializing S3 Event Notifications with RabbitMQ. URL: %v, exchange: %v, routing key: %v\n", cfg.RabbitmqURL, cfg.RabbitmqExchange, cfg.RabbitmqRoutingKey)
default:
return nil, nil
}
s3event/rabbitmq.go
+150
View File
@@ -0,0 +1,150 @@
// Copyright 2023 Versity Software
// This file is licensed under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
package s3event
import (
"encoding/json"
"encoding/xml"
"fmt"
"os"
"sync"
"time"
"github.com/gofiber/fiber/v2"
"github.com/google/uuid"
amqp "github.com/rabbitmq/amqp091-go"
"github.com/versity/versitygw/s3response"
)
// RabbitmqEventSender sends S3 events to a RabbitMQ exchange/queue.
// It mirrors the behavior of the Kafka and NATS implementations: send a
// test event on initialization to validate configuration, filter events,
// and handle multi-delete payloads.
type RabbitmqEventSender struct {
url string
exchange string
routingKey string
conn *amqp.Connection
channel *amqp.Channel
mu sync.Mutex
filter EventFilter
}
// InitRabbitmqEventService creates a RabbitMQ sender. If exchange is blank the
// default (empty string) exchange is used. If routingKey is blank we publish
// with an empty routing key; for delete object multi-events we override the
// routing key with the bucket name (object key not suitable as key routinely has '/')
func InitRabbitmqEventService(url, exchange, routingKey string, filter EventFilter) (S3EventSender, error) {
if url == "" {
return nil, fmt.Errorf("rabbitmq url should be specified")
}
conn, err := amqp.Dial(url)
if err != nil {
return nil, fmt.Errorf("rabbitmq connect: %w", err)
}
ch, err := conn.Channel()
if err != nil {
conn.Close()
return nil, fmt.Errorf("rabbitmq channel: %w", err)
}
// Send a test event to validate publishing works. Use a transient message.
testMsg, err := generateTestEvent()
if err != nil {
ch.Close()
conn.Close()
return nil, fmt.Errorf("rabbitmq generate test event: %w", err)
}
pub := amqp.Publishing{Timestamp: time.Now(), ContentType: fiber.MIMEApplicationJSON, Body: testMsg, MessageId: uuid.NewString()}
if err := ch.Publish(exchange, routingKey, false, false, pub); err != nil {
ch.Close()
conn.Close()
return nil, fmt.Errorf("rabbitmq publish test event: %w", err)
}
return &RabbitmqEventSender{
url: url,
exchange: exchange,
routingKey: routingKey,
conn: conn,
channel: ch,
filter: filter,
}, nil
}
func (rs *RabbitmqEventSender) SendEvent(ctx *fiber.Ctx, meta EventMeta) {
rs.mu.Lock()
defer rs.mu.Unlock()
if rs.filter != nil && !rs.filter.Filter(meta.EventName) {
return
}
if meta.EventName == EventObjectRemovedDeleteObjects {
var dObj s3response.DeleteObjects
if err := xml.Unmarshal(ctx.Body(), &dObj); err != nil {
fmt.Fprintf(os.Stderr, "failed to parse delete objects input payload: %v\n", err.Error())
return
}
for _, obj := range dObj.Objects {
key := *obj.Key
schema := createEventSchema(ctx, meta, ConfigurationIdRabbitMQ)
schema.Records[0].S3.Object.Key = key
schema.Records[0].S3.Object.VersionId = obj.VersionId
go rs.send(schema)
}
return
}
schema := createEventSchema(ctx, meta, ConfigurationIdRabbitMQ)
go rs.send(schema)
}
func (rs *RabbitmqEventSender) Close() error {
var firstErr error
if rs.channel != nil {
if err := rs.channel.Close(); err != nil {
firstErr = err
}
}
if rs.conn != nil {
if err := rs.conn.Close(); err != nil && firstErr == nil {
firstErr = err
}
}
return firstErr
}
func (rs *RabbitmqEventSender) send(event EventSchema) {
body, err := json.Marshal(event)
if err != nil {
fmt.Fprintf(os.Stderr, "failed to marshal event data: %v\n", err.Error())
return
}
msg := amqp.Publishing{
Timestamp: time.Now(),
ContentType: fiber.MIMEApplicationJSON,
Body: body,
MessageId: uuid.NewString(),
}
if err := rs.channel.Publish(rs.exchange, rs.routingKey, false, false, msg); err != nil {
fmt.Fprintf(os.Stderr, "failed to send rabbitmq event: %v\n", err.Error())
}
}