upload-proxy: add basic metrics

Author: iSchluff

cmd/upload-proxy/config.go

@@ -1,6 +1,25 @@
 package main
 
+import (
+	"fmt"
+	"os"
+
+	"github.com/pelletier/go-toml"
+)
+
 type Config struct {
 	ListenAddress string
 	Sinks         []SinkConfig
 }
+
+func (c *Config) ReadFromFile(path string) error {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return fmt.Errorf("read: %w", err)
+	}
+	err = toml.Unmarshal(data, c)
+	if err != nil {
+		return fmt.Errorf("unmarshal: %w", err)
+	}
+	return nil
+}

cmd/upload-proxy/config.toml

@@ -6,12 +6,38 @@ import (
 	"fmt"
 	"os"
 
-	"github.com/pelletier/go-toml"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/voc/stream-api/util"
 )
 
+func main() {
+	conf := Config{}
+	config := flag.String("config", "config.toml", "Set path to proxy config")
+	debug := flag.Bool("debug", false, "sets log level to debug")
+	flag.StringVar(&conf.ListenAddress, "addr", ":8080", "Set listen address")
+	flag.Parse()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	util.HandleSignal(ctx, cancel)
+
+	log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr})
+
+	zerolog.SetGlobalLevel(zerolog.InfoLevel)
+	if *debug {
+		zerolog.SetGlobalLevel(zerolog.DebugLevel)
+	}
+
+	if err := conf.ReadFromFile(*config); err != nil {
+		log.Fatal().Err(err).Msg("config parse failed")
+	}
+
+	// Run proxy
+	if err := run(ctx, &conf); err != nil {
+		log.Fatal().Err(err).Msg("proxy run failed")
+	}
+}
+
 func run(parentCtx context.Context, conf *Config) error {
 	ctx, cancel := context.WithCancel(parentCtx)
 	defer cancel()
@@ -25,7 +51,7 @@ func run(parentCtx context.Context, conf *Config) error {
 			return fmt.Errorf("sink init failed: %w", err)
 		}
 		sinks = append(sinks, sink)
-		log.Info().Str("sink", sink.url.Host).Str("basePath", sink.url.Path).Msg("added sink")
+		log.Info().Str("sink", sink.url.Host).Str("basePath", sink.url.Path).Str("authType", string(sink.conf.AuthType)).Msg("added sink")
 	}
 	proxy, err := NewProxy(ctx, conf.ListenAddress, sinks)
 	if err != nil {
@@ -43,35 +69,3 @@ func run(parentCtx context.Context, conf *Config) error {
 	proxy.Wait()
 	return nil
 }
-
-func main() {
-	conf := Config{}
-	config := flag.String("config", "config.toml", "Set path to proxy config")
-	debug := flag.Bool("debug", false, "sets log level to debug")
-	flag.StringVar(&conf.ListenAddress, "addr", ":8080", "Set listen address")
-	flag.Parse()
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	util.HandleSignal(ctx, cancel)
-
-	log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr})
-
-	zerolog.SetGlobalLevel(zerolog.InfoLevel)
-	if *debug {
-		zerolog.SetGlobalLevel(zerolog.DebugLevel)
-	}
-
-	data, err := os.ReadFile(*config)
-	if err != nil {
-		log.Fatal().Err(err).Msg("config read failed")
-	}
-	err = toml.Unmarshal(data, &conf)
-	if err != nil {
-		log.Fatal().Err(err).Msg("config parse failed")
-	}
-
-	// Run proxy
-	if run(ctx, &conf); err != nil {
-		log.Fatal().Err(err)
-	}
-}

cmd/upload-proxy/main.go

@@ -0,0 +1,5 @@
+# Auto restart server on file changes
+"*.go" {
+    prep: go build -o upload-proxy .
+    daemon +sigterm: ./upload-proxy -addr localhost:8081
+}

cmd/upload-proxy/modd.conf

