Add tailing metrics

pkg/loki/modules.go

@@ -238,7 +238,7 @@ func (t *Loki) initQuerier() (services.Service, error) {
 		return nil, err
 	}
 
-	q, err := querier.New(t.Cfg.Querier, t.Store, t.ingesterQuerier, t.overrides, deleteStore)
+	q, err := querier.New(t.Cfg.Querier, t.Store, t.ingesterQuerier, t.overrides, deleteStore, prometheus.DefaultRegisterer)
 	if err != nil {
 		return nil, err
 	}
@@ -699,7 +699,7 @@ func (t *Loki) initRuler() (_ services.Service, err error) {
 		return nil, err
 	}
 
-	q, err := querier.New(t.Cfg.Querier, t.Store, t.ingesterQuerier, t.overrides, deleteStore)
+	q, err := querier.New(t.Cfg.Querier, t.Store, t.ingesterQuerier, t.overrides, deleteStore, nil)
 	if err != nil {
 		return nil, err
 	}

pkg/querier/metrics.go

@@ -0,0 +1,29 @@
+package querier
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+type Metrics struct {
+	tailsActive         prometheus.Gauge
+	tailedStreamsActive prometheus.Gauge
+	tailedBytesTotal    prometheus.Counter
+}
+
+func NewMetrics(r prometheus.Registerer) *Metrics {
+	return &Metrics{
+		tailsActive: promauto.With(r).NewGauge(prometheus.GaugeOpts{
+			Name: "loki_querier_tail_active",
+			Help: "Number of active tailers",
+		}),
+		tailedStreamsActive: promauto.With(r).NewGauge(prometheus.GaugeOpts{
+			Name: "loki_querier_tail_active_streams",
+			Help: "Number of active streams being tailed",
+		}),
+		tailedBytesTotal: promauto.With(r).NewCounter(prometheus.CounterOpts{
+			Name: "loki_querier_tail_bytes_total",
+			Help: "total bytes tailed",
+		}),
+	}
+}

pkg/querier/querier.go

@@ -6,6 +6,8 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/grafana/loki/pkg/storage/stores/shipper/compactor/deletion"
 
 	"github.com/go-kit/log/level"
@@ -90,20 +92,22 @@ type SingleTenantQuerier struct {
 	limits          *validation.Overrides
 	ingesterQuerier *IngesterQuerier
 	deleteGetter    deleteGetter
+	metrics         *Metrics
 }
 
 type deleteGetter interface {
 	GetAllDeleteRequestsForUser(ctx context.Context, userID string) ([]deletion.DeleteRequest, error)
 }
 
 // New makes a new Querier.
-func New(cfg Config, store storage.Store, ingesterQuerier *IngesterQuerier, limits *validation.Overrides, d deleteGetter) (*SingleTenantQuerier, error) {
+func New(cfg Config, store storage.Store, ingesterQuerier *IngesterQuerier, limits *validation.Overrides, d deleteGetter, r prometheus.Registerer) (*SingleTenantQuerier, error) {
 	return &SingleTenantQuerier{
 		cfg:             cfg,
 		store:           store,
 		ingesterQuerier: ingesterQuerier,
 		limits:          limits,
 		deleteGetter:    d,
+		metrics:         NewMetrics(r),
 	}, nil
 }
 
@@ -454,6 +458,7 @@ func (q *SingleTenantQuerier) Tail(ctx context.Context, req *logproto.TailReques
 		},
 		q.cfg.TailMaxDuration,
 		tailerWaitEntryThrottle,
+		q.metrics,
 	), nil
 }
 

pkg/querier/querier_test.go

@@ -1175,7 +1175,7 @@ func newQuerier(cfg Config, clientCfg client.Config, clientFactory ring_client.P
 		return nil, err
 	}
 
