Skip to content

Commit 9d16f84

Browse files
cosmin-staicuclaude
andcommitted
feat(logger): apply a log config file to the live logger without a restart
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. The file is a declarative overlay on the startup config, not on whatever was applied last: WatchConfigFile snapshots the config once at startup and every file is decoded over that baseline, so a key the file omits falls back to its startup value and emptying the file to `{}` restores the levels the process booted with. Decoding over the config in force instead would make an empty file a no-op once `level: debug` had been applied, and would keep applying a component_levels entry after it disappeared from the file. 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. A file that goes away is deliberately not a reset — a transient read error would otherwise flap levels on a live process; emptying it to `{}` is the reset. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PqszK5GeFJhZx7Hyb2vqcY Signed-off-by: Cosmin Staicu <cosmin.staicu@uipath.com>
1 parent a0a06fb commit 9d16f84

5 files changed

Lines changed: 338 additions & 2 deletions

File tree

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
"github.com/livekit/protocol": patch
3+
"@livekit/protocol": patch
4+
---
5+
6+
logger: apply a log config file to the live logger without a restart, by setting `LK_LOG_CONFIG_PATH`.

logger/config.go

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,30 @@ func (c *Config) Update(o *Config) error {
6767
return nil
6868
}
6969

70+
// snapshot copies the data fields so a file can be unmarshalled over the config in force.
71+
// Update assigns every field, so decoding a partial file into a zero Config would silently reset
72+
// the rest — including ComponentLevels, which is where livekit-server puts pion_level.
73+
func (c *Config) snapshot() *Config {
74+
c.lock.Lock()
75+
defer c.lock.Unlock()
76+
77+
componentLevels := make(map[string]string, len(c.ComponentLevels))
78+
for component, level := range c.ComponentLevels {
79+
componentLevels[component] = level
80+
}
81+
return &Config{
82+
JSON: c.JSON,
83+
Level: c.Level,
84+
Sample: c.Sample,
85+
ComponentLevels: componentLevels,
86+
SampleInitial: c.SampleInitial,
87+
SampleInterval: c.SampleInterval,
88+
ItemSampleSeconds: c.ItemSampleSeconds,
89+
ItemSampleInitial: c.ItemSampleInitial,
90+
ItemSampleInterval: c.ItemSampleInterval,
91+
}
92+
}
93+
7094
func (c *Config) AddUpdateObserver(cb ConfigObserver) {
7195
c.lock.Lock()
7296
defer c.lock.Unlock()

logger/configwatch.go

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
package logger
2+
3+
import (
4+
"bytes"
5+
"os"
6+
"sync"
7+
"time"
8+
9+
"gopkg.in/yaml.v3"
10+
)
11+
12+
const (
13+
// ConfigPathEnv names a file holding the same keys as the service config's `logging` block.
14+
// Point it inside a mounted ConfigMap to change levels without restarting: kubelet refreshes
15+
// the mount in place, and the next poll pushes the new values into the live logger.
16+
ConfigPathEnv = "LK_LOG_CONFIG_PATH"
17+
// ConfigIntervalEnv overrides the poll interval as a Go duration (e.g. "10s").
18+
ConfigIntervalEnv = "LK_LOG_CONFIG_INTERVAL"
19+
20+
defaultConfigWatchInterval = 30 * time.Second
21+
)
22+
23+
var configWatchOnce sync.Once
24+
25+
// startConfigWatchFromEnv wires the watcher for the first logger the process builds, which is the
26+
// one whose Config the service keeps. Every binary that uses this package reaches it through
27+
// newSharedConfig, so none of them need their own flag or call site.
28+
func startConfigWatchFromEnv(conf *Config) {
29+
path := os.Getenv(ConfigPathEnv)
30+
if path == "" {
31+
return
32+
}
33+
interval := defaultConfigWatchInterval
34+
if v := os.Getenv(ConfigIntervalEnv); v != "" {
35+
if d, err := time.ParseDuration(v); err == nil && d > 0 {
36+
interval = d
37+
}
38+
}
39+
configWatchOnce.Do(func() {
40+
WatchConfigFile(conf, path, interval)
41+
})
42+
}
43+
44+
// WatchConfigFile applies path to conf every interval until the returned stop is called.
45+
//
46+
// Polling rather than fsnotify on purpose: a ConfigMap volume update swaps the `..data` symlink
47+
// instead of rewriting the file, so a watch on the file itself never fires.
48+
// The returned stop is synchronous: once it returns, no further apply can be in flight.
49+
func WatchConfigFile(conf *Config, path string, interval time.Duration) (stop func()) {
50+
done := make(chan struct{})
51+
stopped := make(chan struct{})
52+
// The config the process started with. Every file is applied over this, never over whatever
53+
// the previous file left in force, so an empty file restores the startup levels and a
54+
// component_levels entry that disappears from the file stops applying.
55+
baseline := conf.snapshot()
56+
go func() {
57+
defer close(stopped)
58+
ticker := time.NewTicker(interval)
59+
defer ticker.Stop()
60+
var last []byte
61+
for {
62+
select {
63+
case <-done:
64+
return
65+
case <-ticker.C:
66+
// A tick and a close can be ready at once and select would pick either,
67+
// so re-check: after stop the file must not be applied again.
68+
select {
69+
case <-done:
70+
return
71+
default:
72+
}
73+
if applied, ok := applyConfigFile(conf, baseline, path, last); ok {
74+
last = applied
75+
}
76+
}
77+
}
78+
}()
79+
80+
var once sync.Once
81+
return func() {
82+
once.Do(func() { close(done) })
83+
<-stopped
84+
}
85+
}
86+
87+
// applyConfigFile decodes path over baseline and pushes the result into conf when the bytes differ
88+
// from last, returning the bytes it applied. ok is false when nothing was applied (unreadable,
89+
// unchanged or invalid), and the config already in force stays untouched.
90+
//
91+
// Decoding over baseline rather than over conf is what makes the file a declarative overlay: keys
92+
// it omits fall back to the startup values instead of inheriting the previous file's.
93+
func applyConfigFile(conf, baseline *Config, path string, last []byte) (applied []byte, ok bool) {
94+
data, err := os.ReadFile(path)
95+
if err != nil {
96+
// An optional ConfigMap that is not mounted yet is the normal steady state, not something
97+
// to log once per interval forever. A file that goes away is deliberately not treated as a
98+
// reset either: a transient read error would otherwise flap levels. Emptying the file to
99+
// `{}` is the reset.
100+
return nil, false
101+
}
102+
if bytes.Equal(data, last) {
103+
return nil, false
104+
}
105+
106+
next := baseline.snapshot()
107+
if err := yaml.Unmarshal(data, next); err != nil {
108+
Warnw("could not parse log config, keeping the one in force", err, "path", path)
109+
return nil, false
110+
}
111+
if err := conf.Update(next); err != nil {
112+
Warnw("could not apply log config, keeping the one in force", err, "path", path)
113+
return nil, false
114+
}
115+
return data, true
116+
}

logger/configwatch_test.go

Lines changed: 185 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,185 @@
1+
package logger
2+
3+
import (
4+
"os"
5+
"path/filepath"
6+
"testing"
7+
"time"
8+
9+
"github.com/stretchr/testify/require"
10+
"go.uber.org/zap/zapcore"
11+
)
12+
13+
func writeFile(t *testing.T, path, body string) {
14+
t.Helper()
15+
require.NoError(t, os.WriteFile(path, []byte(body), 0o600))
16+
}
17+
18+
func TestApplyConfigFile(t *testing.T) {
19+
t.Run("a level change reaches a live logger", func(t *testing.T) {
20+
conf := &Config{Level: "info"}
21+
l, err := NewZapLogger(conf)
22+
require.NoError(t, err)
23+
core := zapLoggerCore(l)
24+
require.False(t, core.Enabled(zapcore.DebugLevel))
25+
26+
path := filepath.Join(t.TempDir(), "logging.yaml")
27+
writeFile(t, path, "level: debug\n")
28+
29+
applied, ok := applyConfigFile(conf, conf.snapshot(), path, nil)
30+
require.True(t, ok)
31+
require.NotEmpty(t, applied)
32+
require.True(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel),
33+
"the atomic level behind the existing logger must move, not just Config.Level")
34+
})
35+
36+
t.Run("keys absent from the file keep their current values", func(t *testing.T) {
37+
// Update assigns every field, so applying a partial file over a zero Config would wipe
38+
// these. component_levels is where livekit-server lands pion_level.
39+
conf := &Config{
40+
Level: "info",
41+
JSON: true,
42+
Sample: true,
43+
SampleInitial: 7,
44+
ComponentLevels: map[string]string{"pion": "error"},
45+
}
46+
_, err := NewZapLogger(conf)
47+
require.NoError(t, err)
48+
49+
path := filepath.Join(t.TempDir(), "logging.yaml")
50+
writeFile(t, path, "level: warn\n")
51+
52+
_, ok := applyConfigFile(conf, conf.snapshot(), path, nil)
53+
require.True(t, ok)
54+
require.Equal(t, "warn", conf.Level)
55+
require.True(t, conf.JSON)
56+
require.True(t, conf.Sample)
57+
require.Equal(t, 7, conf.SampleInitial)
58+
require.Equal(t, map[string]string{"pion": "error"}, conf.ComponentLevels)
59+
})
60+
61+
t.Run("a component level in the file merges with the existing ones", func(t *testing.T) {
62+
conf := &Config{Level: "info", ComponentLevels: map[string]string{"pion": "error"}}
63+
l, err := NewZapLogger(conf)
64+
require.NoError(t, err)
65+
66+
path := filepath.Join(t.TempDir(), "logging.yaml")
67+
writeFile(t, path, "component_levels:\n psrpc: debug\n")
68+
69+
_, ok := applyConfigFile(conf, conf.snapshot(), path, nil)
70+
require.True(t, ok)
71+
require.Equal(t, "error", conf.ComponentLevels["pion"])
72+
require.Equal(t, "debug", conf.ComponentLevels["psrpc"])
73+
require.True(t, zapLoggerCore(l.WithComponent("psrpc")).Enabled(zapcore.DebugLevel))
74+
})
75+
76+
t.Run("unchanged bytes are not reapplied", func(t *testing.T) {
77+
conf := &Config{Level: "info"}
78+
path := filepath.Join(t.TempDir(), "logging.yaml")
79+
writeFile(t, path, "level: debug\n")
80+
81+
applied, ok := applyConfigFile(conf, conf.snapshot(), path, nil)
82+
require.True(t, ok)
83+
_, ok = applyConfigFile(conf, conf.snapshot(), path, applied)
84+
require.False(t, ok)
85+
})
86+
87+
t.Run("malformed yaml keeps the last good config", func(t *testing.T) {
88+
conf := &Config{Level: "info"}
89+
l, err := NewZapLogger(conf)
90+
require.NoError(t, err)
91+
92+
path := filepath.Join(t.TempDir(), "logging.yaml")
93+
writeFile(t, path, "level: [not, a, string\n")
94+
95+
_, ok := applyConfigFile(conf, conf.snapshot(), path, nil)
96+
require.False(t, ok)
97+
require.Equal(t, "info", conf.Level)
98+
require.False(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel))
99+
})
100+
101+
t.Run("a missing file is tolerated", func(t *testing.T) {
102+
conf := &Config{Level: "info"}
103+
_, ok := applyConfigFile(conf, conf.snapshot(), filepath.Join(t.TempDir(), "absent.yaml"), nil)
104+
require.False(t, ok)
105+
require.Equal(t, "info", conf.Level)
106+
})
107+
}
108+
109+
func TestWatchConfigFile(t *testing.T) {
110+
conf := &Config{Level: "info"}
111+
l, err := NewZapLogger(conf)
112+
require.NoError(t, err)
113+
114+
path := filepath.Join(t.TempDir(), "logging.yaml")
115+
writeFile(t, path, "level: info\n")
116+
117+
stop := WatchConfigFile(conf, path, 5*time.Millisecond)
118+
t.Cleanup(stop)
119+
120+
writeFile(t, path, "level: debug\n")
121+
require.Eventually(t, func() bool {
122+
return zapLoggerCore(l).Enabled(zapcore.DebugLevel)
123+
}, 2*time.Second, 5*time.Millisecond, "watcher should pick up the rewritten file")
124+
125+
stop()
126+
writeFile(t, path, "level: error\n")
127+
time.Sleep(50 * time.Millisecond)
128+
require.True(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel), "stop must end the polling")
129+
}
130+
131+
// Applying config while component levels are being resolved: sharedConfig.ComponentLevel reads
132+
// under its own mutex while Update writes the Config under a different one, so it must be reading
133+
// a copy it owns. Meaningful under -race.
134+
func TestApplyConfigFileWhileResolvingComponents(t *testing.T) {
135+
conf := &Config{Level: "info", ComponentLevels: map[string]string{"pion": "error"}}
136+
l, err := NewZapLogger(conf)
137+
require.NoError(t, err)
138+
139+
dir := t.TempDir()
140+
path := filepath.Join(dir, "logging.yaml")
141+
142+
done := make(chan struct{})
143+
go func() {
144+
defer close(done)
145+
for i := 0; i < 500; i++ {
146+
_ = zapLoggerCore(l.WithComponent("psrpc").WithComponent("Egress"))
147+
}
148+
}()
149+
150+
var last []byte
151+
for i, level := range []string{"debug", "warn", "info", "error"} {
152+
writeFile(t, path, "level: "+level+"\n")
153+
applied, ok := applyConfigFile(conf, conf.snapshot(), path, last)
154+
require.True(t, ok, "iteration %d", i)
155+
last = applied
156+
}
157+
<-done
158+
require.Equal(t, "error", conf.Level)
159+
require.Equal(t, "error", conf.ComponentLevels["pion"])
160+
}
161+
162+
// The reset path the chart documents: emptying the file must put the startup levels back, not
163+
// leave the last override in force. Applying each file over a baseline rather than over the
164+
// config currently in force is what makes this hold.
165+
func TestEmptyFileRestoresStartupConfig(t *testing.T) {
166+
conf := &Config{Level: "info", ComponentLevels: map[string]string{"pion": "error"}}
167+
l, err := NewZapLogger(conf)
168+
require.NoError(t, err)
169+
baseline := conf.snapshot()
170+
171+
path := filepath.Join(t.TempDir(), "logging.yaml")
172+
writeFile(t, path, "level: debug\ncomponent_levels:\n psrpc: debug\n")
173+
applied, ok := applyConfigFile(conf, baseline, path, nil)
174+
require.True(t, ok)
175+
require.Equal(t, "debug", conf.Level)
176+
require.True(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel))
177+
178+
writeFile(t, path, "{}\n")
179+
_, ok = applyConfigFile(conf, baseline, path, applied)
180+
require.True(t, ok)
181+
require.Equal(t, "info", conf.Level)
182+
require.Equal(t, map[string]string{"pion": "error"}, conf.ComponentLevels,
183+
"a component the file no longer names must stop applying")
184+
require.False(t, zapLoggerCore(l).Enabled(zapcore.DebugLevel))
185+
}

