livekit-protocol 0.7.9

// Copyright 2024 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 utils

import (
	"fmt"
	"hash/crc32"
	"os"

	"github.com/fsnotify/fsnotify"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"go.uber.org/atomic"
	"gopkg.in/yaml.v3"

	"github.com/livekit/protocol/logger"
	"github.com/livekit/protocol/utils/events"
)

type ConfigBuilder[T any] interface {
	New() (*T, error)
}

type ConfigDefaulter[T any] interface {
	InitDefaults(*T) error
}

type ConfigObserver[T any] struct {
	builder   ConfigBuilder[T]
	watcher   *fsnotify.Watcher
	observers *events.ObserverList[*T]
	conf      atomic.Pointer[T]
}

func NewConfigObserver[T any](path string, builder ConfigBuilder[T]) (*ConfigObserver[T], *T, error) {
	c := &ConfigObserver[T]{
		builder:   builder,
		observers: events.NewObserverList[*T](events.WithBlocking()),
	}

	conf, err := c.load(path)
	if err != nil {
		return nil, nil, err
	}

	if path != "" {
		c.watcher, err = fsnotify.NewWatcher()
		if err != nil {
			return nil, nil, err
		}
		if err := c.watcher.Add(path); err != nil {
			c.watcher.Close()
			return nil, nil, err
		}
		go c.watch()
	}

	return c, conf, nil
}

func (c *ConfigObserver[T]) Close() {
	if c != nil && c.watcher != nil {
		c.watcher.Close()
	}
}

func (c *ConfigObserver[T]) EmitConfigUpdate(conf *T) {
	c.observers.Emit(conf)
}

func (c *ConfigObserver[T]) Observe(cb func(*T)) func() {
	if c == nil {
		return func() {}
	}
	return c.observers.On(cb)
}

func (c *ConfigObserver[T]) Load() *T {
	return c.conf.Load()
}

func (c *ConfigObserver[T]) watch() {
	for {
		select {
		case event, ok := <-c.watcher.Events:
			if !ok {
				return
			}
			if event.Has(fsnotify.Remove) {
				if err := c.watcher.Add(event.Name); err != nil {
					logger.Errorw("unable to rewatch config file", err, "file", event.Name)
				}
			}
			if event.Has(fsnotify.Write | fsnotify.Remove) {
				if err := c.reload(event.Name); err != nil {
					logger.Errorw("unable to update config file", err, "file", event.Name)
				} else {
					logger.Infow("config file has been updated", "file", event.Name)
				}
			}
		case err, ok := <-c.watcher.Errors:
			if !ok {
				return
			}
			logger.Errorw("config file watcher error", err)
		}
	}
}

func (c *ConfigObserver[T]) reload(path string) error {
	conf, err := c.load(path)
	if err != nil {
		recordConfigReload(path, false)
		return err
	}

	recordConfigReload(path, true)
	c.EmitConfigUpdate(conf)
	return nil
}

func (c *ConfigObserver[T]) load(path string) (conf *T, err error) {
	// always record the load outcome so livekit_config_load_state is populated
	// from the initial load onward and flips on every subsequent load attempt.
	if path != "" {
		defer func() { setConfigLoadState(path, err != nil) }()
	}

	conf, err = c.builder.New()
	if err != nil {
		return nil, err
	}

	var hash uint32
	hasHash := false
	if path != "" {
		b, err := os.ReadFile(path)
		if err != nil {
			return nil, err
		}

		if len(b) == 0 {
			return nil, fmt.Errorf("cannot parse config: file empty")
		}

		if err := yaml.Unmarshal(b, conf); err != nil {
			return nil, fmt.Errorf("cannot parse config: %v", err)
		}

		hash = configHash(b)
		hasHash = true
	}

	if d, ok := c.builder.(ConfigDefaulter[T]); ok {
		d.InitDefaults(conf)
	}

	c.conf.Store(conf)

	if hasHash {
		setConfigHash(path, hash)
	}

	return conf, err
}

var (
	promConfigReloadTotal = promauto.NewCounterVec(prometheus.CounterOpts{
		Namespace: "livekit",
		Subsystem: "config",
		Name:      "reload_total",
		Help:      "Number of times the config file has been reloaded, labeled by file and whether the reload succeeded",
	}, []string{"file", "status"})

	promConfigHash = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "livekit",
		Subsystem: "config",
		Name:      "hash",
		Help:      "Short checksum of the config file currently in use, for detecting changes. Not updated when a reload fails",
	}, []string{"file"})

	promConfigLoadState = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "livekit",
		Subsystem: "config",
		Name:      "load_state",
		Help:      "0 when the config file last loaded successfully, 1 when the last load failed.",
	}, []string{"file"})
)

// configHash returns a small checksum of the config bytes, used only to detect
// when the config content changes (not for any security purpose).
func configHash(b []byte) uint32 {
	return crc32.ChecksumIEEE(b) % 1_000_000
}

func recordConfigReload(file string, success bool) {
	status := "success"
	if !success {
		status = "failure"
	}
	promConfigReloadTotal.WithLabelValues(file, status).Inc()
}

// setConfigLoadState publishes whether the most recent load of file failed:
// 0 on success, 1 on failure.
func setConfigLoadState(file string, failed bool) {
	v := 0.0
	if failed {
		v = 1
	}
	promConfigLoadState.WithLabelValues(file).Set(v)
}

// setConfigHash publishes the checksum of the config currently in use for file.
func setConfigHash(file string, hash uint32) {
	promConfigHash.WithLabelValues(file).Set(float64(hash))
}