Add retry mechanism to mongoDB, Redis, Kafka

pull/518/head
plutoyty 1 year ago
parent a735a70a7b
commit 35b1f75036

1
.gitignore vendored

@ -389,3 +389,4 @@ Sessionx.vim
[._]*.un~
# End of https://www.toptal.com/developers/gitignore/api/go,git,vim,tags,test,emacs,backup,jetbrains
.idea

@ -27,6 +27,11 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
)
const (
maxRetry = 10 //number of retries
)
// NewRedis Initialize redis connection
func NewRedis() (redis.UniversalClient, error) {
if len(config.Config.Redis.Address) == 0 {
return nil, errors.New("redis address is empty")
@ -39,6 +44,7 @@ func NewRedis() (redis.UniversalClient, error) {
Username: config.Config.Redis.Username,
Password: config.Config.Redis.Password, // no password set
PoolSize: 50,
MaxRetries: maxRetry,
})
} else {
rdb = redis.NewClient(&redis.Options{
@ -46,14 +52,17 @@ func NewRedis() (redis.UniversalClient, error) {
Username: config.Config.Redis.Username,
Password: config.Config.Redis.Password, // no password set
DB: 0, // use default DB
PoolSize: 100, // 连接池大小
PoolSize: 100, // connection pool size
MaxRetries: maxRetry,
})
}
var err error = nil
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
err := rdb.Ping(ctx).Err()
err = rdb.Ping(ctx).Err()
if err != nil {
return nil, fmt.Errorf("redis ping %w", err)
}
return rdb, nil
return rdb, err
}

@ -0,0 +1,30 @@
// Copyright © 2023 OpenIM. All rights reserved.
//
// 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 cache
import (
"fmt"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"testing"
)
//TestNewRedis Test redis connection
func TestNewRedis(t *testing.T) {
err := config.InitConfig("config_folder_path")
if err != nil {
fmt.Println("config load error")
return
}
}

