logger: apply a log config file to the live logger without a restart

.changeset/logger-config-file-watch.md

@@ -0,0 +1,6 @@
+---
+"github.com/livekit/protocol": patch
+"@livekit/protocol": patch
+---
+
+logger: apply a log config file to the live logger without a restart, by setting `LK_LOG_CONFIG_PATH`.

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()

logger/configwatch.go

@@ -0,0 +1,116 @@
+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.
+// The returned stop is synchronous: once it returns, no further apply can be in flight.
+func WatchConfigFile(conf *Config, path string, interval time.Duration) (stop func()) {
+	done := make(chan struct{})
+	stopped := 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() {
+		defer close(stopped)
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		var last []byte
+		for {
+			select {
+			case <-done:
+				return
+			case <-ticker.C:
+				// A tick and a close can be ready at once and select would pick either,
+				// so re-check: after stop the file must not be applied again.
+				select {
+				case <-done:
+					return
+				default:
+				}
+				if applied, ok := applyConfigFile(conf, baseline, path, last); ok {
+					last = applied
+				}
+			}
+		}
+	}()
+
+	var once sync.Once
+	return func() {
+		once.Do(func() { close(done) })
+		<-stopped
+	}
+}
+
+// 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
+}

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))
+}

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, ".")