MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-04-20 18:44:21 -04:00
Code Issues Packages Projects Releases Wiki Activity

feat: add userCredentials for nats (#19139)

Browse Source
This commit is contained in:
jiuker 2024-02-28 02:11:55 +08:00 committed by GitHub
parent 1dd8ef09a6
commit 0aae0180fb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 49 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/config/notify/legacy.go
View File

@ -446,6 +446,10 @@ func SetNotifyNATS(s config.Config, natsName string, cfg target.NATSArgs) error
Key: target.NATSUsername, Key: target.NATSUsername,
Value: cfg.Username, Value: cfg.Username,
}, },
config.KV{
Key: target.NATSUserCredentials,
Value: cfg.UserCredentials,
},
config.KV{ config.KV{
Key: target.NATSPassword, Key: target.NATSPassword,
Value: cfg.Password, Value: cfg.Password,

36
internal/config/notify/parse.go
View File

@ -946,6 +946,11 @@ func GetNotifyNATS(natsKVS map[string]config.KVS, rootCAs *x509.CertPool) (map[s
usernameEnv = usernameEnv + config.Default + k usernameEnv = usernameEnv + config.Default + k
} }
userCredentialsEnv := target.NATSUserCredentials
if k != config.Default {
userCredentialsEnv = userCredentialsEnv + config.Default + k
}
passwordEnv := target.EnvNATSPassword passwordEnv := target.EnvNATSPassword
if k != config.Default { if k != config.Default {
passwordEnv = passwordEnv + config.Default + k passwordEnv = passwordEnv + config.Default + k
@ -982,21 +987,22 @@ func GetNotifyNATS(natsKVS map[string]config.KVS, rootCAs *x509.CertPool) (map[s
} }
natsArgs := target.NATSArgs{ natsArgs := target.NATSArgs{
Enable: true, Enable: true,
Address: *address, Address: *address,
Subject: env.Get(subjectEnv, kv.Get(target.NATSSubject)), Subject: env.Get(subjectEnv, kv.Get(target.NATSSubject)),
Username: env.Get(usernameEnv, kv.Get(target.NATSUsername)), Username: env.Get(usernameEnv, kv.Get(target.NATSUsername)),
Password: env.Get(passwordEnv, kv.Get(target.NATSPassword)), UserCredentials: env.Get(userCredentialsEnv, kv.Get(target.NATSUserCredentials)),
CertAuthority: env.Get(certAuthorityEnv, kv.Get(target.NATSCertAuthority)), Password: env.Get(passwordEnv, kv.Get(target.NATSPassword)),
ClientCert: env.Get(clientCertEnv, kv.Get(target.NATSClientCert)), CertAuthority: env.Get(certAuthorityEnv, kv.Get(target.NATSCertAuthority)),
ClientKey: env.Get(clientKeyEnv, kv.Get(target.NATSClientKey)), ClientCert: env.Get(clientCertEnv, kv.Get(target.NATSClientCert)),
Token: env.Get(tokenEnv, kv.Get(target.NATSToken)), ClientKey: env.Get(clientKeyEnv, kv.Get(target.NATSClientKey)),
TLS: env.Get(tlsEnv, kv.Get(target.NATSTLS)) == config.EnableOn, Token: env.Get(tokenEnv, kv.Get(target.NATSToken)),
TLSSkipVerify: env.Get(tlsSkipVerifyEnv, kv.Get(target.NATSTLSSkipVerify)) == config.EnableOn, TLS: env.Get(tlsEnv, kv.Get(target.NATSTLS)) == config.EnableOn,
PingInterval: pingInterval, TLSSkipVerify: env.Get(tlsSkipVerifyEnv, kv.Get(target.NATSTLSSkipVerify)) == config.EnableOn,
QueueDir: env.Get(queueDirEnv, kv.Get(target.NATSQueueDir)), PingInterval: pingInterval,
QueueLimit: queueLimit, QueueDir: env.Get(queueDirEnv, kv.Get(target.NATSQueueDir)),
RootCAs: rootCAs, QueueLimit: queueLimit,
RootCAs: rootCAs,
} }
natsArgs.JetStream.Enable = env.Get(jetStreamEnableEnv, kv.Get(target.NATSJetStream)) == config.EnableOn natsArgs.JetStream.Enable = env.Get(jetStreamEnableEnv, kv.Get(target.NATSJetStream)) == config.EnableOn

40
internal/event/target/nats.go
View File

@ -67,6 +67,7 @@ const (
EnvNATSAddress = "MINIO_NOTIFY_NATS_ADDRESS" EnvNATSAddress = "MINIO_NOTIFY_NATS_ADDRESS"
EnvNATSSubject = "MINIO_NOTIFY_NATS_SUBJECT" EnvNATSSubject = "MINIO_NOTIFY_NATS_SUBJECT"
EnvNATSUsername = "MINIO_NOTIFY_NATS_USERNAME" EnvNATSUsername = "MINIO_NOTIFY_NATS_USERNAME"
NATSUserCredentials = "MINIO_NOTIFY_NATS_USER_CREDENTIALS"
EnvNATSPassword = "MINIO_NOTIFY_NATS_PASSWORD" EnvNATSPassword = "MINIO_NOTIFY_NATS_PASSWORD"
EnvNATSToken = "MINIO_NOTIFY_NATS_TOKEN" EnvNATSToken = "MINIO_NOTIFY_NATS_TOKEN"
EnvNATSTLS = "MINIO_NOTIFY_NATS_TLS" EnvNATSTLS = "MINIO_NOTIFY_NATS_TLS"
@ -90,22 +91,23 @@ const (
// NATSArgs - NATS target arguments. // NATSArgs - NATS target arguments.
type NATSArgs struct { type NATSArgs struct {
Enable bool `json:"enable"` Enable bool `json:"enable"`
Address xnet.Host `json:"address"` Address xnet.Host `json:"address"`
Subject string `json:"subject"` Subject string `json:"subject"`
Username string `json:"username"` Username string `json:"username"`
Password string `json:"password"` UserCredentials string `json:"userCredentials"`
Token string `json:"token"` Password string `json:"password"`
TLS bool `json:"tls"` Token string `json:"token"`
TLSSkipVerify bool `json:"tlsSkipVerify"` TLS bool `json:"tls"`
Secure bool `json:"secure"` TLSSkipVerify bool `json:"tlsSkipVerify"`
CertAuthority string `json:"certAuthority"` Secure bool `json:"secure"`
ClientCert string `json:"clientCert"` CertAuthority string `json:"certAuthority"`
ClientKey string `json:"clientKey"` ClientCert string `json:"clientCert"`
PingInterval int64 `json:"pingInterval"` ClientKey string `json:"clientKey"`
QueueDir string `json:"queueDir"` PingInterval int64 `json:"pingInterval"`
QueueLimit uint64 `json:"queueLimit"` QueueDir string `json:"queueDir"`
JetStream struct { QueueLimit uint64 `json:"queueLimit"`
JetStream struct {
Enable bool `json:"enable"` Enable bool `json:"enable"`
} `json:"jetStream"` } `json:"jetStream"`
Streaming struct { Streaming struct {
@ -167,6 +169,9 @@ func (n NATSArgs) connectNats() (*nats.Conn, error) {
if n.Username != "" && n.Password != "" { if n.Username != "" && n.Password != "" {
connOpts = append(connOpts, nats.UserInfo(n.Username, n.Password)) connOpts = append(connOpts, nats.UserInfo(n.Username, n.Password))
} }
if n.UserCredentials != "" {
connOpts = append(connOpts, nats.UserCredentials(n.UserCredentials))
}
if n.Token != "" { if n.Token != "" {
connOpts = append(connOpts, nats.Token(n.Token)) connOpts = append(connOpts, nats.Token(n.Token))
} }
@ -211,6 +216,9 @@ func (n NATSArgs) connectStan() (stan.Conn, error) {
if n.Streaming.MaxPubAcksInflight > 0 { if n.Streaming.MaxPubAcksInflight > 0 {
connOpts = append(connOpts, stan.MaxPubAcksInflight(n.Streaming.MaxPubAcksInflight)) connOpts = append(connOpts, stan.MaxPubAcksInflight(n.Streaming.MaxPubAcksInflight))
} }
if n.UserCredentials != "" {
connOpts = append(connOpts, stan.NatsOptions(nats.UserCredentials(n.UserCredentials)))
}
return stan.Connect(n.Streaming.ClusterID, clientID, connOpts...) return stan.Connect(n.Streaming.ClusterID, clientID, connOpts...)
} }