feat: add metrics, observability, test fixtures, and scaling playbook artifacts
This commit is contained in:
1 parent
926a663c23
commit
f9aba11b90
32 files changed
+1788
-124
No files matched your search
+25
-25
@@ -62,32 +62,32 @@ func main() {
|
||||
|
||||
// Register routes
|
||||
http.HandleFunc("/ws", hub.HandleConnection)
|
||||
http.HandleFunc("/api/admin/login", handlers.HandleAdminLogin(deps))
|
||||
http.HandleFunc("/api/admin/logout", handlers.HandleAdminLogout(deps))
|
||||
http.HandleFunc("/api/admin/session", handlers.HandleAdminSession(deps))
|
||||
http.HandleFunc("/api/admin/overview", requireAdmin(handlers.HandleAdminOverview(deps)))
|
||||
http.HandleFunc("/api/admin/playlist", requireAdmin(handlers.HandleAdminPlaylist(deps)))
|
||||
http.HandleFunc("/api/admin/moderation/logs", requireAdmin(handlers.HandleAdminModerationLogs(deps)))
|
||||
http.HandleFunc("/api/admin/moderation/actions", requireAdmin(handlers.HandleAdminModerationAction(deps)))
|
||||
http.HandleFunc("/api/admin/analytics", requireAdmin(handlers.HandleAdminAnalytics(deps)))
|
||||
http.HandleFunc("/api/admin/system", requireAdmin(handlers.HandleAdminSystemSummary(deps)))
|
||||
http.HandleFunc("/api/admin/config", requireAdmin(handlers.HandleAdminConfig(deps)))
|
||||
http.HandleFunc("/api/admin/restart", requireAdmin(handlers.HandleAdminRestart(deps)))
|
||||
http.HandleFunc("/api/admin/users", requireAdmin(handlers.HandleAdminUsers(deps)))
|
||||
http.HandleFunc("/api/admin/schedule", requireAdmin(handlers.HandleAdminSchedule(deps)))
|
||||
http.HandleFunc("/api/admin/schedule/overrides", requireAdmin(handlers.HandleAdminScheduleOverrides(deps)))
|
||||
http.HandleFunc("/api/schedule", handlers.HandlePublicSchedule(deps))
|
||||
http.HandleFunc("/api/admin/media", requireAdmin(handlers.HandleAdminMedia(deps)))
|
||||
http.HandleFunc("/api/admin/media/uploads", requireAdmin(handlers.HandleAdminMediaUploads(deps)))
|
||||
http.HandleFunc("/api/admin/media/uploads/", requireAdmin(handlers.HandleAdminMediaUploads(deps)))
|
||||
http.HandleFunc("/api/admin/media/", requireAdmin(handlers.HandleAdminMediaByID(deps)))
|
||||
http.HandleFunc("/api/admin/chat-log/bulk", requireAdmin(handlers.HandleAdminChatLogBulk(deps)))
|
||||
http.HandleFunc("/api/admin/chat-log/export", requireAdmin(handlers.HandleAdminChatLogExport(deps)))
|
||||
http.HandleFunc("/api/admin/chat-log/", requireAdmin(handlers.HandleAdminChatLogByID(deps)))
|
||||
http.HandleFunc("/api/admin/chat-log", requireAdmin(handlers.HandleAdminChatLog(deps)))
|
||||
http.HandleFunc("/api/admin/login", metrics.WrapRequestTiming(handlers.HandleAdminLogin(deps)))
|
||||
http.HandleFunc("/api/admin/logout", metrics.WrapRequestTiming(handlers.HandleAdminLogout(deps)))
|
||||
http.HandleFunc("/api/admin/session", metrics.WrapRequestTiming(handlers.HandleAdminSession(deps)))
|
||||
http.HandleFunc("/api/admin/overview", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminOverview(deps))))
|
||||
http.HandleFunc("/api/admin/playlist", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminPlaylist(deps))))
|
||||
http.HandleFunc("/api/admin/moderation/logs", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminModerationLogs(deps))))
|
||||
http.HandleFunc("/api/admin/moderation/actions", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminModerationAction(deps))))
|
||||
http.HandleFunc("/api/admin/analytics", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminAnalytics(deps))))
|
||||
http.HandleFunc("/api/admin/system", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminSystemSummary(deps))))
|
||||
http.HandleFunc("/api/admin/config", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminConfig(deps))))
|
||||
http.HandleFunc("/api/admin/restart", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminRestart(deps))))
|
||||
http.HandleFunc("/api/admin/users", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminUsers(deps))))
|
||||
http.HandleFunc("/api/admin/schedule", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminSchedule(deps))))
|
||||
http.HandleFunc("/api/admin/schedule/overrides", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminScheduleOverrides(deps))))
|
||||
http.HandleFunc("/api/schedule", metrics.WrapRequestTiming(handlers.HandlePublicSchedule(deps)))
|
||||
http.HandleFunc("/api/admin/media", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminMedia(deps))))
|
||||
http.HandleFunc("/api/admin/media/uploads", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminMediaUploads(deps))))
|
||||
http.HandleFunc("/api/admin/media/uploads/", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminMediaUploads(deps))))
|
||||
http.HandleFunc("/api/admin/media/", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminMediaByID(deps))))
|
||||
http.HandleFunc("/api/admin/chat-log/bulk", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminChatLogBulk(deps))))
|
||||
http.HandleFunc("/api/admin/chat-log/export", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminChatLogExport(deps))))
|
||||
http.HandleFunc("/api/admin/chat-log/", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminChatLogByID(deps))))
|
||||
http.HandleFunc("/api/admin/chat-log", metrics.WrapRequestTiming(requireAdmin(handlers.HandleAdminChatLog(deps))))
|
||||
http.HandleFunc("/metrics", metrics.HandleMetrics(dbClient.DB, redisWrapper, hub, sessionStore, dbDriver, instanceID))
|
||||
http.HandleFunc("/health", metrics.HandleHealth())
|
||||
http.HandleFunc("/", metrics.HandleNotFound())
|
||||
http.HandleFunc("/health", metrics.WrapRequestTiming(metrics.HandleHealth()))
|
||||
http.HandleFunc("/", metrics.WrapRequestTiming(metrics.HandleNotFound()))
|
||||
|
||||
port := config.ChatPort()
|
||||
log.Printf("Chat service starting on :%s", port)
|
||||
|
||||
@@ -3,9 +3,11 @@ package metrics
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
@@ -15,6 +17,62 @@ import (
|
||||
"chat/src/ws"
|
||||
)
|
||||
|
||||
var requestDurationBuckets = []float64{0.05, 0.1, 0.25, 0.5, 1, 2, 5}
|
||||
|
||||
var chatRequestDurationHistogram = newRequestDurationHistogram(requestDurationBuckets)
|
||||
|
||||
type requestDurationHistogram struct {
|
||||
mu sync.Mutex
|
||||
buckets []float64
|
||||
counts []uint64
|
||||
sum float64
|
||||
total uint64
|
||||
}
|
||||
|
||||
func newRequestDurationHistogram(buckets []float64) *requestDurationHistogram {
|
||||
return &requestDurationHistogram{
|
||||
buckets: append([]float64(nil), buckets...),
|
||||
counts: make([]uint64, len(buckets)+1),
|
||||
}
|
||||
}
|
||||
|
||||
func (h *requestDurationHistogram) observe(durationSeconds float64) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
h.total++
|
||||
h.sum += durationSeconds
|
||||
for index, bucket := range h.buckets {
|
||||
if durationSeconds <= bucket {
|
||||
h.counts[index]++
|
||||
return
|
||||
}
|
||||
}
|
||||
h.counts[len(h.counts)-1]++
|
||||
}
|
||||
|
||||
func (h *requestDurationHistogram) snapshot() (counts []uint64, sum float64, total uint64) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
counts = append([]uint64(nil), h.counts...)
|
||||
return counts, h.sum, h.total
|
||||
}
|
||||
|
||||
// ObserveRequestDuration records a chat HTTP request duration in seconds.
|
||||
func ObserveRequestDuration(duration time.Duration) {
|
||||
chatRequestDurationHistogram.observe(duration.Seconds())
|
||||
}
|
||||
|
||||
// WrapRequestTiming records request duration for the provided handler.
|
||||
func WrapRequestTiming(next http.HandlerFunc) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
startedAt := time.Now()
|
||||
defer ObserveRequestDuration(time.Since(startedAt))
|
||||
next(w, r)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleMetrics returns Prometheus-format metrics for the chat service.
|
||||
func HandleMetrics(gormDB *gorm.DB, redis db.RedisClient, hub *ws.Hub, sessionStore *auth.SessionStore, dbDriver, instanceID string) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -36,6 +94,9 @@ func HandleMetrics(gormDB *gorm.DB, redis db.RedisClient, hub *ws.Hub, sessionSt
|
||||
redisEnabled := 0
|
||||
redisUp := 0
|
||||
redisClients := 0
|
||||
redisMemoryUsedBytes := int64(0)
|
||||
redisMemoryMaxBytes := int64(0)
|
||||
redisMemoryUsageRatio := 0.0
|
||||
if redis != nil {
|
||||
redisEnabled = 1
|
||||
if err := redis.Ping(context.Background()); err == nil {
|
||||
@@ -50,6 +111,19 @@ func HandleMetrics(gormDB *gorm.DB, redis db.RedisClient, hub *ws.Hub, sessionSt
|
||||
}
|
||||
}
|
||||
}
|
||||
if info, err := redis.Info(context.Background(), "memory"); err == nil {
|
||||
if value := parseRedisMemoryBytes(info, "used_memory_human"); value > 0 {
|
||||
redisMemoryUsedBytes = value
|
||||
}
|
||||
if value := parseRedisMemoryBytes(info, "maxmemory_human"); value > 0 {
|
||||
redisMemoryMaxBytes = value
|
||||
} else if value := parseRedisMemoryBytes(info, "maxmemory"); value > 0 {
|
||||
redisMemoryMaxBytes = value
|
||||
}
|
||||
if redisMemoryMaxBytes > 0 {
|
||||
redisMemoryUsageRatio = math.Min(float64(redisMemoryUsedBytes)/float64(redisMemoryMaxBytes), 1.0)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -99,6 +173,18 @@ func HandleMetrics(gormDB *gorm.DB, redis db.RedisClient, hub *ws.Hub, sessionSt
|
||||
fmt.Fprintln(w, "# TYPE linearvoid_chat_redis_connected_clients gauge")
|
||||
writeLine("linearvoid_chat_redis_connected_clients", redisClients)
|
||||
|
||||
fmt.Fprintln(w, "# HELP linearvoid_chat_redis_memory_used_bytes Redis used memory reported by INFO memory.")
|
||||
fmt.Fprintln(w, "# TYPE linearvoid_chat_redis_memory_used_bytes gauge")
|
||||
writeLine("linearvoid_chat_redis_memory_used_bytes", redisMemoryUsedBytes)
|
||||
|
||||
fmt.Fprintln(w, "# HELP linearvoid_chat_redis_memory_max_bytes Redis maxmemory limit reported by INFO memory.")
|
||||
fmt.Fprintln(w, "# TYPE linearvoid_chat_redis_memory_max_bytes gauge")
|
||||
writeLine("linearvoid_chat_redis_memory_max_bytes", redisMemoryMaxBytes)
|
||||
|
||||
fmt.Fprintln(w, "# HELP linearvoid_chat_redis_memory_usage_ratio Redis memory usage ratio (used / max).")
|
||||
fmt.Fprintln(w, "# TYPE linearvoid_chat_redis_memory_usage_ratio gauge")
|
||||
writeLine("linearvoid_chat_redis_memory_usage_ratio", redisMemoryUsageRatio)
|
||||
|
||||
fmt.Fprintln(w, "# HELP linearvoid_chat_admin_sessions_active Active unexpired admin sessions.")
|
||||
fmt.Fprintln(w, "# TYPE linearvoid_chat_admin_sessions_active gauge")
|
||||
writeLine("linearvoid_chat_admin_sessions_active", activeSessions)
|
||||
@@ -106,6 +192,19 @@ func HandleMetrics(gormDB *gorm.DB, redis db.RedisClient, hub *ws.Hub, sessionSt
|
||||
fmt.Fprintln(w, "# HELP linearvoid_chat_build_info Static build info for chat service.")
|
||||
fmt.Fprintln(w, "# TYPE linearvoid_chat_build_info gauge")
|
||||
fmt.Fprintf(w, "linearvoid_chat_build_info{service=\"chat\",database_driver=\"%s\",instance_id=\"%s\"} 1\n", dbDriver, instanceID)
|
||||
|
||||
counts, sum, total := chatRequestDurationHistogram.snapshot()
|
||||
fmt.Fprintln(w, "# HELP linearvoid_chat_request_duration_seconds HTTP request duration for chat API handlers.")
|
||||
fmt.Fprintln(w, "# TYPE linearvoid_chat_request_duration_seconds histogram")
|
||||
cumulative := uint64(0)
|
||||
for index, bucket := range requestDurationBuckets {
|
||||
cumulative += counts[index]
|
||||
fmt.Fprintf(w, "linearvoid_chat_request_duration_seconds_bucket{le=\"%g\"} %d\n", bucket, cumulative)
|
||||
}
|
||||
cumulative += counts[len(counts)-1]
|
||||
fmt.Fprintf(w, "linearvoid_chat_request_duration_seconds_bucket{le=\"+Inf\"} %d\n", cumulative)
|
||||
fmt.Fprintf(w, "linearvoid_chat_request_duration_seconds_sum %g\n", sum)
|
||||
fmt.Fprintf(w, "linearvoid_chat_request_duration_seconds_count %d\n", total)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -125,4 +224,55 @@ func HandleNotFound() http.HandlerFunc {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
fmt.Fprint(w, `{"error":"not found"}`)
|
||||
}
|
||||
}
|
||||
|
||||
func parseRedisMemoryBytes(info, key string) int64 {
|
||||
for _, line := range strings.Split(info, "\n") {
|
||||
if !strings.HasPrefix(line, key+":") {
|
||||
continue
|
||||
}
|
||||
value := strings.TrimSpace(strings.TrimPrefix(line, key+":"))
|
||||
if value == "" {
|
||||
return 0
|
||||
}
|
||||
if raw, err := strconv.ParseInt(value, 10, 64); err == nil {
|
||||
return raw
|
||||
}
|
||||
return parseHumanBytes(value)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func parseHumanBytes(value string) int64 {
|
||||
value = strings.TrimSpace(strings.ToUpper(value))
|
||||
if value == "" {
|
||||
return 0
|
||||
}
|
||||
|
||||
multiplier := float64(1)
|
||||
suffixes := map[string]float64{
|
||||
"B": 1,
|
||||
"K": 1024,
|
||||
"KB": 1024,
|
||||
"M": 1024 * 1024,
|
||||
"MB": 1024 * 1024,
|
||||
"G": 1024 * 1024 * 1024,
|
||||
"GB": 1024 * 1024 * 1024,
|
||||
"T": 1024 * 1024 * 1024 * 1024,
|
||||
"TB": 1024 * 1024 * 1024 * 1024,
|
||||
}
|
||||
|
||||
for suffix, factor := range suffixes {
|
||||
if strings.HasSuffix(value, suffix) {
|
||||
multiplier = factor
|
||||
value = strings.TrimSpace(strings.TrimSuffix(value, suffix))
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
parsed, err := strconv.ParseFloat(value, 64)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return int64(parsed * multiplier)
|
||||
}
|
||||
@@ -54,6 +54,10 @@ func TestHandleNotFound(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestHandleMetrics(t *testing.T) {
|
||||
chatRequestDurationHistogram = newRequestDurationHistogram(requestDurationBuckets)
|
||||
ObserveRequestDuration(100 * time.Millisecond)
|
||||
ObserveRequestDuration(750 * time.Millisecond)
|
||||
|
||||
gormDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open test DB: %v", err)
|
||||
@@ -89,8 +93,12 @@ func TestHandleMetrics(t *testing.T) {
|
||||
"linearvoid_chat_redis_enabled",
|
||||
"linearvoid_chat_redis_up",
|
||||
"linearvoid_chat_redis_connected_clients",
|
||||
"linearvoid_chat_redis_memory_used_bytes",
|
||||
"linearvoid_chat_redis_memory_max_bytes",
|
||||
"linearvoid_chat_redis_memory_usage_ratio",
|
||||
"linearvoid_chat_admin_sessions_active",
|
||||
"linearvoid_chat_build_info",
|
||||
"linearvoid_chat_request_duration_seconds",
|
||||
}
|
||||
|
||||
for _, metric := range requiredMetrics {
|
||||
@@ -117,6 +125,12 @@ func TestHandleMetrics(t *testing.T) {
|
||||
if !strings.Contains(body, "linearvoid_chat_redis_enabled 0") {
|
||||
t.Error("redis_enabled should be 0 (nil redis)")
|
||||
}
|
||||
if !strings.Contains(body, "linearvoid_chat_request_duration_seconds_bucket{le=\"1\"} 2") {
|
||||
t.Error("request duration histogram should include both observed samples by the 1s bucket")
|
||||
}
|
||||
if !strings.Contains(body, "linearvoid_chat_request_duration_seconds_count 2") {
|
||||
t.Error("request duration count should be 2")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleMetrics_WrongMethod(t *testing.T) {
|
||||
@@ -132,6 +146,9 @@ func TestHandleMetrics_WrongMethod(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestHandleMetrics_WithRedis(t *testing.T) {
|
||||
chatRequestDurationHistogram = newRequestDurationHistogram(requestDurationBuckets)
|
||||
ObserveRequestDuration(250 * time.Millisecond)
|
||||
|
||||
gormDB, _ := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
|
||||
gormDB.AutoMigrate(&db.Message{}, &db.ModerationLog{})
|
||||
|
||||
@@ -160,6 +177,15 @@ func TestHandleMetrics_WithRedis(t *testing.T) {
|
||||
if !strings.Contains(body, "linearvoid_chat_redis_connected_clients 5") {
|
||||
t.Error("redis_connected_clients should be 5")
|
||||
}
|
||||
if !strings.Contains(body, "linearvoid_chat_redis_memory_used_bytes 1048576") {
|
||||
t.Error("redis_memory_used_bytes should be 1048576")
|
||||
}
|
||||
if !strings.Contains(body, "linearvoid_chat_redis_memory_usage_ratio 1") {
|
||||
t.Error("redis_memory_usage_ratio should be 1 when maxmemory is unavailable or equal to used memory in tests")
|
||||
}
|
||||
if !strings.Contains(body, "linearvoid_chat_request_duration_seconds_count 1") {
|
||||
t.Error("request duration count should be 1")
|
||||
}
|
||||
}
|
||||
|
||||
// metricsMockRedis is a minimal mock for metrics tests.
|
||||
@@ -185,6 +211,6 @@ func (m *metricsMockRedis) Scan(ctx context.Context, cursor uint64, match string
|
||||
return nil, 0, nil
|
||||
}
|
||||
func (m *metricsMockRedis) Info(ctx context.Context, sections ...string) (string, error) {
|
||||
return "connected_clients:5\nused_memory_human:1M\n", nil
|
||||
return "connected_clients:5\nused_memory_human:1M\nmaxmemory_human:1M\n", nil
|
||||
}
|
||||
func (m *metricsMockRedis) Close() error { return nil }
|
||||
@@ -28,6 +28,9 @@ type VideoFile struct {
|
||||
|
||||
var (
|
||||
r2Client *s3.Client
|
||||
putObject = func(client *s3.Client, input *s3.PutObjectInput) (*s3.PutObjectOutput, error) {
|
||||
return client.PutObject(context.TODO(), input)
|
||||
}
|
||||
bucket = os.Getenv("R2_BUCKET_NAME")
|
||||
playoutMode = "starting" // "live", "slate", "degraded"
|
||||
restartCount = 0
|
||||
@@ -39,6 +42,9 @@ var (
|
||||
const (
|
||||
maxRestartsPerMinute = 5
|
||||
slateManifestName = "playlist.m3u8"
|
||||
// Cache-Control header values for R2 uploads.
|
||||
segmentCacheControl = "public, max-age=31536000, immutable"
|
||||
manifestCacheControl = "public, max-age=2"
|
||||
)
|
||||
|
||||
// initR2Client initializes the R2 (S3-compatible) client
|
||||
@@ -203,7 +209,7 @@ func validateKeyframeInterval(videoPath string) error {
|
||||
"-v", "quiet",
|
||||
"-select_streams", "v:0",
|
||||
"-show_frames",
|
||||
"-show_entries", "frame=key_frame,pkt_pts_time",
|
||||
"-show_entries", "frame=key_frame,best_effort_timestamp_time,pkt_pts_time",
|
||||
"-of", "csv=p=0",
|
||||
videoPath)
|
||||
output, err := cmd.Output()
|
||||
@@ -223,8 +229,16 @@ func validateKeyframeInterval(videoPath string) error {
|
||||
if len(parts) < 2 || parts[0] != "1" {
|
||||
continue
|
||||
}
|
||||
pts, err := strconv.ParseFloat(strings.TrimSpace(parts[1]), 64)
|
||||
if err != nil {
|
||||
var pts float64
|
||||
var found bool
|
||||
for _, raw := range parts[1:] {
|
||||
if value, err := strconv.ParseFloat(strings.TrimSpace(raw), 64); err == nil {
|
||||
pts = value
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
keyframes = append(keyframes, kfEntry{pts})
|
||||
@@ -284,7 +298,11 @@ func ValidatePlaylist(playlistPath string) ([]VideoFile, error) {
|
||||
return videos, nil
|
||||
}
|
||||
|
||||
// SyncToR2 uploads a file to R2 and returns the key
|
||||
// SyncToR2 uploads a file to R2 and returns the key.
|
||||
// Cache-Control is set automatically based on file extension:
|
||||
// - .m4s, .ts, .mp4, .m4a -> immutable (1 year)
|
||||
// - .m3u8 -> short-lived (2s, live manifest)
|
||||
// - everything else -> no cache header
|
||||
func SyncToR2(localPath, r2Key string) error {
|
||||
if r2Client == nil {
|
||||
if err := initR2Client(); err != nil {
|
||||
@@ -298,11 +316,23 @@ func SyncToR2(localPath, r2Key string) error {
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
_, err = r2Client.PutObject(context.TODO(), &s3.PutObjectInput{
|
||||
input := &s3.PutObjectInput{
|
||||
Bucket: aws.String(bucket),
|
||||
Key: aws.String(r2Key),
|
||||
Body: file,
|
||||
})
|
||||
}
|
||||
|
||||
ext := strings.ToLower(filepath.Ext(r2Key))
|
||||
switch ext {
|
||||
case ".m4s", ".ts", ".mp4", ".m4a":
|
||||
input.CacheControl = aws.String(segmentCacheControl)
|
||||
input.ContentType = aws.String("video/MP2T")
|
||||
case ".m3u8":
|
||||
input.CacheControl = aws.String(manifestCacheControl)
|
||||
input.ContentType = aws.String("application/vnd.apple.mpegurl")
|
||||
}
|
||||
|
||||
_, err = putObject(r2Client, input)
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
@@ -1,65 +1,97 @@
|
||||
package main
|
||||
|
||||
// Test fixture generation commands (run from repository root):
|
||||
//
|
||||
// Valid 720p30 fixture (1.5s, H.264/AAC, yuv420p, 48000Hz, keyframe every 30 frames):
|
||||
// ffmpeg -f lavfi -i testsrc2=duration=1.5:size=1280x720:rate=30 \
|
||||
// -f lavfi -i sine=frequency=440:duration=1.5 \
|
||||
// -c:v libx264 -preset ultrafast -crf 18 -pix_fmt yuv420p -g 30 -keyint_min 30 \
|
||||
// -c:a aac -ar 48000 -ac 2 -shortest \
|
||||
// backend/playout/testdata/valid_720p30.mp4
|
||||
//
|
||||
// Invalid h265 fixture (1.5s, HEVC, otherwise compliant):
|
||||
// ffmpeg -f lavfi -i testsrc2=duration=1.5:size=1280x720:rate=30 \
|
||||
// -f lavfi -i sine=frequency=440:duration=1.5 \
|
||||
// -c:v libx265 -preset ultrafast -crf 28 -pix_fmt yuv420p -g 30 -keyint_min 30 \
|
||||
// -c:a aac -ar 48000 -ac 2 -shortest \
|
||||
// -tag:v hvc1 backend/playout/testdata/invalid_h265.mp4
|
||||
//
|
||||
// Invalid 24fps fixture (1.5s, correct codec but wrong framerate):
|
||||
// ffmpeg -f lavfi -i testsrc2=duration=1.5:size=1280x720:rate=24 \
|
||||
// -f lavfi -i sine=frequency=440:duration=1.5 \
|
||||
// -c:v libx264 -preset ultrafast -crf 18 -pix_fmt yuv420p -g 24 -keyint_min 24 \
|
||||
// -c:a aac -ar 48000 -ac 2 -shortest \
|
||||
// backend/playout/testdata/invalid_24fps.mp4
|
||||
//
|
||||
// Test playlist fixtures are plain text files with paths relative to /app/media.
|
||||
// See backend/playout/testdata/*.txt.
|
||||
|
||||
import (
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
"github.com/aws/aws-sdk-go-v2/service/s3"
|
||||
)
|
||||
|
||||
func TestLoadPlaylist(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
func ensureFFProbeInPath(t *testing.T) {
|
||||
t.Helper()
|
||||
|
||||
if _, err := exec.LookPath("ffprobe"); err == nil {
|
||||
return
|
||||
}
|
||||
|
||||
windowsFFProbe := `C:\Users\zwits\AppData\Local\Microsoft\WinGet\Packages\Gyan.FFmpeg_Microsoft.Winget.Source_8wekyb3d8bbwe\ffmpeg-9.0-full_build\bin\ffprobe.exe`
|
||||
if _, err := os.Stat(windowsFFProbe); err == nil {
|
||||
t.Setenv("PATH", filepath.Dir(windowsFFProbe)+string(os.PathListSeparator)+os.Getenv("PATH"))
|
||||
return
|
||||
}
|
||||
|
||||
t.Skip("ffprobe not available on PATH and Windows WinGet install path not found")
|
||||
}
|
||||
|
||||
func TestLoadPlaylist(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
content string
|
||||
wantLen int
|
||||
wantErr bool
|
||||
first string
|
||||
name string
|
||||
file string
|
||||
wantLen int
|
||||
wantErr bool
|
||||
first string
|
||||
}{
|
||||
{
|
||||
name: "simple paths",
|
||||
content: "/videos/video1.mp4\n/videos/video2.mp4\n",
|
||||
name: "valid playlist",
|
||||
file: "valid_playlist.txt",
|
||||
wantLen: 2,
|
||||
first: "/videos/video1.mp4",
|
||||
first: "/app/media/demo1.mp4",
|
||||
},
|
||||
{
|
||||
name: "with file prefix",
|
||||
content: "file /videos/video1.mp4\nfile '/videos/video2.mp4'\n",
|
||||
wantLen: 2,
|
||||
first: "/videos/video1.mp4",
|
||||
},
|
||||
{
|
||||
name: "with file prefix double quotes",
|
||||
content: "file \"/videos/video1.mp4\"\n",
|
||||
wantLen: 1,
|
||||
first: "/videos/video1.mp4",
|
||||
},
|
||||
{
|
||||
name: "with comments and blanks",
|
||||
content: "# comment\n\n/videos/video1.mp4\n\n# another\n/videos/video2.mp4\n",
|
||||
wantLen: 2,
|
||||
first: "/videos/video1.mp4",
|
||||
},
|
||||
{
|
||||
name: "empty file",
|
||||
content: "",
|
||||
name: "empty playlist",
|
||||
file: "empty_playlist.txt",
|
||||
wantLen: 0,
|
||||
},
|
||||
{
|
||||
name: "only comments",
|
||||
content: "# comment\n# another\n",
|
||||
name: "comments only",
|
||||
file: "comments_only.txt",
|
||||
wantLen: 0,
|
||||
},
|
||||
{
|
||||
name: "mixed valid invalid",
|
||||
file: "mixed_valid_invalid.txt",
|
||||
wantLen: 3,
|
||||
first: "/app/media/demo1.mp4",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
path := filepath.Join(tmpDir, tt.name+".txt")
|
||||
os.WriteFile(path, []byte(tt.content), 0644)
|
||||
path := filepath.Join("..", "testdata", tt.file)
|
||||
|
||||
videos, err := LoadPlaylist(path)
|
||||
if tt.wantErr && err == nil {
|
||||
@@ -149,10 +181,60 @@ func TestValidateKeyframeInterval(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestValidateVideoStrict(t *testing.T) {
|
||||
// ValidateVideoStrict calls ValidateVideo which calls ffprobe
|
||||
// Test only the result structure when ValidateVideo returns valid=false
|
||||
// Mock not feasible without full ffprobe mock, skip for now
|
||||
t.Skip("requires ffprobe mock or test media fixtures")
|
||||
ensureFFProbeInPath(t)
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
file string
|
||||
wantValid bool
|
||||
wantErr string
|
||||
}{
|
||||
{
|
||||
name: "valid h264 fixture",
|
||||
file: "valid_720p30.mp4",
|
||||
wantValid: true,
|
||||
},
|
||||
{
|
||||
name: "invalid h265 fixture",
|
||||
file: "invalid_h265.mp4",
|
||||
wantErr: "video codec must be h264",
|
||||
},
|
||||
{
|
||||
name: "invalid framerate fixture",
|
||||
file: "invalid_24fps.mp4",
|
||||
wantErr: "framerate must be 30/1",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
path := filepath.Join("..", "testdata", tt.file)
|
||||
result := ValidateVideoStrict(path)
|
||||
|
||||
valid, _ := result["valid"].(bool)
|
||||
if valid != tt.wantValid {
|
||||
t.Fatalf("valid = %v, want %v", valid, tt.wantValid)
|
||||
}
|
||||
|
||||
errs, _ := result["errors"].([]string)
|
||||
if tt.wantErr == "" {
|
||||
if len(errs) != 0 {
|
||||
t.Fatalf("errors = %v, want none", errs)
|
||||
}
|
||||
if got, ok := result["duration_seconds"].(float64); !ok || got <= 0 {
|
||||
t.Fatalf("duration_seconds = %v, want positive duration", result["duration_seconds"])
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if len(errs) == 0 {
|
||||
t.Fatalf("errors = %v, want message containing %q", errs, tt.wantErr)
|
||||
}
|
||||
if !strings.Contains(strings.Join(errs, "; "), tt.wantErr) {
|
||||
t.Fatalf("errors = %v, want message containing %q", errs, tt.wantErr)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHealthHandler(t *testing.T) {
|
||||
@@ -348,6 +430,129 @@ func TestValidatePlaylist(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestSyncToR2(t *testing.T) {
|
||||
// Requires R2/S3 mock
|
||||
t.Skip("requires S3 client mock")
|
||||
prevClient := r2Client
|
||||
prevBucket := bucket
|
||||
prevPutObject := putObject
|
||||
defer func() {
|
||||
r2Client = prevClient
|
||||
bucket = prevBucket
|
||||
putObject = prevPutObject
|
||||
}()
|
||||
|
||||
tmpDir := t.TempDir()
|
||||
localPath := filepath.Join(tmpDir, "segment.m4s")
|
||||
if err := os.WriteFile(localPath, []byte("segment-bytes"), 0o644); err != nil {
|
||||
t.Fatalf("write fixture: %v", err)
|
||||
}
|
||||
|
||||
bucket = "linearvoid-media"
|
||||
r2Client = &s3.Client{}
|
||||
var gotBucket, gotKey, gotCacheControl, gotContentType string
|
||||
var gotBody []byte
|
||||
putObject = func(_ *s3.Client, input *s3.PutObjectInput) (*s3.PutObjectOutput, error) {
|
||||
gotBucket = aws.ToString(input.Bucket)
|
||||
gotKey = aws.ToString(input.Key)
|
||||
gotCacheControl = aws.ToString(input.CacheControl)
|
||||
gotContentType = aws.ToString(input.ContentType)
|
||||
body, err := io.ReadAll(input.Body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
gotBody = body
|
||||
return &s3.PutObjectOutput{}, nil
|
||||
}
|
||||
|
||||
if err := SyncToR2(localPath, "live/segment.m4s"); err != nil {
|
||||
t.Fatalf("SyncToR2 error: %v", err)
|
||||
}
|
||||
|
||||
if gotBucket != "linearvoid-media" {
|
||||
t.Fatalf("bucket = %q, want %q", gotBucket, "linearvoid-media")
|
||||
}
|
||||
if gotKey != "live/segment.m4s" {
|
||||
t.Fatalf("key = %q, want %q", gotKey, "live/segment.m4s")
|
||||
}
|
||||
if string(gotBody) != "segment-bytes" {
|
||||
t.Fatalf("body = %q, want %q", string(gotBody), "segment-bytes")
|
||||
}
|
||||
if gotCacheControl != segmentCacheControl {
|
||||
t.Fatalf("CacheControl = %q, want %q", gotCacheControl, segmentCacheControl)
|
||||
}
|
||||
if gotContentType != "video/MP2T" {
|
||||
t.Fatalf("ContentType = %q, want %q", gotContentType, "video/MP2T")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncToR2_ManifestCacheHeaders(t *testing.T) {
|
||||
prevClient := r2Client
|
||||
prevBucket := bucket
|
||||
prevPutObject := putObject
|
||||
defer func() {
|
||||
r2Client = prevClient
|
||||
bucket = prevBucket
|
||||
putObject = prevPutObject
|
||||
}()
|
||||
|
||||
tmpDir := t.TempDir()
|
||||
localPath := filepath.Join(tmpDir, "playlist.m3u8")
|
||||
if err := os.WriteFile(localPath, []byte("#EXTM3U\n"), 0o644); err != nil {
|
||||
t.Fatalf("write fixture: %v", err)
|
||||
}
|
||||
|
||||
bucket = "linearvoid-media"
|
||||
r2Client = &s3.Client{}
|
||||
var gotCacheControl, gotContentType string
|
||||
putObject = func(_ *s3.Client, input *s3.PutObjectInput) (*s3.PutObjectOutput, error) {
|
||||
gotCacheControl = aws.ToString(input.CacheControl)
|
||||
gotContentType = aws.ToString(input.ContentType)
|
||||
return &s3.PutObjectOutput{}, nil
|
||||
}
|
||||
|
||||
if err := SyncToR2(localPath, "hls/playlist.m3u8"); err != nil {
|
||||
t.Fatalf("SyncToR2 error: %v", err)
|
||||
}
|
||||
|
||||
if gotCacheControl != manifestCacheControl {
|
||||
t.Fatalf("CacheControl = %q, want %q", gotCacheControl, manifestCacheControl)
|
||||
}
|
||||
if gotContentType != "application/vnd.apple.mpegurl" {
|
||||
t.Fatalf("ContentType = %q, want %q", gotContentType, "application/vnd.apple.mpegurl")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncToR2_NoCacheHeadersForUnknownExt(t *testing.T) {
|
||||
prevClient := r2Client
|
||||
prevBucket := bucket
|
||||
prevPutObject := putObject
|
||||
defer func() {
|
||||
r2Client = prevClient
|
||||
bucket = prevBucket
|
||||
putObject = prevPutObject
|
||||
}()
|
||||
|
||||
tmpDir := t.TempDir()
|
||||
localPath := filepath.Join(tmpDir, "data.bin")
|
||||
if err := os.WriteFile(localPath, []byte("raw"), 0o644); err != nil {
|
||||
t.Fatalf("write fixture: %v", err)
|
||||
}
|
||||
|
||||
bucket = "linearvoid-media"
|
||||
r2Client = &s3.Client{}
|
||||
var gotCacheControl, gotContentType string
|
||||
putObject = func(_ *s3.Client, input *s3.PutObjectInput) (*s3.PutObjectOutput, error) {
|
||||
gotCacheControl = aws.ToString(input.CacheControl)
|
||||
gotContentType = aws.ToString(input.ContentType)
|
||||
return &s3.PutObjectOutput{}, nil
|
||||
}
|
||||
|
||||
if err := SyncToR2(localPath, "misc/data.bin"); err != nil {
|
||||
t.Fatalf("SyncToR2 error: %v", err)
|
||||
}
|
||||
|
||||
if gotCacheControl != "" {
|
||||
t.Fatalf("CacheControl = %q, want empty", gotCacheControl)
|
||||
}
|
||||
if gotContentType != "" {
|
||||
t.Fatalf("ContentType = %q, want empty", gotContentType)
|
||||
}
|
||||
}
|
||||
+2
@@ -0,0 +1,2 @@
|
||||
# Linear Void test playlist
|
||||
# No media entries here
|
||||
+1
@@ -0,0 +1 @@
|
||||
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
@@ -0,0 +1,6 @@
|
||||
# Valid entries should still parse
|
||||
file '/app/media/demo1.mp4'
|
||||
|
||||
# Invalid or malformed lines are preserved as literal paths by the parser
|
||||
not-a-file-entry
|
||||
file '/app/media/demo2.mp4'
|
||||
BIN
Binary file not shown.
+2
@@ -0,0 +1,2 @@
|
||||
file '/app/media/demo1.mp4'
|
||||
file '/app/media/demo2.mp4'
|
||||
Reference in new issue
Block a user