c2c: plumb forSpanConfigs arg through streamclient.Create() #106792

msbutler · 2023-07-13T21:55:00Z

This patch plumbs the forSpanConfigs arg through the streamclient.Create()
interface and modifies producer job creation to return a specification for
streaming the spanConfiguration table if that arg is passed.

A future PR will plumb this info through the StreamPartition spec.

It's worth noting that we don't necessarily need this plumbing: if we see that
tenant Id is for the system tenant, then we could automatically set up a span
config stream. I think this extra plumbing is worth it because it makes
everything a bit easier to read.

UPDATE: I think there is a better API which I describe here

Informs #106823

Release note: None

cockroach-teamcity · 2023-07-13T21:55:12Z

This change is

This patch fixes a papercut where we were storing pointers to spans in the producer job proto. This isn't a correctness bug, as protocol buffers will dereference pointers to objects during unmarshalling, but rather, it seems dangerous to work with slices of pointers to spans. Release note: None

pkg/ccl/streamingccl/streamproducer/stream_lifetime.go

lidorcarmel · 2023-07-14T14:53:33Z

pkg/ccl/streamingccl/streamclient/partitioned_stream_client_test.go

@@ -249,7 +249,7 @@ INSERT INTO d.t2 VALUES (2);
 	h.SysSQL.Exec(t, `
 SET CLUSTER SETTING stream_replication.stream_liveness_track_frequency = '200ms';
 `)
-	rps, err = client.Create(ctx, testTenantName)
+	rps, err = client.Create(ctx, testTenantName, false)


test with forSpanConfigs=true in this file?

I plan to in a follow up PR, once I actually teach the partitioned_stream_client to process span config updates.

pkg/ccl/streamingccl/replicationtestutils/replication_helpers.go

pkg/ccl/streamingccl/streamclient/partitioned_stream_client.go

lidorcarmel · 2023-07-14T15:10:53Z

