remove prometheus dependency from tailer package

Author: fstab

grok_exporter.go

@@ -282,5 +282,26 @@ func startTailer(cfg *v2.Config) (tailer.Tailer, error) {
 	default:
 		return nil, fmt.Errorf("Config error: Input type '%v' unknown.", cfg.Input.Type)
 	}
-	return tailer.BufferedTailerWithMetrics(tail), nil
+	return tailer.BufferedTailerWithMetrics(tail, &bufferLoadMetric{}), nil
+}
+
+type bufferLoadMetric struct {
+	bufferLoad prometheus.Summary
+}
+
+func (m *bufferLoadMetric) Register() {
+	bufferLoad := prometheus.NewSummary(prometheus.SummaryOpts{
+		Name: "grok_exporter_line_buffer_peak_load",
+		Help: "Number of lines that are read from the logfile and waiting to be processed. Peak value per second.",
+	})
+	prometheus.MustRegister(bufferLoad)
+	m.bufferLoad = bufferLoad
+}
+
+func (m *bufferLoadMetric) Unregister() {
+	prometheus.Unregister(m.bufferLoad)
+}
+
+func (m *bufferLoadMetric) Observe(currentLoad float64) {
+	m.bufferLoad.Observe(currentLoad)
 }

tailer/bufferedTailer.go

@@ -16,30 +16,33 @@ package tailer
 
 import (
 	"container/list"
-	"github.com/prometheus/client_golang/prometheus"
 	"log"
 	"sync"
 	"time"
 )
 
 // implements Tailer
