implement in-memory streamview counter

Author: iSchluff

cmd/nginx-syslog-stream-counter/README.md

@@ -0,0 +1,29 @@
+# Usage
+```
+log_format json_logs escape=json '{'
+    '"time_local": "$time_local",'
+    '"remote_addr": "$remote_addr",'                            # client IP
+    '"method": "$request_method",'                              # request method, usually “GET” or “POST” 
+    '"protocol": "$server_protocol",'                           # request protocol, usually “HTTP/1.0”, “HTTP/1.1”, “HTTP/2.0”, or “HTTP/3.0” 
+    '"uri": "$uri",'                                            # current URI in request
+    '"status": "$status",'                                      # response status code
+    '"bytes_sent": "$bytes_sent", '                             # the number of bytes sent to a client
+    '"request_length": "$request_length", '                     # request length (including headers and body)
+    '"connection_requests": "$connection_requests",'            # number of requests made in connection
+    '"upstream": "$upstream_addr", '                            # upstream backend server for proxied requests
+    '"upstream_connect_time": "$upstream_connect_time", '       # upstream handshake time incl. TLS
+    '"upstream_header_time": "$upstream_header_time", '         # time spent receiving upstream headers
+    '"upstream_response_time": "$upstream_response_time", '     # time spend receiving upstream body
+    '"upstream_response_length": "$upstream_response_length", ' # upstream response length
+    '"upstream_cache_status": "$upstream_cache_status", '       # cache HIT/MISS where applicable
+    '"ssl_protocol": "$ssl_protocol", '                         # TLS protocol
+    '"ssl_cipher": "$ssl_cipher", '                             # TLS cipher
+    '"scheme": "$scheme", '                                     # http or https
+    '"user_agent": "$http_user_agent"'
+'}';
+
+access_log syslog:server=unix:/var/log/relay.sock json_logs;
+```
+
+# Customization
+The code must be able to derive the sub playlist from the request path. 

cmd/nginx-syslog-stream-counter/main.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/voc/stream-api/logreceiver"
+	"golang.org/x/sys/unix"
+)
+
+// maxSegmentDuration is the maximum duration of a segment to expect
+const maxSegmentDuration = time.Second * 5
+
+func main() {
+	if err := run(); err != nil {
+		slog.Error("collector failed", "err", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	slidingWindow := flag.Duration("sliding-window-duration", time.Second*30, "duration of the sliding window for the counting")
+	prometheusListen := flag.String("prometheus-listen", ":9273", "listen address of the prometheus endpoint")
+	socket := flag.String("socket", "/var/log/relay.sock", "syslog socket")
+	debug := flag.Bool("debug", false, "enable debug mode")
+	flag.Parse()
+
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, unix.SIGTERM)
+	defer cancel()
+
+	logOpts := &slog.HandlerOptions{
+		Level: slog.LevelInfo,
+	}
+	if *debug {
+		logOpts.Level = slog.LevelDebug
+	}
+	slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, logOpts)))
+
+	reg := prometheus.NewPedanticRegistry()
+
+	// Add the standard process and Go metrics to the custom registry.
+	reg.MustRegister(
+		collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
+		collectors.NewGoCollector(),
+	)
+
+	_, err := logreceiver.NewParser(ctx, logreceiver.ParserConfig{
+		SocketPath:        *socket,
+		MetricsRegisterer: reg,
+		SlidingWindow:     *slidingWindow,
+		MinSegments:       int(*slidingWindow/maxSegmentDuration) / 2,
+	})
+	if err != nil {
+		return err
+	}
+
+	// serve metrics
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{}))
+	server := http.Server{
+		Handler: mux,
+	}
+
+	listener, err := net.Listen("tcp", *prometheusListen)
+	if err != nil {
+		return fmt.Errorf("failed to listen on %s: %w", *prometheusListen, err)
+	}
+	defer listener.Close()
+	slog.Info("serving metrics on", "addr", *prometheusListen)
+
+	go func() {
+		if err := server.Serve(listener); err != nil && err != http.ErrServerClosed {
+			slog.Error("failed to serve prometheus metrics", "err", err)
+			cancel()
+		}
+	}()
+	<-ctx.Done()
+	return nil
+}

logreceiver/dash.go

@@ -0,0 +1,122 @@
+package logreceiver
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"net/url"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/zencoder/go-dash/mpd"
+)
+
+type Streams struct {
+	cache sync.Map
+}
+
+type DashManifest struct {
+	LastRefresh time.Time
+	Qualities   map[string]string
+}
+
+func (s *Streams) getManifest(segmentPath string) DashManifest {
+	base := filepath.Dir(segmentPath)
+	val, ok := s.cache.LoadOrStore(base, DashManifest{Qualities: make(map[string]string)})
+	if !ok {
+		// not present in Cache getInBackground
+		go func(b string) {
+			err := s.getManifestInBackground(b)
+			if err == nil {
+				return
+			}
+
+			log.Println(err.Error())
+			s.cache.Delete(b)
+		}(base)
+	}
+
+	return val.(DashManifest)
+}
+
+func (s *Streams) getManifestInBackground(base string) error {
+	path, err := url.JoinPath("http://127.0.0.1/", base, "manifest.mpd")
+
+	log.Printf("try to get %s\n", path)
+
+	if err != nil {
+		return fmt.Errorf("failed to get manifest path: %s", err)
+	}
+
+	resp, err := http.Get(path)
+	if err != nil {
+		return fmt.Errorf("failed to get manifest: %s", err.Error())
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("failed to get manifest: return status is %d", resp.StatusCode)
+	}
+
+	content, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Errorf("failed to get manifest body: %s", err.Error())
+	}
+
+	manifest, err := mpd.ReadFromString(string(content))
+	if err != nil {
+		return fmt.Errorf("failed to decode manifest: %s", err.Error())
+	}
+
+	mD := DashManifest{}
+	mD.Qualities = make(map[string]string)
+	if len(manifest.Periods) != 1 {
+		return fmt.Errorf("not exactly one period")
+	}
+
+	for _, a := range manifest.Periods[0].AdaptationSets {
+		if a == nil {
+			continue
+		}
+
+		if a.ContentType == nil || *a.ContentType != "video" {
+			continue
+		}
+
+		for _, r := range a.Representations {
+			if r == nil || r.Height == nil || r.ID == nil {
+				continue
+			}
+
+			mD.Qualities[*r.ID] = fmt.Sprintf("%d", *r.Height)
+		}
+	}
+
+	mD.LastRefresh = time.Now()
+	s.cache.Store(base, mD)
+	return nil
+}
+
+func (s *Streams) RefreshRoutine() {
+	ticker := time.NewTicker(time.Minute * 30)
+	for {
+		<-ticker.C
+
+		s.cache.Range(func(key, value any) bool {
+			m := value.(DashManifest)
+			base := key.(string)
+			if time.Since(m.LastRefresh) < time.Hour {
+				return true
+			}
+
+			err := s.getManifestInBackground(base)
+			if err != nil {
+				log.Println(err.Error())
+				s.cache.Delete(base)
+			}
+
+			return true
+		})
+	}
+}