reflector.go/reflector/stats.go

package reflector

import (
	"fmt"
	"sync"
	"time"

	"github.com/lbryio/lbry.go/v2/extras/stop"

	log "github.com/sirupsen/logrus"
)

// TODO: store daily stats too. and maybe other intervals

type Stats struct {
	mu      *sync.Mutex
	blobs   int
	streams int
	errors  map[string]int
	started bool

	name    string
	logger  *log.Logger
	logFreq time.Duration
	grp     *stop.Group
}

func NewStatLogger(name string, logger *log.Logger, logFreq time.Duration, parentGrp *stop.Group) *Stats {
	return &Stats{
		mu:      &sync.Mutex{},
		grp:     stop.New(parentGrp),
		logger:  logger,
		logFreq: logFreq,
		errors:  make(map[string]int),
		name:    name,
	}
}

func (s *Stats) Start() {
	s.started = true
	s.grp.Add(1)
	go func() {
		defer s.grp.Done()
		s.runSlackLogger()
	}()
}

func (s *Stats) Shutdown() {
	if !s.started {
		return
	}
	s.log()
	s.grp.StopAndWait()
	s.started = false
}

func (s *Stats) AddBlob() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.blobs++
}
func (s *Stats) AddStream() {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.streams++
}

func (s *Stats) runSlackLogger() {
	t := time.NewTicker(s.logFreq)
	for {
		select {
		case <-s.grp.Ch():
			return
		case <-t.C:
			s.log()
		}
	}
}

func (s *Stats) log() {
	s.mu.Lock()
	blobs, streams := s.blobs, s.streams
	s.blobs, s.streams = 0, 0
	errStr := ""
	for name, count := range s.errors {
		errStr += fmt.Sprintf("%d %s, ", count, name)
		delete(s.errors, name)
	}
	s.mu.Unlock()

	if len(errStr) > 2 {
		errStr = errStr[:len(errStr)-2] // trim last comma and space
	}

	s.logger.Printf("%s stats: %d blobs, %d streams, errors: %s", s.name, blobs, streams, errStr)
}