From 33bf0f86255b4b6e9cb676df56f60cff98c58639 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Fri, 28 Aug 2026 00:31:08 +0800 Subject: [PATCH] feat: support kafka scram queue config --- Dockerfile | 7 +++ server/go.mod | 3 + server/go.sum | 7 +++ server/internal/library/queue/kafkamq.go | 73 ++++++++++++++++++---- server/internal/library/queue/queue.go | 23 ++++--- server/manifest/config/config.example.yaml | 7 ++- server/manifest/config/config.template | 9 ++- server/manifest/docker/Dockerfile | 7 ++- server/manifest/docker/entrypoint.sh | 2 +- 9 files changed, 113 insertions(+), 25 deletions(-) diff --git a/Dockerfile b/Dockerfile index 0362f4f..702b9e6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -36,6 +36,13 @@ RUN apk add --no-cache ca-certificates gettext bash tzdata WORKDIR /app +ENV QUEUE_DRIVER=redis +ENV KAFKA_ADDRESS=127.0.0.1:9092 +ENV KAFKA_VERSION=2.4.0.0 +ENV KAFKA_USERNAME= +ENV KAFKA_PASSWORD= +ENV KAFKA_SASL_MECHANISM=SCRAM-SHA-512 + COPY --from=server-build /src/manifest ./manifest COPY --from=server-build /src/hack ./hack COPY --from=server-build /out/hotgo . diff --git a/server/go.mod b/server/go.mod index 751af09..b483c1f 100644 --- a/server/go.mod +++ b/server/go.mod @@ -37,6 +37,7 @@ require ( github.com/tencentyun/cos-go-sdk-v5 v0.7.66 github.com/tinode/snowflake v1.0.0 github.com/ufilesdk-dev/ufile-gosdk v1.0.6 + github.com/xdg-go/scram v1.1.2 github.com/xuri/excelize/v2 v2.9.1 go.opentelemetry.io/otel v1.38.0 golang.org/x/crypto v0.41.0 @@ -140,6 +141,8 @@ require ( github.com/tjfoc/gmsm v1.4.1 // indirect github.com/tklauser/go-sysconf v0.3.12 // indirect github.com/tklauser/numcpus v0.6.1 // indirect + github.com/xdg-go/pbkdf2 v1.0.0 // indirect + github.com/xdg-go/stringprep v1.0.4 // indirect github.com/xuri/efp v0.0.1 // indirect github.com/xuri/nfp v0.0.1 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect diff --git a/server/go.sum b/server/go.sum index 4709e93..216392c 100644 --- a/server/go.sum +++ b/server/go.sum @@ -448,6 +448,12 @@ github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+F github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY= github.com/ufilesdk-dev/ufile-gosdk v1.0.6 h1:YhTZfA0U5zjKUe6qYrIs72jjxSZryTMbS9wE76fIxMs= github.com/ufilesdk-dev/ufile-gosdk v1.0.6/go.mod h1:s1LIpscsgVNx0JRNbDElNNQhyNXJiumVpzvPzVNS/XI= +github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= +github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY= +github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4= +github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8= +github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= github.com/xuri/efp v0.0.1 h1:fws5Rv3myXyYni8uwj2qKjVaRP30PdjeYe2Y6FDsCL8= github.com/xuri/efp v0.0.1/go.mod h1:ybY/Jr0T0GTCnYjKqmdwxyxn2BQf2RcQIIvex5QldPI= github.com/xuri/excelize/v2 v2.9.1 h1:VdSGk+rraGmgLHGFaGG9/9IWu1nj4ufjJ7uwMDtj8Qw= @@ -620,6 +626,7 @@ golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= diff --git a/server/internal/library/queue/kafkamq.go b/server/internal/library/queue/kafkamq.go index 88f0673..0e4d5ec 100644 --- a/server/internal/library/queue/kafkamq.go +++ b/server/internal/library/queue/kafkamq.go @@ -7,13 +7,20 @@ package queue import ( "context" + "crypto/sha256" + "crypto/sha512" "fmt" + "hash" + "strings" + "time" + "github.com/IBM/sarama" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" + "github.com/xdg-go/scram" + "hotgo/internal/consts" "hotgo/utility/simple" - "time" ) type KafkaMq struct { @@ -31,6 +38,29 @@ type KafkaConfig struct { Version string UserName string Password string + Mechanism string +} + +type xdgSCRAMClient struct { + *scram.ClientConversation + HashGeneratorFcn scram.HashGeneratorFcn +} + +func (x *xdgSCRAMClient) Begin(userName, password, authzID string) error { + client, err := x.HashGeneratorFcn.NewClient(userName, password, authzID) + if err != nil { + return err + } + x.ClientConversation = client.NewConversation() + return nil +} + +func (x *xdgSCRAMClient) Step(challenge string) (string, error) { + return x.ClientConversation.Step(challenge) +} + +func (x *xdgSCRAMClient) Done() bool { + return x.ClientConversation.Done() } // SendMsg 按字符串类型生产数据 @@ -134,11 +164,7 @@ func RegisterKafkaMqConsumer(connOpt KafkaConfig) (client MqConsumer, err error) conf := sarama.NewConfig() conf.Consumer.Return.Errors = true conf.Version = kfkVersion - if connOpt.UserName != "" { - conf.Net.SASL.Enable = true - conf.Net.SASL.User = connOpt.UserName - conf.Net.SASL.Password = connOpt.Password - } + configureKafkaSASL(conf, connOpt) // 默认按随机方式消费 conf.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange @@ -191,11 +217,7 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) { conf.ClientID = connOpt.ClientId conf.Version = kfkVersion - if connOpt.UserName != "" { - conf.Net.SASL.Enable = true - conf.Net.SASL.User = connOpt.UserName - conf.Net.SASL.Password = connOpt.Password - } + configureKafkaSASL(conf, connOpt) mqIns.producerIns, err = sarama.NewAsyncProducer(brokers, conf) if err != nil { @@ -209,6 +231,35 @@ func doRegisterKafkaProducer(connOpt KafkaConfig, mqIns *KafkaMq) (err error) { return } +func configureKafkaSASL(conf *sarama.Config, connOpt KafkaConfig) { + if connOpt.UserName == "" { + return + } + conf.Net.SASL.Enable = true + conf.Net.SASL.User = connOpt.UserName + conf.Net.SASL.Password = connOpt.Password + mechanism := strings.ToUpper(strings.ReplaceAll(connOpt.Mechanism, "_", "-")) + if mechanism == "" { + mechanism = sarama.SASLTypeSCRAMSHA512 + } + switch mechanism { + case sarama.SASLTypeSCRAMSHA512: + conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512 + conf.Net.SASL.SCRAMClientGeneratorFunc = scramClientGenerator(sha512.New) + case sarama.SASLTypeSCRAMSHA256: + conf.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA256 + conf.Net.SASL.SCRAMClientGeneratorFunc = scramClientGenerator(sha256.New) + default: + conf.Net.SASL.Mechanism = sarama.SASLMechanism(mechanism) + } +} + +func scramClientGenerator(hashFunc func() hash.Hash) func() sarama.SCRAMClient { + return func() sarama.SCRAMClient { + return &xdgSCRAMClient{HashGeneratorFcn: scram.HashGeneratorFcn(hashFunc)} + } +} + // validateVersion 验证版本是否有效 func validateVersion(version sarama.KafkaVersion) bool { for _, item := range sarama.SupportedVersions { diff --git a/server/internal/library/queue/queue.go b/server/internal/library/queue/queue.go index fb850bf..595ab4c 100644 --- a/server/internal/library/queue/queue.go +++ b/server/internal/library/queue/queue.go @@ -57,6 +57,9 @@ type RocketmqConf struct { type KafkaConf struct { Address []string `json:"address"` Version string `json:"version"` + Username string `json:"username"` + Password string `json:"password"` + Mechanism string `json:"mechanism"` RandClient bool `json:"randClient"` MultiConsumer bool `json:"multiConsumer"` } @@ -121,9 +124,12 @@ func NewProducer(groupName string) (mqClient MqProducer, err error) { return } mqClient, err = RegisterKafkaProducer(KafkaConfig{ - Brokers: config.Kafka.Address, - GroupID: groupName, - Version: config.Kafka.Version, + Brokers: config.Kafka.Address, + GroupID: groupName, + Version: config.Kafka.Version, + UserName: config.Kafka.Username, + Password: config.Kafka.Password, + Mechanism: config.Kafka.Mechanism, }) case "redis": if _, err = g.Redis().Do(ctx, "ping"); err == nil { @@ -185,10 +191,13 @@ func NewConsumer(groupName string) (mqClient MqConsumer, err error) { } mqClient, err = RegisterKafkaMqConsumer(KafkaConfig{ - Brokers: config.Kafka.Address, - GroupID: groupName, - Version: config.Kafka.Version, - ClientId: clientId, + Brokers: config.Kafka.Address, + GroupID: groupName, + Version: config.Kafka.Version, + ClientId: clientId, + UserName: config.Kafka.Username, + Password: config.Kafka.Password, + Mechanism: config.Kafka.Mechanism, }) case "redis": if _, err = g.Redis().Do(ctx, "ping"); err == nil { diff --git a/server/manifest/config/config.example.yaml b/server/manifest/config/config.example.yaml index 7044e0d..35f1fc2 100644 --- a/server/manifest/config/config.example.yaml +++ b/server/manifest/config/config.example.yaml @@ -212,8 +212,11 @@ queue: retry: 0 # 重试次数 logLevel: "info" # rocketmq日志输出级别，可选：all|close|debug|info|warn|error|fatal kafka: - address: "127.0.0.1:9092" # kafka地址+端口 - version: "2.0.0.0" # kafka专属配置，默认2.0.0.0 + address: ["127.0.0.1:9092"] # kafka地址+端口 + version: "2.4.0.0" # kafka专属配置，默认2.4.0.0 + username: "" # SASL用户名，未开启认证时留空 + password: "" # SASL密码，未开启认证时留空 + mechanism: "SCRAM-SHA-512" # SASL机制，可选：SCRAM-SHA-512|SCRAM-SHA-256 randClient: true # 开启随机生成clientID，可以实现启动多实例同时一起消费相同topic，加速消费能力的特性，默认为true multiConsumer: true # 是否支持创建多个消费者 diff --git a/server/manifest/config/config.template b/server/manifest/config/config.template index f35a467..f266e69 100644 --- a/server/manifest/config/config.template +++ b/server/manifest/config/config.template @@ -192,7 +192,7 @@ token: # 消息队列 queue: switch: true # 队列开关，可选：true|false，默认为true - driver: "redis" # 队列驱动，可选：disk|redis|rocketmq|kafka，默认为disk + driver: "$QUEUE_DRIVER" # 队列驱动，可选：disk|redis|rocketmq|kafka，默认为disk groupName: "hotgo" # mq群组名称 # 磁盘队列 disk: @@ -212,8 +212,11 @@ queue: retry: 0 # 重试次数 logLevel: "info" # rocketmq日志输出级别，可选：all|close|debug|info|warn|error|fatal kafka: - address: "127.0.0.1:9092" # kafka地址+端口 - version: "2.0.0.0" # kafka专属配置，默认2.0.0.0 + address: ["$KAFKA_ADDRESS"] # kafka地址+端口 + version: "$KAFKA_VERSION" # kafka专属配置，默认2.4.0.0 + username: "$KAFKA_USERNAME" # SASL用户名，未开启认证时留空 + password: "$KAFKA_PASSWORD" # SASL密码，未开启认证时留空 + mechanism: "$KAFKA_SASL_MECHANISM" # SASL机制，可选：SCRAM-SHA-512|SCRAM-SHA-256 randClient: true # 开启随机生成clientID，可以实现启动多实例同时一起消费相同topic，加速消费能力的特性，默认为true multiConsumer: true # 是否支持创建多个消费者 diff --git a/server/manifest/docker/Dockerfile b/server/manifest/docker/Dockerfile index f36b107..2cbc538 100644 --- a/server/manifest/docker/Dockerfile +++ b/server/manifest/docker/Dockerfile @@ -5,6 +5,12 @@ FROM loads/alpine:3.8 ############################################################################### ENV WORKDIR /app +ENV QUEUE_DRIVER redis +ENV KAFKA_ADDRESS 127.0.0.1:9092 +ENV KAFKA_VERSION 2.4.0.0 +ENV KAFKA_USERNAME= +ENV KAFKA_PASSWORD= +ENV KAFKA_SASL_MECHANISM SCRAM-SHA-512 ADD hack $WORKDIR/hack/ ADD manifest/config $WORKDIR/manifest/config/ ADD resource $WORKDIR/resource/ @@ -20,4 +26,3 @@ RUN chmod +x $WORKDIR/entrypoint.sh WORKDIR $WORKDIR CMD ./entrypoint.sh - diff --git a/server/manifest/docker/entrypoint.sh b/server/manifest/docker/entrypoint.sh index 74d61b4..608a9fe 100644 --- a/server/manifest/docker/entrypoint.sh +++ b/server/manifest/docker/entrypoint.sh @@ -6,7 +6,7 @@ cd /app # Render the real config from the template using only the specific env vars we # expect — avoids accidentally substituting anything else that might contain a # literal '$' in the YAML. -envsubst '$MYSQL_DSN $REDIS_ADDR $REDIS_DB $REDIS_PASSWORD $TOKEN_SECRET_KEY' \ +envsubst '$MYSQL_DSN $REDIS_ADDR $REDIS_DB $REDIS_PASSWORD $TOKEN_SECRET_KEY $QUEUE_DRIVER $KAFKA_ADDRESS $KAFKA_VERSION $KAFKA_USERNAME $KAFKA_PASSWORD $KAFKA_SASL_MECHANISM' \ < manifest/config/config.template > manifest/config/config.yaml exec ./hotgo