feat(runner/metrics): Add metric broker (#1496)

* Revert "Revert "chore(deps): update frontend dependencies" (#1477)"

This reverts commit ca0034a.

* feat(runner): implement runner registration with gocast

* fix: lint protofile

* feat(runner): implement heartbeat notification

* chore: lint imports/code

* feat(runner): get streaming to work

* feat(runner/stream): rework streaming logic

* feat(runner/metrics): add broker to collect metrics

* chore(runner/metrics): Add documentation and tests

* Revert "Revert "Revert "chore(deps): update frontend dependencies" (#1477)""

This reverts commit fda4d5e.

* fix: re-add comment for runner model

Author: joschahenningsen

runner/cmd/runner/main.go

@@ -20,6 +20,7 @@ func main() {
 	shouldShutdown := false // set to true once we receive a shutdown signal
 
 	currentCount := 0
+
 	go func() {
 		for {
 			currentCount += <-r.JobCount // count Job start/stop

runner/pkg/actions/actions.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"log/slog"
 
+	"github.com/tum-dev/gocast/runner/pkg/metrics"
 	"github.com/tum-dev/gocast/runner/protobuf"
 )
 
@@ -13,4 +14,4 @@ import (
 // Actions should use log for logging and notify for sending messages like their progress to gocast.
 // d contains data passed to the action and is used to pass data to the next actions.
 // Any error, the action returns will be logged. If that error is an AbortingError, the subsequent actions will be skipped.
-type Action func(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notification, d map[string]any) error
+type Action func(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notification, d map[string]any, metrics *metrics.Broker) error

runner/pkg/actions/stream.go

@@ -13,11 +13,12 @@ import (
 	"time"
 
 	"github.com/tum-dev/gocast/runner/config"
+	"github.com/tum-dev/gocast/runner/pkg/metrics"
 	"github.com/tum-dev/gocast/runner/pkg/ptr"
 	"github.com/tum-dev/gocast/runner/protobuf"
 )
 
-func Stream(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notification, d map[string]any) error {
+func Stream(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notification, d map[string]any, metrics *metrics.Broker) error {
 	if ctx.Err() != nil {
 		return AbortingError(ctx.Err())
 	}
@@ -51,6 +52,11 @@ func Stream(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notific
 	}
 	log = log.With("stream_id", streamID)
 
+	metrics.Streams.With(metrics.With().Stream(streamID).Source(input).L()).Inc()
+	defer func() {
+		metrics.Streams.With(metrics.With().Stream(streamID).Source(input).L()).Dec()
+	}()
+
 	// e.g. ./storage/live/1/STREAM_VERSION_COMBINED
 	liveRecDir := path.Join(config.Config.SegmentPath, fmt.Sprintf("%d", streamID), streamVersion)
 	err := os.MkdirAll(liveRecDir, os.ModePerm)
@@ -96,6 +102,7 @@ func Stream(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notific
 	go logCmdPipe(log, stdo, []any{"stream", streamID, "logStream", "stdout"})
 	err = command.Run()
 	if err != nil {
+		metrics.StreamErrors.With(metrics.With().Stream(streamID).Source(input).L()).Inc()
 		return err
 	}
 	return nil

runner/pkg/actions/stream_end.go

@@ -5,13 +5,14 @@ import (
 	"fmt"
 	"log/slog"
 
+	"github.com/tum-dev/gocast/runner/pkg/metrics"
 	"github.com/tum-dev/gocast/runner/protobuf"
 )
 
 // StreamEnd is an action who's sole purpose is to notify gocast about the end of a stream.
 // the only reason it is a separate action is to avoid sending unnecessary
 // stream_end notifications if Stream errors.
-func StreamEnd(_ context.Context, _ *slog.Logger, notify chan *protobuf.Notification, d map[string]any) error {
+func StreamEnd(_ context.Context, _ *slog.Logger, notify chan *protobuf.Notification, d map[string]any, metrics *metrics.Broker) error {
 	streamID, ok := d["streamID"].(uint64)
 	if !ok {
 		return AbortingError(fmt.Errorf("no stream id in context"))

runner/pkg/metrics/broker.go

@@ -0,0 +1,92 @@
+// Package metrics provides Prometheus compatible monitoring.
+package metrics
+
+import (
+	"fmt"
+	"log/slog"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+// Broker manages Prometheus metrics.
+type Broker struct {
+	port         int
+	Streams      *prometheus.GaugeVec
+	StreamErrors *prometheus.CounterVec
+}
+
+// Option represents a functional option for configuring a Broker.
+type Option func(broker *Broker)
+
+// NewBroker initializes a new Broker with optional configurations.
+// It sets up Prometheus all available metrics.
+//
+// Example usage:
+//
+//	b := metrics.NewBroker(metrics.WithPort(8080))
+//	go b.Run()
+//
+//	b.Streams.With(b.With().Stream(123).Input("rtmp://1.2.3.4/src")).Set(5)  // Set active streams
+func NewBroker(options ...Option) *Broker {
+	b := &Broker{
+		port: 9947,
+		Streams: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: "runner",
+			Subsystem: "stream",
+			Name:      "n_streams",
+			Help:      "Number of active streams",
+		}, []string{"stream_id", "input"}),
+
+		StreamErrors: promauto.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "runner",
+			Subsystem: "stream",
+			Name:      "n_errors",
+			Help:      "Number of stream ffmpeg errors",
+		}, []string{"stream_id", "input"}),
+	}
+	for _, option := range options {
+		option(b)
+	}
+	return b
+}
+
+// WithPort configures the Broker to listen on a custom port.
+func WithPort(port int) Option {
+	return func(broker *Broker) {
+		broker.port = port
+	}
+}
+
+// Run starts an HTTP server that exposes the metrics.
+func (b *Broker) Run() {
+	http.Handle("/metrics", promhttp.Handler())
+	slog.Error("Serving metrics", "err", http.ListenAndServe(fmt.Sprintf(":%d", b.port), nil))
+}
+
+// LabelBuilder helps construct Prometheus labels dynamically.
+type LabelBuilder prometheus.Labels
+
+// L converts LabelBuilder to a Prometheus Labels map.
+func (b LabelBuilder) L() prometheus.Labels {
+	return prometheus.Labels(b)
+}
+
+// With initializes a new LabelBuilder.
+func (b *Broker) With() LabelBuilder {
+	return LabelBuilder{}
+}
+
+// Stream adds a "stream_id" label to LabelBuilder.
+func (b LabelBuilder) Stream(streamID uint64) LabelBuilder {
+	b["stream_id"] = fmt.Sprintf("%d", streamID)
+	return b
+}
+
+// Source adds a "source" label to LabelBuilder.
+func (b LabelBuilder) Source(source string) LabelBuilder {
+	b["source"] = source
+	return b
+}

runner/pkg/metrics/broker_test.go

@@ -0,0 +1,18 @@
+package metrics
+
+import "testing"
+
+func TestLabelBuilder(t *testing.T) {
+	b := NewBroker()
+	if b == nil {
+		t.Errorf("broker unexpectedly nil")
+		return
+	}
+	if b.StreamErrors == nil {
+		t.Errorf("broker metric not initialized")
+	}
+	labels := b.With().Stream(123).Source("test").L()
+	if labels["stream_id"] != "123" || labels["source"] != "test" {
+		t.Errorf("Unexpected labels: %+v", labels)
+	}
+}

runner/protobuf/notifications.pb.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/tum-dev/gocast/runner/config"
 	"github.com/tum-dev/gocast/runner/pkg/actions"
+	"github.com/tum-dev/gocast/runner/pkg/metrics"
 	"github.com/tum-dev/gocast/runner/pkg/netutil"
 	"github.com/tum-dev/gocast/runner/pkg/ptr"
 	"github.com/tum-dev/gocast/runner/pkg/vmstat"
@@ -51,6 +52,7 @@ type Runner struct {
 	protobuf.UnimplementedRunnerServiceServer
 
 	notifications chan *protobuf.Notification
+	Metrics       *metrics.Broker
 }
 
 func NewRunner(v string) *Runner {
@@ -70,6 +72,7 @@ func NewRunner(v string) *Runner {
 		stats:         vmstats,
 		StartTime:     start,
 		notifications: make(chan *protobuf.Notification),
+		Metrics:       metrics.NewBroker(),
 	}
 }
 
@@ -85,7 +88,8 @@ func (r *Runner) Run() {
 		config.Config.Port = p
 	}
 	r.log.Info("using port", "port", config.Config.Port)
-
+	
+	go r.Metrics.Run()
 	go r.handleNotifications()
 	go r.InitApiGrpc()
 	go func() {
@@ -147,7 +151,6 @@ func (r *Runner) InitApiGrpc() {
 		r.log.Error("failed to serve", "error", err)
 		os.Exit(1)
 	}
-
 }
 
 func (r *Runner) RunAction(a []actions.Action, data map[string]any) string {
@@ -165,7 +168,7 @@ func (r *Runner) RunAction(a []actions.Action, data map[string]any) string {
 		for _, action := range a {
 			for {
 				log := r.log.With("action", getFunctionName(action)).With("job", job)
-				err := action(c, log, r.notifications, data)
+				err := action(c, log, r.notifications, data, r.Metrics)
 				if err != nil {
 					log.Error("action error", "error", err) // use action specific logger
 					if actions.IsAbortingError(err) {