@@ -20,6 +20,8 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+const MaxFileSize = 50 * 1024 * 1024 // 50 MB
+
 type Proxy struct {
 	sinks     []*Sink
 	errors    chan error
@@ -118,10 +120,10 @@ func getDeadline(path string) time.Time {
 	ext := filepath.Ext(path)
 	now := time.Now()
 	switch ext {
-	case ".m3u8":
-		fallthrough
-	case ".mpd":
+	case ".m3u8", ".mpd":
 		return now.Add(time.Second * 6)
+	case ".jpg", ".jpeg":
+		return now.Add(time.Second * 20)
 	default:
 		return now.Add(time.Second * 60)
 	}
@@ -170,7 +172,7 @@ func (p *Proxy) HandleUpload(w http.ResponseWriter, r *http.Request) {
 	log.Debug().Str("method", r.Method).Str("url", r.URL.Path).Msg("handle")
 	uploadStart := time.Now()
 	var b bytes.Buffer
-	_, err := b.ReadFrom(r.Body)
+	size, err := b.ReadFrom(http.MaxBytesReader(w, r.Body, MaxFileSize))
 	if err != nil {
 		log.Error().Err(err).Msg("read body")
 		w.WriteHeader(500)
@@ -183,8 +185,8 @@ func (p *Proxy) HandleUpload(w http.ResponseWriter, r *http.Request) {
 		return io.NopCloser(bytes.NewReader(b.Bytes())), nil
 	}
 	for _, sink := range p.sinks {
-		req := r.Clone(context.Background())
-		req.ContentLength = r.ContentLength
+		req := r.Clone(p.ctx)
+		req.ContentLength = size
 		req.GetBody = getBody
 		req.Body, _ = getBody()
 		sink.handle(req, getDeadline(r.URL.Path))

cmd/upload-proxy/proxy.go

@@ -52,7 +52,9 @@ func NewSink(conf SinkConfig) (*Sink, error) {
 	if conf.QueueSize == 0 {
 		conf.QueueSize = DefaultQueueSize
 	}
-
+	if conf.Username != "" && (conf.AuthType != AuthTypeBasic && conf.AuthType != AuthTypeDigest) {
+		conf.AuthType = AuthTypeBasic
+	}
 	url, err := url.Parse(conf.Address)
 	if err != nil {
 		return nil, fmt.Errorf("invalid sink address: %w", err)
@@ -109,7 +111,8 @@ outer:
 		default:
 			// drop front of queue
 			<-sink.queue
-			log.Info().Str("sink", sink.url.Host).Msg("queue overflow")
+			sink.metrics.totalNumDropped.Inc()
+			log.Warn().Str("sink", sink.url.Host).Msg("queue full, dropping request")
 			continue
 		}
 	}
@@ -126,71 +129,113 @@ func (sink *Sink) work(ctx context.Context, client *http.Client) {
 		case <-ctx.Done():
 			return
 		case entry := <-sink.queue:
-		retry:
-			for {
-				select {
-				case <-entry.Request.Context().Done():
-					log.Warn().Str("sink", sink.url.Host).Msg("discarding timed out request")
-					break retry
-				default:
-				}
-				res, err := client.Do(entry.Request)
-				if err != nil {
-					log.Error().Str("sink", sink.url.Host).Err(err).Msg("sink error")
-					break retry
-				}
-				res.Body.Close()
-				if res.StatusCode < 200 || res.StatusCode > 299 {
-					log.Warn().
-						Str("sink", sink.url.Host).
-						Str("method", entry.Request.Method).
-						Str("path", entry.Request.URL.Path).
-						Str("status", res.Status).
-						Msg("upload failed")
-
-					// get new body
-					entry.Request.Body, _ = entry.Request.GetBody()
-
-					// retry if we have space in queue
-					time.Sleep(time.Second)
-					continue
-				}
-				// success
-				sink.metrics.totalNumUploaded.Inc()
-				sink.metrics.totalUploadDelay.Add(time.Since(entry.QueuedAt).Seconds())
-				break
+			sink.upload(ctx, entry, client)
+		}
+	}
+}
+
+func (sink *Sink) upload(ctx context.Context, entry SinkEntry, client *http.Client) {
+	for {
+		if entry.Deadline.Before(time.Now()) {
+			log.Warn().Str("sink", sink.url.Host).Msg("discarding timed out request")
+			sink.metrics.totalNumDropped.Inc()
+			return
+		}
+		queuedFor := time.Since(entry.QueuedAt)
+		uploadStart := time.Now()
+		res, err := client.Do(entry.Request)
+		if err != nil {
+			log.Error().Str("sink", sink.url.Host).Err(err).Msg("sink error")
+			sink.metrics.totalNumDropped.Inc()
+			return
+		}
+		_ = res.Body.Close()
+		if res.StatusCode < 200 || res.StatusCode > 299 {
+			log.Warn().
+				Str("sink", sink.url.Host).
+				Str("method", entry.Request.Method).
+				Str("path", entry.Request.URL.Path).
+				Str("status", res.Status).
+				Msg("upload failed")
+			sink.metrics.totalTransientErrors.Inc()
+
+			// get new body
+			entry.Request.Body, _ = entry.Request.GetBody()
+
+			// drop if queue is almost full
+			if len(sink.queue) > cap(sink.queue)-2 {
+				log.Error().Str("sink", sink.url.Host).Err(err).Msg("queue full, dropping request")
+				sink.metrics.totalNumDropped.Inc()
+				return
+			}
+			// retry after short delay
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(time.Second):
 			}
+			continue
 		}
+		// success
+		sink.metrics.totalNumUploaded.Inc()
+		sink.metrics.totalQueueDelay.Add(queuedFor.Seconds())
+		sink.metrics.totalUploadDuration.Add(time.Since(uploadStart).Seconds())
+		sink.metrics.totalBytesUploaded.Add(float64(entry.Request.ContentLength))
+		return
 	}
 }
 
 type SinkMetrics struct {
-	reg              prometheus.Registerer
-	totalNumUploaded prometheus.Counter
-	totalUploadDelay prometheus.Counter
-	queueLength      prometheus.Gauge
+	reg                  prometheus.Registerer
+	totalNumUploaded     prometheus.Counter
+	totalNumDropped      prometheus.Counter
+	totalQueueDelay      prometheus.Counter
+	totalUploadDuration  prometheus.Counter
+	totalBytesUploaded   prometheus.Counter
+	totalTransientErrors prometheus.Counter
 }
 
 func NewSinkMetrics(reg prometheus.Registerer, cfg SinkConfig) *SinkMetrics {
 	return &SinkMetrics{
 		reg: reg,
 		totalNumUploaded: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 			Name: "upload_proxy_sink_uploaded_total",
-			Help: "Total number of uploaded requests",
+			Help: "Number of uploaded requests",
+			ConstLabels: prometheus.Labels{
+				"sink": cfg.Address,
+			},
+		}),
+		totalNumDropped: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "upload_proxy_sink_dropped_total",
+			Help: "Number of dropped requests due to error, timeout or full queue",
+			ConstLabels: prometheus.Labels{
+				"sink": cfg.Address,
+			},
+		}),
+		totalQueueDelay: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "upload_proxy_sink_queue_delay_total",
+			Help: "Queue delay in seconds",
+			ConstLabels: prometheus.Labels{
+				"sink": cfg.Address,
+			},
+		}),
+		totalUploadDuration: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "upload_proxy_sink_upload_duration_total",
+			Help: "Upload duration in seconds",
 			ConstLabels: prometheus.Labels{
 				"sink": cfg.Address,
 			},
 		}),