-	return New(cfg, store, iq, limits, dg)
+	return New(cfg, store, iq, limits, dg, nil)
 }
 
 type mockDeleteGettter struct {

pkg/querier/tail.go

@@ -41,6 +41,10 @@ type Tailer struct {
 	currEntry  logproto.Entry
 	currLabels string
 
+	// keep track of the streams for metrics about active streams
+	seenStreams    map[uint64]struct{}
+	seenStreamsMtx sync.Mutex
+
 	tailDisconnectedIngesters func([]string) (map[string]logproto.Querier_TailClient, error)
 
 	querierTailClients    map[string]logproto.Querier_TailClient // addr -> grpc clients for tailing logs from ingesters
@@ -55,6 +59,7 @@ type Tailer struct {
 	// if we are not seeing any response from ingester,
 	// how long do we want to wait by going into sleep
 	waitEntryThrottle time.Duration
+	metrics           *Metrics
 }
 
 func (t *Tailer) readTailClients() {
@@ -95,8 +100,11 @@ func (t *Tailer) loop() {
 
 		// Read as much entries as we can (up to the max allowed) and populate the
 		// tail response we'll send over the response channel
-		tailResponse := new(loghttp.TailResponse)
-		entriesCount := 0
+		var (
+			tailResponse = new(loghttp.TailResponse)
+			entriesCount = 0
+			entriesSize  = 0
+		)
 
 		for ; entriesCount < maxEntriesPerTailResponse && t.next(); entriesCount++ {
 			// If the response channel channel is blocked, we drop the current entry directly
@@ -106,6 +114,7 @@ func (t *Tailer) loop() {
 				continue
 			}
 
+			entriesSize += len(t.currEntry.Line)
 			tailResponse.Streams = append(tailResponse.Streams, logproto.Stream{
 				Labels:  t.currLabels,
 				Entries: []logproto.Entry{t.currEntry},
@@ -151,6 +160,7 @@ func (t *Tailer) loop() {
 
 		select {
 		case t.responseChan <- tailResponse:
+			t.metrics.tailedBytesTotal.Add(float64(entriesSize))
 			if len(droppedEntries) > 0 {
 				droppedEntries = make([]loghttp.DroppedEntry, 0)
 			}
@@ -239,13 +249,18 @@ func (t *Tailer) next() bool {
 
 	t.currEntry = t.openStreamIterator.Entry()
 	t.currLabels = t.openStreamIterator.Labels()
+	t.recordStream(t.openStreamIterator.StreamHash())
+
 	return true
 }
 
 func (t *Tailer) close() error {
 	t.streamMtx.Lock()
 	defer t.streamMtx.Unlock()
 
+	t.metrics.tailsActive.Dec()
+	t.metrics.tailedStreamsActive.Sub(t.activeStreamCount())
+
 	t.stopped = true
 	return t.openStreamIterator.Close()
 }
@@ -264,25 +279,48 @@ func (t *Tailer) getCloseErrorChan() <-chan error {
 	return t.closeErrChan
 }
 
+func (t *Tailer) recordStream(id uint64) {
+	t.seenStreamsMtx.Lock()
+	defer t.seenStreamsMtx.Unlock()
+
+	if _, ok := t.seenStreams[id]; ok {
+		return
+	}
+
+	t.seenStreams[id] = struct{}{}
+	t.metrics.tailedStreamsActive.Inc()
+}
+
+func (t *Tailer) activeStreamCount() float64 {
+	t.seenStreamsMtx.Lock()
+	defer t.seenStreamsMtx.Unlock()
+
+	return float64(len(t.seenStreams))
+}
+
 func newTailer(
 	delayFor time.Duration,
 	querierTailClients map[string]logproto.Querier_TailClient,
 	historicEntries iter.EntryIterator,
 	tailDisconnectedIngesters func([]string) (map[string]logproto.Querier_TailClient, error),
 	tailMaxDuration time.Duration,
 	waitEntryThrottle time.Duration,
+	m *Metrics,
 ) *Tailer {
 	t := Tailer{
 		openStreamIterator:        iter.NewMergeEntryIterator(context.Background(), []iter.EntryIterator{historicEntries}, logproto.FORWARD),
 		querierTailClients:        querierTailClients,
 		delayFor:                  delayFor,
 		responseChan:              make(chan *loghttp.TailResponse, maxBufferedTailResponses),
 		closeErrChan:              make(chan error),
+		seenStreams:               make(map[uint64]struct{}),
 		tailDisconnectedIngesters: tailDisconnectedIngesters,
 		tailMaxDuration:           tailMaxDuration,
 		waitEntryThrottle:         waitEntryThrottle,
+		metrics:                   m,
 	}
 
+	t.metrics.tailsActive.Inc()
 	t.readTailClients()
 	go t.loop()
 	return &t

pkg/querier/tail_test.go

@@ -161,7 +161,7 @@ func TestTailer(t *testing.T) {
 				tailClients["test"] = test.tailClient
 			}
 
-			tailer := newTailer(0, tailClients, test.historicEntries, tailDisconnectedIngesters, timeout, throttle)
+			tailer := newTailer(0, tailClients, test.historicEntries, tailDisconnectedIngesters, timeout, throttle, NewMetrics(nil))
 			defer tailer.close()
 
 			test.tester(t, tailer, test.tailClient)