sql: Add support for SHOW CHANGEFEED JOBS

[spiffyy99]

Previously, SHOW JOBS did not contain enough changefeed information
for users.
SHOW CHANGEFEED JOBS only shows changefeed jobs as well as the full
table name the changefeed monitors, the changefeed sink URI and the
format as computed columns.

Resolves #64933

Release note (sql change): Added new SHOW CHANGEFEED JOBS command with
additional information about changefeeds for improved user visibility

Specifically, SHOW CHANGEFEED JOBS adds columns sink_uri,
full_table_names, and format as well as high_water_timestamp from
crdb_internal.jobs.

Some of the metrics we wanted to include in here can't be added (i.e. buffer size, # of errors) since they aren't shown on an individual changefeed basis, can create another ticket for that once it is ready.

[cockroach-teamcity]

This change is 

[stevendanna]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @spiffyyeng)

a discussion (no related file):
Thanks for working on this! I think we and our customers are going to get a lot of use out of it going forward.

I left a few minor comments that I hope aren't too annoying. Overall though it looks pretty reasonable. I didn't review the actual query being used too closely since we are going to iterate on what information gets included there over time.

---

---

pkg/jobs/jobs_test.go, line 1732 at r1 (raw file):

	rowResults := sqlDB.Query(t, `SELECT job_id, job_type, sink_uri, full_table_names, format FROM [SHOW CHANGEFEED JOBS]`)

	rowResults.Next()

The sql.Rows interface requires that if Next() returns false that Err() is checked. In this case, it likely doesn't matter much, as the test will still fail at the Scan attempt, but it might produce slightly easier-to-debug failures if we get the error. Perhaps something like (warning, completely untested):

	if !rowResults.Next() {
		t.Fatalf("row expected but not found: %v", rowResult.Err())
	}
	err := rowResults.Scan(&out.id, &out.typ, &out.SinkURI, &out.FulLTableNames, &out.format)
	if err != nil {
		t.Fatal(err)
	}

---

pkg/jobs/jobs_test.go, line 1737 at r1 (raw file):

		t.Fatal(err)
	}
	if out.id != 1 ||

We are asserting the order of the results here, but I don't see an ORDER BY in the query that drives SHOW CHANGEFEED JOBS or in the SELECT above. It almost surely doesn't matter given the table and that this is a single-node test, but might be nice to put an order by on one of them

---

pkg/sql/delegate/show_jobs.go, line 41 at r1 (raw file):

               SELECT job_id, job_type, description, statement, user_name, status,
               running_status, created, started, finished, modified,
               fraction_completed, high_water_timestamp, error, coordinator_id, 

Doesn't need to change in this PR, but I think we found that coordinator_id is always NULL. Since this is a new command where we can change the output, perhaps we can remove it if we expect it to stay NULL? (Perhaps also fraction_completed and statement?)

---

