mirror/hatchet
1
0
Fork 0
mirror of https://github.com/hatchet-dev/hatchet.git synced 2025-12-30 05:09:44 -06:00
Code Issues Packages Projects Releases Wiki Activity

feat: max channels for rabbitmq (#2365)

Browse Source 
* feat: max conns for rabbitmq

* rename conns -> chans
This commit is contained in:
abelanger5
2025-09-30 08:49:45 -04:00
committed by GitHub
parent 806931f8c6
commit 2edeeb10ea
4 changed files with 38 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
internal/msgqueue/v1/rabbitmq/channel_pool.go
View File

@@ -51,7 +51,7 @@ func (p *channelPool) hasActiveConnection() bool {
return p.conn != nil && !p.conn.IsClosed()
}
func newChannelPool(ctx context.Context, l *zerolog.Logger, url string) (*channelPool, error) {
func newChannelPool(ctx context.Context, l *zerolog.Logger, url string, maxChannels int32) (*channelPool, error) {
p := &channelPool{
l: l,
url: url,
@@ -116,7 +116,7 @@ func newChannelPool(ctx context.Context, l *zerolog.Logger, url string) (*channe
}()
// FIXME: this is probably too many channels
maxPoolSize := int32(100)
maxPoolSize := maxChannels
pool, err := puddle.NewPool(&puddle.Config[*amqp.Channel]{
Constructor: constructor,

30
internal/msgqueue/v1/rabbitmq/rabbitmq.go
View File

@@ -59,6 +59,8 @@ type MessageQueueImplOpts struct {
qos int
disableTenantExchangePubs bool
deadLetterBackoff time.Duration
maxPubChannels int32
maxSubChannels int32
}
func defaultMessageQueueImplOpts() *MessageQueueImplOpts {
@@ -83,6 +85,18 @@ func WithURL(url string) MessageQueueImplOpt {
}
}
func WithMaxPubChannels(maxConns int32) MessageQueueImplOpt {
return func(opts *MessageQueueImplOpts) {
opts.maxPubChannels = maxConns
}
}
func WithMaxSubChannels(maxConns int32) MessageQueueImplOpt {
return func(opts *MessageQueueImplOpts) {
opts.maxSubChannels = maxConns
}
}
func WithQos(qos int) MessageQueueImplOpt {
return func(opts *MessageQueueImplOpts) {
opts.qos = qos
@@ -114,14 +128,26 @@ func New(fs ...MessageQueueImplOpt) (func() error, *MessageQueueImpl) {
newLogger := opts.l.With().Str("service", "rabbitmq").Logger()
opts.l = &newLogger
pubChannelPool, err := newChannelPool(ctx, opts.l, opts.url)
pubMaxChans := opts.maxPubChannels
if pubMaxChans <= 0 {
pubMaxChans = 20
}
pubChannelPool, err := newChannelPool(ctx, opts.l, opts.url, pubMaxChans)
if err != nil {
cancel()
return nil, nil
}
subChannelPool, err := newChannelPool(ctx, opts.l, opts.url)
subMaxChans := opts.maxSubChannels
if subMaxChans <= 0 {
subMaxChans = 100
}
subChannelPool, err := newChannelPool(ctx, opts.l, opts.url, subMaxChans)
if err != nil {
pubChannelPool.Close()

2
pkg/config/loader/loader.go
View File

@@ -405,6 +405,8 @@ func createControllerLayer(dc *database.Layer, cf *server.ServerConfigFile, vers
rabbitmqv1.WithLogger(&l),
rabbitmqv1.WithQos(cf.MessageQueue.RabbitMQ.Qos),
rabbitmqv1.WithDisableTenantExchangePubs(cf.Runtime.DisableTenantPubs),
rabbitmqv1.WithMaxPubChannels(cf.MessageQueue.RabbitMQ.MaxPubChans),
rabbitmqv1.WithMaxSubChannels(cf.MessageQueue.RabbitMQ.MaxSubChans),
)
cleanup1 = func() error {

8
pkg/config/server/server.go
View File

@@ -462,8 +462,10 @@ type PostgresMQConfigFile struct {
}
type RabbitMQConfigFile struct {
URL string `mapstructure:"url" json:"url,omitempty" validate:"required"`
Qos int `mapstructure:"qos" json:"qos,omitempty" default:"100"`
URL string `mapstructure:"url" json:"url,omitempty" validate:"required"`
Qos int `mapstructure:"qos" json:"qos,omitempty" default:"100"`
MaxPubChans int32 `mapstructure:"maxPubChans" json:"maxPubChans,omitempty" default:"20"`
MaxSubChans int32 `mapstructure:"maxSubChans" json:"maxSubChans,omitempty" default:"100"`
}
type ConfigFileEmail struct {
@@ -769,6 +771,8 @@ func BindAllEnv(v *viper.Viper) {
_ = v.BindEnv("msgQueue.kind", "SERVER_MSGQUEUE_KIND")
_ = v.BindEnv("msgQueue.rabbitmq.url", "SERVER_MSGQUEUE_RABBITMQ_URL")
_ = v.BindEnv("msgQueue.rabbitmq.maxPubChans", "SERVER_MSGQUEUE_RABBITMQ_MAX_PUB_CHANS")
_ = v.BindEnv("msgQueue.rabbitmq.maxSubChans", "SERVER_MSGQUEUE_RABBITMQ_MAX_SUB_CHANS")
// throughput options
_ = v.BindEnv("msgQueue.rabbitmq.qos", "SERVER_MSGQUEUE_RABBITMQ_QOS")