github_mirrors/go-zero
1
0
Fork 0
mirror of https://github.com/zeromicro/go-zero.git synced 2025-02-02 16:28:39 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: use breaker with ctx to prevent deadline exceeded (#4091)

Browse Source 
Signed-off-by: kevin <wanjunfeng@gmail.com>
This commit is contained in:
Kevin Wan 2024-04-18 23:18:49 +08:00 committed by GitHub
parent 1b9b61f505
commit d371ab5479
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 47 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
core/stores/mon/collection.go
View File

@ -141,7 +141,7 @@ func (c *decoratedCollection) Aggregate(ctx context.Context, pipeline any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
starTime := timex.Now() starTime := timex.Now()
defer func() { defer func() {
c.logDurationSimple(ctx, aggregate, starTime, err) c.logDurationSimple(ctx, aggregate, starTime, err)
@ -161,7 +161,7 @@ func (c *decoratedCollection) BulkWrite(ctx context.Context, models []mongo.Writ
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDurationSimple(ctx, bulkWrite, startTime, err) c.logDurationSimple(ctx, bulkWrite, startTime, err)
@ -181,7 +181,7 @@ func (c *decoratedCollection) CountDocuments(ctx context.Context, filter any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDurationSimple(ctx, countDocuments, startTime, err) c.logDurationSimple(ctx, countDocuments, startTime, err)
@ -201,7 +201,7 @@ func (c *decoratedCollection) DeleteMany(ctx context.Context, filter any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDurationSimple(ctx, deleteMany, startTime, err) c.logDurationSimple(ctx, deleteMany, startTime, err)
@ -221,7 +221,7 @@ func (c *decoratedCollection) DeleteOne(ctx context.Context, filter any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, deleteOne, startTime, err, filter) c.logDuration(ctx, deleteOne, startTime, err, filter)
@ -241,7 +241,7 @@ func (c *decoratedCollection) Distinct(ctx context.Context, fieldName string, fi
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDurationSimple(ctx, distinct, startTime, err) c.logDurationSimple(ctx, distinct, startTime, err)
@ -261,7 +261,7 @@ func (c *decoratedCollection) EstimatedDocumentCount(ctx context.Context,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDurationSimple(ctx, estimatedDocumentCount, startTime, err) c.logDurationSimple(ctx, estimatedDocumentCount, startTime, err)
@ -281,7 +281,7 @@ func (c *decoratedCollection) Find(ctx context.Context, filter any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, find, startTime, err, filter) c.logDuration(ctx, find, startTime, err, filter)
@ -301,7 +301,7 @@ func (c *decoratedCollection) FindOne(ctx context.Context, filter any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, findOne, startTime, err, filter) c.logDuration(ctx, findOne, startTime, err, filter)
@ -322,7 +322,7 @@ func (c *decoratedCollection) FindOneAndDelete(ctx context.Context, filter any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, findOneAndDelete, startTime, err, filter) c.logDuration(ctx, findOneAndDelete, startTime, err, filter)
@ -344,7 +344,7 @@ func (c *decoratedCollection) FindOneAndReplace(ctx context.Context, filter any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, findOneAndReplace, startTime, err, filter, replacement) c.logDuration(ctx, findOneAndReplace, startTime, err, filter, replacement)
@ -365,7 +365,7 @@ func (c *decoratedCollection) FindOneAndUpdate(ctx context.Context, filter, upda
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, findOneAndUpdate, startTime, err, filter, update) c.logDuration(ctx, findOneAndUpdate, startTime, err, filter, update)
@ -386,7 +386,7 @@ func (c *decoratedCollection) InsertMany(ctx context.Context, documents []any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDurationSimple(ctx, insertMany, startTime, err) c.logDurationSimple(ctx, insertMany, startTime, err)
@ -406,7 +406,7 @@ func (c *decoratedCollection) InsertOne(ctx context.Context, document any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, insertOne, startTime, err, document) c.logDuration(ctx, insertOne, startTime, err, document)
@ -426,7 +426,7 @@ func (c *decoratedCollection) ReplaceOne(ctx context.Context, filter, replacemen
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, replaceOne, startTime, err, filter, replacement) c.logDuration(ctx, replaceOne, startTime, err, filter, replacement)
@ -446,7 +446,7 @@ func (c *decoratedCollection) UpdateByID(ctx context.Context, id, update any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, updateByID, startTime, err, id, update) c.logDuration(ctx, updateByID, startTime, err, id, update)
@ -466,7 +466,7 @@ func (c *decoratedCollection) UpdateMany(ctx context.Context, filter, update any
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDurationSimple(ctx, updateMany, startTime, err) c.logDurationSimple(ctx, updateMany, startTime, err)
@ -486,7 +486,7 @@ func (c *decoratedCollection) UpdateOne(ctx context.Context, filter, update any,
endSpan(span, err) endSpan(span, err)
}() }()
err = c.brk.DoWithAcceptable(func() error { err = c.brk.DoWithAcceptableCtx(ctx, func() error {
startTime := timex.Now() startTime := timex.Now()
defer func() { defer func() {
c.logDuration(ctx, updateOne, startTime, err, filter, update) c.logDuration(ctx, updateOne, startTime, err, filter, update)

20
core/stores/mon/model.go
View File

@ -69,7 +69,6 @@ func newModel(name string, cli *mongo.Client, coll Collection, brk breaker.Break
// StartSession starts a new session. // StartSession starts a new session.
func (m *Model) StartSession(opts ...*mopt.SessionOptions) (sess mongo.Session, err error) { func (m *Model) StartSession(opts ...*mopt.SessionOptions) (sess mongo.Session, err error) {
err = m.brk.DoWithAcceptable(func() error {
starTime := timex.Now() starTime := timex.Now()
defer func() { defer func() {
logDuration(context.Background(), m.name, startSession, starTime, err) logDuration(context.Background(), m.name, startSession, starTime, err)
@ -77,19 +76,14 @@ func (m *Model) StartSession(opts ...*mopt.SessionOptions) (sess mongo.Session,
session, sessionErr := m.cli.StartSession(opts...) session, sessionErr := m.cli.StartSession(opts...)
if sessionErr != nil { if sessionErr != nil {
return sessionErr return nil, sessionErr
} }
sess = &wrappedSession{ return &wrappedSession{
Session: session, Session: session,
name: m.name, name: m.name,
brk: m.brk, brk: m.brk,
} }, nil
return nil
}, acceptable)
return
} }
// Aggregate executes an aggregation pipeline. // Aggregate executes an aggregation pipeline.
@ -184,7 +178,7 @@ func (w *wrappedSession) AbortTransaction(ctx context.Context) (err error) {
endSpan(span, err) endSpan(span, err)
}() }()
return w.brk.DoWithAcceptable(func() error { return w.brk.DoWithAcceptableCtx(ctx, func() error {
starTime := timex.Now() starTime := timex.Now()
defer func() { defer func() {
logDuration(ctx, w.name, abortTransaction, starTime, err) logDuration(ctx, w.name, abortTransaction, starTime, err)
@ -201,7 +195,7 @@ func (w *wrappedSession) CommitTransaction(ctx context.Context) (err error) {
endSpan(span, err) endSpan(span, err)
}() }()
return w.brk.DoWithAcceptable(func() error { return w.brk.DoWithAcceptableCtx(ctx, func() error {
starTime := timex.Now() starTime := timex.Now()
defer func() { defer func() {
logDuration(ctx, w.name, commitTransaction, starTime, err) logDuration(ctx, w.name, commitTransaction, starTime, err)
@ -222,7 +216,7 @@ func (w *wrappedSession) WithTransaction(
endSpan(span, err) endSpan(span, err)
}() }()
err = w.brk.DoWithAcceptable(func() error { err = w.brk.DoWithAcceptableCtx(ctx, func() error {
starTime := timex.Now() starTime := timex.Now()
defer func() { defer func() {
logDuration(ctx, w.name, withTransaction, starTime, err) logDuration(ctx, w.name, withTransaction, starTime, err)
@ -243,7 +237,7 @@ func (w *wrappedSession) EndSession(ctx context.Context) {
endSpan(span, err) endSpan(span, err)
}() }()
err = w.brk.DoWithAcceptable(func() error { err = w.brk.DoWithAcceptableCtx(ctx, func() error {
starTime := timex.Now() starTime := timex.Now()
defer func() { defer func() {
logDuration(ctx, w.name, endSession, starTime, err) logDuration(ctx, w.name, endSession, starTime, err)

4
core/stores/redis/breakerhook.go
View File

@ -26,7 +26,7 @@ func (h breakerHook) ProcessHook(next red.ProcessHook) red.ProcessHook {
return next(ctx, cmd) return next(ctx, cmd)
} }
return h.brk.DoWithAcceptable(func() error { return h.brk.DoWithAcceptableCtx(ctx, func() error {
return next(ctx, cmd) return next(ctx, cmd)
}, acceptable) }, acceptable)
} }
@ -34,7 +34,7 @@ func (h breakerHook) ProcessHook(next red.ProcessHook) red.ProcessHook {
func (h breakerHook) ProcessPipelineHook(next red.ProcessPipelineHook) red.ProcessPipelineHook { func (h breakerHook) ProcessPipelineHook(next red.ProcessPipelineHook) red.ProcessPipelineHook {
return func(ctx context.Context, cmds []red.Cmder) error { return func(ctx context.Context, cmds []red.Cmder) error {
return h.brk.DoWithAcceptable(func() error { return h.brk.DoWithAcceptableCtx(ctx, func() error {
return next(ctx, cmds) return next(ctx, cmds)
}, acceptable) }, acceptable)
} }

10
core/stores/sqlx/sqlconn.go
View File

@ -83,7 +83,7 @@ func NewSqlConn(driverName, datasource string, opts ...SqlOption) SqlConn {
} }
// NewSqlConnFromDB returns a SqlConn with the given sql.DB. // NewSqlConnFromDB returns a SqlConn with the given sql.DB.
// Use it with caution, it's provided for other ORM to interact with. // Use it with caution; it's provided for other ORM to interact with.
func NewSqlConnFromDB(db *sql.DB, opts ...SqlOption) SqlConn { func NewSqlConnFromDB(db *sql.DB, opts ...SqlOption) SqlConn {
conn := &commonSqlConn{ conn := &commonSqlConn{
connProv: func() (*sql.DB, error) { connProv: func() (*sql.DB, error) {
@ -120,7 +120,7 @@ func (db *commonSqlConn) ExecCtx(ctx context.Context, q string, args ...any) (
endSpan(span, err) endSpan(span, err)
}() }()
err = db.brk.DoWithAcceptable(func() error { err = db.brk.DoWithAcceptableCtx(ctx, func() error {
var conn *sql.DB var conn *sql.DB
conn, err = db.connProv() conn, err = db.connProv()
if err != nil { if err != nil {
@ -148,7 +148,7 @@ func (db *commonSqlConn) PrepareCtx(ctx context.Context, query string) (stmt Stm
endSpan(span, err) endSpan(span, err)
}() }()
err = db.brk.DoWithAcceptable(func() error { err = db.brk.DoWithAcceptableCtx(ctx, func() error {
var conn *sql.DB var conn *sql.DB
conn, err = db.connProv() conn, err = db.connProv()
if err != nil { if err != nil {
@ -256,7 +256,7 @@ func (db *commonSqlConn) TransactCtx(ctx context.Context, fn func(context.Contex
endSpan(span, err) endSpan(span, err)
}() }()
err = db.brk.DoWithAcceptable(func() error { err = db.brk.DoWithAcceptableCtx(ctx, func() error {
return transact(ctx, db, db.beginTx, fn) return transact(ctx, db, db.beginTx, fn)
}, db.acceptable) }, db.acceptable)
if errors.Is(err, breaker.ErrServiceUnavailable) { if errors.Is(err, breaker.ErrServiceUnavailable) {
@ -287,7 +287,7 @@ func (db *commonSqlConn) acceptable(err error) bool {
func (db *commonSqlConn) queryRows(ctx context.Context, scanner func(*sql.Rows) error, func (db *commonSqlConn) queryRows(ctx context.Context, scanner func(*sql.Rows) error,
q string, args ...any) (err error) { q string, args ...any) (err error) {
var scanFailed bool var scanFailed bool
err = db.brk.DoWithAcceptable(func() error { err = db.brk.DoWithAcceptableCtx(ctx, func() error {
conn, err := db.connProv() conn, err := db.connProv()
if err != nil { if err != nil {
db.onError(ctx, err) db.onError(ctx, err)

4
core/stores/sqlx/stmt.go
View File

@ -65,7 +65,7 @@ func (s statement) ExecCtx(ctx context.Context, args ...any) (result sql.Result,
endSpan(span, err) endSpan(span, err)
}() }()
err = s.brk.DoWithAcceptable(func() error { err = s.brk.DoWithAcceptableCtx(ctx, func() error {
result, err = execStmt(ctx, s.stmt, s.query, args...) result, err = execStmt(ctx, s.stmt, s.query, args...)
return err return err
}, func(err error) bool { }, func(err error) bool {
@ -141,7 +141,7 @@ func (s statement) QueryRowsPartialCtx(ctx context.Context, v any, args ...any)
func (s statement) queryRows(ctx context.Context, scanFn func(any, rowsScanner) error, func (s statement) queryRows(ctx context.Context, scanFn func(any, rowsScanner) error,
v any, args ...any) error { v any, args ...any) error {
var scanFailed bool var scanFailed bool
err := s.brk.DoWithAcceptable(func() error { err := s.brk.DoWithAcceptableCtx(ctx, func() error {
return queryStmt(ctx, s.stmt, func(rows *sql.Rows) error { return queryStmt(ctx, s.stmt, func(rows *sql.Rows) error {
err := scanFn(v, rows) err := scanFn(v, rows)
if isScanFailed(err) { if isScanFailed(err) {

2
rest/httpc/service.go
View File

@ -63,7 +63,7 @@ func (s namedService) do(r *http.Request) (resp *http.Response, err error) {
} }
brk := breaker.GetBreaker(s.name) brk := breaker.GetBreaker(s.name)
err = brk.DoWithAcceptable(func() error { err = brk.DoWithAcceptableCtx(r.Context(), func() error {
resp, err = s.cli.Do(r) resp, err = s.cli.Do(r)
return err return err
}, func(err error) bool { }, func(err error) bool {

2
zrpc/internal/clientinterceptors/breakerinterceptor.go
View File

@ -13,7 +13,7 @@ import (
func BreakerInterceptor(ctx context.Context, method string, req, reply any, func BreakerInterceptor(ctx context.Context, method string, req, reply any,
cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
breakerName := path.Join(cc.Target(), method) breakerName := path.Join(cc.Target(), method)
return breaker.DoWithAcceptable(breakerName, func() error { return breaker.DoWithAcceptableCtx(ctx, breakerName, func() error {
return invoker(ctx, method, req, reply, cc, opts...) return invoker(ctx, method, req, reply, cc, opts...)
}, codes.Acceptable) }, codes.Acceptable)
} }

2
zrpc/internal/serverinterceptors/breakerinterceptor.go
View File

@ -24,7 +24,7 @@ func StreamBreakerInterceptor(svr any, stream grpc.ServerStream, info *grpc.Stre
func UnaryBreakerInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, func UnaryBreakerInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler) (resp any, err error) { handler grpc.UnaryHandler) (resp any, err error) {
breakerName := info.FullMethod breakerName := info.FullMethod
err = breaker.DoWithAcceptable(breakerName, func() error { err = breaker.DoWithAcceptableCtx(ctx, breakerName, func() error {
var err error var err error
resp, err = handler(ctx, req) resp, err = handler(ctx, req)
return err return err