working version with Gatherer

pkg/logentry/counters.go

@@ -0,0 +1,67 @@
+package logentry
+
+import (
+	"sync"
+	"time"
+
+	"github.com/grafana/loki/pkg/util"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/model"
+)
+
+type counters struct {
+	mtx      sync.Mutex
+	counters map[model.Fingerprint]prometheus.Counter
+}
+
+// newCounters Counts log entries by streams.
+func newCounters() *counters {
+	return &counters{
+		counters: map[model.Fingerprint]prometheus.Counter{},
+	}
+}
+
+func (c *counters) Gather() ([]*dto.MetricFamily, error) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	var result []*dto.MetricFamily
+	help := "the total count of log entries"
+	name := "log_entries_total"
+	mtype := dto.MetricType_COUNTER
+	counters := &dto.MetricFamily{
+		Help: &help,
+		Name: &name,
+		Type: &mtype,
+	}
+	result = append(result, counters)
+	for _, m := range c.counters {
+		metric := &dto.Metric{}
+		if err := m.Write(metric); err == nil {
+			counters.Metric = append(counters.Metric, metric)
+		}
+	}
+
+	return result, nil
+}
+
+func (c *counters) Handle(labels model.LabelSet, time time.Time, entry string) error {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+
+	fp := labels.Fingerprint()
+	var ok bool
+	var counter prometheus.Counter
+	if counter, ok = c.counters[fp]; !ok {
+		counter = prometheus.NewCounter(prometheus.CounterOpts{
+			Help:        "the total count of log entries",
+			Name:        "log_entries_total",
+			ConstLabels: util.ModelLabelSetToMap(labels),
+		})
+		c.counters[fp] = counter
+	}
+	counter.Inc()
+	return nil
+}

pkg/logentry/counters_test.go

@@ -0,0 +1,46 @@
+package logentry
+
+import (
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	testutil "github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/prometheus/common/model"
+)
+
+var expected = `# HELP log_entries_total the total count of log entries
+# TYPE log_entries_total counter
+log_entries_total 10.0
+log_entries_total{foo="bar"} 5.0
+log_entries_total{bar="foo"} 5.0
+log_entries_total{bar="foo",foo="bar"} 5.0
+`
+
+func Test_newCounters(t *testing.T) {
+	t.Parallel()
+
+	handler := newCounters()
+
+	workerCount := 5
+	var wg sync.WaitGroup
+	for i := 0; i < workerCount; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_ = handler.Handle(model.LabelSet(map[model.LabelName]model.LabelValue{}), time.Now(), "")
+			_ = handler.Handle(model.LabelSet(map[model.LabelName]model.LabelValue{"foo": "bar"}), time.Now(), "")
+			_ = handler.Handle(model.LabelSet(map[model.LabelName]model.LabelValue{"bar": "foo"}), time.Now(), "")
+			_ = handler.Handle(model.LabelSet(map[model.LabelName]model.LabelValue{"bar": "foo", "foo": "bar"}), time.Now(), "")
+			_ = handler.Handle(model.LabelSet(map[model.LabelName]model.LabelValue{}), time.Now(), "")
+
+		}()
+	}
+	wg.Wait()
+
+	if err := testutil.GatherAndCompare(handler, strings.NewReader(expected), "log_entries_total"); err != nil {
+		t.Fatalf("missmatch metrics: %v", err)
+	}
+
+}

pkg/util/conv.go

@@ -2,6 +2,7 @@ package util
 
 import (
 	"github.com/cortexproject/cortex/pkg/ingester/client"
+	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/promql"
 )
 
@@ -14,3 +15,12 @@ func ToClientLabels(labels string) ([]client.LabelAdapter, error) {
 
 	return client.FromLabelsToLabelAdapaters(ls), nil
 }
+
+// ModelLabelSetToMap convert a model.LabelSet to a map[string]string
+func ModelLabelSetToMap(m model.LabelSet) map[string]string {
+	result := map[string]string{}
+	for k, v := range m {
+		result[string(k)] = string(v)
+	}
+	return result
+}