logger/logger.go

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -150,11 +150,12 @@ type sharedConfig struct {
150150
func newSharedConfig(conf *Config) *sharedConfig {
151151
sc := &sharedConfig{
152152
level: zap.NewAtomicLevelAt(ParseZapLevel(conf.Level)),
153-
config: conf,
153+
config: conf.snapshot(),
154154
componentLevels: make(map[string]zap.AtomicLevel),
155155
}
156156
conf.AddUpdateObserver(sc.onConfigUpdate)
157157
_ = sc.onConfigUpdate(conf)
158+
startConfigWatchFromEnv(conf)
158159
return sc
159160
}
160161

@@ -164,7 +165,11 @@ func (c *sharedConfig) onConfigUpdate(conf *Config) error {
164165

165166
// we have to update alla existing component levels
166167
c.mu.Lock()
167-
c.config = conf
168+
// Snapshot, not the caller's live Config: Update writes that object's fields under its own
169+
// lock, while ComponentLevel reads them under c.mu. Holding a private copy keeps the two
170+
// mutexes from guarding the same memory now that Update is actually reachable (the file
171+
// watcher calls it; before that nothing ever did).
172+
c.config = conf.snapshot()
168173
for component, atomicLevel := range c.componentLevels {
169174
effectiveLevel := c.level.Level()
170175
parts := strings.Split(component, ".")

0 commit comments

Comments
 (0)