Add go leak check to Cassandra and Kafka e2e tests

pkg/testutils/leakcheck.go

@@ -23,6 +23,17 @@ func IgnoreOpenCensusWorkerLeak() goleak.Option {
 	return goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start")
 }
 
+// IgnoreGoMetricsMeterLeak prevents the leak created by go-metrics which is
+// used by Sarama (Kafka Client) in Jaeger v1. This reason of this leak is
+// not Jaeger but the go-metrics used by Samara.
+// See these issues for the context
+// - https://github.com/IBM/sarama/issues/1321
+// - https://github.com/IBM/sarama/issues/1340
+// - https://github.com/IBM/sarama/issues/2832
+func IgnoreGoMetricsMeterLeak() goleak.Option {
+	return goleak.IgnoreTopFunction("github.com/rcrowley/go-metrics.(*meterArbiter).tick")
+}
+
 // VerifyGoLeaks verifies that unit tests do not leak any goroutines.
 // It should be called in TestMain.
 func VerifyGoLeaks(m *testing.M) {
@@ -36,5 +47,5 @@ func VerifyGoLeaks(m *testing.M) {
 //
 //	defer testutils.VerifyGoLeaksOnce(t)
 func VerifyGoLeaksOnce(t *testing.T) {
-	goleak.VerifyNone(t, IgnoreGlogFlushDaemonLeak(), IgnoreOpenCensusWorkerLeak())
+	goleak.VerifyNone(t, IgnoreGlogFlushDaemonLeak(), IgnoreOpenCensusWorkerLeak(), IgnoreGoMetricsMeterLeak())
 }

plugin/storage/integration/cassandra_test.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
+	"github.com/jaegertracing/jaeger/pkg/testutils"
 	"github.com/jaegertracing/jaeger/plugin/storage/cassandra"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
 )
@@ -80,6 +81,7 @@ func (s *CassandraStorageIntegration) initializeCassandra(t *testing.T) {
 	s.initializeDependencyReaderAndWriter(t, f)
 	t.Cleanup(func() {
 		require.NoError(t, f.Close())
+		testutils.VerifyGoLeaksOnce(t)
 	})
 }
 

plugin/storage/integration/kafka_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/kafka/consumer"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
+	"github.com/jaegertracing/jaeger/pkg/testutils"
 	"github.com/jaegertracing/jaeger/plugin/storage/kafka"
 	"github.com/jaegertracing/jaeger/plugin/storage/memory"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
@@ -56,7 +57,6 @@ func (s *KafkaIntegrationTestSuite) initialize(t *testing.T) {
 
 	spanWriter, err := f.CreateSpanWriter()
 	require.NoError(t, err)
-
 	v, command = config.Viperize(app.AddFlags)
 	err = command.ParseFlags([]string{
 		"--kafka.consumer.topic",
@@ -88,6 +88,11 @@ func (s *KafkaIntegrationTestSuite) initialize(t *testing.T) {
 	s.SpanReader = &ingester{traceStore}
 	s.CleanUp = func(_ *testing.T) {}
 	s.SkipArchiveTest = true
+	t.Cleanup(func() {
+		require.NoError(t, f.Close())
+		require.NoError(t, spanConsumer.Close())
+		testutils.VerifyGoLeaksOnce(t)
+	})
 }
 
 // The ingester consumes spans from kafka and writes them to an in-memory traceStore