pkg/sql/delegate/show_jobs.go, line 45 at r1 (raw file):

               ARRAY(SELECT concat(database_name, '.', schema_name, '.', name) from crdb_internal.tables WHERE table_id = any(descriptor_ids)) AS full_table_names,
               btrim(jsonb_pretty(json_extract_path(payload_json, 'changefeed', 'opts', 'format')), '"') AS format
				FROM crdb_internal.jobs INNER JOIN payload ON id = job_id`

nit: I think future us might appreciate a little bit more formatting here. Perhaps something like has been done in show_table.go?

[miretskiy]

Reviewed 1 of 6 files at r1.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @spiffyyeng and @stevendanna)

---

pkg/sql/delegate/show_jobs.go, line 45 at r1 (raw file):

Previously, stevendanna (Steven Danna) wrote…

nit: I think future us might appreciate a little bit more formatting here. Perhaps something like has been done in show_table.go?

I'll +1 this.

---

pkg/sql/delegate/show_jobs.go, line 74 at r1 (raw file):

	var sqlStmt string
	if n.WhichJobs == tree.ChangefeedJobs {
		sqlStmt = fmt.Sprintf("%s %s %s", selectClauseChangefeed, whereClause, orderbyClause)

You probably need to update parse tests: pkg/sql/parser/testdata/show

---

pkg/sql/sem/tree/show.go, line 255 at r1 (raw file):

const (

dro extra \n please

---

pkg/sql/sem/tree/show.go, line 302 at r1 (raw file):

	ctx.WriteString("SHOW ")
	if node.WhichJobs != NonAutomaticJobs {
		ctx.FormatNode(&node.WhichJobs)

you can probably call format node without above "if" check (you have default clause in job type format)

[spiffyy99]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

---

pkg/jobs/jobs_test.go, line 1732 at r1 (raw file):

Previously, stevendanna (Steven Danna) wrote…

The sql.Rows interface requires that if Next() returns false that Err() is checked. In this case, it likely doesn't matter much, as the test will still fail at the Scan attempt, but it might produce slightly easier-to-debug failures if we get the error. Perhaps something like (warning, completely untested):

	if !rowResults.Next() {
		t.Fatalf("row expected but not found: %v", rowResult.Err())
	}
	err := rowResults.Scan(&out.id, &out.typ, &out.SinkURI, &out.FulLTableNames, &out.format)
	if err != nil {
		t.Fatal(err)
	}

good catch. looks like Err() actually returns nil if there's no error (could just be no rows left), so accounted for this as well.

---

pkg/jobs/jobs_test.go, line 1737 at r1 (raw file):

Previously, stevendanna (Steven Danna) wrote…

We are asserting the order of the results here, but I don't see an ORDER BY in the query that drives SHOW CHANGEFEED JOBS or in the SELECT above. It almost surely doesn't matter given the table and that this is a single-node test, but might be nice to put an order by on one of them

Done.

---

pkg/sql/delegate/show_jobs.go, line 41 at r1 (raw file):

Previously, stevendanna (Steven Danna) wrote…

Doesn't need to change in this PR, but I think we found that coordinator_id is always NULL. Since this is a new command where we can change the output, perhaps we can remove it if we expect it to stay NULL? (Perhaps also fraction_completed and statement?)

i'll create a ticket

---

pkg/sql/sem/tree/show.go, line 255 at r1 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

dro extra \n please

Done.

---

pkg/sql/sem/tree/show.go, line 302 at r1 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

you can probably call format node without above "if" check (you have default clause in job type format)

Done.

[miretskiy]

Reviewed 2 of 6 files at r1, 2 of 5 files at r2.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @spiffyyeng and @stevendanna)

---

pkg/ccl/changefeedccl/changefeedbase/options.go, line 95 at r2 (raw file):

	SinkParamFileSize           = `file_size`
	SinkParamRegistryClientCert = `registry_client_cert`
	SinkParamRegistryClientKey  = `registry_client_cert`

why did we change these options?

---

pkg/jobs/jobs_test.go, line 1650 at r2 (raw file):

}

func TestShowChangefeedJobs(t *testing.T) {

I honestly don't think this test belongs here. I think it belongs in the changefeedccl package.

---

pkg/jobs/jobs_test.go, line 1724 at r2 (raw file):

		sqlDB.Exec(t,
			`INSERT INTO system.jobs (id, status, payload) VALUES ($1, $2, $3)`,
			in.id, in.status, inPayload,

I'm not too thrilled about inserting jobs like this. The registry could adopt this job; and start running it for example.

I think it would be better if we executed "CREATE CHANGEFEED" statement instead.
There are few ways we can prevent those "fake" changefeeds from being executed. But the one that I think makes most sense is to muck with TestingResumerCreationKnobs -- see example in import_stmt_test (sets up resumer for import statement).
Basically, you could write a "fake changefeed" resumer, which probably blocks until you cancel context.
After you created changefeed, you can examine it.

---

pkg/jobs/jobs_test.go, line 1753 at r2 (raw file):

			"but found id:%d, type:%s, sinkUri:%s, fullTableNames:%s and format:%s",
			1, "CHANGEFEED", "kafka://localhost:9092", "{defaultdb.public.foo}", "json",
			out.id, out.typ, out.SinkURI, string(out.FulLTableNames), out.format)

I like using require.Equal -- this prints expected/found automatically, and you can add extra information too
require.Equal(t, "json", out.format, "extra error information: %s ....", ...)

---

pkg/jobs/jobs_test.go, line 1777 at r2 (raw file):

			2, "CHANGEFEED", "experimental-s3://fake-bucket-name/fake/path", "{defaultdb.public.foo,defaultdb.public.bar}", "avro",
			out.id, out.typ, out.SinkURI, string(out.FulLTableNames), out.format)
	}

We are also missing tests when you try to show changefeed job 123 when the job is not a changefeed job.
We should also test how the statement behaves when the job is paused, cancelled, etc.

---

pkg/sql/delegate/show_jobs.go, line 72 at r2 (raw file):

    ), 
    '"'
  ) AS sink_uri, 

This is probably not okay. We need to make sure we don't show sensitive
information here -- and URIs can contain e.g. access keys.

For example, see func (m ScheduledBackupExecutionArgs) MarshalJSONPB(x *jsonpb.Marshaler) ([]byte, error) {
(in created_scheduled_backup) -- we had to write custom marshaller so that we won't display those secrets.

---

pkg/sql/delegate/show_jobs.go, line 97 at r2 (raw file):

  INNER JOIN payload ON id = job_id`
	)

