test: add integration test for dbmode strategy

go.mod

@@ -48,6 +48,7 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.9.0
 	github.com/testcontainers/testcontainers-go v0.33.0
+	github.com/testcontainers/testcontainers-go/modules/postgres v0.33.0
 	github.com/tonglil/buflogr v1.1.1
 	go.uber.org/zap v1.27.0
 	google.golang.org/api v0.193.0

go.sum

@@ -224,6 +224,14 @@ github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4=
 github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
+github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.5.4 h1:Xp2aQS8uXButQdnCMWNmvx6UysWQQC+u1EoizjguY+8=
+github.com/jackc/pgx/v5 v5.5.4/go.mod h1:ez9gk+OAat140fv9ErkZDYFWmXLfV+++K0uAOiwgm1A=
+github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
+github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/jpillora/backoff v1.0.0 h1:uvFg412JmmHBHw7iwprIxkPMI+sGQ4kzOWsMeHnm2EA=
@@ -254,6 +262,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de h1:9TO3cAIGXtEhnIaL+V+BEER86oLrvS+kWobKpbJuye0=
 github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de/go.mod h1:zAbeS9B/r2mtpb6U+EI2rYA5OAXxsYw6wTamcNW+zcE=
 github.com/lithammer/dedent v1.1.0 h1:VNzHMVCBNG1j0fh3OrsFRkVUwStdDArbgBWoPAffktY=
@@ -402,6 +412,8 @@ github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsT
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/testcontainers/testcontainers-go v0.33.0 h1:zJS9PfXYT5O0ZFXM2xxXfk4J5UMw/kRiISng037Gxdw=
 github.com/testcontainers/testcontainers-go v0.33.0/go.mod h1:W80YpTa8D5C3Yy16icheD01UTDu+LmXIA2Keo+jWtT8=
+github.com/testcontainers/testcontainers-go/modules/postgres v0.33.0 h1:c+Gt+XLJjqFAejgX4hSpnHIpC9eAhvgI/TFWL/PbrFI=
+github.com/testcontainers/testcontainers-go/modules/postgres v0.33.0/go.mod h1:I4DazHBoWDyf69ByOIyt3OdNjefiUx372459txOpQ3o=
 github.com/tidwall/gjson v1.17.3 h1:bwWLZU7icoKRG+C+0PNwIKC6FCJO/Q3p2pZvuP0jN94=
 github.com/tidwall/gjson v1.17.3/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=

internal/dataplane/sendconfig/dbmode.go

