From e227b1a6caa348211e3976191eef4bf7e2d01da6 Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Mon, 15 Jun 2026 12:24:53 +0300 Subject: [PATCH 1/3] redis: add credentials provider option and built-in Entra ID auth --- go.mod | 7 ++++ go.sum | 23 ++++++++++++ redis/redis.go | 69 ++++++++++++++++++++++++++++------ redis/redis_test.go | 92 +++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 179 insertions(+), 12 deletions(-) create mode 100644 redis/redis_test.go 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/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) +} From 524da91d67796e24b74895c4e16d78d11465284c Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Thu, 3 Sep 2026 14:50:16 +0300 Subject: [PATCH 2/3] feat(logger): apply a log config file to the live logger without a restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pieces for changing levels at runtime were already here — the root level and every component level are zap.AtomicLevel, Config.Update pushes new values into them, and newSharedConfig registers itself as an update observer — but nothing ever called Update, so a level change meant restarting the process. Add the missing trigger: when LK_LOG_CONFIG_PATH is set, poll that file and apply it to the Config the service is holding (LK_LOG_CONFIG_INTERVAL overrides the 30s default). The hook sits in newSharedConfig, which is the one path every consumer reaches — livekit-server via InitFromConfig, livekit-sip via NewZapLogger — so no binary needs its own flag or call site. Polling rather than fsnotify because the target is a mounted ConfigMap: kubelet swaps the ..data symlink instead of rewriting the file, so a watch on the file never fires. Two details that would otherwise bite: - Update assigns every field, so a partial file decoded into a zero Config would silently reset the rest. Config.snapshot copies the data fields and the file is unmarshalled over that, leaving unspecified keys — including ComponentLevels, where livekit-server puts pion_level — as they were. - sharedConfig kept the caller's live *Config and read ComponentLevels from it under its own mutex, while Update writes those fields under Config.lock. Two mutexes over the same memory was harmless while Update was unreachable; now it is reachable, so sharedConfig holds a snapshot it owns instead. An unreadable file (an optional ConfigMap not yet mounted), unchanged bytes and malformed YAML all leave the config in force untouched, without logging once per interval. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FgV5yPwbxmFhMuN62ybthK Signed-off-by: Cosmin Staicu --- logger/config.go | 24 ++++++ logger/configwatch.go | 94 ++++++++++++++++++++++ logger/configwatch_test.go | 160 +++++++++++++++++++++++++++++++++++++ logger/logger.go | 9 ++- 4 files changed, 285 insertions(+), 2 deletions(-) create mode 100644 logger/configwatch.go create mode 100644 logger/configwatch_test.go 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..671065984 --- /dev/null +++ b/logger/configwatch.go @@ -0,0 +1,94 @@ +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{}) + 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, path, last); ok { + last = applied + } + } + } + }() + + var once sync.Once + return func() { once.Do(func() { close(done) }) } +} + +// applyConfigFile pushes path into conf when its 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. +func applyConfigFile(conf *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. + return nil, false + } + if bytes.Equal(data, last) { + return nil, false + } + + next := conf.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..43caa1586 --- /dev/null +++ b/logger/configwatch_test.go @@ -0,0 +1,160 @@ +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, 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, 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, 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, path, nil) + require.True(t, ok) + _, ok = applyConfigFile(conf, 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, 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, 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, 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"]) +} 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, ".") From e736f58e29c458a53b0fc868d4334b049b598f6f Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Thu, 3 Sep 2026 15:49:28 +0300 Subject: [PATCH 3/3] fix(logger): apply the config file over the startup config, not the last one Emptying the watched file was supposed to restore the levels the process booted with. It did not: each file was decoded over a snapshot of the config currently in force, so once `level: debug` had been applied, an empty file snapshotted debug and changed nothing. A component_levels entry that disappeared from the file kept applying for the same reason. WatchConfigFile now snapshots the startup config once and applyConfigFile decodes every file over that baseline, which makes the file a declarative overlay: keys it omits fall back to the startup values. A file that goes away is still not a reset. A transient read error would otherwise flap levels on a live process, so the reset is emptying the file to `{}`, which now works. Reported by Copilot on UiPath/cx#823. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FgV5yPwbxmFhMuN62ybthK Signed-off-by: Cosmin Staicu --- logger/configwatch.go | 23 ++++++++++++++------- logger/configwatch_test.go | 41 ++++++++++++++++++++++++++++++-------- 2 files changed, 49 insertions(+), 15 deletions(-) diff --git a/logger/configwatch.go b/logger/configwatch.go index 671065984..b9d1151b3 100644 --- a/logger/configwatch.go +++ b/logger/configwatch.go @@ -47,6 +47,10 @@ func startConfigWatchFromEnv(conf *Config) { // 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() @@ -56,7 +60,7 @@ func WatchConfigFile(conf *Config, path string, interval time.Duration) (stop fu case <-done: return case <-ticker.C: - if applied, ok := applyConfigFile(conf, path, last); ok { + if applied, ok := applyConfigFile(conf, baseline, path, last); ok { last = applied } } @@ -67,21 +71,26 @@ func WatchConfigFile(conf *Config, path string, interval time.Duration) (stop fu return func() { once.Do(func() { close(done) }) } } -// applyConfigFile pushes path into conf when its 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. -func applyConfigFile(conf *Config, path string, last []byte) (applied []byte, ok bool) { +// 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. + // 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 := conf.snapshot() + 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 diff --git a/logger/configwatch_test.go b/logger/configwatch_test.go index 43caa1586..ac7f7df11 100644 --- a/logger/configwatch_test.go +++ b/logger/configwatch_test.go @@ -26,7 +26,7 @@ func TestApplyConfigFile(t *testing.T) { path := filepath.Join(t.TempDir(), "logging.yaml") writeFile(t, path, "level: debug\n") - applied, ok := applyConfigFile(conf, path, nil) + applied, ok := applyConfigFile(conf, conf.snapshot(), path, nil) require.True(t, ok) require.NotEmpty(t, applied) require.True(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel), @@ -49,7 +49,7 @@ func TestApplyConfigFile(t *testing.T) { path := filepath.Join(t.TempDir(), "logging.yaml") writeFile(t, path, "level: warn\n") - _, ok := applyConfigFile(conf, path, nil) + _, ok := applyConfigFile(conf, conf.snapshot(), path, nil) require.True(t, ok) require.Equal(t, "warn", conf.Level) require.True(t, conf.JSON) @@ -66,7 +66,7 @@ func TestApplyConfigFile(t *testing.T) { path := filepath.Join(t.TempDir(), "logging.yaml") writeFile(t, path, "component_levels:\n psrpc: debug\n") - _, ok := applyConfigFile(conf, path, nil) + _, 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"]) @@ -78,9 +78,9 @@ func TestApplyConfigFile(t *testing.T) { path := filepath.Join(t.TempDir(), "logging.yaml") writeFile(t, path, "level: debug\n") - applied, ok := applyConfigFile(conf, path, nil) + applied, ok := applyConfigFile(conf, conf.snapshot(), path, nil) require.True(t, ok) - _, ok = applyConfigFile(conf, path, applied) + _, ok = applyConfigFile(conf, conf.snapshot(), path, applied) require.False(t, ok) }) @@ -92,7 +92,7 @@ func TestApplyConfigFile(t *testing.T) { path := filepath.Join(t.TempDir(), "logging.yaml") writeFile(t, path, "level: [not, a, string\n") - _, ok := applyConfigFile(conf, path, nil) + _, 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)) @@ -100,7 +100,7 @@ func TestApplyConfigFile(t *testing.T) { t.Run("a missing file is tolerated", func(t *testing.T) { conf := &Config{Level: "info"} - _, ok := applyConfigFile(conf, filepath.Join(t.TempDir(), "absent.yaml"), nil) + _, ok := applyConfigFile(conf, conf.snapshot(), filepath.Join(t.TempDir(), "absent.yaml"), nil) require.False(t, ok) require.Equal(t, "info", conf.Level) }) @@ -150,7 +150,7 @@ func TestApplyConfigFileWhileResolvingComponents(t *testing.T) { var last []byte for i, level := range []string{"debug", "warn", "info", "error"} { writeFile(t, path, "level: "+level+"\n") - applied, ok := applyConfigFile(conf, path, last) + applied, ok := applyConfigFile(conf, conf.snapshot(), path, last) require.True(t, ok, "iteration %d", i) last = applied } @@ -158,3 +158,28 @@ func TestApplyConfigFileWhileResolvingComponents(t *testing.T) { 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)) +}