cdc/server/integration_test.go: added in a new test for the CDC serve…

…r using the workload.Checker

Integrated the logic from the existing CDC server integration test and the
latest developments for the pglogical integration tests to create a test based
on the workload.Checker that validates the CDC server. There are variants of
this test that support various changefeed configuration options such as
key_in_value == diff == queries == true or setting webhook == false, among others.
This gives confidence that various ways that customers define changefeeds will
work with the replicator CDC server.

There are now helper methods that help to create changefeeds for a variety
of configurations, along with relevant tests that verify the create changefeed
string is as expected. There are also new helpers added to help gate when the
workload checker runs since there is a minimum version requirement for its
operation. These shared utilities now live within the sinktest package.

Resolves: #906
Release Note: None

internal/sinktest/changefeed.go

@@ -0,0 +1,118 @@
+// Copyright 2024 The Cockroach Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package sinktest
+
+import (
+	"fmt"
+	"net/url"
+	"strings"
+
+	"github.com/cockroachdb/replicator/internal/util/ident"
+	"github.com/cockroachdb/replicator/internal/util/stdpool"
+)
+
+// ChangefeedStatement is a struct that represents a changefeed statement and it
+// allows the caller to specify various configuration options and parameters
+// useful for creating changefeeds on CRDB sources.
+type ChangefeedStatement struct {
+	Diff                   bool
+	Host                   string
+	KeyInValue             bool
+	QueryProjectionColumns []ident.Ident
+	SourceVersion          string
+	Tables                 []ident.Table
+	Target                 ident.Table
+	Token                  string
+	Webhook                bool
+}
+
+// String returns a string representation of the changefeed statement.
+func (cfs *ChangefeedStatement) String() string {
+	params := make(url.Values)
+	var feedURL url.URL
+	var pathIdent ident.Identifier
+	createStmt := "CREATE CHANGEFEED"
+
+	if len(cfs.QueryProjectionColumns) > 0 {
+		pathIdent = cfs.Target
+	} else {
+		// Creating the comma-separated table string required by the changefeed.
+		tablesStr := ""
+		for i, table := range cfs.Tables {
+			if i > 0 {
+				tablesStr += ", "
+			}
+			tablesStr += table.String()
+		}
+		pathIdent = cfs.Target.Schema()
+		createStmt += fmt.Sprintf(" FOR TABLE %s", tablesStr)
+	}
+
+	if cfs.Webhook {
+		params.Set("insecure_tls_skip_verify", "true")
+		feedURL = url.URL{
+			Scheme:   "webhook-https",
+			Host:     cfs.Host,
+			Path:     ident.Join(pathIdent, ident.Raw, '/'),
+			RawQuery: params.Encode(),
+		}
+		createStmt += " INTO '" + feedURL.String() + "' " +
+			" WITH updated," +
+			"     resolved='1s'," +
+			"     webhook_auth_header='Bearer " + cfs.Token + "'"
+	} else {
+		params.Set("access_token", cfs.Token)
+		feedURL = url.URL{
+			Scheme:   "experimental-http",
+			Host:     cfs.Host,
+			Path:     ident.Join(pathIdent, ident.Raw, '/'),
+			RawQuery: params.Encode(),
+		}
+		createStmt += " INTO '" + feedURL.String() + "' " +
+			"WITH updated, resolved='1s'"
+	}
+
+	if cfs.Diff {
+		createStmt += ", diff"
+	}
+
+	if cfs.KeyInValue {
+		createStmt += ", key_in_value"
+	}
+
+	if ok, err := SupportsMinCheckpointFrequency(cfs.SourceVersion); err == nil && ok {
+		createStmt += ", min_checkpoint_frequency='1s'"
+	}
+
+	if len(cfs.QueryProjectionColumns) > 0 {
+		createStmt += ",envelope='wrapped',format='json'"
+		var columns []string
+		for _, col := range cfs.QueryProjectionColumns {
+			columns = append(columns, col.String())
+		}
+		createStmt += " AS SELECT " + strings.Join(columns, ", ")
+		createStmt += fmt.Sprintf(" FROM %s", cfs.Tables[0].String())
+	}
+
+	return createStmt
+}
+
+// SupportsMinCheckpointFrequency checks if a certain version of CRDB supports the
+// min_checkpoint_frequency option.
+func SupportsMinCheckpointFrequency(version string) (bool, error) {
+	return stdpool.CockroachMinVersion(version, "v22.1")
+}

internal/sinktest/changefeed_test.go

