cdc-sink: Overhaul #73

bobvawter · 2021-12-13T14:08:28Z

This change overhauls the cdc-sink code to split it into well-defined packages
and APIs.

Notable functional changes:

A cdc-sink endpoint now operates on an entire database at once, since this is
the use-case that has been most prominent in discussions.

An "immediate" mode is supported, which applies incoming data without waiting
for resolved timestamps. This is intended for use when backfilling large
datasets or if a high-volume changefeed must catch up after an outage. It is
not expected to be the default configuration for cdc-sink.

The cdc-sink code can now detect and recover from a limited amount of
structural schema drift between the source and target databases. The schema for
each target database is held in memory and refreshed from time to time. Drift
is checked during resolved-timestamp flushes, which will effectively pause a
changefeed until the stored payloads are at least structurally-compatible with
the target tables.

Notes to reviewers:

The sinktypes package defines interfaces for the major moving parts of the
revised cdc-sink code base. At present, each interface is implemented by a
single type, but having small APIs has been useful in identifying the
independent parts of cdc-sink.

Similarly, the package structure may be overly fine-grained in favor of
identifying specific, small portions of code that compose easily.

A comment at the top of most files indicates where code was repackaged from.

Recommended package review order:
sinktypes, cdc/handler, backend/mutation, backend/apply, backend/timestamp

This change is

codecov-commenter · 2021-12-16T22:13:01Z

Codecov Report

❗ No coverage uploaded for pull request base (master@2458a1c). Click here to learn what that means.
The diff coverage is n/a.

@@            Coverage Diff            @@
##             master      #73   +/-   ##
=========================================
  Coverage          ?   77.14%           
=========================================
  Files             ?       25           
  Lines             ?     1260           
  Branches          ?        0           
=========================================
  Hits              ?      972           
  Misses            ?      211           
  Partials          ?       77

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 2458a1c...551f48a. Read the comment docs.

bobvawter · 2021-12-16T22:17:11Z

I have pushed some revisions based on the desk review yesterday.

cdc-sink is now based around a single changefeed per SCHEMA (i.e. the second-level namespace in each DATABASE). The url format for sending changes to cdc-sink now leads with /targetDB/targetSchema/ to specify the destination table(s).
The README file has been overhauled to reflect the new state of the code. I have revalidated the quickstart flow.
Added support for the new webhook-https:// scheme in CockroachDB v21.2. cdc-sink can optionally speak TLS.

sravotto

Reviewed 18 of 41 files at r1, 18 of 22 files at r3, 1 of 1 files at r5, 5 of 5 files at r6, 2 of 3 files at r7, 1 of 1 files at r8, 24 of 24 files at r9, all commit messages.
Reviewable status: all files reviewed, 2 unresolved discussions (waiting on @bobvawter and @BramGruneir)

internal/sinktypes/sinktypes.go, line 15 at r8 (raw file):

// the types into this package is to make it easy to compose
// functionality as the cdc-sink project evolves.
package sinktypes

Should we call this package simply "types" (or model/interfaces)?

internal/sinktypes/sinktypes.go, line 79 at r8 (raw file):

