logger lock should be more granular (#14901)

This PR simplifies few things by splitting
the locks between audit, logger targets to
avoid potential contention between them.

any failures inside audit/logger HTTP
targets must only log to console instead
of other targets to avoid cyclical dependency.

avoids unneeded atomic variables instead
uses RWLock to differentiate a more common
read phase v/s lock phase.
This commit is contained in:
Harshavardhana 2022-05-12 07:20:58 -07:00 committed by GitHub
parent 88dd83a365
commit 9341201132
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 114 additions and 58 deletions

View File

@ -639,7 +639,7 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
userAgent := getUserAgent(getMinioMode()) userAgent := getUserAgent(getMinioMode())
for n, l := range loggerCfg.HTTP { for n, l := range loggerCfg.HTTP {
if l.Enabled { if l.Enabled {
l.LogOnce = logger.LogOnceIf l.LogOnce = logger.LogOnceConsoleIf
l.UserAgent = userAgent l.UserAgent = userAgent
l.Transport = NewGatewayHTTPTransportWithClientCerts(l.ClientCert, l.ClientKey) l.Transport = NewGatewayHTTPTransportWithClientCerts(l.ClientCert, l.ClientKey)
loggerCfg.HTTP[n] = l loggerCfg.HTTP[n] = l
@ -657,7 +657,7 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
userAgent := getUserAgent(getMinioMode()) userAgent := getUserAgent(getMinioMode())
for n, l := range loggerCfg.AuditWebhook { for n, l := range loggerCfg.AuditWebhook {
if l.Enabled { if l.Enabled {
l.LogOnce = logger.LogOnceIf l.LogOnce = logger.LogOnceConsoleIf
l.UserAgent = userAgent l.UserAgent = userAgent
l.Transport = NewGatewayHTTPTransportWithClientCerts(l.ClientCert, l.ClientKey) l.Transport = NewGatewayHTTPTransportWithClientCerts(l.ClientCert, l.ClientKey)
loggerCfg.AuditWebhook[n] = l loggerCfg.AuditWebhook[n] = l

View File

@ -53,6 +53,7 @@ import (
"sync" "sync"
"testing" "testing"
"time" "time"
"unsafe"
"github.com/fatih/color" "github.com/fatih/color"
@ -1183,12 +1184,11 @@ func newTestSignedRequestV4(method, urlStr string, contentLength int64, body io.
return req, nil return req, nil
} }
// Function to generate random string for bucket/object names. var src = rand.NewSource(time.Now().UnixNano())
func randString(n int) string {
src := rand.NewSource(UTCNow().UnixNano())
func randString(n int) string {
b := make([]byte, n) b := make([]byte, n)
// A rand.Int63() generates 63 random bits, enough for letterIdxMax letters! // A src.Int63() generates 63 random bits, enough for letterIdxMax characters!
for i, cache, remain := n-1, src.Int63(), letterIdxMax; i >= 0; { for i, cache, remain := n-1, src.Int63(), letterIdxMax; i >= 0; {
if remain == 0 { if remain == 0 {
cache, remain = src.Int63(), letterIdxMax cache, remain = src.Int63(), letterIdxMax
@ -1200,7 +1200,8 @@ func randString(n int) string {
cache >>= letterIdxBits cache >>= letterIdxBits
remain-- remain--
} }
return string(b)
return *(*string)(unsafe.Pointer(&b))
} }
// generate random object name. // generate random object name.

View File

@ -24,7 +24,6 @@ import (
"io" "io"
"net/http" "net/http"
"strconv" "strconv"
"sync/atomic"
"time" "time"
"github.com/klauspost/compress/gzhttp" "github.com/klauspost/compress/gzhttp"
@ -158,13 +157,12 @@ func GetAuditEntry(ctx context.Context) *audit.Entry {
// AuditLog - logs audit logs to all audit targets. // AuditLog - logs audit logs to all audit targets.
func AuditLog(ctx context.Context, w http.ResponseWriter, r *http.Request, reqClaims map[string]interface{}, filterKeys ...string) { func AuditLog(ctx context.Context, w http.ResponseWriter, r *http.Request, reqClaims map[string]interface{}, filterKeys ...string) {
// Fast exit if there is not audit target configured auditTgts := AuditTargets()
if atomic.LoadInt32(&nAuditTargets) == 0 { if len(auditTgts) == 0 {
return return
} }
var entry audit.Entry var entry audit.Entry
if w != nil && r != nil { if w != nil && r != nil {
reqInfo := GetReqInfo(ctx) reqInfo := GetReqInfo(ctx)
if reqInfo == nil { if reqInfo == nil {
@ -234,7 +232,7 @@ func AuditLog(ctx context.Context, w http.ResponseWriter, r *http.Request, reqCl
} }
// Send audit logs only to http targets. // Send audit logs only to http targets.
for _, t := range AuditTargets() { for _, t := range auditTgts {
if err := t.Send(entry, string(All)); err != nil { if err := t.Send(entry, string(All)); err != nil {
LogAlwaysIf(context.Background(), fmt.Errorf("event(%v) was not sent to Audit target (%v): %v", entry, t, err), All) LogAlwaysIf(context.Background(), fmt.Errorf("event(%v) was not sent to Audit target (%v): %v", entry, t, err), All)
} }

View File

@ -278,12 +278,7 @@ func LogIf(ctx context.Context, err error, errKind ...interface{}) {
logIf(ctx, err, errKind...) logIf(ctx, err, errKind...)
} }
// logIf prints a detailed error message during func errToEntry(ctx context.Context, err error, errKind ...interface{}) log.Entry {
// the execution of the server.
func logIf(ctx context.Context, err error, errKind ...interface{}) {
if Disable {
return
}
logKind := string(Minio) logKind := string(Minio)
if len(errKind) > 0 { if len(errKind) > 0 {
if ek, ok := errKind[0].(Kind); ok { if ek, ok := errKind[0].(Kind); ok {
@ -357,8 +352,37 @@ func logIf(ctx context.Context, err error, errKind ...interface{}) {
entry.Trace.Variables = make(map[string]interface{}) entry.Trace.Variables = make(map[string]interface{})
} }
return entry
}
// consoleLogIf prints a detailed error message during
// the execution of the server.
func consoleLogIf(ctx context.Context, err error, errKind ...interface{}) {
if Disable {
return
}
if consoleTgt != nil {
entry := errToEntry(ctx, err, errKind...)
consoleTgt.Send(entry, entry.LogKind)
}
}
// logIf prints a detailed error message during
// the execution of the server.
func logIf(ctx context.Context, err error, errKind ...interface{}) {
if Disable {
return
}
systemTgts := SystemTargets()
if len(systemTgts) == 0 {
return
}
entry := errToEntry(ctx, err, errKind...)
// Iterate over all logger targets to send the log entry // Iterate over all logger targets to send the log entry
for _, t := range SystemTargets() { for _, t := range systemTgts {
if err := t.Send(entry, entry.LogKind); err != nil { if err := t.Send(entry, entry.LogKind); err != nil {
if consoleTgt != nil { if consoleTgt != nil {
entry.Trace.Message = fmt.Sprintf("event(%#v) was not sent to Logger target (%#v): %#v", entry, t, err) entry.Trace.Message = fmt.Sprintf("event(%#v) was not sent to Logger target (%#v): %#v", entry, t, err)

View File

@ -31,6 +31,27 @@ type logOnceType struct {
sync.Mutex sync.Mutex
} }
func (l *logOnceType) logOnceConsoleIf(ctx context.Context, err error, id interface{}, errKind ...interface{}) {
if err == nil {
return
}
l.Lock()
shouldLog := false
prevErr := l.IDMap[id]
if prevErr == nil {
l.IDMap[id] = err
shouldLog = true
} else if prevErr.Error() != err.Error() {
l.IDMap[id] = err
shouldLog = true
}
l.Unlock()
if shouldLog {
consoleLogIf(ctx, err, errKind...)
}
}
// One log message per error. // One log message per error.
func (l *logOnceType) logOnceIf(ctx context.Context, err error, id interface{}, errKind ...interface{}) { func (l *logOnceType) logOnceIf(ctx context.Context, err error, id interface{}, errKind ...interface{}) {
if err == nil { if err == nil {
@ -91,3 +112,20 @@ func LogOnceIf(ctx context.Context, err error, id interface{}, errKind ...interf
logOnce.logOnceIf(ctx, err, id, errKind...) logOnce.logOnceIf(ctx, err, id, errKind...)
} }
// LogOnceConsoleIf - similar to LogOnceIf but exclusively only logs to console target.
func LogOnceConsoleIf(ctx context.Context, err error, id interface{}, errKind ...interface{}) {
if err == nil {
return
}
if errors.Is(err, context.Canceled) {
return
}
if err.Error() == http.ErrServerClosed.Error() || err.Error() == "disk not found" {
return
}
logOnce.logOnceConsoleIf(ctx, err, id, errKind...)
}

View File

@ -19,7 +19,6 @@ package logger
import ( import (
"sync" "sync"
"sync/atomic"
"github.com/minio/minio/internal/logger/target/http" "github.com/minio/minio/internal/logger/target/http"
"github.com/minio/minio/internal/logger/target/kafka" "github.com/minio/minio/internal/logger/target/kafka"
@ -39,8 +38,8 @@ type Target interface {
} }
var ( var (
// swapMu must be held while reading slice info or swapping targets or auditTargets. swapAuditMuRW sync.RWMutex
swapMu sync.Mutex swapSystemMuRW sync.RWMutex
// systemTargets is the set of enabled loggers. // systemTargets is the set of enabled loggers.
// Must be immutable at all times. // Must be immutable at all times.
@ -49,33 +48,25 @@ var (
// This is always set represent /dev/console target // This is always set represent /dev/console target
consoleTgt Target consoleTgt Target
nTargets int32 // atomic count of len(targets)
) )
// SystemTargets returns active targets. // SystemTargets returns active targets.
// Returned slice may not be modified in any way. // Returned slice may not be modified in any way.
func SystemTargets() []Target { func SystemTargets() []Target {
if atomic.LoadInt32(&nTargets) == 0 { swapSystemMuRW.RLock()
// Lock free if none... defer swapSystemMuRW.RUnlock()
return nil
}
swapMu.Lock()
res := systemTargets res := systemTargets
swapMu.Unlock()
return res return res
} }
// AuditTargets returns active audit targets. // AuditTargets returns active audit targets.
// Returned slice may not be modified in any way. // Returned slice may not be modified in any way.
func AuditTargets() []Target { func AuditTargets() []Target {
if atomic.LoadInt32(&nAuditTargets) == 0 { swapAuditMuRW.RLock()
// Lock free if none... defer swapAuditMuRW.RUnlock()
return nil
}
swapMu.Lock()
res := auditTargets res := auditTargets
swapMu.Unlock()
return res return res
} }
@ -83,8 +74,7 @@ func AuditTargets() []Target {
// Must be immutable at all times. // Must be immutable at all times.
// Can be swapped to another while holding swapMu // Can be swapped to another while holding swapMu
var ( var (
auditTargets = []Target{} auditTargets = []Target{}
nAuditTargets int32 // atomic count of len(auditTargets)
) )
// AddSystemTarget adds a new logger target to the // AddSystemTarget adds a new logger target to the
@ -93,7 +83,10 @@ func AddSystemTarget(t Target) error {
if err := t.Init(); err != nil { if err := t.Init(); err != nil {
return err return err
} }
swapMu.Lock()
swapSystemMuRW.Lock()
defer swapSystemMuRW.Unlock()
if consoleTgt == nil { if consoleTgt == nil {
if t.Type() == types.TargetConsole { if t.Type() == types.TargetConsole {
consoleTgt = t consoleTgt = t
@ -102,8 +95,6 @@ func AddSystemTarget(t Target) error {
updated := append(make([]Target, 0, len(systemTargets)+1), systemTargets...) updated := append(make([]Target, 0, len(systemTargets)+1), systemTargets...)
updated = append(updated, t) updated = append(updated, t)
systemTargets = updated systemTargets = updated
atomic.StoreInt32(&nTargets, int32(len(updated)))
swapMu.Unlock()
return nil return nil
} }
@ -142,24 +133,26 @@ func initKafkaTargets(cfgMap map[string]kafka.Config) (tgts []Target, err error)
// UpdateSystemTargets swaps targets with newly loaded ones from the cfg // UpdateSystemTargets swaps targets with newly loaded ones from the cfg
func UpdateSystemTargets(cfg Config) error { func UpdateSystemTargets(cfg Config) error {
updated, err := initSystemTargets(cfg.HTTP) newTgts, err := initSystemTargets(cfg.HTTP)
if err != nil { if err != nil {
return err return err
} }
swapMu.Lock() swapSystemMuRW.Lock()
defer swapSystemMuRW.Unlock()
for _, tgt := range systemTargets { for _, tgt := range systemTargets {
// Preserve console target when dynamically updating // Preserve console target when dynamically updating
// other HTTP targets, console target is always present. // other HTTP targets, console target is always present.
if tgt.Type() == types.TargetConsole { if tgt.Type() == types.TargetConsole {
updated = append(updated, tgt) newTgts = append(newTgts, tgt)
break break
} }
} }
atomic.StoreInt32(&nTargets, int32(len(updated)))
cancelAllSystemTargets() // cancel running targets cancelAllSystemTargets() // cancel running targets
systemTargets = updated systemTargets = newTgts
swapMu.Unlock()
return nil return nil
} }
@ -183,18 +176,19 @@ func existingAuditTargets(t types.TargetType) []Target {
// UpdateAuditWebhookTargets swaps audit webhook targets with newly loaded ones from the cfg // UpdateAuditWebhookTargets swaps audit webhook targets with newly loaded ones from the cfg
func UpdateAuditWebhookTargets(cfg Config) error { func UpdateAuditWebhookTargets(cfg Config) error {
updated, err := initSystemTargets(cfg.AuditWebhook) newTgts, err := initSystemTargets(cfg.AuditWebhook)
if err != nil { if err != nil {
return err return err
} }
// retain kafka targets
updated = append(existingAuditTargets(types.TargetKafka), updated...)
swapMu.Lock() // retain kafka targets
atomic.StoreInt32(&nAuditTargets, int32(len(updated))) swapAuditMuRW.Lock()
defer swapAuditMuRW.Unlock()
newTgts = append(existingAuditTargets(types.TargetKafka), newTgts...)
cancelAuditTargetType(types.TargetHTTP) // cancel running targets cancelAuditTargetType(types.TargetHTTP) // cancel running targets
auditTargets = updated auditTargets = newTgts
swapMu.Unlock()
return nil return nil
} }
@ -204,13 +198,14 @@ func UpdateAuditKafkaTargets(cfg Config) error {
if err != nil { if err != nil {
return err return err
} }
swapAuditMuRW.Lock()
defer swapAuditMuRW.Unlock()
// retain HTTP targets // retain HTTP targets
updated = append(existingAuditTargets(types.TargetHTTP), updated...) updated = append(existingAuditTargets(types.TargetHTTP), updated...)
swapMu.Lock()
atomic.StoreInt32(&nAuditTargets, int32(len(updated)))
cancelAuditTargetType(types.TargetKafka) // cancel running targets cancelAuditTargetType(types.TargetKafka) // cancel running targets
auditTargets = updated auditTargets = updated
swapMu.Unlock()
return nil return nil
} }