@@ -0,0 +1,121 @@
+// Copyright 2024 The Cockroach Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package sinktest
+
+import (
+	"testing"
+
+	"github.com/cockroachdb/replicator/internal/util/ident"
+	"github.com/stretchr/testify/require"
+)
+
+// Note that the `sourceVersion` must contain the platform in the semver regex
+// (i.e. "CockroachDB CCL v24.2.1 (platform)").
+func TestCreateChangefeedStatement(t *testing.T) {
+	tests := []struct {
+		name string
+		stmt ChangefeedStatement
+		want string
+	}{
+		{
+			name: "basic no changefeed configs",
+			stmt: ChangefeedStatement{
+				Host:          "localHost:8080",
+				SourceVersion: "CockroachDB CCL v24.2.1 (platform)",
+				Target: ident.NewTable(ident.MustSchema(ident.New("target"),
+					ident.New("public")), ident.New("tbl1")),
+				Token: "my_token",
+				Tables: []ident.Table{ident.NewTable(ident.MustSchema(ident.New("source"),
+					ident.New("public")), ident.New("tbl1"))},
+			},
+			want: `CREATE CHANGEFEED FOR TABLE "source"."public"."tbl1" INTO ` +
+				`'experimental-http://localHost:8080/target/public?access_token=my_token' ` +
+				`WITH updated, resolved='1s', min_checkpoint_frequency='1s'`,
+		},
+		{
+			name: "basic webhook",
+			stmt: ChangefeedStatement{
+				Host:          "localHost:8080",
+				SourceVersion: "CockroachDB CCL v24.2.1 (platform)",
+				Target: ident.NewTable(ident.MustSchema(ident.New("target"), ident.New("public")),
+					ident.New("tbl1")),
+				Token: "my_token",
+				Tables: []ident.Table{
+					ident.NewTable(ident.MustSchema(ident.New("source"), ident.New("public")),
+						ident.New("tbl1")),
+					ident.NewTable(ident.MustSchema(ident.New("source"), ident.New("public")),
+						ident.New("tbl2")),
+				},
+				Webhook: true,
+			},
+			want: `CREATE CHANGEFEED FOR TABLE "source"."public"."tbl1", "source"."public"."tbl2" ` +
+				`INTO 'webhook-https://localHost:8080/target/public?insecure_tls_skip_verify=true'  ` +
+				`WITH updated,     resolved='1s',     webhook_auth_header='Bearer my_token', ` +
+				`min_checkpoint_frequency='1s'`,
+		},
+		{
+			name: "webhook and diff and key_in_value enabled",
+			stmt: ChangefeedStatement{
+				Diff:          true,
+				Host:          "localHost:8080",
+				KeyInValue:    true,
+				SourceVersion: "CockroachDB CCL v24.2.1 (platform)",
+				Target: ident.NewTable(ident.MustSchema(ident.New("target"), ident.New("public")),
+					ident.New("tbl1")),
+				Token: "my_token",
+				Tables: []ident.Table{
+					ident.NewTable(ident.MustSchema(ident.New("source"), ident.New("public")),
+						ident.New("tbl1")),
+					ident.NewTable(ident.MustSchema(ident.New("source"), ident.New("public")),
+						ident.New("tbl2")),
+				},
+				Webhook: true,
+			},
+			want: `CREATE CHANGEFEED FOR TABLE "source"."public"."tbl1", "source"."public"."tbl2" ` +
+				`INTO 'webhook-https://localHost:8080/target/public?insecure_tls_skip_verify=true'  ` +
+				`WITH updated,     resolved='1s',     webhook_auth_header='Bearer my_token', diff, key_in_value, ` +
+				`min_checkpoint_frequency='1s'`,
+		},
+		{
+			name: "basic webhook CDC queries",
+			stmt: ChangefeedStatement{
+				Host:                   "localHost:8080",
+				QueryProjectionColumns: []ident.Ident{ident.New("pk"), ident.New("val")},
+				SourceVersion:          "CockroachDB CCL v24.2.1 (platform)",
+				Target: ident.NewTable(ident.MustSchema(ident.New("target"), ident.New("public")),
+					ident.New("tbl1")),
+				Token: "my_token",
+				Tables: []ident.Table{
+					ident.NewTable(ident.MustSchema(ident.New("source"), ident.New("public")),
+						ident.New("tbl1")),
+				},
+				Webhook: true,
+			},
+			want: `CREATE CHANGEFEED INTO ` +
+				`'webhook-https://localHost:8080/target/public/tbl1?insecure_tls_skip_verify=true'  ` +
+				`WITH updated,     resolved='1s',     webhook_auth_header='Bearer my_token', ` +
+				`min_checkpoint_frequency='1s',envelope='wrapped',format='json' AS SELECT "pk", "val" ` +
+				`FROM "source"."public"."tbl1"`,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.stmt.String()
+			require.Equal(t, tt.want, got)
+		})
+	}
+}