pkg/ccl/streamingccl/streamclient/partitioned_stream_client.go

 ) (streampb.ReplicationProducerSpec, error) {
 	ctx, sp := tracing.ChildSpan(ctx, "streamclient.Client.Create")
 	defer sp.Finish()

 	p.mu.Lock()
 	defer p.mu.Unlock()
 	var rawReplicationProducerSpec []byte
-	row := p.mu.srcConn.QueryRow(ctx, `SELECT crdb_internal.start_replication_stream($1)`, tenantName)
+	row := p.mu.srcConn.QueryRow(ctx, `SELECT crdb_internal.start_replication_stream($1,$2)`,
+		tenantName, forSpanConfig)


is it allowed to use forSpanConfig=true with a tenantName other than system?

good point. this query now fails if spanConfig==true and the tenanName is not system

thanks, I thought so, in that case, what do you think about splitting the API to have 2 funcs, one for span configs and a "normal" one? so that you would not need the boolean. something like:

Create(ctx context.Context, tenant roachpb.TenantName) (streampb.ReplicationProducerSpec, error) CreateForSpanConfigs(ctx context.Context) (streampb.ReplicationProducerSpec, error)

I don't think this is needed all over.. you'll decide where it makes more sense. overall I think it would be nicer than a bool, and callers will not be able to mix span configs and non-system tenant. another benefit might be that if we'll have crdb_internal.start_replication_stream_for_span_configs we'll get separate metrics for the 2 flavors (once we have metrics..).

We chatted offline and decided that it's fine to keep the API as is-- there's only one caller of this interface, the stream ingestion job, and it will only ever call Create() with and without the bool. If we need to add another stream flavor, it would be nicer to add some sort of opts struct to Create() . @stevendanna when you get a chance, please do chime in here if you have opinions on how I'm changing the api.

@lidorcarmel after thinking about this a bit more, I realized we need to persist the app tenant ID asap on the producer side. I think I need to modify this PR to do the following:

add a new field to the SpanConfigsForTenantID to the producer job record which persists the replicating app tenant id whose span config updates we want. The TenantID field will continue to be the system tenant id.

modify the startReplicationProducerJob(), to persist the SpanConfigsForTenantID field if necessary.

From the client's perspective, we can either add:

CreateForSpanConfigs(ctx context.Context, appTenant roachpb.TenantName) (streampb.ReplicationProducerSpec, error)

or modify the current impl in the PR to:

Create(ctx context.Context, appTenant roachpb.TenantName, forSpanConfigs bool) (streampb.ReplicationProducerSpec, error)

i.e. the client always passes in the appTenant name, never a system tenant.

We can discuss this on Monday.

msbutler

@lidorcarmel thanks for the quick review!

pkg/ccl/streamingccl/streamproducer/stream_lifetime.go

pkg/ccl/streamingccl/replicationtestutils/replication_helpers.go

pkg/ccl/streamingccl/streamclient/partitioned_stream_client.go

msbutler · 2023-07-14T15:43:22Z

pkg/ccl/streamingccl/streamclient/partitioned_stream_client.go

 ) (streampb.ReplicationProducerSpec, error) {
 	ctx, sp := tracing.ChildSpan(ctx, "streamclient.Client.Create")
 	defer sp.Finish()

 	p.mu.Lock()
 	defer p.mu.Unlock()
 	var rawReplicationProducerSpec []byte
-	row := p.mu.srcConn.QueryRow(ctx, `SELECT crdb_internal.start_replication_stream($1)`, tenantName)
+	row := p.mu.srcConn.QueryRow(ctx, `SELECT crdb_internal.start_replication_stream($1,$2)`,
+		tenantName, forSpanConfig)


good point. this query now fails if spanConfig==true and the tenanName is not system

msbutler · 2023-07-14T15:45:40Z

pkg/ccl/streamingccl/streamclient/partitioned_stream_client_test.go

@@ -249,7 +249,7 @@ INSERT INTO d.t2 VALUES (2);
 	h.SysSQL.Exec(t, `
 SET CLUSTER SETTING stream_replication.stream_liveness_track_frequency = '200ms';
 `)
-	rps, err = client.Create(ctx, testTenantName)
+	rps, err = client.Create(ctx, testTenantName, false)


I plan to in a follow up PR, once I actually teach the partitioned_stream_client to process span config updates.

This patch plumbs the forSpanConfigs arg through the streamclient.Create() interface and modifies producer job creation to return a specification for streaming the spanConfiguration table if that arg is passed. A future PR will plumb this info through the StreamPartition spec. It's worth noting that we don't necessarily need this plumbing: if we see that tenant Id is for the system tenant, then we could automatically set up a span config stream. I think this extra plumbing is worth it because it makes everything a bit easier to read. Informs cockroachdb#106823 Release note: None

msbutler · 2023-07-17T15:30:13Z

unrelated essential ci flake

msbutler · 2023-07-18T15:59:26Z

closing in favor of #107077

msbutler · 2023-07-18T16:36:17Z

unrelated flakes

msbutler self-assigned this Jul 13, 2023

msbutler force-pushed the butler-mr-span-config-plumbing branch 3 times, most recently from 2843d22 to 78100c7 Compare July 14, 2023 13:47

msbutler changed the title ~~Butler mr span config plumbing~~ c2c: plumb forSpanConfigs arg through streamclient.Create() Jul 14, 2023

msbutler force-pushed the butler-mr-span-config-plumbing branch from 78100c7 to f346995 Compare July 14, 2023 14:08

msbutler marked this pull request as ready for review July 14, 2023 14:09

msbutler requested review from a team as code owners July 14, 2023 14:09

msbutler requested review from lidorcarmel and stevendanna and removed request for a team July 14, 2023 14:09

msbutler added the T-disaster-recovery label Jul 14, 2023

lidorcarmel reviewed Jul 14, 2023

View reviewed changes

msbutler commented Jul 14, 2023

View reviewed changes

msbutler added the do-not-merge bors won't merge a PR with this label. label Jul 14, 2023

msbutler force-pushed the butler-mr-span-config-plumbing branch from 73ee93a to 298eca7 Compare July 14, 2023 16:58

msbutler removed the do-not-merge bors won't merge a PR with this label. label Jul 14, 2023

msbutler force-pushed the butler-mr-span-config-plumbing branch from 298eca7 to e285254 Compare July 14, 2023 20:45

msbutler force-pushed the butler-mr-span-config-plumbing branch from e285254 to 94ac2eb Compare July 14, 2023 21:07

msbutler closed this Jul 18, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

c2c: plumb forSpanConfigs arg through streamclient.Create() #106792

c2c: plumb forSpanConfigs arg through streamclient.Create() #106792

msbutler commented Jul 13, 2023 •

edited

Loading

cockroach-teamcity commented Jul 13, 2023

lidorcarmel Jul 14, 2023

msbutler Jul 14, 2023

lidorcarmel Jul 14, 2023

msbutler Jul 14, 2023

lidorcarmel Jul 14, 2023

msbutler Jul 14, 2023

msbutler Jul 14, 2023

msbutler left a comment

msbutler Jul 14, 2023

msbutler Jul 14, 2023

msbutler commented Jul 17, 2023

msbutler commented Jul 18, 2023

msbutler commented Jul 18, 2023

c2c: plumb forSpanConfigs arg through streamclient.Create() #106792

c2c: plumb forSpanConfigs arg through streamclient.Create() #106792

Conversation

msbutler commented Jul 13, 2023 • edited Loading

cockroach-teamcity commented Jul 13, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

msbutler left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

msbutler commented Jul 17, 2023

msbutler commented Jul 18, 2023

msbutler commented Jul 18, 2023

msbutler commented Jul 13, 2023 •

edited

Loading