@ -100,7 +100,7 @@ func (e *ExtendMsgSetMongoDriver) GetExtendMsgSet(
return &setList[0], nil
}
// first modify msg
// InsertExtendMsg first modify msg.
func (e *ExtendMsgSetMongoDriver) InsertExtendMsg(
ctx context.Context,
conversationID string,
@ -130,7 +130,7 @@ func (e *ExtendMsgSetMongoDriver) InsertExtendMsg(
return utils.Wrap(err, "")
}
// insert or update
// InsertOrUpdateReactionExtendMsgSet insert or update.
func (e *ExtendMsgSetMongoDriver) InsertOrUpdateReactionExtendMsgSet(
ctx context.Context,
conversationID string,
@ -163,7 +163,7 @@ func (e *ExtendMsgSetMongoDriver) InsertOrUpdateReactionExtendMsgSet(
return utils.Wrap(err, "")
}
// delete TypeKey
// DeleteReactionExtendMsgSet delete TypeKey.
func (e *ExtendMsgSetMongoDriver) DeleteReactionExtendMsgSet(
ctx context.Context,
conversationID string,

@ -31,19 +31,21 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
)
const (
maxRetry = 10 //number of retries
)
type Mongo struct {
db *mongo.Client
}
// NewMongo Initialize MongoDB connection
func NewMongo() (*Mongo, error) {
specialerror.AddReplace(mongo.ErrNoDocuments, errs.ErrRecordNotFound)
uri := "mongodb://sample.host:27017/?maxPoolSize=20&w=majority"
url := "mongodb://sample.host:27017/?maxPoolSize=20&w=majority"
if config.Config.Mongo.Uri != "" {
// example:
// mongodb://$user:$password@mongo1.mongo:27017,mongo2.mongo:27017,mongo3.mongo:27017/$DBDatabase/?replicaSet=rs0&readPreference=secondary&authSource=admin&maxPoolSize=$DBMaxPoolSize
uri = config.Config.Mongo.Uri
url = config.Config.Mongo.Uri
} else {
//mongodb://mongodb1.example.com:27317,mongodb2.example.com:27017/?replicaSet=mySet&authSource=authDB
mongodbHosts := ""
for i, v := range config.Config.Mongo.Address {
if i == len(config.Config.Mongo.Address)-1 {
@ -53,23 +55,34 @@ func NewMongo() (*Mongo, error) {
}
}
if config.Config.Mongo.Password != "" && config.Config.Mongo.Username != "" {
uri = fmt.Sprintf("mongodb://%s:%s@%s/%s?maxPoolSize=%d&authSource=admin",
url = fmt.Sprintf("mongodb://%s:%s@%s/%s?maxPoolSize=%d&authSource=admin",
config.Config.Mongo.Username, config.Config.Mongo.Password, mongodbHosts,
config.Config.Mongo.Database, config.Config.Mongo.MaxPoolSize)
} else {
uri = fmt.Sprintf("mongodb://%s/%s/?maxPoolSize=%d&authSource=admin",
url = fmt.Sprintf("mongodb://%s/%s/?maxPoolSize=%d&authSource=admin",
mongodbHosts, config.Config.Mongo.Database,
config.Config.Mongo.MaxPoolSize)
}
}
fmt.Println("mongo:", uri)
ctx, cancel := context.WithTimeout(context.Background(), time.Second*60)
fmt.Println("mongo:", url)
var mongoClient *mongo.Client
var err error = nil
for i := 0; i <= maxRetry; i++ {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
mongoClient, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
if err != nil {
mongoClient, err = mongo.Connect(ctx, options.Client().ApplyURI(url))
if err == nil {
return &Mongo{db: mongoClient}, nil
}
if cmdErr, ok := err.(mongo.CommandError); ok {
if cmdErr.Code == 13 || cmdErr.Code == 18 {
return nil, err
} else {
fmt.Printf("Failed to connect to MongoDB: %s\n", err)
}
return &Mongo{db: mongoClient}, nil
}
}
return nil, err
}
func (m *Mongo) GetClient() *mongo.Client {

@ -1,9 +1,18 @@
/*
** description("").
** copyright('open-im,www.open-im.io').
** author("fg,Gordon@tuoyun.net").
** time(2021/5/27 10:31).
*/package http
// Copyright © 2023 OpenIM. All rights reserved.
//
// 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 http
import (
"bytes"

@ -17,12 +17,12 @@ package kafka
import (
"context"
"errors"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/config"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/constant"
log "github.com/OpenIMSDK/Open-IM-Server/pkg/common/log"
"github.com/OpenIMSDK/Open-IM-Server/pkg/common/mcontext"
"github.com/OpenIMSDK/Open-IM-Server/pkg/utils"
"time"
"github.com/Shopify/sarama"
"google.golang.org/protobuf/proto"
@ -30,6 +30,10 @@ import (
prome "github.com/OpenIMSDK/Open-IM-Server/pkg/common/prome"
)
const (
maxRetry = 10 //number of retries
)
var errEmptyMsg = errors.New("binary msg is empty")
type Producer struct {
@ -39,6 +43,7 @@ type Producer struct {
producer sarama.SyncProducer
}
// NewKafkaProducer Initialize kafka producer
func NewKafkaProducer(addr []string, topic string) *Producer {
p := Producer{}
p.config = sarama.NewConfig() //Instantiate a sarama Config
@ -53,9 +58,23 @@ func NewKafkaProducer(addr []string, topic string) *Producer {
}
p.addr = addr
p.topic = topic
producer, err := sarama.NewSyncProducer(p.addr, p.config) //Initialize the client
if err != nil {
panic(err.Error())
var producer sarama.SyncProducer
var err error
for i := 0; i <= maxRetry; i++ {
producer, err = sarama.NewSyncProducer(p.addr, p.config) //Initialize the client
if err == nil {
p.producer = producer
return &p
}
//TODO If the password is wrong, exit directly
//if packetErr, ok := err.(*sarama.PacketEncodingError); ok {
//if _, ok := packetErr.Err.(sarama.AuthenticationError); ok {
// fmt.Println("Kafka password is wrong.")
//}
//} else {
// fmt.Printf("Failed to create Kafka producer: %v\n", err)
//}
time.Sleep(time.Duration(1) * time.Second)
}
p.producer = producer
return &p

@ -31,6 +31,7 @@ import (
"github.com/OpenIMSDK/Open-IM-Server/pkg/errs"
)
// CorsHandler gin cross-domain configuration.
func CorsHandler() gin.HandlerFunc {
return func(c *gin.Context) {
c.Writer.Header().Set("Access-Control-Allow-Origin", "*")
@ -39,19 +40,19 @@ func CorsHandler() gin.HandlerFunc {
c.Header(
"Access-Control-Expose-Headers",
"Content-Length, Access-Control-Allow-Origin, Access-Control-Allow-Headers,Cache-Control,Content-Language,Content-Type,Expires,Last-Modified,Pragma,FooBar",
) // 跨域关键设置 让浏览器可以解析
) // Cross-domain key settings allow browsers to resolve.
c.Header(
"Access-Control-Max-Age",
"172800",
) // 缓存请求信息 单位为秒
) // Cache request information in seconds.
c.Header(
"Access-Control-Allow-Credentials",
"false",
) // 跨域请求是否需要带cookie信息 默认设置为true
) // Whether cross-domain requests need to carry cookie information, the default setting is true.
c.Header(
"content-type",
"application/json",
) // 设置返回格式是json
) // Set the return format to json.
//Release all option pre-requests
if c.Request.Method == http.MethodOptions {
c.JSON(http.StatusOK, "Options Request!")

Loading…
Cancel
Save