Fixing all goroutine leaks with client closings

Also adds uber's goroutine leaks checker to almost all tests

The biggest change here is adding context to sidecar pull, which is then
cancelled if the container creation fails. This prevents the sidecar goroutine from leaking.
The same for making the sidecar channel buffered; the goroutine is not stuck
when nobody reads from it. (It is not a memory leak either, GC will remove the channel.)

Other changes are just annoying work; putting client.Close() everywhere,
db.Close() when Ping() is not successful, client closes in tests/presets, etc.

internal/gnomockd/ tests still show goroutine leaks for http transport, but I have no idea what is gnomockd even doing,
let alone the tests, so I let that be.

.golangci.yml

@@ -37,6 +37,8 @@ linters:
  - wsl
 
 issues:
+ include:
+ - EXC0001 # require error check on Close()
  exclude-rules:
  - path: _test\.go
  linters:

docker.go

@@ -137,15 +137,17 @@ func (d *docker) startContainer(ctx context.Context, image string, ports NamedPo
  return nil, fmt.Errorf("can't prepare container: %w", err)
  }
 
- sidecarChan := d.setupContainerCleanup(resp.ID, cfg)
+ sidecarChan, cleanupCancel := d.setupContainerCleanup(resp.ID, cfg)
 
  err = d.client.ContainerStart(ctx, resp.ID, container.StartOptions{})
  if err != nil {
+ cleanupCancel()
  return nil, fmt.Errorf("can't start container %s: %w", resp.ID, err)
  }
 
  container, err := d.waitForContainerNetwork(ctx, resp.ID, ports)
  if err != nil {
+ cleanupCancel()
  return nil, fmt.Errorf("container network isn't ready: %w", err)
  }
 
@@ -158,8 +160,9 @@ func (d *docker) startContainer(ctx context.Context, image string, ports NamedPo
  return container, nil
 }
 
