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

@@ -9,12 +9,14 @@ import (
 	"os"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 
 	"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"
 )
@@ -65,6 +67,9 @@ func (s *CassandraStorageIntegration) initializeCassandra(t *testing.T) {
 		"--cassandra-archive.password=" + password,
 		"--cassandra-archive.username=" + username,
 	})
+	t.Cleanup(func() {
+		assert.NoError(t, f.Close())
+	})
 	s.factory = f
 	var err error
 	s.SpanWriter, err = f.CreateSpanWriter()
@@ -78,9 +83,6 @@ func (s *CassandraStorageIntegration) initializeCassandra(t *testing.T) {
 	s.SamplingStore, err = f.CreateSamplingStore(0)
 	require.NoError(t, err)
 	s.initializeDependencyReaderAndWriter(t, f)
-	t.Cleanup(func() {
-		require.NoError(t, f.Close())
-	})
 }
 
 func (s *CassandraStorageIntegration) initializeDependencyReaderAndWriter(t *testing.T, f *cassandra.Factory) {
@@ -98,6 +100,9 @@ func (s *CassandraStorageIntegration) initializeDependencyReaderAndWriter(t *tes
 }
 
 func TestCassandraStorage(t *testing.T) {
+	t.Cleanup(func() {
+		testutils.VerifyGoLeaksOnce(t)
+	})
 	SkipUnlessEnv(t, "cassandra")
 	s := newCassandraStorageIntegration()
 	s.initializeCassandra(t)

plugin/storage/integration/kafka_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
@@ -20,6 +21,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"
@@ -40,6 +42,9 @@ func (s *KafkaIntegrationTestSuite) initialize(t *testing.T) {
 	topic := "jaeger-kafka-integration-test-" + strconv.FormatInt(time.Now().UnixNano(), 10)
 
 	f := kafka.NewFactory()
+	t.Cleanup(func() {
+		assert.NoError(t, f.Close())
+	})
 	v, command := config.Viperize(f.AddFlags)
 	err := command.ParseFlags([]string{
 		"--kafka.producer.topic",
@@ -56,7 +61,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",
@@ -82,6 +86,9 @@ func (s *KafkaIntegrationTestSuite) initialize(t *testing.T) {
 	traceStore := memory.NewStore()
 	spanConsumer, err := builder.CreateConsumer(logger, metrics.NullFactory, traceStore, options)
 	require.NoError(t, err)
+	t.Cleanup(func() {
+		assert.NoError(t, spanConsumer.Close())
+	})
 	spanConsumer.Start()
 
 	s.SpanWriter = spanWriter
@@ -119,6 +126,9 @@ func (*ingester) FindTraceIDs(context.Context, *spanstore.TraceQueryParameters)
 }
 
 func TestKafkaStorage(t *testing.T) {
+	t.Cleanup(func() {
+		testutils.VerifyGoLeaksOnce(t)
+	})
 	SkipUnlessEnv(t, "kafka")
 	s := &KafkaIntegrationTestSuite{}
 	s.initialize(t)