diff --git a/go.mod b/go.mod index cce8163f1..335c81df1 100644 --- a/go.mod +++ b/go.mod @@ -28,6 +28,7 @@ require ( github.com/prometheus/client_model v0.6.2 github.com/prometheus/procfs v0.21.1 github.com/puzpuzpuz/xsync/v4 v4.5.0 + github.com/redis/go-redis-entraid v1.0.7 github.com/redis/go-redis/v9 v9.22.0 github.com/stretchr/testify v1.12.0 github.com/twitchtv/twirp v8.1.3+incompatible @@ -54,6 +55,10 @@ require ( buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20260415201107-50325440f8f2.1 // indirect buf.build/go/protovalidate v1.2.0 // indirect cel.dev/expr v0.25.2 // indirect + github.com/Azure/azure-sdk-for-go/sdk/azcore v1.18.0 // indirect + github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.10.0 // indirect + github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.1 // indirect + github.com/AzureAD/microsoft-authentication-library-for-go v1.4.2 // indirect github.com/antlr4-go/antlr/v4 v4.13.1 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect @@ -65,6 +70,7 @@ require ( github.com/hashicorp/go-cleanhttp v0.5.2 // indirect github.com/klauspost/compress v1.19.1 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/nats-io/nats.go v1.52.0 // indirect github.com/nats-io/nkeys v0.4.16 // indirect @@ -82,6 +88,7 @@ require ( github.com/pion/stun/v3 v3.1.6 // indirect github.com/pion/transport/v4 v4.0.2 // indirect github.com/pion/turn/v5 v5.0.12 // indirect + github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect github.com/prometheus/common v0.70.1 // indirect github.com/wlynxg/anet v0.0.5 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect diff --git a/go.sum b/go.sum index b96e9dc88..c0f19ed63 100644 --- a/go.sum +++ b/go.sum @@ -6,6 +6,18 @@ buf.build/go/protoyaml v0.7.0 h1:z4oVoFicbpPefhT7WAykxUdfp0yEQlhMQ2mCZOY5V38= buf.build/go/protoyaml v0.7.0/go.mod h1:+a0cavd0uMvirb87xdu2ZMMmjlIQoiH/N2Ich5MGSQ0= cel.dev/expr v0.25.2 h1:K6j46C81hXtZQfuX60cVWQFBJahKSE2gfRbNuvr5bFs= cel.dev/expr v0.25.2/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= +github.com/Azure/azure-sdk-for-go/sdk/azcore v1.18.0 h1:Gt0j3wceWMwPmiazCa8MzMA0MfhmPIz0Qp0FJ6qcM0U= +github.com/Azure/azure-sdk-for-go/sdk/azcore v1.18.0/go.mod h1:Ot/6aikWnKWi4l9QB7qVSwa8iMphQNqkWALMoNT3rzM= +github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.10.0 h1:j8BorDEigD8UFOSZQiSqAMOOleyQOOQPnUAwV+Ls1gA= +github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.10.0/go.mod h1:JdM5psgjfBf5fo2uWOZhflPWyDBZ/O/CNAH9CtsuZE4= +github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.2 h1:yz1bePFlP5Vws5+8ez6T3HWXPmwOK7Yvq8QxDBD3SKY= +github.com/Azure/azure-sdk-for-go/sdk/azidentity/cache v0.3.2/go.mod h1:Pa9ZNPuoNu/GztvBSKk9J1cDJW6vk/n0zLtV4mgd8N8= +github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.1 h1:FPKJS1T+clwv+OLGt13a8UjqeRuh0O4SJ3lUriThc+4= +github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.1/go.mod h1:j2chePtV91HrC22tGoRX3sGY42uF13WzmmV80/OdVAA= +github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1 h1:WJTmL004Abzc5wDB5VtZG2PJk5ndYDgVacGqfirKxjM= +github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1/go.mod h1:tCcJZ0uHAmvjsVYzEFivsRTN00oz5BEsRgQHu5JZ9WE= +github.com/AzureAD/microsoft-authentication-library-for-go v1.4.2 h1:oygO0locgZJe7PpYPXT5A29ZkwJaPqcva7BVeemZOZs= +github.com/AzureAD/microsoft-authentication-library-for-go v1.4.2/go.mod h1:wP83P5OoQ5p6ip3ScPr0BAq0BvuPAvacpEuSzyouqAI= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= @@ -73,6 +85,8 @@ github.com/hashicorp/go-retryablehttp v0.7.8 h1:ylXZWnqa7Lhqpk0L1P1LzDtGcCR0rPVU github.com/hashicorp/go-retryablehttp v0.7.8/go.mod h1:rjiScheydd+CxvumBsIrFKlx3iS0jrZ7LvzFGFmuKbw= github.com/jxskiss/base62 v1.1.0 h1:A5zbF8v8WXx2xixnAKD2w+abC+sIzYJX+nxmhA6HWFw= github.com/jxskiss/base62 v1.1.0/go.mod h1:HhWAlUXvxKThfOlZbcuFzsqwtF5TcqS9ru3y5GfjWAc= +github.com/keybase/go-keychain v0.0.1 h1:way+bWYa6lDppZoZcgMbYsvC7GxljxrskdNInRtuthU= +github.com/keybase/go-keychain v0.0.1/go.mod h1:PdEILRW3i9D8JcdM+FmY6RwkHGnhHxXwkPPMeUgOK1k= github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= @@ -81,6 +95,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/lithammer/shortuuid/v4 v4.2.0 h1:LMFOzVB3996a7b8aBuEXxqOBflbfPQAiVzkIcHO0h8c= github.com/lithammer/shortuuid/v4 v4.2.0/go.mod h1:D5noHZ2oFw/YaKCfGy0YxyE7M0wMbezmMjPdhyEFe6Y= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5ATTo469PQPkqzdoU7be46ryiCDO3boc= @@ -155,6 +171,8 @@ github.com/pion/turn/v5 v5.0.12 h1:6+b69ivQQXSlyfkp2AKripqD2k3W32qXK8QzCzpJWPI= github.com/pion/turn/v5 v5.0.12/go.mod h1:CQACsRDJtjQ+6RSrGHrS2PCIerLwbW3uqXRqOvtjAFg= github.com/pion/webrtc/v4 v4.2.18 h1:smA/3g6Gy4RohM0VIZ5KKY/12TQbxv3XFgpUMyb2EUI= github.com/pion/webrtc/v4 v4.2.18/go.mod h1:vmzi6s+rvhoIuT94DPqivB+0xJXs9rG4QRD+4MgBtlY= +github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ= +github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= @@ -167,6 +185,8 @@ github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+ github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= github.com/puzpuzpuz/xsync/v4 v4.5.0 h1:vOSWu6b57/emh+L/Cw0BeQfvxa/cogFywXHeGUxQxAg= github.com/puzpuzpuz/xsync/v4 v4.5.0/go.mod h1:VJDmTCJMBt8igNxnkQd86r+8KUeN1quSfNKu5bLYFQo= +github.com/redis/go-redis-entraid v1.0.7 h1:cU+XXeCyZ8w1AcRSlCFHXMNJNmB4TA3hKwMe/HQQLTM= +github.com/redis/go-redis-entraid v1.0.7/go.mod h1:OS6s3V1DdSRzOJEIjpK38/w4chZpl/Sy+1pzby+6nEk= github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= github.com/rodaine/protogofakeit v0.1.1 h1:ZKouljuRM3A+TArppfBqnH8tGZHOwM/pjvtXe9DaXH8= @@ -177,6 +197,8 @@ github.com/sclevine/spec v1.4.0 h1:z/Q9idDcay5m5irkZ28M7PtQM4aOISzOpj4bUPkDee8= github.com/sclevine/spec v1.4.0/go.mod h1:LvpgJaFyvQzRvc1kaDs0bulYwzC70PbiYjC4QnFHkOM= github.com/shoenig/test v1.7.0 h1:eWcHtTXa6QLnBvm0jgEabMRN/uJ4DMV3M8xUGgRkZmk= github.com/shoenig/test v1.7.0/go.mod h1:UxJ6u/x2v/TNs/LoLxBNJRV9DiwBBKYxXSyczsBHFoI= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= github.com/stretchr/testify v1.12.0 h1:K6Mr6jO9JICuend/5xzTM03ydSV3vdNRYAdPSukj8uI= github.com/stretchr/testify v1.12.0/go.mod h1:bOYBZb5qJ00vPzWfIqBUZPaxK8jWiXc6d3ErP4Ca9Gw= github.com/twitchtv/twirp v8.1.3+incompatible h1:+F4TdErPgSUbMZMwp13Q/KgDVuI7HJXP61mNV3/7iuU= @@ -231,6 +253,7 @@ golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= diff --git a/logger/config.go b/logger/config.go index 1d1e7dee0..8a8d87bbf 100644 --- a/logger/config.go +++ b/logger/config.go @@ -67,6 +67,30 @@ func (c *Config) Update(o *Config) error { return nil } +// snapshot copies the data fields so a file can be unmarshalled over the config in force. +// Update assigns every field, so decoding a partial file into a zero Config would silently reset +// the rest — including ComponentLevels, which is where livekit-server puts pion_level. +func (c *Config) snapshot() *Config { + c.lock.Lock() + defer c.lock.Unlock() + + componentLevels := make(map[string]string, len(c.ComponentLevels)) + for component, level := range c.ComponentLevels { + componentLevels[component] = level + } + return &Config{ + JSON: c.JSON, + Level: c.Level, + Sample: c.Sample, + ComponentLevels: componentLevels, + SampleInitial: c.SampleInitial, + SampleInterval: c.SampleInterval, + ItemSampleSeconds: c.ItemSampleSeconds, + ItemSampleInitial: c.ItemSampleInitial, + ItemSampleInterval: c.ItemSampleInterval, + } +} + func (c *Config) AddUpdateObserver(cb ConfigObserver) { c.lock.Lock() defer c.lock.Unlock() diff --git a/logger/configwatch.go b/logger/configwatch.go new file mode 100644 index 000000000..b9d1151b3 --- /dev/null +++ b/logger/configwatch.go @@ -0,0 +1,103 @@ +package logger + +import ( + "bytes" + "os" + "sync" + "time" + + "gopkg.in/yaml.v3" +) + +const ( + // ConfigPathEnv names a file holding the same keys as the service config's `logging` block. + // Point it inside a mounted ConfigMap to change levels without restarting: kubelet refreshes + // the mount in place, and the next poll pushes the new values into the live logger. + ConfigPathEnv = "LK_LOG_CONFIG_PATH" + // ConfigIntervalEnv overrides the poll interval as a Go duration (e.g. "10s"). + ConfigIntervalEnv = "LK_LOG_CONFIG_INTERVAL" + + defaultConfigWatchInterval = 30 * time.Second +) + +var configWatchOnce sync.Once + +// startConfigWatchFromEnv wires the watcher for the first logger the process builds, which is the +// one whose Config the service keeps. Every binary that uses this package reaches it through +// newSharedConfig, so none of them need their own flag or call site. +func startConfigWatchFromEnv(conf *Config) { + path := os.Getenv(ConfigPathEnv) + if path == "" { + return + } + interval := defaultConfigWatchInterval + if v := os.Getenv(ConfigIntervalEnv); v != "" { + if d, err := time.ParseDuration(v); err == nil && d > 0 { + interval = d + } + } + configWatchOnce.Do(func() { + WatchConfigFile(conf, path, interval) + }) +} + +// WatchConfigFile applies path to conf every interval until the returned stop is called. +// +// Polling rather than fsnotify on purpose: a ConfigMap volume update swaps the `..data` symlink +// instead of rewriting the file, so a watch on the file itself never fires. +func WatchConfigFile(conf *Config, path string, interval time.Duration) (stop func()) { + done := make(chan struct{}) + // The config the process started with. Every file is applied over this, never over whatever + // the previous file left in force, so an empty file restores the startup levels and a + // component_levels entry that disappears from the file stops applying. + baseline := conf.snapshot() + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + var last []byte + for { + select { + case <-done: + return + case <-ticker.C: + if applied, ok := applyConfigFile(conf, baseline, path, last); ok { + last = applied + } + } + } + }() + + var once sync.Once + return func() { once.Do(func() { close(done) }) } +} + +// applyConfigFile decodes path over baseline and pushes the result into conf when the bytes differ +// from last, returning the bytes it applied. ok is false when nothing was applied (unreadable, +// unchanged or invalid), and the config already in force stays untouched. +// +// Decoding over baseline rather than over conf is what makes the file a declarative overlay: keys +// it omits fall back to the startup values instead of inheriting the previous file's. +func applyConfigFile(conf, baseline *Config, path string, last []byte) (applied []byte, ok bool) { + data, err := os.ReadFile(path) + if err != nil { + // An optional ConfigMap that is not mounted yet is the normal steady state, not something + // to log once per interval forever. A file that goes away is deliberately not treated as a + // reset either: a transient read error would otherwise flap levels. Emptying the file to + // `{}` is the reset. + return nil, false + } + if bytes.Equal(data, last) { + return nil, false + } + + next := baseline.snapshot() + if err := yaml.Unmarshal(data, next); err != nil { + Warnw("could not parse log config, keeping the one in force", err, "path", path) + return nil, false + } + if err := conf.Update(next); err != nil { + Warnw("could not apply log config, keeping the one in force", err, "path", path) + return nil, false + } + return data, true +} diff --git a/logger/configwatch_test.go b/logger/configwatch_test.go new file mode 100644 index 000000000..ac7f7df11 --- /dev/null +++ b/logger/configwatch_test.go @@ -0,0 +1,185 @@ +package logger + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/zap/zapcore" +) + +func writeFile(t *testing.T, path, body string) { + t.Helper() + require.NoError(t, os.WriteFile(path, []byte(body), 0o600)) +} + +func TestApplyConfigFile(t *testing.T) { + t.Run("a level change reaches a live logger", func(t *testing.T) { + conf := &Config{Level: "info"} + l, err := NewZapLogger(conf) + require.NoError(t, err) + core := zapLoggerCore(l) + require.False(t, core.Enabled(zapcore.DebugLevel)) + + path := filepath.Join(t.TempDir(), "logging.yaml") + writeFile(t, path, "level: debug\n") + + applied, ok := applyConfigFile(conf, conf.snapshot(), path, nil) + require.True(t, ok) + require.NotEmpty(t, applied) + require.True(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel), + "the atomic level behind the existing logger must move, not just Config.Level") + }) + + t.Run("keys absent from the file keep their current values", func(t *testing.T) { + // Update assigns every field, so applying a partial file over a zero Config would wipe + // these. component_levels is where livekit-server lands pion_level. + conf := &Config{ + Level: "info", + JSON: true, + Sample: true, + SampleInitial: 7, + ComponentLevels: map[string]string{"pion": "error"}, + } + _, err := NewZapLogger(conf) + require.NoError(t, err) + + path := filepath.Join(t.TempDir(), "logging.yaml") + writeFile(t, path, "level: warn\n") + + _, ok := applyConfigFile(conf, conf.snapshot(), path, nil) + require.True(t, ok) + require.Equal(t, "warn", conf.Level) + require.True(t, conf.JSON) + require.True(t, conf.Sample) + require.Equal(t, 7, conf.SampleInitial) + require.Equal(t, map[string]string{"pion": "error"}, conf.ComponentLevels) + }) + + t.Run("a component level in the file merges with the existing ones", func(t *testing.T) { + conf := &Config{Level: "info", ComponentLevels: map[string]string{"pion": "error"}} + l, err := NewZapLogger(conf) + require.NoError(t, err) + + path := filepath.Join(t.TempDir(), "logging.yaml") + writeFile(t, path, "component_levels:\n psrpc: debug\n") + + _, ok := applyConfigFile(conf, conf.snapshot(), path, nil) + require.True(t, ok) + require.Equal(t, "error", conf.ComponentLevels["pion"]) + require.Equal(t, "debug", conf.ComponentLevels["psrpc"]) + require.True(t, zapLoggerCore(l.WithComponent("psrpc")).Enabled(zapcore.DebugLevel)) + }) + + t.Run("unchanged bytes are not reapplied", func(t *testing.T) { + conf := &Config{Level: "info"} + path := filepath.Join(t.TempDir(), "logging.yaml") + writeFile(t, path, "level: debug\n") + + applied, ok := applyConfigFile(conf, conf.snapshot(), path, nil) + require.True(t, ok) + _, ok = applyConfigFile(conf, conf.snapshot(), path, applied) + require.False(t, ok) + }) + + t.Run("malformed yaml keeps the last good config", func(t *testing.T) { + conf := &Config{Level: "info"} + l, err := NewZapLogger(conf) + require.NoError(t, err) + + path := filepath.Join(t.TempDir(), "logging.yaml") + writeFile(t, path, "level: [not, a, string\n") + + _, ok := applyConfigFile(conf, conf.snapshot(), path, nil) + require.False(t, ok) + require.Equal(t, "info", conf.Level) + require.False(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel)) + }) + + t.Run("a missing file is tolerated", func(t *testing.T) { + conf := &Config{Level: "info"} + _, ok := applyConfigFile(conf, conf.snapshot(), filepath.Join(t.TempDir(), "absent.yaml"), nil) + require.False(t, ok) + require.Equal(t, "info", conf.Level) + }) +} + +func TestWatchConfigFile(t *testing.T) { + conf := &Config{Level: "info"} + l, err := NewZapLogger(conf) + require.NoError(t, err) + + path := filepath.Join(t.TempDir(), "logging.yaml") + writeFile(t, path, "level: info\n") + + stop := WatchConfigFile(conf, path, 5*time.Millisecond) + t.Cleanup(stop) + + writeFile(t, path, "level: debug\n") + require.Eventually(t, func() bool { + return zapLoggerCore(l).Enabled(zapcore.DebugLevel) + }, 2*time.Second, 5*time.Millisecond, "watcher should pick up the rewritten file") + + stop() + writeFile(t, path, "level: error\n") + time.Sleep(50 * time.Millisecond) + require.True(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel), "stop must end the polling") +} + +// Applying config while component levels are being resolved: sharedConfig.ComponentLevel reads +// under its own mutex while Update writes the Config under a different one, so it must be reading +// a copy it owns. Meaningful under -race. +func TestApplyConfigFileWhileResolvingComponents(t *testing.T) { + conf := &Config{Level: "info", ComponentLevels: map[string]string{"pion": "error"}} + l, err := NewZapLogger(conf) + require.NoError(t, err) + + dir := t.TempDir() + path := filepath.Join(dir, "logging.yaml") + + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 500; i++ { + _ = zapLoggerCore(l.WithComponent("psrpc").WithComponent("Egress")) + } + }() + + var last []byte + for i, level := range []string{"debug", "warn", "info", "error"} { + writeFile(t, path, "level: "+level+"\n") + applied, ok := applyConfigFile(conf, conf.snapshot(), path, last) + require.True(t, ok, "iteration %d", i) + last = applied + } + <-done + require.Equal(t, "error", conf.Level) + require.Equal(t, "error", conf.ComponentLevels["pion"]) +} + +// The reset path the chart documents: emptying the file must put the startup levels back, not +// leave the last override in force. Applying each file over a baseline rather than over the +// config currently in force is what makes this hold. +func TestEmptyFileRestoresStartupConfig(t *testing.T) { + conf := &Config{Level: "info", ComponentLevels: map[string]string{"pion": "error"}} + l, err := NewZapLogger(conf) + require.NoError(t, err) + baseline := conf.snapshot() + + path := filepath.Join(t.TempDir(), "logging.yaml") + writeFile(t, path, "level: debug\ncomponent_levels:\n psrpc: debug\n") + applied, ok := applyConfigFile(conf, baseline, path, nil) + require.True(t, ok) + require.Equal(t, "debug", conf.Level) + require.True(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel)) + + writeFile(t, path, "{}\n") + _, ok = applyConfigFile(conf, baseline, path, applied) + require.True(t, ok) + require.Equal(t, "info", conf.Level) + require.Equal(t, map[string]string{"pion": "error"}, conf.ComponentLevels, + "a component the file no longer names must stop applying") + require.False(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel)) +} diff --git a/logger/logger.go b/logger/logger.go index c60d030a3..ffde17d80 100644 --- a/logger/logger.go +++ b/logger/logger.go @@ -150,11 +150,12 @@ type sharedConfig struct { func newSharedConfig(conf *Config) *sharedConfig { sc := &sharedConfig{ level: zap.NewAtomicLevelAt(ParseZapLevel(conf.Level)), - config: conf, + config: conf.snapshot(), componentLevels: make(map[string]zap.AtomicLevel), } conf.AddUpdateObserver(sc.onConfigUpdate) _ = sc.onConfigUpdate(conf) + startConfigWatchFromEnv(conf) return sc } @@ -164,7 +165,11 @@ func (c *sharedConfig) onConfigUpdate(conf *Config) error { // we have to update alla existing component levels c.mu.Lock() - c.config = conf + // Snapshot, not the caller's live Config: Update writes that object's fields under its own + // lock, while ComponentLevel reads them under c.mu. Holding a private copy keeps the two + // mutexes from guarding the same memory now that Update is actually reachable (the file + // watcher calls it; before that nothing ever did). + c.config = conf.snapshot() for component, atomicLevel := range c.componentLevels { effectiveLevel := c.level.Level() parts := strings.Split(component, ".") diff --git a/redis/redis.go b/redis/redis.go index 6881ba1d0..b1da0189e 100644 --- a/redis/redis.go +++ b/redis/redis.go @@ -21,7 +21,9 @@ import ( "fmt" "time" + entraid "github.com/redis/go-redis-entraid" "github.com/redis/go-redis/v9" + "github.com/redis/go-redis/v9/auth" "github.com/livekit/protocol/xtls" @@ -50,6 +52,7 @@ type RedisConfig struct { MaxRedirects *int `yaml:"max_redirects,omitempty"` PoolTimeout time.Duration `yaml:"pool_timeout,omitempty"` PoolSize int `yaml:"pool_size,omitempty"` + AzureEntra bool `yaml:"azure_entra,omitempty"` } func (r *RedisConfig) IsConfigured() bool { @@ -72,19 +75,26 @@ func (r *RedisConfig) GetMaxRedirects() int { return 2 } -func GetRedisClient(conf *RedisConfig) (redis.UniversalClient, error) { - if conf == nil { - return nil, nil - } +type clientOptions struct { + streamingCredentialsProvider auth.StreamingCredentialsProvider +} - if !conf.IsConfigured() { - return nil, ErrNotConfigured +type Option func(*clientOptions) + +func WithStreamingCredentialsProvider(p auth.StreamingCredentialsProvider) Option { + return func(o *clientOptions) { + o.streamingCredentialsProvider = p } +} - var rcOptions *redis.UniversalOptions - var rc redis.UniversalClient - var tlsConfig *tls.Config +var azureEntraProviderFactory = newAzureEntraCredentialsProvider + +func newAzureEntraCredentialsProvider() (auth.StreamingCredentialsProvider, error) { + return entraid.NewDefaultAzureCredentialsProvider(entraid.DefaultAzureCredentialsProviderOptions{}) +} +func buildRedisOptions(conf *RedisConfig, co clientOptions) (*redis.UniversalOptions, error) { + var tlsConfig *tls.Config if conf.TLS != nil && conf.TLS.Enabled { var err error tlsConfig, err = conf.TLS.ClientTLSConfig() @@ -97,6 +107,7 @@ func GetRedisClient(conf *RedisConfig) (redis.UniversalClient, error) { } } + var rcOptions *redis.UniversalOptions if len(conf.SentinelAddresses) > 0 { logger.Infow("connecting to redis", "sentinel", true, "addr", conf.SentinelAddresses, "masterName", conf.MasterName) @@ -153,12 +164,46 @@ func GetRedisClient(conf *RedisConfig) (redis.UniversalClient, error) { PoolSize: conf.PoolSize, } } - rc = redis.NewUniversalClient(rcOptions) - if err := rc.Ping(context.Background()).Err(); err != nil { - err = fmt.Errorf("unable to connect to redis: %w", err) + provider := co.streamingCredentialsProvider + if provider == nil && conf.AzureEntra { + p, err := azureEntraProviderFactory() + if err != nil { + return nil, fmt.Errorf("unable to create Azure Entra credentials provider: %w", err) + } + provider = p + } + if provider != nil { + rcOptions.StreamingCredentialsProvider = provider + } + + return rcOptions, nil +} + +func GetRedisClient(conf *RedisConfig, opts ...Option) (redis.UniversalClient, error) { + if conf == nil { + return nil, nil + } + + if !conf.IsConfigured() { + return nil, ErrNotConfigured + } + + var co clientOptions + for _, opt := range opts { + opt(&co) + } + + rcOptions, err := buildRedisOptions(conf, co) + if err != nil { return nil, err } + rc := redis.NewUniversalClient(rcOptions) + + if err := rc.Ping(context.Background()).Err(); err != nil { + return nil, fmt.Errorf("unable to connect to redis: %w", err) + } + return rc, nil } diff --git a/redis/redis_test.go b/redis/redis_test.go new file mode 100644 index 000000000..24442c9e2 --- /dev/null +++ b/redis/redis_test.go @@ -0,0 +1,92 @@ +// Copyright 2023 LiveKit, Inc. +// +// 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 redis + +import ( + "testing" + + "github.com/redis/go-redis/v9/auth" + "github.com/stretchr/testify/require" +) + +type fakeStreamingProvider struct{} + +func (f *fakeStreamingProvider) Subscribe(_ auth.CredentialsListener) (auth.Credentials, auth.UnsubscribeFunc, error) { + return nil, nil, nil +} + +func TestBuildRedisOptions_ClusterWithoutProvider(t *testing.T) { + opts, err := buildRedisOptions(&RedisConfig{ + ClusterAddresses: []string{"host:10000"}, + Username: "user", + Password: "pass", + UseTLS: true, + }, clientOptions{}) + require.NoError(t, err) + require.True(t, opts.IsClusterMode) + require.Nil(t, opts.StreamingCredentialsProvider) + require.Equal(t, "user", opts.Username) + require.Equal(t, "pass", opts.Password) + require.NotNil(t, opts.TLSConfig) +} + +func TestBuildRedisOptions_WithStreamingCredentialsProvider(t *testing.T) { + fake := &fakeStreamingProvider{} + var co clientOptions + WithStreamingCredentialsProvider(fake)(&co) + + opts, err := buildRedisOptions(&RedisConfig{ + ClusterAddresses: []string{"host:10000"}, + }, co) + require.NoError(t, err) + require.Same(t, fake, opts.StreamingCredentialsProvider) +} + +func TestBuildRedisOptions_AzureEntraFlag(t *testing.T) { + fake := &fakeStreamingProvider{} + orig := azureEntraProviderFactory + azureEntraProviderFactory = func() (auth.StreamingCredentialsProvider, error) { + return fake, nil + } + t.Cleanup(func() { azureEntraProviderFactory = orig }) + + opts, err := buildRedisOptions(&RedisConfig{ + ClusterAddresses: []string{"host:10000"}, + UseTLS: true, + AzureEntra: true, + }, clientOptions{}) + require.NoError(t, err) + require.Same(t, fake, opts.StreamingCredentialsProvider) +} + +func TestBuildRedisOptions_ExplicitProviderBeatsAzureEntraFlag(t *testing.T) { + explicit := &fakeStreamingProvider{} + azureFromFactory := &fakeStreamingProvider{} + orig := azureEntraProviderFactory + azureEntraProviderFactory = func() (auth.StreamingCredentialsProvider, error) { + return azureFromFactory, nil + } + t.Cleanup(func() { azureEntraProviderFactory = orig }) + + var co clientOptions + WithStreamingCredentialsProvider(explicit)(&co) + + opts, err := buildRedisOptions(&RedisConfig{ + ClusterAddresses: []string{"host:10000"}, + AzureEntra: true, + }, co) + require.NoError(t, err) + require.Same(t, explicit, opts.StreamingCredentialsProvider) +}