chore: unexpected EOF errors with postgres container (#319)

testhelper/docker/resource/postgres/config.go

@@ -20,8 +20,22 @@ func WithShmSize(shmSize int64) Opt {
 	}
 }
 
+func WithMemory(memory int64) Opt {
+	return func(c *Config) {
+		c.Memory = memory
+	}
+}
+
+func WithOOMKillDisable(disable bool) Opt {
+	return func(c *Config) {
+		c.OOMKillDisable = disable
+	}
+}
+
 type Config struct {
-	Tag     string
-	Options []string
-	ShmSize int64
+	Tag            string
+	Options        []string
+	ShmSize        int64
+	Memory         int64
+	OOMKillDisable bool
 }

testhelper/docker/resource/postgres/postgres.go

@@ -1,12 +1,14 @@
 package postgres
 
 import (
+	"bytes"
 	"database/sql"
 	_ "encoding/json"
 	"fmt"
 
 	_ "github.com/lib/pq"
 	"github.com/ory/dockertest/v3"
+	"github.com/ory/dockertest/v3/docker"
 	dc "github.com/ory/dockertest/v3/docker"
 
 	"github.com/rudderlabs/rudder-go-kit/bytesize"
@@ -27,6 +29,9 @@ type Resource struct {
 	User     string
 	Host     string
 	Port     string
+
+	ContainerName string
+	ContainerID   string
 }
 
 func Setup(pool *dockertest.Pool, d resource.Cleaner, opts ...func(*Config)) (*Resource, error) {
@@ -54,12 +59,30 @@ func Setup(pool *dockertest.Pool, d resource.Cleaner, opts ...func(*Config)) (*R
 		Cmd: cmd,
 	}, func(hc *dc.HostConfig) {
 		hc.ShmSize = c.ShmSize
+		hc.OOMKillDisable = c.OOMKillDisable
+		hc.Memory = c.Memory
 	})
 	if err != nil {
 		return nil, err
 	}
 
 	d.Cleanup(func() {
+		if d.Failed() {
+			if c, found := pool.ContainerByName(postgresContainer.Container.Name); found {
+				d.Log(fmt.Sprintf("%q postgres container state: %+v", c.Container.Name, c.Container.State))
+				b := bytes.NewBufferString("")
+				if err := pool.Client.Logs(docker.LogsOptions{
+					Container:    c.Container.ID,
+					Stdout:       true,
+					Stderr:       true,
+					OutputStream: b,
+					ErrorStream:  b,
+				}); err != nil {
+					_, _ = b.Write([]byte(fmt.Sprintf("could not get logs: %s", err)))
+				}
+				d.Log(fmt.Sprintf("%q postgres container logs:\n%s", c.Container.Name, b.String()))
+			}
+		}
 		if err := pool.Purge(postgresContainer); err != nil {
 			d.Log("Could not purge resource:", err)
 		}
@@ -72,21 +95,50 @@ func Setup(pool *dockertest.Pool, d resource.Cleaner, opts ...func(*Config)) (*R
 	var db *sql.DB
 	// exponential backoff-retry, because the application in the container might not be ready to accept connections yet
 	err = pool.Retry(func() (err error) {
-		if db, err = sql.Open("postgres", dbDSN); err != nil {
+		// 1. use pg_isready
+		var w bytes.Buffer
+		code, err := postgresContainer.Exec([]string{
+			"bash",
+			"-c",
+			fmt.Sprintf("pg_isready -d %[1]s -U %[2]s", postgresDefaultDB, postgresDefaultUser),
+		}, dockertest.ExecOptions{StdOut: &w, StdErr: &w})
+		if err != nil {
 			return err
 		}
-		return db.Ping()
+		if code != 0 {
+			return fmt.Errorf("postgres not ready:\n%s" + w.String())
+		}
+
+		// 2. create a sql.DB and verify connection
+		if db, err = sql.Open("postgres", dbDSN); err != nil {
+			return fmt.Errorf("opening database: %w", err)
+		}
+		defer func() {
+			if err != nil {
+				_ = db.Close()
+			}
+		}()
+		if err = db.Ping(); err != nil {
+			return fmt.Errorf("pinging database: %w", err)
+		}
+		var one int
+		if err = db.QueryRow("SELECT 1").Scan(&one); err != nil {
+			return fmt.Errorf("querying database: %w", err)
+		}
+		return nil
 	})
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("waiting for database to startup: %w", err)
 	}
 	return &Resource{
-		DB:       db,
-		DBDsn:    dbDSN,
-		Database: postgresDefaultDB,
-		User:     postgresDefaultUser,
-		Password: postgresDefaultPassword,
-		Host:     "localhost",
-		Port:     postgresContainer.GetPort("5432/tcp"),
+		DB:            db,
+		DBDsn:         dbDSN,
+		Database:      postgresDefaultDB,
+		User:          postgresDefaultUser,
+		Password:      postgresDefaultPassword,
+		Host:          "localhost",
+		Port:          postgresContainer.GetPort("5432/tcp"),
+		ContainerName: postgresContainer.Container.Name,
+		ContainerID:   postgresContainer.Container.ID,
 	}, nil
 }

testhelper/docker/resource/postgres/postgres_test.go

@@ -0,0 +1,64 @@
+package postgres_test
+
+import (
+	"database/sql"
+	"fmt"
+	"testing"
+
+	"github.com/ory/dockertest/v3"
+	"github.com/stretchr/testify/require"
+
+	"github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/postgres"
+)
+
+func TestPostgres(t *testing.T) {
+	pool, err := dockertest.NewPool("")
+	require.NoError(t, err)
+
+	for i := 1; i <= 6; i++ {
+		t.Run(fmt.Sprintf("iteration %d", i), func(t *testing.T) {
+			postgresContainer, err := postgres.Setup(pool, t)
+			require.NoError(t, err)
+			defer func() { _ = postgresContainer.DB.Close() }()
+
+			db, err := sql.Open("postgres", postgresContainer.DBDsn)
+			require.NoError(t, err)
+			_, err = db.Exec("CREATE TABLE test (id int)")
+			require.NoError(t, err)
+
+			var count int
+			err = db.QueryRow("SELECT count(*) FROM test").Scan(&count)
+			require.NoError(t, err)
+		})
+	}
+
+	t.Run("with test failure", func(t *testing.T) {
+		cl := &testCleaner{T: t, failed: true}
+		r, err := postgres.Setup(pool, cl)
+		require.NoError(t, err)
+		err = pool.Client.StopContainer(r.ContainerID, 10)
+		require.NoError(t, err)
+		cl.cleanup()
+		require.Contains(t, cl.logs, "postgres container state: {Status:exited")
+		require.Contains(t, cl.logs, "postgres container logs:")
+	})
+}
+
+type testCleaner struct {
+	*testing.T
+	cleanup func()
+	failed  bool
+	logs    string
+}
+
+func (t *testCleaner) Cleanup(f func()) {
+	t.cleanup = f
+}
+
+func (t *testCleaner) Failed() bool {
+	return t.failed
+}
+
+func (t *testCleaner) Log(args ...any) {
+	t.logs = t.logs + fmt.Sprint(args...)
+}

testhelper/docker/resource/types.go

@@ -4,9 +4,14 @@ type Logger interface {
 	Log(...interface{})
 }
 
+type FailIndicator interface {
+	Failed() bool
+}
+
 type Cleaner interface {
 	Cleanup(func())
 	Logger
+	FailIndicator
 }
 
 type NOPLogger struct{}

throttling/util_test.go

@@ -15,6 +15,7 @@ type tester interface {
 	Log(...interface{})
 	Errorf(format string, args ...interface{})
 	Fatalf(format string, args ...any)
+	Failed() bool
 	FailNow()
 	Cleanup(f func())
 }