-func (d *docker) setupContainerCleanup(id string, cfg *Options) chan string {
- sidecarChan := make(chan string)
+func (d *docker) setupContainerCleanup(id string, cfg *Options) (chan string, context.CancelFunc) {
+ sidecarChan := make(chan string, 1)
+ bctx, bcancel := context.WithCancel(context.Background())
 
  go func() {
  defer close(sidecarChan)
@@ -174,9 +177,10 @@ func (d *docker) setupContainerCleanup(id string, cfg *Options) chan string {
  WithHealthCheck(func(ctx context.Context, c *Container) error {
  return health.HTTPGet(ctx, c.DefaultAddress())
  }),
- WithInit(func(ctx context.Context, c *Container) error {
- return cleaner.Notify(ctx, c.DefaultAddress(), id)
+ WithInit(func(_ context.Context, c *Container) error {
+ return cleaner.Notify(bctx, c.DefaultAddress(), id)
  }),
+ WithContext(bctx),
  }
  if cfg.UseLocalImagesFirst {
  opts = append(opts, WithUseLocalImagesFirst())
@@ -190,7 +194,7 @@ func (d *docker) setupContainerCleanup(id string, cfg *Options) chan string {
  }
  }()
 
- return sidecarChan
+ return sidecarChan, bcancel
 }
 
 func (d *docker) prepareContainer(
@@ -453,6 +457,10 @@ func (d *docker) stopContainer(ctx context.Context, id string) error {
  return nil
 }
 
+func (d *docker) stopClient() error {
+ return d.client.Close()
+}
+
 func (d *docker) removeContainer(ctx context.Context, id string) error {
  d.lock.Lock()
  defer d.lock.Unlock()

gnomock.go

@@ -114,6 +114,8 @@ func newContainer(g *g, image string, ports NamedPorts, config *Options) (c *Con
  return nil, fmt.Errorf("can't create docker client: %w", err)
  }
 
+ defer func() { _ = cli.stopClient() }()
+
  c, err = cli.startContainer(ctx, image, ports, config)
  if err != nil {
  return nil, fmt.Errorf("can't start container: %w", err)
@@ -230,13 +232,16 @@ func (g *g) stop(c *Container) error {
  return fmt.Errorf("can't create docker client: %w", err)
  }
 
+ defer func() { _ = cli.stopClient() }()
+
  id, sidecar := parseID(c.ID)
  if sidecar != "" {
  go func() {
  // stop sidecar container when the main one is requested to stop;
  // error in this case won't matter, the container has a self-destruct
  // timer
  _ = cli.stopContainer(context.Background(), sidecar)
+ _ = cli.stopClient()
  }()
  }
 

gnomock_internal_test.go

@@ -8,10 +8,15 @@ import (
  "time"
 
  "github.com/stretchr/testify/require"
+ "go.uber.org/goleak"
 )
 
 const testImage = "docker.io/orlangure/gnomock-test-image"
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestWaitForContainerNetwork(t *testing.T) {
  t.Parallel()
 
@@ -36,6 +41,10 @@ func TestWaitForContainerNetwork(t *testing.T) {
  d, err := gg.dockerConnect()
  require.NoError(t, err)
 
+ defer func() {
+ require.NoError(t, d.stopClient())
+ }()
+
  ctx := context.Background()
 
  t.Run("fails after context cancellation", func(t *testing.T) {
@@ -57,6 +66,8 @@ func TestWaitForContainerNetwork(t *testing.T) {
  require.Error(t, err)
  require.True(t, errors.Is(err, ErrPortNotFound), err.Error())
  })
+
+ require.NoError(t, Stop(container))
 }
 
 func TestEnvAwareClone(t *testing.T) {

gnomock_test.go

@@ -167,6 +167,7 @@ func TestGnomock_withDebugMode(t *testing.T) {
  containerList, err = testutil.ListContainerByID(cli, container.ID)
  require.NoError(t, err)
  require.Len(t, containerList, 0)
+ require.NoError(t, cli.Close())
 }
 
 func TestGnomock_withLogWriter(t *testing.T) {

go.mod

@@ -26,22 +26,20 @@ require (
  github.com/moby/term v0.0.0-20210619224110-3f7ff695adc6 // indirect
  github.com/morikuni/aec v1.0.0 // indirect
  github.com/pkg/errors v0.9.1 // indirect
+ github.com/rabbitmq/amqp091-go v1.9.0
  github.com/segmentio/kafka-go v0.4.47
  github.com/stretchr/testify v1.9.0
  go.mongodb.org/mongo-driver v1.15.0
+ go.uber.org/goleak v1.3.0
  go.uber.org/multierr v1.10.0 // indirect
  go.uber.org/zap v1.26.0
+ golang.org/x/mod v0.14.0
  golang.org/x/sync v0.6.0
  k8s.io/api v0.29.1
  k8s.io/apimachinery v0.29.1
  k8s.io/client-go v0.29.1
 )
 
-require (
- github.com/rabbitmq/amqp091-go v1.9.0
- golang.org/x/mod v0.14.0
-)
-
 require (
  github.com/Azure/azure-sdk-for-go/sdk/azcore v1.9.1 // indirect
  github.com/Azure/azure-sdk-for-go/sdk/internal v1.5.1 // indirect

go.sum

@@ -323,8 +323,9 @@ go.opentelemetry.io/otel/trace v1.25.0 h1:tqukZGLwQYRIFtSQM2u2+yfMVTgGVeqRLPUYx1
 go.opentelemetry.io/otel/trace v1.25.0/go.mod h1:hCCs70XM/ljO+BeQkyFnbK28SBIJ/Emuha+ccrCRT7I=
 go.opentelemetry.io/proto/otlp v1.1.0 h1:2Di21piLrCqJ3U3eXGCTPHE9R8Nh+0uglSnOyxikMeI=
 go.opentelemetry.io/proto/otlp v1.1.0/go.mod h1:GpBHCBWiqvVLDqmHZsoMM3C5ySeKTC7ej/RNTae6MdY=
-go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
 go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.10.0 h1:S0h4aNzvfcFsC3dRF1jLoaov7oRaKqRGC/pUEJ2yvPQ=
 go.uber.org/multierr v1.10.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo=

internal/cleaner/cleaner_test.go

@@ -11,8 +11,14 @@ import (
  "github.com/orlangure/gnomock/internal/health"
  "github.com/orlangure/gnomock/internal/testutil"
  "github.com/stretchr/testify/require"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestCleaner(t *testing.T) {
  t.Parallel()
 
@@ -53,6 +59,8 @@ func TestCleaner(t *testing.T) {
  containerList, err = testutil.ListContainerByID(cli, cleanerContainer.ID)
  require.NoError(t, err)
  require.Len(t, containerList, 0)
+
+ require.NoError(t, cli.Close())
 }
 
 func TestCleaner_wrongRequest(t *testing.T) {

internal/errors/errors_test.go

@@ -8,8 +8,14 @@ import (
  "github.com/orlangure/gnomock"
  "github.com/orlangure/gnomock/internal/errors"
  "github.com/stretchr/testify/require"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestPresetNotFoundError(t *testing.T) {
  err := errors.NewPresetNotFoundError("invalid")
  require.Equal(t, "preset 'invalid' not found", err.Error())

internal/health/health_test.go

@@ -8,8 +8,14 @@ import (
 
  "github.com/orlangure/gnomock/internal/health"
  "github.com/stretchr/testify/require"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestHTTPGet(t *testing.T) {
  ctx := context.Background()
 

internal/registry/registry_test.go

@@ -6,8 +6,13 @@ import (
  "github.com/orlangure/gnomock"
  "github.com/orlangure/gnomock/internal/registry"
  "github.com/stretchr/testify/require"
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 var p gnomock.Preset
 
 func TestRegistry(t *testing.T) {

preset/azurite/preset_test.go

@@ -8,10 +8,15 @@ import (
  "github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
  "github.com/orlangure/gnomock/preset/azurite"
  "github.com/stretchr/testify/require"
+ "go.uber.org/goleak"
 
  "github.com/orlangure/gnomock"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestPreset_Blobstorage(t *testing.T) {
  t.Parallel()
 

preset/cassandra/preset_test.go

@@ -7,8 +7,14 @@ import (
  "github.com/orlangure/gnomock"
  "github.com/orlangure/gnomock/preset/cassandra"
  "github.com/stretchr/testify/require"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestPreset(t *testing.T) {
  t.Parallel()
 

preset/cockroachdb/preset.go

@@ -87,6 +87,10 @@ func (p *P) setDefaults() {
 func healthcheck(_ context.Context, c *gnomock.Container) error {
  db, err := connect(c, "")
  if err != nil {
+ if db != nil {
+ _ = db.Close()
+ }
+
  return err
  }
 
@@ -108,6 +112,7 @@ func (p *P) initf() gnomock.InitFunc {
 
  _, err = db.Exec("create database " + p.DB)
  if err != nil {
+ _ = db.Close()
  return err
  }
 

preset/cockroachdb/preset_test.go

@@ -8,8 +8,14 @@ import (
  "github.com/orlangure/gnomock"
  "github.com/orlangure/gnomock/preset/cockroachdb"
  "github.com/stretchr/testify/require"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestPreset(t *testing.T) {
  t.Parallel()
 
@@ -54,6 +60,7 @@ func testPreset(version string) func(t *testing.T) {
  require.Equal(t, float64(2), avg)
  require.Equal(t, float64(1), min)
  require.Equal(t, float64(3), count)
+ require.NoError(t, db.Close())
  }
 }
 

preset/elastic/preset_test.go

@@ -9,8 +9,14 @@ import (
  "github.com/orlangure/gnomock"
  "github.com/orlangure/gnomock/preset/elastic"
  "github.com/stretchr/testify/require"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 // these tests have trouble running in parallel, probably due to limited
 // resources
 

preset/influxdb/preset_test.go

@@ -10,8 +10,14 @@ import (
  "github.com/orlangure/gnomock"
  "github.com/orlangure/gnomock/preset/influxdb"
  "github.com/stretchr/testify/require"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestPreset(t *testing.T) {
  for _, version := range []string{"alpine"} {
  t.Run("version", func(t *testing.T) {
@@ -96,5 +102,6 @@ func testPreset(version string, useDefaults bool) func(t *testing.T) {
  }
 
  require.Contains(t, orgNames, org)
+ client.Close()
  }
 }

preset/k3s/preset_test.go

@@ -11,8 +11,14 @@ import (
  v1 "k8s.io/api/core/v1"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  "k8s.io/client-go/kubernetes"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestPreset(t *testing.T) {
  t.Parallel()
 

preset/kafka/preset.go

@@ -146,7 +146,8 @@ func (p *P) healthcheck(ctx context.Context, c *gnomock.Container) (err error) {
  if err != nil {
  return fmt.Errorf("can't create consumer group: %w", err)
  }
- defer group.Close()
+
+ defer func() { _ = group.Close() }()
 
  if _, err := group.Next(ctx); err != nil {
  return fmt.Errorf("can't read next consumer group: %w", err)

preset/kafka/preset_test.go

@@ -10,8 +10,14 @@ import (
  "github.com/orlangure/gnomock/preset/kafka"
  kafkaclient "github.com/segmentio/kafka-go"
  "github.com/stretchr/testify/require"
+
+ "go.uber.org/goleak"
 )
 
+func TestMain(m *testing.M) {
+ goleak.VerifyTestMain(m)
+}
+
 func TestPreset(t *testing.T) {
  versions := []string{"3.3.1-L0", "3.6.1-L0"}