vbatts/ntfy
1
0
Fork 0
forked from mirrors/ntfy
Code Pull requests Activity

More RWLock. Jeff wins again

Browse source
This commit is contained in:
binwiederhier 2023-02-09 20:49:45 -05:00
parent 057c4a3239
commit d8dd4c92bf
6 changed files with 132 additions and 92 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
cmd/app.go
View file

@ -28,7 +28,7 @@ var flagsDefault = []cli.Flag{
}
var (
logLevelOverrideRegex = regexp.MustCompile(`(?i)^([^=]+)\s*=\s*(\S+)\s*->\s*(TRACE|DEBUG|INFO|WARN|ERROR)$`)
logLevelOverrideRegex = regexp.MustCompile(`(?i)^([^=\s]+)(?:\s*=\s*(\S+))?\s*->\s*(TRACE|DEBUG|INFO|WARN|ERROR)$`)
)
// New creates a new CLI application
@ -76,11 +76,15 @@ func initLogFunc(c *cli.Context) error {
func applyLogLevelOverrides(rawOverrides []string) error {
for _, override := range rawOverrides {
m := logLevelOverrideRegex.FindStringSubmatch(override)
if len(m) != 4 {
if len(m) == 4 {
field, value, level := m[1], m[2], m[3]
log.SetLevelOverride(field, value, log.ToLevel(level))
} else if len(m) == 3 {
field, level := m[1], m[2]
log.SetLevelOverride(field, "", log.ToLevel(level)) // Matches any value
} else {
return fmt.Errorf(`invalid log level override "%s", must be "field=value -> loglevel", e.g. "user_id=u_123 -> DEBUG"`, override)
}
field, value, level := m[1], m[2], m[3]
log.SetLevelOverride(field, value, log.ToLevel(level))
}
return nil
}

4
log/event.go
View file

@ -206,9 +206,7 @@ func (e *Event) globalLevelWithOverride() Level {
for field, override := range ov {
value, exists := e.fields[field]
if exists {
if value == override.value {
return override.level
} else if fmt.Sprintf("%v", value) == override.value {
if override.value == "" || override.value == value || override.value == fmt.Sprintf("%v", value) {
return override.level
}
}

21
log/log_test.go
View file

@ -151,6 +151,27 @@ func TestLog_Timing(t *testing.T) {
require.Contains(t, out.String(), `{"time":"1970-01-01T00:00:12Z","level":"INFO","message":"A thing that takes a while","time_taken_ms":`)
}
func TestLog_LevelOverrideAny(t *testing.T) {
t.Cleanup(resetState)
var out bytes.Buffer
SetOutput(&out)
SetFormat(JSONFormat)
SetLevelOverride("this_one", "", DebugLevel)
SetLevelOverride("time_taken_ms", "", TraceLevel)
Time(time.Unix(11, 0).UTC()).Field("this_one", "11").Debug("this is logged")
Time(time.Unix(12, 0).UTC()).Field("not_this", "11").Debug("this is not logged")
Time(time.Unix(13, 0).UTC()).Field("this_too", "11").Info("this is also logged")
Time(time.Unix(14, 0).UTC()).Field("time_taken_ms", 0).Info("this is also logged")
expected := `{"time":"1970-01-01T00:00:11Z","level":"DEBUG","message":"this is logged","this_one":"11"}
{"time":"1970-01-01T00:00:13Z","level":"INFO","message":"this is also logged","this_too":"11"}
{"time":"1970-01-01T00:00:14Z","level":"INFO","message":"this is also logged","time_taken_ms":0}
`
require.Equal(t, expected, out.String())
}
type fakeError struct {
Code int
Message string

92
server/server.go
View file

@ -44,6 +44,7 @@ import (
- MEDIUM: Test new token endpoints & never-expiring token
- LOW: UI: Flickering upgrade banner when logging in
- LOW: Menu item -> popup click should not open page
- LOW: get rid of reservation id, replace with DELETE X-Topic: ...
*/
@ -143,8 +144,8 @@ const (
tagPublish = "publish"
tagSubscribe = "subscribe"
tagFirebase = "firebase"
tagEmail = "email" // Send email
tagSMTP = "smtp" // Receive email
tagEmail = "email" // Send email
tagFileCache = "file_cache"
tagMessageCache = "message_cache"
tagStripe = "stripe"
@ -323,48 +324,61 @@ func (s *Server) closeDatabases() {
s.messageCache.Close()
}
// handle is the main entry point for all HTTP requests
func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
v, err := s.maybeAuthenticate(r) // Note: Always returns v, even when error is returned
if err == nil {
logvr(v, r).Debug("Dispatching request")
if log.IsTrace() {
logvr(v, r).Trace("Entire request (headers and body):\n%s", renderHTTPRequest(r))
}
err = s.handleInternal(w, r, v)
}
if err != nil {
if websocket.IsWebSocketUpgrade(r) {
isNormalError := strings.Contains(err.Error(), "i/o timeout")
if isNormalError {
logvr(v, r).Tag(tagWebsocket).Err(err).Fields(websocketErrorContext(err)).Debug("WebSocket error (this error is okay, it happens a lot): %s", err.Error())
} else {
logvr(v, r).Tag(tagWebsocket).Err(err).Fields(websocketErrorContext(err)).Info("WebSocket error: %s", err.Error())
}
return // Do not attempt to write to upgraded connection
}
if matrixErr, ok := err.(*errMatrix); ok {
writeMatrixError(w, r, v, matrixErr)
return
}
httpErr, ok := err.(*errHTTP)
if !ok {
httpErr = errHTTPInternalError
}
isNormalError := httpErr.HTTPCode == http.StatusNotFound || httpErr.HTTPCode == http.StatusBadRequest || httpErr.HTTPCode == http.StatusTooManyRequests
if isNormalError {
logvr(v, r).
Err(httpErr).
Debug("Connection closed with HTTP %d (ntfy error %d)", httpErr.HTTPCode, httpErr.Code)
} else {
logvr(v, r).
Err(httpErr).
Info("Connection closed with HTTP %d (ntfy error %d)", httpErr.HTTPCode, httpErr.Code)
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Access-Control-Allow-Origin", s.config.AccessControlAllowOrigin) // CORS, allow cross-origin requests
w.WriteHeader(httpErr.HTTPCode)
io.WriteString(w, httpErr.JSON()+"\n")
s.handleError(w, r, v, err)
return
}
if log.IsTrace() {
logvr(v, r).Field("http_request", renderHTTPRequest(r)).Trace("HTTP request started")
} else if log.IsDebug() {
logvr(v, r).Debug("HTTP request started")
}
logvr(v, r).
Timing(func() {
if err := s.handleInternal(w, r, v); err != nil {
s.handleError(w, r, v, err)
return
}
}).
Debug("HTTP request finished")
}
func (s *Server) handleError(w http.ResponseWriter, r *http.Request, v *visitor, err error) {
if websocket.IsWebSocketUpgrade(r) {
isNormalError := strings.Contains(err.Error(), "i/o timeout")
if isNormalError {
logvr(v, r).Tag(tagWebsocket).Err(err).Fields(websocketErrorContext(err)).Debug("WebSocket error (this error is okay, it happens a lot): %s", err.Error())
} else {
logvr(v, r).Tag(tagWebsocket).Err(err).Fields(websocketErrorContext(err)).Info("WebSocket error: %s", err.Error())
}
return // Do not attempt to write to upgraded connection
}
if matrixErr, ok := err.(*errMatrix); ok {
writeMatrixError(w, r, v, matrixErr)
return
}
httpErr, ok := err.(*errHTTP)
if !ok {
httpErr = errHTTPInternalError
}
isNormalError := httpErr.HTTPCode == http.StatusNotFound || httpErr.HTTPCode == http.StatusBadRequest || httpErr.HTTPCode == http.StatusTooManyRequests
if isNormalError {
logvr(v, r).
Err(httpErr).
Debug("Connection closed with HTTP %d (ntfy error %d)", httpErr.HTTPCode, httpErr.Code)
} else {
logvr(v, r).
Err(httpErr).
Info("Connection closed with HTTP %d (ntfy error %d)", httpErr.HTTPCode, httpErr.Code)
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Access-Control-Allow-Origin", s.config.AccessControlAllowOrigin) // CORS, allow cross-origin requests
w.WriteHeader(httpErr.HTTPCode)
io.WriteString(w, httpErr.JSON()+"\n")
}
func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request, v *visitor) error {

5
server/server.yml
View file

@ -257,7 +257,9 @@
# Be aware that "debug" (and particularly "trace") can be VERY CHATTY. Only turn them on briefly for debugging purposes.
# - log-level-overrides lets you override the log level if certain fields match. This is incredibly powerful
# for debugging certain parts of the system (e.g. only the account management, or only a certain visitor).
# This is an array of strings in the format "field=value -> level", e.g. "tag=manager -> trace".
# This is an array of strings in the format:
# - "field=value -> level" to match a value exactly, e.g. "tag=manager -> trace"
# - "field -> level" to match any value, e.g. "time_taken_ms -> debug"
# Warning: Using log-level-overrides has a performance penalty. Only use it for temporary debugging.
#
# Example (good for production):
@ -269,6 +271,7 @@
# log-level-overrides:
# - "tag=manager -> trace"
# - "visitor_ip=1.2.3.4 -> debug"
# - "time_taken_ms -> debug"
#
# log-level: info
# log-level-overrides:

90
server/visitor.go
View file

@ -62,7 +62,7 @@ type visitor struct {
authLimiter *rate.Limiter // Limiter for incorrect login attempts, may be nil
firebase time.Time // Next allowed Firebase message
seen time.Time // Last seen time of this visitor (needed for removal of stale visitors)
mu sync.Mutex
mu sync.RWMutex
}
type visitorInfo struct {
@ -133,8 +133,8 @@ func newVisitor(conf *Config, messageCache *messageCache, userManager *user.Mana
}
func (v *visitor) Context() log.Context {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
return v.contextNoLock()
}
@ -184,14 +184,14 @@ func visitorExtendedInfoContext(info *visitorInfo) log.Context {
}
func (v *visitor) RequestAllowed() bool {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
return v.requestLimiter.Allow()
}
func (v *visitor) FirebaseAllowed() bool {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
return !time.Now().Before(v.firebase)
}
@ -202,27 +202,27 @@ func (v *visitor) FirebaseTemporarilyDeny() {
}
func (v *visitor) MessageAllowed() bool {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
return v.messagesLimiter.Allow()
}
func (v *visitor) EmailAllowed() bool {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
return v.emailsLimiter.Allow()
}
func (v *visitor) SubscriptionAllowed() bool {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
return v.subscriptionLimiter.Allow()
}
// AuthAllowed returns true if an auth request can be attempted (> 1 token available)
func (v *visitor) AuthAllowed() bool {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
if v.authLimiter == nil {
return true
}
@ -231,8 +231,8 @@ func (v *visitor) AuthAllowed() bool {
// AuthFailed records an auth failure
func (v *visitor) AuthFailed() {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
if v.authLimiter != nil {
v.authLimiter.Allow()
}
@ -240,8 +240,8 @@ func (v *visitor) AuthFailed() {
// AccountCreationAllowed returns true if a new account can be created
func (v *visitor) AccountCreationAllowed() bool {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
if v.accountLimiter == nil || (v.accountLimiter != nil && v.accountLimiter.Tokens() < 1) {
return false
}
@ -250,22 +250,22 @@ func (v *visitor) AccountCreationAllowed() bool {
// AccountCreated decreases the account limiter. This is to be called after an account was created.
func (v *visitor) AccountCreated() {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
if v.accountLimiter != nil {
v.accountLimiter.Allow()
}
}
func (v *visitor) BandwidthAllowed(bytes int64) bool {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
return v.bandwidthLimiter.AllowN(bytes)
}
func (v *visitor) RemoveSubscription() {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
v.subscriptionLimiter.AllowN(-1)
}
@ -276,20 +276,20 @@ func (v *visitor) Keepalive() {
}
func (v *visitor) BandwidthLimiter() util.Limiter {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
return v.bandwidthLimiter
}
func (v *visitor) Stale() bool {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
return time.Since(v.seen) > visitorExpungeAfter
}
func (v *visitor) Stats() *user.Stats {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
return &user.Stats{
Messages: v.messagesLimiter.Value(),
Emails: v.emailsLimiter.Value(),
@ -297,30 +297,30 @@ func (v *visitor) Stats() *user.Stats {
}
func (v *visitor) ResetStats() {
v.mu.Lock() // limiters could be replaced!
defer v.mu.Unlock()
v.mu.RLock() // limiters could be replaced!
defer v.mu.RUnlock()
v.emailsLimiter.Reset()
v.messagesLimiter.Reset()
}
// User returns the visitor user, or nil if there is none
func (v *visitor) User() *user.User {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
return v.user // May be nil
}
// IP returns the visitor IP address
func (v *visitor) IP() netip.Addr {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
return v.ip
}
// Authenticated returns true if a user successfully authenticated
func (v *visitor) Authenticated() bool {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
return v.user != nil
}
@ -338,8 +338,8 @@ func (v *visitor) SetUser(u *user.User) {
// MaybeUserID returns the user ID of the visitor (if any). If this is an anonymous visitor,
// an empty string is returned.
func (v *visitor) MaybeUserID() string {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
if v.user != nil {
return v.user.ID
}
@ -369,8 +369,8 @@ func (v *visitor) resetLimitersNoLock(messages, emails int64, enqueueUpdate bool
}
func (v *visitor) Limits() *visitorLimits {
v.mu.Lock()
defer v.mu.Unlock()
v.mu.RLock()
defer v.mu.RUnlock()
return v.limitsNoLock()
}
@ -422,9 +422,9 @@ func configBasedVisitorLimits(conf *Config) *visitorLimits {
}
func (v *visitor) Info() (*visitorInfo, error) {
v.mu.Lock()
v.mu.RLock()
info := v.infoLightNoLock()
v.mu.Unlock()
v.mu.RUnlock()
// Attachment stats from database
var attachmentsBytesUsed int64