-		totalUploadDelay: promauto.With(reg).NewCounter(prometheus.CounterOpts{
-			Name: "upload_proxy_sink_upload_delay_total",
-			Help: "Total upload delay in seconds",
+		totalBytesUploaded: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "upload_proxy_sink_bytes_uploaded_total",
+			Help: "Number of bytes uploaded",
 			ConstLabels: prometheus.Labels{
 				"sink": cfg.Address,
 			},
 		}),
-		queueLength: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
-			Name: "upload_proxy_sink_queue_length",
-			Help: "Number of entries in the sink queue",
+		totalTransientErrors: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "upload_proxy_sink_transient_errors_total",
+			Help: "Number of transient errors",
 			ConstLabels: prometheus.Labels{
 				"sink": cfg.Address,
 			},
@@ -200,6 +245,9 @@ func NewSinkMetrics(reg prometheus.Registerer, cfg SinkConfig) *SinkMetrics {
 
 func (m *SinkMetrics) deregister() {
 	m.reg.Unregister(m.totalNumUploaded)
-	m.reg.Unregister(m.totalUploadDelay)
-	m.reg.Unregister(m.queueLength)
+	m.reg.Unregister(m.totalNumDropped)
+	m.reg.Unregister(m.totalQueueDelay)
+	m.reg.Unregister(m.totalUploadDuration)
+	m.reg.Unregister(m.totalBytesUploaded)
+	m.reg.Unregister(m.totalTransientErrors)
 }