-type bufferedTailerWithMetrics struct {
+type bufferedTailer struct {
 	out  chan string
 	orig Tailer
 }
 
-func (b *bufferedTailerWithMetrics) Lines() chan string {
+func (b *bufferedTailer) Lines() chan string {
 	return b.out
 }
 
-func (b *bufferedTailerWithMetrics) Errors() chan Error {
+func (b *bufferedTailer) Errors() chan Error {
 	return b.orig.Errors()
 }
 
-func (b *bufferedTailerWithMetrics) Close() {
+func (b *bufferedTailer) Close() {
 	b.orig.Close()
 }
 
+func BufferedTailer(orig Tailer) Tailer {
+	return BufferedTailerWithMetrics(orig, &noopMetric{})
+}
+
 // Wrapper around a tailer that consumes the lines channel quickly.
 // The idea is that the original tailer can continue reading lines from the logfile,
 // and does not need to wait until the lines are processed.
@@ -99,42 +102,38 @@ func (b *bufferedTailerWithMetrics) Close() {
 //
 // To minimize the risk, use the buffered tailer to make sure file system events are handled
 // as quickly as possible without waiting for the grok patterns to be processed.
-func BufferedTailerWithMetrics(orig Tailer) Tailer {
+func BufferedTailerWithMetrics(orig Tailer, bufferLoadMetric BufferLoadMetric) Tailer {
 	buffer := list.New()
 	bufferSync := sync.NewCond(&sync.Mutex{}) // coordinate producer and consumer
 	out := make(chan string)
 
 	// producer
 	go func() {
-		bufferLoad := prometheus.NewSummary(prometheus.SummaryOpts{
-			Name: "grok_exporter_line_buffer_peak_load",
-			Help: "Number of lines that are read from the logfile and waiting to be processed. Peak value per second.",
-		})
-		prometheus.MustRegister(bufferLoad)
+		bufferLoadMetric.Register()
 		bufferLoadPeakValue := 0
 		tick := time.NewTicker(1 * time.Second)
 		for {
 			select {
 			case line, ok := <-orig.Lines():
 				if ok {
 					bufferSync.L.Lock()
+					buffer.PushBack(line)
 					if buffer.Len() > bufferLoadPeakValue {
 						bufferLoadPeakValue = buffer.Len()
 					}
-					buffer.PushBack(line)
 					bufferSync.Signal()
 					bufferSync.L.Unlock()
 				} else {
 					bufferSync.L.Lock()
 					buffer = nil // make the consumer quit
 					bufferSync.Signal()
 					bufferSync.L.Unlock()
-					prometheus.Unregister(bufferLoad)
+					bufferLoadMetric.Unregister()
 					tick.Stop()
 					return
 				}
 			case <-tick.C:
-				bufferLoad.Observe(float64(bufferLoadPeakValue))
+				bufferLoadMetric.Observe(float64(bufferLoadPeakValue))
 				bufferLoadPeakValue = 0
 			}
 		}
@@ -164,8 +163,20 @@ func BufferedTailerWithMetrics(orig Tailer) Tailer {
 			}
 		}
 	}()
-	return &bufferedTailerWithMetrics{
+	return &bufferedTailer{
 		out:  out,
 		orig: orig,
 	}
 }
+
+type BufferLoadMetric interface {
+	Register()
+	Observe(currentLoad float64)
+	Unregister()
+}
+
+type noopMetric struct{}
+
+func (m *noopMetric) Register()                   {}
+func (m *noopMetric) Observe(currentLoad float64) {}
+func (m *noopMetric) Unregister()                 {}

tailer/bufferedTailer_test.go

@@ -41,16 +41,19 @@ func (tail *sourceTailer) Close() {
 // First produce 10,000 lines, then consume 10,000 lines.
 func TestLineBufferSequential(t *testing.T) {
 	src := &sourceTailer{lines: make(chan string)}
-	buffered := BufferedTailerWithMetrics(src)
-	for i := 0; i < 10000; i++ {
+	metric := &peakLoadMetric{}
+	buffered := BufferedTailerWithMetrics(src, metric)
+	for i := 1; i <= 10000; i++ {
 		src.lines <- fmt.Sprintf("This is line number %v.", i)
 	}
-	for i := 0; i < 10000; i++ {
+	for i := 1; i <= 10000; i++ {
 		line := <-buffered.Lines()
 		if line != fmt.Sprintf("This is line number %v.", i) {
 			t.Errorf("Expected 'This is line number %v', but got '%v'.", i, line)
 		}
 	}
+	// wait until peak load is observed (buffered tailer observes the max of each 1 Sec interval)
+	time.Sleep(1100 * time.Millisecond)
 	buffered.Close()
 	_, stillOpen := <-buffered.Lines()
 	if stillOpen {
@@ -60,16 +63,26 @@ func TestLineBufferSequential(t *testing.T) {
 	if stillOpen {
 		t.Error("Source tailer was not closed.")
 	}
+	if !metric.registerCalled {
+		t.Error("metric.Register() not called.")
+	}
+	if !metric.unregisterCalled {
+		t.Error("metric.Unregister() not called.")
+	}
+	// The peak load should be 9999 or 9998, depending on how quick
+	// the consumer loop started reading
+	fmt.Printf("peak load: %v\n", metric.peakLoad)
 }
 
 // Produce and consume in parallel.
 func TestLineBufferParallel(t *testing.T) {
 	src := &sourceTailer{lines: make(chan string)}
-	buffered := BufferedTailerWithMetrics(src)
+	metric := &peakLoadMetric{}
+	buffered := BufferedTailerWithMetrics(src, metric)
 	var wg sync.WaitGroup
 	go func() {
 		start := time.Now()
-		for i := 0; i < 10000; i++ {
+		for i := 1; i <= 10000; i++ {
 			src.lines <- fmt.Sprintf("This is line number %v.", i)
 			if rand.Int()%64 == 0 { // Sleep from time to time
 				time.Sleep(10 * time.Millisecond)
@@ -80,7 +93,7 @@ func TestLineBufferParallel(t *testing.T) {
 	}()
 	go func() {
 		start := time.Now()
-		for i := 0; i < 10000; i++ {
+		for i := 1; i <= 10000; i++ {
 			line := <-buffered.Lines()
 			if line != fmt.Sprintf("This is line number %v.", i) {
 				t.Errorf("Expected 'This is line number %v', but got '%v'.", i, line)
@@ -94,6 +107,8 @@ func TestLineBufferParallel(t *testing.T) {
 	}()
 	wg.Add(2)
 	wg.Wait()
+	// wait until peak load is observed (buffered tailer observes the max of each 1 Sec interval)
+	time.Sleep(1100 * time.Millisecond)
 	buffered.Close()
 	_, stillOpen := <-buffered.Lines()
 	if stillOpen {
@@ -103,4 +118,31 @@ func TestLineBufferParallel(t *testing.T) {
 	if stillOpen {
 		t.Error("Source tailer was not closed.")
 	}
+	if !metric.registerCalled {
+		t.Error("metric.Register() not called.")
+	}
+	if !metric.unregisterCalled {
+		t.Error("metric.Unregister() not called.")
+	}
+	// Should be much less than 10000, because consumer and producer work in parallel.
+	fmt.Printf("peak load: %v\n", metric.peakLoad)
+}
+
+type peakLoadMetric struct {
+	registerCalled, unregisterCalled bool
+	peakLoad                         float64
+}
+
+func (m *peakLoadMetric) Register() {
+	m.registerCalled = true
+}
+
+func (m *peakLoadMetric) Observe(currentLoad float64) {
+	if currentLoad > m.peakLoad {
+		m.peakLoad = currentLoad
+	}
+}
+
+func (m *peakLoadMetric) Unregister() {
+	m.unregisterCalled = true
 }

tailer/fileTailer_test.go

@@ -245,7 +245,7 @@ func testLogrotate(t *testing.T, log simpleLogger, watcherOpt watcherType, logro
 	case polling:
 		tail = RunPollingFileTailer(logfile, true, true, 10*time.Millisecond, log)
 	}
-	tail = BufferedTailerWithMetrics(tail)
+	tail = BufferedTailer(tail)
 	defer tail.Close()
 
 	// We don't expect errors. However, start a go-routine listening on