feat: boot sequence health check and Prometheus health metrics

Add a boot sequence health check that will cause the app to abort if any
of the following requirements aren't met:
  - Filesystem write permissions
  - Redis connectivity
  - LiveKit connectivity (when enabled - see livekit.healthCheck)

Additionally, add a health metric to the Prometheus exporter called
`recorder_component_health`. It's a gauge with two possible labels:
  - livekit
  - redis.
1 means healthy, 0 means unhealthy.

Author: prlanzarin

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 All notable changes to this project will be documented in this file.
 
+### UNRELEASED
+
+* feat: boot sequence health check and Prometheus health metrics
+
 ### v0.9.4
 
 * feat(livekit): add Prom counters for reconnects and sub failures

config/bbb-webrtc-recorder.yml

@@ -51,3 +51,6 @@ livekit:
   host: ws://localhost:7880
   apiKey: ""
   apiSecret: ""
+  healthCheck:
+    enable: false
+    interval: 1m

internal/app/app.go

@@ -5,12 +5,15 @@ import (
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
 
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal"
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/appstats"
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/config"
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/pubsub"
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/server"
+	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/webrtc/livekit"
+	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/webrtc/recorder"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
 	flag "github.com/spf13/pflag"
@@ -82,6 +85,43 @@ func Run() {
 
 	ps = pubsub.NewPubSub(cfg.PubSub)
 
+	if err := ps.Check(); err != nil {
+		log.Fatalf("failed to connect to pubsub: %v", err)
+	}
+
+	if err := recorder.CheckFsPermissions(cfg.Recorder); err != nil {
+		log.Fatalf("failed to check recorder filesystem permissions: %v", err)
+	}
+
+	if cfg.LiveKit.HealthCheck.Enable {
+		log.WithField("interval", cfg.LiveKit.HealthCheck.Interval).
+			WithField("host", cfg.LiveKit.Host).
+			Debug("LiveKit health check enabled")
+
+		if err := livekit.CheckConnectivity(cfg.LiveKit); err != nil {
+			log.Fatalf("failed to connect to LiveKit: %v", err)
+		}
+
+		appstats.SetComponentHealth("livekit", true)
+		log.WithField("host", cfg.LiveKit.Host).
+			Info("LiveKit is ready")
+
+		go func() {
+			ticker := time.NewTicker(cfg.LiveKit.HealthCheck.Interval)
+			defer ticker.Stop()
+			for {
+				<-ticker.C
+				if err := livekit.CheckConnectivity(cfg.LiveKit); err != nil {
+					log.Warnf("livekit health check failed: %v", err)
+					appstats.SetComponentHealth("livekit", false)
+				} else {
+					log.Trace("livekit health check succeeded")
+					appstats.SetComponentHealth("livekit", true)
+				}
+			}
+		}()
+	}
+
 	if cfg.HTTP.Enable {
 		hs := server.NewHTTPServer(cfg, ps)
 		hs.Serve()

internal/appstats/prometheus.go

@@ -190,6 +190,15 @@ var (
 		[]string{
 			"error", // error string
 		})
+
+	ComponentHealth = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Subsystem: "recorder",
+		Name:      "component_health",
+		Help:      "Health status of recorder components (1 = healthy, 0 = unhealthy)",
+	},
+		[]string{
+			"component",
+		})
 )
 
 func Init() {
@@ -211,6 +220,7 @@ func Init() {
 	prometheus.MustRegister(ParticipantReconnects)
 	prometheus.MustRegister(TrackSubscriptionFailures)
 	prometheus.MustRegister(ParticipantReconnectingEvents)
+	prometheus.MustRegister(ComponentHealth)
 }
 
 func newMetricsHandler() *metricsHandler {
@@ -322,6 +332,16 @@ func OnServerResponse(msg interface{}) {
 	}
 }
 
+func SetComponentHealth(component string, healthy bool) {
+	status := 0.0
+
+	if healthy {
+		status = 1.0
+	}
+
+	ComponentHealth.WithLabelValues(component).Set(status)
+}
+
 func UpdateCaptureMetrics(stats *CaptureStats) {
 	if stats == nil {
 		return

internal/config/config.go

@@ -80,6 +80,10 @@ func (cfg *Config) SetDefaults() {
 		APISecret:             "",
 		PacketReadTimeout:     500 * time.Millisecond,
 		PreferredVideoQuality: livekit.VideoQuality_HIGH,
+		HealthCheck: HealthCheck{
+			Enable:   false,
+			Interval: 1 * time.Minute,
+		},
 	}
 }
 
@@ -131,11 +135,17 @@ type Prometheus struct {
 }
 
 type LiveKit struct {
-	Host                  string               `mapstructure:"host"`
-	APIKey                string               `mapstructure:"api_key"`
-	APISecret             string               `mapstructure:"api_secret"`
-	PacketReadTimeout     time.Duration        `mapstructure:"packet_read_timeout"`
-	PreferredVideoQuality livekit.VideoQuality `mapstructure:"preferred_video_quality"`
+	Host                  string               `yaml:"host,omitempty" mapstructure:"host"`
+	APIKey                string               `yaml:"apiKey,omitempty" mapstructure:"api_key"`
+	APISecret             string               `yaml:"apiSecret,omitempty" mapstructure:"api_secret"`
+	PacketReadTimeout     time.Duration        `yaml:"packetReadTimeout,omitempty" mapstructure:"packet_read_timeout"`
+	PreferredVideoQuality livekit.VideoQuality `yaml:"preferredVideoQuality,omitempty" mapstructure:"preferred_video_quality"`
+	HealthCheck           HealthCheck          `yaml:"healthCheck,omitempty"`
+}
+
+type HealthCheck struct {
+	Enable   bool          `yaml:"enable,omitempty"`
+	Interval time.Duration `yaml:"interval,omitempty"`
 }
 
 type LogConfig struct {

internal/pubsub/pubsub.go

@@ -13,6 +13,7 @@ type PubSub interface {
 	Subscribe(channel string, handler PubSubHandler, onStart func() error) error
 	Publish(channel string, message []byte) error
 	Close() error
+	Check() error
 }
 
 type PubSubHandler func(ctx context.Context, message []byte)

internal/pubsub/redis.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/appstats"
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/config"
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/pubsub/redis"
 	"github.com/cenkalti/backoff/v4"
@@ -37,6 +38,7 @@ func (r *Redis) Subscribe(channel string, handler PubSubHandler, onStart func()
 				log.Infof("Subscribed to pubsub %s on redis %s", channel, r.config.Address)
 				r.hasConnected = true
 				eb.Reset()
+				appstats.SetComponentHealth("redis", true)
 
 				return onStart()
 			},
@@ -59,6 +61,7 @@ func (r *Redis) Subscribe(channel string, handler PubSubHandler, onStart func()
 			log.Errorf("failed to subscribe to pubsub %s on %s: %s", channel, r.config.Address, err)
 			return err
 		} else {
+			appstats.SetComponentHealth("redis", false)
 			log.Errorf("failed to subscribe to pubsub %s on %s: %s - retrying in %s", channel, r.config.Address, err, next)
 		}
 
@@ -79,6 +82,10 @@ func (r *Redis) Publish(channel string, message []byte) error {
 	return r.pubsub.Publish(channel, message)
 }
 
+func (r *Redis) Check() error {
+	return r.pubsub.Check()
+}
+
 func NewRedis(cfg config.Redis) *Redis {
 	r := &Redis{config: cfg}
 	if p, err := redis.NewPubSub(cfg.Network, cfg.Address, cfg.Password); err != nil {

internal/pubsub/redis/redis.go

@@ -119,6 +119,22 @@ loop:
 	return <-done
 }
 
+func (p *PubSub) Check() error {
+	c, err := p.dial()
+
+	if err != nil {
+		return err
+	}
+
+	defer c.Close()
+
+	if _, err = c.Do("PING"); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (p *PubSub) Publish(channel string, message []byte) error {
 	c, err := p.dial()
 	if err != nil {

internal/webrtc/recorder/recorder.go

@@ -12,6 +12,7 @@ import (
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/config"
 	"github.com/bigbluebutton/bbb-webrtc-recorder/internal/webrtc/interfaces"
 	"github.com/pion/rtp"
+	log "github.com/sirupsen/logrus"
 )
 
 // KeyframeRequester defines the interface for requesting keyframes
@@ -34,11 +35,41 @@ type Recorder interface {
 	Close() time.Duration
 }
 
+func CheckFsPermissions(cfg config.Recorder) error {
+	dir := path.Clean(cfg.Directory)
+
+	if err := checkDirectory(dir); err != nil {
+		return err
+	}
+
+	fileMode, _ := parseFileMode(cfg.FileMode)
+
+	tmpFile, err := os.CreateTemp(dir, ".rec-file-perm-check-*")
+
+	if err != nil {
+		return fmt.Errorf("recorder directory is not writable: %w", err)
+	}
+
+	defer func() {
+		_ = tmpFile.Close()
+		if err := os.Remove(tmpFile.Name()); err != nil {
+			log.WithField("file", tmpFile.Name()).Warnf("could not remove permission check file: %v", err)
+		}
+	}()
+
+	// Check if the configured file mode can be applied
+	if err := tmpFile.Chmod(fileMode); err != nil {
+		return fmt.Errorf("cannot apply file mode %s: %w", cfg.FileMode, err)
+	}
+
+	return nil
+}
+
 func ValidateAndPrepareFile(ctx context.Context, cfg config.Recorder, file string) (string, os.FileMode, error) {
 	dir := path.Clean(cfg.Directory)
 
-	if _, err := os.Stat(dir); os.IsNotExist(err) {
-		return "", 0, fmt.Errorf("directory does not exist %s", cfg.Directory)
+	if err := checkDirectory(dir); err != nil {
+		return "", 0, err
 	}
 
 	if !cfg.WriteToDevNull {
@@ -108,3 +139,29 @@ func NewRecorder(ctx context.Context, cfg config.Recorder, file string) (Recorde
 	}
 	return r, nil
 }
+
+func parseFileMode(mode string) (os.FileMode, error) {
+	if parsedFileMode, err := strconv.ParseUint(mode, 0, 32); err != nil {
+		return 0, fmt.Errorf("invalid file mode %s", mode)
+	} else {
+		return os.FileMode(parsedFileMode), nil
+	}
+}
+
+func checkDirectory(dir string) error {
+	if fileInfo, err := os.Stat(dir); err != nil {
+		if os.IsNotExist(err) {
+			return fmt.Errorf("recorder directory does not exist: %s", dir)
+		}
+
+		if err != nil {
+			return fmt.Errorf("could not stat recorder directory %s: %w", dir, err)
+		}
+	} else {
+		if !fileInfo.IsDir() {
+			return fmt.Errorf("recorder path is not a directory: %s", dir)
+		}
+	}
+
+	return nil
+}