Feat: implement spanner CDC for Bento

[anicoll]

TODO: add integration tests. Added

[jem-davies]

Hi @anicoll - thanks for the contribution! 🚀 - I should (or @gregfurman) get around to taking a look in a few days. Looks like there is just one lint check failing atm - is this something you could take a look at 🙏

[anicoll]

Hi @anicoll - thanks for the contribution! 🚀 - I should (or @gregfurman) get around to taking a look in a few days. Looks like there is just one lint check failing atm - is this something you could take a look at 🙏

I have fixed the lint issue and added some extra metadata tests to the READ func. Also made those variables.

[jem-davies]

Pretty much just doc changes / nits on this review, I have taken a look at the implementation and mostly looks good but want to check again regarding the use of contexts - how does this handle shutdown gracefully etc.

Thanks

[jem-davies]

I think that there is a problem with gracefully shutting down this component?

I set up a gcr.io/cloud-spanner-emulator/emulator similar to the integration test then manually sent SIGTERM to the bento process and got the logs in the log.txt file I have attached.

logs.txt

Also I think that the connect logic doesn't retry because I did get this once as well:

go run main.go -c ./tmp/0605_gcp_spanner_cdc_test/config.yaml
INFO Running main config from specified file       @service=bento bento_version="" path=./tmp/0605_gcp_spanner_cdc_test/config.yaml
INFO Listening for HTTP requests at: http://0.0.0.0:4195  @service=bento
ERRO Service closing due to: failed to init input <no label> path root.input: rpc error: code = FailedPrecondition desc = Schema change operation rejected because a concurrent schema change operation or read-write transaction is already in progress.  @service=bento
exit status 1

I think normally we put most of the connection logic in the Connect function of the component then the bento engine will retry to connect indefinitely rather than terminate?

There is also a small fix needed for the tests at the moment.

[anicoll]

@jem-davies
I have fixed said tests (apologies) and please see attached screenshot of the handling of a sigterm now.

If you want me to alter the log levels to anything specific please let me know!

I tested it like so
`
func main() {
ctx := context.Background()

go func() {
	time.Sleep(15 * time.Second)
	p, err := os.FindProcess(os.Getpid())
	if err != nil {
		panic(err)
	}
	if err := p.Signal(syscall.SIGTERM); err != nil {
		panic(err)
	}
}()

service.RunCLI(ctx)

}
`

Any more feedback is welcome..

[anicoll]

This PR resolves #341

[anicoll]

@jem-davies ping

[jem-davies]

@jem-davies ping

👋 @anicoll - haven't forgotten about - just working through a few other things on the todo list then will come back to this

If it's a blocker for you - I would suggest looking at implementing as a plugin.

[jem-davies]

I think that the issue regarding graceful shutdown is still an issue, was going to get round to taking a proper look and seeing if I could give more advice but currently taking a look at some other things.