Add logs to Logs page (#151)

* Add websocket connection

* Add logs to the log page

* Make debug color more readable

* Remove TODO from front page

* Put all log entries in latest first order

* Add filtering of log entries by level

* Limit log entries and throttle updates

* Fix logger not throttling broadcasts

* Remove now unnecessary UI-side log throttling

* Filter incoming logs by log level

* Make log view more terminal-like

pkg/api/resolver_query_logs.go

@@ -0,0 +1,23 @@
+package api
+
+import (
+	"context"
+
+	"github.com/stashapp/stash/pkg/logger"
+	"github.com/stashapp/stash/pkg/models"
+)
+
+func (r *queryResolver) Logs(ctx context.Context) ([]*models.LogEntry, error) {
+	logCache := logger.GetLogCache()
+	ret := make([]*models.LogEntry, len(logCache))
+
+	for i, entry := range logCache {
+		ret[i] = &models.LogEntry{
+			Time:    entry.Time,
+			Level:   getLogLevel(entry.Type),
+			Message: entry.Message,
+		}
+	}
+
+	return ret, nil
+}

pkg/api/resolver_subscription_logging.go

@@ -0,0 +1,60 @@
+package api
+
+import (
+	"context"
+
+	"github.com/stashapp/stash/pkg/logger"
+	"github.com/stashapp/stash/pkg/models"
+)
+
+func getLogLevel(logType string) models.LogLevel {
+	if logType == "progress" {
+		return models.LogLevelProgress
+	} else if logType == "debug" {
+		return models.LogLevelDebug
+	} else if logType == "info" {
+		return models.LogLevelInfo
+	} else if logType == "warn" {
+		return models.LogLevelWarning
+	} else if logType == "error" {
+		return models.LogLevelError
+	}
+
+	// default to debug
+	return models.LogLevelDebug
+}
+
+func logEntriesFromLogItems(logItems []logger.LogItem) []*models.LogEntry {
+	ret := make([]*models.LogEntry, len(logItems))
+
+	for i, entry := range logItems {
+		ret[i] = &models.LogEntry{
+			Time:    entry.Time,
+			Level:   getLogLevel(entry.Type),
+			Message: entry.Message,
+		}
+	}
+
+	return ret
+}
+
+func (r *subscriptionResolver) LoggingSubscribe(ctx context.Context) (<-chan []*models.LogEntry, error) {
+	ret := make(chan []*models.LogEntry, 100)
+	stop := make(chan int, 1)
+	logSub := logger.SubscribeToLog(stop)
+
+	go func() {
+		for {
+			select {
+			case logEntries := <-logSub:
+				ret <- logEntriesFromLogItems(logEntries)
+			case <-ctx.Done():
+				stop <- 0
+				close(ret)
+				return
+			}
+		}
+	}()
+
+	return ret, nil
+}

pkg/logger/logger.go

@@ -2,13 +2,16 @@ package logger
 
 import (
 	"fmt"
-	"github.com/sirupsen/logrus"
 	"sync"
+	"time"
+
+	"github.com/sirupsen/logrus"
 )
 
 type LogItem struct {
-	Type    string `json:"type"`
-	Message string `json:"message"`
+	Time    time.Time `json:"time"`
+	Type    string    `json:"type"`
+	Message string    `json:"message"`
 }
 
 var logger = logrus.New()
@@ -16,14 +19,99 @@ var progressLogger = logrus.New()
 
 var LogCache []LogItem
 var mutex = &sync.Mutex{}
+var logSubs []chan []LogItem
+var waiting = false
+var lastBroadcast = time.Now()
+var logBuffer []LogItem
 
 func addLogItem(l *LogItem) {
 	mutex.Lock()
+	l.Time = time.Now()
 	LogCache = append([]LogItem{*l}, LogCache...)
 	if len(LogCache) > 30 {
 		LogCache = LogCache[:len(LogCache)-1]
 	}
 	mutex.Unlock()
+	go broadcastLogItem(l)
+}
+
+func GetLogCache() []LogItem {
+	mutex.Lock()
+
+	ret := make([]LogItem, len(LogCache))
+	copy(ret, LogCache)
+
+	mutex.Unlock()
+
+	return ret
+}
+
+func SubscribeToLog(stop chan int) <-chan []LogItem {
+	ret := make(chan []LogItem, 100)
+
+	go func() {
+		<-stop
+		unsubscribeFromLog(ret)
+	}()
+
+	mutex.Lock()
+	logSubs = append(logSubs, ret)
+	mutex.Unlock()
+
+	return ret
+}
+
+func unsubscribeFromLog(toRemove chan []LogItem) {
+	mutex.Lock()
+	for i, c := range logSubs {
+		if c == toRemove {
+			logSubs = append(logSubs[:i], logSubs[i+1:]...)
+		}
+	}
+	close(toRemove)
+	mutex.Unlock()
+}
+
+func doBroadcastLogItems() {
+	// assumes mutex held
+
+	for _, c := range logSubs {
+		// don't block waiting to broadcast
+		select {
+		case c <- logBuffer:
+		default:
+		}
+	}
+
+	logBuffer = nil
+	waiting = false
+	lastBroadcast = time.Now()
+}
+
+func broadcastLogItem(l *LogItem) {
+	mutex.Lock()
+
+	logBuffer = append(logBuffer, *l)
+
+	// don't send more than once per second
+	if !waiting {
+		// if last broadcast was under a second ago, wait until a second has
+		// passed
+		timeSinceBroadcast := time.Since(lastBroadcast)
+		if timeSinceBroadcast.Seconds() < 1 {
+			waiting = true
+			time.AfterFunc(time.Second-timeSinceBroadcast, func() {
+				mutex.Lock()
+				doBroadcastLogItems()
+				mutex.Unlock()
+			})
+		} else {
+			doBroadcastLogItems()
+		}
+	}
+	// if waiting then adding it to the buffer is sufficient
+
+	mutex.Unlock()
 }
 
 func init() {