mirror of
https://github.com/tailscale/tailscale.git
synced 2025-07-31 16:23:44 +00:00
Replace time-dependent cache limiting with LRU
To keep memory usage as constant as possible, the previous cache purging at periodic time intervals has been replaced by an LRU that discards the oldest string when the capacity of the cache is reached. Also fixed some formatting and commenting issues, and added better testing for the rate limiter. Signed-off-by: Wendi Yu <wendi.yu@yahoo.ca>
This commit is contained in:
parent
ca6675386b
commit
8a94ccc5a2
@ -51,8 +51,8 @@ func main() {
|
|||||||
statepath := getopt.StringLong("state", 0, paths.DefaultTailscaledStateFile(), "Path of state file")
|
statepath := getopt.StringLong("state", 0, paths.DefaultTailscaledStateFile(), "Path of state file")
|
||||||
socketpath := getopt.StringLong("socket", 's', paths.DefaultTailscaledSocket(), "Path of the service unix socket")
|
socketpath := getopt.StringLong("socket", 's', paths.DefaultTailscaledSocket(), "Path of the service unix socket")
|
||||||
|
|
||||||
rlPrint := logger.RateLimitedFn(log.Printf, 1, 1)
|
logf := wgengine.RusagePrefixLog(log.Printf)
|
||||||
logf := wgengine.RusagePrefixLog(rlPrint)
|
logf = logger.RateLimitedFn(logf, 1, 1, 100)
|
||||||
|
|
||||||
err := fixconsole.FixConsoleIfNeeded()
|
err := fixconsole.FixConsoleIfNeeded()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -193,15 +193,15 @@ type testNode struct {
|
|||||||
func newNode(t *testing.T, prefix string, https *httptest.Server, weirdPrefs bool) testNode {
|
func newNode(t *testing.T, prefix string, https *httptest.Server, weirdPrefs bool) testNode {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
ulogfe := func(fmt string, args ...interface{}) {
|
logfe := func(fmt string, args ...interface{}) {
|
||||||
t.Logf(prefix+".e: "+fmt, args...)
|
t.Logf(prefix+".e: "+fmt, args...)
|
||||||
}
|
}
|
||||||
logfe := logger.RateLimitedFn(ulogfe, 1, 1)
|
logfe := logger.RateLimitedFn(logfe, 1, 1)
|
||||||
|
|
||||||
ulogf := func(fmt string, args ...interface{}) {
|
logf := func(fmt string, args ...interface{}) {
|
||||||
t.Logf(prefix+": "+fmt, args...)
|
t.Logf(prefix+": "+fmt, args...)
|
||||||
}
|
}
|
||||||
logf := logger.RateLimitedFn(ulogf, 1, 1)
|
logf := logger.RateLimitedFn(logf, 1, 1, 100)
|
||||||
|
|
||||||
var err error
|
var err error
|
||||||
httpc := https.Client()
|
httpc := https.Client()
|
||||||
|
@ -40,7 +40,7 @@ func TestRunMultipleAccepts(t *testing.T) {
|
|||||||
t.Logf(format, args...)
|
t.Logf(format, args...)
|
||||||
}
|
}
|
||||||
|
|
||||||
logf := logger.RateLimitedFn(ulogf, 1, 1)
|
logf := logger.RateLimitedFn(ulogf, 1, 1, 100)
|
||||||
|
|
||||||
connect := func() {
|
connect := func() {
|
||||||
for i := 1; i <= 2; i++ {
|
for i := 1; i <= 2; i++ {
|
||||||
|
@ -8,16 +8,18 @@
|
|||||||
package logger
|
package logger
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"container/list"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"log"
|
"log"
|
||||||
"time"
|
"sync"
|
||||||
|
|
||||||
"golang.org/x/time/rate"
|
"golang.org/x/time/rate"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Logf is the basic Tailscale logger type: a printf-like func.
|
// Logf is the basic Tailscale logger type: a printf-like func.
|
||||||
// Like log.Printf, the format need not end in a newline.
|
// Like log.Printf, the format need not end in a newline.
|
||||||
|
// Logf functions should be safe for concurrent use.
|
||||||
type Logf func(format string, args ...interface{})
|
type Logf func(format string, args ...interface{})
|
||||||
|
|
||||||
// WithPrefix wraps f, prefixing each format with the provided prefix.
|
// WithPrefix wraps f, prefixing each format with the provided prefix.
|
||||||
@ -47,60 +49,66 @@ func (w funcWriter) Write(p []byte) (int, error) {
|
|||||||
// Discard is a Logf that throws away the logs given to it.
|
// Discard is a Logf that throws away the logs given to it.
|
||||||
func Discard(string, ...interface{}) {}
|
func Discard(string, ...interface{}) {}
|
||||||
|
|
||||||
|
// limitData is used to keep track of each format string's associated
|
||||||
|
// rate-limiting data.
|
||||||
type limitData struct {
|
type limitData struct {
|
||||||
lim *rate.Limiter
|
lim *rate.Limiter // the token bucket associated with this string
|
||||||
lastAccessed time.Time
|
msgBlocked bool // whether a "duplicate error" message has already been logged
|
||||||
msgBlocked bool
|
ele *list.Element // list element used to access this string in the cache
|
||||||
}
|
}
|
||||||
|
|
||||||
// RateLimitedFn implements rate limiting by fstring on a given Logf.
|
// RateLimitedFn implements rate limiting by fstring on a given Logf.
|
||||||
// Messages are allowed through at a maximum of r messages/second, in
|
// Messages are allowed through at a maximum of f messages/second, in
|
||||||
// bursts of up to b messages at a time.
|
// bursts of up to b messages at a time. Up to m strings will be held at a time.
|
||||||
func RateLimitedFn(logf Logf, f float64, b int) Logf {
|
func RateLimitedFn(logf Logf, f float64, b int, m int) Logf {
|
||||||
r := rate.Limit(f)
|
r := rate.Limit(f)
|
||||||
msgList := make(map[string]limitData)
|
msgLim := make(map[string]*limitData)
|
||||||
lastPurge := time.Now()
|
msgCache := list.New() // a rudimentary LRU that limits the size of the map
|
||||||
|
mu := &sync.Mutex{}
|
||||||
rlLogf := func(s string, args ...interface{}) {
|
|
||||||
if rl, ok := msgList[s]; ok {
|
|
||||||
|
|
||||||
// Fields of structs contained in maps can't be modified; this is
|
|
||||||
// the workaround. See issue https://github.com/golang/go/issues/3117
|
|
||||||
temp := msgList[s]
|
|
||||||
temp.lastAccessed = time.Now()
|
|
||||||
msgList[s] = temp
|
|
||||||
|
|
||||||
|
return func(format string, args ...interface{}) {
|
||||||
|
mu.Lock()
|
||||||
|
rl, ok := msgLim[format]
|
||||||
|
if ok {
|
||||||
|
msgCache.MoveToFront(rl.ele)
|
||||||
if rl.lim.Allow() {
|
if rl.lim.Allow() {
|
||||||
|
mu.Lock()
|
||||||
rl.msgBlocked = false
|
rl.msgBlocked = false
|
||||||
logf(s, args)
|
mu.Unlock()
|
||||||
|
logf(format, args...)
|
||||||
} else {
|
} else {
|
||||||
if !rl.msgBlocked {
|
if !rl.msgBlocked {
|
||||||
temp = msgList[s]
|
rl.msgBlocked = true
|
||||||
temp.msgBlocked = true
|
mu.Unlock()
|
||||||
msgList[s] = temp
|
logf("Repeated messages were suppressed by rate limiting. Original message: %s",
|
||||||
logf("Repeated messages were suppressed by rate limiting. Original message: " +
|
fmt.Sprintf(format, args...))
|
||||||
fmt.Sprintf(s, args))
|
} else {
|
||||||
|
mu.Unlock()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
msgList[s] = limitData{rate.NewLimiter(r, b), time.Now(), false}
|
msgLim[format] = &limitData{rate.NewLimiter(r, b), false, msgCache.PushFront(format)}
|
||||||
msgList[s].lim.Allow()
|
msgLim[format].lim.Allow()
|
||||||
logf(s, args)
|
mu.Unlock()
|
||||||
|
logf(format, args...)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Purge msgList of outdated keys to reduce overhead. Must be done by copying
|
mu.Lock()
|
||||||
// over to a new map, since deleting in maps is done through a zombie flag
|
if msgCache.Len() > m {
|
||||||
if time.Since(lastPurge) >= time.Minute {
|
msgCache.Remove(msgCache.Back())
|
||||||
newList := make(map[string]limitData)
|
}
|
||||||
for k, v := range msgList {
|
|
||||||
if time.Since(v.lastAccessed) < 5*time.Second {
|
// Purge msgLim of outdated keys to reduce overhead. Must be done by copying
|
||||||
newList[k] = v
|
// over to a new map and allowing the garbage collector to eat the entire old one,
|
||||||
}
|
// since deleting keys in maps is done through a "zombie flag" on the data rather than
|
||||||
|
// actually clearing it from memory. See https://github.com/golang/go/issues/20135
|
||||||
|
if len(msgLim)-msgCache.Len() > 100 {
|
||||||
|
newList := make(map[string]*limitData)
|
||||||
|
for e := msgCache.Front(); e != nil; e = e.Next() {
|
||||||
|
newList[e.Value.(string)] = msgLim[e.Value.(string)]
|
||||||
}
|
}
|
||||||
msgList = nil
|
msgLim = newList
|
||||||
msgList = newList
|
|
||||||
}
|
}
|
||||||
|
mu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
return rlLogf
|
|
||||||
}
|
}
|
||||||
|
@ -5,9 +5,9 @@
|
|||||||
package logger
|
package logger
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestFuncWriter(t *testing.T) {
|
func TestFuncWriter(t *testing.T) {
|
||||||
@ -22,17 +22,41 @@ func TestStdLogger(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestRateLimiter(t *testing.T) {
|
func TestRateLimiter(t *testing.T) {
|
||||||
lg := RateLimitedFn(t.Logf, 1, 1)
|
|
||||||
|
// Testing function. args[0] should indicate what should
|
||||||
|
logTester := func(want []string) Logf {
|
||||||
|
i := 0
|
||||||
|
return func(format string, args ...interface{}) {
|
||||||
|
got := fmt.Sprintf(format, args...)
|
||||||
|
if i >= len(want) {
|
||||||
|
t.Fatalf("Logging continued past end of expected input: %s", got)
|
||||||
|
}
|
||||||
|
if got != want[i] {
|
||||||
|
t.Fatalf("wanted: %s \n got: %s", want[i], got)
|
||||||
|
}
|
||||||
|
i++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
want := []string{
|
||||||
|
"boring string with constant formatting (constant)",
|
||||||
|
"templated format string no. 0",
|
||||||
|
"Repeated messages were suppressed by rate limiting. Original message: boring string with constant formatting (constant)",
|
||||||
|
"Repeated messages were suppressed by rate limiting. Original message: templated format string no. 1",
|
||||||
|
"Make sure this string makes it through the rest (that are blocked) 4",
|
||||||
|
"4 shouldn't get filtered.",
|
||||||
|
}
|
||||||
|
|
||||||
|
lg := RateLimitedFn(logTester(want), 1, 1, 50)
|
||||||
var prefixed Logf
|
var prefixed Logf
|
||||||
for i := 0; i < 10; i++ {
|
for i := 0; i < 10; i++ {
|
||||||
lg("boring string with no formatting")
|
lg("boring string with constant formatting %s", "(constant)")
|
||||||
lg("templated format string no. %d", i)
|
lg("templated format string no. %d", i)
|
||||||
if i == 4 {
|
if i == 4 {
|
||||||
lg("Make sure this string makes it through the rest (that are blocked) %d", i)
|
lg("Make sure this string makes it through the rest (that are blocked) %d", i)
|
||||||
|
prefixed = WithPrefix(lg, string('0'+i))
|
||||||
|
prefixed(" shouldn't get filtered.")
|
||||||
}
|
}
|
||||||
prefixed = WithPrefix(lg, string('0'+i))
|
|
||||||
prefixed(" shouldn't get filtered.")
|
|
||||||
time.Sleep(200 * time.Millisecond)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
Loading…
x
Reference in New Issue
Block a user