This query is LARGE. And it's likely that it will grow larger still.
What's also likely, that it might no longer be feasible (or convenient) to use a 1 shot query to display
everything we may want to display. As such, I think it might make sense to move these changes into e.g. show_changefeed.go file. I believe you should be able to introduce tree.ShowChangefeed node (like ShowSchedules one).

[spiffyy99]

fixed now, not sure what happened there 😂

[spiffyy99]

Dismissed @miretskiy from 3 discussions.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

---

pkg/jobs/jobs_test.go, line 1650 at r2 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

I honestly don't think this test belongs here. I think it belongs in the changefeedccl package.

moved to changefeedccl

---

pkg/jobs/jobs_test.go, line 1724 at r2 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

I'm not too thrilled about inserting jobs like this. The registry could adopt this job; and start running it for example.

I think it would be better if we executed "CREATE CHANGEFEED" statement instead.
There are few ways we can prevent those "fake" changefeeds from being executed. But the one that I think makes most sense is to muck with TestingResumerCreationKnobs -- see example in import_stmt_test (sets up resumer for import statement).
Basically, you could write a "fake changefeed" resumer, which probably blocks until you cancel context.
After you created changefeed, you can examine it.

changed

---

pkg/sql/delegate/show_jobs.go, line 74 at r1 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

You probably need to update parse tests: pkg/sql/parser/testdata/show

Done.

---

pkg/sql/delegate/show_jobs.go, line 72 at r2 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

This is probably not okay. We need to make sure we don't show sensitive
information here -- and URIs can contain e.g. access keys.

For example, see func (m ScheduledBackupExecutionArgs) MarshalJSONPB(x *jsonpb.Marshaler) ([]byte, error) {
(in created_scheduled_backup) -- we had to write custom marshaller so that we won't display those secrets.

added this method for ChangfeedDetails, sinkuri is now redacted

---

pkg/sql/delegate/show_jobs.go, line 97 at r2 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

This query is LARGE. And it's likely that it will grow larger still.
What's also likely, that it might no longer be feasible (or convenient) to use a 1 shot query to display
everything we may want to display. As such, I think it might make sense to move these changes into e.g. show_changefeed.go file. I believe you should be able to introduce tree.ShowChangefeed node (like ShowSchedules one).

moved and simplified the query a bit

[miretskiy]

Reviewed 13 of 15 files at r3.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 28 at r3 (raw file):

	"github.com/stretchr/testify/require"
)

Overall -- good job on tests. Few nits/improvements here.

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 48 at r3 (raw file):

}

func waitForJobOperation(

nit: i'd call it waitForJobStatus

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 115 at r3 (raw file):

	// Cannot use kafka for tests right now because of leaked goroutine issue
	query = `CREATE CHANGEFEED FOR TABLE foo, bar INTO 
		'experimental-s3://fake-bucket-name/fake/path?AWS_ACCESS_KEY_ID=123&AWS_SECRET_ACCESS_KEY=456'`

the query itself returns jobID -- you should read that instead of querying jobs table below.

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 204 at r3 (raw file):

	sqlDB.Exec(t, query, changefeedID)

	waitForJobOperation(sqlDB, t, changefeedID, "paused")

you probably want to run show changefeed statements again after pausing/cancelling (and also you need to resume too).
You should verify that show changefeed job does something reasonable. Otherwise, why bother pausing/resuming? You're not
testing jobs system here.

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 212 at r3 (raw file):

}