@@ -40,8 +40,8 @@ func NewUpdateStrategyDBMode(
 	version semver.Version,
 	concurrency int,
 	logger logr.Logger,
-) UpdateStrategyDBMode {
-	return UpdateStrategyDBMode{
+) *UpdateStrategyDBMode {
+	return &UpdateStrategyDBMode{
 		client:            client,
 		dumpConfig:        dumpConfig,
 		version:           version,
@@ -58,13 +58,13 @@ func NewUpdateStrategyDBModeKonnect(
 	version semver.Version,
 	concurrency int,
 	logger logr.Logger,
-) UpdateStrategyDBMode {
+) *UpdateStrategyDBMode {
 	s := NewUpdateStrategyDBMode(client, dumpConfig, version, concurrency, logger)
 	s.isKonnect = true
 	return s
 }
 
-func (s UpdateStrategyDBMode) Update(ctx context.Context, targetContent ContentWithHash) error {
+func (s *UpdateStrategyDBMode) Update(ctx context.Context, targetContent ContentWithHash) error {
 	cs, err := s.currentState(ctx)
 	if err != nil {
 		return fmt.Errorf("failed getting current state for %s: %w", s.client.BaseRootURL(), err)
@@ -89,7 +89,7 @@ func (s UpdateStrategyDBMode) Update(ctx context.Context, targetContent ContentW
 	}
 
 	ctx, cancel := context.WithCancel(ctx)
-	go s.HandleEvents(ctx, syncer.GetResultChan())
+	go s.handleEvents(ctx, syncer.GetResultChan())
 
 	_, errs, _ := syncer.Solve(ctx, s.concurrency, false, false)
 	cancel()
@@ -115,9 +115,9 @@ func (s UpdateStrategyDBMode) Update(ctx context.Context, targetContent ContentW
 	return nil
 }
 
-// HandleEvents handles logging and error reporting for individual entity change events generated during a sync by
+// handleEvents handles logging and error reporting for individual entity change events generated during a sync by
 // looping over an event channel. It terminates when its context dies.
-func (s *UpdateStrategyDBMode) HandleEvents(ctx context.Context, events chan diff.EntityAction) {
+func (s *UpdateStrategyDBMode) handleEvents(ctx context.Context, events chan diff.EntityAction) {
 	s.resourceErrorLock.Lock()
 	for {
 		select {
@@ -187,22 +187,22 @@ func resourceErrorFromEntityAction(event diff.EntityAction) (ResourceError, erro
 		// has "methods", but we'd need to do string parsing to extract it, and we may not catch all possible error types.
 		// This lazier approach just dumps the full error string as a single problem, which is probably good enough.
 		Problems: map[string]string{
-			"": fmt.Sprintf("%s", event.Error),
+			fmt.Sprintf("%s:%s", event.Entity.Kind, event.Entity.Name): fmt.Sprintf("%s", event.Error),
 		},
 	}
 
 	return parseRawResourceError(raw)
 }
 
-func (s UpdateStrategyDBMode) MetricsProtocol() metrics.Protocol {
+func (s *UpdateStrategyDBMode) MetricsProtocol() metrics.Protocol {
 	return metrics.ProtocolDeck
 }
 
-func (s UpdateStrategyDBMode) Type() string {
+func (s *UpdateStrategyDBMode) Type() string {
 	return "DBMode"
 }
 
-func (s UpdateStrategyDBMode) currentState(ctx context.Context) (*state.KongState, error) {
+func (s *UpdateStrategyDBMode) currentState(ctx context.Context) (*state.KongState, error) {
 	rawState, err := dump.Get(ctx, s.client, s.dumpConfig)
 	if err != nil {
 		return nil, fmt.Errorf("loading configuration from kong: %w", err)
@@ -211,7 +211,7 @@ func (s UpdateStrategyDBMode) currentState(ctx context.Context) (*state.KongStat
 	return state.Get(rawState)
 }
 
-func (s UpdateStrategyDBMode) targetState(
+func (s *UpdateStrategyDBMode) targetState(
 	ctx context.Context,
 	currentState *state.KongState,
 	targetContent *file.Content,

internal/dataplane/sendconfig/error_handling.go

@@ -0,0 +1,44 @@
+package sendconfig
+
+import (
+	"fmt"
+
+	"github.com/go-logr/logr"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8stypes "k8s.io/apimachinery/pkg/types"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/logging"
+)
+
+// resourceErrorsToResourceFailures translates a slice of ResourceError to a slice of failures.ResourceFailure.
+func resourceErrorsToResourceFailures(resourceErrors []ResourceError, logger logr.Logger) []failures.ResourceFailure {
+	var out []failures.ResourceFailure
+	for _, ee := range resourceErrors {
+		obj := metav1.PartialObjectMetadata{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       ee.Kind,
+				APIVersion: ee.APIVersion,
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: ee.Namespace,
+				Name:      ee.Name,
+				UID:       k8stypes.UID(ee.UID),
+			},
+		}
+		for problemSource, problem := range ee.Problems {
+			logger.V(logging.DebugLevel).Info("Adding failure", "resource_name", ee.Name, "source", problemSource, "problem", problem)
+			resourceFailure, failureCreateErr := failures.NewResourceFailure(
+				fmt.Sprintf("invalid %s: %s", problemSource, problem),
+				&obj,
+			)
+			if failureCreateErr != nil {
+				logger.Error(failureCreateErr, "Could not create resource failure event")
+			} else {
+				out = append(out, resourceFailure)
+			}
+		}
+	}
+
+	return out
+}

internal/dataplane/sendconfig/inmemory_error_handling.go

@@ -6,12 +6,8 @@ import (
 	"strings"
 
 	"github.com/go-logr/logr"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	k8stypes "k8s.io/apimachinery/pkg/types"
 
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/logging"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
 	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
 	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
@@ -206,35 +202,3 @@ func gvkIsClusterScoped(gvk schema.GroupVersionKind) bool {
 	}
 	return false
 }
-
-// resourceErrorsToResourceFailures translates a slice of ResourceError to a slice of failures.ResourceFailure.
-func resourceErrorsToResourceFailures(resourceErrors []ResourceError, logger logr.Logger) []failures.ResourceFailure {
-	var out []failures.ResourceFailure
-	for _, ee := range resourceErrors {
-		obj := metav1.PartialObjectMetadata{
-			TypeMeta: metav1.TypeMeta{
-				Kind:       ee.Kind,
-				APIVersion: ee.APIVersion,
-			},
-			ObjectMeta: metav1.ObjectMeta{
-				Namespace: ee.Namespace,
-				Name:      ee.Name,
-				UID:       k8stypes.UID(ee.UID),
-			},
-		}
-		for problemSource, problem := range ee.Problems {
-			logger.V(logging.DebugLevel).Info("Adding failure", "resource_name", ee.Name, "source", problemSource, "problem", problem)
-			resourceFailure, failureCreateErr := failures.NewResourceFailure(
-				fmt.Sprintf("invalid %s: %s", problemSource, problem),
-				&obj,
-			)
-			if failureCreateErr != nil {
-				logger.Error(failureCreateErr, "Could not create resource failure event")
-			} else {
-				out = append(out, resourceFailure)
-			}
-		}
-	}
-
-	return out
-}

test/envtest/configerrorevent_envtest_test.go

@@ -315,7 +315,7 @@ func TestConfigErrorEventGenerationDBMode(t *testing.T) {
 			return e.Reason == dataplane.KongConfigurationApplyFailedEventReason &&
 				e.InvolvedObject.Kind == "KongConsumer" &&
 				e.InvolvedObject.Name == consumer.Name &&
-				e.Message == "invalid : HTTP status 400 (message: \"2 schema violations (at least one of these fields must be non-empty: 'custom_id', 'username'; fake: unknown field)\")"
+				e.Message == fmt.Sprintf("invalid consumer:%s: HTTP status 400 (message: \"2 schema violations (at least one of these fields must be non-empty: 'custom_id', 'username'; fake: unknown field)\")", consumer.Name)
 		})
 		if lo.Count(matches, true) != 1 {
 			t.Logf("not all events matched: %+v", matches)

test/kongintegration/containers/kong.go

@@ -35,6 +35,18 @@ func KongWithRouterFlavor(flavor string) KongOpt {
 	}
 }
 
+func KongWithDBMode(networkName string) KongOpt {
+	return func(req *testcontainers.ContainerRequest) {
+		req.Networks = []string{networkName}
+
+		req.Env["KONG_DATABASE"] = "postgres"
+		req.Env["KONG_PG_DATABASE"] = postgresDatabase
+		req.Env["KONG_PG_USER"] = postgresUser
+		req.Env["KONG_PG_PASSWORD"] = postgresPassword
+		req.Env["KONG_PG_HOST"] = postgresContainerNetworkAlias
+	}
+}
+
 // Kong represents a docker container running Kong.
 type Kong struct {
 	container testcontainers.Container

test/kongintegration/containers/kong_postgres.go

@@ -0,0 +1,131 @@
+package containers
+
+import (
+	"context"
+	"io"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/postgres"
+	"github.com/testcontainers/testcontainers-go/network"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+const (
+	// postgresImage is the default image used for the Postgres container.
+	postgresImage = "postgres:16-alpine"
+
+	// postgresUser is the default user for the Postgres container.
+	postgresUser = "postgres"
+
+	// postgresPassword is the default password for the Postgres container.
+	postgresPassword = "pass"
+
+	// postgresDatabase is the default database for the Postgres container.
+	postgresDatabase = "kong"
+
+	// postgres is the default port for the Postgres container.
+	postgresPort = 5432
+
+	// postgresContainerNetworkAlias is the hostname alias for the Postgres container.
+	postgresContainerNetworkAlias = "db"
+)
+
+type Postgres struct {
+	container testcontainers.Container
+}
+
+func NewPostgres(ctx context.Context, t *testing.T, net *testcontainers.DockerNetwork) *Postgres {
+	postgresC, err := postgres.Run(ctx,
+		postgresImage,
+		network.WithNetwork([]string{postgresContainerNetworkAlias}, net),
+		testcontainers.WithEnv(map[string]string{
+			"POSTGRES_USER":     postgresUser,
+			"POSTGRES_PASSWORD": postgresPassword,
+			"POSTGRES_DB":       postgresDatabase,
+		}),
+		testcontainers.WithWaitStrategy(
+			wait.ForLog("database system is ready to accept connections"),
+		),
+	)
+	require.NoError(t, err)
+	t.Logf("Postgres container ID: %s", postgresC.GetContainerID())
+
+	t.Cleanup(func() { //nolint:contextcheck
+		// If the container is already terminated, we don't need to terminate it again.
+		if postgresC.IsRunning() {
+			assert.NoError(t, postgresC.Terminate(context.Background()))
+		}
+	})
+
+	runKongDBMigrations(ctx, t, net.Name)
+
+	return &Postgres{
+		container: postgresC,
+	}
+}
+
+// runKongDBMigrations runs the Kong migrations bootstrap command in a container against a Postgres database container.
+func runKongDBMigrations(ctx context.Context, t *testing.T, networkName string) {
+	// Run Kong migrations bootstrap command in a container.
+	kongMigrationsC, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
+		ContainerRequest: testcontainers.ContainerRequest{
+			Image: kongImageUnderTest(),
+			Env: map[string]string{
+				"KONG_DATABASE":    "postgres",
+				"KONG_PG_HOST":     postgresContainerNetworkAlias,
+				"KONG_PG_PORT":     strconv.Itoa(postgresPort),
+				"KONG_PG_USER":     postgresUser,
+				"KONG_PG_PASSWORD": postgresPassword,
+				"KONG_PG_DATABASE": postgresDatabase,
+			},
+			Cmd: []string{
+				"kong", "migrations", "bootstrap",
+				"--yes", "--force",
+				"--db-timeout", "30",
+			},
+			Networks: []string{networkName},
+		},
+		Started: true,
+	})
+	require.NoError(t, err)
+	t.Logf("Kong migrations container ID: %s", kongMigrationsC.GetContainerID())
+
+	// Wait for migrations to finish successfully (status == "exited" and exit code == 0).
+	const (
+		timeout = 30 * time.Second
+		period  = 1 * time.Second
+	)
+	timer := time.After(timeout)
+	ticker := time.NewTicker(period)
+	defer ticker.Stop()
+	for range ticker.C {
+		select {
+		case <-timer:
+			assert.Fail(t, "Kong migrations bootstrap timed out")
+			return
+		default:
+		}
+
+		state, err := kongMigrationsC.State(ctx)
+		require.NoError(t, err)
+		if state.Status == "exited" {
+			if !assert.Equal(t, 0, state.ExitCode, "Kong migrations bootstrap failed") {
+				logs, err := kongMigrationsC.Logs(ctx)
+				require.NoError(t, err)
+
+				logsB, err := io.ReadAll(logs)
+				require.NoError(t, err)
+
+				t.Logf("Kong migrations bootstrap logs: %s", string(logsB))
+			}
+			return
+		}
+
+		t.Logf("Waiting for Kong migrations to finish, current state: %s", state.Status)
+	}
+}