// A TimeSwapper maintains a durable map of string keys to timestamps.
type TimeSwapper interface {

Perhaps we can call it TimeKeeper? And the method simply Update?

internal/util/ident/ident.go, line 12 at r8 (raw file):

// Package ident contains types for safely representing SQL identifiers.
package ident

Like this approach. Makes the (rest of the) code much more robust and easier to read.

bobvawter

Reviewable status: 35 of 52 files reviewed, 2 unresolved discussions (waiting on @BramGruneir and @sravotto)

internal/sinktypes/sinktypes.go, line 15 at r8 (raw file):

Previously, sravotto (silvano) wrote…

Should we call this package simply "types" (or model/interfaces)?

Renamed it to "types".

internal/sinktypes/sinktypes.go, line 79 at r8 (raw file):

Previously, sravotto (silvano) wrote…

Perhaps we can call it TimeKeeper? And the method simply Update?

Switch it to TimeKeeper.Put() since it's map-like and one puts into a map.

BramGruneir · 2021-12-20T14:45:08Z

internal/frontend/cdc/resolved.go, line 128 at r10 (raw file):

		for table := range targetSchema {
			// Only perform work on target tables which match the
			// database and schema name specified in the resolved URL.

Do we throw an error if we get one that doesn't match? A table we were not expecting?

BramGruneir

Reviewed 15 of 41 files at r1, 5 of 22 files at r3, 1 of 5 files at r6, 1 of 3 files at r7, 1 of 1 files at r8, 12 of 24 files at r9, 17 of 17 files at r10, all commit messages.
Reviewable status: all files reviewed, 36 unresolved discussions (waiting on @bobvawter and @sravotto)

internal/backend/apply/apply.go, line 162 at r10 (raw file):

	defer res.Close()

	for i, j := 0, batch.Len(); i < j; i++ {

Is the assignment of j needed here?

internal/backend/apply/apply.go, line 194 at r10 (raw file):

			rawColName := col.Name.Raw()
			decoded, ok := temp[rawColName]
			delete(temp, rawColName)

These seems slightly out of order. I think they'll work, (delete something that doesn't exist,) but the logic doesn't flow cleanly.

internal/backend/apply/apply.go, line 252 at r10 (raw file):

	_, _ = fmt.Fprintf(&delete, "DELETE FROM %s WHERE (", a.target)
	_, _ = fmt.Fprintf(&upsert, "UPSERT INTO %s (", a.target)
	for i := range colData {

why not
for i, col :=

cleans up a lot of the colData[i] you're using all over all the place

internal/backend/apply/apply.go, line 270 at r10 (raw file):

	delete.WriteString(") IN (SELECT ")
	upsert.WriteString(") SELECT ")
	for i := range colData {

Same here

internal/backend/apply/factory.go, line 39 at r10 (raw file):

// Get returns a memoized instance of the Applier for the table.
func (f *factory) Get(

I'm a little confused by the api here.

Get and New are the only exposed functions?

Get --> getUnlocked
New --> returns a factory

But no Create?
? --> createUnlocked

internal/backend/mutation/factory.go, line 43 at r10 (raw file):

}

func (f *factory) createUnlocked(ctx context.Context, table ident.Table) (*store, error) {

Are createUnlocked and getUnlocked called from anywhere else? Do you need them to be their own functions? I'm guessing for unit tests?

internal/backend/mutation/store.go, line 50 at r10 (raw file):

// newStore constructs a new mutation store that will track pending
// mutations to be applied to the given target table.
func newStore(

Shouldn't this be New?

internal/backend/mutation/store.go, line 133 at r10 (raw file):

	batch := &pgx.Batch{}

	for i := range mutations {

Why not use for i, mut := range mutations instead of indexing in every time?

internal/backend/mutation/store_test.go, line 79 at r10 (raw file):

	// Dequeue.
	ret, err := s.Drain(ctx, dbInfo.Pool(),

style nit, I thought our usual style was to put only the ctx on the first line then the rest on the 2nd. Or to put everything on the 2nd line with the close bracket on the 3rd

Hmm... can we get the style guide from cockroach?

internal/backend/schemawatch/coldata.go, line 37 at r10 (raw file):

// Retrieve the primary key columns in their index-order, then append
// any remaining non-generated columns.
const sqlColumnsQuery = `

Wow... is the most efficient way? It makes sense but I'm just surprised you need this complex a query.

internal/backend/schemawatch/coldata_test.go, line 90 at r10 (raw file):

	// Virtual columns not supported before v21.1
	if !strings.Contains(dbInfo.Version(), "v20.2.") {

If you have a few more tests that need version numbers, it makes sense to create a utility to determine if the version is newer, older or equal. Can you just add a comment about how the older version supported is v20.2? Perhaps that info should go in the readme as well.

internal/backend/schemawatch/watcher.go, line 48 at r10 (raw file):

	mu struct {
		sync.RWMutex
		cond sync.Cond // Conditional on the RLocker

Should cond be under mu if a lock isn't required for its use?
I see it's a weird mixed thing... Just needs a read lock.

So I find this a little confusing. Simplifying the locking here might be worthwhile.

internal/backend/schemawatch/watcher.go, line 112 at r10 (raw file):

	w.mu.data = data
	w.mu.Unlock()
	w.mu.cond.Broadcast()

Why take this out of the lock?

internal/backend/schemawatch/watcher.go, line 153 at r10 (raw file):

			// Respond to context cancellation or dropping the table.
			if !ok || ctx.Err() != nil {
				return

Shouldn't this do something with the error?

internal/backend/schemawatch/watcher_test.go, line 41 at r10 (raw file):

	// Bootstrap column.
	tblInfo, err := sinktest.CreateTable(ctx, dbName, "CREATE TABLE %s (pk INT PRIMARY KEY)")

Do you delete this table on test cleanup?

internal/backend/schemawatch/watcher_test.go, line 59 at r10 (raw file):

	select {
	case <-time.After(10 * time.Second):

Consider setting these 10s much smaller, as this is unit test.

internal/backend/schemawatch/watchers.go, line 35 at r10 (raw file):

// NewWatchers creates a Watchers factory.
func NewWatchers(pool *pgxpool.Pool) (_ *Watchers, cancel func()) {

So in some factories you call this New, in other the name of the struct you're creating and in this case, NewStruct. Can you do a quick audit and make them all match in style?

internal/backend/sinktest/info.go, line 23 at r10 (raw file):

// DBInfo encapsulates metadata and a connection to a database.
type DBInfo struct {

perhaps name this file dbinfo to match?

internal/backend/sinktest/info.go, line 45 at r10 (raw file):

}

// DeleteAll deletes (not TRUNCATEs) all rows in the table.

Why no Create here? Just Drop?

internal/backend/sinktest/sinktest.go, line 102 at r10 (raw file):

	cancel = func() {
		err := retry.Execute(ctx, db, fmt.Sprintf("DROP DATABASE IF EXISTS %s CASCADE", name))

Ah, dropping happens here. Good, but perhaps this whole function should be part of dbinfo?

I'm having a bit an of an issue understanding the line between this file and info.go

internal/backend/timestamp/store.go, line 30 at r10 (raw file):

// store implements a simple key/value store for HLC timestamps.
type store struct {

Store is very generic. Should we call it timekeeper instead? The file and the type?

internal/backend/timestamp/store_test.go, line 43 at r10 (raw file):

	prev := hlc.Zero()
	for i := 0; i <= count; i++ {
		next := hlc.New(int64(1000*i), i)

Please add comment on exactly what this test is testing.

internal/frontend/cdc/handler_test.go, line 34 at r10 (raw file):

}

func testHandler(t *testing.T, immediate bool) {

Please add some comments within the tests.

internal/frontend/cdc/webhook.go, line 16 at r10 (raw file):

)

var (

So should we be using webhook or https? It it worth it to support both?

internal/frontend/cdc/webhook.go, line 70 at r10 (raw file):

	}

	// First, we'll aggregate the mutations by target table. We know

You have a first, but not a second or third.

internal/frontend/server/integration_test.go, line 49 at r10 (raw file):

	// Prefer the webhook format for current versions of CRDB.
	useWebhook := true
	if strings.Contains(dbInfo.Version(), "v20.2.") || strings.Contains(dbInfo.Version(), "v21.1.") {

Again, a nice version function for comparing would be better.

internal/frontend/server/integration_test.go, line 56 at r10 (raw file):

	defer func() { *GenerateSelfSigned = false }()

	srv, err := newServer(ctx, "127.0.0.1:0", dbInfo.Pool().Config().ConnString(), false)

Do we need to worry if that's already in use?

internal/frontend/server/integration_test.go, line 84 at r10 (raw file):

	if useWebhook {
		feedURL = url.URL{
			Scheme:   "webhook-https",

So webhook is beta, http is experimental. What is the difference? (This is not a question for you.)

internal/frontend/server/server.go, line 45 at r10 (raw file):

	ConnectionString = flag.String(
		"conn",
		"postgresql://root@localhost:26257/?sslmode=disable",

We're going to need to get this working with certs and a non-root user soon.

internal/frontend/server/server.go, line 49 at r10 (raw file):

	)

	IgnoreResolved = flag.Bool("ignoreResolved", false,

This shouldn't be a flag on all of cdc-sink, but instead on a per feed basis.

internal/frontend/server/server.go, line 136 at r10 (raw file):

	}

	go func() {

Go really needs a better closer. Or at least some syntactic sugar over this.

internal/frontend/server/tls_config.go, line 42 at r10 (raw file):

	// Loosely based on https://golang.org/src/crypto/tls/generate_cert.go
	if *GenerateSelfSigned {
		priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)

I'm just going to trust this works.

internal/util/batches/batches.go, line 29 at r10 (raw file):

// of values in a batch-oriented fashion. The indexes provided to
// the callback function are a half-open range [begin , end).
func Batch(count int, fn func(begin, end int) error) error {

This is really nice.

internal/util/batches/batches.go, line 29 at r10 (raw file):

// of values in a batch-oriented fashion. The indexes provided to
// the callback function are a half-open range [begin , end).
func Batch(count int, fn func(begin, end int) error) error {

Please add tests for batch.

internal/util/ident/ident.go, line 79 at r10 (raw file):

// Empty returns true if the identifier is empty.
func (n Ident) Empty() bool {

Shouldn't this be IsEmpty()

internal/util/ident/ident.go, line 85 at r10 (raw file):

// Raw returns the original, raw value.
func (n Ident) Raw() string {
	return strings.ReplaceAll(n.q[1:len(n.q)-1], `""`, `"`)

Why not just store the raw? Because this is clearly a lie :)

internal/util/ident/ident_test.go, line 49 at r10 (raw file):

		expectError bool
		qual        Qualification
	}{

Please add some more test cases that would produce errors. Empty schema, empty db, empty table, etc.

internal/util/retry/retry.go, line 11 at r10 (raw file):

// licenses/APL.txt.

// Package retry contains utility code for retrying database transactions.

Is there any way to not re-implement this ourselves?

BramGruneir

That's a lot of code, I went through it all.
Not many changes needed I think.

I still don't like the frontend/backend names.

How about sources/target?

Reviewable status: all files reviewed, 36 unresolved discussions (waiting on @bobvawter and @sravotto)

This change overhauls the cdc-sink code to split it into well-defined packages and APIs. Notable functional changes: A cdc-sink endpoint now operates on an entire database schema (i.e. the second-level namespace) at once, since this is the use-case that has been most prominent in discussions around our preferred microservice architecture. For users that only ever use the "public" schema in their databases, this will handle whole-database use cases. An "immediate" mode is supported, which applies incoming data without waiting for resolved timestamps. This is intended for use when backfilling large datasets or if a high-volume changefeed must catch up after an outage. It is not expected to be the default configuration for cdc-sink. The cdc-sink code can now detect and optionally recover from a limited amount of structural schema drift between the source and target databases. The schema for each target database is held in memory and refreshed from time to time. Drift is checked during resolved-timestamp flushes, which will effectively pause a changefeed until the stored payloads are at least structurally-compatible with the target tables. The new webhook-https:// scheme in CockroachDB v21.2 is now supported. This necessitates that cdc-sink can use a TLS-enabled HTTP server. The option now exists to load a certificate and private key from disk. For testing purposes, a self-signed certifcate can be internally generated by cdc-sink. Notes to reviewers: The internal/types package defines interfaces for the major moving parts of the revised cdc-sink code base. At present, each interface is implemented by a single type, but having small APIs has been useful in identifying the independent parts of cdc-sink. Similarly, the package structure may be overly fine-grained in favor of identifying specific, small portions of code that compose easily. A comment at the top of most files indicates where code was repackaged from. Recommended package review order: types, source/cdc, target/stage, target/apply, target/timestamp

bobvawter

TFTR. PTAL. TANSTAAFL.

Renamed the mid-level packages to to sources and target.

Reviewable status: 20 of 55 files reviewed, 36 unresolved discussions (waiting on @BramGruneir and @sravotto)

internal/util/batches/batches.go, line 29 at r10 (raw file):