func TestShowChangefeedJobsNoResults(t *testing.T) {

I think you should also create a different type of job (say backup), but run show changefeed job on that.
Make sure you get sane results.

---

pkg/sql/delegate/show_changefeed_jobs.go, line 26 at r3 (raw file):

	const (
		selectClause = `
WITH payload AS (

I'm still

---

pkg/sql/delegate/show_changefeed_jobs.go, line 38 at r3 (raw file):

SELECT 
  job_id, 
  job_type, 

probably not needed since you know it's a changefeed job?

---

pkg/sql/delegate/show_changefeed_jobs.go, line 49 at r3 (raw file):

  high_water_timestamp, 
  error, 
	replace(

nit: formatting a bit off?

---

pkg/sql/delegate/show_changefeed_jobs.go, line 52 at r3 (raw file):

		changefeed_details->>'sink_uri',
		'\u0026', '&'
	) AS sink_uri, 

doesn't sink_uri need to be redacted?
Ahh... I see. You have changefeed_details produced as pb_to_json -- and that did the redaction.
Perhaps add a comment?

---

pkg/sql/delegate/show_changefeed_jobs.go, line 66 at r3 (raw file):

	changefeed_details->'opts'->>'format' AS format 
FROM 
  crdb_internal.jobs 

I wonder why we need to query crdb_internal.jobs at all?
Actually, scratch that -- there is reason why we want to query that instead of system.jobs -- namely permission checking.
But that makes me a bit sad (cause I'm not a fan of crdb_internal.jobs)

Long time ago, I tried rewriting show jobs, and I was able to emulate permission checking directly in sql by doing something like:

+       const decodedJobs = `
+SELECT
+   id, status, created,
+   crdb_internal.pb_to_json('cockroach.sql.jobs.jobspb.Payload', payload) as payload,
+   crdb_internal.pb_to_json('cockroach.sql.jobs.jobspb.Progress', progress) as progress
+FROM system.jobs

+  permissions AS (
+    SELECT
+      crdb_internal.is_admin() as isAdmin,
+      crdb_internal.has_role_option('CONTROLJOB') as hasJobControl
+  ),
+  jobs AS (
+    SELECT *, %s
+    FROM (WITH decodedJobs AS (%s %s) SELECT * FROM decodedJobs), permissions
+    WHERE permissions.isAdmin OR permissions.hasJobControl OR payload->>'username' = current_user()
+  )

I don't know. Let me think a bit more about this query, and whether we should just not use crdb_internal.jobs.
For now, let's address other comments, but we might have to come back to this particular query.

---

pkg/sql/sem/tree/stmt.go, line 1254 at r3 (raw file):

// StatementTag returns a short string identifying the type of statement.
func (*ShowChangefeedJobs) StatementTag() string { return "SHOW JOBS" }

SHOW CHANGEFEED JOBS?

[spiffyy99]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

---

pkg/sql/delegate/show_changefeed_jobs.go, line 66 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

I wonder why we need to query crdb_internal.jobs at all?
Actually, scratch that -- there is reason why we want to query that instead of system.jobs -- namely permission checking.
But that makes me a bit sad (cause I'm not a fan of crdb_internal.jobs)

Long time ago, I tried rewriting show jobs, and I was able to emulate permission checking directly in sql by doing something like:

+       const decodedJobs = `
+SELECT
+   id, status, created,
+   crdb_internal.pb_to_json('cockroach.sql.jobs.jobspb.Payload', payload) as payload,
+   crdb_internal.pb_to_json('cockroach.sql.jobs.jobspb.Progress', progress) as progress
+FROM system.jobs

+  permissions AS (
+    SELECT
+      crdb_internal.is_admin() as isAdmin,
+      crdb_internal.has_role_option('CONTROLJOB') as hasJobControl
+  ),
+  jobs AS (
+    SELECT *, %s
+    FROM (WITH decodedJobs AS (%s %s) SELECT * FROM decodedJobs), permissions
+    WHERE permissions.isAdmin OR permissions.hasJobControl OR payload->>'username' = current_user()
+  )

I don't know. Let me think a bit more about this query, and whether we should just not use crdb_internal.jobs.
For now, let's address other comments, but we might have to come back to this particular query.

If we're reconsidering using crdb_internal.jobs, should we also consider it's use in SHOW JOBS? (maybe in another ticket)

[spiffyy99]

Dismissed @miretskiy from a discussion.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 204 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

you probably want to run show changefeed statements again after pausing/cancelling (and also you need to resume too).
You should verify that show changefeed job does something reasonable. Otherwise, why bother pausing/resuming? You're not
testing jobs system here.

Discussed offline.

[spiffyy99]

Dismissed @miretskiy from 5 discussions.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

---

pkg/sql/delegate/show_changefeed_jobs.go, line 38 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

probably not needed since you know it's a changefeed job?

Removed

---

pkg/sql/sem/tree/stmt.go, line 1254 at r3 (raw file):

Previously, miretskiy (Yevgeniy Miretskiy) wrote…

SHOW CHANGEFEED JOBS?

Nice catch

[miretskiy]

Reviewed 3 of 3 files at r4.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

a discussion (no related file):

Previously, stevendanna (Steven Danna) wrote…

Thanks for working on this! I think we and our customers are going to get a lot of use out of it going forward.

I left a few minor comments that I hope aren't too annoying. Overall though it looks pretty reasonable. I didn't review the actual query being used too closely since we are going to iterate on what information gets included there over time.

I think it's fine to use crdb_internal.jobs for now. Just small minor comments.
Let's see if @stevendanna has more to say on this PR.

---

---

pkg/sql/delegate/show_changefeed_jobs.go, line 24 at r4 (raw file):

	sqltelemetry.IncrementShowCounter(sqltelemetry.Jobs)

	// sink_uri has sensitive information redacted when JSON is marshaled from PB

Do you mind expanding this comment?
Something like:
Note: changefeed_details contains sensitive information in sink_url. This information is redacted by implementing MarshalJSONPB.

Also, perhaps we should move that method here?

[spiffyy99]

@stevendanna any more comments?

[miretskiy]

-- Feel free to update your test as needed.

[stevendanna]

Reviewable status: complete! 2 of 0 LGTMs obtained (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

a discussion (no related file):

---

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 114 at r5 (raw file):

	var changefeedID jobspb.JobID

	// Cannot use kafka for tests right now because of leaked goroutine issue

Do we have an open issue for this?

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 125 at r5 (raw file):

	require.Equal(t, changefeedID, out.id, "Expected id:%d but found id:%d", changefeedID, out.id)
	require.Equal(t, "experimental-s3://fake-bucket-name/fake/path?AWS_ACCESS_KEY_ID=123&AWS_SECRET_ACCESS_KEY=redacted", out.SinkURI, "Expected sinkUri:%s but found sinkUri:%s", "experimental-s3://fake-bucket-name/fake/path?AWS_ACCESS_KEY_ID=123&AWS_SECRET_ACCESS_KEY=redacted", out.SinkURI)

I think that require.Equal will already produce pretty nice output that contains both the expected and actual values, so you can probably avoid doing that manually if you'd like.

---

pkg/sql/delegate/show_changefeed_jobs.go, line 79 at r5 (raw file):

		// - then all completed jobs sorted in order of completion time.
		whereClause = fmt.Sprintf(
			`WHERE %s AND (finished IS NULL OR finished > now() - '12h':::interval)`, typePredicate)

We should be sure to document this 12 hour bit. I like the idea of limiting it by default.

[spiffyy99]

Reviewable status: complete! 0 of 0 LGTMs obtained (and 2 stale) (waiting on @miretskiy, @spiffyyeng, and @stevendanna)

---

pkg/ccl/changefeedccl/show_changefeed_jobs_test.go, line 114 at r5 (raw file):

Previously, stevendanna (Steven Danna) wrote…

Do we have an open issue for this?

no, but i can create one

---

pkg/sql/delegate/show_changefeed_jobs.go, line 79 at r5 (raw file):

Previously, stevendanna (Steven Danna) wrote…

We should be sure to document this 12 hour bit. I like the idea of limiting it by default.

it's on our website for SHOW JOBS: https://www.cockroachlabs.com/docs/v20.2/show-jobs.html#considerations

[stevendanna]

it's on our website for SHOW JOBS

Ah, cool, thanks!

[spiffyy99]

bors r+

[craig]

Build succeeded:

• GitHub CI (Cockroach)