func (w *batchWatcher) watch() {
	t := time.NewTicker(checkRatioInterval)
	defer t.Stop()
	for {
		select {
		case <-t.C:
		case <-w.stop:
			return
		}
		w.outgoingBatches.Tick()
		w.incomingBatches.Tick()
		outgoingRate := w.outgoingBatches.Rate()
		incomingRate := w.incomingBatches.Rate()
		if outgoingRate > 0 {
			r := incomingRate / outgoingRate
			if r > reportRatioThreshold {
				warnHighRatio(r, incomingRate, outgoingRate)
			}
		}
		if w.shouldReportLastDuration.Load().(bool) {
			warnSlowIngestion(w.lastIngestionDuration.Load().(time.Duration))
			w.shouldReportLastDuration.Store(false)
		}
	}
}
watchBatches watches the ratio between incomingBatches and outgoingBatches every checkRatioInterval.
package ingestor
import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/timescale/promscale/pkg/ewma"
	"github.com/timescale/promscale/pkg/log"
)
const (
	reportRatioThreshold = 3
	checkRatioInterval   = time.Minute
	reportDurationThreshold = time.Minute
)
type batchWatcher struct {
	incomingBatches *ewma.Rate
	outgoingBatches *ewma.Rate
	shouldReportLastDuration atomic.Value
	lastIngestionDuration    atomic.Value
	stop chan struct{}
}
var (
	watcher  *batchWatcher
	tWatcher = new(sync.Once)
)
func runBatchWatcher(stop chan struct{}) {
	tWatcher.Do(func() {
		watcher := newThroughputWatcher(stop)
		go watcher.watch()
	})
}
func newThroughputWatcher(stop chan struct{}) *batchWatcher {
	watcher = &batchWatcher{
		incomingBatches: ewma.NewEWMARate(1, checkRatioInterval),
		outgoingBatches: ewma.NewEWMARate(1, checkRatioInterval),
		stop:            stop,
	}
	watcher.shouldReportLastDuration.Store(false)
	return watcher
}