Persist time series viewer metrics (#1752)

* WIP persisting time series viewer metrics. Closes #1478

* Remove unused var, move around initial collection

metrics/metrics.go

@@ -13,15 +13,11 @@ import (
 // How often we poll for updates.
 const metricsPollingInterval = 1 * time.Minute
 
-var _getStatus func() models.Status
-
 // CollectedMetrics stores different collected + timestamped values.
 type CollectedMetrics struct {
 	CPUUtilizations  []timestampedValue `json:"cpu"`
 	RAMUtilizations  []timestampedValue `json:"memory"`
 	DiskUtilizations []timestampedValue `json:"disk"`
-
-	Viewers []timestampedValue `json:"-"`
 }
 
 // Metrics is the shared Metrics instance.
@@ -29,8 +25,6 @@ var Metrics *CollectedMetrics
 
 // Start will begin the metrics collection and alerting.
 func Start(getStatus func() models.Status) {
-	_getStatus = getStatus
-
 	host := data.GetServerURL()
 	if host == "" {
 		host = "unknown"
@@ -72,7 +66,6 @@ func Start(getStatus func() models.Status) {
 
 	Metrics = new(CollectedMetrics)
 	go startViewerCollectionMetrics()
-	handlePolling()
 
 	for range time.Tick(metricsPollingInterval) {
 		handlePolling()

metrics/timestampedValue.go

@@ -1,8 +1,21 @@
 package metrics
 
-import "time"
+import (
+	"time"
+
+	"github.com/nakabonne/tstorage"
+)
 
 type timestampedValue struct {
 	Time  time.Time `json:"time"`
 	Value int       `json:"value"`
 }
+
+func makeTimestampedValuesFromDatapoints(dp []*tstorage.DataPoint) []timestampedValue {
+	tv := []timestampedValue{}
+	for _, d := range dp {
+		tv = append(tv, timestampedValue{Time: time.Unix(d.Timestamp, 0), Value: int(d.Value)})
+	}
+
+	return tv
+}

metrics/viewers.go

@@ -3,15 +3,27 @@ package metrics
 import (
 	"time"
 
+	"github.com/nakabonne/tstorage"
+	"github.com/owncast/owncast/core"
 	"github.com/owncast/owncast/core/chat"
 	"github.com/owncast/owncast/core/data"
+	log "github.com/sirupsen/logrus"
 )
 
 // How often we poll for updates.
 const viewerMetricsPollingInterval = 2 * time.Minute
 
+var storage tstorage.Storage
+
 func startViewerCollectionMetrics() {
+	storage, _ = tstorage.NewStorage(
+		tstorage.WithTimestampPrecision(tstorage.Seconds),
+		tstorage.WithDataPath("./data/metrics"),
+	)
+	defer storage.Close()
+
 	collectViewerCount()
+	handlePolling()
 
 	for range time.Tick(viewerMetricsPollingInterval) {
 		collectViewerCount()
@@ -19,16 +31,15 @@ func startViewerCollectionMetrics() {
 }
 
 func collectViewerCount() {
-	if len(Metrics.Viewers) > maxCollectionValues {
-		Metrics.Viewers = Metrics.Viewers[1:]
+	// Don't collect metrics for viewers if there's no stream active.
+	if !core.GetStatus().Online {
+		return
 	}
+}
 
-	count := _getStatus().ViewerCount
-	value := timestampedValue{
-		Value: count,
-		Time:  time.Now(),
-	}
-	Metrics.Viewers = append(Metrics.Viewers, value)
+func collectChatClientCount() {
+	count := len(chat.GetClients())
+	activeChatClientCount.Set(float64(count))
 
 	// Save to our Prometheus collector.
 	activeViewerCount.Set(float64(count))
@@ -40,9 +51,24 @@ func collectViewerCount() {
 	// Total user count
 	uc := data.GetUsersCount()
 	chatUserCount.Set(float64(uc))
+
+	if err := storage.InsertRows([]tstorage.Row{
+		{
+			Metric:    "viewercount",
+			DataPoint: tstorage.DataPoint{Timestamp: time.Now().Unix(), Value: float64(count)},
+		},
+	}); err != nil {
+		log.Errorln(err)
+	}
 }
 
-func collectChatClientCount() {
-	count := len(chat.GetClients())
-	activeChatClientCount.Set(float64(count))
+// GetViewersOverTime will return a window of viewer counts over time.
+func GetViewersOverTime(start, end time.Time) []timestampedValue {
+	p, err := storage.Select("viewercount", nil, start.Unix(), end.Unix())
+	if err != nil && err != tstorage.ErrNoDataPoints {
+		log.Errorln(err)
+	}
+	datapoints := makeTimestampedValuesFromDatapoints(p)
+
+	return datapoints
 }